Multithreaded FIFO Gate (Python recipe) by Anand

ActiveState Code (http://code.activestate.com/recipes/576418/)

I don't understand some things in revision 2 of this code:

• position (I guess it should be renamed/retyped, as "is_open"/boolean for example) is never reset to 1 imho. Besides, there could be a race condition if position were reset to 1 in the open() at the location "position=0" is now (i'd say it should be before the sem.release)

• race condition in enter: test for "position == 0" not atomic with all your synchros.

• count is never decreased

• If you protect the count incrementation with a mutex, you should also protect it in getcount(), it would be cleaner

• I really really doubt count would be accurate as it is now, even if it were decreased, because it's based on a synchro completely isolated from the other sets of synchros. So I guess there would be transitory periods when it would report an inaccurate count

One thing about which I need to think more: I have the intuition that maybe 2 levels of lock (mutex / binary semaphore) is not needed here. But, well, I am not sure at all about that: for now I don't have any idea to suggest.

I thought a little about this syncho. Here are some additional comments on rev 2 of the code.

First of all, I have to remove the last statement in my previous comment: imho a 2-level synchro seems inevitable. Worse, I think that enforcing the Fifo ordering requires to have one lock for each thread waiting... this, if we consider we don't know anything about the scheduler (which is my case). If we can make any Fifo assumption about the scheduler, then, I still do think that a 2-level synchro is needed, as you implemented, but that there is no need to have one lock per queued thread (a simple condition+binary semaphore should be enough). So, please reconsider the last 2 sentences of my previous comment.

Second and more importantly, to me it appears the protocol for this gate synchro is not really rigourous as it is now. Waking up the threads in Fifo order does not tell anything about the order in which they will actually resume on the CPU, unless we use appropriate mutual exclusion when the threads start their execution: I didn't see any guarantee that the scheduler is Fifo. That's probably why you added the part you wanted "Fifo" inside the synchro itself (inside the enter(): you put your call to sleep() there, to show it's Fifo), but nothing guarantees anything after the enter(). So imho, you might see some cases where the order recorded in your exitlog is not the one effectively followed by the open() function.

That's why I think a synchro with an explicit enter/leave protocol would be more rigourous here: at least you can guarantee some exclusion inside the enter/leave which can enforce some Fifo ordering wrt the order in which the threads will be woken up by open().

This also means that, after the open, the threads will be serialized (mutual exclusion)... and raises the question "what to do if we have still threads in enter/leave, the gate is open, and new threads call enter...": do these new threads have to wait for the threads that are still in enter() ? Or can they execute right away, thus by-passing the Fifo order ?

Besides this, I still think that the implementation is broken as it is. For example calling open() doesn't re-open the gate: it just allows the blocked threads to be woken up. But any other threads coming after an open() will be queued as if the gate was still closed. This was not the case before close() was called the first time (in that case, the threads were not blocked at all).

Proposed implementation of the enter/leave protocol:

from collections import deque
...
class ThreadGate(object):
    """The unprotected_* methods are expected to be called with lock held. Use the non-unprotected_* variant otherwise"""

    def __init__(self, lock = None):
        if lock is None: lock = threading.Lock()
        self.lock       = lock
        self.owner      = None # threading ID
        self.waiters    = deque() # list of locks
        self.per_thread = threading.local()
        self.open       = self._synchronized(self.unprotected_open)
        self.close      = self._synchronized(self.unprotected_close)
        self.leave      = self._synchronized(self.unprotected_leave)
        self.count      = self._synchronized(self.unprotected_count)
        self.is_open    = self._synchronized(self.unprotected_is_open)

    def _synchronized(self, f):
        def new_f(*args, **kw):
            self.lock.acquire()
            try:
                return f(*args, **kw)
            finally:
                self.lock.release()
        return new_f

    def unprotected_close(self):
        if self.owner is not None:
            raise ThreadGateException,"trying to close an already closed gate"
        self.owner = threading.currentThread()

    def unprotected_open(self):
        if self.owner is None:
            raise ThreadGateException,"trying to open an already opened gate"
        if threading.currentThread() != self.owner:
            raise ThreadGateException,"not owner, cannot open gate"
        self.owner = None
        self.__wakeup_head()

    def enter(self):
        """Lock expected NOT to be acquired, return True if was blocked"""
        self.lock.acquire()
        try:
            # The gate is not closed: don't block only if no others are blocked
            self.per_thread.is_blocked = (self.owner is not None) # and len(self.waiters) == 0
            if not self.per_thread.is_blocked: return False

            # Create our lock
            waiter = threading.Lock()
            waiter.acquire()
            self.waiters.append(waiter)

        finally:
            self.lock.release()

        # Go to sleep, waiting for some thread to wake-us up
        waiter.acquire()
        return True # Was blocked

    def unprotected_leave(self):
        if self.per_thread.is_blocked: self.__wakeup_head()

    def __wakeup_head(self):
        """Lock expected to be held by caller"""
        if not self.waiters: return # Nobody to wake up
        self.waiters.popleft().release()

    def unprotected_count(self):
        return len(self.waiters)

    def unprotected_is_open(self):
        return (self.owner is None)

I didn't have enough space to add comments. To the question "what to do if we have still threads in enter/leave, the gate is open, and new threads call enter?"... this implementation answers "they execute right away". If you want to have them queued after those that are still exiting, you can uncomment the:

and len(self.waiters) == 0

Following the answer to the previous question, it means that threads entering after the open() will be woken up in an arbitrary order. For that matter, the check in main() will have to be updated to take into account those threads that have been blocked by close() (hence the return value for enter()):

def test():
    """ Test code """
    NTHREADS = 255

    import random

    print 'test()...'

    enterlog = []
    exitlog = []
    gate = ThreadGate()

    random.seed()

    class SampleThread(threading.Thread):
        """ Sample thread class for gate demo """

        def __init__(self, index):
            self.index = index
            threading.Thread.__init__(self, name = 'Thread %d' % self.index)

        def run(self):

            # Sleep randomly
            time.sleep(random.choice(range(1,10)))
            # Mark entry to gate
            ct = threading.currentThread()

            print 'Thread %s - Entering Gate...' % ct.getName()
            if not gate.is_open():
                # Might catch more threads than those really blocked
                enterlog.append(self.index)            
            was_blocked = gate.enter()

            time.sleep(.1)

            # Mark exit out of gate
            print 'Thread %s - Exiting Gate (was blocked: %s)' \
                  % (ct.getName(), was_blocked)
            if was_blocked:
                exitlog.append(self.index)

            gate.leave()

    # Close the gate
    print 'Closing gate...'
    gate.close()

    print 'Starting threads...'
    # Create 10 threads
    threads = []
    for x in range(NTHREADS):
        threads.append(SampleThread(x))

    # Start threads and join
    for x in range(NTHREADS):
        threads[x].start()        

    # Open gate after sleeping some time
    time.sleep(5)
    print 'Opening gate'
    gate.open()

    # Join with threads
    for x in range(NTHREADS):
        threads[x].join()

    print 'Joined with threads'

    # Exit and entry logs must be same
    print "Entered and probably slept:", enterlog
    print "Exited after having slept:", exitlog
    assert([e for e in enterlog if e in exitlog] == exitlog)

Please note that there is a race condition in this test: the assertion may fail, because enterlog() is filled outside from the lock.

Hi david,

I did not read your comments fully. But here is a version which I think fixes at least one of the problem you mentioned, namely that of managing the thread count. It also adds a size check to the gate which means that the owner can proceed from "close" only after either a fixed number of threads join the gate or a timeout occurs. The test cases illustrate both situations.

Please give your valuable comments.

Btw, I don't think this is any longer a FIFO, because of blocking semantics on the semaphore. Still I have retained the title as "Multithreaded FIFO gate".