diff options
Diffstat (limited to 'Demo/threads/sync.py')
| -rw-r--r-- | Demo/threads/sync.py | 599 |
1 files changed, 0 insertions, 599 deletions
diff --git a/Demo/threads/sync.py b/Demo/threads/sync.py deleted file mode 100644 index 90fff2e..0000000 --- a/Demo/threads/sync.py +++ /dev/null @@ -1,599 +0,0 @@ -# Defines classes that provide synchronization objects. Note that use of -# this module requires that your Python support threads. -# -# condition(lock=None) # a POSIX-like condition-variable object -# barrier(n) # an n-thread barrier -# event() # an event object -# semaphore(n=1) # a semaphore object, with initial count n -# mrsw() # a multiple-reader single-writer lock -# -# CONDITIONS -# -# A condition object is created via -# import this_module -# your_condition_object = this_module.condition(lock=None) -# -# As explained below, a condition object has a lock associated with it, -# used in the protocol to protect condition data. You can specify a -# lock to use in the constructor, else the constructor will allocate -# an anonymous lock for you. Specifying a lock explicitly can be useful -# when more than one condition keys off the same set of shared data. -# -# Methods: -# .acquire() -# acquire the lock associated with the condition -# .release() -# release the lock associated with the condition -# .wait() -# block the thread until such time as some other thread does a -# .signal or .broadcast on the same condition, and release the -# lock associated with the condition. The lock associated with -# the condition MUST be in the acquired state at the time -# .wait is invoked. -# .signal() -# wake up exactly one thread (if any) that previously did a .wait -# on the condition; that thread will awaken with the lock associated -# with the condition in the acquired state. If no threads are -# .wait'ing, this is a nop. If more than one thread is .wait'ing on -# the condition, any of them may be awakened. -# .broadcast() -# wake up all threads (if any) that are .wait'ing on the condition; -# the threads are woken up serially, each with the lock in the -# acquired state, so should .release() as soon as possible. If no -# threads are .wait'ing, this is a nop. -# -# Note that if a thread does a .wait *while* a signal/broadcast is -# in progress, it's guaranteeed to block until a subsequent -# signal/broadcast. -# -# Secret feature: `broadcast' actually takes an integer argument, -# and will wake up exactly that many waiting threads (or the total -# number waiting, if that's less). Use of this is dubious, though, -# and probably won't be supported if this form of condition is -# reimplemented in C. -# -# DIFFERENCES FROM POSIX -# -# + A separate mutex is not needed to guard condition data. Instead, a -# condition object can (must) be .acquire'ed and .release'ed directly. -# This eliminates a common error in using POSIX conditions. -# -# + Because of implementation difficulties, a POSIX `signal' wakes up -# _at least_ one .wait'ing thread. Race conditions make it difficult -# to stop that. This implementation guarantees to wake up only one, -# but you probably shouldn't rely on that. -# -# PROTOCOL -# -# Condition objects are used to block threads until "some condition" is -# true. E.g., a thread may wish to wait until a producer pumps out data -# for it to consume, or a server may wish to wait until someone requests -# its services, or perhaps a whole bunch of threads want to wait until a -# preceding pass over the data is complete. Early models for conditions -# relied on some other thread figuring out when a blocked thread's -# condition was true, and made the other thread responsible both for -# waking up the blocked thread and guaranteeing that it woke up with all -# data in a correct state. This proved to be very delicate in practice, -# and gave conditions a bad name in some circles. -# -# The POSIX model addresses these problems by making a thread responsible -# for ensuring that its own state is correct when it wakes, and relies -# on a rigid protocol to make this easy; so long as you stick to the -# protocol, POSIX conditions are easy to "get right": -# -# A) The thread that's waiting for some arbitrarily-complex condition -# (ACC) to become true does: -# -# condition.acquire() -# while not (code to evaluate the ACC): -# condition.wait() -# # That blocks the thread, *and* releases the lock. When a -# # condition.signal() happens, it will wake up some thread that -# # did a .wait, *and* acquire the lock again before .wait -# # returns. -# # -# # Because the lock is acquired at this point, the state used -# # in evaluating the ACC is frozen, so it's safe to go back & -# # reevaluate the ACC. -# -# # At this point, ACC is true, and the thread has the condition -# # locked. -# # So code here can safely muck with the shared state that -# # went into evaluating the ACC -- if it wants to. -# # When done mucking with the shared state, do -# condition.release() -# -# B) Threads that are mucking with shared state that may affect the -# ACC do: -# -# condition.acquire() -# # muck with shared state -# condition.release() -# if it's possible that ACC is true now: -# condition.signal() # or .broadcast() -# -# Note: You may prefer to put the "if" clause before the release(). -# That's fine, but do note that anyone waiting on the signal will -# stay blocked until the release() is done (since acquiring the -# condition is part of what .wait() does before it returns). -# -# TRICK OF THE TRADE -# -# With simpler forms of conditions, it can be impossible to know when -# a thread that's supposed to do a .wait has actually done it. But -# because this form of condition releases a lock as _part_ of doing a -# wait, the state of that lock can be used to guarantee it. -# -# E.g., suppose thread A spawns thread B and later wants to wait for B to -# complete: -# -# In A: In B: -# -# B_done = condition() ... do work ... -# B_done.acquire() B_done.acquire(); B_done.release() -# spawn B B_done.signal() -# ... some time later ... ... and B exits ... -# B_done.wait() -# -# Because B_done was in the acquire'd state at the time B was spawned, -# B's attempt to acquire B_done can't succeed until A has done its -# B_done.wait() (which releases B_done). So B's B_done.signal() is -# guaranteed to be seen by the .wait(). Without the lock trick, B -# may signal before A .waits, and then A would wait forever. -# -# BARRIERS -# -# A barrier object is created via -# import this_module -# your_barrier = this_module.barrier(num_threads) -# -# Methods: -# .enter() -# the thread blocks until num_threads threads in all have done -# .enter(). Then the num_threads threads that .enter'ed resume, -# and the barrier resets to capture the next num_threads threads -# that .enter it. -# -# EVENTS -# -# An event object is created via -# import this_module -# your_event = this_module.event() -# -# An event has two states, `posted' and `cleared'. An event is -# created in the cleared state. -# -# Methods: -# -# .post() -# Put the event in the posted state, and resume all threads -# .wait'ing on the event (if any). -# -# .clear() -# Put the event in the cleared state. -# -# .is_posted() -# Returns 0 if the event is in the cleared state, or 1 if the event -# is in the posted state. -# -# .wait() -# If the event is in the posted state, returns immediately. -# If the event is in the cleared state, blocks the calling thread -# until the event is .post'ed by another thread. -# -# Note that an event, once posted, remains posted until explicitly -# cleared. Relative to conditions, this is both the strength & weakness -# of events. It's a strength because the .post'ing thread doesn't have to -# worry about whether the threads it's trying to communicate with have -# already done a .wait (a condition .signal is seen only by threads that -# do a .wait _prior_ to the .signal; a .signal does not persist). But -# it's a weakness because .clear'ing an event is error-prone: it's easy -# to mistakenly .clear an event before all the threads you intended to -# see the event get around to .wait'ing on it. But so long as you don't -# need to .clear an event, events are easy to use safely. -# -# SEMAPHORES -# -# A semaphore object is created via -# import this_module -# your_semaphore = this_module.semaphore(count=1) -# -# A semaphore has an integer count associated with it. The initial value -# of the count is specified by the optional argument (which defaults to -# 1) passed to the semaphore constructor. -# -# Methods: -# -# .p() -# If the semaphore's count is greater than 0, decrements the count -# by 1 and returns. -# Else if the semaphore's count is 0, blocks the calling thread -# until a subsequent .v() increases the count. When that happens, -# the count will be decremented by 1 and the calling thread resumed. -# -# .v() -# Increments the semaphore's count by 1, and wakes up a thread (if -# any) blocked by a .p(). It's an (detected) error for a .v() to -# increase the semaphore's count to a value larger than the initial -# count. -# -# MULTIPLE-READER SINGLE-WRITER LOCKS -# -# A mrsw lock is created via -# import this_module -# your_mrsw_lock = this_module.mrsw() -# -# This kind of lock is often useful with complex shared data structures. -# The object lets any number of "readers" proceed, so long as no thread -# wishes to "write". When a (one or more) thread declares its intention -# to "write" (e.g., to update a shared structure), all current readers -# are allowed to finish, and then a writer gets exclusive access; all -# other readers & writers are blocked until the current writer completes. -# Finally, if some thread is waiting to write and another is waiting to -# read, the writer takes precedence. -# -# Methods: -# -# .read_in() -# If no thread is writing or waiting to write, returns immediately. -# Else blocks until no thread is writing or waiting to write. So -# long as some thread has completed a .read_in but not a .read_out, -# writers are blocked. -# -# .read_out() -# Use sometime after a .read_in to declare that the thread is done -# reading. When all threads complete reading, a writer can proceed. -# -# .write_in() -# If no thread is writing (has completed a .write_in, but hasn't yet -# done a .write_out) or reading (similarly), returns immediately. -# Else blocks the calling thread, and threads waiting to read, until -# the current writer completes writing or all the current readers -# complete reading; if then more than one thread is waiting to -# write, one of them is allowed to proceed, but which one is not -# specified. -# -# .write_out() -# Use sometime after a .write_in to declare that the thread is done -# writing. Then if some other thread is waiting to write, it's -# allowed to proceed. Else all threads (if any) waiting to read are -# allowed to proceed. -# -# .write_to_read() -# Use instead of a .write_in to declare that the thread is done -# writing but wants to continue reading without other writers -# intervening. If there are other threads waiting to write, they -# are allowed to proceed only if the current thread calls -# .read_out; threads waiting to read are only allowed to proceed -# if there are are no threads waiting to write. (This is a -# weakness of the interface!) - -import _thread as thread - -class condition: - def __init__(self, lock=None): - # the lock actually used by .acquire() and .release() - if lock is None: - self.mutex = thread.allocate_lock() - else: - if hasattr(lock, 'acquire') and \ - hasattr(lock, 'release'): - self.mutex = lock - else: - raise TypeError('condition constructor requires ' \ - 'a lock argument') - - # lock used to block threads until a signal - self.checkout = thread.allocate_lock() - self.checkout.acquire() - - # internal critical-section lock, & the data it protects - self.idlock = thread.allocate_lock() - self.id = 0 - self.waiting = 0 # num waiters subject to current release - self.pending = 0 # num waiters awaiting next signal - self.torelease = 0 # num waiters to release - self.releasing = 0 # 1 iff release is in progress - - def acquire(self): - self.mutex.acquire() - - def release(self): - self.mutex.release() - - def wait(self): - mutex, checkout, idlock = self.mutex, self.checkout, self.idlock - if not mutex.locked(): - raise ValueError("condition must be .acquire'd when .wait() invoked") - - idlock.acquire() - myid = self.id - self.pending = self.pending + 1 - idlock.release() - - mutex.release() - - while 1: - checkout.acquire(); idlock.acquire() - if myid < self.id: - break - checkout.release(); idlock.release() - - self.waiting = self.waiting - 1 - self.torelease = self.torelease - 1 - if self.torelease: - checkout.release() - else: - self.releasing = 0 - if self.waiting == self.pending == 0: - self.id = 0 - idlock.release() - mutex.acquire() - - def signal(self): - self.broadcast(1) - - def broadcast(self, num = -1): - if num < -1: - raise ValueError('.broadcast called with num %r' % (num,)) - if num == 0: - return - self.idlock.acquire() - if self.pending: - self.waiting = self.waiting + self.pending - self.pending = 0 - self.id = self.id + 1 - if num == -1: - self.torelease = self.waiting - else: - self.torelease = min( self.waiting, - self.torelease + num ) - if self.torelease and not self.releasing: - self.releasing = 1 - self.checkout.release() - self.idlock.release() - -class barrier: - def __init__(self, n): - self.n = n - self.togo = n - self.full = condition() - - def enter(self): - full = self.full - full.acquire() - self.togo = self.togo - 1 - if self.togo: - full.wait() - else: - self.togo = self.n - full.broadcast() - full.release() - -class event: - def __init__(self): - self.state = 0 - self.posted = condition() - - def post(self): - self.posted.acquire() - self.state = 1 - self.posted.broadcast() - self.posted.release() - - def clear(self): - self.posted.acquire() - self.state = 0 - self.posted.release() - - def is_posted(self): - self.posted.acquire() - answer = self.state - self.posted.release() - return answer - - def wait(self): - self.posted.acquire() - if not self.state: - self.posted.wait() - self.posted.release() - -class semaphore: - def __init__(self, count=1): - if count <= 0: - raise ValueError('semaphore count %d; must be >= 1' % count) - self.count = count - self.maxcount = count - self.nonzero = condition() - - def p(self): - self.nonzero.acquire() - while self.count == 0: - self.nonzero.wait() - self.count = self.count - 1 - self.nonzero.release() - - def v(self): - self.nonzero.acquire() - if self.count == self.maxcount: - raise ValueError('.v() tried to raise semaphore count above ' \ - 'initial value %r' % self.maxcount) - self.count = self.count + 1 - self.nonzero.signal() - self.nonzero.release() - -class mrsw: - def __init__(self): - # critical-section lock & the data it protects - self.rwOK = thread.allocate_lock() - self.nr = 0 # number readers actively reading (not just waiting) - self.nw = 0 # number writers either waiting to write or writing - self.writing = 0 # 1 iff some thread is writing - - # conditions - self.readOK = condition(self.rwOK) # OK to unblock readers - self.writeOK = condition(self.rwOK) # OK to unblock writers - - def read_in(self): - self.rwOK.acquire() - while self.nw: - self.readOK.wait() - self.nr = self.nr + 1 - self.rwOK.release() - - def read_out(self): - self.rwOK.acquire() - if self.nr <= 0: - raise ValueError('.read_out() invoked without an active reader') - self.nr = self.nr - 1 - if self.nr == 0: - self.writeOK.signal() - self.rwOK.release() - - def write_in(self): - self.rwOK.acquire() - self.nw = self.nw + 1 - while self.writing or self.nr: - self.writeOK.wait() - self.writing = 1 - self.rwOK.release() - - def write_out(self): - self.rwOK.acquire() - if not self.writing: - raise ValueError('.write_out() invoked without an active writer') - self.writing = 0 - self.nw = self.nw - 1 - if self.nw: - self.writeOK.signal() - else: - self.readOK.broadcast() - self.rwOK.release() - - def write_to_read(self): - self.rwOK.acquire() - if not self.writing: - raise ValueError('.write_to_read() invoked without an active writer') - self.writing = 0 - self.nw = self.nw - 1 - self.nr = self.nr + 1 - if not self.nw: - self.readOK.broadcast() - self.rwOK.release() - -# The rest of the file is a test case, that runs a number of parallelized -# quicksorts in parallel. If it works, you'll get about 600 lines of -# tracing output, with a line like -# test passed! 209 threads created in all -# as the last line. The content and order of preceding lines will -# vary across runs. - -def _new_thread(func, *args): - global TID - tid.acquire(); id = TID = TID+1; tid.release() - io.acquire(); alive.append(id); \ - print('starting thread', id, '--', len(alive), 'alive'); \ - io.release() - thread.start_new_thread( func, (id,) + args ) - -def _qsort(tid, a, l, r, finished): - # sort a[l:r]; post finished when done - io.acquire(); print('thread', tid, 'qsort', l, r); io.release() - if r-l > 1: - pivot = a[l] - j = l+1 # make a[l:j] <= pivot, and a[j:r] > pivot - for i in range(j, r): - if a[i] <= pivot: - a[j], a[i] = a[i], a[j] - j = j + 1 - a[l], a[j-1] = a[j-1], pivot - - l_subarray_sorted = event() - r_subarray_sorted = event() - _new_thread(_qsort, a, l, j-1, l_subarray_sorted) - _new_thread(_qsort, a, j, r, r_subarray_sorted) - l_subarray_sorted.wait() - r_subarray_sorted.wait() - - io.acquire(); print('thread', tid, 'qsort done'); \ - alive.remove(tid); io.release() - finished.post() - -def _randarray(tid, a, finished): - io.acquire(); print('thread', tid, 'randomizing array'); \ - io.release() - for i in range(1, len(a)): - wh.acquire(); j = randint(0,i); wh.release() - a[i], a[j] = a[j], a[i] - io.acquire(); print('thread', tid, 'randomizing done'); \ - alive.remove(tid); io.release() - finished.post() - -def _check_sort(a): - if a != range(len(a)): - raise ValueError('a not sorted', a) - -def _run_one_sort(tid, a, bar, done): - # randomize a, and quicksort it - # for variety, all the threads running this enter a barrier - # at the end, and post `done' after the barrier exits - io.acquire(); print('thread', tid, 'randomizing', a); \ - io.release() - finished = event() - _new_thread(_randarray, a, finished) - finished.wait() - - io.acquire(); print('thread', tid, 'sorting', a); io.release() - finished.clear() - _new_thread(_qsort, a, 0, len(a), finished) - finished.wait() - _check_sort(a) - - io.acquire(); print('thread', tid, 'entering barrier'); \ - io.release() - bar.enter() - io.acquire(); print('thread', tid, 'leaving barrier'); \ - io.release() - io.acquire(); alive.remove(tid); io.release() - bar.enter() # make sure they've all removed themselves from alive - ## before 'done' is posted - bar.enter() # just to be cruel - done.post() - -def test(): - global TID, tid, io, wh, randint, alive - import random - randint = random.randint - - TID = 0 # thread ID (1, 2, ...) - tid = thread.allocate_lock() # for changing TID - io = thread.allocate_lock() # for printing, and 'alive' - wh = thread.allocate_lock() # for calls to random - alive = [] # IDs of active threads - - NSORTS = 5 - arrays = [] - for i in range(NSORTS): - arrays.append( range( (i+1)*10 ) ) - - bar = barrier(NSORTS) - finished = event() - for i in range(NSORTS): - _new_thread(_run_one_sort, arrays[i], bar, finished) - finished.wait() - - print('all threads done, and checking results ...') - if alive: - raise ValueError('threads still alive at end', alive) - for i in range(NSORTS): - a = arrays[i] - if len(a) != (i+1)*10: - raise ValueError('length of array', i, 'screwed up') - _check_sort(a) - - print('test passed!', TID, 'threads created in all') - -if __name__ == '__main__': - test() - -# end of module |