Thread Coordination Practical

Communication Between Threads

• The threads run concurrently

  • The data fetching thread runs continually
  • The progress bar thread waits for information
  • The processor thread waits until all the data has been received

• When the download is complete

  • The fetching thread terminates
  • The progress bar thread terminates
  • The processor thread runs

Data Sharing Between Threads

• The downloaded data is shared by all three threads

  • The data fetching appends to it
  • The progress bar thread calculates its size
  • The processor thread uses the data

• Potential Data Race

  • Multiple threads
  • Modification

Coordination of Threads

• We will use two bools to coordinate the threads

• "progress" flag

  • The fetching thread sets this when it has new data
  • The progress bar thread checks this flag

• "completed" flag

  • The fetching thread sets this when it finishes
  • The other two threads check this flag

• Potential Data Race

  • Multiple threads
  • Modification

• Use mutexes

Hot Loop

• We need to lock the mutex while checking a bool

// In progress bar task function
std::lock_guard data_lck(data_mutex);
while (!update_progress) {}

• The thread will run flat out

  • The processor core will run at 100%
  • Other threads cannot do useful work
  • Uses a lot of electricity

• The fetcher thread cannot set the flag

Hot Loop Avoidance

• To avoid this, unlock the mutex inside the Loop

std::unique_lock<std::mutex> data_lck(data_mutex);

while (!update_progress) {
  data_lck.unlock();
  std::this_thread::sleep_for(10ms);
  data_lck.lock();
}

• Sleeping allows other threads to use the core
• The fetcher thread can set the flag

#include <cstddef>
#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;

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

    std::lock_guard<std::mutex> data_lck(data_mutex);
    sdata += "Block" + std::to_string(i + 1);
    std::cout << "sdata: " << sdata << '\n';
    update_progress = true;
  }

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

  // Tell the progress bar thread to exit and wake up the processing thread
  std::lock_guard<std::mutex> completed_lck(completed_mutex);
  completed = true;
}

// progress bar thread
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);
    while (!update_progress) {
      data_lck.unlock();
      std::this_thread::sleep_for(10ms);
      data_lck.lock();
    }

    len = sdata.size();
    update_progress = false;
    data_lck.unlock();

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

    // Terminate when download has finished
    std::lock_guard<std::mutex> completed_lck(completed_mutex);
    if (completed) {
      std::cout << "Progress bar thread has ended\n";
      break;
    }
  }
}

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

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

  while (!completed) {
    completed_lck.unlock();
    std::this_thread::sleep_for(10ms);
    completed_lck.lock();
  }

  completed_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 processor(process_data);

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

Implementation with Mutex

• This is not ideal

  • Too many loops
  • Too much explicit locking and unlocking
  • How do we choose the sleep duration?

• Better solution

  • Thread A indicates that it is waiting for something
  • Thread B does the "something"
  • Thread A is woken up and resumes