wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-22 13:00:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

Allow arbitrary number of connections (#13)

Browse Source 
* 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.
This commit is contained in:
Philipp Moritz
2016-09-07 20:19:37 -07:00
committed by Robert Nishihara
parent dd23a68f79
commit a62c0f8fac
13 changed files with 595 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+7 -7
View File
@@ -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
src/event_loop.c
+85
View File
@@ -0,0 +1,85 @@
#include "event_loop.h"
#include <assert.h>
#include <unistd.h>
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);
}
src/event_loop.h
+36
View File
@@ -0,0 +1,36 @@
#ifndef EVENT_LOOP_H
#define EVENT_LOOP_H
#include <poll.h>
#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
src/example.c
+8 -8
View File
@@ -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 <stdlib.h>
#include <getopt.h>
src/fling.c
+4 -4
View File
@@ -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;
src/fling.h
+16 -17
View File
@@ -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 <unistd.h>
#include <errno.h>
@@ -15,7 +15,7 @@
#include <sys/socket.h>
#include <sys/un.h>
// 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);
src/plasma.h
+17 -11
View File
@@ -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);
src/plasma_client.c
+8 -7
View File
@@ -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 <assert.h>
#include <stdlib.h>
@@ -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);
}
src/plasma_manager.c
+89 -100
View File
@@ -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 <stdio.h>
#include <stdlib.h>
@@ -21,46 +20,33 @@
#include <netinet/in.h>
#include <netdb.h>
#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) {
src/plasma_manager.h
+22 -27
View File
@@ -2,41 +2,36 @@
#define PLASMA_MANAGER_H
#include <poll.h>
#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
src/plasma_store.c
+54 -75
View File
@@ -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 <stdio.h>
@@ -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[]) {
src/utarray.h
+238
View File
@@ -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 <stddef.h> /* size_t */
#include <string.h> /* memset, etc */
#include <stdlib.h> /* 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 */
test/test.py
+11 -8
View File
@@ -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):