# 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()

    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()