libevent-based framework for distributed DL computation

In this commit:

- controller, which coordinates the computation
- dpnode, which connects to the controller to learn which subproblem
  to listen for DPs for
- worker, which connects to the controller to learn which subproblem
  to work on, and which dpnodes to connect to to send the DPs it finds

The worker does not yet do any actual work, and the dpnode does not yet
read DPs from its listening socket (it does set up the socket, though).

Makefile

@@ -15,16 +15,18 @@
 #  You should have received a copy of the GNU General Public License
 #  along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-CXXFLAGS=-g -Wall -O2
-NVCCOPTS=-g -arch sm_20 --ptxas-options=-v -O2
+LIBEVENT=/usr/local
+
+WORDS = 24
+
+CXXFLAGS=-g -Wall -O0 -DWORDS=$(WORDS) -I$(LIBEVENT)/include
+NVCCOPTS=-g -arch sm_20 --ptxas-options=-v -O2 -DWORDS=$(WORDS)
 NVCC=nvcc $(NVCCOPTS)
 
 CXXFILES = dlrho.cc gen_N.cc
 CUFILES = parrhoasm.cu dpstream.cu
-OFILES = dlrho.o gen_N.o cudadl.o
-TARGETS = gen_N dlrho
-
-WORDS = 24
+OFILES = dlrho.o gen_N.o cudadl.o controller.o evutils.o dpnode.o worker.o
+TARGETS = gen_N dlrho controller dpnode worker
 
 all: $(TARGETS)
 
@@ -38,11 +40,20 @@ dlrho.o: dlrho.cc
 	g++ $(CXXFLAGS) -I /usr/local/cuda/include $^ -c -o $@
 
 cudadl.o: parrhoasm.cu cios.asm dpstream.cu
-	$(NVCC) -c parrhoasm.cu -DWORDS=$(WORDS) -o cudadl.o
+	$(NVCC) -c parrhoasm.cu -o cudadl.o
 
 cios.asm: gencios_reg_20
 	./gencios_reg_20 $(WORDS) > $@
 
+controller: controller.o evutils.o
+	g++ -g -Wall $^ -o $@ -L$(LIBEVENT)/lib -levent -lntl -lgmp
+
+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
+
 clean:
 	-rm -f $(OFILES)
 

controller.cc

