How to properly synchronize threads in c/c++ application?

Question

I need to do proper synchronization over several threads in my application. The threads are devided into a group of threads - graup A which may contain more then one thread and thread B. Thread B is supposed to be unlocker thread while only one thread from group A at the same time is supposed to be unlocked by thread B. I tryied to achive stable solution using pthread_mutex_t with code like this:

// thread group A
...
while(...)
{
    pthread_mutex_lock(&lock) ;
    // only one thread at the same time allowed from here
    ...
}

// thread B
while(...)
{
    pthread_mutex_unlock(&lock)
    ...
}

...
int main()
{
    ...
    pthread_mutex_init(&lock, NULL) ;
    pthread_mutex_lock(&lock) ;
    ...
    // start threads
    ...
}

This solution works but is unstable and sometimes causes deadlock because if it happens that

pthread_mutex_unlock(&lock) ;

is called before

pthread_mutex_lock(&lock) ;

then mutex stays locked and causes deadlock because

pthread_mutex_unlock(&lock) ;

has no effect if it is called before

pthread_mutex_lock(&lock) ;

I found one crappy solution to this but it's crappy because it eats additional cpu time needlessly. Such solution is this:

bool lock_cond ;

// thread group A
...
while(...)
{
    lock_cond = true ;
    pthread_mutex_lock(&lock) ;
    lock_cond = false ;
    // only one thread at the same time allowed from here
    ...
}

// thread B
while(...)
{
    while(!lock_cond)
    ;
    pthread_mutex_unlock(&lock)
    ...
}

...
int main()
{
    ...
    pthread_mutex_init(&lock, NULL) ;
    pthread_mutex_lock(&lock) ;
    ...
    // start threads
    ...
}

So my question is how to properly implement threads synchronization in such scenario ?. Can I use

pthread_mutex_t

variables for that or does I have to use semaphore ? Please explain with code examples.

Answer 1

There are many kinds of synchronization patterns between different threads.

Your scenario seems to be a good fit for a binary semaphore rather than a mutex:

• Thread B doesn't "lock and release" - it just signals threads in the A group that they may proceed with their work.
• It's not clear that a thread in A, once done with its own work, allows other threads in A to start work.

C++ will have an std::binary_semaphore in the next language standard version. Until then, you'll need to use a C++ library implementing them (perhaps this one? I haven't tried it myself), or using POSIX semaphores in C-style coding.

Answer 2

After studying and modifying code samples taken from

https://en.cppreference.com/w/cpp/thread/condition_variable

for my needs I created the following:

#include <iostream>
#include <string>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <unistd.h>
#include <random>
#include <ctime>

std::mutex m, m1;
std::condition_variable cv, cv1;
bool ready = false, ready2 = false;
bool processed = false;

pthread_mutex_t only_one ;
bool done, done2 ;

class Task
{
    public:

    void thread_groupA(std::string msg)
    {
        while(!done)
        {
            pthread_mutex_lock(&only_one) ;

            {
                std::lock_guard<std::mutex> lk(m1);
                ready2 = true;
            }
            cv1.notify_one();

            std::cout << msg << std::endl ;

            std::cout << "before sleep 1 second" << std::endl ;
            sleep(1); // sleep for demonstration that it really works
            std::cout << "after sleep 1 second" << std::endl ;

            std::cout << "before cv.wait()" << std::endl ;
            std::unique_lock<std::mutex> lk(m);
            cv.wait(lk, []{return ready;});

            pthread_mutex_unlock(&only_one) ;

            std::cout << "after cv.wait()" << std::endl ;
            ready = false ;
            processed = true;
            lk.unlock();
            cv.notify_one();

            int val = rand() % 10000 ;
            usleep(val) ;  // server clients timing simulation
                           // different clients provide different data so clients timing isn't the same.
                           // fastest client's thread gets passed through 'pthread_mutex_lock(&only_one)'
         }
    }
} ;

void threadB()
{
    int aa = 2, bb = 0 ;
    while(!done2)
    {
        std::unique_lock<std::mutex> lk(m1);
        cv1.wait(lk, []{return ready2;});
        ready2 = false ;

        if(done2)
        break ;

        if(bb % aa)
        {
            std::cout << "before sleep 5 seconds" << std::endl ;
            sleep(5);  // sleep for demonstration that it really works
            std::cout << "after sleep 5 seconds" << std::endl ;
        }
        {
            std::lock_guard<std::mutex> lk(m);
            ready = true;
        }
        cv.notify_one();
        {
            std::unique_lock<std::mutex> lk(m);
            cv.wait(lk, []{return processed;});
            processed = false ;
        }
        ++bb ;
    }
}

int main()
{
    pthread_mutex_init(&only_one, NULL) ;

    done = false ;
    done2 = false ;
    srand(time(0)) ;

    Task * taskPtr1 = new Task();
    Task * taskPtr2 = new Task();

    std::thread worker1(&Task::thread_groupA, taskPtr1, "thread 1");
    std::thread worker2(&Task::thread_groupA, taskPtr2, "thread 2");
    std::thread signal(threadB);

    std::string s ;
    do
    {
        getline(std::cin, s) ;
    }
    while(s.compare("stop") != 0) ;
    done = true ;

    worker1.join();
    worker2.join();

    done2 = true ;

    {
        std::lock_guard<std::mutex> lk(m1);
        ready2 = true;
    }
    cv1.notify_one();

    signal.join();
}

Now based on this code I can make implementation to my app. I hope this will work pretty stable.