From a62c0f8fac7c45f80cb47a6adda4205d97a610a9 Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Wed, 7 Sep 2016 20:19:37 -0700 Subject: [PATCH] Allow arbitrary number of connections (#13) * refactor plasma to use an event loop * unify comment style * Clean up Makefile flags. * Randomize socket names in tests so multiple copies of the tests can be run in parallel without conflict. --- Makefile | 14 +-- src/event_loop.c | 85 ++++++++++++++++ src/event_loop.h | 36 +++++++ src/example.c | 16 +-- src/fling.c | 8 +- src/fling.h | 33 +++--- src/plasma.h | 28 +++-- src/plasma_client.c | 15 +-- src/plasma_manager.c | 189 ++++++++++++++++------------------ src/plasma_manager.h | 49 ++++----- src/plasma_store.c | 129 ++++++++++------------- src/utarray.h | 238 +++++++++++++++++++++++++++++++++++++++++++ test/test.py | 19 ++-- 13 files changed, 595 insertions(+), 264 deletions(-) create mode 100644 src/event_loop.c create mode 100644 src/event_loop.h create mode 100644 src/utarray.h diff --git a/Makefile b/Makefile index f2c7ff9a5..64f66c980 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ CC = gcc -CFLAGS = -g -Wall +CFLAGS = -g -Wall --std=c99 -D_XOPEN_SOURCE=500 BUILD = build all: $(BUILD)/plasma_store $(BUILD)/plasma_manager $(BUILD)/plasma_client.so $(BUILD)/example @@ -7,14 +7,14 @@ all: $(BUILD)/plasma_store $(BUILD)/plasma_manager $(BUILD)/plasma_client.so $(B clean: rm -r $(BUILD)/* -$(BUILD)/plasma_store: src/plasma_store.c src/plasma.h src/fling.h src/fling.c - $(CC) $(CFLAGS) --std=c99 -D_XOPEN_SOURCE=500 src/plasma_store.c src/fling.c -o $(BUILD)/plasma_store +$(BUILD)/plasma_store: src/plasma_store.c src/plasma.h src/event_loop.h src/event_loop.c src/fling.h src/fling.c + $(CC) $(CFLAGS) src/plasma_store.c src/event_loop.c src/fling.c -o $(BUILD)/plasma_store -$(BUILD)/plasma_manager: src/plasma_manager.c src/plasma.h src/plasma_client.c src/fling.h src/fling.c - $(CC) $(CFLAGS) --std=c99 -D_XOPEN_SOURCE=500 src/plasma_manager.c src/plasma_client.c src/fling.c -o $(BUILD)/plasma_manager +$(BUILD)/plasma_manager: src/plasma_manager.c src/event_loop.h src/event_loop.c src/plasma.h src/plasma_client.c src/fling.h src/fling.c + $(CC) $(CFLAGS) src/plasma_manager.c src/event_loop.c src/plasma_client.c src/fling.c -o $(BUILD)/plasma_manager $(BUILD)/plasma_client.so: src/plasma_client.c src/fling.h src/fling.c - $(CC) $(CFLAGS) --std=c99 -D_XOPEN_SOURCE=500 src/plasma_client.c src/fling.c -fPIC -shared -o $(BUILD)/plasma_client.so + $(CC) $(CFLAGS) src/plasma_client.c src/fling.c -fPIC -shared -o $(BUILD)/plasma_client.so $(BUILD)/example: src/plasma_client.c src/plasma.h src/example.c src/fling.h src/fling.c - $(CC) $(CFLAGS) --std=c99 -D_XOPEN_SOURCE=500 src/plasma_client.c src/example.c src/fling.c -o $(BUILD)/example + $(CC) $(CFLAGS) src/plasma_client.c src/example.c src/fling.c -o $(BUILD)/example diff --git a/src/event_loop.c b/src/event_loop.c new file mode 100644 index 000000000..2fb6a856b --- /dev/null +++ b/src/event_loop.c @@ -0,0 +1,85 @@ +#include "event_loop.h" + +#include +#include + +UT_icd item_icd = { sizeof(event_loop_item), NULL, NULL, NULL }; +UT_icd poll_icd = { sizeof(struct pollfd), NULL, NULL, NULL }; + +/* Initializes the event loop. + * This function needs to be called before any other event loop function. */ +void event_loop_init(event_loop *loop) { + utarray_new(loop->items, &item_icd); + utarray_new(loop->waiting, &poll_icd); +} + +/* Add a new file descriptor fd to the event loop. + * This function sets a user defined type and id for the file descriptor + * which can be queried using event_loop_type and event_loop_id. The parameter + * events is the same as in http://linux.die.net/man/2/poll. + * Returns the index of the item in the event loop. */ +int64_t event_loop_attach(event_loop *loop, int type, data_connection* connection, int fd, int events) { + assert(utarray_len(loop->items) == utarray_len(loop->waiting)); + int64_t index = utarray_len(loop->items); + event_loop_item item = { .type = type }; + if (connection) { + item.connection = *connection; + } + utarray_push_back(loop->items, &item ); + struct pollfd waiting = { .fd = fd, .events = events }; + utarray_push_back(loop->waiting, &waiting); + return index; +} + +/* Detach a file descriptor from the event loop. + * This invalidates all other indices into the event loop items, but leaves + * the ids of the event loop items valid. */ +void event_loop_detach(event_loop *loop, int64_t index, int shall_close) { + struct pollfd *waiting_item = (struct pollfd*) utarray_eltptr(loop->waiting, index); + struct pollfd *waiting_back = (struct pollfd*) utarray_back(loop->waiting); + if (shall_close) { + close(waiting_item->fd); + } + *waiting_item = *waiting_back; + utarray_pop_back(loop->waiting); + + event_loop_item *items_item = (event_loop_item*) utarray_eltptr(loop->items, index); + event_loop_item *items_back = (event_loop_item*) utarray_back(loop->items); + *items_item = *items_back; + utarray_pop_back(loop->items); +} + +/* Poll the file descriptors associated to this event loop. + * See http://linux.die.net/man/2/poll */ +int event_loop_poll(event_loop *loop) { + return poll((struct pollfd*) utarray_front(loop->waiting), utarray_len(loop->waiting), -1); +} + +/* Get the total number of file descriptors participating in the event loop. */ +int64_t event_loop_size(event_loop *loop) { + return utarray_len(loop->waiting); +} + +/* Get the pollfd structure associated to a file descriptor participating in the event loop. */ +struct pollfd *event_loop_get(event_loop *loop, int64_t index) { + return (struct pollfd*) utarray_eltptr(loop->waiting, index); +} + +/* Set the data connection information for participant in the event loop. */ +void event_loop_set_connection(event_loop *loop, int64_t index, const data_connection* conn) { + event_loop_item *item = (event_loop_item*) utarray_eltptr(loop->items, index); + item->connection = *conn; +} + +/* Get the data connection information for participant in the event loop. */ +data_connection* event_loop_get_connection(event_loop *loop, int64_t index) { + event_loop_item *item = (event_loop_item*) utarray_eltptr(loop->items, index); + return &item->connection; +} + +/* Free the space associated to the event loop. + * Does not free the event_loop datastructure itself. */ +void event_loop_free(event_loop *loop) { + utarray_free(loop->items); + utarray_free(loop->waiting); +} diff --git a/src/event_loop.h b/src/event_loop.h new file mode 100644 index 000000000..fbb3aab97 --- /dev/null +++ b/src/event_loop.h @@ -0,0 +1,36 @@ +#ifndef EVENT_LOOP_H +#define EVENT_LOOP_H + +#include + +#include "utarray.h" +#include "plasma.h" +#include "plasma_manager.h" + +typedef struct { + /* The type of connection (e.g. redis, client, manager, data transfer). */ + int type; + /* If type is data transfer, this contains information about the status + * of the transfer. */ + data_connection connection; +} event_loop_item; + +typedef struct { + /* Array of event_loop_items that hold information for connections. */ + UT_array *items; + /* Array of file descriptors that are waiting, corresponding to items. */ + UT_array *waiting; +} event_loop; + +/* Event loop functions. */ +void event_loop_init(event_loop *loop); +void event_loop_free(event_loop *loop); +int64_t event_loop_attach(event_loop *loop, int type, data_connection* connection, int fd, int events); +void event_loop_detach(event_loop *loop, int64_t index, int shall_close); +int event_loop_poll(event_loop *loop); +int64_t event_loop_size(event_loop *loop); +struct pollfd *event_loop_get(event_loop *loop, int64_t index); +void event_loop_set_connection(event_loop *loop, int64_t index, const data_connection* conn); +data_connection *event_loop_get_connection(event_loop *loop, int64_t index); + +#endif diff --git a/src/example.c b/src/example.c index 1d8d74b5c..f42e26729 100644 --- a/src/example.c +++ b/src/example.c @@ -1,11 +1,11 @@ -// A simple example on how to use the plasma store -// -// Can be called in the following way: -// -// cd build -// ./plasma_store -s /tmp/plasma_socket -// ./example -s /tmp/plasma_socket -g -// ./example -s /tmp/plasma_socket -c -f +/* A simple example on how to use the plasma store + * + * Can be called in the following way: + * + * cd build + * ./plasma_store -s /tmp/plasma_socket + * ./example -s /tmp/plasma_socket -g + * ./example -s /tmp/plasma_socket -c -f */ #include #include diff --git a/src/fling.c b/src/fling.c index d614a051d..6b6e773de 100644 --- a/src/fling.c +++ b/src/fling.c @@ -26,7 +26,7 @@ int send_fd(int conn, int fd, const char* payload, int size) { header->cmsg_len = CMSG_LEN(sizeof(int)); *(int *)CMSG_DATA(header) = fd; - // send file descriptor and payload + /* send file descriptor and payload */ return sendmsg(conn, &msg, 0) != -1 && send(conn, payload, size, 0) == -1; } @@ -55,9 +55,9 @@ int recv_fd(int conn, char* payload, int size) { } } - // The sender sent us more than one file descriptor. We've closed - // them all to prevent fd leaks but notify the caller that we got - // a bad message. + /* The sender sent us more than one file descriptor. We've closed + * them all to prevent fd leaks but notify the caller that we got + * a bad message. */ if (oh_noes) { close(found_fd); errno = EBADMSG; diff --git a/src/fling.h b/src/fling.h index f6dc8a268..cc38f05ce 100644 --- a/src/fling.h +++ b/src/fling.h @@ -1,13 +1,13 @@ -// FLING: Exchanging file descriptors over sockets -// -// This is a little library for sending file descriptors over a socket -// between processes. The reason for doing that (as opposed to using -// filenames to share the files) is so (a) no files remain in the -// filesystem after all the processes terminate, (b) to make sure that -// there are no name collisions and (c) to be able to control who has -// access to the data. -// -// Most of the code is from https://github.com/sharvil/flingfd +/* FLING: Exchanging file descriptors over sockets + * + * This is a little library for sending file descriptors over a socket + * between processes. The reason for doing that (as opposed to using + * filenames to share the files) is so (a) no files remain in the + * filesystem after all the processes terminate, (b) to make sure that + * there are no name collisions and (c) to be able to control who has + * access to the data. + * + * Most of the code is from https://github.com/sharvil/flingfd */ #include #include @@ -15,7 +15,7 @@ #include #include -// This is neccessary for Mac OS X, see http://www.apuebook.com/faqs2e.html (10). +/* This is neccessary for Mac OS X, see http://www.apuebook.com/faqs2e.html (10). */ #if !defined(CMSG_SPACE) && !defined(CMSG_LEN) #define CMSG_SPACE(len) (__DARWIN_ALIGN32(sizeof(struct cmsghdr)) + __DARWIN_ALIGN32(len)) #define CMSG_LEN(len) (__DARWIN_ALIGN32(sizeof(struct cmsghdr)) + (len)) @@ -24,12 +24,11 @@ void init_msg(struct msghdr *msg, struct iovec *iov, char *buf, size_t buf_len); -// Send a file descriptor "fd" and a payload "payload" of size "size" -// over the socket "conn". Return 0 on success. +/* Send a file descriptor "fd" and a payload "payload" of size "size" + * over the socket "conn". Return 0 on success. */ int send_fd(int conn, int fd, const char* payload, int size); -// Receive a file descriptor and a payload of size up to "size" from a -// socket "conn". The payload will be written to "payload" and the file -// descriptor will be returned. Returns -1 on failure. +/* Receive a file descriptor and a payload of size up to "size" from a + * socket "conn". The payload will be written to "payload" and the file + * descriptor will be returned. Returns -1 on failure. */ int recv_fd(int conn, char* payload, int size); - diff --git a/src/plasma.h b/src/plasma.h index 9fdec86c2..eda883aa6 100644 --- a/src/plasma.h +++ b/src/plasma.h @@ -26,18 +26,22 @@ typedef struct { int64_t construct_duration; } plasma_object_info; -// Represents an object id hash, can hold a full SHA1 hash +/* Represents an object id hash, can hold a full SHA1 hash */ typedef struct { unsigned char id[20]; } plasma_id; -// these values must be in sync with the ones in plasma.py (can we have a test for that?) enum plasma_request_type { - PLASMA_CREATE, // create a new object - PLASMA_GET, // get an object - PLASMA_SEAL, // seal an object - PLASMA_TRANSFER, // request transfer to another store - PLASMA_DATA, // header for sending data + /* Create a new object. */ + PLASMA_CREATE, + /* Get an object. */ + PLASMA_GET, + /* seal an object */ + PLASMA_SEAL, + /* request transfer to another store */ + PLASMA_TRANSFER, + /* Header for sending data */ + PLASMA_DATA, }; typedef struct { @@ -49,8 +53,10 @@ typedef struct { } plasma_request; enum plasma_reply_type { - PLASMA_OBJECT, // the file descriptor represents an object - PLASMA_FUTURE, // the file descriptor represents a future + /* the file descriptor represents an object */ + PLASMA_OBJECT, + /* the file descriptor represents a future */ + PLASMA_FUTURE, }; typedef struct { @@ -65,10 +71,10 @@ typedef struct { int writable; } plasma_buffer; -// Connect to the local plasma store UNIX domain socket +/* Connect to the local plasma store UNIX domain socket */ int plasma_store_connect(const char* socket_name); -// Connect to a possibly remote plasma manager +/* Connect to a possibly remote plasma manager */ int plasma_manager_connect(const char* addr, int port); void plasma_create(int store, plasma_id object_id, int64_t size, void **data); diff --git a/src/plasma_client.c b/src/plasma_client.c index e8e160d70..d4f2d207f 100644 --- a/src/plasma_client.c +++ b/src/plasma_client.c @@ -1,4 +1,4 @@ -// PLASMA CLIENT: Client library for using the plasma store and manager +/* PLASMA CLIENT: Client library for using the plasma store and manager */ #include #include @@ -42,7 +42,7 @@ void plasma_get(int conn, plasma_id object_id, int64_t *size, void **data) { plasma_request req = { .type = PLASMA_GET, .object_id = object_id }; plasma_send(conn, &req); plasma_reply reply; - // the following loop is run at most twice + /* The following loop is run at most twice. */ int fd = recv_fd(conn, (char*)&reply, sizeof(plasma_reply)); if (reply.type == PLASMA_FUTURE) { int new_fd = recv_fd(fd, (char*)&reply, sizeof(plasma_reply)); @@ -74,17 +74,17 @@ int plasma_store_connect(const char* socket_name) { memset(&addr, 0, sizeof(addr)); addr.sun_family = AF_UNIX; strncpy(addr.sun_path, socket_name, sizeof(addr.sun_path)-1); - // Try to connect to the Plasma store. If unsuccessful, retry several times. + /* Try to connect to the Plasma store. If unsuccessful, retry several times. */ int connected_successfully = 0; for (int num_attempts = 0; num_attempts < 50; ++num_attempts) { if (connect(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0) { connected_successfully = 1; break; } - // Sleep for 100 milliseconds. + /* Sleep for 100 milliseconds. */ usleep(100000); } - // If we could not connect to the Plasma store, exit. + /* If we could not connect to the Plasma store, exit. */ if (!connected_successfully) { LOG_ERR("could not connect to store %s", socket_name); exit(-1); @@ -101,7 +101,7 @@ int plasma_manager_connect(const char* ip_addr, int port) { exit(-1); } - struct hostent *manager = gethostbyname(ip_addr); // TODO(pcm): cache this + struct hostent *manager = gethostbyname(ip_addr); /* TODO(pcm): cache this */ if (!manager) { LOG_ERR("plasma manager %s not found", ip_addr); exit(-1); @@ -125,7 +125,8 @@ void plasma_transfer(int manager, const char* addr, int port, plasma_id object_i char* end = NULL; for (int i = 0; i < 4; ++i) { req.addr[i] = strtol(end ? end : addr, &end, 10); - end += 1; // skip the '.' + /* skip the '.' */ + end += 1; } plasma_send(manager, &req); } diff --git a/src/plasma_manager.c b/src/plasma_manager.c index 90e0d4087..321f40931 100644 --- a/src/plasma_manager.c +++ b/src/plasma_manager.c @@ -1,10 +1,9 @@ -// PLASMA MANAGER: Local to a node, connects to other managers to send and -// receive objects from them -// -// The storage manager listens on its main listening port, and if a request for -// transfering an object to another object store comes in, it ships the data -// using a new connection to the target object manager. Also keeps a list of -// other object managers. +/* PLASMA MANAGER: Local to a node, connects to other managers to send and + * receive objects from them + * + * The storage manager listens on its main listening port, and if a request for + * transfering an object to another object store comes in, it ships the data + * using a new connection to the target object manager. */ #include #include @@ -21,46 +20,33 @@ #include #include +#include "event_loop.h" #include "plasma.h" #include "plasma_manager.h" -void init_manager_state(plasma_manager_state *s, const char* store_socket_name) { - memset(&s->waiting, 0, sizeof(s->waiting)); - memset(&s->conn, 0, sizeof(s->conn)); - s->num_conn = 0; +typedef struct { + /* Name of the socket connecting to local plasma store. */ + const char* store_socket_name; + /* Event loop. */ + event_loop* loop; +} plasma_manager_state; + +/* Initialize the plasma manager. This function initializes the event loop + * of the plasma manager, and stores the address 'store_socket_name' of + * the local plasma store socket. */ +void init_plasma_manager(plasma_manager_state *s, const char* store_socket_name) { + s->loop = malloc(sizeof(event_loop)); + event_loop_init(s->loop); s->store_socket_name = store_socket_name; } -// Add connection for sending commands or data to another plasma manager -// (returns the connection index). -int add_conn(plasma_manager_state* s, int type, int fd, int events, plasma_buffer* buf) { - s->waiting[s->num_conn].fd = fd; - s->waiting[s->num_conn].events = events; - s->conn[s->num_conn].type = type; - if (buf) { - s->conn[s->num_conn].buf = *buf; - } - s->conn[s->num_conn].cursor = 0; - return s->num_conn++; -} - -// Remove connection with index i by swapping it with the last element. -void remove_conn(plasma_manager_state* s, int i) { - memcpy(&s->waiting[i], &s->waiting[s->num_conn-1], sizeof(struct pollfd)); - memset(&s->waiting[s->num_conn-1], 0, sizeof(struct pollfd)); - memcpy(&s->conn[i], &s->conn[s->num_conn-1], sizeof(conn_state)); - memset(&s->conn[s->num_conn-1], 0, sizeof(conn_state)); -} - -#define BUFSIZE 4096 - -// Start transfering data to another object store manager. This establishes -// a connection to both the manager and the local object store and sends -// the data header to the other object manager. -void initiate_transfer(plasma_manager_state* state, plasma_request* req) { - int c = plasma_store_connect(state->store_socket_name); +/* Start transfering data to another object store manager. This establishes + * a connection to both the manager and the local object store and sends + * the data header to the other object manager. */ +void initiate_transfer(plasma_manager_state* s, plasma_request* req) { + int store_conn = plasma_store_connect(s->store_socket_name); plasma_buffer buf = { .object_id = req->object_id, .writable = 0 }; - plasma_get(c, req->object_id, &buf.size, &buf.data); + plasma_get(store_conn, req->object_id, &buf.size, &buf.data); char ip_addr[16]; snprintf(ip_addr, 32, "%d.%d.%d.%d", @@ -68,28 +54,27 @@ void initiate_transfer(plasma_manager_state* state, plasma_request* req) { req->addr[2], req->addr[3]); int fd = plasma_manager_connect(&ip_addr[0], req->port); - - add_conn(state, CONN_WRITE_DATA, fd, POLLOUT, &buf); + data_connection conn = { .type = DATA_CONNECTION_WRITE, .store_conn = store_conn, .buf = buf, .cursor = 0 }; + event_loop_attach(s->loop, CONNECTION_DATA, &conn, fd, POLLOUT); plasma_request manager_req = { .type = PLASMA_DATA, .object_id = req->object_id, .size = buf.size }; - LOG_INFO("filedescriptor is %d", fd); plasma_send(fd, &manager_req); } -void setup_data_connection(int conn_idx, plasma_manager_state* state, plasma_request* req) { - int store_conn = plasma_store_connect(state->store_socket_name); - state->conn[conn_idx].type = CONN_READ_DATA; - state->conn[conn_idx].store_conn = store_conn; - state->conn[conn_idx].buf.object_id = req->object_id; - state->conn[conn_idx].buf.size = req->size; - state->conn[conn_idx].buf.writable = 1; - plasma_create(store_conn, req->object_id, req->size, &state->conn[conn_idx].buf.data); - state->conn[conn_idx].cursor = 0; +/* Start reading data from another object manager. + * Initializes the object we are going to write to in the + * local plasma store and then switches the data socket to reading mode. */ +void start_reading_data(int64_t index, plasma_manager_state* s, plasma_request* req) { + int store_conn = plasma_store_connect(s->store_socket_name); + plasma_buffer buf = { .object_id = req->object_id, .size = req->size, .writable = 1 }; + plasma_create(store_conn, req->object_id, req->size, &buf.data); + data_connection conn = { .type = DATA_CONNECTION_READ, .store_conn = store_conn, .buf = buf, .cursor = 0 }; + event_loop_set_connection(s->loop, index, &conn); } -// Handle a command request that came in through a socket (transfering data, -// registering object managers, accepting incoming data). -void process_command(int conn_idx, plasma_manager_state* state, plasma_request* req) { +/* Handle a command request that came in through a socket (transfering data, + * or accepting incoming data). */ +void process_command(int64_t id, plasma_manager_state* state, plasma_request* req) { switch (req->type) { case PLASMA_TRANSFER: LOG_INFO("transfering object to manager with port %d", req->port); @@ -97,7 +82,7 @@ void process_command(int conn_idx, plasma_manager_state* state, plasma_request* break; case PLASMA_DATA: LOG_INFO("starting to stream data"); - setup_data_connection(conn_idx, state, req); + start_reading_data(id, state, req); break; default: LOG_ERR("invalid request %d", req->type); @@ -105,58 +90,59 @@ void process_command(int conn_idx, plasma_manager_state* state, plasma_request* } } -// Handle data or command event incoming on socket with index i. -void read_from_socket(plasma_manager_state* state, int i, plasma_request* req) { +/* Handle data or command event incoming on socket with index "index". */ +void read_from_socket(plasma_manager_state* state, struct pollfd *waiting, int64_t index, plasma_request* req) { ssize_t r, s; - switch (state->conn[i].type) { - case CONN_CONTROL: - r = read(state->waiting[i].fd, req, sizeof(plasma_request)); - if (r == 1) { + data_connection *conn = event_loop_get_connection(state->loop, index); + switch (conn->type) { + case DATA_CONNECTION_HEADER: + r = read(waiting->fd, req, sizeof(plasma_request)); + if (r == -1) { LOG_ERR("read error"); } else if (r == 0) { - LOG_INFO("connection with index %d disconnected", i); - remove_conn(state, i); + LOG_INFO("connection with id %" PRId64 " disconnected", index); + event_loop_detach(state->loop, index, 1); } else { - process_command(i, state, req); + process_command(index, state, req); } break; - case CONN_READ_DATA: - LOG_DEBUG("polled CONN_READ_DATA"); - r = read(state->waiting[i].fd, state->conn[i].buf.data + state->conn[i].cursor, BUFSIZE); + case DATA_CONNECTION_READ: + LOG_DEBUG("polled DATA_CONNECTION_READ"); + r = read(waiting->fd, conn->buf.data + conn->cursor, BUFSIZE); if (r == -1) { LOG_ERR("read error"); } else if (r == 0) { LOG_INFO("end of file"); } else { - state->conn[i].cursor += r; + conn->cursor += r; } if (r == 0) { - close(state->waiting[i].fd); - state->waiting[i].fd = 0; - state->waiting[i].events = 0; - plasma_seal(state->conn[i].store_conn, state->conn[i].buf.object_id); + LOG_DEBUG("reading on channel %" PRId64 " finished", index); + plasma_seal(conn->store_conn, conn->buf.object_id); + close(conn->store_conn); + event_loop_detach(state->loop, index, 1); } break; - case CONN_WRITE_DATA: - LOG_DEBUG("polled CONN_WRITE_DATA"); - s = state->conn[i].buf.size - state->conn[i].cursor; + case DATA_CONNECTION_WRITE: + LOG_DEBUG("polled DATA_CONNECTION_WRITE"); + s = conn->buf.size - conn->cursor; if (s > BUFSIZE) s = BUFSIZE; - r = write(state->waiting[i].fd, state->conn[i].buf.data + state->conn[i].cursor, s); + r = write(waiting->fd, conn->buf.data + conn->cursor, s); if (r != s) { if (r > 0) { - LOG_ERR("partial write on fd %d", state->waiting[i].fd); + LOG_ERR("partial write on fd %d", waiting->fd); } else { LOG_ERR("write error"); exit(-1); } } else { - state->conn[i].cursor += r; + conn->cursor += r; } if (r == 0) { - close(state->waiting[i].fd); - state->waiting[i].fd = 0; - state->waiting[i].events = 0; + LOG_DEBUG("writing on channel %" PRId64 " finished", index); + close(conn->store_conn); + event_loop_detach(state->loop, index, 1); } break; default: @@ -165,22 +151,23 @@ void read_from_socket(plasma_manager_state* state, int i, plasma_request* req) { } } -// Main event loop of the plasma manager. -void event_loop(int sock, plasma_manager_state* state) { - // Add listening socket. - add_conn(state, CONN_CONTROL, sock, POLLIN, NULL); +/* Main event loop of the plasma manager. */ +void run_event_loop(int sock, plasma_manager_state* s) { + /* Add listening socket. */ + event_loop_attach(s->loop, CONNECTION_LISTENER, NULL, sock, POLLIN); plasma_request req; while (1) { - int num_ready = poll(state->waiting, state->num_conn, -1); + int num_ready = event_loop_poll(s->loop); if (num_ready < 0) { LOG_ERR("poll failed"); exit(-1); } - for (int i = 0; i < state->num_conn; ++i) { - if (state->waiting[i].revents == 0) + for (int i = 0; i < event_loop_size(s->loop); ++i) { + struct pollfd *waiting = event_loop_get(s->loop, i); + if (waiting->revents == 0) continue; - if (state->waiting[i].fd == sock) { - // Handle new incoming connections. + if (waiting->fd == sock) { + /* Handle new incoming connections. */ int new_socket = accept(sock, NULL, NULL); if (new_socket < 0) { if (errno != EWOULDBLOCK) { @@ -189,10 +176,11 @@ void event_loop(int sock, plasma_manager_state* state) { } break; } - int conn_id = add_conn(state, CONN_CONTROL, new_socket, POLLIN, NULL); - LOG_INFO("new connection with id %d", conn_id); + data_connection conn = { .type = DATA_CONNECTION_HEADER }; + event_loop_attach(s->loop, CONNECTION_DATA, &conn, new_socket, POLLIN); + LOG_INFO("new connection with id %" PRId64, event_loop_size(s->loop)); } else { - read_from_socket(state, i, &req); + read_from_socket(s, waiting, i, &req); } } } @@ -209,12 +197,13 @@ void start_server(const char *store_socket_name, const char* master_addr, int po name.sin_port = htons(port); name.sin_addr.s_addr = htonl(INADDR_ANY); int on = 1; - // TODO(pcm): http://stackoverflow.com/q/1150635 + /* TODO(pcm): http://stackoverflow.com/q/1150635 */ if (ioctl(sock, FIONBIO, (char*) &on) < 0) { LOG_ERR("ioctl failed"); close(sock); exit(-1); } + setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &on, sizeof (on)); if (bind(sock, (struct sockaddr*) &name, sizeof(name)) < 0) { LOG_ERR("could not bind socket"); exit(-1); @@ -225,16 +214,16 @@ void start_server(const char *store_socket_name, const char* master_addr, int po exit(-1); } plasma_manager_state state; - init_manager_state(&state, store_socket_name); - event_loop(sock, &state); + init_plasma_manager(&state, store_socket_name); + run_event_loop(sock, &state); } int main(int argc, char* argv[]) { - // Socket name of the plasma store this manager is connected to. + /* Socket name of the plasma store this manager is connected to. */ char *store_socket_name = NULL; - // IP address of this node + /* IP address of this node. */ char *master_addr = NULL; - // Port number the manager should use + /* Port number the manager should use. */ int port; int c; while ((c = getopt(argc, argv, "s:m:p:")) != -1) { diff --git a/src/plasma_manager.h b/src/plasma_manager.h index efa326cb9..d0efe4b65 100644 --- a/src/plasma_manager.h +++ b/src/plasma_manager.h @@ -2,41 +2,36 @@ #define PLASMA_MANAGER_H #include +#include "utarray.h" -#define MAX_CONNECTIONS 2048 +/* The buffer size in bytes. Data will get transfered in multiples of this */ +#define BUFSIZE 4096 -enum conn_type { - // Connection to send commands to the manager. - CONN_CONTROL, - // Connection to send data to another manager. - CONN_WRITE_DATA, - // Connection to receive data from another manager. - CONN_READ_DATA +enum connection_type { + CONNECTION_REDIS, + CONNECTION_LISTENER, + CONNECTION_DATA +}; + +enum data_connection_type { + /* Connection to send commands and metadata to the manager. */ + DATA_CONNECTION_HEADER, + /* Connection to send data to another manager. */ + DATA_CONNECTION_WRITE, + /* Connection to receive data from another manager. */ + DATA_CONNECTION_READ }; typedef struct { - // Of type conn_type. + /* Of type data_connection_type. */ int type; - // Socket of the plasma store that is accessed for reading or writing data for - // this connection. + /* Local socket of the plasma store that is accessed for reading or writing + * data for this connection. */ int store_conn; - // Buffer this connection is reading from or writing to. + /* Buffer this connection is reading from or writing to. */ plasma_buffer buf; - // Current position in the buffer. + /* Current position in the buffer. */ int64_t cursor; -} conn_state; - -typedef struct { - // ID of this manager - int64_t manager_id; - // Name of the socket connecting to local plasma store. - const char* store_socket_name; - // Number of connections. - int num_conn; - // For the "poll" system call. - struct pollfd waiting[MAX_CONNECTIONS]; - // Status of connections (both control and data). - conn_state conn[MAX_CONNECTIONS]; -} plasma_manager_state; +} data_connection; #endif diff --git a/src/plasma_store.c b/src/plasma_store.c index d67420c24..1ddc6b06a 100644 --- a/src/plasma_store.c +++ b/src/plasma_store.c @@ -1,13 +1,13 @@ -// PLASMA STORE: This is a simple object store server process -// -// It accepts incoming client connections on a unix domain socket -// (name passed in via the -s option of the executable) and uses a -// single thread to serve the clients. Each client establishes a -// connection and can create objects, wait for objects and seal -// objects through that connection. -// -// It keeps a hash table that maps object_ids (which are 20 byte long, -// just enough to store and SHA1 hash) to memory mapped files. +/* PLASMA STORE: This is a simple object store server process + * + * It accepts incoming client connections on a unix domain socket + * (name passed in via the -s option of the executable) and uses a + * single thread to serve the clients. Each client establishes a + * connection and can create objects, wait for objects and seal + * objects through that connection. + * + * It keeps a hash table that maps object_ids (which are 20 byte long, + * just enough to store and SHA1 hash) to memory mapped files. */ #include @@ -24,77 +24,54 @@ #include "uthash.h" #include "fling.h" #include "plasma.h" +#include "event_loop.h" -#define MAX_NUM_CLIENTS 2048 +#define MAX_NUM_CLIENTS 100000 typedef struct { - // Number of clients connected. - int num_clients; - // Unique identifier for the clients. - int client_id[MAX_NUM_CLIENTS]; - // Data structure for polling. - struct pollfd waiting[MAX_NUM_CLIENTS]; + /* Event loop for the plasma store. */ + event_loop *loop; } plasma_store_state; void init_state(plasma_store_state* s) { - memset(&s->waiting, 0, sizeof(s->waiting)); - memset(&s->client_id, 0, sizeof(s->client_id)); - s->num_clients = 0; -} - -int add_client(plasma_store_state* s, int fd) { - static int curr_id = 0; - s->waiting[s->num_clients].fd = fd; - s->waiting[s->num_clients].events = POLLIN; - s->client_id[s->num_clients] = curr_id; - s->num_clients += 1; - return curr_id++; -} - -// Remove the client at index i by swapping it with the -// client at index num_clients-1 and zeroing the latter out. -void remove_client(plasma_store_state* s, int i) { - memcpy(&s->waiting[i], &s->waiting[s->num_clients-1], sizeof(struct pollfd)); - memset(&s->waiting[s->num_clients-1], 0, sizeof(struct pollfd)); - s->client_id[i] = s->client_id[s->num_clients-1]; - s->client_id[s->num_clients-1] = 0; - s->num_clients -= 1; + s->loop = malloc(sizeof(event_loop)); + event_loop_init(s->loop); } typedef struct { - // Object id of this object. + /* Object id of this object. */ plasma_id object_id; - // Object info like size, creation time and owner. + /* Object info like size, creation time and owner. */ plasma_object_info info; - // Memory mapped file containing the object. + /* Memory mapped file containing the object. */ int fd; - // Handle for the uthash table. + /* Handle for the uthash table. */ UT_hash_handle handle; } object_table_entry; -// objects that are still being written by their owner process +/* Objects that are still being written by their owner process. */ object_table_entry* open_objects = NULL; -// Objects that have already been sealed by their owner process and -// can now be shared with other processes. +/* Objects that have already been sealed by their owner process and + * can now be shared with other processes. */ object_table_entry* sealed_objects = NULL; typedef struct { - // Object id of this object. + /* Object id of this object. */ plasma_id object_id; - // Number of processes waiting for the object. + /* Number of processes waiting for the object. */ int num_waiting; - // Socket connections to waiting clients. + /* Socket connections to waiting clients. */ int conn[MAX_NUM_CLIENTS]; - // Handle for the uthash table. + /* Handle for the uthash table. */ UT_hash_handle handle; } object_notify_entry; -// Objects that processes are waiting for. +/* Objects that processes are waiting for. */ object_notify_entry* objects_notify = NULL; -// Create a buffer. This is creating a temporary file and then -// immediately unlinking it so we do not leave traces in the system. +/* Create a buffer. This is creating a temporary file and then + * immediately unlinking it so we do not leave traces in the system. */ int create_buffer(int64_t size) { static char template[] = "/tmp/plasmaXXXXXX"; char file_name[32]; @@ -118,9 +95,9 @@ int create_buffer(int64_t size) { return fd; } -// Create a new object buffer in the hash table. +/* Create a new object buffer in the hash table. */ void create_object(int conn, plasma_request* req) { - LOG_INFO("creating object"); // TODO(pcm): add object_id here + LOG_INFO("creating object"); /* TODO(pcm): add object_id here */ int fd = create_buffer(req->size); if (fd < 0) { LOG_ERR("could not create shared memory buffer"); @@ -129,14 +106,14 @@ void create_object(int conn, plasma_request* req) { object_table_entry *entry = malloc(sizeof(object_table_entry)); memcpy(&entry->object_id, &req->object_id, 20); entry->info.size = req->size; - // TODO(pcm): set the other fields + /* TODO(pcm): set the other fields */ entry->fd = fd; HASH_ADD(handle, open_objects, object_id, sizeof(plasma_id), entry); plasma_reply reply = { PLASMA_OBJECT, req->size }; send_fd(conn, fd, (char*) &reply, sizeof(plasma_reply)); } -// Get an object from the hash table. +/* Get an object from the hash table. */ void get_object(int conn, plasma_request* req) { object_table_entry *entry; HASH_FIND(handle, sealed_objects, &req->object_id, sizeof(plasma_id), entry); @@ -157,19 +134,19 @@ void get_object(int conn, plasma_request* req) { } } -// Seal an object that has been created in the hash table. +/* Seal an object that has been created in the hash table. */ void seal_object(int conn, plasma_request* req) { LOG_INFO("sealing object"); // TODO(pcm): add object_id here object_table_entry *entry; HASH_FIND(handle, open_objects, &req->object_id, sizeof(plasma_id), entry); if (!entry) { - return; // TODO(pcm): return error + return; /* TODO(pcm): return error */ } HASH_DELETE(handle, open_objects, entry); int64_t size = entry->info.size; int fd = entry->fd; HASH_ADD(handle, sealed_objects, object_id, sizeof(plasma_id), entry); - // Inform processes that the object is ready now. + /* Inform processes that the object is ready now. */ object_notify_entry* notify_entry; HASH_FIND(handle, objects_notify, &req->object_id, sizeof(plasma_id), notify_entry); if (!notify_entry) { @@ -178,6 +155,7 @@ void seal_object(int conn, plasma_request* req) { plasma_reply reply = { PLASMA_OBJECT, size }; for (int i = 0; i < notify_entry->num_waiting; ++i) { send_fd(notify_entry->conn[i], fd, (char*) &reply, sizeof(plasma_reply)); + close(notify_entry->conn[i]); } HASH_DELETE(handle, objects_notify, notify_entry); free(notify_entry); @@ -200,23 +178,24 @@ void process_event(int conn, plasma_request* req) { } } -void event_loop(int socket) { +void run_event_loop(int socket) { plasma_store_state state; init_state(&state); - add_client(&state, socket); + event_loop_attach(state.loop, 0, NULL, socket, POLLIN); plasma_request req; while (1) { - int num_ready = poll(state.waiting, state.num_clients, -1); + int num_ready = event_loop_poll(state.loop); if (num_ready < 0) { LOG_ERR("poll failed"); exit(-1); } - for (int i = 0; i < state.num_clients; ++i) { - if (state.waiting[i].revents == 0) + for (int i = 0; i < event_loop_size(state.loop); ++i) { + struct pollfd *waiting = event_loop_get(state.loop, i); + if (waiting->revents == 0) continue; - if (state.waiting[i].fd == socket) { + if (waiting->fd == socket) { while (1) { - // Handle new incoming connections. + /* Handle new incoming connections. */ int new_socket = accept(socket, NULL, NULL); if (new_socket < 0) { if (errno != EWOULDBLOCK) { @@ -225,19 +204,19 @@ void event_loop(int socket) { } break; } - int client_id = add_client(&state, new_socket); - LOG_INFO("adding new client with id %d", client_id); + event_loop_attach(state.loop, 0, NULL, new_socket, POLLIN); + LOG_INFO("adding new client"); } } else { - int r = read(state.waiting[i].fd, &req, sizeof(plasma_request)); + int r = read(waiting->fd, &req, sizeof(plasma_request)); if (r == -1) { LOG_ERR("read error"); continue; } else if (r == 0) { - LOG_INFO("client with id %d disconnected", state.client_id[i]); - remove_client(&state, i); + LOG_INFO("connection %d disconnected", i); + event_loop_detach(state.loop, i, 1); } else { - process_event(state.waiting[i].fd, &req); + process_event(waiting->fd, &req); } } } @@ -256,7 +235,7 @@ void start_server(char* socket_name) { close(fd); exit(-1); } - // TODO(pcm): http://stackoverflow.com/q/1150635 + /* TODO(pcm): http://stackoverflow.com/q/1150635 */ if (ioctl(fd, FIONBIO, (char*) &on) < 0) { LOG_ERR("ioctl failed"); close(fd); @@ -269,7 +248,7 @@ void start_server(char* socket_name) { unlink(socket_name); bind(fd, (struct sockaddr*)&addr, sizeof(addr)); listen(fd, 5); - event_loop(fd); + run_event_loop(fd); } int main(int argc, char* argv[]) { diff --git a/src/utarray.h b/src/utarray.h new file mode 100644 index 000000000..979e99e98 --- /dev/null +++ b/src/utarray.h @@ -0,0 +1,238 @@ +/* +Copyright (c) 2008-2016, Troy D. Hanson http://troydhanson.github.com/uthash/ +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS +IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED +TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A +PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER +OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +/* a dynamic array implementation using macros + */ +#ifndef UTARRAY_H +#define UTARRAY_H + +#define UTARRAY_VERSION 2.0.1 + +#ifdef __GNUC__ +#define _UNUSED_ __attribute__ ((__unused__)) +#else +#define _UNUSED_ +#endif + +#include /* size_t */ +#include /* memset, etc */ +#include /* exit */ + +#ifndef oom +#define oom() exit(-1) +#endif + +typedef void (ctor_f)(void *dst, const void *src); +typedef void (dtor_f)(void *elt); +typedef void (init_f)(void *elt); +typedef struct { + size_t sz; + init_f *init; + ctor_f *copy; + dtor_f *dtor; +} UT_icd; + +typedef struct { + unsigned i,n;/* i: index of next available slot, n: num slots */ + UT_icd icd; /* initializer, copy and destructor functions */ + char *d; /* n slots of size icd->sz*/ +} UT_array; + +#define utarray_init(a,_icd) do { \ + memset(a,0,sizeof(UT_array)); \ + (a)->icd = *(_icd); \ +} while(0) + +#define utarray_done(a) do { \ + if ((a)->n) { \ + if ((a)->icd.dtor) { \ + unsigned _ut_i; \ + for(_ut_i=0; _ut_i < (a)->i; _ut_i++) { \ + (a)->icd.dtor(utarray_eltptr(a,_ut_i)); \ + } \ + } \ + free((a)->d); \ + } \ + (a)->n=0; \ +} while(0) + +#define utarray_new(a,_icd) do { \ + (a) = (UT_array*)malloc(sizeof(UT_array)); \ + if ((a) == NULL) oom(); \ + utarray_init(a,_icd); \ +} while(0) + +#define utarray_free(a) do { \ + utarray_done(a); \ + free(a); \ +} while(0) + +#define utarray_reserve(a,by) do { \ + if (((a)->i+(by)) > (a)->n) { \ + char *utarray_tmp; \ + while (((a)->i+(by)) > (a)->n) { (a)->n = ((a)->n ? (2*(a)->n) : 8); } \ + utarray_tmp=(char*)realloc((a)->d, (a)->n*(a)->icd.sz); \ + if (utarray_tmp == NULL) oom(); \ + (a)->d=utarray_tmp; \ + } \ +} while(0) + +#define utarray_push_back(a,p) do { \ + utarray_reserve(a,1); \ + if ((a)->icd.copy) { (a)->icd.copy( _utarray_eltptr(a,(a)->i++), p); } \ + else { memcpy(_utarray_eltptr(a,(a)->i++), p, (a)->icd.sz); }; \ +} while(0) + +#define utarray_pop_back(a) do { \ + if ((a)->icd.dtor) { (a)->icd.dtor( _utarray_eltptr(a,--((a)->i))); } \ + else { (a)->i--; } \ +} while(0) + +#define utarray_extend_back(a) do { \ + utarray_reserve(a,1); \ + if ((a)->icd.init) { (a)->icd.init(_utarray_eltptr(a,(a)->i)); } \ + else { memset(_utarray_eltptr(a,(a)->i),0,(a)->icd.sz); } \ + (a)->i++; \ +} while(0) + +#define utarray_len(a) ((a)->i) + +#define utarray_eltptr(a,j) (((j) < (a)->i) ? _utarray_eltptr(a,j) : NULL) +#define _utarray_eltptr(a,j) ((a)->d + ((a)->icd.sz * (j))) + +#define utarray_insert(a,p,j) do { \ + if ((j) > (a)->i) utarray_resize(a,j); \ + utarray_reserve(a,1); \ + if ((j) < (a)->i) { \ + memmove( _utarray_eltptr(a,(j)+1), _utarray_eltptr(a,j), \ + ((a)->i - (j))*((a)->icd.sz)); \ + } \ + if ((a)->icd.copy) { (a)->icd.copy( _utarray_eltptr(a,j), p); } \ + else { memcpy(_utarray_eltptr(a,j), p, (a)->icd.sz); }; \ + (a)->i++; \ +} while(0) + +#define utarray_inserta(a,w,j) do { \ + if (utarray_len(w) == 0) break; \ + if ((j) > (a)->i) utarray_resize(a,j); \ + utarray_reserve(a,utarray_len(w)); \ + if ((j) < (a)->i) { \ + memmove(_utarray_eltptr(a,(j)+utarray_len(w)), \ + _utarray_eltptr(a,j), \ + ((a)->i - (j))*((a)->icd.sz)); \ + } \ + if ((a)->icd.copy) { \ + unsigned _ut_i; \ + for(_ut_i=0;_ut_i<(w)->i;_ut_i++) { \ + (a)->icd.copy(_utarray_eltptr(a, (j) + _ut_i), _utarray_eltptr(w, _ut_i)); \ + } \ + } else { \ + memcpy(_utarray_eltptr(a,j), _utarray_eltptr(w,0), \ + utarray_len(w)*((a)->icd.sz)); \ + } \ + (a)->i += utarray_len(w); \ +} while(0) + +#define utarray_resize(dst,num) do { \ + unsigned _ut_i; \ + if ((dst)->i > (unsigned)(num)) { \ + if ((dst)->icd.dtor) { \ + for (_ut_i = (num); _ut_i < (dst)->i; ++_ut_i) { \ + (dst)->icd.dtor(_utarray_eltptr(dst, _ut_i)); \ + } \ + } \ + } else if ((dst)->i < (unsigned)(num)) { \ + utarray_reserve(dst, (num) - (dst)->i); \ + if ((dst)->icd.init) { \ + for (_ut_i = (dst)->i; _ut_i < (unsigned)(num); ++_ut_i) { \ + (dst)->icd.init(_utarray_eltptr(dst, _ut_i)); \ + } \ + } else { \ + memset(_utarray_eltptr(dst, (dst)->i), 0, (dst)->icd.sz*((num) - (dst)->i)); \ + } \ + } \ + (dst)->i = (num); \ +} while(0) + +#define utarray_concat(dst,src) do { \ + utarray_inserta(dst, src, utarray_len(dst)); \ +} while(0) + +#define utarray_erase(a,pos,len) do { \ + if ((a)->icd.dtor) { \ + unsigned _ut_i; \ + for (_ut_i = 0; _ut_i < (len); _ut_i++) { \ + (a)->icd.dtor(utarray_eltptr(a, (pos) + _ut_i)); \ + } \ + } \ + if ((a)->i > ((pos) + (len))) { \ + memmove(_utarray_eltptr(a, pos), _utarray_eltptr(a, (pos) + (len)), \ + ((a)->i - ((pos) + (len))) * (a)->icd.sz); \ + } \ + (a)->i -= (len); \ +} while(0) + +#define utarray_renew(a,u) do { \ + if (a) utarray_clear(a); \ + else utarray_new(a, u); \ +} while(0) + +#define utarray_clear(a) do { \ + if ((a)->i > 0) { \ + if ((a)->icd.dtor) { \ + unsigned _ut_i; \ + for(_ut_i=0; _ut_i < (a)->i; _ut_i++) { \ + (a)->icd.dtor(_utarray_eltptr(a, _ut_i)); \ + } \ + } \ + (a)->i = 0; \ + } \ +} while(0) + +#define utarray_sort(a,cmp) do { \ + qsort((a)->d, (a)->i, (a)->icd.sz, cmp); \ +} while(0) + +#define utarray_find(a,v,cmp) bsearch((v),(a)->d,(a)->i,(a)->icd.sz,cmp) + +#define utarray_front(a) (((a)->i) ? (_utarray_eltptr(a,0)) : NULL) +#define utarray_next(a,e) (((e)==NULL) ? utarray_front(a) : ((((a)->i) > (utarray_eltidx(a,e)+1)) ? _utarray_eltptr(a,utarray_eltidx(a,e)+1) : NULL)) +#define utarray_prev(a,e) (((e)==NULL) ? utarray_back(a) : ((utarray_eltidx(a,e) > 0) ? _utarray_eltptr(a,utarray_eltidx(a,e)-1) : NULL)) +#define utarray_back(a) (((a)->i) ? (_utarray_eltptr(a,(a)->i-1)) : NULL) +#define utarray_eltidx(a,e) (((char*)(e) >= (a)->d) ? (((char*)(e) - (a)->d)/(a)->icd.sz) : -1) + +/* last we pre-define a few icd for common utarrays of ints and strings */ +static void utarray_str_cpy(void *dst, const void *src) { + char **_src = (char**)src, **_dst = (char**)dst; + *_dst = (*_src == NULL) ? NULL : strdup(*_src); +} +static void utarray_str_dtor(void *elt) { + char **eltc = (char**)elt; + if (*eltc != NULL) free(*eltc); +} +static const UT_icd ut_str_icd _UNUSED_ = {sizeof(char*),NULL,utarray_str_cpy,utarray_str_dtor}; +static const UT_icd ut_int_icd _UNUSED_ = {sizeof(int),NULL,NULL,NULL}; +static const UT_icd ut_ptr_icd _UNUSED_ = {sizeof(void*),NULL,NULL,NULL}; + + +#endif /* UTARRAY_H */ diff --git a/test/test.py b/test/test.py index 82ba406a8..0fb54d3f2 100644 --- a/test/test.py +++ b/test/test.py @@ -17,9 +17,10 @@ class TestPlasmaClient(unittest.TestCase): def setUp(self): # Start Plasma. plasma_store_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_store") - self.p = subprocess.Popen([plasma_store_executable, "-s", "/tmp/store"]) + store_name = "/tmp/store{}".format(random.randint(0, 10000)) + self.p = subprocess.Popen([plasma_store_executable, "-s", store_name]) # Connect to Plasma. - self.plasma_client = plasma.PlasmaClient("/tmp/store") + self.plasma_client = plasma.PlasmaClient(store_name) def tearDown(self): # Kill the plasma store process. @@ -67,18 +68,20 @@ class TestPlasmaManager(unittest.TestCase): def setUp(self): # Start two PlasmaStores. plasma_store_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_store") - self.p2 = subprocess.Popen([plasma_store_executable, "-s", "/tmp/store1"]) - self.p3 = subprocess.Popen([plasma_store_executable, "-s", "/tmp/store2"]) + store_name1 = "/tmp/store{}".format(random.randint(0, 10000)) + store_name2 = "/tmp/store{}".format(random.randint(0, 10000)) + self.p2 = subprocess.Popen([plasma_store_executable, "-s", store_name1]) + self.p3 = subprocess.Popen([plasma_store_executable, "-s", store_name2]) # Start two PlasmaManagers. self.port1 = random.randint(10000, 50000) self.port2 = random.randint(10000, 50000) plasma_manager_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_manager") - self.p4 = subprocess.Popen([plasma_manager_executable, "-s", "/tmp/store1", "-m", "127.0.0.1", "-p", str(self.port1)]) - self.p5 = subprocess.Popen([plasma_manager_executable, "-s", "/tmp/store2", "-m", "127.0.0.1", "-p", str(self.port2)]) + self.p4 = subprocess.Popen([plasma_manager_executable, "-s", store_name1, "-m", "127.0.0.1", "-p", str(self.port1)]) + self.p5 = subprocess.Popen([plasma_manager_executable, "-s", store_name2, "-m", "127.0.0.1", "-p", str(self.port2)]) time.sleep(0.1) # Connect two PlasmaClients. - self.client1 = plasma.PlasmaClient("/tmp/store1", "127.0.0.1", self.port1) - self.client2 = plasma.PlasmaClient("/tmp/store2", "127.0.0.1", self.port2) + self.client1 = plasma.PlasmaClient(store_name1, "127.0.0.1", self.port1) + self.client2 = plasma.PlasmaClient(store_name2, "127.0.0.1", self.port2) time.sleep(0.5) def tearDown(self):