diff --git a/Makefile b/Makefile index 5b6172e..b915433 100644 --- a/Makefile +++ b/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=) diff --git a/src/ch03/alarm_cond_cpp.cpp b/src/ch03/alarm_cond_cpp.cpp new file mode 100644 index 0000000..7dfc89e --- /dev/null +++ b/src/ch03/alarm_cond_cpp.cpp @@ -0,0 +1,173 @@ + +//#include "errors.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using ClockType = std::chrono::high_resolution_clock; +using DurationType = std::chrono::duration; +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; + +std::mutex alarm_mutex; +std::condition_variable alarm_cond; +std::list 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 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 l(alarm_mutex); + + alarm_insert(std::move(alarm)); + } +} diff --git a/src/ch03/alarm_mutex_cpp.cpp b/src/ch03/alarm_mutex_cpp.cpp index 9cd8c95..70210e7 100644 --- a/src/ch03/alarm_mutex_cpp.cpp +++ b/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 -#include -#include "errors.h" +#include "errors.hpp" + +#include +#include +#include #include +#include +#include +#include +#include + +using ClockType = std::chrono::high_resolution_clock; +using DurationType = std::chrono::duration; #define MIN(X, Y) (((X) < (Y)) ? (X) : (Y)) @@ -26,117 +35,109 @@ * been on the list. */ struct alarm_tag { - int seconds; - time_t time; /* seconds from EPOCH */ - char message[64]; + unsigned seconds = -1; + ClockType::time_point time; + std::string message; }; using alarm_t = alarm_tag; -pthread_mutex_t alarm_mutex = PTHREAD_MUTEX_INITIALIZER; +std::mutex alarm_mutex; std::list alarm_list; /* * The alarm thread's start routine. */ -void* -alarm_thread(void* arg) +void +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) { - status = pthread_mutex_lock(&alarm_mutex); - if (status != 0) - err_abort(status, "Lock mutex"); + while (true) { + DurationType sleep_time(0.0); - now = time(NULL); - sleep_time = 0; + { + std::lock_guard l(alarm_mutex); - for (auto it = alarm_list.begin(); it != alarm_list.end();) { - if (it->time <= now) { - printf("(%d) %s\n", it->seconds, it->message); - it = alarm_list.erase(it); - } - else { - sleep_time = MIN(it->time - now, sleep_time); - ++it; + auto now = ClockType::now(); + + for (auto it = alarm_list.begin(); it != alarm_list.end();) { + if (it->time <= now) { + std::cout << it->message << std::endl; + it = alarm_list.erase(it); + } + else { + DurationType remaining = it->time - ClockType::now(); + sleep_time = MIN(remaining, sleep_time); + ++it; + } } + + /* + * Unlock the mutex before waiting, so that the main + * thread can lock it to insert a new alarm request. If + * the sleep_time is 0, then call sched_yield, giving + * the main thread a chance to run if it has been + * readied by user input, without delaying the message + * if there's no input. + */ } - /* - * Unlock the mutex before waiting, so that the main - * thread can lock it to insert a new alarm request. If - * the sleep_time is 0, then call sched_yield, giving - * the main thread a chance to run if it has been - * 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 > 0) - sleep(sleep_time); + if (sleep_time.count() > 0) + std::this_thread::sleep_for(sleep_time); else - sched_yield(); + std::this_thread::yield(); } } int main(int argc, char* argv[]) { - int status; - char line[128]; - pthread_t thread; + std::thread t(alarm_thread); + t.detach(); - status = pthread_create(&thread, NULL, alarm_thread, NULL); - if (status != 0) - err_abort(status, "Create alarm thread"); - while (1) { - printf("alarm> "); - if (fgets(line, sizeof(line), stdin) == NULL) + std::string line; + while (true) { + std::cout << "Alarm> "; + + std::getline(std::cin, line); + if (!std::cin.good()) exit(0); - if (strlen(line) <= 1) + + if (line.empty()) continue; + + std::istringstream in(line); + alarm_t alarm; - /* - * Parse input line into seconds (%d) and a message - * (%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"); + in >> alarm.seconds; + alarm.message = in.str(); + + if (alarm.seconds < 1 || alarm.message.empty()) { + std::cout << "Bad command\n"; + continue; } - else { - alarm.time = time(NULL) + alarm.seconds; - status = pthread_mutex_lock(&alarm_mutex); - if (status != 0) - err_abort(status, "Lock mutex"); + alarm.time = ClockType::now() + std::chrono::seconds(alarm.seconds); - /* - * Insert the new alarm into the list of alarms. - */ - alarm_list.push_back(alarm); + std::lock_guard l(alarm_mutex); + + /* + * Insert the new alarm into the list of alarms. + */ + alarm_list.push_back(alarm); #ifdef DEBUG - printf("[list: "); - for (const auto& iAlarm : alarm_list) - printf("%d(%d)[\"%s\"] ", - iAlarm.time, - 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 << "[list: "; + for (const auto& iAlarm : alarm_list) { + DurationType remaining = iAlarm.time - ClockType::now(); + std::cout << "[\"" << iAlarm.message << "\": " << remaining.count() + << "], "; } + std::cout << "]\n"; +#endif } } diff --git a/src/ch03/cond_cpp.cpp b/src/ch03/cond_cpp.cpp new file mode 100644 index 0000000..1cdb06d --- /dev/null +++ b/src/ch03/cond_cpp.cpp @@ -0,0 +1,82 @@ + +#include "errors.hpp" + +#include +#include +#include +#include +#include + +using ClockType = std::chrono::high_resolution_clock; +using DurationType = std::chrono::duration; + +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 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 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; +}