libphobos/libdruntime/core/sync/condition.d

/**
 * The condition module provides a primitive for synchronized condition
 * checking.
 *
 * Copyright: Copyright Sean Kelly 2005 - 2009.
 * License:   $(LINK2 http://www.boost.org/LICENSE_1_0.txt, Boost License 1.0)
 * Authors:   Sean Kelly
 * Source:    $(DRUNTIMESRC core/sync/_condition.d)
 */

/*          Copyright Sean Kelly 2005 - 2009.
 * Distributed under the Boost Software License, Version 1.0.
 *    (See accompanying file LICENSE or copy at
 *          http://www.boost.org/LICENSE_1_0.txt)
 */
module core.sync.condition;


public import core.sync.exception;
public import core.sync.mutex;
public import core.time;

import core.exception : AssertError, staticError;


version (Windows)
{
    import core.sync.semaphore;
    import core.sys.windows.basetsd /+: HANDLE+/;
    import core.sys.windows.winbase /+: CloseHandle, CreateSemaphoreA, CRITICAL_SECTION,
        DeleteCriticalSection, EnterCriticalSection, INFINITE, InitializeCriticalSection,
        LeaveCriticalSection, ReleaseSemaphore, WAIT_OBJECT_0, WaitForSingleObject+/;
    import core.sys.windows.windef /+: BOOL, DWORD+/;
    import core.sys.windows.winerror /+: WAIT_TIMEOUT+/;
}
else version (Posix)
{
    import core.stdc.errno : EAGAIN, ETIMEDOUT;
    import core.sync.config;
    import core.sys.posix.pthread : pthread_cond_broadcast, pthread_cond_destroy, pthread_cond_init,
        pthread_cond_signal, pthread_cond_t, pthread_cond_timedwait, pthread_cond_wait;
    import core.sys.posix.time : timespec;
}
else
{
    static assert(false, "Platform not supported");
}


////////////////////////////////////////////////////////////////////////////////
// Condition
//
// void wait();
// void notify();
// void notifyAll();
////////////////////////////////////////////////////////////////////////////////

/**
 * This class represents a condition variable as conceived by C.A.R. Hoare.  As
 * per Mesa type monitors however, "signal" has been replaced with "notify" to
 * indicate that control is not transferred to the waiter when a notification
 * is sent.
 */
class Condition
{
    ////////////////////////////////////////////////////////////////////////////
    // Initialization
    ////////////////////////////////////////////////////////////////////////////

    /**
     * Initializes a condition object which is associated with the supplied
     * mutex object.
     *
     * Params:
     *  m = The mutex with which this condition will be associated.
     *
     * Throws:
     *  SyncError on error.
     */
    this( Mutex m ) nothrow @safe @nogc
    {
        this(m, true);
    }

    /// ditto
    this( shared Mutex m ) shared nothrow @safe @nogc
    {
        import core.atomic : atomicLoad;
        this(atomicLoad(m), true);
    }

