/** * The event module provides a primitive for lightweight signaling of other threads * (emulating Windows events on Posix) * * Copyright: Copyright (c) 2019 D Language Foundation * License: Distributed under the *    $(LINK2 http://www.boost.org/LICENSE_1_0.txt, Boost Software License 1.0). *    (See accompanying file LICENSE) * Authors: Rainer Schuetze * Source:    $(DRUNTIMESRC core/sync/event.d) */module core.sync.event; version (Windows){    import core.sys.windows.basetsd /+: HANDLE +/;    import core.sys.windows.winerror /+: WAIT_TIMEOUT +/;    import core.sys.windows.winbase /+: CreateEvent, CloseHandle, SetEvent, ResetEvent,        WaitForSingleObject, INFINITE, WAIT_OBJECT_0+/;}else version (Posix){    import core.sys.posix.pthread : pthread_cond_broadcast, pthread_cond_destroy, pthread_cond_init,        pthread_cond_timedwait, pthread_cond_wait, pthread_mutex_destroy, pthread_mutex_init, pthread_mutex_lock,        pthread_mutex_unlock;    import core.sys.posix.sys.types : pthread_cond_t, pthread_mutex_t;    import core.sys.posix.time : timespec;}else{    static assert(false, "Platform not supported");} import core.internal.abort : abort;import core.time; /** * represents an event. Clients of an event are suspended while waiting * for the event to be "signaled". * * Implemented using `pthread_mutex` and `pthread_condition` on Posix and * `CreateEvent` and `SetEvent` on Windows.---import core.sync.event, core.thread, std.file; struct ProcessFile{    ThreadGroup group;    Event event;    void[] buffer;     void doProcess()    {        event.wait();        // process buffer    }     void process(string filename)    {        event.initialize(true, false);        group = new ThreadGroup;        for (int i = 0; i < 10; ++i)            group.create(&doProcess);         buffer = std.file.read(filename);        event.setIfInitialized();        group.joinAll();        event.terminate();    }}--- */struct Event{nothrow @nogc:    /**     * Creates an event object.     *     * Params:     *  manualReset  = the state of the event is not reset automatically after resuming waiting clients     *  initialState = initial state of the signal     */    this(bool manualReset, bool initialState)    {        initialize(manualReset, initialState);    }     /**     * Initializes an event object. Does nothing if the event is already initialized.     *     * Params:     *  manualReset  = the state of the event is not reset automatically after resuming waiting clients     *  initialState = initial state of the signal     */    void initialize(bool manualReset, bool initialState)    {        version (Windows)        {            if (m_event)                return;            m_event = CreateEvent(null, manualReset, initialState, null);            m_event || abort("Error: CreateEvent failed.");        }        else version (Posix)        {            if (m_initalized)                return;            pthread_mutex_init(cast(pthread_mutex_t*) &m_mutex, null) == 0 ||                abort("Error: pthread_mutex_init failed.");             static if ( is( typeof( imported!"core.sys.posix.pthread".pthread_condattr_setclock ) ) )            {                import core.sys.posix.pthread : CLOCK_MONOTONIC, pthread_condattr_destroy, pthread_condattr_init,                    pthread_condattr_setclock;                import core.sys.posix.sys.types : pthread_condattr_t;                 pthread_condattr_t attr = void;                pthread_condattr_init(&attr) == 0 ||                    abort("Error: pthread_condattr_init failed.");                pthread_condattr_setclock(&attr, CLOCK_MONOTONIC) == 0 ||                    abort("Error: pthread_condattr_setclock failed.");                pthread_cond_init(&m_cond, &attr) == 0 ||                    abort("Error: pthread_cond_init failed.");                pthread_condattr_destroy(&attr) == 0 ||                    abort("Error: pthread_condattr_destroy failed.");            }            else            {                pthread_cond_init(&m_cond, null) == 0 ||                    abort("Error: pthread_cond_init failed.");            }            m_state = initialState;            m_manualReset = manualReset;            m_initalized = true;        }    }     // copying not allowed, can produce resource leaks    @disable this(this);    @disable void opAssign(Event);     ~this()    {        terminate();    }     /**     * deinitialize event. Does nothing if the event is not initialized. There must not be     * threads currently waiting for the event to be signaled.    */    void terminate()    {        version (Windows)        {            if (m_event)                CloseHandle(m_event);            m_event = null;        }        else version (Posix)        {            if (m_initalized)            {                pthread_mutex_destroy(&m_mutex) == 0 ||                    abort("Error: pthread_mutex_destroy failed.");                pthread_cond_destroy(&m_cond) == 0 ||                    abort("Error: pthread_cond_destroy failed.");                m_initalized = false;            }        }    }     deprecated ("Use setIfInitialized() instead") void set()    {        setIfInitialized();    }     /// Set the event to "signaled", so that waiting clients are resumed    void setIfInitialized()    {        version (Windows)        {            if (m_event)                SetEvent(m_event);        }        else version (Posix)        {            if (m_initalized)            {                pthread_mutex_lock(&m_mutex);                m_state = true;                pthread_cond_broadcast(&m_cond);                pthread_mutex_unlock(&m_mutex);            }        }    }     /// Reset the event manually    void reset()    {        version (Windows)        {            if (m_event)                ResetEvent(m_event);        }        else version (Posix)        {            if (m_initalized)            {                pthread_mutex_lock(&m_mutex);                m_state = false;                pthread_mutex_unlock(&m_mutex);            }        }    }     /**     * Wait for the event to be signaled without timeout.     *     * Returns:     *  `true` if the event is in signaled state, `false` if the event is uninitialized or another error occured     */    bool wait()    {        version (Windows)        {            return m_event && WaitForSingleObject(m_event, INFINITE) == WAIT_OBJECT_0;        }        else version (Posix)        {            return wait(Duration.max);        }    }     /**     * Wait for the event to be signaled with timeout.     *     * Params:     *  tmout = the maximum time to wait     * Returns:     *  `true` if the event is in signaled state, `false` if the event was nonsignaled for the given time or     *  the event is uninitialized or another error occured     */    bool wait(Duration tmout)    {        version (Windows)        {            if (!m_event)                return false;             auto maxWaitMillis = dur!("msecs")(uint.max - 1);             while (tmout > maxWaitMillis)            {                auto res = WaitForSingleObject(m_event, uint.max - 1);                if (res != WAIT_TIMEOUT)                    return res == WAIT_OBJECT_0;                tmout -= maxWaitMillis;            }            auto ms = cast(uint)(tmout.total!"msecs");            return WaitForSingleObject(m_event, ms) == WAIT_OBJECT_0;        }        else version (Posix)        {            if (!m_initalized)                return false;             pthread_mutex_lock(&m_mutex);             int result = 0;            if (!m_state)            {                if (tmout == Duration.max)                {                    result = pthread_cond_wait(&m_cond, &m_mutex);                }                else                {                    import core.sync.config;                     timespec t = void;                    mktspec(t, tmout);                     result = pthread_cond_timedwait(&m_cond, &m_mutex, &t);                }            }            if (result == 0 && !m_manualReset)                m_state = false;             pthread_mutex_unlock(&m_mutex);             return result == 0;        }    } private:    version (Windows)    {        HANDLE m_event;    }    else version (Posix)    {        pthread_mutex_t m_mutex;        pthread_cond_t m_cond;        bool m_initalized;        bool m_state;        bool m_manualReset;    }} // Test single-thread (non-shared) use.@nogc nothrow unittest{    // auto-reset, initial state false    Event ev1 = Event(false, false);    assert(!ev1.wait(1.dur!"msecs"));    ev1.setIfInitialized();    assert(ev1.wait());    assert(!ev1.wait(1.dur!"msecs"));     // manual-reset, initial state true    Event ev2 = Event(true, true);    assert(ev2.wait());    assert(ev2.wait());    ev2.reset();    assert(!ev2.wait(1.dur!"msecs"));} unittest{    import core.atomic;    import core.thread;     scope event      = new Event(true, false);    int  numThreads = 10;    shared int numRunning = 0;     void testFn()    {        event.wait(8.dur!"seconds"); // timeout below limit for druntime test_runner        numRunning.atomicOp!"+="(1);    }     auto group = new ThreadGroup;     for (int i = 0; i < numThreads; ++i)        group.create(&testFn);     auto start = MonoTime.currTime;    assert(numRunning == 0);     event.setIfInitialized();    group.joinAll();     assert(numRunning == numThreads);     assert(MonoTime.currTime - start < 5.dur!"seconds");}