# 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 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()Lines 1–500 of 603. Load more
# 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 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()