    //
    private this(this Q, M)( M m, bool _unused_ ) nothrow @trusted @nogc
        if ((is(Q == Condition) && is(M == Mutex)) ||
            (is(Q == shared Condition) && is(M == shared Mutex)))
    {
        version (Windows)
        {
            static if (is(Q == Condition))
            {
                alias HANDLE_TYPE = void*;
            }
            else
            {
                alias HANDLE_TYPE = shared(void*);
            }
            m_blockLock = cast(HANDLE_TYPE) CreateSemaphoreA( null, 1, 1, null );
            if ( m_blockLock == m_blockLock.init )
                throw staticError!AssertError("Unable to initialize condition", __FILE__, __LINE__);
            scope(failure) CloseHandle( cast(void*) m_blockLock );

            m_blockQueue = cast(HANDLE_TYPE) CreateSemaphoreA( null, 0, int.max, null );
            if ( m_blockQueue == m_blockQueue.init )
                throw staticError!AssertError("Unable to initialize condition", __FILE__, __LINE__);
            scope(failure) CloseHandle( cast(void*) m_blockQueue );

            InitializeCriticalSection( cast(RTL_CRITICAL_SECTION*) &m_unblockLock );
            m_assocMutex = m;
        }
        else version (Posix)
        {
            static if (is(Q == shared))
            {
                import core.atomic : atomicLoad;
                m_assocMutex = atomicLoad(m);
            }
            else
            {
                m_assocMutex = m;
            }
            static if ( is( typeof( imported!"core.sys.posix.pthread".pthread_condattr_setclock ) ) )
            {
                import core.sys.posix.pthread : pthread_condattr_destroy, pthread_condattr_init,
                    pthread_condattr_setclock;
                import core.sys.posix.sys.types : pthread_condattr_t;
                import core.sys.posix.time : CLOCK_MONOTONIC;
                () @trusted
                {
                    pthread_condattr_t attr = void;
                    int rc  = pthread_condattr_init( &attr );
                    if ( rc )
                        throw staticError!AssertError("Unable to initialize condition", __FILE__, __LINE__);
                    rc = pthread_condattr_setclock( &attr, CLOCK_MONOTONIC );
                    if ( rc )
                        throw staticError!AssertError("Unable to initialize condition", __FILE__, __LINE__);
                    rc = pthread_cond_init( cast(pthread_cond_t*) &m_hndl, &attr );
                    if ( rc )
                        throw staticError!AssertError("Unable to initialize condition", __FILE__, __LINE__);
                    rc = pthread_condattr_destroy( &attr );
                    if ( rc )
                        throw staticError!AssertError("Unable to initialize condition", __FILE__, __LINE__);
                } ();
            }
            else
            {
                int rc = pthread_cond_init( cast(pthread_cond_t*) &m_hndl, null );
                if ( rc )
                    throw staticError!AssertError("Unable to initialize condition", __FILE__, __LINE__);
            }
        }
    }

    ~this() @nogc
    {
        version (Windows)
        {
            BOOL rc = CloseHandle( m_blockLock );
            assert( rc, "Unable to destroy condition" );
            rc = CloseHandle( m_blockQueue );
            assert( rc, "Unable to destroy condition" );
            DeleteCriticalSection( &m_unblockLock );
        }
        else version (Posix)
        {
            int rc = pthread_cond_destroy( &m_hndl );
            assert( !rc, "Unable to destroy condition" );
        }
    }


    ////////////////////////////////////////////////////////////////////////////
    // General Properties
    ////////////////////////////////////////////////////////////////////////////


    /**
     * Gets the mutex associated with this condition.
     *
     * Returns:
     *  The mutex associated with this condition.
     */
    @property Mutex mutex()
    {
        return m_assocMutex;
    }

    /// ditto
    @property shared(Mutex) mutex() shared
    {
        import core.atomic : atomicLoad;
        return atomicLoad(m_assocMutex);
    }

    // undocumented function for internal use
    final @property Mutex mutex_nothrow() pure nothrow @safe @nogc
    {
        return m_assocMutex;
    }

    // ditto
    final @property shared(Mutex) mutex_nothrow() shared pure nothrow @safe @nogc
    {
        import core.atomic : atomicLoad;
        return atomicLoad(m_assocMutex);
    }

    ////////////////////////////////////////////////////////////////////////////
    // General Actions
    ////////////////////////////////////////////////////////////////////////////


    /**
     * Wait until notified.
     *
     * Throws:
     *  SyncError on error.
     */
    void wait()
    {
        wait!(typeof(this))(true);
    }

    /// ditto
    void wait() shared
    {
        wait!(typeof(this))(true);
    }

    /// ditto
    void wait(this Q)( bool _unused_ )
        if (is(Q == Condition) || is(Q == shared Condition))
    {
        version (Windows)
        {
            timedWait( INFINITE );
        }
        else version (Posix)
        {
            int rc = pthread_cond_wait( cast(pthread_cond_t*) &m_hndl, (cast(Mutex) mutex()).handleAddr() );
            if ( rc )
                throw staticError!AssertError("Unable to wait for condition", __FILE__, __LINE__);
        }
    }

