Prepare the worker to be multithreaded

The worker will have two threads: one doing the networking, and one
doing the computation.  This commit doesn't actually start the computation
thread yet, but it gets the scaffolding ready.

Makefile

@@ -52,7 +52,7 @@ dpnode: dpnode.o evutils.o
 	g++ -g -Wall $^ -o $@ -L$(LIBEVENT)/lib -levent -lntl -lgmp
 
 worker: worker.o evutils.o
-	g++ -g -Wall $^ -o $@ -L$(LIBEVENT)/lib -levent -lntl -lgmp
+	g++ -g -Wall $^ -o $@ -L$(LIBEVENT)/lib -levent -lntl -lgmp -lpthread
 
 clean:
 	-rm -f $(OFILES)

worker.cc

@@ -5,6 +5,7 @@ extern "C" {
 }
 
 #include <vector>
+#include <pthread.h>
 #include <stdio.h>
 
 #include "evutils.h"
@@ -19,21 +20,55 @@ typedef enum {
 
 struct WrkControllerConnInfo {
     WrkCCState state;
-    unsigned char subproblem[SUBPROBLEM_DESC_LEN];
     unsigned short num_dpnodes;
 
-    WrkControllerConnInfo() : state(WRKCCSTATE_AWAITCMD) {}
+    WrkControllerConnInfo() : state(WRKCCSTATE_AWAITCMD), num_dpnodes(0) {}
 };
 
+typedef enum { WT_NOT_RUNNING, WT_RUNNING, WT_SHOULD_STOP} WTState;
+
 static struct WrkControllerState {
     Subproblem *current_problem;
     unsigned short num_expected_dpnodes;
     vector<struct bufferevent *> dpnodes;
     unsigned short num_connected_dpnodes;
+    WTState worker_thread_state;
+    pthread_t worker_thread;
 
-    WrkControllerState(): current_problem(NULL) {}
+    WrkControllerState(): current_problem(NULL),
+	worker_thread_state(WT_NOT_RUNNING) {}
 } wrkctrlstate;
 
+static void stop_working(void)
+{
+    cerr << "Stopping work\n";
+    if (wrkctrlstate.worker_thread_state != WT_NOT_RUNNING) {
+	// Tell the worker thread to stop after its next kernel launch
+	wrkctrlstate.worker_thread_state = WT_SHOULD_STOP;
+	pthread_join(wrkctrlstate.worker_thread, NULL);
+	wrkctrlstate.worker_thread_state = WT_NOT_RUNNING;
+    }
+
+    // Close the connections to the dpnodes
+    vector<struct bufferevent *>::iterator bevit;
+    for (bevit = wrkctrlstate.dpnodes.begin();
+	    bevit != wrkctrlstate.dpnodes.end(); ++bevit) {
+	bufferevent_free(*bevit);
+    }
+    wrkctrlstate.dpnodes.clear();
+    wrkctrlstate.num_connected_dpnodes = 0;
+    wrkctrlstate.num_expected_dpnodes = 0;
+    // Careful!  Subproblem uses NTL, so we must be sure we're not
+    // multithreaded at this point.
+    delete wrkctrlstate.current_problem;
+    wrkctrlstate.current_problem = NULL;
+}
+
+static void start_working(void)
+{
+    cerr << "Starting work\n";
+}
+
 static void dpconn_event_cb(struct bufferevent *bev, short events,
     void *ctx)
 {
@@ -42,14 +77,12 @@ static void dpconn_event_cb(struct bufferevent *bev, short events,
 	++wrkctrlstate.num_connected_dpnodes;
 	if (wrkctrlstate.num_connected_dpnodes ==
 		wrkctrlstate.num_expected_dpnodes) {
-	    cerr << "Starting work\n";
-	    // start_working();
+	    start_working();
 	}
     } else if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
-	fprintf(stderr, "Closing connection to controller and restarting\n");
+	fprintf(stderr, "Closing connection to dpnode\n");
 	bufferevent_free(bev);
-	cerr << "Stopping work\n";
-	// stop_working();
+	stop_working();
     }
 }
 
@@ -58,6 +91,7 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
     struct evbuffer *input = bufferevent_get_input(bev);
     WrkControllerConnInfo *info = (WrkControllerConnInfo *)ctx;
     unsigned char cmd[1];
+    unsigned char subproblem[SUBPROBLEM_DESC_LEN];
 
     while(1) {
 	size_t len = evbuffer_get_length(input);
@@ -70,7 +104,7 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 			info->state = WRKCCSTATE_RDPROBLEM;
 			break;
 		    case 'S':
-			// stop_working();
+			stop_working();
 			break;
 		    default:
 			/* Unknown command received */
@@ -83,9 +117,12 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 
 	    case WRKCCSTATE_RDPROBLEM:
 		if (len < SUBPROBLEM_DESC_LEN+2) return;
-		// stop_working();
-		bufferevent_read(bev, info->subproblem, SUBPROBLEM_DESC_LEN);
+		stop_working();
+		bufferevent_read(bev, subproblem, SUBPROBLEM_DESC_LEN);
 		bufferevent_read(bev, &(info->num_dpnodes), 2);
+		// Careful!  Subproblem uses NTL, so we must be sure
+		// we're not multithreaded at this point.
+		wrkctrlstate.current_problem = new Subproblem(subproblem);
 		info->state = WRKCCSTATE_RDDPNODES;
 		/* FALLTHROUGH */
 
@@ -97,7 +134,6 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 		    for(i=0;i<info->num_dpnodes;++i) {
 			unsigned char ipport[6];
 			bufferevent_read(bev, ipport, 6);
-			// XXX: Start a connection to this DPnode
 			cerr << "Connecting to DPnode " <<
 				    int(ipport[0]) << "." <<
 				    int(ipport[1]) << "." <<
@@ -111,7 +147,7 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 			if (dpbev) {
 			    wrkctrlstate.dpnodes.push_back(dpbev);
 			} else {
-			    // stop_working();
+			    stop_working();
 			}
 		    }
 		}