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;
}