@@ -0,0 +1,620 @@
+extern "C" {
+#include <event2/listener.h>
+#include <event2/bufferevent.h>
+#include <event2/buffer.h>
+}
+
+#include <NTL/vec_ZZ.h>
+#include <NTL/ZZ.h>
+#include <NTL/ZZ_p.h>
+
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+
+#include <fstream>
+#include <vector>
+#include <set>
+#include <map>
+#include <stdlib.h>
+#include <string.h>
+
+#include "evutils.h"
+#include "subproblem.h"
+
+NTL_CLIENT
+
+struct SubproblemProgress;
+
+typedef std::set<struct bufferevent *> BESet;
+typedef std::map<struct bufferevent *, SubproblemProgress *> BEMap;
+
+void besetdump(const BESet &bes, ostream &os)
+{
+    BESet::const_iterator besit;
+
+    os << hex << "    ";
+    for (besit = bes.begin(); besit != bes.end(); ++besit) {
+	os << *besit << " ";
+    }
+    os << dec << "\n";
+}
+
+void bemapdump(const BEMap &bem, ostream &os)
+{
+    BEMap::const_iterator bemit;
+
+    os << hex << "    ";
+    for (bemit = bem.begin(); bemit != bem.end(); ++bemit) {
+	os << bemit->first << "->" << bemit->second << " ";
+    }
+    os << dec << "\n";
+}
+
+struct Statuses {
+    BESet idle;
+    BEMap working;
+
+    // Dump the state for debug purposes
+    void dump(ostream &os) const {
+	os << "  idle (" << idle.size() << "):\n";
+	besetdump(idle, os);
+	os << "  working (" << working.size() << "):\n";
+	bemapdump(working, os);
+    }
+};
+
+struct FactorDecomp {
+    ZZ factor;
+    vec_ZZ fvec;
+};
+
+void vsppdump(const vector<SubproblemProgress> &spv, ostream &os);
+
+static struct ControllerState {
+    ZZ rho;
+    FactorDecomp p, q;
+    int working;
+    vector<SubproblemProgress> subproblems_p, subproblems_q;
+    Statuses dpnodes, workers;
+
+    ControllerState() : working(0) {}
+
+    // Dump the state for debug purposes
+    void dump(ostream &os) const {
+	if (!working) {
+	    os << "Not working\n";
+	    return;
+	}
+	os << "P:\n";
+	vsppdump(subproblems_p, os);
+	os << "Q:\n";
+	vsppdump(subproblems_q, os);
+	os << "dpnodes:\n";
+	dpnodes.dump(os);
+	os << "workers:\n";
+	workers.dump(os);
+    }
+} ctrlstate;
+
+struct IPPort {
+    unsigned char ipport[6];
+
+    IPPort(unsigned char *ipp) {
+	memmove(ipport, ipp, 6);
+    }
+
+    void dump(ostream &os) const {
+	os << int(ipport[0]) << "." << int(ipport[1]) << "." <<
+	    int(ipport[2]) << "." << int(ipport[3]) << ":" <<
+	    ((ipport[4] << 8) + ipport[5]) << " ";
+    }
+};
+
+typedef vector<IPPort> IPPortSet;
+
+void ipportsetdump(const IPPortSet &ipps, ostream &os)
+{
+    IPPortSet::const_iterator ippsit;
+
+    os << "    ";
+    for (ippsit = ipps.begin(); ippsit != ipps.end(); ++ippsit) {
+	ippsit->dump(os);
+    }
+    os << "\n";
+}
+
+struct SubproblemProgress : Subproblem {
+
+    // The sets of dpnodes and workers currently working on this subproblem
+    BESet dpnodes, workers;
+    // The dpnode IPPorts registered for this subproblem
+    IPPortSet ipports;
+
+    // The desired number of DPnodes for this subproblem
+    unsigned short desired_dpnodes;
+    // The maximum number of workers useful for this subproblem
+    unsigned int max_workers;
+
+    // Have we found a solution?
+    int solved;
+    // The solution, if found.
+    ZZ solution;
+
+    SubproblemProgress(unsigned short id, const ZZ &b, const ZZ &t,
+	    const ZZ &m, const ZZ &o, unsigned int dpf) :
+	    Subproblem(id, b, t, m, o, dpf), solved(0) {
+	// How many DPnodes should we use for a problem of this size?
+	desired_dpnodes = 2;
+	// How many workers would we like to use?
+	ZZ sorder = SqrRoot(order >> 46);
+	if (NumBits(sorder) > 30) {
+	    // Just use all the workers we can find
+	    max_workers = 4294967295U;  // 2^32 - 1
+	} else {
+	    max_workers = trunc_long(sorder,31) + 1;
+	}
+    }
+
+    // Stop all dpnodes and workers and reset to unstarted state
+    void reset(void) {
+	BESet::iterator iter;
+	unsigned char stopcmd[1] = { 'S' };
+
+	for (BESet::iterator iter = dpnodes.begin(); iter != dpnodes.end();
+		++iter) {
+	    bufferevent_write(*iter, stopcmd, 1);
+	    ctrlstate.dpnodes.working.erase(*iter);
+	    ctrlstate.dpnodes.idle.insert(*iter);
+	}
+	for (BESet::iterator iter = workers.begin(); iter != workers.end();
+		++iter) {
+	    bufferevent_write(*iter, stopcmd, 1);
+	    ctrlstate.workers.working.erase(*iter);
+	    ctrlstate.workers.idle.insert(*iter);
+	}
+	dpnodes.clear();
+	workers.clear();
+	ipports.clear();
+    }
+
+    // Dump for debugging purposes
+    void dump(ostream &os) const {
+	os << "    dpnodes (" << dpnodes.size() << "):\n";
+	besetdump(dpnodes, os);
+	os << "    workers (" << workers.size() << "):\n";
+	besetdump(workers, os);
+	os << "    ipports (" << ipports.size() << "):\n";
+	ipportsetdump(ipports, os);
+    }
+
+    void worker_write(struct bufferevent *bev) {
+	bev_write(bev);
+	unsigned short num_ipports = ipports.size();
+	bufferevent_write(bev, &num_ipports, 2);
+	for (unsigned short i = 0; i < num_ipports; ++i) {
+	    bufferevent_write(bev, ipports[i].ipport, 6);
+	}
+    }
+};
+
+// Dump the state for debug purposes
+void vsppdump(const vector<SubproblemProgress> &spv, ostream &os)
+{
+    vector<SubproblemProgress>::const_iterator spiter;
+    int count = 0;
+
+    for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
+	++count;
+	os << "  " << count << ":\n";
+	spiter->dump(os);
+    }
+    os << "\n";
+}
+
+// Find a subproblem in the given vector that could use another DPnode,
+// and give it one of the idle ones.  Only allocate it to a subproblem
+// with no current DPnodes if consider_empty is true.
+static void find_subproblem_for_dpnode(vector<SubproblemProgress> &spv,
+	bool consider_empty)
+{
+    vector<SubproblemProgress>::iterator spiter;
+
+    for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
+	if (spiter->solved) continue;
+	if (spiter->dpnodes.size() == 0 && consider_empty == false) continue;
+	// How many DPnodes would we like to have for this subproblem?
+	while (spiter->dpnodes.size() < spiter->desired_dpnodes &&
+		ctrlstate.dpnodes.idle.size() > 0) {
+	    // Get the first idle DPnode
+	    BESet::iterator beviter = ctrlstate.dpnodes.idle.begin();
+
+	    // Allocate it to the subproblem
+	    spiter->dpnodes.insert(*beviter);
+	    ctrlstate.dpnodes.working[*beviter] = &(*spiter);
+	    ctrlstate.dpnodes.idle.erase(*beviter);
+
+	    // Tell it to start listening for DPs
+	    spiter->bev_write(*beviter);
+	}
+    }
+}
+
+// Find a subproblem in the given vector that has all of its DPnodes and
+// could use another worker, and give it one of the idle ones.
+static void find_subproblem_for_worker(vector<SubproblemProgress> &spv)
+{
+    vector<SubproblemProgress>::iterator spiter;
+
+    for (spiter = spv.begin(); spiter != spv.end(); ++spiter) {
+	if (spiter->solved) continue;
+
+	while (spiter->ipports.size() == spiter->desired_dpnodes &&
+		spiter->workers.size() < spiter->max_workers &&
+		ctrlstate.workers.idle.size() > 0) {
+	    // Get the first idle worker
+	    BESet::iterator beviter = ctrlstate.workers.idle.begin();
+
+	    // Allocate it to the subproblem
+	    spiter->workers.insert(*beviter);
+	    ctrlstate.workers.working[*beviter] = &(*spiter);
+	    ctrlstate.workers.idle.erase(*beviter);
+
+	    // Tell it to start working on the subproblem
+	    spiter->worker_write(*beviter);
+	}
+    }
+}
+
+// See if there are any idle DPnodes or workers we can put to use
+void schedule(void)
+{
+    cerr << "Before schedule:\n"; ctrlstate.dump(cerr);
+
+    // Check the DPnodes
+
+    // Iterate through the subproblems, looking for one that can use
+    // another DPnode.  First look for subproblems that already have
+    // some, but not all, of their DPnodes
+    if (ctrlstate.dpnodes.idle.size() > 0) {
+	find_subproblem_for_dpnode(ctrlstate.subproblems_p, false);
+    }
+    if (ctrlstate.dpnodes.idle.size() > 0) {
+	find_subproblem_for_dpnode(ctrlstate.subproblems_q, false);
+    }
+    // If there are still more dpnodes to place, start assigning them to
+    // subproblems with no current dpnodes
+    if (ctrlstate.dpnodes.idle.size() > 0) {
+	find_subproblem_for_dpnode(ctrlstate.subproblems_p, true);
+    }
+    if (ctrlstate.dpnodes.idle.size() > 0) {
+	find_subproblem_for_dpnode(ctrlstate.subproblems_q, true);
+    }
+
+    // Check the workers
+
+    // Iterate through the subproblems, looking for one that can use
+    // another worker.
+    if (ctrlstate.workers.idle.size() > 0) {
+	find_subproblem_for_worker(ctrlstate.subproblems_p);
+    }
+    if (ctrlstate.workers.idle.size() > 0) {
+	find_subproblem_for_worker(ctrlstate.subproblems_q);
+    }
+
+    cerr << "After schedule:\n"; ctrlstate.dump(cerr);
+}
+
+typedef enum {
+    CCSTATE_START,
+    CCSTATE_DPWAITRESP,
+    CCSTATE_DPLISTENING,
+    CCSTATE_END
+} CCState;
+
+struct ControllerConnInfo {
+    CCState state;
+
+    ControllerConnInfo() : state(CCSTATE_DPWAITRESP) {}
+};
+
+static void controller_dpnode_reader(struct bufferevent *bev, void *ctx)
+{
+    struct evbuffer *input = bufferevent_get_input(bev);
+    ControllerConnInfo *info = (ControllerConnInfo *)ctx;
+    unsigned char cmd[1];
+
+    while(1) {
+	size_t len = evbuffer_get_length(input);
+	switch (info->state) {
+	    case CCSTATE_START:
+	    case CCSTATE_DPWAITRESP:
+		    if (len < 1) return;
+		    bufferevent_read(bev, cmd, 1);
+		    switch (cmd[0]) {
+			case 'L':
+			    info->state = CCSTATE_DPLISTENING;
+			    break;
+			default:
+			    /* Unknown DPnode command received */
+			    fprintf(stderr, "Unknown command in "
+				    "controller_dpnode_reader: "
+				    "%c\n", cmd[0]);
+			    info->state = CCSTATE_END;
+			    break;
+		    }
+		    break;
+
+	    case CCSTATE_DPLISTENING:
+		// Read 6 bytes
+		if (len < 6) return;
+		unsigned char ipport[6];
+		unsigned int DPip;
+		unsigned short DPport;
+		bufferevent_read(bev, ipport, 6);
+		memmove(&DPip, ipport, 4);
+		memmove(&DPport, ipport+4, 2);
+		{
+		    struct in_addr DPaddr = { DPip };
+		    printf("DP node at %s:%d\n", inet_ntoa(DPaddr), ntohs(DPport));
+		    if (ctrlstate.dpnodes.working.count(bev) > 0) {
+			ctrlstate.dpnodes.working[bev]->ipports.push_back(
+			    IPPort(ipport));
+			schedule();
+		    }
+		}
+		info->state = CCSTATE_DPWAITRESP;
+		break;
+
+	    case CCSTATE_END:
+		// Shut down the connection
+		delete info;
+		bufferevent_free(bev);
+		return;
+	}
+    }
+
+}
+
+static void controller_worker_reader(struct bufferevent *bev, void *ctx)
+{
+    struct evbuffer *input = bufferevent_get_input(bev);
+    size_t len = evbuffer_get_length(input);
+
+}
+
+static void controller_dpnode_event_cb(struct bufferevent *bev, short events,
+    void *ctx)
+{
+    if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
+	ControllerConnInfo *info = (ControllerConnInfo*)ctx;
+	fprintf(stderr, "Closing dpnode connection\n");
+	if (ctrlstate.dpnodes.working.count(bev)) {
+	    // If we lose a dpnode from an active computation, the
+	    // computation is useless.
+	    SubproblemProgress *spp = ctrlstate.dpnodes.working[bev];
+	    ctrlstate.dpnodes.working.erase(bev);
+	    spp->dpnodes.erase(bev);
+	    spp->reset();
+	} else {
+	    ctrlstate.dpnodes.idle.erase(bev);
+	}
+	delete info;
+	bufferevent_free(bev);
+	schedule();
+    }
+}
+
+static void controller_worker_event_cb(struct bufferevent *bev, short events,
+    void *ctx)
+{
+    if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
+	ControllerConnInfo *info = (ControllerConnInfo*)ctx;
+	fprintf(stderr, "Closing worker connection\n");
+	if (ctrlstate.workers.working.count(bev)) {
+	    SubproblemProgress *spp = ctrlstate.workers.working[bev];
+	    ctrlstate.workers.working.erase(bev);
+	    spp->workers.erase(bev);
+	} else {
+	    ctrlstate.workers.idle.erase(bev);
+	}
+	delete info;
+	bufferevent_free(bev);
+	schedule();
+    }
+}
+
+static void controller_event_cb(struct bufferevent *bev, short events,
+    void *ctx)
+{
+    if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
+	fprintf(stderr, "Closing connection\n");
+	ControllerConnInfo *info = (ControllerConnInfo*)ctx;
+	delete info;
+	bufferevent_free(bev);
+    }
+}
+
+// We're just going to read a single byte that will tell us whether the
+// peer is a DPnode or a Worker
+static void controller_master_reader(struct bufferevent *bev, void *ctx)
+{
+    struct evbuffer *input = bufferevent_get_input(bev);
+
+    size_t len = evbuffer_get_length(input);
+
+    if (len < 1) return;
+
+    char indata[1];
+    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,
+		    controller_dpnode_event_cb, ctx);
+	    controller_dpnode_reader(bev, ctx);
+	    schedule();
+	    return;
+	case 'W':
+	    printf("Worker\n");
+	    ctrlstate.workers.idle.insert(bev);
+	    bufferevent_setcb(bev, controller_worker_reader, NULL,
+		    controller_worker_event_cb, ctx);
+	    controller_worker_reader(bev, ctx);
+	    schedule();
+	    return;
+	default:
+	    fprintf(stderr, "Unknown command in controller_master_reader: "
+		    "%c\n", indata[0]);
+
+	    ControllerConnInfo *info = (ControllerConnInfo*)ctx;
+	    delete info;
+	    bufferevent_free(bev);
+	    return;
+    }
+}
+
+static void controller_accept_cb(struct evconnlistener *listener,
+    evutil_socket_t fd, struct sockaddr *address, int socklen,
+    void *ctx)
+{
+    // Create the state of the new connection
+    ControllerConnInfo *info = new ControllerConnInfo();
+
+    // Create a bufferevent for the new connection
+    struct event_base *base = evconnlistener_get_base(listener);
+    struct bufferevent *bev = bufferevent_socket_new(
+	    base, fd, BEV_OPT_CLOSE_ON_FREE);
+
+    bufferevent_setcb(bev, controller_master_reader, NULL,
+	    controller_event_cb, info);
+
+    bufferevent_enable(bev, EV_READ|EV_WRITE);
+}
+
+// Create a new controller socket.  bindport is the port to bind to (in
+// host byte order), or 0 if any port will do.  ip and boundport are set
+// to the IP and port of the socket, in network byte order.
+void *controller_create(struct event_base *evbase, unsigned short bindport,
+    unsigned int *ip, unsigned short *boundport)
+{
+    return listener_create(evbase, bindport, controller_accept_cb, NULL,
+	ip, boundport);
+}
+
+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);
+	}
+	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
+    unsigned char randbuf[1024];
+    ifstream urand("/dev/urandom");
+    urand.read((char *)randbuf, sizeof(randbuf));
+    urand.close();
+    ZZ randzz = ZZFromBytes(randbuf, sizeof(randbuf));
+    SetSeed(randzz);
+
+    // Read the modulus and the factorization of its totient from cin
+    cin >> ctrlstate.rho >> ctrlstate.p.factor >> ctrlstate.p.fvec >>
+	ctrlstate.q.factor >> ctrlstate.q.fvec;
+
+    unsigned short bindport = 0;
+
+    if (argc > 1) {
+	bindport = strtoul(argv[1], NULL, 10);
+    }
+
+    struct event_base *evbase = event_base_new();
+
+    unsigned int myip;
+    unsigned short myport;
+
+    controller_create(evbase, bindport, &myip, &myport);
+    struct in_addr myaddr = { myip };
+    printf("Bound to %s:%d\n", inet_ntoa(myaddr), ntohs(myport));
+
+    // Kick off the first problem to solve
+    generate_problem(evbase);
+
+    event_base_dispatch(evbase);
+
+    return 0;
+}

