extern "C" { #include #include #include } #include #include #include #include #include #include #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) { struct evbuffer *input = bufferevent_get_input(bev); unsigned int dp[WORDS+6]; int num_read = 0; while(1) { size_t len = evbuffer_get_length(input); if (len < ((WORDS+6)*sizeof(unsigned int))) break; bufferevent_read(bev, dp, (WORDS+6)*sizeof(unsigned int)); ++num_read; } // cerr << num_read << " DPs read\n"; } 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, false); } 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 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::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, false); }