    /**
     * Suspends the calling thread until a notification occurs or until the
     * supplied time period has elapsed.
     *
     * Params:
     *  val = The time to wait.
     *
     * In:
     *  val must be non-negative.
     *
     * Throws:
     *  SyncError on error.
     *
     * Returns:
     *  true if notified before the timeout and false if not.
     */
    bool wait( Duration val )
    {
        return wait!(typeof(this))(val, true);
    }

    /// ditto
    bool wait( Duration val ) shared
    {
        return wait!(typeof(this))(val, true);
    }

    /// ditto
    bool wait(this Q)( Duration val, bool _unused_ )
        if (is(Q == Condition) || is(Q == shared Condition))
    in
    {
        assert( !val.isNegative );
    }
    do
    {
        version (Windows)
        {
            auto maxWaitMillis = dur!("msecs")( uint.max - 1 );

            while ( val > maxWaitMillis )
            {
                if ( timedWait( cast(uint)
                               maxWaitMillis.total!"msecs" ) )
                    return true;
                val -= maxWaitMillis;
            }
            return timedWait( cast(uint) val.total!"msecs" );
        }
        else version (Posix)
        {
            timespec t = void;
            mktspec( t, val );

            int rc = pthread_cond_timedwait( cast(pthread_cond_t*) &m_hndl,
                                             (cast(Mutex) mutex()).handleAddr(),
                                             &t );
            if ( !rc )
                return true;
            if ( rc == ETIMEDOUT )
                return false;
            throw staticError!AssertError("Unable to wait for condition", __FILE__, __LINE__);
        }
    }

    /**
     * Notifies one waiter.
     *
     * Throws:
     *  SyncError on error.
     */
    void notify()
    {
        notify!(typeof(this))(true);
    }

    /// ditto
    void notify() shared
    {
        notify!(typeof(this))(true);
    }

    /// ditto
    void notify(this Q)( bool _unused_ )
        if (is(Q == Condition) || is(Q == shared Condition))
    {
        version (Windows)
        {
            notify_( false );
        }
        else version (Posix)
        {
            // Since OS X 10.7 (Lion), pthread_cond_signal returns EAGAIN after retrying 8192 times,
            // so need to retrying while it returns EAGAIN.
            //
            // 10.7.0 (Lion):          http://www.opensource.apple.com/source/Libc/Libc-763.11/pthreads/pthread_cond.c
            // 10.8.0 (Mountain Lion): http://www.opensource.apple.com/source/Libc/Libc-825.24/pthreads/pthread_cond.c
            // 10.10.0 (Yosemite):     http://www.opensource.apple.com/source/libpthread/libpthread-105.1.4/src/pthread_cond.c
            // 10.11.0 (El Capitan):   http://www.opensource.apple.com/source/libpthread/libpthread-137.1.1/src/pthread_cond.c
            // 10.12.0 (Sierra):       http://www.opensource.apple.com/source/libpthread/libpthread-218.1.3/src/pthread_cond.c
            // 10.13.0 (High Sierra):  http://www.opensource.apple.com/source/libpthread/libpthread-301.1.6/src/pthread_cond.c
            // 10.14.0 (Mojave):       http://www.opensource.apple.com/source/libpthread/libpthread-330.201.1/src/pthread_cond.c
            // 10.14.1 (Mojave):       http://www.opensource.apple.com/source/libpthread/libpthread-330.220.2/src/pthread_cond.c

            int rc;
            do {
                rc = pthread_cond_signal( cast(pthread_cond_t*) &m_hndl );
            } while ( rc == EAGAIN );
            if ( rc )
                throw staticError!AssertError("Unable to notify condition", __FILE__, __LINE__);
        }
    }