dpnode.cc

@@ -0,0 +1,205 @@
+extern "C" {
+#include <event2/listener.h>
+#include <event2/bufferevent.h>
+#include <event2/buffer.h>
+}
+
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+
+#include <set>
+#include <stdlib.h>
+#include <string.h>
+
+#include "evutils.h"
+#include "subproblem.h"
+
+typedef enum {
+    DPSTATE_START,
+    DPSTATE_END
+} DPState;
+
+struct DPNodeConnInfo {
+    DPState state;
+
+    DPNodeConnInfo() : state(DPSTATE_START) {}
+};
+
+static void dpnode_event_cb(struct bufferevent *bev, short events,
+    void *ctx)
+{
+    if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
+	fprintf(stderr, "Closing connection\n");
+	DPNodeConnInfo *info = (DPNodeConnInfo*)ctx;
+	delete info;
+	bufferevent_free(bev);
+    }
+}
+
+static void dpnode_reader(struct bufferevent *bev, void *ctx)
+{
+}
+
+static void dpnode_accept_cb(struct evconnlistener *listener,
+    evutil_socket_t fd, struct sockaddr *address, int socklen,
+    void *ctx)
+{
+    DPNodeConnInfo *info = new DPNodeConnInfo();
+
+    // Create a bufferevent for the new connection
+    struct event_base *base = evconnlistener_get_base(listener);
+    struct bufferevent *bev = bufferevent_socket_new(
+	    base, fd, BEV_OPT_CLOSE_ON_FREE);
+
+    bufferevent_setcb(bev, dpnode_reader, NULL,
+	    dpnode_event_cb, info);
+
+    bufferevent_enable(bev, EV_READ|EV_WRITE);
+}
+
+// Create a new DPnode socket.  ip and boundport are set to the IP and
+// port of the socket, in network byte order.
+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);
+}
+
+typedef enum {
+    DPCCSTATE_AWAITCMD,
+    DPCCSTATE_RDPROBLEM,
+    DPCCSTATE_END
+} DPCCState;
+
+struct DPControllerConnInfo {
+    DPCCState state;
+
+    DPControllerConnInfo() : state(DPCCSTATE_AWAITCMD) {}
+};
+
+static struct DPControllerState {
+    Subproblem *current_problem;
+    struct evconnlistener *listener;
+    std::set<struct bufferevent *> workers;
+
+    DPControllerState() : current_problem(NULL), listener(NULL) {}
+} dpctrlstate;
+
+static void stop_problem(void)
+{
+    if (dpctrlstate.current_problem) {
+	dpctrlstate.current_problem = NULL;
+    }
+    if (dpctrlstate.listener) {
+	evconnlistener_free(dpctrlstate.listener);
+	dpctrlstate.listener = NULL;
+    }
+    std::set<struct bufferevent *>::iterator wit;
+    for (wit = dpctrlstate.workers.begin(); wit != dpctrlstate.workers.end();
+	    ++wit) {
+	bufferevent_free(*wit);
+    }
+    dpctrlstate.workers.clear();
+}
+
+static void start_problem(struct bufferevent *bev,
+	const unsigned char *subproblem)
+{
+    unsigned int myip;
+    unsigned short myport;
+
+    stop_problem();
+    dpctrlstate.current_problem = new Subproblem(subproblem);
+    dpctrlstate.current_problem->dump(cerr);
+
+    // Create the DPNode server socket
+    dpctrlstate.listener = dpnode_create(bufferevent_get_base(bev),
+	    &myip, &myport);
+    struct in_addr myaddr = { myip };
+    fprintf(stderr, "Bound to %s:%d\n", inet_ntoa(myaddr), ntohs(myport));
+    unsigned char idstring[7];
+    idstring[0] = 'L';
+    memmove(idstring+1, &myip, 4);
+    memmove(idstring+5, &myport, 2);
+
+    bufferevent_write(bev, idstring, 7);
+}
+
+static void controllerconn_reader(struct bufferevent *bev, void *ctx)
+{
+    struct evbuffer *input = bufferevent_get_input(bev);
+    DPControllerConnInfo *info = (DPControllerConnInfo *)ctx;
+    unsigned char cmd[1];
+    unsigned char subproblem[SUBPROBLEM_DESC_LEN];
+
+    while(1) {
+	size_t len = evbuffer_get_length(input);
+	switch(info->state) {
+	    case DPCCSTATE_AWAITCMD:
+		if (len < 1) return;
+		bufferevent_read(bev, cmd, 1);
+		switch(cmd[0]) {
+		    case 'P':
+			info->state = DPCCSTATE_RDPROBLEM;
+			break;
+		    case 'S':
+			stop_problem();
+			break;
+		    default:
+			/* Unknown command received */
+			fprintf(stderr, "Unknown command in "
+				"controllerconn_reader: %c\n", cmd[0]);
+			info->state = DPCCSTATE_END;
+			break;
+		}
+		break;
+
+	    case DPCCSTATE_RDPROBLEM:
+		if (len < SUBPROBLEM_DESC_LEN) return;
+		bufferevent_read(bev, subproblem, SUBPROBLEM_DESC_LEN);
+		start_problem(bev, subproblem);
+		info->state = DPCCSTATE_AWAITCMD;
+		break;
+
+	    case DPCCSTATE_END:
+		// Shut down
+		delete info;
+		event_base_loopbreak(bufferevent_get_base(bev));
+		bufferevent_free(bev);
+		return;
+	}
+    }
+}
+
+static void controllerconn_event_cb(struct bufferevent *bev, short events,
+    void *ctx)
+{
+    if (events & BEV_EVENT_CONNECTED) {
+	// We have successfully connected to the controller
+	char id[1] = { 'D' };
+
+        bufferevent_enable(bev, EV_READ|EV_WRITE);
+	bufferevent_write(bev, id, 1);
+	bufferevent_setcb(bev, controllerconn_reader, NULL,
+		controllerconn_event_cb, new DPControllerConnInfo());
+    } 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));
+	bufferevent_free(bev);
+    }
+}
+
+int main(int argc, char **argv)
+{
+    if (argc != 3) {
+	fprintf(stderr, "Usage: %s controller_host controller_port\n", argv[0]);
+	return 1;
+    }
+
+    unsigned short controller_port = strtoul(argv[2], NULL, 10);
+
+    return controller_client(argv[1], controller_port,
+				controllerconn_event_cb);
+}

