C ++ 0x não possui semáforos? Como sincronizar threads?

É verdade que o C ++ 0x virá sem semáforos? Já existem algumas perguntas no Stack Overflow sobre o uso de semáforos. Eu os uso (semáforos posix) o tempo todo para deixar um thread aguardar algum evento em outro thread:

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

  event1.wait();

  ...
}

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

  event1.post();

  ...
}

Se eu faria isso com um mutex:

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

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

  ...
}

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

  doSomethingth1();

  event1.unlock();

  ...
}

Problema: É feio e não é garantido que o thread1 bloqueie o mutex primeiro (como o mesmo thread deve bloquear e desbloquear um mutex, você também não pode bloquear o evento1 antes do início do thread0 e do thread1).

Portanto, como o boost também não possui semáforos, qual é a maneira mais simples de conseguir o que foi mencionado acima?

Você pode criar facilmente um a partir de um mutex e uma variável de condição:

#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;
    }
};

Com base na resposta de Maxim Yegorushkin , tentei fazer o exemplo no 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;
};

Decidi escrever o semáforo C ++ 11 mais robusto / genérico que eu pudesse, no estilo do padrão, tanto quanto eu poderia (observe using semaphore = ..., você normalmente usaria o nome semaphoresemelhante ao normalmente usando stringnão 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 acordo com os semáforos posix, eu acrescentaria

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

E eu prefiro usar um mecanismo de sincronização em um nível conveniente de abstração, em vez de sempre copiar colando uma versão costurada usando operadores mais básicos.

Você também pode conferir o cpp11-on-multicore - ele possui uma implementação de semáforo portátil e ideal.

O repositório também contém outros itens de threading que complementam o threading do c ++ 11.

Você pode trabalhar com variáveis ​​mutex e condição. Você obtém acesso exclusivo com o mutex, verifique se deseja continuar ou precisa esperar pelo outro lado. Se precisar esperar, aguarde em uma condição. Quando o outro encadeamento determina que você pode continuar, ele sinaliza a condição.

Há um pequeno exemplo na biblioteca boost :: thread que você provavelmente pode copiar (as bibliotecas C ++ 0x e boost thread são muito semelhantes).

Também pode ser útil o invólucro de semáforo RAII nos threads:

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;
};

Exemplo de uso no aplicativo multithread:

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();

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

Estes terão (pelo menos) os seguintes métodos:

• acquire() (bloqueando)
• try_acquire() (sem bloqueio, retorna imediatamente)
• try_acquire_for() (sem bloqueio, dura uma duração)
• try_acquire_until() (sem bloqueio, leva um tempo para parar de tentar)
• release()

Isso ainda não está listado na cppreference, mas você pode ler os slides da apresentação do CppCon 2019 ou assistir ao vídeo . Há também a proposta oficial P0514R4 , mas não tenho certeza de que seja a versão mais atualizada.

Eu encontrei o shared_ptr e o fraco_ptr, um longo com uma lista, fez o trabalho que eu precisava. Meu problema era que eu tinha vários clientes que desejavam interagir com os dados internos de um host. Normalmente, o host atualiza os dados por conta própria; no entanto, se um cliente solicita, o host precisa parar de atualizar até que nenhum cliente esteja acessando os dados do host. Ao mesmo tempo, um cliente pode solicitar acesso exclusivo, para que nenhum outro cliente, nem o host, possa modificar os dados do host.

Como fiz isso, criei uma struct:

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

Cada cliente teria um membro desses:

UpdateLock::ptr m_myLock;

Em seguida, o host teria um membro fraco_ptr para exclusividade e uma lista de fraco_ptrs para bloqueios não exclusivos:

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

Há uma função para ativar o bloqueio e outra função para verificar se o host está bloqueado:

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

Testo bloqueios em LockUpdate, IsUpdateLocked e periodicamente na rotina de atualização do host. Testar um bloqueio é tão simples quanto verificar se o fraco_ptr expirou e remover qualquer um que expirou da lista m_locks (eu só faço isso durante a atualização do host); posso verificar se a lista está vazia; ao mesmo tempo, recebo o desbloqueio automático quando um cliente redefine o shared_ptr no qual está pendurado, o que também acontece quando um cliente é destruído automaticamente.

O efeito geral é que, como os clientes raramente precisam de exclusividade (normalmente reservados apenas para adições e exclusões), na maioria das vezes uma solicitação para LockUpdate (false), ou seja, não exclusiva, é bem-sucedida desde que (! M_exclusiveLock). E um LockUpdate (true), um pedido de exclusividade, é bem-sucedido apenas quando ambos (! M_exclusiveLock) e (m_locks.empty ()).

Uma fila pode ser adicionada para atenuar entre bloqueios exclusivos e não exclusivos; no entanto, até o momento não tive colisões, então pretendo esperar até que isso aconteça para adicionar a solução (principalmente para que eu tenha uma condição de teste no mundo real).

Até agora, isso está funcionando bem para minhas necessidades; Eu posso imaginar a necessidade de expandir isso, e alguns problemas que podem surgir sobre o uso expandido, no entanto, isso foi rápido de implementar e exigiu muito pouco código personalizado.

Caso alguém esteja interessado na versão atômica, aqui está a implementação. O desempenho é esperado melhor que a versão da variável mutex & condition.

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};
};