add condition variable examples

Makefile

@@ -75,6 +75,8 @@ ch02/lifecycle_cpp.cpp \
 ch03/backoff_cpp.cpp \
 ch03/trylock_cpp.cpp \
 ch03/alarm_mutex_cpp.cpp \
+ch03/cond_cpp.cpp \
+ch03/alarm_cond_cpp.cpp \
 ch04/pipe_cpp.cpp
 
 NAMES_CXX=$(SOURCES_CXX:.cpp=)

src/ch03/alarm_cond_cpp.cpp

@@ -0,0 +1,173 @@
+
+//#include "errors.hpp"
+
+#include <algorithm>
+#include <chrono>
+#include <condition_variable>
+#include <iostream>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <sstream>
+#include <string>
+#include <thread>
+
+using ClockType    = std::chrono::high_resolution_clock;
+using DurationType = std::chrono::duration<double>;
+using TimePoint    = ClockType::time_point;
+
+/*
+ * The "alarm" structure now contains the time_t (time since the
+ * Epoch, in seconds) for each alarm, so that they can be
+ * sorted. Storing the requested number of seconds would not be
+ * enough, since the "alarm thread" cannot tell how long it has
+ * been on the list.
+ */
+struct alarm_tag {
+  unsigned seconds = -1;
+  TimePoint time;
+  std::string message;
+};
+
+using alarm_t = alarm_tag;
+using alarm_p = std::unique_ptr<alarm_t>;
+
+std::mutex alarm_mutex;
+std::condition_variable alarm_cond;
+std::list<alarm_p> alarm_list;
+
+const TimePoint FarFuture =
+    ClockType::now() + std::chrono::hours(365 * 24 * 10);
+TimePoint current_alarm = FarFuture;
+
+/*
+ * Insert alarm entry on list, in order.
+ */
+void
+alarm_insert(alarm_p alarm)
+{
+  /*
+   * LOCKING PROTOCOL:
+   *
+   * This routine requires that the caller have locked the
+   * alarm_mutex!
+   */
+
+  auto alarm_ptr = alarm.get();
+
+  auto it = alarm_list.insert(std::find_if(alarm_list.begin(),
+                                           alarm_list.end(),
+                                           [&alarm_ptr](const auto& a) {
+                                             return a->time > alarm_ptr->time;
+                                           }),
+                              std::move(alarm));
+
+#if 1  // DEBUG
+  std::cout << "[list: ";
+  for (const auto& iAlarm : alarm_list) {
+    DurationType remaining = iAlarm->time - ClockType::now();
+    std::cout << "[\"" << iAlarm->message << "\": " << remaining.count()
+              << "], ";
+  }
+  std::cout << "]\n";
+#endif
+
+  /*
+   * Wake the alarm thread if it is not busy (that is, if
+   * current_alarm is 0, signifying that it's waiting for
+   * work), or if the new alarm comes before the one on
+   * which the alarm thread is waiting.
+   */
+  if ((*it)->time < current_alarm) {
+    current_alarm = (*it)->time;
+    alarm_cond.notify_one();
+  }
+}
+
+/*
+ * The alarm thread's start routine.
+ */
+void
+alarm_thread()
+{
+  /*
+   * Loop forever, processing commands. The alarm thread will
+   * be disintegrated when the process exits. Lock the mutex
+   * at the start -- it will be unlocked during condition
+   * waits, so the main thread can insert alarms.
+   */
+  std::unique_lock<std::mutex> lk(alarm_mutex);
+
+  while (true) {
+    /*
+     * If the alarm list is empty, wait until an alarm is
+     * added. Setting current_alarm to 0 informs the insert
+     * routine that the thread is not busy.
+     */
+    current_alarm = FarFuture;
+    alarm_cond.wait(lk, []() { return !alarm_list.empty(); });
+
+    auto now = ClockType::now();
+
+    bool expired = false;
+
+    auto it           = alarm_list.begin();
+    const auto& alarm = *it;
+
+    if (alarm->time > now) {
+#if 1  // def DEBUG
+      std::cout << "waiting for \"" << alarm->message
+                << "\":" << DurationType(alarm->time - now).count()
+                << std::endl;
+#endif
+
+      current_alarm = alarm->time;
+      if (alarm_cond.wait_until(lk, alarm->time) == std::cv_status::timeout) {
+        expired = true;
+      }
+    }
+    else
+      expired = true;
+    if (expired) {
+      std::cout << alarm->message << std::endl;
+      alarm_list.erase(it);
+    }
+  }
+}
+
+int
+main(int argc, char* argv[])
+{
+  std::thread t(alarm_thread);
+  t.detach();
+
+  std::string line;
+  while (true) {
+    std::cout << "Alarm> ";
+
+    std::getline(std::cin, line);
+    if (!std::cin.good())
+      exit(0);
+
+    if (line.empty())
+      continue;
+
+    std::istringstream in(line);
+
+    alarm_p alarm(new alarm_t());
+
+    in >> alarm->seconds;
+    alarm->message = in.str();
+
+    if (alarm->seconds < 1 || alarm->message.empty()) {
+      std::cout << "Bad command\n";
+      continue;
+    }
+
+    alarm->time = ClockType::now() + std::chrono::seconds(alarm->seconds);
+
+    std::lock_guard<std::mutex> l(alarm_mutex);
+
+    alarm_insert(std::move(alarm));
+  }
+}

