C ++ 0x no tiene semáforos? ¿Cómo sincronizar hilos?

¿Es cierto que C ++ 0x vendrá sin semáforos? Ya hay algunas preguntas sobre Stack Overflow con respecto al uso de semáforos. Los uso (semáforos posix) todo el tiempo para dejar que un hilo espere algún evento en otro hilo:

void thread0(...)
{
  doSomething0();

  event1.wait();

  ...
}

void thread1(...)
{
  doSomething1();

  event1.post();

  ...
}

Si hiciera eso con un mutex:

void thread0(...)
{
  doSomething0();

  event1.lock(); event1.unlock();

  ...
}

void thread1(...)
{
  event1.lock();

  doSomethingth1();

  event1.unlock();

  ...
}

Problema: es feo y no se garantiza que thread1 bloquee el mutex primero (dado que el mismo thread debería bloquear y desbloquear un mutex, tampoco puede bloquear event1 antes de que se inicien thread0 y thread1).

Entonces, dado que boost tampoco tiene semáforos, ¿cuál es la forma más simple de lograr lo anterior?

Puede crear fácilmente uno a partir de un mutex y una variable de condición:

#include <mutex>
#include <condition_variable>

class semaphore
{
private:
    std::mutex mutex_;
    std::condition_variable condition_;
    unsigned long count_ = 0; // Initialized as locked.

public:
    void notify() {
        std::lock_guard<decltype(mutex_)> lock(mutex_);
        ++count_;
        condition_.notify_one();
    }

    void wait() {
        std::unique_lock<decltype(mutex_)> lock(mutex_);
        while(!count_) // Handle spurious wake-ups.
            condition_.wait(lock);
        --count_;
    }

    bool try_wait() {
        std::lock_guard<decltype(mutex_)> lock(mutex_);
        if(count_) {
            --count_;
            return true;
        }
        return false;
    }
};

Basado en la respuesta de Maxim Yegorushkin , traté de hacer el ejemplo en estilo C ++ 11.

#include <mutex>
#include <condition_variable>

class Semaphore {
public:
    Semaphore (int count_ = 0)
        : count(count_) {}

    inline void notify()
    {
        std::unique_lock<std::mutex> lock(mtx);
        count++;
        cv.notify_one();
    }

    inline void wait()
    {
        std::unique_lock<std::mutex> lock(mtx);

        while(count == 0){
            cv.wait(lock);
        }
        count--;
    }

private:
    std::mutex mtx;
    std::condition_variable cv;
    int count;
};

Decidí escribir el semáforo C ++ 11 más robusto / genérico que pude, en el estilo del estándar tanto como pude (tenga en cuenta using semaphore = ...que normalmente usaría el nombre semaphoresimilar a stringno usar normalmente basic_string):

template <typename Mutex, typename CondVar>
class basic_semaphore {
public:
    using native_handle_type = typename CondVar::native_handle_type;

    explicit basic_semaphore(size_t count = 0);
    basic_semaphore(const basic_semaphore&) = delete;
    basic_semaphore(basic_semaphore&&) = delete;
    basic_semaphore& operator=(const basic_semaphore&) = delete;
    basic_semaphore& operator=(basic_semaphore&&) = delete;

    void notify();
    void wait();
    bool try_wait();
    template<class Rep, class Period>
    bool wait_for(const std::chrono::duration<Rep, Period>& d);
    template<class Clock, class Duration>
    bool wait_until(const std::chrono::time_point<Clock, Duration>& t);

    native_handle_type native_handle();

private:
    Mutex   mMutex;
    CondVar mCv;
    size_t  mCount;
};

using semaphore = basic_semaphore<std::mutex, std::condition_variable>;

template <typename Mutex, typename CondVar>
basic_semaphore<Mutex, CondVar>::basic_semaphore(size_t count)
    : mCount{count}
{}