    /**
     * Notifies all waiters.
     *
     * Throws:
     *  SyncError on error.
     */
    void notifyAll()
    {
        notifyAll!(typeof(this))(true);
    }

    /// ditto
    void notifyAll() shared
    {
        notifyAll!(typeof(this))(true);
    }

    /// ditto
    void notifyAll(this Q)( bool _unused_ )
        if (is(Q == Condition) || is(Q == shared Condition))
    {
        version (Windows)
        {
            notify_( true );
        }
        else version (Posix)
        {
            // Since OS X 10.7 (Lion), pthread_cond_broadcast returns EAGAIN after retrying 8192 times,
            // so need to retrying while it returns EAGAIN.
            //
            // 10.7.0 (Lion):          http://www.opensource.apple.com/source/Libc/Libc-763.11/pthreads/pthread_cond.c
            // 10.8.0 (Mountain Lion): http://www.opensource.apple.com/source/Libc/Libc-825.24/pthreads/pthread_cond.c
            // 10.10.0 (Yosemite):     http://www.opensource.apple.com/source/libpthread/libpthread-105.1.4/src/pthread_cond.c
            // 10.11.0 (El Capitan):   http://www.opensource.apple.com/source/libpthread/libpthread-137.1.1/src/pthread_cond.c
            // 10.12.0 (Sierra):       http://www.opensource.apple.com/source/libpthread/libpthread-218.1.3/src/pthread_cond.c
            // 10.13.0 (High Sierra):  http://www.opensource.apple.com/source/libpthread/libpthread-301.1.6/src/pthread_cond.c
            // 10.14.0 (Mojave):       http://www.opensource.apple.com/source/libpthread/libpthread-330.201.1/src/pthread_cond.c
            // 10.14.1 (Mojave):       http://www.opensource.apple.com/source/libpthread/libpthread-330.220.2/src/pthread_cond.c

            int rc;
            do {
                rc = pthread_cond_broadcast( cast(pthread_cond_t*) &m_hndl );
            } while ( rc == EAGAIN );
            if ( rc )
                throw staticError!AssertError("Unable to notify condition", __FILE__, __LINE__);
        }
    }

private:
    version (Windows)
    {
        bool timedWait(this Q)( DWORD timeout )
            if (is(Q == Condition) || is(Q == shared Condition))
        {
            static if (is(Q == Condition))
            {
                auto op(string o, T, V1)(ref T val, V1 mod)
                {
                    return mixin("val " ~ o ~ "mod");
                }
            }
            else
            {
                auto op(string o, T, V1)(ref shared T val, V1 mod)
                {
                    import core.atomic: atomicOp;
                    return atomicOp!o(val, mod);
                }
            }

            int   numSignalsLeft;
            int   numWaitersGone;
            DWORD rc;

            rc = WaitForSingleObject( cast(HANDLE) m_blockLock, INFINITE );
            assert( rc == WAIT_OBJECT_0 );

            op!"+="(m_numWaitersBlocked, 1);

            rc = ReleaseSemaphore( cast(HANDLE) m_blockLock, 1, null );
            assert( rc );

            m_assocMutex.unlock();
            scope(failure) m_assocMutex.lock();

            rc = WaitForSingleObject( cast(HANDLE) m_blockQueue, timeout );
            assert( rc == WAIT_OBJECT_0 || rc == WAIT_TIMEOUT );
            bool timedOut = (rc == WAIT_TIMEOUT);

            EnterCriticalSection( &m_unblockLock );
            scope(failure) LeaveCriticalSection( &m_unblockLock );

            if ( (numSignalsLeft = m_numWaitersToUnblock) != 0 )
            {
                if ( timedOut )
                {
                    // timeout (or canceled)
                    if ( m_numWaitersBlocked != 0 )
                    {
                        op!"-="(m_numWaitersBlocked, 1);
                        // do not unblock next waiter below (already unblocked)
                        numSignalsLeft = 0;
                    }
                    else
                    {
                        // spurious wakeup pending!!
                        m_numWaitersGone = 1;
                    }
                }
                if ( op!"-="(m_numWaitersToUnblock, 1) == 0 )
                {
                    if ( m_numWaitersBlocked != 0 )
                    {
                        // open the gate
                        rc = ReleaseSemaphore( cast(HANDLE) m_blockLock, 1, null );
                        assert( rc );
                        // do not open the gate below again
                        numSignalsLeft = 0;
                    }
                    else if ( (numWaitersGone = m_numWaitersGone) != 0 )
                    {
                        m_numWaitersGone = 0;
                    }
                }
            }
            else if ( op!"+="(m_numWaitersGone, 1) == int.max / 2 )
            {
                // timeout/canceled or spurious event :-)
                rc = WaitForSingleObject( cast(HANDLE) m_blockLock, INFINITE );
                assert( rc == WAIT_OBJECT_0 );
                // something is going on here - test of timeouts?
                op!"-="(m_numWaitersBlocked, m_numWaitersGone);
                rc = ReleaseSemaphore( cast(HANDLE) m_blockLock, 1, null );
                assert( rc == WAIT_OBJECT_0 );
                m_numWaitersGone = 0;
            }

            LeaveCriticalSection( &m_unblockLock );

            if ( numSignalsLeft == 1 )
            {
                // better now than spurious later (same as ResetEvent)
                for ( ; numWaitersGone > 0; --numWaitersGone )
                {
                    rc = WaitForSingleObject( cast(HANDLE) m_blockQueue, INFINITE );
                    assert( rc == WAIT_OBJECT_0 );
                }
                // open the gate
                rc = ReleaseSemaphore( cast(HANDLE) m_blockLock, 1, null );
                assert( rc );
            }
            else if ( numSignalsLeft != 0 )
            {
                // unblock next waiter
                rc = ReleaseSemaphore( cast(HANDLE) m_blockQueue, 1, null );
                assert( rc );
            }
            m_assocMutex.lock();
            return !timedOut;
        }


        void notify_(this Q)( bool all )
            if (is(Q == Condition) || is(Q == shared Condition))
        {
            static if (is(Q == Condition))
            {
                auto op(string o, T, V1)(ref T val, V1 mod)
                {
                    return mixin("val " ~ o ~ "mod");
                }
            }
            else
            {
                auto op(string o, T, V1)(ref shared T val, V1 mod)
                {
                    import core.atomic: atomicOp;
                    return atomicOp!o(val, mod);
                }
            }

            DWORD rc;

            EnterCriticalSection( &m_unblockLock );
            scope(failure) LeaveCriticalSection( &m_unblockLock );

            if ( m_numWaitersToUnblock != 0 )
            {
                if ( m_numWaitersBlocked == 0 )
                {
                    LeaveCriticalSection( &m_unblockLock );
                    return;
                }
                if ( all )
                {
                    op!"+="(m_numWaitersToUnblock, m_numWaitersBlocked);
                    m_numWaitersBlocked = 0;
                }
                else
                {
                    op!"+="(m_numWaitersToUnblock, 1);
                    op!"-="(m_numWaitersBlocked, 1);
                }
                LeaveCriticalSection( &m_unblockLock );
            }
            else if ( m_numWaitersBlocked > m_numWaitersGone )
            {
                rc = WaitForSingleObject( cast(HANDLE) m_blockLock, INFINITE );
                assert( rc == WAIT_OBJECT_0 );
                if ( 0 != m_numWaitersGone )
                {
                    op!"-="(m_numWaitersBlocked, m_numWaitersGone);
                    m_numWaitersGone = 0;
                }
                if ( all )
                {
                    m_numWaitersToUnblock = m_numWaitersBlocked;
                    m_numWaitersBlocked = 0;
                }
                else
                {
                    m_numWaitersToUnblock = 1;
                    op!"-="(m_numWaitersBlocked, 1);
                }
                LeaveCriticalSection( &m_unblockLock );
                rc = ReleaseSemaphore( cast(HANDLE) m_blockQueue, 1, null );
                assert( rc );
            }
            else
            {
                LeaveCriticalSection( &m_unblockLock );
            }
        }


        // NOTE: This implementation uses Algorithm 8c as described here:
        //       http://groups.google.com/group/comp.programming.threads/
        //              browse_frm/thread/1692bdec8040ba40/e7a5f9d40e86503a
        HANDLE              m_blockLock;    // auto-reset event (now semaphore)
        HANDLE              m_blockQueue;   // auto-reset event (now semaphore)
        Mutex               m_assocMutex;   // external mutex/CS
        CRITICAL_SECTION    m_unblockLock;  // internal mutex/CS
        int                 m_numWaitersGone        = 0;
        int                 m_numWaitersBlocked     = 0;
        int                 m_numWaitersToUnblock   = 0;
    }
    else version (Posix)
    {
        Mutex               m_assocMutex;
        pthread_cond_t      m_hndl;
    }
}


