rework server-worker communication

OK, this is a major rework.

* tcp_server: connection threads are not detached anymore, the caller has
  to clean them up. This was done so that the server can clean up the
  threads cleanly.
* run_queue: simple refactoring, run_queue_entry is called just run now.
* server: worker threads are now killed when a run is assigned to a
  worker.
* worker: the connection to server is no longer persistent. A worker
  sends "new-worker", waits for a task, closes the connection, and when
  it's done, sends the "complete" message and waits for a new task.

This is supposed to improve resilience, since the worker-server
connections don't have to be maintained while the worker is doing a CI
run.

1 files changed, 31 insertions, 16 deletions

diff --git a/src/run_queue.c b/src/run_queue.c
index 8b5052b..0455c92 100644
--- a/src/run_queue.c
+++ b/src/run_queue.c
@@ -7,20 +7,21 @@
 
 #include "run_queue.h"
 #include "log.h"
+#include "msg.h"
 
 #include <stdlib.h>
 #include <string.h>
 #include <sys/queue.h>
 
-struct run_queue_entry {
+struct run {
 	char *url;
 	char *rev;
-	STAILQ_ENTRY(run_queue_entry) entries;
+	STAILQ_ENTRY(run) entries;
 };
 
-int run_queue_entry_create(struct run_queue_entry **_entry, const char *_url, const char *_rev)
+int run_create(struct run **_entry, const char *_url, const char *_rev)
 {
-	struct run_queue_entry *entry = malloc(sizeof(struct run_queue_entry));
+	struct run *entry = malloc(sizeof(struct run));
 	if (!entry) {
 		log_errno("malloc");
 		goto fail;
@@ -54,19 +55,33 @@ fail:
 	return -1;
 }
 
-void run_queue_entry_destroy(struct run_queue_entry *entry)
+int run_from_msg(struct run **run, const struct msg *msg)
+{
+	size_t msg_len = msg_get_length(msg);
+
+	if (msg_len != 3) {
+		log_err("Invalid number of arguments for a message: %zu\n", msg_len);
+		msg_dump(msg);
+		return -1;
+	}
+
+	const char **words = msg_get_words(msg);
+	return run_create(run, words[1], words[2]);
+}
+
+void run_destroy(struct run *entry)
 {
 	free(entry->rev);
 	free(entry->url);
 	free(entry);
 }
 
-const char *run_queue_entry_get_url(const struct run_queue_entry *entry)
+const char *run_get_url(const struct run *entry)
 {
 	return entry->url;
 }
 
-const char *run_queue_entry_get_rev(const struct run_queue_entry *entry)
+const char *run_get_rev(const struct run *entry)
 {
 	return entry->rev;
 }
@@ -78,10 +93,10 @@ void run_queue_create(struct run_queue *queue)
 
 void run_queue_destroy(struct run_queue *queue)
 {
-	struct run_queue_entry *entry1 = STAILQ_FIRST(queue);
+	struct run *entry1 = STAILQ_FIRST(queue);
 	while (entry1) {
-		struct run_queue_entry *entry2 = STAILQ_NEXT(entry1, entries);
-		run_queue_entry_destroy(entry1);
+		struct run *entry2 = STAILQ_NEXT(entry1, entries);
+		run_destroy(entry1);
 		entry1 = entry2;
 	}
 	STAILQ_INIT(queue);
@@ -92,19 +107,19 @@ int run_queue_is_empty(const struct run_queue *queue)
 	return STAILQ_EMPTY(queue);
 }
 
-void run_queue_add_last(struct run_queue *queue, struct run_queue_entry *entry)
+void run_queue_add_first(struct run_queue *queue, struct run *entry)
 {
-	STAILQ_INSERT_TAIL(queue, entry, entries);
+	STAILQ_INSERT_HEAD(queue, entry, entries);
 }
 
-void run_queue_add_first(struct run_queue *queue, struct run_queue_entry *entry)
+void run_queue_add_last(struct run_queue *queue, struct run *entry)
 {
-	STAILQ_INSERT_HEAD(queue, entry, entries);
+	STAILQ_INSERT_TAIL(queue, entry, entries);
 }
 
-struct run_queue_entry *run_queue_remove_first(struct run_queue *queue)
+struct run *run_queue_remove_first(struct run_queue *queue)
 {
-	struct run_queue_entry *entry = STAILQ_FIRST(queue);
+	struct run *entry = STAILQ_FIRST(queue);
 	STAILQ_REMOVE_HEAD(queue, entries);
 	return entry;
 }