From 63027dd857a47415099d12d2ae21951a155350d3 Mon Sep 17 00:00:00 2001 From: Bassem Girgis Date: Wed, 17 Oct 2018 11:12:34 -0500 Subject: [PATCH] Adding cpp mocks --- .cproject | 106 +++++++++++++ .project | 26 ++++ .settings/language.settings.xml | 25 +++ Makefile | 8 +- alarm_mutex.cpp | 127 +++++++++++++++ lifecycle.c | 4 + pipe.cpp | 268 ++++++++++++++++++++++++++++++++ trylock.c | 139 ++++++++--------- 8 files changed, 623 insertions(+), 80 deletions(-) create mode 100644 .cproject create mode 100644 .project create mode 100644 .settings/language.settings.xml create mode 100644 alarm_mutex.cpp create mode 100644 pipe.cpp diff --git a/.cproject b/.cproject new file mode 100644 index 0000000..93945a4 --- /dev/null +++ b/.cproject @@ -0,0 +1,106 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/.project b/.project new file mode 100644 index 0000000..9bc0a2c --- /dev/null +++ b/.project @@ -0,0 +1,26 @@ + + + Programming-POSIX-Threads + + + + + + org.eclipse.cdt.managedbuilder.core.genmakebuilder + clean,full,incremental, + + + + + org.eclipse.cdt.managedbuilder.core.ScannerConfigBuilder + full,incremental, + + + + + + org.eclipse.cdt.core.cnature + org.eclipse.cdt.managedbuilder.core.managedBuildNature + org.eclipse.cdt.managedbuilder.core.ScannerConfigNature + + diff --git a/.settings/language.settings.xml b/.settings/language.settings.xml new file mode 100644 index 0000000..ea24223 --- /dev/null +++ b/.settings/language.settings.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/Makefile b/Makefile index 485587b..b2cc03c 100644 --- a/Makefile +++ b/Makefile @@ -1,11 +1,15 @@ # Digital UNIX 4.0 compilation flags: -CFLAGS=-std1 -pthread -g -w1 $(DEBUGFLAGS) -RTFLAGS=-lrt +#CFLAGS=-std1 -pthread -g -w1 $(DEBUGFLAGS) +#RTFLAGS=-lrt # Solaris 2.5 compilation flags: #CFLAGS=-D_POSIX_C_SOURCE=199506 -D_REENTRANT -Xa -lpthread -g $(DEBUGFLAGS) #RTFLAGS=-lposix4 +# Linux compilation flags: +CFLAGS=-pthread -O3 $(DEBUGFLAGS) +RTFLAGS=-lrt + SOURCES=alarm.c alarm_cond.c alarm_fork.c alarm_mutex.c \ alarm_thread.c atfork.c backoff.c \ barrier_main.c cancel.c cancel_async.c cancel_cleanup\ diff --git a/alarm_mutex.cpp b/alarm_mutex.cpp new file mode 100644 index 0000000..a6222b1 --- /dev/null +++ b/alarm_mutex.cpp @@ -0,0 +1,127 @@ +/* + * alarm_mutex.c + * + * This is an enhancement to the alarm_thread.c program, which + * created an "alarm thread" for each alarm command. This new + * version uses a single alarm thread, which reads the next + * entry in a list. The main thread places new requests onto the + * list, in order of absolute expiration time. The list is + * protected by a mutex, and the alarm thread sleeps for at + * 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 + +#define MIN(X, Y) (((X) < (Y)) ? (X) : (Y)) + +/* + * 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 { + int seconds; + time_t time; /* seconds from EPOCH */ + char message[64]; +}; + +using alarm_t = alarm_tag; + +pthread_mutex_t alarm_mutex = PTHREAD_MUTEX_INITIALIZER; +std::list alarm_list; + +/* + * The alarm thread's start routine. + */ +void *alarm_thread(void *arg) { + 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"); + + now = time(NULL); + sleep_time = 0; + + 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; + } + } + + /* + * 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); + else + sched_yield(); + } +} + +int main(int argc, char *argv[]) { + int status; + char line[128]; + pthread_t thread; + + 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) exit(0); + if (strlen(line) <= 1) continue; + 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"); + } else { + alarm.time = time(NULL) + alarm.seconds; + + status = pthread_mutex_lock(&alarm_mutex); + if (status != 0) err_abort(status, "Lock 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"); + } + } +} diff --git a/lifecycle.c b/lifecycle.c index 8b99120..78f772a 100644 --- a/lifecycle.c +++ b/lifecycle.c @@ -12,6 +12,7 @@ */ void *thread_routine (void *arg) { + printf("Inside thread %i\n", pthread_self()); return arg; } @@ -21,10 +22,13 @@ main (int argc, char *argv[]) void *thread_result; int status; + printf("Inside thread %i\n", pthread_self()); status = pthread_create ( &thread_id, NULL, thread_routine, NULL); if (status != 0) err_abort (status, "Create thread"); + sleep(1.0); + printf("Main thread created thread %i\n", thread_id); status = pthread_join (thread_id, &thread_result); if (status != 0) diff --git a/pipe.cpp b/pipe.cpp new file mode 100644 index 0000000..338b445 --- /dev/null +++ b/pipe.cpp @@ -0,0 +1,268 @@ +/* + * pipe.c + * + * Simple demonstration of a pipeline. main() is a loop that + * feeds the pipeline with integer values. Each stage of the + * pipeline increases the integer by one before passing it along + * to the next. Entering the command "=" reads the pipeline + * result. (Notice that too many '=' commands will hang.) + */ +#include +#include "errors.h" + +#include +#include +#include + +/* + * Internal structure describing a "stage" in the + * pipeline. One for each thread, plus a "result + * stage" where the final thread can stash the value. + */ +struct stage_tag { + pthread_mutex_t mutex; /* Protect data */ + pthread_cond_t dataIsAvail; /* Data available */ + pthread_cond_t threadIsIdle; /* Ready for data */ + bool isDataAvail; /* Data present */ + long data; /* Data to process */ + pthread_t thread; /* Thread for stage */ + stage_tag *next; +}; + +using stage_t = stage_tag; + +using StageList = std::list; + +/* + * External structure representing the entire + * pipeline. + */ +struct pipe_tag { + pthread_mutex_t mutex; /* Mutex to protect pipe */ + StageList stageList; /* stages list */ + int nActive; /* Active data elements */ +}; + +using pipe_t = pipe_tag; + +/* + * Internal function to send a "message" to the + * specified pipe stage. Threads use this to pass + * along the modified data item. + */ +int pipe_send(stage_t &stage, long data) { + int status; + + status = pthread_mutex_lock(&stage.mutex); + if (status != 0) return status; + + /* + * If there's data in the pipe stage, wait for it + * to be consumed. + */ + while (stage.isDataAvail) { + status = pthread_cond_wait(&stage.threadIsIdle, &stage.mutex); + if (status != 0) { + pthread_mutex_unlock(&stage.mutex); + return status; + } + } + + /* + * Send the new data + */ + stage.data = data; + stage.isDataAvail = true; + + status = pthread_cond_signal(&stage.dataIsAvail); + if (status != 0) { + pthread_mutex_unlock(&stage.mutex); + return status; + } + + status = pthread_mutex_unlock(&stage.mutex); + return status; +} + +/* + * The thread start routine for pipe stage threads. + * Each will wait for a data item passed from the + * caller or the previous stage, modify the data + * and pass it along to the next (or final) stage. + */ +void *pipe_stage(void *arg) { + stage_t &stage = *(stage_t *)arg; + stage_t &nextStage = *stage.next; + int status; + + status = pthread_mutex_lock(&stage.mutex); + if (status != 0) err_abort(status, "Lock pipe stage"); + + while (1) { + while (!stage.isDataAvail) { + status = pthread_cond_wait(&stage.dataIsAvail, &stage.mutex); + if (status != 0) err_abort(status, "Wait for previous stage"); + } + + pipe_send(nextStage, stage.data + 1); + + stage.isDataAvail = false; + + status = pthread_cond_signal(&stage.threadIsIdle); + if (status != 0) err_abort(status, "Wake next stage"); + } + + /* + * Notice that the routine never unlocks the stage->mutex. + * The call to pthread_cond_wait implicitly unlocks the + * mutex while the thread is waiting, allowing other threads + * to make progress. Because the loop never terminates, this + * function has no need to unlock the mutex explicitly. + */ +} + +/* + * External interface to create a pipeline. All the + * data is initialized and the threads created. They'll + * wait for data. + */ +int pipe_create(pipe_t &pipe, size_t nStages) { + int status = pthread_mutex_init(&pipe.mutex, NULL); + if (status != 0) err_abort(status, "Init pipe mutex"); + + pipe.stageList.resize(nStages); + pipe.nActive = 0; + + for (auto &iStage : pipe.stageList) { + status = pthread_mutex_init(&iStage.mutex, NULL); + if (status != 0) err_abort(status, "Init stage mutex"); + status = pthread_cond_init(&iStage.dataIsAvail, NULL); + if (status != 0) err_abort(status, "Init dataIsAvail condition"); + status = pthread_cond_init(&iStage.threadIsIdle, NULL); + if (status != 0) err_abort(status, "Init threadIsIdle condition"); + iStage.isDataAvail = false; + } + + /* + * Create the threads for the pipe stages only after all + * the data is initialized (including all links). Note + * that the last stage doesn't get a thread, it's just + * a receptacle for the final pipeline value. + * + * At this point, proper cleanup on an error would take up + * more space than worthwhile in a "simple example", so + * instead of cancelling and detaching all the threads + * already created, plus the synchronization object and + * memory cleanup done for earlier errors, it will simply + * abort. + */ + for (auto it = pipe.stageList.begin(), ite = --pipe.stageList.end(); + it != ite;) { + auto iit = it++; + iit->next = &(*it); + + status = pthread_create(&iit->thread, NULL, pipe_stage, (void *)&*iit); + if (status != 0) err_abort(status, "Create pipe stage"); + } + return 0; +} + +/* + * Collect the result of the pipeline. Wait for a + * result if the pipeline hasn't produced one. + */ +int pipe_result(pipe_t &pipe) { + int status; + + status = pthread_mutex_lock(&pipe.mutex); + if (status != 0) err_abort(status, "Lock pipe mutex"); + + bool isEmpty = false; + if (pipe.nActive <= 0) + isEmpty = true; + else + pipe.nActive--; + + status = pthread_mutex_unlock(&pipe.mutex); + if (status != 0) err_abort(status, "Unlock pipe mutex"); + + if (isEmpty) { + printf("Pipe is empty\n"); + return 0; + } + + auto &tail = pipe.stageList.back(); + + status = pthread_mutex_lock(&tail.mutex); + if (status != 0) err_abort(status, "Lock pipe tail mutex"); + + while (!tail.isDataAvail) { + pthread_cond_wait(&tail.dataIsAvail, &tail.mutex); + } + + long result = tail.data; + tail.isDataAvail = false; + + status = pthread_cond_signal(&tail.threadIsIdle); + if (status != 0) err_abort(status, "Signal pipe tail threadIsIdle"); + + status = pthread_mutex_unlock(&tail.mutex); + if (status != 0) err_abort(status, "Unlock pipe tail mutex"); + + printf("Result is %ld\n", result); + + return 1; +} + +/* + * External interface to start a pipeline by passing + * data to the first stage. The routine returns while + * the pipeline processes in parallel. Call the + * pipe_result return to collect the final stage values + * (note that the pipe will stall when each stage fills, + * until the result is collected). + */ +void pipe_start(pipe_t &pipe, long value) { + int status = pthread_mutex_lock(&pipe.mutex); + if (status != 0) err_abort(status, "Lock pipe mutex"); + + if (pipe.nActive > pipe.stageList.size() - 1) { + status = pthread_mutex_unlock(&pipe.mutex); + pipe_result(pipe); + status = pthread_mutex_lock(&pipe.mutex); + } + + pipe.nActive++; + + status = pthread_mutex_unlock(&pipe.mutex); + if (status != 0) err_abort(status, "Unlock pipe mutex"); + + pipe_send(pipe.stageList.front(), value); +} + +/* + * The main program to "drive" the pipeline... + */ +int main(int argc, char *argv[]) { + pipe_t my_pipe; + pipe_create(my_pipe, 2); + + printf("Enter integer values, or \"=\" for next result\n"); + + char line[128]; + while (1) { + printf("Data> "); + if (fgets(line, sizeof(line), stdin) == NULL) exit(0); + printf(line); + if (strlen(line) <= 1) continue; + if (strlen(line) <= 2 && line[0] == '=') { + pipe_result(my_pipe); + } else { + long value; + if (sscanf(line, "%ld", &value) < 1) + fprintf(stderr, "Enter an integer value\n"); + else + pipe_start(my_pipe, value); + } + } +} diff --git a/trylock.c b/trylock.c index 8d8860a..4d2421f 100644 --- a/trylock.c +++ b/trylock.c @@ -24,31 +24,27 @@ time_t end_time; * Thread start routine that repeatedly locks a mutex and * increments a counter. */ -void *counter_thread (void *arg) -{ - int status; - int spin; +void *counter_thread(void *arg) { + int status; + int spin; - /* - * Until end_time, increment the counter each - * second. Instead of just incrementing the counter, it - * sleeps for another second with the mutex locked, to give - * monitor_thread a reasonable chance of running. - */ - while (time (NULL) < end_time) - { - status = pthread_mutex_lock (&mutex); - if (status != 0) - err_abort (status, "Lock mutex"); - for (spin = 0; spin < SPIN; spin++) - counter++; - status = pthread_mutex_unlock (&mutex); - if (status != 0) - err_abort (status, "Unlock mutex"); - sleep (1); - } - printf ("Counter is %#lx\n", counter); - return NULL; + /* + * Until end_time, increment the counter each + * second. Instead of just incrementing the counter, it + * sleeps for another second with the mutex locked, to give + * monitor_thread a reasonable chance of running. + */ + while (time(NULL) < end_time) { + status = pthread_mutex_lock(&mutex); + if (status != 0) err_abort(status, "Lock mutex"); + for (spin = 0; spin < SPIN; spin++) counter++; + sleep(4); + status = pthread_mutex_unlock(&mutex); + if (status != 0) err_abort(status, "Unlock mutex"); + sleep(1); + } + printf("Counter is %#lx\n", counter); + return NULL; } /* @@ -56,65 +52,52 @@ void *counter_thread (void *arg) * seconds, try to lock the mutex and read the counter. If the * trylock fails, skip this cycle. */ -void *monitor_thread (void *arg) -{ - int status; - int misses = 0; +void *monitor_thread(void *arg) { + int status; + int misses = 0; - - /* - * Loop until end_time, checking the counter every 3 - * seconds. - */ - while (time (NULL) < end_time) - { - sleep (3); - status = pthread_mutex_trylock (&mutex); - if (status != EBUSY) - { - if (status != 0) - err_abort (status, "Trylock mutex"); - printf ("Counter is %ld\n", counter/SPIN); - status = pthread_mutex_unlock (&mutex); - if (status != 0) - err_abort (status, "Unlock mutex"); - } else - misses++; /* Count "misses" on the lock */ - } - printf ("Monitor thread missed update %d times.\n", misses); - return NULL; + /* + * Loop until end_time, checking the counter every 3 + * seconds. + */ + while (time(NULL) < end_time) { + sleep(3); + status = pthread_mutex_trylock(&mutex); + if (status != EBUSY) { + if (status != 0) err_abort(status, "Trylock mutex"); + printf("Counter is %ld\n", counter / SPIN); + status = pthread_mutex_unlock(&mutex); + if (status != 0) err_abort(status, "Unlock mutex"); + } else + misses++; /* Count "misses" on the lock */ + } + printf("Monitor thread missed update %d times.\n", misses); + return NULL; } -int main (int argc, char *argv[]) -{ - int status; - pthread_t counter_thread_id; - pthread_t monitor_thread_id; +int main(int argc, char *argv[]) { + int status; + pthread_t counter_thread_id; + pthread_t monitor_thread_id; #ifdef sun - /* - * On Solaris 2.5, threads are not timesliced. To ensure - * that our threads can run concurrently, we need to - * increase the concurrency level to 2. - */ - DPRINTF (("Setting concurrency level to 2\n")); - thr_setconcurrency (2); + /* + * On Solaris 2.5, threads are not timesliced. To ensure + * that our threads can run concurrently, we need to + * increase the concurrency level to 2. + */ + DPRINTF(("Setting concurrency level to 2\n")); + thr_setconcurrency(2); #endif - end_time = time (NULL) + 60; /* Run for 1 minute */ - status = pthread_create ( - &counter_thread_id, NULL, counter_thread, NULL); - if (status != 0) - err_abort (status, "Create counter thread"); - status = pthread_create ( - &monitor_thread_id, NULL, monitor_thread, NULL); - if (status != 0) - err_abort (status, "Create monitor thread"); - status = pthread_join (counter_thread_id, NULL); - if (status != 0) - err_abort (status, "Join counter thread"); - status = pthread_join (monitor_thread_id, NULL); - if (status != 0) - err_abort (status, "Join monitor thread"); - return 0; + end_time = time(NULL) + 60; /* Run for 1 minute */ + status = pthread_create(&counter_thread_id, NULL, counter_thread, NULL); + if (status != 0) err_abort(status, "Create counter thread"); + status = pthread_create(&monitor_thread_id, NULL, monitor_thread, NULL); + if (status != 0) err_abort(status, "Create monitor thread"); + status = pthread_join(counter_thread_id, NULL); + if (status != 0) err_abort(status, "Join counter thread"); + status = pthread_join(monitor_thread_id, NULL); + if (status != 0) err_abort(status, "Join monitor thread"); + return 0; }