////////////////////////////////////////////////////////////////////////////////
// Unit Tests
////////////////////////////////////////////////////////////////////////////////

unittest
{
    import core.sync.mutex;
    import core.sync.semaphore;
    import core.thread;


    void testNotify()
    {
        auto mutex      = new Mutex;
        auto condReady  = new Condition( mutex );
        auto semDone    = new Semaphore;
        auto synLoop    = new Object;
        int  numWaiters = 10;
        int  numTries   = 10;
        int  numReady   = 0;
        int  numTotal   = 0;
        int  numDone    = 0;
        int  numPost    = 0;

        void waiter()
        {
            for ( int i = 0; i < numTries; ++i )
            {
                synchronized( mutex )
                {
                    while ( numReady < 1 )
                    {
                        condReady.wait();
                    }
                    --numReady;
                    ++numTotal;
                }

                synchronized( synLoop )
                {
                    ++numDone;
                }
                semDone.wait();
            }
        }

        auto group = new ThreadGroup;

        for ( int i = 0; i < numWaiters; ++i )
            group.create( &waiter );

        for ( int i = 0; i < numTries; ++i )
        {
            for ( int j = 0; j < numWaiters; ++j )
            {
                synchronized( mutex )
                {
                    ++numReady;
                    condReady.notify();
                }
            }
            while ( true )
            {
                synchronized( synLoop )
                {
                    if ( numDone >= numWaiters )
                        break;
                }
                Thread.yield();
            }
            for ( int j = 0; j < numWaiters; ++j )
            {
                semDone.notify();
            }
        }

        group.joinAll();
        assert( numTotal == numWaiters * numTries );
    }


    void testNotifyAll()
    {
        auto mutex      = new Mutex;
        auto condReady  = new Condition( mutex );
        int  numWaiters = 10;
        int  numReady   = 0;
        int  numDone    = 0;
        bool alert      = false;

        void waiter()
        {
            synchronized( mutex )
            {
                ++numReady;
                while ( !alert )
                    condReady.wait();
                ++numDone;
            }
        }

        auto group = new ThreadGroup;

        for ( int i = 0; i < numWaiters; ++i )
            group.create( &waiter );

        while ( true )
        {
            synchronized( mutex )
            {
                if ( numReady >= numWaiters )
                {
                    alert = true;
                    condReady.notifyAll();
                    break;
                }
            }
            Thread.yield();
        }
        group.joinAll();
        assert( numReady == numWaiters && numDone == numWaiters );
    }


    void testWaitTimeout()
    {
        auto mutex      = new Mutex;
        auto condReady  = new Condition( mutex );
        bool waiting    = false;
        bool alertedOne = true;
        bool alertedTwo = true;

        void waiter()
        {
            synchronized( mutex )
            {
                waiting    = true;
                // we never want to miss the notification (30s)
                alertedOne = condReady.wait( dur!"seconds"(30) );
                // but we don't want to wait long for the timeout (10ms)
                alertedTwo = condReady.wait( dur!"msecs"(10) );
            }
        }

        auto thread = new Thread( &waiter );
        thread.start();

        while ( true )
        {
            synchronized( mutex )
            {
                if ( waiting )
                {
                    condReady.notify();
                    break;
                }
            }
            Thread.yield();
        }
        thread.join();
        assert( waiting );
        assert( alertedOne );
        assert( !alertedTwo );
    }

    testNotify();
    testNotifyAll();
    testWaitTimeout();
}

