/** * 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 );

Lines 1–500 of 956. Load more