src/ch03/alarm_mutex_cpp.cpp

@@ -10,11 +10,20 @@
  * least 1 second, each iteration, to ensure that the main
  * thread can lock the mutex to add new work to the list.
  */
-#include <pthread.h>
-#include <time.h>
-#include "errors.h"
 
+#include "errors.hpp"
+
+#include <algorithm>
+#include <chrono>
+#include <iostream>
 #include <list>
+#include <mutex>
+#include <sstream>
+#include <string>
+#include <thread>
+
+using ClockType    = std::chrono::high_resolution_clock;
+using DurationType = std::chrono::duration<double, std::milli>;
 
 #define MIN(X, Y) (((X) < (Y)) ? (X) : (Y))
 
@@ -26,45 +35,42 @@
  * been on the list.
  */
 struct alarm_tag {
-  int seconds;
+  unsigned seconds = -1;
-  time_t time; /* seconds from EPOCH */
+  ClockType::time_point time;
-  char message[64];
+  std::string message;
 };
 
 using alarm_t = alarm_tag;
 
-pthread_mutex_t alarm_mutex = PTHREAD_MUTEX_INITIALIZER;
+std::mutex alarm_mutex;
 std::list<alarm_t> alarm_list;
 
 /*
  * The alarm thread's start routine.
  */
