condition_variable.hpp

/* SPDX-License-Identifier: MPL-2.0 */

#ifndef __ZMQ_CONDITON_VARIABLE_HPP_INCLUDED__
#define __ZMQ_CONDITON_VARIABLE_HPP_INCLUDED__

#include "err.hpp"
#include "mutex.hpp"

//  Condition variable class encapsulates OS mutex in a platform-independent way.

#if defined(ZMQ_USE_CV_IMPL_NONE)

namespace zmq
{
class condition_variable_t
{
  public:
    inline condition_variable_t () { zmq_assert (false); }

    inline int wait (mutex_t *mutex_, int timeout_)
    {
        zmq_assert (false);
        return -1;
    }

    inline void broadcast () { zmq_assert (false); }

    ZMQ_NON_COPYABLE_NOR_MOVABLE (condition_variable_t)
};
}

#elif defined(ZMQ_USE_CV_IMPL_WIN32API)

#include "windows.hpp"

namespace zmq
{
class condition_variable_t
{
  public:
    inline condition_variable_t () { InitializeConditionVariable (&_cv); }

    inline int wait (mutex_t *mutex_, int timeout_)
    {
        int rc = SleepConditionVariableCS (&_cv, mutex_->get_cs (), timeout_);

        if (rc != 0)
            return 0;

        rc = GetLastError ();

        if (rc != ERROR_TIMEOUT)
            win_assert (rc);

        errno = EAGAIN;
        return -1;
    }

    inline void broadcast () { WakeAllConditionVariable (&_cv); }

  private:
    CONDITION_VARIABLE _cv;

    ZMQ_NON_COPYABLE_NOR_MOVABLE (condition_variable_t)
};
}

#elif defined(ZMQ_USE_CV_IMPL_STL11)

#include <condition_variable>

namespace zmq
{
class condition_variable_t
{
  public:
    condition_variable_t () ZMQ_DEFAULT;

    int wait (mutex_t *mutex_, int timeout_)
    {
        // this assumes that the mutex mutex_ has been locked by the caller
        int res = 0;
        if (timeout_ == -1) {
            _cv.wait (
              *mutex_); // unlock mtx and wait cv.notify_all(), lock mtx after cv.notify_all()
        } else if (_cv.wait_for (*mutex_, std::chrono::milliseconds (timeout_))
                   == std::cv_status::timeout) {
            // time expired
            errno = EAGAIN;
            res = -1;
        }
        return res;
    }

    void broadcast ()
    {
        // this assumes that the mutex associated with _cv has been locked by the caller
        _cv.notify_all ();
    }

  private:
    std::condition_variable_any _cv;

    ZMQ_NON_COPYABLE_NOR_MOVABLE (condition_variable_t)
};
}

#elif defined(ZMQ_USE_CV_IMPL_VXWORKS)

#include <sysLib.h>

namespace zmq
{
class condition_variable_t
{
  public:
    inline condition_variable_t () ZMQ_DEFAULT;

    inline ~condition_variable_t ()
    {
        scoped_lock_t l (_listenersMutex);
        for (size_t i = 0; i < _listeners.size (); i++) {
            semDelete (_listeners[i]);
        }
    }

    inline int wait (mutex_t *mutex_, int timeout_)
    {
        //Atomically releases lock, blocks the current executing thread,
        //and adds it to the list of threads waiting on *this. The thread
        //will be unblocked when broadcast() is executed.
        //It may also be unblocked spuriously. When unblocked, regardless
        //of the reason, lock is reacquired and wait exits.

        SEM_ID sem = semBCreate (SEM_Q_PRIORITY, SEM_EMPTY);
        {
            scoped_lock_t l (_listenersMutex);
            _listeners.push_back (sem);
        }
        mutex_->unlock ();

        int rc;
        if (timeout_ < 0)
            rc = semTake (sem, WAIT_FOREVER);
        else {
            int ticksPerSec = sysClkRateGet ();
            int timeoutTicks = (timeout_ * ticksPerSec) / 1000 + 1;
            rc = semTake (sem, timeoutTicks);
        }

        {
            scoped_lock_t l (_listenersMutex);
            // remove sem from listeners
            for (size_t i = 0; i < _listeners.size (); i++) {
                if (_listeners[i] == sem) {
                    _listeners.erase (_listeners.begin () + i);
                    break;
                }
            }
            semDelete (sem);
        }
        mutex_->lock ();

        if (rc == 0)
            return 0;

        if (rc == S_objLib_OBJ_TIMEOUT) {
            errno = EAGAIN;
            return -1;
        }

        return -1;
    }

    inline void broadcast ()
    {
        scoped_lock_t l (_listenersMutex);
        for (size_t i = 0; i < _listeners.size (); i++) {
            semGive (_listeners[i]);
        }
    }

  private:
    mutex_t _listenersMutex;
    std::vector<SEM_ID> _listeners;

    ZMQ_NON_COPYABLE_NOR_MOVABLE (condition_variable_t)
};
}

#elif defined(ZMQ_USE_CV_IMPL_PTHREADS)

#include <pthread.h>

#if defined(__ANDROID_API__) && __ANDROID_API__ < 21
#define ANDROID_LEGACY
extern "C" int pthread_cond_timedwait_monotonic_np (pthread_cond_t *,
                                                    pthread_mutex_t *,
                                                    const struct timespec *);
#endif

namespace zmq
{
class condition_variable_t
{
  public:
    inline condition_variable_t ()
    {
        pthread_condattr_t attr;
        pthread_condattr_init (&attr);
#if !defined(ZMQ_HAVE_OSX) && !defined(ANDROID_LEGACY)
        pthread_condattr_setclock (&attr, CLOCK_MONOTONIC);
#endif
        int rc = pthread_cond_init (&_cond, &attr);
        posix_assert (rc);
    }

    inline ~condition_variable_t ()
    {
        int rc = pthread_cond_destroy (&_cond);
        posix_assert (rc);
    }

    inline int wait (mutex_t *mutex_, int timeout_)
    {
        int rc;

        if (timeout_ != -1) {
            struct timespec timeout;

#ifdef ZMQ_HAVE_OSX
            timeout.tv_sec = 0;
            timeout.tv_nsec = 0;
#else
            rc = clock_gettime (CLOCK_MONOTONIC, &timeout);
            posix_assert (rc);
#endif

            timeout.tv_sec += timeout_ / 1000;
            timeout.tv_nsec += (timeout_ % 1000) * 1000000;

            if (timeout.tv_nsec >= 1000000000) {
                timeout.tv_sec++;
                timeout.tv_nsec -= 1000000000;
            }
#ifdef ZMQ_HAVE_OSX
            rc = pthread_cond_timedwait_relative_np (
              &_cond, mutex_->get_mutex (), &timeout);
#elif defined(ANDROID_LEGACY)
            rc = pthread_cond_timedwait_monotonic_np (
              &_cond, mutex_->get_mutex (), &timeout);
#else
            rc =
              pthread_cond_timedwait (&_cond, mutex_->get_mutex (), &timeout);
#endif
        } else
            rc = pthread_cond_wait (&_cond, mutex_->get_mutex ());

        if (rc == 0)
            return 0;

        if (rc == ETIMEDOUT) {
            errno = EAGAIN;
            return -1;
        }

        posix_assert (rc);
        return -1;
    }

    inline void broadcast ()
    {
        int rc = pthread_cond_broadcast (&_cond);
        posix_assert (rc);
    }

  private:
    pthread_cond_t _cond;

    ZMQ_NON_COPYABLE_NOR_MOVABLE (condition_variable_t)
};
}

#endif

#endif