Thread synchronization with atomic

Introduction

In earlier blog, I demonstrate the use of conditional_variable and unique_lock to block current thread until other threads provides a positive condition. Once the current block wakes up, the current thread can continue its tasks. There are some drawbacks with using conditional_variable.

In this blog, I tried to achieve the same goal, using atomic.

Concepts

Drawback from using conditional_variable

There are 2 problems with conditional_variables explained in "Effective Modern C++":

If the detecting task notifies the conditional_variable before the reacting task waits,
the reacting task will hang. In order for notification of a conditional_variable to wake
another task, the other task must be waiting on that conditional_variable . If the detecting
task happens to execute the notification before the reacting task executes the
wait, the reacting task will miss the notification, and it will wait forever.
The wait statement fails to account for spurious wakeups. A fact of life in threading APIs (in many languages—not just C++) is that code waiting on a conditional_variable may be awakened even if the conditional_variable wasn’t notified. Such awakenings are known as spurious wakeups

Atomic

Atomic is a template class. According to "Effective Modern C++":

Once a std::atomic object has been constructed, operations on it behave as if they were inside a mutex-protected critical section, but the operations are generally implemented using special machine instructions that are more efficient than would be the case if a mutex were employed.

The following program demonstrates how atomic could be used in place of mutex to prevent race condition

#include <thread>
#include <iostream>
#include <atomic>
using namespace std;

static std::atomic<int> glob (0) ;

static void threadFunc(int loop)
{

    for (int j = 0; j < loop; j++) {
        glob++;
    }

}
int main(int argc, char *argv[])
{
     
     int loops=10000000;
    
     std::thread t1 (threadFunc, loops);
     std::thread t2 (threadFunc, loops);
     
     t1.join();
     t2.join();
     
     std::cout<<glob<<endl;
     return 0;
}

Atomic wrapper of integer value glob allows atomic operation, which means only one of 2 threads could access and modify the value at once. The above program prints output of 20,000,000 on every run.

Therefore, instead of using mutex + conditional variable, we could use a boolean and atomic variable as a flag that basically performs the same goals.

Using atomic instead of Conditional Variable.

The following program re-implements main thread and worker threads operation as described in Thread synchronization with Conditional Variables and Unique_Lock – My sky (freewindcode.com)

Main thread does:
- spawns the worker thread.
- sends data to worker thread.
- waits for worker thread to process data. Notification is presented as boolean value data=true.
- after receiving notification from worker threads, wakes up, prints data, then exit program.
Worker thread does:
- waits for data from main threads
- process data and notify main thread. Notification is presented as boolean value processed=true.
- terminate the thread.

The 2 global boolean and atomic values, defaulted as false, are used by conditional_variables:

bool data: set to be true by main thread. Worker thread is blocked and waits until notified by main thread. Worker thread then checks data if it is true and wakes up.
bool processes: set to be true by worker threads. Main thread is blocked and waits until notified by main thread. Main thread then checks data if it is true and wakes up.

#include <iostream>
#include <condition_variable>
#include <mutex>
#include <thread>
#include <string>
#include <atomic>
 
std::mutex m;
std::condition_variable cv;
std::string datastream;
std::atomic<bool> data (false);
std::atomic<bool> processed  (false);
 
void worker_thread_task()
{
    // 1. Worker threads waits until main() sends data
    std::cout<<"Worker threadID: "<<std::this_thread::get_id()<<std::endl;
    while(!data){
        std::this_thread::sleep_for(std::chrono::seconds(2));
    }
    if(!processed & data){

        // 2. after the wait from main, worker thread wakes up
        std::cout<<"    Worker thread is processing data"<<std::endl;
        datastream += " after processing 4,5,6";
     
        // 3. send data back to main()
        processed = true;
        data = false;
        std::cout<<"    Worker thread signals data processing completed"<<std::endl;

    }
}

void main_thread_sending_data(){
    // 1. main_thread sends data to Worker
    // until it received from workers that processed = true
    if (!data && !processed){
        std::this_thread::sleep_for(std::chrono::seconds(2));
        std::cout<<"main() signals data ready for processing, threadID "<<std::this_thread::get_id()<<std::endl;
        data = true;
    }
    
}

void main_thread_wait_for_processed_data(){
    // 1. main thread sleeps until worker thread sends notification
    while (!processed && data){
        std::this_thread::sleep_for(std::chrono::seconds(2));
    }
    // 2. main thread wakes up after worker thread processed data
    if (processed){
        std::cout<<"Back in main(), threadID "<<std::this_thread::get_id()<<std::endl;
        std::cout<<"    data = "<<datastream<<std::endl;
        processed = false;
        data = false;
    }
}
 
int main()
{
    std::thread worker(worker_thread_task);
 
    datastream = "Example data 1,2,3...";
    // send data to the worker thread
    main_thread_sending_data();
 
    // wait for the worker
    main_thread_wait_for_processed_data();
    
    worker.join();
}

Output: the output is printed as same as if the program is implemented with conditional_variable and mutex:

Worker threadID: 123790922454592
main() signals data ready for processing, threadID 123790922458944
    Worker thread is processing data
    Worker thread signals data processing completed
Back in main(), threadID 123790922458944
    data = Example data 1,2,3... after processing 4,5,6


...Program finished with exit code 0
Press ENTER to exit console.