-void*
+void
-alarm_thread(void* arg)
+alarm_thread()
 {
-  int sleep_time;
-  time_t now;
-  int status;
-
   /*
    * Loop forever, processing commands. The alarm thread will
    * be disintegrated when the process exits.
    */
-  while (1) {
+  while (true) {
-    status = pthread_mutex_lock(&alarm_mutex);
+    DurationType sleep_time(0.0);
-    if (status != 0)
-      err_abort(status, "Lock mutex");
 
-    now        = time(NULL);
+    {
-    sleep_time = 0;
+      std::lock_guard<std::mutex> l(alarm_mutex);
+
+      auto now = ClockType::now();
 
       for (auto it = alarm_list.begin(); it != alarm_list.end();) {
         if (it->time <= now) {
-        printf("(%d) %s\n", it->seconds, it->message);
+          std::cout << it->message << std::endl;
           it = alarm_list.erase(it);
         }
         else {
-        sleep_time = MIN(it->time - now, sleep_time);
+          DurationType remaining = it->time - ClockType::now();
+          sleep_time             = MIN(remaining, sleep_time);
           ++it;
         }
       }
@@ -77,48 +83,47 @@ alarm_thread(void* arg)
        * readied by user input, without delaying the message
        * if there's no input.
        */
-    status = pthread_mutex_unlock(&alarm_mutex);
+    }
-    if (status != 0)
+
-      err_abort(status, "Unlock mutex");
+    if (sleep_time.count() > 0)
-    if (sleep_time > 0)
+      std::this_thread::sleep_for(sleep_time);
-      sleep(sleep_time);
     else
-      sched_yield();
+      std::this_thread::yield();
   }
 }
 
 int
 main(int argc, char* argv[])
 {
-  int status;
+  std::thread t(alarm_thread);
-  char line[128];
+  t.detach();
-  pthread_t thread;
 
-  status = pthread_create(&thread, NULL, alarm_thread, NULL);
+  std::string line;
-  if (status != 0)
+  while (true) {
-    err_abort(status, "Create alarm thread");
+    std::cout << "Alarm> ";
-  while (1) {
+
-    printf("alarm> ");
+    std::getline(std::cin, line);
-    if (fgets(line, sizeof(line), stdin) == NULL)
+    if (!std::cin.good())
       exit(0);
-    if (strlen(line) <= 1)
+
+    if (line.empty())
       continue;
+
+    std::istringstream in(line);
+
     alarm_t alarm;
 
-    /*
+    in >> alarm.seconds;
-     * Parse input line into seconds (%d) and a message
+    alarm.message = in.str();
-     * (%64[^\n]), consisting of up to 64 characters
-     * separated from the seconds by whitespace.
-     */
-    if (sscanf(line, "%d %64[^\n]", &alarm.seconds, alarm.message) < 2) {
-      fprintf(stderr, "Bad command\n");
-    }
-    else {
-      alarm.time = time(NULL) + alarm.seconds;
 
-      status = pthread_mutex_lock(&alarm_mutex);
+    if (alarm.seconds < 1 || alarm.message.empty()) {
-      if (status != 0)
+      std::cout << "Bad command\n";
-        err_abort(status, "Lock mutex");
+      continue;
+    }
+
+    alarm.time = ClockType::now() + std::chrono::seconds(alarm.seconds);
+
+    std::lock_guard<std::mutex> l(alarm_mutex);
 
     /*
      * Insert the new alarm into the list of alarms.
@@ -126,17 +131,13 @@ main(int argc, char* argv[])
     alarm_list.push_back(alarm);
 
 #ifdef DEBUG
-      printf("[list: ");
+    std::cout << "[list: ";
-      for (const auto& iAlarm : alarm_list)
+    for (const auto& iAlarm : alarm_list) {
-        printf("%d(%d)[\"%s\"] ",
+      DurationType remaining = iAlarm.time - ClockType::now();
-               iAlarm.time,
+      std::cout << "[\"" << iAlarm.message << "\": " << remaining.count()
-               iAlarm.time - time(NULL),
+                << "], ";
-               iAlarm.message);
-      printf("]\n");
-#endif
-      status = pthread_mutex_unlock(&alarm_mutex);
-      if (status != 0)
-        err_abort(status, "Unlock mutex");
     }
+    std::cout << "]\n";
+#endif
   }
 }

src/ch03/cond_cpp.cpp

@@ -0,0 +1,82 @@
+
+#include "errors.hpp"
+
+#include <chrono>
+#include <condition_variable>
+#include <iostream>
+#include <mutex>
+#include <thread>
+
+using ClockType    = std::chrono::high_resolution_clock;
+using DurationType = std::chrono::duration<double, std::milli>;
+
+struct my_struct_tag {
+  std::mutex mutex;             /* Protects access to value */
+  std::condition_variable cond; /* Signals change to value */
+  int value = 0;                /* Access protected by mutex */
+};
+
+using my_struct_t = my_struct_tag;
+
+my_struct_t data;
+
+DurationType hibernation(1000.0); /* Default to 1 second */
+
+/*
+ * Thread start routine. It will set the main thread's predicate
+ * and signal the condition variable.
+ */
+void
+wait_thread()
+{
+  int status;
+
+  std::this_thread::sleep_for(hibernation);
+  std::lock_guard<std::mutex> lk(data.mutex);
+
+  data.value = 1; /* Set predicate */
+  data.cond.notify_one();
+}
+
+int
+main(int argc, char* argv[])
+{
+  /*
+   * If an argument is specified, interpret it as the number
+   * of seconds for wait_thread to sleep before signaling the
+   * condition variable.  You can play with this to see the
+   * condition wait below time out or wake normally.
+   */
+  if (argc > 1) {
+    int xHibernation = atoi(argv[1]);
+    hibernation      = DurationType(1000.0 * xHibernation);
+  }
+
+  /*
+   * Create wait_thread.
+   */
+  std::thread t(wait_thread);
+
+  /*
+   * Wait on the condition variable for 2 seconds, or until
+   * signaled by the wait_thread. Normally, wait_thread
+   * should signal. If you raise "hibernation" above 2
+   * seconds, it will time out.
+   */
+  auto timeout = ClockType::now() + std::chrono::seconds(2);
+
+  {
+    std::unique_lock<std::mutex> lk(data.mutex);
+
+    if (!data.cond.wait_until(lk, timeout, []() { return data.value != 0; })) {
+      std::cout << "Condition wait timed out.\n";
+    }
+
+    if (data.value != 0)
+      std::cout << "Condition was signaled.\n";
+  }
+
+  t.join();
+
+  return 0;
+}