evutils.cc

@@ -0,0 +1,134 @@
+extern "C" {
+#include <event2/listener.h>
+#include <event2/bufferevent.h>
+}
+
+#include "evutils.h"
+
+#include <string.h>
+#include <netdb.h>
+#include <unistd.h>
+
+unsigned int hostlookup(const char *hostname)
+{
+    struct hostent *he = gethostbyname(hostname);
+    if (!he) {
+	// Could not resolve hostname
+	return 0;
+    }
+    struct in_addr *addr = (struct in_addr*)(he->h_addr_list[0]);
+    return addr->s_addr;
+}
+
+static unsigned int getlocalIP(void)
+{
+    char hostname[257];
+    if (gethostname(hostname, 256)) {
+	// Could not determine hostname
+	return 0;
+    }
+    hostname[256] = '\0';
+    return hostlookup(hostname);
+}
+
+// Create a new listener socket.  bindport is the port to bind to (in
+// host byte order), or 0 if any port will do.  ip and boundport are set
+// to the IP and port of the socket, in network byte order.
+struct evconnlistener *listener_create(struct event_base *evbase,
+    unsigned short bindport, evconnlistener_cb cb, void *ctx,
+    unsigned int *ip, unsigned short *boundport)
+{
+    struct sockaddr_in sin;
+    sin.sin_family = AF_INET;
+    sin.sin_addr.s_addr = htonl(0);
+    sin.sin_port = htons(bindport);
+
+    if (ip) {
+	*ip = getlocalIP();
+	if (*ip == 0) {
+	    perror("Determining local IP");
+	    return NULL;
+	}
+    }
+
+    struct evconnlistener *listener = evconnlistener_new_bind(evbase,
+	cb, ctx, LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE, -1,
+	(struct sockaddr*)&sin, sizeof(sin));
+    if (!listener) {
+	perror("Unable to create listener");
+	return NULL;
+    }
+    int localfd = evconnlistener_get_fd(listener);
+    if (boundport) {
+	socklen_t addrsize = sizeof(sin);
+	if (getsockname(localfd, (struct sockaddr*)&sin, &addrsize)) {
+	    perror("Unable to learn local port");
+	    return NULL;
+	}
+	*boundport = sin.sin_port;
+    }
+
+    return listener;
+}
+
+// Create a client connection to the given 6-byte ipport (4 byte IP, 2
+// byte port in network byte order).
+struct bufferevent *client_create(struct event_base *evbase,
+    unsigned char ipport[6], bufferevent_event_cb event_handler)
+{
+    // Create a Controller client socket
+    struct sockaddr_in client_sin;
+    client_sin.sin_family = AF_INET;
+    memmove(&client_sin.sin_addr.s_addr, ipport, 4);
+    memmove(&client_sin.sin_port, ipport+4, 2);
+
+    struct bufferevent *client_bev = bufferevent_socket_new(evbase,
+	-1, BEV_OPT_CLOSE_ON_FREE);
+
+    if (bufferevent_socket_connect(client_bev,
+	    (struct sockaddr *)&client_sin, sizeof(client_sin)) < 0) {
+	bufferevent_free(client_bev);
+	fprintf(stderr, "Unable to connect to server\n");
+	return NULL;
+    }
+
+    bufferevent_setcb(client_bev, NULL, NULL, event_handler, NULL);
+
+    return client_bev;
+}
+
+// Create a client connection to a controller, with event_handler set as
+// the event callback.  It will be called when the connection succeeds or
+// fails.  This function calls the libevent main loop, so it will only return
+// when the program is finished.
+int controller_client(const char *controller_host,
+    unsigned short controller_port, bufferevent_event_cb event_handler)
+{
+    struct event_base *evbase = event_base_new();
+
+    // Create a Controller client socket
+    struct sockaddr_in controller_sin;
+    controller_sin.sin_family = AF_INET;
+    controller_sin.sin_addr.s_addr = hostlookup(controller_host);
+    if (controller_sin.sin_addr.s_addr == 0) {
+	fprintf(stderr, "Unknown host %s\n", controller_host);
+	return 1;
+    }
+    controller_sin.sin_port = htons(controller_port);
+
+    struct bufferevent *controller_bev = bufferevent_socket_new(evbase,
+	-1, BEV_OPT_CLOSE_ON_FREE);
+
+    if (bufferevent_socket_connect(controller_bev,
+	    (struct sockaddr *)&controller_sin, sizeof(controller_sin)) < 0) {
+	bufferevent_free(controller_bev);
+	fprintf(stderr, "Unable to connect to controller\n");
+	return 1;
+    }
+
+    bufferevent_setcb(controller_bev, NULL, NULL, event_handler, NULL);
+
+    event_base_dispatch(evbase);
+
+    return 0;
+}

