[boost] win32 mutex and condition variable algorithms (was: [threads] Permission given to change to Boost.License)

From: Anthony Williams (anthony_w.geo_at_[hidden])
Date: 2006-09-19 10:38:35

"Peter Dimov" <pdimov_at_[hidden]> writes:

> Inventing mutex and condition variable algorithms is a slight cause of 
> concern, I have to admit. The easiest way to get a free review of the 
> algorithms is to post the code on comp.programming.threads and hope that 
> someone of the experts there has enough time to look at them. CVs in 
> particular have proven to be quite hard to implement correctly, and 
> Alexander Terekhov's "8a" algorithm has emerged as a de-facto standard. 

The mutex is simple, and you can see the code in
boost/thread/win32/basic_mutex.hpp on the thread_rewrite branch.

The algorithm is:

init():
   active_count=0;
   no semaphore

lock():
   atomic increment active_count
   if new active_count ==1, that's us, so we've got the lock
   else
        get semaphore, and wait
        now we've got the lock

unlock():
   atomic decrement active_count
   if new active_count >0, then other threads are waiting, 
       so release semaphore.

locked():
   return active_count>0

get_semaphore():
   if there's already a semaphore associated with this mutex, return that
   else
       create new semaphore.
       use atomic compare-and-swap to make this the associated semaphore if
           none
       if another thread beat us to it, and already set a semaphore, destroy
           new one, and return already-set one
       else return the new semaphore

The condition variable algorithm is actually a variant of Alexander Terekhov's
algorithm, and uses the same "gate" idea. However, my implementation uses
QueueUserAPC to wake up the waiting threads, rather than have them wait on an
event. This gets round what I think is a deficiency in Alexander's algorithm,
which is that once a notify operation is underway, no new threads can wait on
the condition until all notified threads have actually woken up, and the last
one has "opened the gate". By using QueueUserAPC, we can notify the threads to
wake up, and open the gate, without having to wait for all the notified
threads to resume. The flipside is that the first thread to wait is the first
to be notified. Whether this is a good or bad thing is open to debate.

The code is in boost/thread/win32/condition.hpp on the thread_rewrite branch.

The algorithm is as follows:

waiting_list_node:
    waiting_list_node* next, prev
    HANDLE thread_handle
    bool notified

waiting_list: doubly-linked list of waiting_list_node
gate: mutex

init():
    waiting_list.next=waiting_list.prev=&waiting_list
    init mutex

wait(external_mutex, timeout):
    create a new waiting_list_node
    new_node.thread_handle=thread handle for this thread
    new_node.prev=&waiting_list
    lock(gate)
    new_node.next=waiting_list.next
    waiting_list.next=&new_node
    new_node.next->prev=&new_node
    unlock(external_mutex)
    unlock(gate)

    // Any APC will break the sleep, so keep sleeping until we've been
    // notified, or we've timed out
    while(!atomic_read(new_node.notified)
        && SleepEx(milliseconds_until(timeout), true)==WAIT_IO_COMPLETION);

    lock(gate)
    unlink(new_node)
    unlock(gate)
    lock(external_mutex)
    return new_node.notified // did we timeout, or were we notified?

unlink(node)
    // unlink the node from the list
    node.next->prev=new_node.prev
    node.prev->next=new_node.next
    node.next=node.prev=&node

notify_and_unlink_entry(node)
    atomic_set(node->notified,true)
    unlink(node)
    // wake the node's thread by queueing an APC
    // the APC func doesn't have to do anything to wake the thread
    QueueUserAPC(NOP(),node->thread_handle)

notify_one()
    lock(gate)
    if(waiting_list.prev==&waiting_list) do nothing
    else
        notify_and_unlink_entry(waiting_list.prev)
    unlock(gate)

notify_all()
    create a waiting_list_node for new_list
    lock(gate)
    // transfer the existing list over to new_list
    new_list.prev=waiting_list.prev
    new_list.next=waiting_list.next
    new_list.next->prev=&new_list
    new_list.prev->next=&new_list
    // the waiting_list is now empty
    waiting_list.next=waiting_list.prev=&waiting_list
    unlock(gate) // noone has to wait for us any more

    while(new_list.prev!=&new_list) // loop until the captured list is empty
        notify_and_unlink_entry(new_list.prev)

I'll post the algorithms over at comp.programming.threads for comment.

Anthony
-- 
Anthony Williams
Software Developer
Just Software Solutions Ltd
http://www.justsoftwaresolutions.co.uk