C ++ 0x nie ma semaforów? Jak synchronizować wątki?

Czy to prawda, że ​​C ++ 0x przyjdzie bez semaforów? Istnieje już kilka pytań dotyczących przepełnienia stosu dotyczących używania semaforów. Używam ich (semaforów posix) cały czas, aby pozwolić wątkowi czekać na jakieś zdarzenie w innym wątku:

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

  event1.wait();

  ...
}

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

  event1.post();

  ...
}

Gdybym zrobił to z muteksem:

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

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

  ...
}

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

  doSomethingth1();

  event1.unlock();

  ...
}

Problem: Jest brzydki i nie ma gwarancji, że Thread1 najpierw zablokuje muteks (biorąc pod uwagę, że ten sam wątek powinien blokować i odblokowywać muteks, nie możesz również zablokować zdarzenia1 przed uruchomieniem wątku0 i wątku1).

Skoro więc doładowanie również nie ma semaforów, jaki jest najprostszy sposób osiągnięcia powyższego?

Możesz łatwo zbudować go z muteksu i zmiennej warunku:

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

Bazując na odpowiedzi Maxima Yegorushkina , spróbowałem zrobić przykład w stylu 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;
};

Zdecydowałem się napisać najsolidniejszy / generyczny semafor C ++ 11, jaki mogłem, w stylu standardowym, jak tylko potrafię (uwaga using semaphore = ..., normalnie używałbyś nazwy semaphorepodobnej do normalnie używanej stringnie 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();
}

zgodnie z semaforami posix, dodałbym

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

I zdecydowanie wolę używać mechanizmu synchronizacji na dogodnym poziomie abstrakcji, zamiast zawsze kopiować wklejanie zszytej wersji przy użyciu bardziej podstawowych operatorów.

Możesz również sprawdzić cpp11-on-multicore - ma przenośną i optymalną implementację semafora.

Repozytorium zawiera również inne dodatki do wątków, które uzupełniają wątki w języku c ++ 11.

Możesz pracować ze zmiennymi mutex i warunkowymi. Uzyskujesz wyłączny dostęp z muteksem, sprawdź, czy chcesz kontynuować, czy musisz czekać na drugi koniec. Jeśli musisz czekać, czekasz w stanie. Gdy inny wątek stwierdzi, że możesz kontynuować, sygnalizuje stan.

W bibliotece boost :: thread jest krótki przykład , który najprawdopodobniej możesz po prostu skopiować (biblioteki wątków C ++ 0x i boost są bardzo podobne).

Może być również przydatna otoka semafora RAII w wątkach:

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

Przykład użycia w aplikacji wielowątkowej:

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 wreszcie będzie miał semafory - std::counting_semaphore<max_count> .

Będą one miały (przynajmniej) następujące metody:

• acquire() (bloking)
• try_acquire() (bez blokowania, zwraca natychmiast)
• try_acquire_for() (bez blokowania, trwa pewien czas)
• try_acquire_until() (bez blokowania, zajmuje trochę czasu, aby przestać próbować)
• release()

Tego jeszcze nie ma na liście cppreference, ale możesz przeczytać te slajdy prezentacji CppCon 2019 lub obejrzeć wideo . Jest też oficjalna propozycja P0514R4 , ale nie jestem pewien, czy jest to najbardziej aktualna wersja.

Okazało się, że shared_ptr i słabe_ptr, długie z listą, wykonały pracę, której potrzebowałem. Mój problem polegał na tym, że kilku klientów chciało wchodzić w interakcje z wewnętrznymi danymi hosta. Zazwyczaj host aktualizuje dane samodzielnie, jednak jeśli klient tego zażąda, musi przerwać aktualizację, dopóki żaden klient nie uzyska dostępu do danych hosta. W tym samym czasie klient może poprosić o wyłączny dostęp, aby żaden inny klient ani host nie mógł modyfikować danych hosta.

Jak to zrobiłem, stworzyłem strukturę:

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

Każdy klient miałby członka takiego:

UpdateLock::ptr m_myLock;

Wtedy host miałby element slaby_ptr na wyłączność i listę slabych_ptrs dla blokad niewyłącznych:

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

Dostępna jest funkcja umożliwiająca blokowanie i inna funkcja sprawdzająca, czy host jest zablokowany:

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

Testuję blokady w LockUpdate, IsUpdateLocked i okresowo w ramach procedury aktualizacji hosta. Testowanie blokady jest tak proste, jak sprawdzenie, czy słaby_ptr wygasł i usunięcie dowolnego wygasłego z listy m_locks (robię to tylko podczas aktualizacji hosta). Mogę sprawdzić, czy lista jest pusta; w tym samym czasie otrzymuję automatyczne odblokowanie, gdy klient resetuje shared_ptr, na którym się trzyma, co również ma miejsce, gdy klient zostaje automatycznie zniszczony.

Ogólny efekt jest taki, że ponieważ klienci rzadko potrzebują wyłączności (zwykle zarezerwowanej tylko dla dodatków i usunięć), w większości przypadków żądanie LockUpdate (false), to znaczy niewyłączne, kończy się sukcesem, o ile (! M_exclusiveLock). LockUpdate (true), żądanie wyłączności, kończy się sukcesem tylko wtedy, gdy zarówno (! M_exclusiveLock), jak i (m_locks.empty ()).

Można dodać kolejkę, aby złagodzić blokady między wyłącznymi i niewyłącznymi blokadami, jednak do tej pory nie miałem żadnych kolizji, więc zamierzam poczekać, aż to się stanie, aby dodać rozwiązanie (głównie dlatego, że mam rzeczywisty warunek testu).

Jak dotąd działa to dobrze na moje potrzeby; Mogę sobie wyobrazić potrzebę rozszerzenia tego i pewne problemy, które mogą pojawić się podczas rozszerzonego użytkowania, jednak było to szybkie do wdrożenia i wymagało bardzo mało niestandardowego kodu.

Jeśli ktoś jest zainteresowany wersją atomową, oto implementacja. Oczekuje się, że wydajność będzie lepsza niż wersja ze zmiennymi muteksami i warunkami.

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