evutils.h

@@ -0,0 +1,31 @@
+#ifndef __EVUTILS_H__
+#define __EVUTILS_H__
+
+extern "C" {
+#include <event2/listener.h>
+#include <event2/bufferevent.h>
+}
+
+unsigned int hostlookup(const char *hostname);
+
+
+// Create a new listener socket.  bindport is the port to bind to (in
+// host byte order), or 0 if any port will do.  ip and boundport are set
+// to the IP and port of the socket, in network byte order.
+struct evconnlistener *listener_create(struct event_base *evbase,
+    unsigned short bindport, evconnlistener_cb cb, void *ctx,
+    unsigned int *ip, unsigned short *boundport);
+
+// Create a client connection to the given 6-byte ipport (4 byte IP, 2
+// byte port in network byte order).
+struct bufferevent *client_create(struct event_base *evbase,
+    unsigned char ipport[6], bufferevent_event_cb event_handler);
+
+// Create a client connection to a controller, with event_handler set as
+// the event callback.  It will be called when the connection succeeds or
+// fails.  This function calls the libevent main loop, so it will only return
+// when the program is finished.
+int controller_client(const char *controller_host,
+    unsigned short controller_port, bufferevent_event_cb event_handler);
+
+#endif

subproblem.h

@@ -0,0 +1,75 @@
+#ifndef __SUBPROBLEM_H__
+#define __SUBPROBLEM_H__
+
+extern "C" {
+#include <event2/bufferevent.h>
+}
+
+#include <NTL/ZZ.h>
+#include <ostream>
+#include <string.h>
+
+NTL_CLIENT
+
+#define SUBPROBLEM_DESC_LEN (2 + (WORDS*3 + 3 + 1)*sizeof(unsigned int))
+
+struct Subproblem {
+    unsigned short problemid;
+    ZZ base;
+    ZZ target;
+    ZZ modulus;
+    ZZ order;
+    unsigned int dpfreq;
+    unsigned char desc[1 + SUBPROBLEM_DESC_LEN];
+
+    Subproblem(unsigned short id, const ZZ &b, const ZZ &t, const ZZ &m,
+	    const ZZ &o, unsigned int dpf) : problemid(id), base(b),
+	    target(t), modulus(m), order(o), dpfreq(dpf) {
+	desc[0] = 'P';
+	memmove(desc+1, &problemid, 2);
+	BytesFromZZ(desc+3, base, WORDS*sizeof(unsigned int));
+	BytesFromZZ(desc+3+WORDS*sizeof(unsigned int), target,
+	    WORDS*sizeof(unsigned int));
+	BytesFromZZ(desc+3+2*WORDS*sizeof(unsigned int), modulus,
+	    WORDS*sizeof(unsigned int));
+	BytesFromZZ(desc+3+3*WORDS*sizeof(unsigned int), order,
+	    3*sizeof(unsigned int));
+	memmove(desc+3+(3*WORDS+3)*sizeof(unsigned int), &dpfreq,
+	    sizeof(unsigned int));
+	++problemid;
+    }
+
+    // Initilize the Subproblem from the binary description, *without*
+    // the leading 'P'
+    Subproblem(const unsigned char *descnoP) {
+	desc[0] = 'P';
+	memmove(desc+1, descnoP, SUBPROBLEM_DESC_LEN);
+	memmove(&problemid, desc+1, 2);
+	ZZFromBytes(base, desc+3, WORDS*sizeof(unsigned int));
+	ZZFromBytes(target, desc+3+WORDS*sizeof(unsigned int),
+	    WORDS*sizeof(unsigned int));
+	ZZFromBytes(modulus, desc+3+2*WORDS*sizeof(unsigned int),
+	    WORDS*sizeof(unsigned int));
+	ZZFromBytes(order, desc+3+3*WORDS*sizeof(unsigned int),
+	    3*sizeof(unsigned int));
+	memmove(&dpfreq, desc+3+(3*WORDS+3)*sizeof(unsigned int),
+	    sizeof(unsigned int));
+    }
+
+    // Dump for debug purposes
+    void dump(ostream &os) const {
+	os << "Subproblem " << problemid << "\n";
+	os << "base = " << base << "\n";
+	os << "target = " << target << "\n";
+	os << "modulus = " << modulus << "\n";
+	os << "order = " << order << "\n";
+	os << "dpfreq = " << dpfreq << "\n";
+    }
+
+    // Write the subproblem to the given bufferevent
+    void bev_write(struct bufferevent *bev) {
+	bufferevent_write(bev, desc, 1 + SUBPROBLEM_DESC_LEN);
+    }
+};
+
+#endif

