mirror of
https://github.com/python/cpython.git
synced 2025-10-31 10:26:02 +00:00
599 lines
21 KiB
Python
599 lines
21 KiB
Python
# 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
|