Python code coverage for Lib/sched.py

#	count	content
1	n/a	"""A generally useful event scheduler class.
2	n/a
3	n/a	Each instance of this class manages its own queue.
4	n/a	No multi-threading is implied; you are supposed to hack that
5	n/a	yourself, or use a single instance per application.
6	n/a
7	n/a	Each instance is parametrized with two functions, one that is
8	n/a	supposed to return the current time, one that is supposed to
9	n/a	implement a delay. You can implement real-time scheduling by
10	n/a	substituting time and sleep from built-in module time, or you can
11	n/a	implement simulated time by writing your own functions. This can
12	n/a	also be used to integrate scheduling with STDWIN events; the delay
13	n/a	function is allowed to modify the queue. Time can be expressed as
14	n/a	integers or floating point numbers, as long as it is consistent.
15	n/a
16	n/a	Events are specified by tuples (time, priority, action, argument, kwargs).
17	n/a	As in UNIX, lower priority numbers mean higher priority; in this
18	n/a	way the queue can be maintained as a priority queue. Execution of the
19	n/a	event means calling the action function, passing it the argument
20	n/a	sequence in "argument" (remember that in Python, multiple function
21	n/a	arguments are be packed in a sequence) and keyword parameters in "kwargs".
22	n/a	The action function may be an instance method so it
23	n/a	has another way to reference private data (besides global variables).
24	n/a	"""
25	n/a
26	n/a	import time
27	n/a	import heapq
28	n/a	from collections import namedtuple
29	n/a	try:
30	n/a	import threading
31	n/a	except ImportError:
32	n/a	import dummy_threading as threading
33	n/a	from time import monotonic as _time
34	n/a
35	n/a	__all__ = ["scheduler"]
36	n/a
37	n/a	class Event(namedtuple('Event', 'time, priority, action, argument, kwargs')):
38	n/a	__slots__ = []
39	n/a	def __eq__(s, o): return (s.time, s.priority) == (o.time, o.priority)
40	n/a	def __lt__(s, o): return (s.time, s.priority) < (o.time, o.priority)
41	n/a	def __le__(s, o): return (s.time, s.priority) <= (o.time, o.priority)
42	n/a	def __gt__(s, o): return (s.time, s.priority) > (o.time, o.priority)
43	n/a	def __ge__(s, o): return (s.time, s.priority) >= (o.time, o.priority)
44	n/a
45	n/a	Event.time.__doc__ = ('''Numeric type compatible with the return value of the
46	n/a	timefunc function passed to the constructor.''')
47	n/a	Event.priority.__doc__ = ('''Events scheduled for the same time will be executed
48	n/a	in the order of their priority.''')
49	n/a	Event.action.__doc__ = ('''Executing the event means executing
50	n/a	action(argument, *kwargs)''')
51	n/a	Event.argument.__doc__ = ('''argument is a sequence holding the positional
52	n/a	arguments for the action.''')
53	n/a	Event.kwargs.__doc__ = ('''kwargs is a dictionary holding the keyword
54	n/a	arguments for the action.''')
55	n/a
56	n/a	_sentinel = object()
57	n/a
58	n/a	class scheduler:
59	n/a
60	n/a	def __init__(self, timefunc=_time, delayfunc=time.sleep):
61	n/a	"""Initialize a new instance, passing the time and delay
62	n/a	functions"""
63	n/a	self._queue = []
64	n/a	self._lock = threading.RLock()
65	n/a	self.timefunc = timefunc
66	n/a	self.delayfunc = delayfunc
67	n/a
68	n/a	def enterabs(self, time, priority, action, argument=(), kwargs=_sentinel):
69	n/a	"""Enter a new event in the queue at an absolute time.
70	n/a
71	n/a	Returns an ID for the event which can be used to remove it,
72	n/a	if necessary.
73	n/a
74	n/a	"""
75	n/a	if kwargs is _sentinel:
76	n/a	kwargs = {}
77	n/a	event = Event(time, priority, action, argument, kwargs)
78	n/a	with self._lock:
79	n/a	heapq.heappush(self._queue, event)
80	n/a	return event # The ID
81	n/a
82	n/a	def enter(self, delay, priority, action, argument=(), kwargs=_sentinel):
83	n/a	"""A variant that specifies the time as a relative time.
84	n/a
85	n/a	This is actually the more commonly used interface.
86	n/a
87	n/a	"""
88	n/a	time = self.timefunc() + delay
89	n/a	return self.enterabs(time, priority, action, argument, kwargs)
90	n/a
91	n/a	def cancel(self, event):
92	n/a	"""Remove an event from the queue.
93	n/a
94	n/a	This must be presented the ID as returned by enter().
95	n/a	If the event is not in the queue, this raises ValueError.
96	n/a
97	n/a	"""
98	n/a	with self._lock:
99	n/a	self._queue.remove(event)
100	n/a	heapq.heapify(self._queue)
101	n/a
102	n/a	def empty(self):
103	n/a	"""Check whether the queue is empty."""
104	n/a	with self._lock:
105	n/a	return not self._queue
106	n/a
107	n/a	def run(self, blocking=True):
108	n/a	"""Execute events until the queue is empty.
109	n/a	If blocking is False executes the scheduled events due to
110	n/a	expire soonest (if any) and then return the deadline of the
111	n/a	next scheduled call in the scheduler.
112	n/a
113	n/a	When there is a positive delay until the first event, the
114	n/a	delay function is called and the event is left in the queue;
115	n/a	otherwise, the event is removed from the queue and executed
116	n/a	(its action function is called, passing it the argument). If
117	n/a	the delay function returns prematurely, it is simply
118	n/a	restarted.
119	n/a
120	n/a	It is legal for both the delay function and the action
121	n/a	function to modify the queue or to raise an exception;
122	n/a	exceptions are not caught but the scheduler's state remains
123	n/a	well-defined so run() may be called again.
124	n/a
125	n/a	A questionable hack is added to allow other threads to run:
126	n/a	just after an event is executed, a delay of 0 is executed, to
127	n/a	avoid monopolizing the CPU when other threads are also
128	n/a	runnable.
129	n/a
130	n/a	"""
131	n/a	# localize variable access to minimize overhead
132	n/a	# and to improve thread safety
133	n/a	lock = self._lock
134	n/a	q = self._queue
135	n/a	delayfunc = self.delayfunc
136	n/a	timefunc = self.timefunc
137	n/a	pop = heapq.heappop
138	n/a	while True:
139	n/a	with lock:
140	n/a	if not q:
141	n/a	break
142	n/a	time, priority, action, argument, kwargs = q[0]
143	n/a	now = timefunc()
144	n/a	if time > now:
145	n/a	delay = True
146	n/a	else:
147	n/a	delay = False
148	n/a	pop(q)
149	n/a	if delay:
150	n/a	if not blocking:
151	n/a	return time - now
152	n/a	delayfunc(time - now)
153	n/a	else:
154	n/a	action(argument, *kwargs)
155	n/a	delayfunc(0) # Let other threads run
156	n/a
157	n/a	@property
158	n/a	def queue(self):
159	n/a	"""An ordered list of upcoming events.
160	n/a
161	n/a	Events are named tuples with fields for:
162	n/a	time, priority, action, arguments, kwargs
163	n/a
164	n/a	"""
165	n/a	# Use heapq to sort the queue rather than using 'sorted(self._queue)'.
166	n/a	# With heapq, two events scheduled at the same time will show in
167	n/a	# the actual order they would be retrieved.
168	n/a	with self._lock:
169	n/a	events = self._queue[:]
170	n/a	return list(map(heapq.heappop, [events]*len(events)))