worker.cc

@@ -0,0 +1,160 @@
+extern "C" {
+#include <event2/bufferevent.h>
+#include <event2/buffer.h>
+#include <event2/event.h>
+}
+
+#include <vector>
+#include <stdio.h>
+
+#include "evutils.h"
+#include "subproblem.h"
+
+typedef enum {
+    WRKCCSTATE_AWAITCMD,
+    WRKCCSTATE_RDPROBLEM,
+    WRKCCSTATE_RDDPNODES,
+    WRKCCSTATE_END
+} WrkCCState;
+
+struct WrkControllerConnInfo {
+    WrkCCState state;
+    unsigned char subproblem[SUBPROBLEM_DESC_LEN];
+    unsigned short num_dpnodes;
+
+    WrkControllerConnInfo() : state(WRKCCSTATE_AWAITCMD) {}
+};
+
+static struct WrkControllerState {
+    Subproblem *current_problem;
+    unsigned short num_expected_dpnodes;
+    vector<struct bufferevent *> dpnodes;
+    unsigned short num_connected_dpnodes;
+
+    WrkControllerState(): current_problem(NULL) {}
+} wrkctrlstate;
+
+static void dpconn_event_cb(struct bufferevent *bev, short events,
+    void *ctx)
+{
+    if (events & BEV_EVENT_CONNECTED) {
+	// We have successfully connected to the dpnode
+	++wrkctrlstate.num_connected_dpnodes;
+	if (wrkctrlstate.num_connected_dpnodes ==
+		wrkctrlstate.num_expected_dpnodes) {
+	    cerr << "Starting work\n";
+	    // start_working();
+	}
+    } else if (events & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) {
+	fprintf(stderr, "Closing connection to controller and restarting\n");
+	bufferevent_free(bev);
+	cerr << "Stopping work\n";
+	// stop_working();
+    }
+}
+
+static void controllerconn_reader(struct bufferevent *bev, void *ctx)
+{
+    struct evbuffer *input = bufferevent_get_input(bev);
+    WrkControllerConnInfo *info = (WrkControllerConnInfo *)ctx;
+    unsigned char cmd[1];
+
+    while(1) {
+	size_t len = evbuffer_get_length(input);
+	switch(info->state) {
+	    case WRKCCSTATE_AWAITCMD:
+		if (len < 1) return;
+		bufferevent_read(bev, cmd, 1);
+		switch(cmd[0]) {
+		    case 'P':
+			info->state = WRKCCSTATE_RDPROBLEM;
+			break;
+		    case 'S':
+			// stop_working();
+			break;
+		    default:
+			/* Unknown command received */
+			fprintf(stderr, "Unknown command in "
+				"controllerconn_reader: %c\n", cmd[0]);
+			info->state = WRKCCSTATE_END;
+			break;
+		}
+		break;
+
+	    case WRKCCSTATE_RDPROBLEM:
+		if (len < SUBPROBLEM_DESC_LEN+2) return;
+		// stop_working();
+		bufferevent_read(bev, info->subproblem, SUBPROBLEM_DESC_LEN);
+		bufferevent_read(bev, &(info->num_dpnodes), 2);
+		info->state = WRKCCSTATE_RDDPNODES;
+		/* FALLTHROUGH */
+
+	    case WRKCCSTATE_RDDPNODES:
+		if (len < 6*(info->num_dpnodes)) return;
+		wrkctrlstate.num_expected_dpnodes = info->num_dpnodes;
+		{
+		    unsigned short i;
+		    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]) << "." <<
+			            int(ipport[2]) << "." <<
+				    int(ipport[3]) << ":" <<
+				    ((ipport[4] << 8) + ipport[5]) <<
+				    "\n";
+			struct bufferevent *dpbev = client_create(
+				bufferevent_get_base(bev), ipport,
+				dpconn_event_cb);
+			if (dpbev) {
+			    wrkctrlstate.dpnodes.push_back(dpbev);
+			} else {
+			    // stop_working();
+			}
+		    }
+		}
+		info->state = WRKCCSTATE_AWAITCMD;
+		break;
+
+	    case WRKCCSTATE_END:
+		// Shut down
+		delete info;
+		event_base_loopbreak(bufferevent_get_base(bev));
+		bufferevent_free(bev);
+		return;
+	}
+    }
+}
+
+static void controllerconn_event_cb(struct bufferevent *bev, short events,
+    void *ctx)
+{
+    if (events & BEV_EVENT_CONNECTED) {
+	// We have successfully connected to the controller
+	char id[1] = { 'W' };
+
+        bufferevent_enable(bev, EV_READ|EV_WRITE);
+	bufferevent_write(bev, id, 1);
+	bufferevent_setcb(bev, controllerconn_reader, NULL,
+		controllerconn_event_cb, new WrkControllerConnInfo());
+    } 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));
+	bufferevent_free(bev);
+    }
+}
+
+int main(int argc, char **argv)
+{
+    if (argc != 3) {
+	fprintf(stderr, "Usage: %s controller_host controller_port\n", argv[0]);
+	return 1;
+    }
+
+    unsigned short controller_port = strtoul(argv[2], NULL, 10);
+
+    return controller_client(argv[1], controller_port,
+				controllerconn_event_cb);
+}