cudadl/evutils.cc

extern "C" {
#include <event2/listener.h>
#include <event2/bufferevent.h>
}

#include "evutils.h"

#include <iostream>

#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)
{
    unsigned int const hostname_len = 257+50;
    char hostname[hostname_len];
    if (gethostname(hostname, 256)) {
	// Could not determine hostname
	return 0;
    }
    hostname[256] = '\0';
#if defined(RIPPLE)
    // Use the 4*40Gbps bonded interface on RIPPLE
    strcat(hostname, "-data0");
#elif defined(GRAHAM)
    // Use the Infiniband interface on GRAHAM
    char temp_hostname[hostname_len];
    strcpy(temp_hostname, hostname);
    snprintf(hostname, sizeof(hostname), "%s%s", "ic-", temp_hostname);
#endif
    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, bool multithread)
{
    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|
	         (multithread ? LEV_OPT_THREADSAFE : 0),
        -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,
    bool multithread)
{
    // 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 |
	    (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) {
	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,
    bool multithread, void (*ev_loop_empty_cb)())
{
    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|(multithread ? BEV_OPT_THREADSAFE : 0));

    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);

    if (ev_loop_empty_cb == NULL) {
	event_base_dispatch(evbase);
    } else {
	while (true) {
	    event_base_loop(evbase, EVLOOP_ONCE);
	    ev_loop_empty_cb();
	    if (event_base_got_exit(evbase) || event_base_got_break(evbase)) {
		break;
	    }
	}
    }

    return 0;
}