Threading and Conditions

[Jonathan Giddy]

Ken Seehof asks
>Both notify() and wait() require a thread lock.  By my understanding,
>this
>means that a notify cannot occur while a wait is in progress (i.e. until
>we are
>done waiting for the notify), which means I'll be waiting a very long
>time :-).
>What's my confusion?

The wait() releases the lock on entry, and reacquires it on exit.  If the
program had to release the lock before the wait, a race condition would
exist if a notify() occurred between the release() and the wait().

[Ken then produces some real example code]

Modified code is below.  But first, what I changed:

I used a Python list as a queue for the items.  Even though you sleep for 
a second between putting items on the queue, it is possible (unlikely in 
this case, but true in general) for more than one item to be "made" before 
one item is "got".

You also have a race condition in your code.  After the client tests
"while not an_item_is_available()", the server thread can add an item and 
notify().  The client code now goes into a wait() even though there is an 
item waiting and an indeterminate time until the next notify() will wake it 
up.  The solution is to move this test inside the locked region.

Also, a subtlety that exists in the documentation sample that you changed
was the repeated test after wait() wakes up.  It's a good idea to assume 
that wait() can be woken up spuriously, since that's what the Pthread
standard says.  For this reason, you need to repeat the test for existence
of an item before trying to remove it.

Finally, I also put the client "while not done" test inside the locked 
region.  Otherwise, if the server runs really fast (i.e. if you remove the 
sleep()), the variable done is set before the client has a chance to read
the queue, and the client exits.

from threading import *
from time import sleep
import sys

def an_item_is_available():
    return len(item) > 0

def make_an_item_available():
    global item, done
    if len(x)>0:
        item.append(x[0])
        del x[0]
    else:
        item.append('end')
        done = 1

def get_an_available_item():
    print item[0]
    del item[0]

def server():
    print 'starting server'
    sys.stdout.flush()
    while not done:
        for i in range(1000):
            sleep(0.001)

        # Produce one item
        cv.acquire()
        make_an_item_available()
        cv.notify()
        cv.release()

def client():
    print 'starting client'
    sys.stdout.flush()
    cv.acquire()
    while not done or an_item_is_available():
        while not an_item_is_available():
            cv.wait()   # releases lock, allowing server thread to run
        get_an_available_item()
    cv.release()

def test():
    global cv, td, x, item, done
    x = ['alpha', 'beta', 'gamma', 'delta']
    item = []
    done = 0

    td = Thread(target = server)
    td.setDaemon(1)
    cv = Condition()
    td.start()
    client()
-- 
Jonathan Giddy
Distributed Systems Technology Centre
Level 5, C Block, Monash University
Caulfield, Victoria 3145 Australia
ph:+61 3 9903 1966   fax:+61 3 9903 2863
jon at dstc.edu.au