/usr/lib/python3/dist-packages/simpy/util.py is in python3-simpy3 3.0.10-2.
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 | """
A collection of utility functions:
.. autosummary::
start_delayed
"""
def start_delayed(env, generator, delay):
"""Return a helper process that starts another process for *generator*
after a certain *delay*.
:meth:`~simpy.core.Environment.process()` starts a process at the current
simulation time. This helper allows you to start a process after a delay of
*delay* simulation time units::
>>> from simpy import Environment
>>> from simpy.util import start_delayed
>>> def my_process(env, x):
... print('%s, %s' % (env.now, x))
... yield env.timeout(1)
...
>>> env = Environment()
>>> proc = start_delayed(env, my_process(env, 3), 5)
>>> env.run()
5, 3
Raise a :exc:`ValueError` if ``delay <= 0``.
"""
if delay <= 0:
raise ValueError('delay(=%s) must be > 0.' % delay)
def starter():
yield env.timeout(delay)
proc = env.process(generator)
env.exit(proc)
return env.process(starter())
def subscribe_at(event):
"""Register at the *event* to receive an interrupt when it occurs.
The most common use case for this is to pass
a :class:`~simpy.events.Process` to get notified when it terminates.
Raise a :exc:`RuntimeError` if ``event`` has already occurred.
"""
env = event.env
subscriber = env.active_process
def signaller(signaller, receiver):
result = yield signaller
if receiver.is_alive:
receiver.interrupt((signaller, result))
if event.callbacks is not None:
env.process(signaller(event, subscriber))
else:
raise RuntimeError('%s has already terminated.' % event)
|