multithreading

Make espeakup a multi-threaded program.  One thread reads from the softsynth
device, queuing text and synthesis commands.  The other thread processes
items from the queue.

Makefile

@@ -9,7 +9,7 @@ SRCS = \
 
 OBJS = $(SRCS:.c=.o)
 
-LDLIBS = -lespeak
+LDLIBS = -lespeak -lpthread
 
 PREFIX = /usr
 MANDIR = $(PREFIX)/share/man/man8

espeakup.c

@@ -23,6 +23,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
+#include <pthread.h>
 
 #include "espeakup.h"
 
@@ -90,10 +91,13 @@ void espeakup_sighandler(int sig)
 
 int main(int argc, char **argv)
 {
+pthread_t queue_thread_id;
 	struct synth_t s = {
 		.voice = "",
 	};
 
+	/* Spawn our queue-processing thread. */
+
 	/* process command line options */
 	process_cli(argc, argv);
 
@@ -114,6 +118,11 @@ int main(int argc, char **argv)
 		}
 	}
 
+	int err = pthread_create(&queue_thread_id, NULL, &queue_runner, &s);
+	if (err != 0) {
+		return 4;
+	}
+
 	/* open the softsynth. */
 	open_softsynth();
 

espeakup.h

@@ -61,7 +61,6 @@ extern void queue_clear(void);
 extern void queue_add_cmd(enum command_t cmd, enum adjust_t adj,
 						  int value);
 extern void queue_add_text(char *txt, size_t length);
-extern void queue_process_entry(struct synth_t *s);
 extern espeak_ERROR set_frequency(struct synth_t *s, int freq,
 								  enum adjust_t adj);
 extern espeak_ERROR set_pitch(struct synth_t *s, int pitch,
@@ -78,5 +77,6 @@ extern espeak_ERROR speak_text(struct synth_t *s);
 extern void open_softsynth(void);
 extern void close_softsynth(void);
 extern void main_loop(struct synth_t *s);
+extern void * queue_runner(void *arg);
 
 #endif

queue.c

@@ -21,9 +21,13 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <pthread.h>
 
 #include "espeakup.h"
 
+pthread_cond_t runner_awake = PTHREAD_COND_INITIALIZER;
+pthread_mutex_t queue_guard = PTHREAD_MUTEX_INITIALIZER;
+
 struct queue_entry_t {
 	enum command_t cmd;
 	enum adjust_t adjust;
@@ -38,6 +42,7 @@ static struct queue_entry_t *last = NULL;
 
 static void queue_add(struct queue_entry_t *entry)
 {
+	pthread_mutex_lock(&queue_guard);
 	assert(entry);
 	entry->next = NULL;
 	if (!last)
@@ -48,26 +53,45 @@ static void queue_add(struct queue_entry_t *entry)
 		first->next = entry;
 		first = first->next;
 	}
+	pthread_mutex_unlock(&queue_guard);
+	pthread_cond_signal(&runner_awake);
 }
 
-static void queue_remove(void)
+static void free_entry(struct queue_entry_t *entry)
 {
-	struct queue_entry_t *temp;
+	if (entry->cmd == CMD_SPEAK_TEXT)
+		free(entry->buf);
+	free(entry);
+}
 
-	assert(last);
-	temp = last;
-	last = temp->next;
-	if (temp->cmd == CMD_SPEAK_TEXT)
-		free(temp->buf);
-	free(temp);
-	if (!last)
-		first = last;
+/* Remove and return the entry at the head of the queue.
+ * Return NULL if queue is empty. */
+
+static struct queue_entry_t *queue_remove(void)
+{
+	struct queue_entry_t *temp = NULL;
+
+	if(last) {
+		temp = last;
+		last = temp->next;
+
+		if (!last)
+			first = last;
+	}
+
+	return temp;
 }
 
 void queue_clear(void)
 {
-	while (last)
-		queue_remove();
+	pthread_mutex_lock(&queue_guard);
+	while (last) {
+		struct queue_entry_t *entry = queue_remove();
+		if(entry)
+			free_entry(entry);
+	}
+	pthread_mutex_unlock(&queue_guard);
+	/* We aren't adding data to the queue, so no need to signal. */
 }
 
 void queue_add_cmd(enum command_t cmd, enum adjust_t adj, int value)
@@ -106,39 +130,76 @@ void queue_add_text(char *txt, size_t length)
 	queue_add(entry);
 }
 
