Reconstruct answer and launch new problems

Compute and check the final answer, then launch a new subproblem (for
now, forever; this should change to a counter).

controller.cc

@@ -24,6 +24,8 @@ extern "C" {
 
 NTL_CLIENT
 
+#define DEBUG
+
 struct SubproblemProgress;
 
 typedef std::set<struct bufferevent *> BESet;
@@ -74,11 +76,23 @@ void vsppdump(const vector<SubproblemProgress> &spv, ostream &os);
 static struct ControllerState {
     ZZ rho;
     FactorDecomp p, q;
-    int working;
+    ZZ base, target;
+    bool working;
     vector<SubproblemProgress> subproblems_p, subproblems_q;
     Statuses dpnodes, workers;
-
-    ControllerState() : working(0) {}
+    unsigned int num_unsolved_subproblems;
+
+    ControllerState() : working(false), num_unsolved_subproblems(0) {}
+
+    // Reset the state for a new problem with the same modulus
+    void reset(void) {
+	base = 0;
+	target = 0;
+	working = false;
+	subproblems_p.clear();
+	subproblems_q.clear();
+	num_unsolved_subproblems = 0;
+    }
 
     // Dump the state for debug purposes
     void dump(ostream &os) const {
@@ -199,6 +213,10 @@ struct SubproblemProgress : Subproblem {
 	for (unsigned short i = 0; i < num_ipports; ++i) {
 	    bufferevent_write(bev, ipports[i].ipport, 6);
 	}
+#ifdef DEBUG
+	cerr << "Added worker " << bev << "to subproblem "
+		<< problemid << "\n";
+#endif
     }
 };
 
@@ -273,10 +291,8 @@ static void find_subproblem_for_worker(vector<SubproblemProgress> &spv)
 }
 
 // See if there are any idle DPnodes or workers we can put to use
-void schedule(void)
+static void schedule(void)
 {
-    cerr << "Before schedule:\n"; ctrlstate.dump(cerr);
-
     // Check the DPnodes
 
     // Iterate through the subproblems, looking for one that can use
@@ -308,7 +324,141 @@ void schedule(void)
 	find_subproblem_for_worker(ctrlstate.subproblems_q);
     }
 
-    cerr << "After schedule:\n"; ctrlstate.dump(cerr);
+    // cerr << "After schedule:\n"; ctrlstate.dump(cerr);
+}
+
+static ZZ computation_complete_p(const vector<SubproblemProgress> &v)
+{
+    ZZ curmodulus, curexp;
+    curmodulus = 2;
+    curexp = 0;
+
+    vector<SubproblemProgress>::const_iterator vit;
+    for(vit = v.begin(); vit != v.end(); ++vit) {
+	CRT(curexp, curmodulus, vit->solution, vit->order);
+    }
+
+    if (curexp < 0) {
+	curexp += curmodulus;
+    }
+
+    return curexp;
+}
+
+// All subproblems are solved.  Combine the results.
+static void computation_complete(void)
+{
+    ZZ exp_p = computation_complete_p(ctrlstate.subproblems_p);
+    ZZ exp_q = computation_complete_p(ctrlstate.subproblems_q);
+    ZZ pm1 = (ctrlstate.p.factor - 1)/2;
+    ZZ qm1 = (ctrlstate.q.factor - 1)/2;
+    CRT(exp_p, pm1, exp_q, qm1);
+    if (exp_p < 0) {
+	exp_p += pm1;
+    }
+    ZZ& expon = exp_p;
+
+    cout << "expon = " << expon << "\n";
+    ZZ base_exp = PowerMod(ctrlstate.base, expon, ctrlstate.rho);
+    if (base_exp == ctrlstate.target) {
+	cout << "CORRECT!\n";
+    } else {
+	cout << "INCORRECT:\nbase^exp = " << base_exp << "\n";
+	cout << "target   = " << ctrlstate.target << "\n";
+    }
+
+    ctrlstate.reset();
+}
+
+static unsigned short curproblemid = 0;
+
+// Take base and target mod f.factor, then decompose that into small
+// subproblems given our knowledge of the factors of phi(f.factor)
+static vector<SubproblemProgress> decomp(const ZZ_p &base, const ZZ_p &target,
+    const FactorDecomp &f)
+{
+    vector<SubproblemProgress> ret;
+
+    // Compute phi(factor)
+    const int fveclen = f.fvec.length();
+    ZZ phi = to_ZZ(2);
+    for (int i = 0; i < fveclen; ++i) {
+	phi *= f.fvec[i];
+    }
+
+    ZZ_p::init(f.factor);
+    for (int i = 0; i < fveclen; ++i) {
+	const ZZ& order = f.fvec[i];
+	ZZ quotient = phi / order;
+	ZZ_p subgroup_base = to_ZZ_p(rep(base));
+	subgroup_base = power(subgroup_base, quotient);
+	ZZ_p subgroup_target = to_ZZ_p(rep(target));
+	subgroup_target = power(subgroup_target, quotient);
+
+	if (subgroup_base == 1) {
+	    // The original base wasn't a generator of the whole group
+	    if (subgroup_target == 1) {
+		// But the target is in the subgroup.  Lucky us.
+		continue;
+	    } else {
+		ret.clear();
+		return ret;
+	    }
+	}
+
+	// By default, 1 in 1000 points are distinguihed points.  The
+	// number in the next line is 2^32/1000
+	unsigned int dpfreq = 4294967;
+	if (order < 1000) {
+	    // Just make every point a DP
+	    dpfreq = 4294967295U;
+	} else if (NumBits(order) < 27) {
+	    // The frequency of DPs should be 10/sqrt(order) to avoid
+	    // a DP-free cycle, so dpfreq = (10*2^32)/sqrt(order)
+	    ZZ f = (to_ZZ(10) << 32) / SqrRoot(order);
+	    dpfreq = trunc_long(f, 31);
+	}
+	++curproblemid;
+	ret.push_back(SubproblemProgress(curproblemid, rep(subgroup_base),
+				    rep(subgroup_target),
+				    f.factor, order, dpfreq));
+    }
+
+    return ret;
+}
+
+static int generate_problem(struct event_base *evbase)
+{
+    // If there's already a problem on the go, don't generate another one
+    if (ctrlstate.working == true) {
+	return -1;
+    }
+    ctrlstate.working = true;
+
+    int num_subproblems_p = 0;
+    int num_subproblems_q = 0;
+
+    // Generate a DLP mod rho (in the large odd-order subgroup)
+    ZZ_p::init(ctrlstate.rho);
+    do {
+	ZZ_p base = power(random_ZZ_p(), 2);
+	ZZ_p target = power(random_ZZ_p(), 2);
+	ctrlstate.base = rep(base);
+	ctrlstate.target = rep(target);
+
+	// Decompose it mod p and mod q
+	ctrlstate.subproblems_p = decomp(base, target, ctrlstate.p);
+	ctrlstate.subproblems_q = decomp(base, target, ctrlstate.q);
+	num_subproblems_p = ctrlstate.subproblems_p.size();
+	num_subproblems_q = ctrlstate.subproblems_q.size();
+    } while (num_subproblems_p == 0 || num_subproblems_q == 0);
+
+    ctrlstate.num_unsolved_subproblems =
+	    num_subproblems_p + num_subproblems_q;
+
+    schedule();
+
+    return 0;
 }
 
 typedef enum {
@@ -367,7 +517,9 @@ static void controller_dpnode_reader(struct bufferevent *bev, void *ctx)
 		memmove(&DPport, ipport+4, 2);
 		{
 		    struct in_addr DPaddr = { DPip };
-		    printf("DP node at %s:%d\n", inet_ntoa(DPaddr), ntohs(DPport));
+#ifdef DEBUG
+		    fprintf(stderr, "DP node at %s:%d\n", inet_ntoa(DPaddr), ntohs(DPport));
+#endif
 		    if (ctrlstate.dpnodes.working.count(bev) > 0) {
 			ctrlstate.dpnodes.working[bev]->ipports.push_back(
 			    IPPort(ipport));
@@ -389,12 +541,19 @@ static void controller_dpnode_reader(struct bufferevent *bev, void *ctx)
 		if (ctrlstate.dpnodes.working.count(bev) > 0) {
 		    SubproblemProgress *spp = ctrlstate.dpnodes.working[bev];
 		    if (spp->problemid == problemid &&
+			    spp->solved == false &&
 			    spp->target ==
 				PowerMod(spp->base, expon, spp->modulus)) {
 			// Subproblem solved!
 			spp->solution = expon;
 			spp->solved = true;
 			spp->stop();
+			ctrlstate.num_unsolved_subproblems--;
+			if (ctrlstate.num_unsolved_subproblems == 0) {
+			    computation_complete();
+			    generate_problem(bufferevent_get_base(bev));
+			    ctrlstate.dump(cerr);
+			}
 			schedule();
 		    }
 		}
@@ -477,7 +636,6 @@ static void controller_master_reader(struct bufferevent *bev, void *ctx)
     bufferevent_read(bev, indata, 1);
     switch(indata[0]) {
 	case 'D':
-	    printf("DPnode\n");
 	    /* Add this DPnode to the list of available ones */
 	    ctrlstate.dpnodes.idle.insert(bev);
 	    bufferevent_setcb(bev, controller_dpnode_reader, NULL,
@@ -486,7 +644,6 @@ static void controller_master_reader(struct bufferevent *bev, void *ctx)
 	    schedule();
 	    return;
 	case 'W':
-	    printf("Worker\n");
 	    ctrlstate.workers.idle.insert(bev);
 	    // We don't actually read anything from workers
 	    bufferevent_enable(bev, EV_WRITE);
@@ -533,85 +690,6 @@ void *controller_create(struct event_base *evbase, unsigned short bindport,
 	ip, boundport, false);
 }
 
-static unsigned short curproblemid = 0;
-
-// Take base and target mod f.factor, then decompose that into small
-// subproblems given our knowledge of the factors of phi(f.factor)
-static vector<SubproblemProgress> decomp(const ZZ_p &base, const ZZ_p &target,
-    const FactorDecomp &f)
-{
-    vector<SubproblemProgress> ret;
-
-    // Compute phi(factor)
-    const int fveclen = f.fvec.length();
-    ZZ phi = to_ZZ(2);
-    for (int i = 0; i < fveclen; ++i) {
-	phi *= f.fvec[i];
-    }
-
-    ZZ_p::init(f.factor);
-    for (int i = 0; i < fveclen; ++i) {
-	const ZZ& order = f.fvec[i];
-	ZZ quotient = phi / order;
-	ZZ_p subgroup_base = to_ZZ_p(rep(base));
-	subgroup_base = power(subgroup_base, quotient);
-	ZZ_p subgroup_target = to_ZZ_p(rep(target));
-	subgroup_target = power(subgroup_target, quotient);
-
-	if (subgroup_base == 1) {
-	    // The original base wasn't a generator of the whole group
-	    if (subgroup_target == 1) {
-		// But the target is in the subgroup.  Lucky us.
-		continue;
-	    } else {
-		ret.clear();
-		return ret;
-	    }
-	}
-
-	// By default, 1 in 1000 points are distinguihed points.  The
-	// number in the next line is 2^32/1000
-	unsigned int dpfreq = 4294967;
-	if (order < 1000) {
-	    // Just make every point a DP
-	    dpfreq = 4294967295U;
-	} else if (NumBits(order) < 27) {
-	    // The frequency of DPs should be 10/sqrt(order) to avoid
-	    // a DP-free cycle, so dpfreq = (10*2^32)/sqrt(order)
-	    ZZ f = (to_ZZ(10) << 32) / SqrRoot(order);
-	    dpfreq = trunc_long(f, 31);
-	}
-	++curproblemid;
-	ret.push_back(SubproblemProgress(curproblemid, rep(subgroup_base),
-				    rep(subgroup_target),
-				    f.factor, order, dpfreq));
-    }
-
-    return ret;
-}
-
-static int generate_problem(struct event_base *evbase)
-{
-    // If there's already a problem on the go, don't generate another one
-    if (ctrlstate.working == 1) {
-	return -1;
-    }
-    ctrlstate.working = 1;
-
-    // Generate a DLP mod rho (in the large odd-order subgroup)
-    ZZ_p::init(ctrlstate.rho);
-    ZZ_p base = power(random_ZZ_p(), 2);
-    ZZ_p target = power(random_ZZ_p(), 2);
-
-    // Decompose it mod p and mod q
-    ctrlstate.subproblems_p = decomp(base, target, ctrlstate.p);
-    ctrlstate.subproblems_q = decomp(base, target, ctrlstate.q);
-
-    schedule();
-
-    return 0;
-}
-
 int main(int argc, char **argv)
 {
     // Initialize the prng with some randomness from the kernel

dpnode.cc

@@ -2,6 +2,7 @@ extern "C" {
 #include <event2/listener.h>
 #include <event2/bufferevent.h>
 #include <event2/buffer.h>
+#include <event2/util.h>
 }
 
 #include <sys/socket.h>
@@ -12,6 +13,7 @@ extern "C" {
 #include <map>
 #include <stdlib.h>
 #include <string.h>
+#include <errno.h>
 
 #include "evutils.h"
 #include "subproblem.h"
@@ -45,11 +47,17 @@ static void dpnode_event_cb(struct bufferevent *bev, short events,
     void *ctx)
 {
     if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
-	fprintf(stderr, "Closing connection (%s%s)\n",
-	    (events & BEV_EVENT_EOF) ? "EOF" : "",
-	    (events & BEV_EVENT_ERROR) ? "ERR" : "");
+	cerr << "Closing connection " << bev << " ";
+	if (events & BEV_EVENT_EOF) {
+	    cerr << "EOF";
+	}
+	if (events & BEV_EVENT_ERROR) {
+	    cerr << "ERR (" << evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR()) << ")";
+	}
+	cerr << "\n";
 	DPNodeConnInfo *info = (DPNodeConnInfo*)ctx;
 	delete info;
+	dpctrlstate.workers.erase(bev);
 	bufferevent_free(bev);
     }
 }
@@ -118,11 +126,14 @@ static void dpnode_accept_cb(struct evconnlistener *listener,
     struct event_base *base = evconnlistener_get_base(listener);
     struct bufferevent *bev = bufferevent_socket_new(
 	    base, fd, BEV_OPT_CLOSE_ON_FREE);
+    cerr << "accepted connection " << bev << "\n";
 
     bufferevent_setcb(bev, dpnode_reader, NULL,
 	    dpnode_event_cb, info);
 
     bufferevent_enable(bev, EV_READ);
+
+    dpctrlstate.workers.insert(bev);
 }
 
 // Create a new DPnode socket.  ip and boundport are set to the IP and
@@ -130,8 +141,12 @@ static void dpnode_accept_cb(struct evconnlistener *listener,
 struct evconnlistener *dpnode_create(struct event_base *evbase,
     unsigned int *ip, unsigned short *boundport)
 {
-    return listener_create(evbase, 0, dpnode_accept_cb, NULL,
-	ip, boundport, false);
+    struct evconnlistener *ecl = listener_create(evbase, 0,
+	dpnode_accept_cb, NULL, ip, boundport, false);
+
+    cerr << "Listening at " << ecl << "\n";
+
+    return ecl;
 }
 
 typedef enum {
@@ -148,17 +163,20 @@ struct DPControllerConnInfo {
 
 static void stop_problem(void)
 {
+    cerr << "Stopping problem\n";
     if (dpctrlstate.current_problem) {
 	delete dpctrlstate.current_problem;
 	dpctrlstate.current_problem = NULL;
     }
     if (dpctrlstate.listener) {
+	cerr << "Closing listener " << dpctrlstate.listener << "\n";
 	evconnlistener_free(dpctrlstate.listener);
 	dpctrlstate.listener = NULL;
     }
     std::set<struct bufferevent *>::iterator wit;
     for (wit = dpctrlstate.workers.begin(); wit != dpctrlstate.workers.end();
 	    ++wit) {
+	cerr << "Closing connection " << *wit << "\n";
 	bufferevent_free(*wit);
     }
     dpctrlstate.workers.clear();
@@ -202,6 +220,7 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 	    case DPCCSTATE_AWAITCMD:
 		if (len < 1) return;
 		bufferevent_read(bev, cmd, 1);
+		cerr << "Received command " << cmd[0] << "\n";
 		switch(cmd[0]) {
 		    case 'P':
 			info->state = DPCCSTATE_RDPROBLEM;
@@ -229,6 +248,8 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 		// Shut down
 		delete info;
 		event_base_loopbreak(bufferevent_get_base(bev));
+		cerr << "END conenction " << bev << "\n";
+		dpctrlstate.workers.erase(bev);
 		bufferevent_free(bev);
 		return;
 	}

evutils.cc

@@ -85,7 +85,9 @@ struct bufferevent *client_create(struct event_base *evbase,
     memmove(&client_sin.sin_port, ipport+4, 2);
 
     struct bufferevent *client_bev = bufferevent_socket_new(evbase,
-	-1, BEV_OPT_CLOSE_ON_FREE|(multithread ? BEV_OPT_THREADSAFE : 0));
+	-1, BEV_OPT_CLOSE_ON_FREE |
+	    (multithread ? BEV_OPT_DEFER_CALLBACKS|BEV_OPT_UNLOCK_CALLBACKS|
+	                   BEV_OPT_THREADSAFE : 0));
 
     if (bufferevent_socket_connect(client_bev,
 	    (struct sockaddr *)&client_sin, sizeof(client_sin)) < 0) {

worker.cc

@@ -5,12 +5,15 @@ extern "C" {
 #include <event2/bufferevent.h>
 #include <event2/buffer.h>
 #include <event2/event.h>
+#include <event2/util.h>
 }
 
 #include <NTL/ZZ_p.h>
 
 #include <vector>
 #include <stdio.h>
+#include <errno.h>
+#include <signal.h>
 
 #include "cudadl.h"
 
@@ -19,6 +22,8 @@ extern "C" {
 
 NTL_CLIENT
 
+#define DEBUG
+
 typedef enum {
     WRKCCSTATE_AWAITCMD,
     WRKCCSTATE_RDPROBLEM,
@@ -96,7 +101,9 @@ static void *worker_thread_start(void *data)
 
 static void stop_working(void)
 {
+#ifdef DEBUG
     cerr << "Stopping work\n";
+#endif
     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;
@@ -108,6 +115,9 @@ static void stop_working(void)
     vector<struct bufferevent *>::iterator bevit;
     for (bevit = wrkctrlstate.dpnodes.begin();
 	    bevit != wrkctrlstate.dpnodes.end(); ++bevit) {
+#ifdef DEBUG
+	cerr << "Closing connection to " << *bevit << "\n";
+#endif
 	bufferevent_free(*bevit);
     }
     wrkctrlstate.dpnodes.clear();
@@ -121,7 +131,9 @@ static void stop_working(void)
 
 static void start_working(void)
 {
+#ifdef DEBUG
     cerr << "Starting work\n";
+#endif
     wrkctrlstate.worker_thread_state = WT_RUNNING;
     if (pthread_create(&wrkctrlstate.worker_thread, NULL,
 			worker_thread_start, NULL)) {
@@ -135,6 +147,9 @@ static void dpconn_event_cb(struct bufferevent *bev, short events,
 {
     if (events & BEV_EVENT_CONNECTED) {
 	// We have successfully connected to the dpnode
+#ifdef DEBUG
+	cerr << "Connection established to dpnode " << bev << "\n";
+#endif
 	bufferevent_enable(bev, EV_WRITE);
 	++wrkctrlstate.num_connected_dpnodes;
 	if (wrkctrlstate.num_connected_dpnodes ==
@@ -142,7 +157,16 @@ static void dpconn_event_cb(struct bufferevent *bev, short events,
 	    start_working();
 	}
     } else if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
-	fprintf(stderr, "Closing connection to dpnode\n");
+#ifdef DEBUG
+	cerr << "Closing connection to dpnode " << bev << " ";
+	if (events & BEV_EVENT_EOF) {
+	    cerr << "EOF";
+	}
+	if (events & BEV_EVENT_ERROR) {
+	    cerr << "ERR (" << evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR()) << ")";
+	}
+	cerr << "\n";
+#endif
 	stop_working();
     }
 }
@@ -160,6 +184,9 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 	    case WRKCCSTATE_AWAITCMD:
 		if (len < 1) return;
 		bufferevent_read(bev, cmd, 1);
+#ifdef DEBUG
+		cerr << "Command " << cmd[0] << " received\n";
+#endif
 		switch(cmd[0]) {
 		    case 'P':
 			info->state = WRKCCSTATE_RDPROBLEM;
@@ -195,6 +222,7 @@ 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);
+#ifdef DEBUG
 			cerr << "Connecting to DPnode " <<
 				    int(ipport[0]) << "." <<
 				    int(ipport[1]) << "." <<
@@ -202,9 +230,13 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 				    int(ipport[3]) << ":" <<
 				    ((ipport[4] << 8) + ipport[5]) <<
 				    "\n";
+#endif
 			struct bufferevent *dpbev = client_create(
 				bufferevent_get_base(bev), ipport,
 				dpconn_event_cb, true);
+#ifdef DEBUG
+			cerr << "Starting connection to dpnode " << dpbev << "\n";
+#endif
 			if (dpbev) {
 			    wrkctrlstate.dpnodes.push_back(dpbev);
 			} else {
@@ -219,6 +251,9 @@ static void controllerconn_reader(struct bufferevent *bev, void *ctx)
 		// Shut down
 		delete info;
 		event_base_loopbreak(bufferevent_get_base(bev));
+#ifdef DEBUG
+		cerr << "Closing connection to " << bev << "\n";
+#endif
 		bufferevent_free(bev);
 		return;
 	}
@@ -239,6 +274,9 @@ static void controllerconn_event_cb(struct bufferevent *bev, short events,
     } else if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
 	fprintf(stderr, "Closing connection to controller and exiting\n");
 	event_base_loopbreak(bufferevent_get_base(bev));
+#ifdef DEBUG
+	cerr << "Closing connection to " << bev << "\n";
+#endif
 	bufferevent_free(bev);
     }
 }
@@ -254,6 +292,8 @@ int main(int argc, char **argv)
 
     evthread_use_pthreads();
 
+    signal(SIGPIPE, SIG_IGN);
+
     return controller_client(argv[1], controller_port,
 				controllerconn_event_cb, true);
 }