unittest
{
    import core.sync.mutex;
    import core.sync.semaphore;
    import core.thread;


    void testNotify()
    {
        auto mutex      = new shared Mutex;
        auto condReady  = new shared Condition( mutex );
        auto semDone    = new Semaphore;
        auto synLoop    = new Object;
        int  numWaiters = 10;
        int  numTries   = 10;
        int  numReady   = 0;
        int  numTotal   = 0;
        int  numDone    = 0;
        int  numPost    = 0;

        void waiter()
        {
            for ( int i = 0; i < numTries; ++i )
            {
                synchronized( mutex )
                {
                    while ( numReady < 1 )
                    {
                        condReady.wait();
                    }
                    --numReady;
                    ++numTotal;
                }

                synchronized( synLoop )
                {
                    ++numDone;
                }
                semDone.wait();
            }
        }

        auto group = new ThreadGroup;

        for ( int i = 0; i < numWaiters; ++i )
            group.create( &waiter );

        for ( int i = 0; i < numTries; ++i )
        {
            for ( int j = 0; j < numWaiters; ++j )
            {
                synchronized( mutex )
                {
                    ++numReady;
                    condReady.notify();
                }
            }
            while ( true )
            {
                synchronized( synLoop )
                {
                    if ( numDone >= numWaiters )
                        break;
                }
                Thread.yield();
            }
            for ( int j = 0; j < numWaiters; ++j )
            {
                semDone.notify();
            }
        }

        group.joinAll();
        assert( numTotal == numWaiters * numTries );
    }


    void testNotifyAll()
    {
        auto mutex      = new shared Mutex;
        auto condReady  = new shared Condition( mutex );
        int  numWaiters = 10;
        int  numReady   = 0;
        int  numDone    = 0;
        bool alert      = false;

        void waiter()
        {
            synchronized( mutex )
            {
                ++numReady;
                while ( !alert )
                    condReady.wait();
                ++numDone;
            }
        }

        auto group = new ThreadGroup;

        for ( int i = 0; i < numWaiters; ++i )
            group.create( &waiter );

        while ( true )
        {
            synchronized( mutex )
            {
                if ( numReady >= numWaiters )
                {
                    alert = true;
                    condReady.notifyAll();
                    break;
                }
            }
            Thread.yield();
        }
        group.joinAll();
        assert( numReady == numWaiters && numDone == numWaiters );
    }


    void testWaitTimeout()
    {
        auto mutex      = new shared Mutex;
        auto condReady  = new shared Condition( mutex );
        bool waiting    = false;
        bool alertedOne = true;
        bool alertedTwo = true;

        void waiter()
        {
            synchronized( mutex )
            {
                waiting    = true;
                // we never want to miss the notification (30s)
                alertedOne = condReady.wait( dur!"seconds"(30) );
                // but we don't want to wait long for the timeout (10ms)
                alertedTwo = condReady.wait( dur!"msecs"(10) );
            }
        }

        auto thread = new Thread( &waiter );
        thread.start();

        while ( true )
        {
            synchronized( mutex )
            {
                if ( waiting )
                {
                    condReady.notify();
                    break;
                }
            }
            Thread.yield();
        }
        thread.join();
        assert( waiting );
        assert( alertedOne );
        assert( !alertedTwo );
    }

    testNotify();
    testNotifyAll();
    testWaitTimeout();
}