template <typename Mutex, typename CondVar>
void basic_semaphore<Mutex, CondVar>::notify() {
    std::lock_guard<Mutex> lock{mMutex};
    ++mCount;
    mCv.notify_one();
}

template <typename Mutex, typename CondVar>
void basic_semaphore<Mutex, CondVar>::wait() {
    std::unique_lock<Mutex> lock{mMutex};
    mCv.wait(lock, [&]{ return mCount > 0; });
    --mCount;
}

template <typename Mutex, typename CondVar>
bool basic_semaphore<Mutex, CondVar>::try_wait() {
    std::lock_guard<Mutex> lock{mMutex};

    if (mCount > 0) {
        --mCount;
        return true;
    }

    return false;
}

template <typename Mutex, typename CondVar>
template<class Rep, class Period>
bool basic_semaphore<Mutex, CondVar>::wait_for(const std::chrono::duration<Rep, Period>& d) {
    std::unique_lock<Mutex> lock{mMutex};
    auto finished = mCv.wait_for(lock, d, [&]{ return mCount > 0; });

    if (finished)
        --mCount;

    return finished;
}

template <typename Mutex, typename CondVar>
template<class Clock, class Duration>
bool basic_semaphore<Mutex, CondVar>::wait_until(const std::chrono::time_point<Clock, Duration>& t) {
    std::unique_lock<Mutex> lock{mMutex};
    auto finished = mCv.wait_until(lock, t, [&]{ return mCount > 0; });

    if (finished)
        --mCount;

    return finished;
}

template <typename Mutex, typename CondVar>
typename basic_semaphore<Mutex, CondVar>::native_handle_type basic_semaphore<Mutex, CondVar>::native_handle() {
    return mCv.native_handle();
}

de acuerdo con los semáforos posix, agregaría

class semaphore
{
    ...
    bool trywait()
    {
        boost::mutex::scoped_lock lock(mutex_);
        if(count_)
        {
            --count_;
            return true;
        }
        else
        {
            return false;
        }
    }
};

Y prefiero usar un mecanismo de sincronización a un nivel conveniente de abstracción, en lugar de siempre copiar y pegar una versión unida usando operadores más básicos.

También puede consultar cpp11-on-multicore : tiene una implementación de semáforo portátil y óptima.

El repositorio también contiene otros objetos de subprocesamiento que complementan el subproceso de c ++ 11.

Puede trabajar con mutex y variables de condición. Obtiene acceso exclusivo con el mutex, verifique si desea continuar o si necesita esperar al otro extremo. Si necesita esperar, espere en una condición. Cuando el otro hilo determina que puede continuar, señala la condición.

Hay un breve ejemplo en la biblioteca boost :: thread que probablemente puedas copiar (las bibliotecas C ++ 0x y boost thread son muy similares).

También puede ser útil envoltorio de semáforo RAII en hilos:

class ScopedSemaphore
{
public:
    explicit ScopedSemaphore(Semaphore& sem) : m_Semaphore(sem) { m_Semaphore.Wait(); }
    ScopedSemaphore(const ScopedSemaphore&) = delete;
    ~ScopedSemaphore() { m_Semaphore.Notify(); }

   ScopedSemaphore& operator=(const ScopedSemaphore&) = delete;

private:
    Semaphore& m_Semaphore;
};

Ejemplo de uso en la aplicación multiproceso:

boost::ptr_vector<std::thread> threads;
Semaphore semaphore;

for (...)
{
    ...
    auto t = new std::thread([..., &semaphore]
    {
        ScopedSemaphore scopedSemaphore(semaphore);
        ...
    }
    );
    threads.push_back(t);
}

for (auto& t : threads)
    t.join();

C ++ 20 finalmente tendrá semáforos - std::counting_semaphore<max_count> .

Estos tendrán (al menos) los siguientes métodos:

• acquire() (bloqueo)
• try_acquire() (sin bloqueo, devuelve inmediatamente)
• try_acquire_for() (sin bloqueo, toma una duración)
• try_acquire_until() (sin bloqueo, toma un tiempo para dejar de intentarlo)
• release()