-void queue_process_entry(struct synth_t *s)
+static void queue_process_entry(struct synth_t *s)
 {
 	espeak_ERROR error;
+	struct queue_entry_t *current = queue_remove();
 
-	if (!last)
-		return;
+	pthread_mutex_unlock(&queue_guard); /* So "reader" can go. */
 
-	switch (last->cmd) {
-	case CMD_SET_FREQUENCY:
-		error = set_frequency(s, last->value, last->adjust);
-		break;
-	case CMD_SET_PITCH:
-		error = set_pitch(s, last->value, last->adjust);
-		break;
-	case CMD_SET_PUNCTUATION:
-		error = set_punctuation(s, last->value, last->adjust);
-		break;
-	case CMD_SET_RATE:
-		error = set_rate(s, last->value, last->adjust);
-		break;
-	case CMD_SET_VOICE:
-		break;
-	case CMD_SET_VOLUME:
-		error = set_volume(s, last->value, last->adjust);
-		break;
-	case CMD_SPEAK_TEXT:
-		s->buf = last->buf;
-		s->len = last->len;
-		error = speak_text(s);
-		break;
-	default:
-		break;
+	if(current) {
+		switch (current->cmd) {
+		case CMD_SET_FREQUENCY:
+			error = set_frequency(s, current->value, current->adjust);
+			break;
+		case CMD_SET_PITCH:
+			error = set_pitch(s, current->value, current->adjust);
+			break;
+		case CMD_SET_PUNCTUATION:
+			error = set_punctuation(s, current->value, current->adjust);
+			break;
+		case CMD_SET_RATE:
+			error = set_rate(s, current->value, current->adjust);
+			break;
+		case CMD_SET_VOICE:
+			break;
+		case CMD_SET_VOLUME:
+			error = set_volume(s, current->value, current->adjust);
+			break;
+		case CMD_SPEAK_TEXT:
+			s->buf = current->buf;
+			s->len = current->len;
+			error = speak_text(s);
+			break;
+		default:
+			break;
+		}
+
+		free_entry(current);
 	}
-	if (error == EE_OK)
-		queue_remove();
+}
+
+/* queue_runner is the "main" function of our secondary (queue-processing)
+ * thread.
+ * First, lock queue_guard, because it needs to be locked when we call
+ * pthread_cond_wait on the runner_awake condition variable.
+ * Next, enter an infinite loop.
+ * The wait call also unlocks queue_guard, so that the other thread can
+ * manipulate the queue.
+ * When runner_awake is signaled, the pthread_cond_wait call re-locks
+ * queue_guard, and the "queue processor" thread has access to the queue.
+ * While there is an entry in the queue, call queue_process_entry.
+ * queue_process_entry unlocks queue_guard after removing an item from the
+ * queue, so that the main thread doesn't have to wait for us to finish
+ * processing the entry.  So re-lock queue_guard after each call to
+ * queue_process_entry.
+ *
+ * The main thread can add items to the queue in exactly two situations:
+ * 1. We are waiting on runner_awake, or
+ * 2. We are processing an entry that has just been removed from the queue.
+*/
+
+void *queue_runner(void *arg) {
+	struct synth_t *synth = (struct synth_t *) arg;
+		pthread_mutex_lock(&queue_guard);
+	while(1) {
+		pthread_cond_wait(&runner_awake, &queue_guard);
+
+		while(last) {
+			queue_process_entry(synth);
+			pthread_mutex_lock(&queue_guard);
+		}
+	}
+
+	return NULL;
 }

softsynth.c

@@ -142,19 +142,15 @@ void close_softsynth(void)
 void main_loop(struct synth_t *s)
 {
 	fd_set set;
-	struct timeval tv;
 	ssize_t length;
 	char buf[maxBufferSize];
 	char *cp;
 
 	while (1) {
-		queue_process_entry(s);
 
 		FD_ZERO(&set);
 		FD_SET(softFD, &set);
-		tv.tv_sec = 0;
-		tv.tv_usec = 500;
-		if (select(softFD + 1, &set, NULL, NULL, &tv) < 0) {
+		if (select(softFD + 1, &set, NULL, NULL, NULL) < 0) {
 			if (errno == EINTR)
 				continue;
 			perror("Select failed");
@@ -182,3 +178,4 @@ void main_loop(struct synth_t *s)
 		process_buffer(s, buf, length);
 	}
 }
+