|
1959
|
1 # Defines classes that provide synchronization objects. Note that use of
|
|
|
2 # this module requires that your Python support threads.
|
|
|
3 #
|
|
|
4 # condition() # a POSIX-like condition-variable object
|
|
|
5 # barrier(n) # an n-thread barrier
|
|
|
6 # event() # an event object
|
|
|
7 # semaphore(n=1)# a semaphore object, with initial count n
|
|
|
8 #
|
|
|
9 # CONDITIONS
|
|
|
10 #
|
|
|
11 # A condition object is created via
|
|
|
12 # import this_module
|
|
|
13 # your_condition_object = this_module.condition()
|
|
|
14 #
|
|
|
15 # Methods:
|
|
|
16 # .acquire()
|
|
|
17 # acquire the lock associated with the condition
|
|
|
18 # .release()
|
|
|
19 # release the lock associated with the condition
|
|
|
20 # .wait()
|
|
|
21 # block the thread until such time as some other thread does a
|
|
|
22 # .signal or .broadcast on the same condition, and release the
|
|
|
23 # lock associated with the condition. The lock associated with
|
|
|
24 # the condition MUST be in the acquired state at the time
|
|
|
25 # .wait is invoked.
|
|
|
26 # .signal()
|
|
|
27 # wake up exactly one thread (if any) that previously did a .wait
|
|
|
28 # on the condition; that thread will awaken with the lock associated
|
|
|
29 # with the condition in the acquired state. If no threads are
|
|
|
30 # .wait'ing, this is a nop. If more than one thread is .wait'ing on
|
|
|
31 # the condition, any of them may be awakened.
|
|
|
32 # .broadcast()
|
|
|
33 # wake up all threads (if any) that are .wait'ing on the condition;
|
|
|
34 # the threads are woken up serially, each with the lock in the
|
|
|
35 # acquired state, so should .release() as soon as possible. If no
|
|
|
36 # threads are .wait'ing, this is a nop.
|
|
|
37 #
|
|
|
38 # Note that if a thread does a .wait *while* a signal/broadcast is
|
|
|
39 # in progress, it's guaranteeed to block until a subsequent
|
|
|
40 # signal/broadcast.
|
|
|
41 #
|
|
|
42 # Secret feature: `broadcast' actually takes an integer argument,
|
|
|
43 # and will wake up exactly that many waiting threads (or the total
|
|
|
44 # number waiting, if that's less). Use of this is dubious, though,
|
|
|
45 # and probably won't be supported if this form of condition is
|
|
|
46 # reimplemented in C.
|
|
|
47 #
|
|
|
48 # DIFFERENCES FROM POSIX
|
|
|
49 #
|
|
|
50 # + A separate mutex is not needed to guard condition data. Instead, a
|
|
|
51 # condition object can (must) be .acquire'ed and .release'ed directly.
|
|
|
52 # This eliminates a common error in using POSIX conditions.
|
|
|
53 #
|
|
|
54 # + Because of implementation difficulties, a POSIX `signal' wakes up
|
|
|
55 # _at least_ one .wait'ing thread. Race conditions make it difficult
|
|
|
56 # to stop that. This implementation guarantees to wake up only one,
|
|
|
57 # but you probably shouldn't rely on that.
|
|
|
58 #
|
|
|
59 # PROTOCOL
|
|
|
60 #
|
|
|
61 # Condition objects are used to block threads until "some condition" is
|
|
|
62 # true. E.g., a thread may wish to wait until a producer pumps out data
|
|
|
63 # for it to consume, or a server may wish to wait until someone requests
|
|
|
64 # its services, or perhaps a whole bunch of threads want to wait until a
|
|
|
65 # preceding pass over the data is complete. Early models for conditions
|
|
|
66 # relied on some other thread figuring out when a blocked thread's
|
|
|
67 # condition was true, and made the other thread responsible both for
|
|
|
68 # waking up the blocked thread and guaranteeing that it woke up with all
|
|
|
69 # data in a correct state. This proved to be very delicate in practice,
|
|
|
70 # and gave conditions a bad name in some circles.
|
|
|
71 #
|
|
|
72 # The POSIX model addresses these problems by making a thread responsible
|
|
|
73 # for ensuring that its own state is correct when it wakes, and relies
|
|
|
74 # on a rigid protocol to make this easy; so long as you stick to the
|
|
|
75 # protocol, POSIX conditions are easy to "get right":
|
|
|
76 #
|
|
|
77 # A) The thread that's waiting for some arbitrarily-complex condition
|
|
|
78 # (ACC) to become true does:
|
|
|
79 #
|
|
|
80 # condition.acquire()
|
|
|
81 # while not (code to evaluate the ACC):
|
|
|
82 # condition.wait()
|
|
|
83 # # That blocks the thread, *and* releases the lock. When a
|
|
|
84 # # condition.signal() happens, it will wake up some thread that
|
|
|
85 # # did a .wait, *and* acquire the lock again before .wait
|
|
|
86 # # returns.
|
|
|
87 # #
|
|
|
88 # # Because the lock is acquired at this point, the state used
|
|
|
89 # # in evaluating the ACC is frozen, so it's safe to go back &
|
|
|
90 # # reevaluate the ACC.
|
|
|
91 #
|
|
|
92 # # At this point, ACC is true, and the thread has the condition
|
|
|
93 # # locked.
|
|
|
94 # # So code here can safely muck with the shared state that
|
|
|
95 # # went into evaluating the ACC -- if it wants to.
|
|
|
96 # # When done mucking with the shared state, do
|
|
|
97 # condition.release()
|
|
|
98 #
|
|
|
99 # B) Threads that are mucking with shared state that may affect the
|
|
|
100 # ACC do:
|
|
|
101 #
|
|
|
102 # condition.acquire()
|
|
|
103 # # muck with shared state
|
|
|
104 # condition.release()
|
|
|
105 # if it's possible that ACC is true now:
|
|
|
106 # condition.signal() # or .broadcast()
|
|
|
107 #
|
|
|
108 # Note: You may prefer to put the "if" clause before the release().
|
|
|
109 # That's fine, but do note that anyone waiting on the signal will
|
|
|
110 # stay blocked until the release() is done (since acquiring the
|
|
|
111 # condition is part of what .wait() does before it returns).
|
|
|
112 #
|
|
|
113 # TRICK OF THE TRADE
|
|
|
114 #
|
|
|
115 # With simpler forms of conditions, it can be impossible to know when
|
|
|
116 # a thread that's supposed to do a .wait has actually done it. But
|
|
|
117 # because this form of condition releases a lock as _part_ of doing a
|
|
|
118 # wait, the state of that lock can be used to guarantee it.
|
|
|
119 #
|
|
|
120 # E.g., suppose thread A spawns thread B and later wants to wait for B to
|
|
|
121 # complete:
|
|
|
122 #
|
|
|
123 # In A: In B:
|
|
|
124 #
|
|
|
125 # B_done = condition() ... do work ...
|
|
|
126 # B_done.acquire() B_done.acquire(); B_done.release()
|
|
|
127 # spawn B B_done.signal()
|
|
|
128 # ... some time later ... ... and B exits ...
|
|
|
129 # B_done.wait()
|
|
|
130 #
|
|
|
131 # Because B_done was in the acquire'd state at the time B was spawned,
|
|
|
132 # B's attempt to acquire B_done can't succeed until A has done its
|
|
|
133 # B_done.wait() (which releases B_done). So B's B_done.signal() is
|
|
|
134 # guaranteed to be seen by the .wait(). Without the lock trick, B
|
|
|
135 # may signal before A .waits, and then A would wait forever.
|
|
|
136 #
|
|
|
137 # BARRIERS
|
|
|
138 #
|
|
|
139 # A barrier object is created via
|
|
|
140 # import this_module
|
|
|
141 # your_barrier = this_module.barrier(num_threads)
|
|
|
142 #
|
|
|
143 # Methods:
|
|
|
144 # .enter()
|
|
|
145 # the thread blocks until num_threads threads in all have done
|
|
|
146 # .enter(). Then the num_threads threads that .enter'ed resume,
|
|
|
147 # and the barrier resets to capture the next num_threads threads
|
|
|
148 # that .enter it.
|
|
|
149 #
|
|
|
150 # EVENTS
|
|
|
151 #
|
|
|
152 # An event object is created via
|
|
|
153 # import this_module
|
|
|
154 # your_event = this_module.event()
|
|
|
155 #
|
|
|
156 # An event has two states, `posted' and `cleared'. An event is
|
|
|
157 # created in the cleared state.
|
|
|
158 #
|
|
|
159 # Methods:
|
|
|
160 #
|
|
|
161 # .post()
|
|
|
162 # Put the event in the posted state, and resume all threads
|
|
|
163 # .wait'ing on the event (if any).
|
|
|
164 #
|
|
|
165 # .clear()
|
|
|
166 # Put the event in the cleared state.
|
|
|
167 #
|
|
|
168 # .is_posted()
|
|
|
169 # Returns 0 if the event is in the cleared state, or 1 if the event
|
|
|
170 # is in the posted state.
|
|
|
171 #
|
|
|
172 # .wait()
|
|
|
173 # If the event is in the posted state, returns immediately.
|
|
|
174 # If the event is in the cleared state, blocks the calling thread
|
|
|
175 # until the event is .post'ed by another thread.
|
|
|
176 #
|
|
|
177 # Note that an event, once posted, remains posted until explicitly
|
|
|
178 # cleared. Relative to conditions, this is both the strength & weakness
|
|
|
179 # of events. It's a strength because the .post'ing thread doesn't have to
|
|
|
180 # worry about whether the threads it's trying to communicate with have
|
|
|
181 # already done a .wait (a condition .signal is seen only by threads that
|
|
|
182 # do a .wait _prior_ to the .signal; a .signal does not persist). But
|
|
|
183 # it's a weakness because .clear'ing an event is error-prone: it's easy
|
|
|
184 # to mistakenly .clear an event before all the threads you intended to
|
|
|
185 # see the event get around to .wait'ing on it. But so long as you don't
|
|
|
186 # need to .clear an event, events are easy to use safely.
|
|
|
187 #
|
|
|
188 # SEMAPHORES
|
|
|
189 #
|
|
|
190 # A semaphore object is created via
|
|
|
191 # import this_module
|
|
|
192 # your_semaphore = this_module.semaphore(count=1)
|
|
|
193 #
|
|
|
194 # A semaphore has an integer count associated with it. The initial value
|
|
|
195 # of the count is specified by the optional argument (which defaults to
|
|
|
196 # 1) passed to the semaphore constructor.
|
|
|
197 #
|
|
|
198 # Methods:
|
|
|
199 #
|
|
|
200 # .p()
|
|
|
201 # If the semaphore's count is greater than 0, decrements the count
|
|
|
202 # by 1 and returns.
|
|
|
203 # Else if the semaphore's count is 0, blocks the calling thread
|
|
|
204 # until a subsequent .v() increases the count. When that happens,
|
|
|
205 # the count will be decremented by 1 and the calling thread resumed.
|
|
|
206 #
|
|
|
207 # .v()
|
|
|
208 # Increments the semaphore's count by 1, and wakes up a thread (if
|
|
|
209 # any) blocked by a .p(). It's an (detected) error for a .v() to
|
|
|
210 # increase the semaphore's count to a value larger than the initial
|
|
|
211 # count.
|
|
|
212
|
|
|
213 import thread
|
|
|
214
|
|
|
215 class condition:
|
|
|
216 def __init__(self):
|
|
|
217 # the lock actually used by .acquire() and .release()
|
|
|
218 self.mutex = thread.allocate_lock()
|
|
|
219
|
|
|
220 # lock used to block threads until a signal
|
|
|
221 self.checkout = thread.allocate_lock()
|
|
|
222 self.checkout.acquire()
|
|
|
223
|
|
|
224 # internal critical-section lock, & the data it protects
|
|
|
225 self.idlock = thread.allocate_lock()
|
|
|
226 self.id = 0
|
|
|
227 self.waiting = 0 # num waiters subject to current release
|
|
|
228 self.pending = 0 # num waiters awaiting next signal
|
|
|
229 self.torelease = 0 # num waiters to release
|
|
|
230 self.releasing = 0 # 1 iff release is in progress
|
|
|
231
|
|
|
232 def acquire(self):
|
|
|
233 self.mutex.acquire()
|
|
|
234
|
|
|
235 def release(self):
|
|
|
236 self.mutex.release()
|
|
|
237
|
|
|
238 def wait(self):
|
|
|
239 mutex, checkout, idlock = self.mutex, self.checkout, self.idlock
|
|
|
240 if not mutex.locked():
|
|
|
241 raise ValueError, \
|
|
|
242 "condition must be .acquire'd when .wait() invoked"
|
|
|
243
|
|
|
244 idlock.acquire()
|
|
|
245 myid = self.id
|
|
|
246 self.pending = self.pending + 1
|
|
|
247 idlock.release()
|
|
|
248
|
|
|
249 mutex.release()
|
|
|
250
|
|
|
251 while 1:
|
|
|
252 checkout.acquire(); idlock.acquire()
|
|
|
253 if myid < self.id:
|
|
|
254 break
|
|
|
255 checkout.release(); idlock.release()
|
|
|
256
|
|
|
257 self.waiting = self.waiting - 1
|
|
|
258 self.torelease = self.torelease - 1
|
|
|
259 if self.torelease:
|
|
|
260 checkout.release()
|
|
|
261 else:
|
|
|
262 self.releasing = 0
|
|
|
263 if self.waiting == self.pending == 0:
|
|
|
264 self.id = 0
|
|
|
265 idlock.release()
|
|
|
266 mutex.acquire()
|
|
|
267
|
|
|
268 def signal(self):
|
|
|
269 self.broadcast(1)
|
|
|
270
|
|
|
271 def broadcast(self, num = -1):
|
|
|
272 if num < -1:
|
|
|
273 raise ValueError, '.broadcast called with num ' + `num`
|
|
|
274 if num == 0:
|
|
|
275 return
|
|
|
276 self.idlock.acquire()
|
|
|
277 if self.pending:
|
|
|
278 self.waiting = self.waiting + self.pending
|
|
|
279 self.pending = 0
|
|
|
280 self.id = self.id + 1
|
|
|
281 if num == -1:
|
|
|
282 self.torelease = self.waiting
|
|
|
283 else:
|
|
|
284 self.torelease = min( self.waiting,
|
|
|
285 self.torelease + num )
|
|
|
286 if self.torelease and not self.releasing:
|
|
|
287 self.releasing = 1
|
|
|
288 self.checkout.release()
|
|
|
289 self.idlock.release()
|
|
|
290
|
|
|
291 class barrier:
|
|
|
292 def __init__(self, n):
|
|
|
293 self.n = n
|
|
|
294 self.togo = n
|
|
|
295 self.full = condition()
|
|
|
296
|
|
|
297 def enter(self):
|
|
|
298 full = self.full
|
|
|
299 full.acquire()
|
|
|
300 self.togo = self.togo - 1
|
|
|
301 if self.togo:
|
|
|
302 full.wait()
|
|
|
303 else:
|
|
|
304 self.togo = self.n
|
|
|
305 full.broadcast()
|
|
|
306 full.release()
|
|
|
307
|
|
|
308 class event:
|
|
|
309 def __init__(self):
|
|
|
310 self.state = 0
|
|
|
311 self.posted = condition()
|
|
|
312
|
|
|
313 def post(self):
|
|
|
314 self.posted.acquire()
|
|
|
315 self.state = 1
|
|
|
316 self.posted.broadcast()
|
|
|
317 self.posted.release()
|
|
|
318
|
|
|
319 def clear(self):
|
|
|
320 self.posted.acquire()
|
|
|
321 self.state = 0
|
|
|
322 self.posted.release()
|
|
|
323
|
|
|
324 def is_posted(self):
|
|
|
325 self.posted.acquire()
|
|
|
326 answer = self.state
|
|
|
327 self.posted.release()
|
|
|
328 return answer
|
|
|
329
|
|
|
330 def wait(self):
|
|
|
331 self.posted.acquire()
|
|
|
332 if not self.state:
|
|
|
333 self.posted.wait()
|
|
|
334 self.posted.release()
|
|
|
335
|
|
|
336 class semaphore:
|
|
|
337 def __init__(self, count=1):
|
|
|
338 if count <= 0:
|
|
|
339 raise ValueError, 'semaphore count %d; must be >= 1' % count
|
|
|
340 self.count = count
|
|
|
341 self.maxcount = count
|
|
|
342 self.nonzero = condition()
|
|
|
343
|
|
|
344 def p(self):
|
|
|
345 self.nonzero.acquire()
|
|
|
346 while self.count == 0:
|
|
|
347 self.nonzero.wait()
|
|
|
348 self.count = self.count - 1
|
|
|
349 self.nonzero.release()
|
|
|
350
|
|
|
351 def v(self):
|
|
|
352 self.nonzero.acquire()
|
|
|
353 if self.count == self.maxcount:
|
|
|
354 raise ValueError, '.v() tried to raise semaphore count above ' \
|
|
|
355 'initial value ' + `maxcount`
|
|
|
356 self.count = self.count + 1
|
|
|
357 self.nonzero.signal()
|
|
|
358 self.nonzero.release()
|
|
|
359
|
|
|
360 # The rest of the file is a test case, that runs a number of parallelized
|
|
|
361 # quicksorts in parallel. If it works, you'll get about 600 lines of
|
|
|
362 # tracing output, with a line like
|
|
|
363 # test passed! 209 threads created in all
|
|
|
364 # as the last line. The content and order of preceding lines will
|
|
|
365 # vary across runs.
|
|
|
366
|
|
|
367 def _new_thread(func, *args):
|
|
|
368 global TID
|
|
|
369 tid.acquire(); id = TID = TID+1; tid.release()
|
|
|
370 io.acquire(); alive.append(id); \
|
|
|
371 print 'starting thread', id, '--', len(alive), 'alive'; \
|
|
|
372 io.release()
|
|
|
373 thread.start_new_thread( func, (id,) + args )
|
|
|
374
|
|
|
375 def _qsort(tid, a, l, r, finished):
|
|
|
376 # sort a[l:r]; post finished when done
|
|
|
377 io.acquire(); print 'thread', tid, 'qsort', l, r; io.release()
|
|
|
378 if r-l > 1:
|
|
|
379 pivot = a[l]
|
|
|
380 j = l+1 # make a[l:j] <= pivot, and a[j:r] > pivot
|
|
|
381 for i in range(j, r):
|
|
|
382 if a[i] <= pivot:
|
|
|
383 a[j], a[i] = a[i], a[j]
|
|
|
384 j = j + 1
|
|
|
385 a[l], a[j-1] = a[j-1], pivot
|
|
|
386
|
|
|
387 l_subarray_sorted = event()
|
|
|
388 r_subarray_sorted = event()
|
|
|
389 _new_thread(_qsort, a, l, j-1, l_subarray_sorted)
|
|
|
390 _new_thread(_qsort, a, j, r, r_subarray_sorted)
|
|
|
391 l_subarray_sorted.wait()
|
|
|
392 r_subarray_sorted.wait()
|
|
|
393
|
|
|
394 io.acquire(); print 'thread', tid, 'qsort done'; \
|
|
|
395 alive.remove(tid); io.release()
|
|
|
396 finished.post()
|
|
|
397
|
|
|
398 def _randarray(tid, a, finished):
|
|
|
399 io.acquire(); print 'thread', tid, 'randomizing array'; \
|
|
|
400 io.release()
|
|
|
401 for i in range(1, len(a)):
|
|
|
402 wh.acquire(); j = randint(0,i); wh.release()
|
|
|
403 a[i], a[j] = a[j], a[i]
|
|
|
404 io.acquire(); print 'thread', tid, 'randomizing done'; \
|
|
|
405 alive.remove(tid); io.release()
|
|
|
406 finished.post()
|
|
|
407
|
|
|
408 def _check_sort(a):
|
|
|
409 if a != range(len(a)):
|
|
|
410 raise ValueError, ('a not sorted', a)
|
|
|
411
|
|
|
412 def _run_one_sort(tid, a, bar, done):
|
|
|
413 # randomize a, and quicksort it
|
|
|
414 # for variety, all the threads running this enter a barrier
|
|
|
415 # at the end, and post `done' after the barrier exits
|
|
|
416 io.acquire(); print 'thread', tid, 'randomizing', a; \
|
|
|
417 io.release()
|
|
|
418 finished = event()
|
|
|
419 _new_thread(_randarray, a, finished)
|
|
|
420 finished.wait()
|
|
|
421
|
|
|
422 io.acquire(); print 'thread', tid, 'sorting', a; io.release()
|
|
|
423 finished.clear()
|
|
|
424 _new_thread(_qsort, a, 0, len(a), finished)
|
|
|
425 finished.wait()
|
|
|
426 _check_sort(a)
|
|
|
427
|
|
|
428 io.acquire(); print 'thread', tid, 'entering barrier'; \
|
|
|
429 io.release()
|
|
|
430 bar.enter()
|
|
|
431 io.acquire(); print 'thread', tid, 'leaving barrier'; \
|
|
|
432 io.release()
|
|
|
433 io.acquire(); alive.remove(tid); io.release()
|
|
|
434 bar.enter() # make sure they've all removed themselves from alive
|
|
|
435 ## before 'done' is posted
|
|
|
436 bar.enter() # just to be cruel
|
|
|
437 done.post()
|
|
|
438
|
|
|
439 def test():
|
|
|
440 global TID, tid, io, wh, randint, alive
|
|
|
441 import whrandom
|
|
|
442 randint = whrandom.randint
|
|
|
443
|
|
|
444 TID = 0 # thread ID (1, 2, ...)
|
|
|
445 tid = thread.allocate_lock() # for changing TID
|
|
|
446 io = thread.allocate_lock() # for printing, and 'alive'
|
|
|
447 wh = thread.allocate_lock() # for calls to whrandom
|
|
|
448 alive = [] # IDs of active threads
|
|
|
449
|
|
|
450 NSORTS = 5
|
|
|
451 arrays = []
|
|
|
452 for i in range(NSORTS):
|
|
|
453 arrays.append( range( (i+1)*10 ) )
|
|
|
454
|
|
|
455 bar = barrier(NSORTS)
|
|
|
456 finished = event()
|
|
|
457 for i in range(NSORTS):
|
|
|
458 _new_thread(_run_one_sort, arrays[i], bar, finished)
|
|
|
459 finished.wait()
|
|
|
460
|
|
|
461 print 'all threads done, and checking results ...'
|
|
|
462 if alive:
|
|
|
463 raise ValueError, ('threads still alive at end', alive)
|
|
|
464 for i in range(NSORTS):
|
|
|
465 a = arrays[i]
|
|
|
466 if len(a) != (i+1)*10:
|
|
|
467 raise ValueError, ('length of array', i, 'screwed up')
|
|
|
468 _check_sort(a)
|
|
|
469
|
|
|
470 print 'test passed!', TID, 'threads created in all'
|
|
|
471
|
|
|
472 if __name__ == '__main__':
|
|
|
473 test()
|
|
|
474
|
|
|
475 # end of module
|