Esto aún no figura en cppreference, pero puede leer estas diapositivas de presentación de CppCon 2019 o ver el video . También está la propuesta oficial P0514R4 , pero no estoy seguro de que sea la versión más actualizada.

Encontré que shared_ptr y weak_ptr, un largo con una lista, hicieron el trabajo que necesitaba. Mi problema era que tenía varios clientes que querían interactuar con los datos internos de un host. Por lo general, el host actualiza los datos por sí mismo, sin embargo, si un cliente lo solicita, el host debe dejar de actualizarse hasta que ningún cliente acceda a los datos del host. Al mismo tiempo, un cliente puede solicitar acceso exclusivo, de modo que ningún otro cliente, ni el host, puedan modificar los datos del host.

Cómo hice esto fue, creé una estructura:

struct UpdateLock
{
    typedef std::shared_ptr< UpdateLock > ptr;
};

Cada cliente tendría un miembro de tales:

UpdateLock::ptr m_myLock;

Luego, el host tendría un miembro weak_ptr para exclusividad y una lista de weak_ptrs para bloqueos no exclusivos:

std::weak_ptr< UpdateLock > m_exclusiveLock;
std::list< std::weak_ptr< UpdateLock > > m_locks;

Hay una función para habilitar el bloqueo y otra función para verificar si el host está bloqueado:

UpdateLock::ptr LockUpdate( bool exclusive );       
bool IsUpdateLocked( bool exclusive ) const;

Compruebo los bloqueos en LockUpdate, IsUpdateLocked y periódicamente en la rutina de actualización del host. Probar un bloqueo es tan simple como verificar si los débiles_ptr han expirado, y eliminar cualquier vencido de la lista m_locks (solo hago esto durante la actualización del host), puedo verificar si la lista está vacía; Al mismo tiempo, obtengo desbloqueo automático cuando un cliente restablece el shared_ptr en el que están colgando, lo que también ocurre cuando un cliente se destruye automáticamente.

El efecto general es que, dado que los clientes rara vez necesitan exclusividad (generalmente reservado solo para adiciones y eliminaciones), la mayoría de las veces una solicitud a LockUpdate (falso), es decir, no exclusiva, tiene éxito siempre que (! M_exclusiveLock). Y un LockUpdate (verdadero), una solicitud de exclusividad, tiene éxito solo cuando ambos (! M_exclusiveLock) y (m_locks.empty ()).

Se podría agregar una cola para mitigar entre bloqueos exclusivos y no exclusivos, sin embargo, no he tenido colisiones hasta ahora, por lo que tengo la intención de esperar hasta que eso suceda para agregar la solución (principalmente para tener una condición de prueba del mundo real).

Hasta ahora esto está funcionando bien para mis necesidades; Puedo imaginar la necesidad de expandir esto, y algunos problemas que podrían surgir sobre el uso expandido, sin embargo, esto fue rápido de implementar y requirió muy poco código personalizado.

En caso de que alguien esté interesado en la versión atómica, aquí está la implementación. El rendimiento se espera mejor que la versión mutex y la variable de condición.

class semaphore_atomic
{
public:
    void notify() {
        count_.fetch_add(1, std::memory_order_release);
    }

    void wait() {
        while (true) {
            int count = count_.load(std::memory_order_relaxed);
            if (count > 0) {
                if (count_.compare_exchange_weak(count, count-1, std::memory_order_acq_rel, std::memory_order_relaxed)) {
                    break;
                }
            }
        }
    }

    bool try_wait() {
        int count = count_.load(std::memory_order_relaxed);
        if (count > 0) {
            if (count_.compare_exchange_strong(count, count-1, std::memory_order_acq_rel, std::memory_order_relaxed)) {
                return true;
            }
        }
        return false;
    }
private:
    std::atomic_int count_{0};
};