Condition Variables Practical

Coordination of Threads

• We will use two condition variables

• data_cv

  • The fetching thread notifies this when it has new data
  • The progress bar waits on it and updates itself

• completed_cv

  • The fetching thread notifies this when the download completes
  • The progress bar waits on it and exits
  • The processing thread waits on it and processes the data

• We use predicates with the condition variables

  • Avoid lost and spurious wake-ups

Progress Bar

• Implemented as a loop:

  • Wait on data_cv
  • Update progress
  • Wait on completed_cv
  • If the download is complete, exit

• Use blocking wait() on data_cv

• Use non blocking wait-for() on completed_cv

// One thread fetches the data
// Another thread displays a progress bar
// A third thread processes the data when the download is complete
// Implemented using a condition variable to communicate between the threads

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <string>
#include <thread>

using namespace std::literals;

std::string sdata;

bool update_progress = false;
bool completed = false;

std::mutex data_mutex;
std::mutex completed_mutex;

std::condition_variable data_cv;
std::condition_variable completed_cv;

void fetch_data() {
  for (int i = 0; i < 5; ++i) {
    std::cout << "Fetcher thread waiting for data\n";
    std::this_thread::sleep_for(2s);

    // update sdata, then notify the progress bar thread
    std::unique_lock<std::mutex> uniq_lck(data_mutex);
    sdata += "block" + std::to_string(i + 1);
    std::cout << "Fetched sdata: " << sdata << '\n';
    update_progress = true;
    uniq_lck.unlock();
    data_cv.notify_all();
  }

  std::cout << "Fetch sdata has ended\n";

  // tell the progress bar thread to exit and wake up the processing thread
  std::lock_guard<std::mutex> lg(completed_mutex);
  completed = true;
  completed_cv.notify_all();
}

void progress_bar() {
  size_t len = 0;

  while (true) {
    std::cout << "Progress bar thread waiting for data...\n";

    // wait until there is some new data to display
    std::unique_lock<std::mutex> data_lck(data_mutex);
    data_cv.wait(data_lck, [] { return update_progress; });

    // Wake up and use the new value
    len = sdata.size();

    update_progress = false;
    data_lck.unlock();

    std::cout << "Recieved " << len << " bytes so far\n";

    // check if download has finished
    std::unique_lock<std::mutex> compl_lock(completed_mutex);

    // Us wait_for() to avoid blocking
    if (completed_cv.wait_for(compl_lock, 10ms, [] { return completed; })) {
      std::cout << "Progress bar thread has ended\n";
      break;
    }
  }
}

void process_data() {
  std::this_thread::sleep_for(200ms);
  std::cout << "Processing thread waiting for data...\n";

  // Wait until download is complete
  std::unique_lock<std::mutex> compl_lck(completed_mutex);

  completed_cv.wait(compl_lck, [] { return completed; });
  compl_lck.unlock();

  std::lock_guard<std::mutex> data_lck(data_mutex);
  std::cout << "Processing sdata: " << sdata << '\n';
}

int main() {
  std::thread fetcher(fetch_data);
  std::thread prog(progress_bar);
  std::thread processing(process_data);

  fetcher.join();
  prog.join();
  processing.join();
}