diff options
author | Devtools Arcadia <[email protected]> | 2022-02-07 18:08:42 +0300 |
---|---|---|
committer | Devtools Arcadia <[email protected]> | 2022-02-07 18:08:42 +0300 |
commit | 1110808a9d39d4b808aef724c861a2e1a38d2a69 (patch) | |
tree | e26c9fed0de5d9873cce7e00bc214573dc2195b7 /contrib/tools/python3/src/Lib/sched.py |
intermediate changes
ref:cde9a383711a11544ce7e107a78147fb96cc4029
Diffstat (limited to 'contrib/tools/python3/src/Lib/sched.py')
-rw-r--r-- | contrib/tools/python3/src/Lib/sched.py | 167 |
1 files changed, 167 insertions, 0 deletions
diff --git a/contrib/tools/python3/src/Lib/sched.py b/contrib/tools/python3/src/Lib/sched.py new file mode 100644 index 00000000000..ff87874a3a4 --- /dev/null +++ b/contrib/tools/python3/src/Lib/sched.py @@ -0,0 +1,167 @@ +"""A generally useful event scheduler class. + +Each instance of this class manages its own queue. +No multi-threading is implied; you are supposed to hack that +yourself, or use a single instance per application. + +Each instance is parametrized with two functions, one that is +supposed to return the current time, one that is supposed to +implement a delay. You can implement real-time scheduling by +substituting time and sleep from built-in module time, or you can +implement simulated time by writing your own functions. This can +also be used to integrate scheduling with STDWIN events; the delay +function is allowed to modify the queue. Time can be expressed as +integers or floating point numbers, as long as it is consistent. + +Events are specified by tuples (time, priority, action, argument, kwargs). +As in UNIX, lower priority numbers mean higher priority; in this +way the queue can be maintained as a priority queue. Execution of the +event means calling the action function, passing it the argument +sequence in "argument" (remember that in Python, multiple function +arguments are be packed in a sequence) and keyword parameters in "kwargs". +The action function may be an instance method so it +has another way to reference private data (besides global variables). +""" + +import time +import heapq +from collections import namedtuple +import threading +from time import monotonic as _time + +__all__ = ["scheduler"] + +class Event(namedtuple('Event', 'time, priority, action, argument, kwargs')): + __slots__ = [] + def __eq__(s, o): return (s.time, s.priority) == (o.time, o.priority) + def __lt__(s, o): return (s.time, s.priority) < (o.time, o.priority) + def __le__(s, o): return (s.time, s.priority) <= (o.time, o.priority) + def __gt__(s, o): return (s.time, s.priority) > (o.time, o.priority) + def __ge__(s, o): return (s.time, s.priority) >= (o.time, o.priority) + +Event.time.__doc__ = ('''Numeric type compatible with the return value of the +timefunc function passed to the constructor.''') +Event.priority.__doc__ = ('''Events scheduled for the same time will be executed +in the order of their priority.''') +Event.action.__doc__ = ('''Executing the event means executing +action(*argument, **kwargs)''') +Event.argument.__doc__ = ('''argument is a sequence holding the positional +arguments for the action.''') +Event.kwargs.__doc__ = ('''kwargs is a dictionary holding the keyword +arguments for the action.''') + +_sentinel = object() + +class scheduler: + + def __init__(self, timefunc=_time, delayfunc=time.sleep): + """Initialize a new instance, passing the time and delay + functions""" + self._queue = [] + self._lock = threading.RLock() + self.timefunc = timefunc + self.delayfunc = delayfunc + + def enterabs(self, time, priority, action, argument=(), kwargs=_sentinel): + """Enter a new event in the queue at an absolute time. + + Returns an ID for the event which can be used to remove it, + if necessary. + + """ + if kwargs is _sentinel: + kwargs = {} + event = Event(time, priority, action, argument, kwargs) + with self._lock: + heapq.heappush(self._queue, event) + return event # The ID + + def enter(self, delay, priority, action, argument=(), kwargs=_sentinel): + """A variant that specifies the time as a relative time. + + This is actually the more commonly used interface. + + """ + time = self.timefunc() + delay + return self.enterabs(time, priority, action, argument, kwargs) + + def cancel(self, event): + """Remove an event from the queue. + + This must be presented the ID as returned by enter(). + If the event is not in the queue, this raises ValueError. + + """ + with self._lock: + self._queue.remove(event) + heapq.heapify(self._queue) + + def empty(self): + """Check whether the queue is empty.""" + with self._lock: + return not self._queue + + def run(self, blocking=True): + """Execute events until the queue is empty. + If blocking is False executes the scheduled events due to + expire soonest (if any) and then return the deadline of the + next scheduled call in the scheduler. + + When there is a positive delay until the first event, the + delay function is called and the event is left in the queue; + otherwise, the event is removed from the queue and executed + (its action function is called, passing it the argument). If + the delay function returns prematurely, it is simply + restarted. + + It is legal for both the delay function and the action + function to modify the queue or to raise an exception; + exceptions are not caught but the scheduler's state remains + well-defined so run() may be called again. + + A questionable hack is added to allow other threads to run: + just after an event is executed, a delay of 0 is executed, to + avoid monopolizing the CPU when other threads are also + runnable. + + """ + # localize variable access to minimize overhead + # and to improve thread safety + lock = self._lock + q = self._queue + delayfunc = self.delayfunc + timefunc = self.timefunc + pop = heapq.heappop + while True: + with lock: + if not q: + break + time, priority, action, argument, kwargs = q[0] + now = timefunc() + if time > now: + delay = True + else: + delay = False + pop(q) + if delay: + if not blocking: + return time - now + delayfunc(time - now) + else: + action(*argument, **kwargs) + delayfunc(0) # Let other threads run + + @property + def queue(self): + """An ordered list of upcoming events. + + Events are named tuples with fields for: + time, priority, action, arguments, kwargs + + """ + # Use heapq to sort the queue rather than using 'sorted(self._queue)'. + # With heapq, two events scheduled at the same time will show in + # the actual order they would be retrieved. + with self._lock: + events = self._queue[:] + return list(map(heapq.heappop, [events]*len(events))) |