/usr/lib/python2.7/dist-packages/PySPH-1.0a4.dev0-py2.7-linux-x86_64.egg/pysph/solver/controller.py is in python-pysph 0~20160514.git91867dc-4build1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 | ''' Implement infrastructure for the solver to add various interfaces '''
from functools import wraps
import threading
try:
from thread import LockType
except ImportError:
from _thread import LockType
from pysph.base.particle_array import ParticleArray
import logging
logger = logging.getLogger(__name__)
class DummyComm(object):
''' A dummy MPI.Comm implementation as placeholder for for serial runs '''
def Get_size(self):
''' return the size of the comm (1) '''
return 1
def Get_rank(self):
''' return the rank of the process (0) '''
return 0
def send(self, data, pid):
''' dummy send implementation '''
self.data = data
def recv(self, pid):
''' dummy recv implementation '''
data = self.data
del self.data
return data
def bcast(self, data):
''' bcast (broadcast) implementation for serial run '''
return data
def gather(self, data):
''' gather implementation for serial run '''
return [data]
def synchronized(lock_or_func):
''' decorator for synchronized (thread safe) function
Usage:
- sync_func = synchronized(lock)(func) # sync with an existing lock
- sync_func = synchronized(func) # sync with a new private lock
'''
if isinstance(lock_or_func, LockType):
lock = lock_or_func
def synchronized_inner(func):
@wraps(func)
def wrapped(*args, **kwargs):
with lock:
return func(*args, **kwargs)
return wrapped
return synchronized_inner
else:
func = lock_or_func
lock = threading.Lock()
return synchronized(lock)(func)
def wrap_dispatcher(obj, meth, *args2, **kwargs2):
@wraps(meth)
def wrapped(*args, **kwargs):
kw = {}
kw.update(kwargs2)
kw.update(kwargs)
return meth(obj.block, *(args2+args), **kw)
return wrapped
class Controller(object):
''' Controller class acts a a proxy to control the solver
This is passed as an argument to the interface
**Methods available**:
- get -- get the value of a solver parameter
- set -- set the value of a solver parameter
- get_result -- return result of a queued command
- pause_on_next -- pause solver thread on next iteration
- wait -- wait (block) calling thread till solver is paused
(call after `pause_on_next`)
- cont -- continue solver thread (call after `pause_on_next`)
Various other methods are also available as listed in
:data:`CommandManager.dispatch_dict` which perform different functions.
- The methods in CommandManager.active_methods do their operation and return
the result (if any) immediately
- The methods in CommandManager.lazy_methods do their later when solver
thread is available and return a task-id. The result of the task can be
obtained later using the blocking call `get_result()` which waits till
result is available and returns the result.
The availability of the result can be checked using the lock returned
by `get_task_lock()` method
FIXME: wait/cont currently do not work in parallel
'''
def __init__(self, command_manager, block=True):
super(Controller, self).__init__()
self.__command_manager = command_manager
self.daemon = True
self.block = block
self._set_methods()
def _set_methods(self):
for prop in self.__command_manager.solver_props:
setattr(self, 'get_'+prop, wrap_dispatcher(self, self.__command_manager.dispatch, 'get', prop))
setattr(self, 'set_'+prop, wrap_dispatcher(self, self.__command_manager.dispatch, 'set', prop))
for meth in self.__command_manager.solver_methods:
setattr(self, meth, wrap_dispatcher(self, self.__command_manager.dispatch, meth))
for meth in self.__command_manager.lazy_methods:
setattr(self, meth, wrap_dispatcher(self, self.__command_manager.dispatch, meth))
for meth in self.__command_manager.active_methods:
setattr(self, meth, wrap_dispatcher(self, self.__command_manager.dispatch, meth))
def get(self, name):
''' get a solver property; returns immediately '''
return self.__command_manager.dispatch(self.block, 'get', name)
def set(self, name, value):
''' set a solver property; returns immediately; '''
return self.__command_manager.dispatch(self.block, 'set', name, value)
def pause_on_next(self):
''' pause the solver thread on next iteration '''
return self.__command_manager.pause_on_next()
def wait(self):
''' block the calling thread until the solver thread pauses
call this only after calling the `pause_on_next` method to tell
the controller to pause the solver thread'''
self.__command_manager.wait()
return True
def get_prop_names(self):
return list(self.__command_manager.solver_props)
def cont(self):
''' continue solver thread after it has been paused by `pause_on_next`
call this only after calling the `pause_on_next` method '''
return self.__command_manager.cont()
def get_result(self, task_id):
''' get the result of a previously queued command '''
return self.__command_manager.get_result(task_id)
def set_blocking(self, block):
''' set the blocking mode to True/False
In blocking mode (block=True) all methods other than getting of
solver properties block until the command is executed by the solver
and return the results. The blocking time can vary depending on the
time taken by solver per iteration and the command_interval
In non-blocking mode, these methods queue the command for later
and return a string corresponding to the task_id of the operation.
The result can be later obtained by a (blocking) call to get_result
with the task_id as argument
'''
if block != self.block:
self.block = block
self._set_methods()
return self.block
def get_blocking(self):
''' get the blocking mode ( True/False ) '''
return self.block
def ping(self):
return True
def on_root_proc(f):
''' run the decorated function only on the root proc '''
@wraps(f)
def wrapper(self, *args, **kwds):
if self.comm.Get_rank()==0:
return f(self, *args, **kwds)
return wrapper
def in_parallel(f):
''' return a list of results of running decorated function on all procs '''
@wraps(f)
def wrapper(self, *args, **kwds):
return self.comm.gather(f(self, *args, **kwds))
return wrapper
class CommandManager(object):
''' Class to manage and synchronize commands from various Controllers '''
solver_props = set(('t', 'tf', 'dt', 'count', 'pfreq', 'fname',
'detailed_output', 'output_directory', 'command_interval'))
solver_methods = set(('dump_output',))
lazy_methods = set(('get_particle_array_names', 'get_named_particle_array',
'get_particle_array_combined', 'get_particle_array_from_procs'))
active_methods = set(('get_status', 'get_task_lock', 'set_log_level'))
def __init__(self, solver, comm=None):
if comm is not None:
self.comm = comm
self.rank = comm.Get_rank()
else:
try:
self.comm = solver.particles.cell_manager.parallel_controller.comm
except AttributeError:
self.comm = DummyComm()
self.rank = 0
logger.debug('CommandManager: using comm: %s'%self.comm)
self.solver = solver
self.interfaces = []
self.func_dict = {}
self.rlock = threading.RLock()
self.res_lock = threading.Lock()
self.plock = threading.Condition()
self.qlock = threading.Condition() # queue lock
self.queue = []
self.queue_dict = {}
self.queue_lock_map = {}
self.results = {}
self.pause = set([])
@on_root_proc
def add_interface(self, callable, block=True):
''' Add a callable interface to the controller
The callable must accept an Controller instance argument.
The callable is called in a new thread of its own and it can
do various actions with methods defined on the Controller
instance passed to it
The new created thread is set to daemon mode and returned
'''
logger.debug('adding_interface: %s'%callable)
control = Controller(self, block)
thr = threading.Thread(target=callable, args=(control,))
thr.daemon = True
thr.start()
return thr
def add_function(self, callable, interval=1):
''' add a function to to be called every `interval` iterations '''
l = self.func_dict[interval] = self.func_dict.get(interval, [])
l.append(callable)
def execute_commands(self, solver):
''' called by the solver after each timestep '''
# TODO: first synchronize all the controllers in different processes
# using mpi
self.sync_commands()
with self.qlock:
self.run_queued_commands()
if self.rank == 0:
logger.debug('control handler: count=%d'%solver.count)
for interval in self.func_dict:
if solver.count%interval == 0:
for func in self.func_dict[interval]:
func(solver)
self.wait_for_cmd()
def wait_for_cmd(self):
''' wait for command from any interface '''
with self.qlock:
while self.pause:
with self.plock:
self.plock.notify_all()
self.qlock.wait()
self.run_queued_commands()
def sync_commands(self):
''' send the pending commands to all the procs in parallel run '''
self.queue_dict, self.queue, self.pause = self.comm.bcast((self.queue_dict, self.queue, self.pause))
def run_queued_commands(self):
while self.queue:
lock_id = self.queue.pop(0)
meth, args, kwargs = self.queue_dict[lock_id]
with self.res_lock:
try:
self.results[lock_id] = self.run_command(meth, args, kwargs)
finally:
del self.queue_dict[lock_id]
if self.comm.Get_rank()==0:
self.queue_lock_map[lock_id].release()
def run_command(self, cmd, args=[], kwargs={}):
res = self.dispatch_dict[cmd](self, *args, **kwargs)
logger.debug('controller: running_command: %s %s %s %s'%(
cmd, args, kwargs, res))
return res
def pause_on_next(self):
''' pause and wait for command on the next control interval '''
if self.comm.Get_size() > 1:
logger.debug('pause/continue not yet supported in parallel runs')
return False
with self.plock:
self.pause.add(threading.current_thread().ident)
self.plock.notify()
return True
def wait(self):
with self.plock:
self.plock.wait()
def cont(self):
''' continue after a pause command '''
if self.comm.Get_size() > 1:
logger.debug('pause/continue noy yet supported in parallel runs')
return
with self.plock:
self.pause.remove(threading.current_thread().ident)
self.plock.notify()
with self.qlock:
self.qlock.notify_all()
def get_result(self, lock_id):
''' get the result of a previously queued command '''
lock_id = int(lock_id)
lock = self.queue_lock_map[lock_id]
with lock:
with self.res_lock:
ret = self.results[lock_id]
del self.results[lock_id]
del self.queue_lock_map[lock_id]
return ret
def get_task_lock(self, lock_id):
''' get the Lock instance associated with a command '''
return self.queue_lock_map[int(lock_id)]
def get_prop(self, name):
''' get a solver property '''
return getattr(self.solver, name)
def set_prop(self, name, value):
''' set a solver property '''
return setattr(self.solver, name, value)
def solver_method(self, name, *args, **kwargs):
''' execute a method on the solver '''
ret = getattr(self.solver, name)(*args, **kwargs)
ret = self.comm.gather(ret)
return ret
def get_particle_array_names(self):
''' get the names of the particle arrays '''
return [pa.name for pa in self.solver.particles]
def get_named_particle_array(self, name, props=None):
for pa in self.solver.particles:
if pa.name == name:
if props:
return [getattr(pa, p) for p in props if hasattr(pa, p)]
else:
return pa
def get_particle_array_index(self, name):
''' get the index of the named particle array '''
for i,pa in enumerate(self.solver.particles):
if pa.name == name:
return i
def get_particle_array_from_procs(self, idx, procs=None):
''' get particle array at index from all processes
specifying processes is currently not implemented
'''
if procs is None:
procs = list(range(self.comm.size))
pa = self.solver.particles[idx]
pas = self.comm.gather(pa)
return pas
def get_particle_array_combined(self, idx, procs=None):
''' get a single particle array with combined data from all procs
specifying processes is currently not implemented
'''
if procs is None:
procs = list(range(self.comm.size))
pa = self.solver.particles[idx]
pas = self.comm.gather(pa)
pa = ParticleArray(name=pa.name)
for p in pas:
pa.append_parray(p)
return pa
def get_status(self):
''' get the status of the controller '''
return 'commands queued: %d'%len(self.queue)
def set_log_level(self, level):
''' set the logging level '''
logger.setLevel(level)
dispatch_dict = {'get':get_prop, 'set':set_prop}
for meth in solver_methods:
dispatch_dict[meth] = solver_method
for meth in lazy_methods:
dispatch_dict[meth] = locals()[meth]
for meth in active_methods:
dispatch_dict[meth] = locals()[meth]
@synchronized
def dispatch(self, block, meth, *args, **kwargs):
''' execute/queue a command with specified arguments '''
if meth in self.dispatch_dict:
if meth=='get' or meth=='set':
prop = args[0]
if prop not in self.solver_props:
raise RuntimeError('Invalid dispatch on method: %s with '
'non-existant property: %s '%(meth,prop))
if block or meth=='get' or meth in self.active_methods:
logger.debug('controller: immediate dispatch(): %s %s %s'%(
meth, args, kwargs))
return self.dispatch_dict[meth](self, *args, **kwargs)
else:
lock = threading.Lock()
lock.acquire()
lock_id = id(lock)
with self.qlock:
self.queue_lock_map[lock_id] = lock
self.queue_dict[lock_id] = (meth, args, kwargs)
self.queue.append(lock_id)
logger.debug('controller: dispatch(%d): %s %s %s'%(
lock_id, meth, args, kwargs))
return str(lock_id)
else:
raise RuntimeError('Invalid dispatch on method: '+meth)
|