From 23327a18e0f8ecb6de56ad646d058fbd2c420573 Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Wed, 17 Aug 2016 12:54:34 -0700 Subject: [PATCH 1/3] add plasma manager --- Makefile | 5 +- lib/python/plasma.py | 25 ++- src/plasma.h | 10 ++ src/plasma_manager.c | 392 +++++++++++++++++++++++++++++++++++++++++++ src/plasma_store.c | 53 +++--- test/nameserver.py | 33 ++++ 6 files changed, 495 insertions(+), 23 deletions(-) create mode 100644 src/plasma_manager.c create mode 100644 test/nameserver.py diff --git a/Makefile b/Makefile index e303d65b5..f2c7ff9a5 100644 --- a/Makefile +++ b/Makefile @@ -2,7 +2,7 @@ CC = gcc CFLAGS = -g -Wall BUILD = build -all: $(BUILD)/plasma_store $(BUILD)/plasma_client.so $(BUILD)/example +all: $(BUILD)/plasma_store $(BUILD)/plasma_manager $(BUILD)/plasma_client.so $(BUILD)/example clean: rm -r $(BUILD)/* @@ -10,6 +10,9 @@ clean: $(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_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_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 diff --git a/lib/python/plasma.py b/lib/python/plasma.py index 69361d63c..83de5f1b8 100644 --- a/lib/python/plasma.py +++ b/lib/python/plasma.py @@ -9,8 +9,13 @@ ID = ctypes.c_ubyte * 20 class PlasmaID(ctypes.Structure): _fields_ = [("plasma_id", ID)] -# these must be in sync with plasma_request_type in plasma.h +# these must be in sync with plasma_request_type in plasma.h (can we have a test for that?) PLASMA_CREATE = 0 +PLASMA_GET = 1 +PLASMA_SEAL = 2 +PLASMA_TRANSFER = 3 +PLASMA_DATA = 4 +PLASMA_REGISTER = 5 class PlasmaRequest(ctypes.Structure): _fields_ = [("type", ctypes.c_int), @@ -32,6 +37,24 @@ def make_plasma_id(string): object_id = map(ord, string) return PlasmaID(plasma_id=ID(*object_id)) +class PlasmaManager(object): + + def __init__(self, addr, port): + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.connect((addr, port)) + + def register(self, manager_id, addr, port): + "Register another object manager." + req = PlasmaRequest(type=PLASMA_REGISTER, manager_id=manager_id, + addr=Addr(*map(int, addr.split("."))), port=port) + self.sock.send(buffer(req)[:]) + + def transfer(self, manager_id, object_id): + "Transfer local object with id object_id to manager with id manager_id." + req = PlasmaRequest(type=PLASMA_TRANSFER, manager_id=manager_id, + object_id=make_plasma_id(object_id)) + self.sock.send(buffer(req)[:]) + class PlasmaClient(object): def __init__(self, socket_name): plasma_client_library = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../../build/plasma_client.so") diff --git a/src/plasma.h b/src/plasma.h index 42ac2c6f5..2e4fbfdcd 100644 --- a/src/plasma.h +++ b/src/plasma.h @@ -6,6 +6,13 @@ #include #include +#ifdef NDEBUG + #define LOG_DEBUG(M, ...) +#else + #define LOG_DEBUG(M, ...) \ + fprintf(stderr, "[DEBUG] (%s:%d) " M "\n", __FILE__, __LINE__, ##__VA_ARGS__) +#endif + #define LOG_ERR(M, ...) \ fprintf(stderr, "[ERROR] (%s:%d: errno: %s) " M "\n", \ __FILE__, __LINE__, errno == 0 ? "None" : strerror(errno), ##__VA_ARGS__) @@ -24,6 +31,7 @@ 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 @@ -64,4 +72,6 @@ plasma_buffer plasma_create(int conn, plasma_id object_id, int64_t size); plasma_buffer plasma_get(int conn, plasma_id object_id); void plasma_seal(int fd, plasma_id object_id); +void plasma_send(int fd, plasma_request *req); + #endif diff --git a/src/plasma_manager.c b/src/plasma_manager.c new file mode 100644 index 000000000..7ad2585b6 --- /dev/null +++ b/src/plasma_manager.c @@ -0,0 +1,392 @@ +// 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. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "plasma.h" + +#define MAX_CONNECTIONS 2048 +#define MAX_NUM_MANAGERS 1024 + +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 +}; + +typedef struct { + // Unique identifier for the connection. + int id; + // Of type conn_type. + int type; + // 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. + plasma_buffer buf; + // Current position in the buffer. + int64_t cursor; +} conn_state; + +typedef struct { + // Address of the manager. + struct sockaddr_in name; + // Is this manager connected? + int connected; +} manager_state; + +typedef struct { + // 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]; + // Other plasma managers in the cluster. + manager_state managers[MAX_NUM_MANAGERS]; +} plasma_manager_state; + +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)); + memset(&s->managers, 0, sizeof(s->managers)); + s->num_conn = 0; + s->store_socket_name = store_socket_name; +} + +#define h_addr h_addr_list[0] + +// Add name info for another plasma manager from the cluster. +void add_manager(plasma_manager_state *s, int manager_id, char *ip_address, int port) { + assert(ip_address); + assert(s); + struct hostent *manager = gethostbyname(ip_address); + if (!manager) { + LOG_ERR("plasma manager %s not found", ip_address); + exit(-1); + } + s->managers[manager_id].connected = 1; + struct sockaddr_in *name = &s->managers[manager_id].name; + name->sin_family = AF_INET; + bcopy(manager->h_addr, &name->sin_addr.s_addr, manager->h_length); + name->sin_port = htons(port); +} + +// Add connection for sending commands or data to another plasma manager +// (returns the connection id). +int add_conn(plasma_manager_state* s, int type, int fd, int events, plasma_buffer* buf) { + static int conn_id = 0; + s->waiting[s->num_conn].fd = fd; + s->waiting[s->num_conn].events = events; + s->conn[s->num_conn].id = conn_id; + s->conn[s->num_conn].type = type; + if (buf) { + s->conn[s->num_conn].buf = *buf; + } + s->conn[s->num_conn].cursor = 0; + s->num_conn += 1; + return conn_id++; +} + +// 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 manager_id = req->manager_id; + int c = plasma_store_connect(state->store_socket_name); + plasma_buffer buf = plasma_get(c, req->object_id); + + int fd = socket(PF_INET, SOCK_STREAM, 0); + if (fd < 0) { + LOG_ERR("could not create socket"); + exit(-1); + } + int r = connect(fd, (struct sockaddr*) &state->managers[manager_id].name, sizeof(state->managers[manager_id].name)); + if (r < 0) { + LOG_ERR("could not establish connection to manager with id %d", manager_id); + exit(-1); + } + + add_conn(state, CONN_WRITE_DATA, fd, POLLOUT, &buf); + + 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 = plasma_create(store_conn, req->object_id, req->size); + state->conn[conn_idx].cursor = 0; +} + +// 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) { + switch (req->type) { + case PLASMA_TRANSFER: + LOG_INFO("transfering object to manager with id %d", req->manager_id); + initiate_transfer(state, req); + break; + case PLASMA_REGISTER: { + char buff[16]; + snprintf(buff, 32, "%d.%d.%d.%d", + req->addr[0], req->addr[1], + req->addr[2], req->addr[3]); + if (req->manager_id >= MAX_NUM_MANAGERS) { + LOG_ERR("manager_id %d out of bounds", req->manager_id); + } else { + add_manager(state, req->manager_id, buff, req->port); + LOG_INFO("registering %s:%d with id %d", buff, req->port, req->manager_id); + } + } + break; + case PLASMA_DATA: + LOG_INFO("starting to stream data"); + setup_data_connection(conn_idx, state, req); + break; + default: + LOG_ERR("invalid request %d", req->type); + exit(-1); + } +} + +// Handle data or command event incoming on socket with index i. +void read_from_socket(plasma_manager_state* state, int i, 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) { + LOG_ERR("read error"); + } else if (r == 0) { + LOG_INFO("connection with id %d disconnected", state->conn[i].id); + remove_conn(state, i); + } else { + process_command(i, 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); + if (r == -1) { + LOG_ERR("read error"); + } else if (r == 0) { + LOG_INFO("end of file"); + } else { + state->conn[i].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); + } + break; + case CONN_WRITE_DATA: + LOG_DEBUG("polled CONN_WRITE_DATA"); + s = state->conn[i].buf.size - state->conn[i].cursor; + if (s > BUFSIZE) + s = BUFSIZE; + r = write(state->waiting[i].fd, state->conn[i].buf.data + state->conn[i].cursor, s); + if (r != s) { + if (r > 0) { + LOG_ERR("partial write on fd %d", state->waiting[i].fd); + } else { + LOG_ERR("write error"); + exit(-1); + } + } else { + state->conn[i].cursor += r; + } + if (r == 0) { + close(state->waiting[i].fd); + state->waiting[i].fd = 0; + state->waiting[i].events = 0; + } + break; + default: + LOG_ERR("invalid connection type"); + exit(-1); + } +} + +// 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); + plasma_request req; + while (1) { + int num_ready = poll(state->waiting, state->num_conn, -1); + 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) + continue; + if (state->waiting[i].fd == sock) { + // Handle new incoming connections. + int new_socket = accept(sock, NULL, NULL); + if (new_socket < 0) { + if (errno != EWOULDBLOCK) { + LOG_ERR("accept failed"); + exit(-1); + } + break; + } + int conn_id = add_conn(state, CONN_CONTROL, new_socket, POLLIN, NULL); + LOG_INFO("new connection with id %d", conn_id); + } else { + read_from_socket(state, i, &req); + } + } + } +} + +// Register this plasma manager with the nameserver. +void register_with_nameserver(const char* nameserver_addr, int nameserver_port, + const char* manager_addr, int manager_port) { + int fd = socket(PF_INET, SOCK_STREAM, 0); + if (fd < 0) { + LOG_ERR("socket for nameserver connection could not be established"); + exit(-1); + } + struct hostent *host = gethostbyname(nameserver_addr); + if (!host) { + LOG_ERR("nameserver %s not found", nameserver_addr); + exit(-1); + } + struct sockaddr_in nameserver; + memset(&nameserver, 0, sizeof(struct sockaddr_in)); + nameserver.sin_family = AF_INET; + bcopy(host->h_addr, &nameserver.sin_addr.s_addr, host->h_length); + nameserver.sin_port = htons(nameserver_port); + if (connect(fd, (struct sockaddr*) &nameserver, sizeof(nameserver)) == -1) { + LOG_ERR("could not connect to nameserver %s:%d", nameserver_addr, nameserver_port); + exit(-1); + } + plasma_request req = { .type = PLASMA_REGISTER, .port = manager_port }; + // TODO(pcm): input validation + sscanf(manager_addr, "%" SCNu8 ".%" SCNu8 ".%" SCNu8 ".%" SCNu8, &req.addr[0], &req.addr[1], &req.addr[2], &req.addr[3]); + plasma_send(fd, &req); + close(fd); +} + +void start_server(const char *store_socket_name, const char* master_addr, + const char* nameserver_addr, int nameserver_port) { + struct sockaddr_in name; + int sock = socket(PF_INET, SOCK_STREAM, 0); + if (sock < 0) { + LOG_ERR("could not create socket"); + exit(-1); + } + name.sin_family = AF_INET; + name.sin_port = 0; + name.sin_addr.s_addr = htonl(INADDR_ANY); + int on = 1; + // TODO(pcm): http://stackoverflow.com/q/1150635 + if (ioctl(sock, FIONBIO, (char*) &on) < 0) { + LOG_ERR("ioctl failed"); + close(sock); + exit(-1); + } + if (bind(sock, (struct sockaddr*) &name, sizeof(name)) < 0) { + LOG_ERR("could not bind socket"); + exit(-1); + } + socklen_t len = sizeof(name); + if (getsockname(sock, (struct sockaddr*) &name, &len) == -1) { + LOG_ERR("getsockname failed"); + } else { + LOG_INFO("listening on port %d", ntohs(name.sin_port)); + } + if (listen(sock, 5) == -1) { + LOG_ERR("could not listen to socket"); + exit(-1); + } + register_with_nameserver(nameserver_addr, nameserver_port, master_addr, ntohs(name.sin_port)); + plasma_manager_state state; + init_manager_state(&state, store_socket_name); + event_loop(sock, &state); +} + + + +int main(int argc, char* argv[]) { + // Socket name of the plasma store this manager is connected to. + char *store_socket_name = NULL; + // IP address and port of the nameserver. + char *nameserver_addr_port = NULL; + // IP address this host can be reached at from the outside. + char *master_addr = NULL; + int c; + while ((c = getopt(argc, argv, "n:s:m:")) != -1) { + switch (c) { + case 's': + store_socket_name = optarg; + break; + case 'n': + nameserver_addr_port = optarg; + break; + case 'm': + master_addr = optarg; + break; + default: + LOG_ERR("unknown option %c", c); + exit(-1); + } + } + if (!store_socket_name) { + LOG_ERR("please specify socket for connecting to the plasma store with -s switch"); + exit(-1); + } + if (!master_addr) { + LOG_ERR("please specify ip address of the current host in the format 123.456.789.10 with -m switch"); + exit(-1); + } + // Parse nameserver address and port. + const char *format = "%15[0-9.]:%5[0-9]"; + char nameserver_addr[16] = { 0 }; + char nameserver_port[6] = { 0 }; + if(!nameserver_addr_port || sscanf(nameserver_addr_port, format, nameserver_addr, nameserver_port) != 2) { + LOG_ERR("need to specify nameserver address in the format 123.456.789.10:12345 with -n switch"); + exit(-1); + } + start_server(store_socket_name, master_addr, nameserver_addr, atoi(nameserver_port)); +} diff --git a/src/plasma_store.c b/src/plasma_store.c index 3260cde9e..d67420c24 100644 --- a/src/plasma_store.c +++ b/src/plasma_store.c @@ -28,9 +28,12 @@ #define MAX_NUM_CLIENTS 2048 typedef struct { - int num_clients; // number of clients connected - int client_id[MAX_NUM_CLIENTS]; // unique identifier for the clients - struct pollfd waiting[MAX_NUM_CLIENTS]; // data structure for polling + // 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]; } plasma_store_state; void init_state(plasma_store_state* s) { @@ -48,8 +51,8 @@ int add_client(plasma_store_state* s, int fd) { 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 +// 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)); @@ -59,27 +62,35 @@ void remove_client(plasma_store_state* s, int i) { } typedef struct { - plasma_id object_id; // object id of this object - plasma_object_info info; // object info like size, creation time and owner - int fd; // memory mapped file containing the object - UT_hash_handle handle; // handle for the uthash table + // Object id of this object. + plasma_id object_id; + // Object info like size, creation time and owner. + plasma_object_info info; + // Memory mapped file containing the object. + int fd; + // Handle for the uthash table. + UT_hash_handle handle; } object_table_entry; // 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 { - plasma_id object_id; // object id of this object - int num_waiting; // number of processes waiting for the object - int conn[MAX_NUM_CLIENTS]; // socket connections to waiting clients - UT_hash_handle handle; // handle for the uthash table + // Object id of this object. + plasma_id object_id; + // Number of processes waiting for the object. + int num_waiting; + // Socket connections to waiting clients. + int conn[MAX_NUM_CLIENTS]; + // 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 @@ -107,7 +118,7 @@ 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 int fd = create_buffer(req->size); @@ -125,7 +136,7 @@ void create_object(int conn, plasma_request* req) { 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); @@ -146,7 +157,7 @@ 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; @@ -158,7 +169,7 @@ void seal_object(int conn, plasma_request* req) { 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) { @@ -205,7 +216,7 @@ void event_loop(int socket) { continue; if (state.waiting[i].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) { diff --git a/test/nameserver.py b/test/nameserver.py new file mode 100644 index 000000000..6fcd965ca --- /dev/null +++ b/test/nameserver.py @@ -0,0 +1,33 @@ +import collections +import socket +import ctypes +import plasma + +DEFAULT_PORT = 16121 + +Connection = collections.namedtuple("Connection", ["address", "port"]) + +# list of IP addresses and ports of managers +object_managers = [] + +def send_addresses(conn, object_managers): + manager = plasma.PlasmaManager(conn.address, conn.port) + for (manager_id, object_manager) in enumerate(object_managers): + manager.register(manager_id, object_manager.address, object_manager.port) + +if __name__ == '__main__': + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(('', DEFAULT_PORT)) + sock.listen(5) + + while True: + (client, address) = sock.accept() + request = plasma.PlasmaRequest() + client.recv_into(request) + address = ".".join(map(str, request.addr[:])) + conn = Connection(address=address, port=request.port) + print "object manager " + str(conn) + " connected" + object_managers.append(conn) + for c in object_managers: + send_addresses(c, object_managers) + From 3d6d1e8fe5b3017dca78048f51c1296d0f241add Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Thu, 18 Aug 2016 09:56:20 -0700 Subject: [PATCH 2/3] Test plasma manager. --- lib/python/plasma.py | 49 +++++++++++++++++++- src/plasma_manager.c | 8 ++-- test/nameserver.py | 35 +++++++++----- test/test.py | 106 ++++++++++++++++++++++++++++++++++++++++--- 4 files changed, 174 insertions(+), 24 deletions(-) diff --git a/lib/python/plasma.py b/lib/python/plasma.py index 83de5f1b8..28b3f086d 100644 --- a/lib/python/plasma.py +++ b/lib/python/plasma.py @@ -16,6 +16,8 @@ PLASMA_SEAL = 2 PLASMA_TRANSFER = 3 PLASMA_DATA = 4 PLASMA_REGISTER = 5 +PLASMA_GET_MANAGER_PORT = 6 +PLASMA_RETURN_MANAGER_PORT = 7 class PlasmaRequest(ctypes.Structure): _fields_ = [("type", ctypes.c_int), @@ -38,24 +40,43 @@ def make_plasma_id(string): return PlasmaID(plasma_id=ID(*object_id)) class PlasmaManager(object): + """The PlasmaManager is used to manage a PlasmaStore. + + There should be one PlasmaManager per PlasmaStore. The PlasmaManager is + responsible for interfacing with other PlasmaManagers in order to transfer + objects between PlasmaStores. This class sends commands to the C + implementation of the PlasmaManager using sockets. + + Attributes: + sock: The socket used to communicate with the C implementation of the + PlasmaManager. + """ def __init__(self, addr, port): + """Initialize the PlasmaManager.""" self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((addr, port)) def register(self, manager_id, addr, port): - "Register another object manager." + """Register another object manager.""" req = PlasmaRequest(type=PLASMA_REGISTER, manager_id=manager_id, addr=Addr(*map(int, addr.split("."))), port=port) self.sock.send(buffer(req)[:]) def transfer(self, manager_id, object_id): - "Transfer local object with id object_id to manager with id manager_id." + """Transfer local object with id object_id to manager with id manager_id.""" req = PlasmaRequest(type=PLASMA_TRANSFER, manager_id=manager_id, object_id=make_plasma_id(object_id)) self.sock.send(buffer(req)[:]) class PlasmaClient(object): + """The PlasmaClient is used to interface with a PlasmaStore. + + The PlasmaClient can ask the PlasmaStore to allocate a new buffer, seal a + buffer, and get a buffer. Buffers are referred to by object IDs, which are + strings. + """ + def __init__(self, socket_name): plasma_client_library = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../../build/plasma_client.so") self.client = ctypes.cdll.LoadLibrary(plasma_client_library) @@ -82,12 +103,36 @@ class PlasmaClient(object): self.sock = self.client.plasma_store_connect(socket_name) def create(self, object_id, size): + """Create a new buffer in the PlasmaStore for a particular object ID. + + The returned buffer is mutable until seal is called. + + Args: + object_id (str): A string used to identify an object. + size (int): The size in bytes of the created buffer. + """ buf = self.client.plasma_create(self.sock, make_plasma_id(object_id), size) return self.buffer_from_read_write_memory(buf.data, buf.size) def get(self, object_id): + """Create a buffer from the PlasmaStore based on object ID. + + This method can only be called after the buffer has been sealed. The + retrieved buffer is immutable. + + Args: + object_id (str): A string used to identify an object. + """ buf = self.client.plasma_get(self.sock, make_plasma_id(object_id)) return self.buffer_from_memory(buf.data, buf.size) def seal(self, object_id): + """Seal the buffer in the PlasmaStore for a particular object ID. + + Once a buffer has been sealed, the buffer is immutable and can only be + accessed through get. + + Args: + object_id (str): A string used to identify an object. + """ self.client.plasma_seal(self.sock, make_plasma_id(object_id)) diff --git a/src/plasma_manager.c b/src/plasma_manager.c index 7ad2585b6..b439fcf35 100644 --- a/src/plasma_manager.c +++ b/src/plasma_manager.c @@ -128,7 +128,7 @@ void initiate_transfer(plasma_manager_state* state, plasma_request* req) { int manager_id = req->manager_id; int c = plasma_store_connect(state->store_socket_name); plasma_buffer buf = plasma_get(c, req->object_id); - + int fd = socket(PF_INET, SOCK_STREAM, 0); if (fd < 0) { LOG_ERR("could not create socket"); @@ -274,7 +274,7 @@ void event_loop(int sock, plasma_manager_state* state) { LOG_INFO("new connection with id %d", conn_id); } else { read_from_socket(state, i, &req); - } + } } } } @@ -346,8 +346,6 @@ void start_server(const char *store_socket_name, const char* master_addr, event_loop(sock, &state); } - - int main(int argc, char* argv[]) { // Socket name of the plasma store this manager is connected to. char *store_socket_name = NULL; @@ -384,7 +382,7 @@ int main(int argc, char* argv[]) { const char *format = "%15[0-9.]:%5[0-9]"; char nameserver_addr[16] = { 0 }; char nameserver_port[6] = { 0 }; - if(!nameserver_addr_port || sscanf(nameserver_addr_port, format, nameserver_addr, nameserver_port) != 2) { + if (!nameserver_addr_port || sscanf(nameserver_addr_port, format, nameserver_addr, nameserver_port) != 2) { LOG_ERR("need to specify nameserver address in the format 123.456.789.10:12345 with -n switch"); exit(-1); } diff --git a/test/nameserver.py b/test/nameserver.py index 6fcd965ca..66fa6cf52 100644 --- a/test/nameserver.py +++ b/test/nameserver.py @@ -1,6 +1,8 @@ import collections import socket import ctypes +import atexit + import plasma DEFAULT_PORT = 16121 @@ -15,19 +17,30 @@ def send_addresses(conn, object_managers): for (manager_id, object_manager) in enumerate(object_managers): manager.register(manager_id, object_manager.address, object_manager.port) -if __name__ == '__main__': +if __name__ == "__main__": sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.bind(('', DEFAULT_PORT)) + sock.bind(("", DEFAULT_PORT)) sock.listen(5) - + + def cleanup(): + sock.shutdown(socket.SHUT_RDWR) + sock.close() + atexit.register(cleanup) + while True: - (client, address) = sock.accept() + client, address = sock.accept() request = plasma.PlasmaRequest() client.recv_into(request) - address = ".".join(map(str, request.addr[:])) - conn = Connection(address=address, port=request.port) - print "object manager " + str(conn) + " connected" - object_managers.append(conn) - for c in object_managers: - send_addresses(c, object_managers) - + if request.type == plasma.PLASMA_REGISTER: + address = ".".join(map(str, request.addr[:])) + conn = Connection(address=address, port=request.port) + print "object manager " + str(conn) + " connected" + object_managers.append(conn) + for c in object_managers: + send_addresses(c, object_managers) + elif request.type == plasma.PLASMA_GET_MANAGER_PORT: + port = object_managers[request.manager_id].port + req = plasma.PlasmaRequest(type=plasma.PLASMA_RETURN_MANAGER_PORT, port=port) + client.send(buffer(req)[:]) + else: + raise Exception("This code should be unreachable.") diff --git a/test/test.py b/test/test.py index 6d4bee003..ba9789e81 100644 --- a/test/test.py +++ b/test/test.py @@ -1,11 +1,17 @@ import os +import socket import subprocess import sys import unittest +import random +import time import plasma -class TestPlasmaAPI(unittest.TestCase): +def random_object_id(): + return "".join([chr(random.randint(0, 256)) for _ in range(20)]) + +class TestPlasmaClient(unittest.TestCase): def setUp(self): # Start Plasma. @@ -15,12 +21,12 @@ class TestPlasmaAPI(unittest.TestCase): self.plasma_client = plasma.PlasmaClient("/tmp/store") def tearDown(self): - # Kill the plasma stoe process. + # Kill the plasma store process. self.p.kill() def test_create(self): - # Create an object string. - object_id = "id" + 18 * "x" + # Create an object id string. + object_id = random_object_id() # Create a new buffer and write to it. length = 1000 memory_buffer = self.plasma_client.create(object_id, length) @@ -34,8 +40,8 @@ class TestPlasmaAPI(unittest.TestCase): self.assertEqual(memory_buffer[i], chr(i % 256)) def test_illegal_functionality(self): - # Create an object string. - object_id = "id" + 18 * "x" + # Create an object id string. + object_id = random_object_id() # Create a new buffer and write to it. length = 1000 memory_buffer = self.plasma_client.create(object_id, length) @@ -55,5 +61,93 @@ class TestPlasmaAPI(unittest.TestCase): memory_buffer[0] = chr(0) self.assertRaises(Exception, illegal_assignment) +class TestPlasmaManager(unittest.TestCase): + + def setUp(self): + # Start the nameserver. + nameserver_path = os.path.join(os.path.abspath(os.path.dirname(__file__)), "nameserver.py") + self.p1 = subprocess.Popen(["python", nameserver_path]) + # 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"]) + # Connect two PlasmaClients. + self.client1 = plasma.PlasmaClient("/tmp/store1") + self.client2 = plasma.PlasmaClient("/tmp/store2") + # Start two PlasmaManagers. + time.sleep(0.1) + plasma_manager_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_manager") + self.p4 = subprocess.Popen([plasma_manager_executable, "-n", "127.0.0.1:16121", "-s", "/tmp/store1", "-m", "127.0.0.1"]) + self.p5 = subprocess.Popen([plasma_manager_executable, "-n", "127.0.0.1:16121", "-s", "/tmp/store2", "-m", "127.0.0.1"]) + time.sleep(0.1) + # Connect to the nameserver. + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect(("127.0.0.1", 16121)) + # Get the port for the first PlasmaManager. + req = plasma.PlasmaRequest(type=plasma.PLASMA_GET_MANAGER_PORT, manager_id=0) + sock.send(buffer(req)[:]) + request = plasma.PlasmaRequest() + sock.recv_into(request) + port1 = request.port + time.sleep(0.1) + # Get the port for the second PlasmaManager. + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect(("127.0.0.1", 16121)) + req = plasma.PlasmaRequest(type=plasma.PLASMA_GET_MANAGER_PORT, manager_id=1) + sock.send(buffer(req)[:]) + request = plasma.PlasmaRequest() + sock.recv_into(request) + port2 = request.port + # Connect two Python PlasmaManagers. + self.manager1 = plasma.PlasmaManager("127.0.0.1", port1) + self.manager2 = plasma.PlasmaManager("127.0.0.1", port2) + + def tearDown(self): + # Kill the nameserver, PlasmaStore and PlasmaManager processes. + self.p1.kill() + self.p2.kill() + self.p3.kill() + self.p4.kill() + self.p5.kill() + + def test_transfer(self): + # Create an object id string. + object_id1 = random_object_id() + # Create a new buffer and write to it. + memory_buffer = self.client1.create(object_id1, 20000) + for i in range(len(memory_buffer)): + memory_buffer[i] = chr(i % 10) + # Seal the buffer. + self.client1.seal(object_id1) + # Transfer the buffer to the the other PlasmaStore. + self.manager1.transfer(1, object_id1) + # Compare the two buffers. + self.assertEqual(self.client1.get(object_id1)[:], self.client2.get(object_id1)[:]) + # Transfer the buffer again. + self.manager1.transfer(1, object_id1) + # Compare the two buffers. + self.assertEqual(self.client1.get(object_id1)[:], self.client2.get(object_id1)[:]) + # Create a new object id string. + object_id2 = random_object_id() + # Create a new buffer and write to it. + memory_buffer = self.client2.create(object_id2, 20000) + for i in range(len(memory_buffer)): + memory_buffer[i] = chr(i % 10) + # Seal the buffer. + self.client2.seal(object_id2) + # Transfer the buffer to the the other PlasmaStore. + self.manager2.transfer(0, object_id2) + # Compare the two buffers. + self.assertEqual(self.client1.get(object_id2)[:], self.client2.get(object_id2)[:]) + + def test_illegal_functionality(self): + # Create an object id string. + object_id = random_object_id() + # Create a new buffer. + memory_buffer = self.client1.create(object_id, 20000) + # This test is commented out because it currently fails. + # # Transferring the buffer before sealing it should fail. + # self.assertRaises(Exception, lambda : self.manager1.transfer(1, object_id)) + if __name__ == "__main__": unittest.main(verbosity=2) From e1627319b3ea03c4811e5c91178edf4efa193a9c Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Mon, 22 Aug 2016 15:30:16 -0700 Subject: [PATCH 3/3] clean up tests --- lib/python/plasma.py | 16 ++---- src/plasma.h | 2 - src/plasma_manager.c | 127 ++++++++++--------------------------------- test/nameserver.py | 46 ---------------- test/test.py | 43 ++++----------- 5 files changed, 45 insertions(+), 189 deletions(-) delete mode 100644 test/nameserver.py diff --git a/lib/python/plasma.py b/lib/python/plasma.py index 28b3f086d..713c3f3ad 100644 --- a/lib/python/plasma.py +++ b/lib/python/plasma.py @@ -16,12 +16,9 @@ PLASMA_SEAL = 2 PLASMA_TRANSFER = 3 PLASMA_DATA = 4 PLASMA_REGISTER = 5 -PLASMA_GET_MANAGER_PORT = 6 -PLASMA_RETURN_MANAGER_PORT = 7 class PlasmaRequest(ctypes.Structure): _fields_ = [("type", ctypes.c_int), - ("manager_id", ctypes.c_int), ("object_id", PlasmaID), ("size", ctypes.c_int64), ("addr", Addr), @@ -57,16 +54,11 @@ class PlasmaManager(object): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((addr, port)) - def register(self, manager_id, addr, port): - """Register another object manager.""" - req = PlasmaRequest(type=PLASMA_REGISTER, manager_id=manager_id, - addr=Addr(*map(int, addr.split("."))), port=port) - self.sock.send(buffer(req)[:]) - - def transfer(self, manager_id, object_id): + def transfer(self, addr, port, object_id): """Transfer local object with id object_id to manager with id manager_id.""" - req = PlasmaRequest(type=PLASMA_TRANSFER, manager_id=manager_id, - object_id=make_plasma_id(object_id)) + req = PlasmaRequest(type=PLASMA_TRANSFER, object_id=make_plasma_id(object_id), + addr=Addr(*map(int, addr.split("."))), port=port) + print "sending port", port self.sock.send(buffer(req)[:]) class PlasmaClient(object): diff --git a/src/plasma.h b/src/plasma.h index 2e4fbfdcd..39d04700f 100644 --- a/src/plasma.h +++ b/src/plasma.h @@ -38,12 +38,10 @@ enum plasma_request_type { PLASMA_SEAL, // seal an object PLASMA_TRANSFER, // request transfer to another store PLASMA_DATA, // header for sending data - PLASMA_REGISTER // register a plasma manager }; typedef struct { int type; - int manager_id; plasma_id object_id; int64_t size; uint8_t addr[4]; diff --git a/src/plasma_manager.c b/src/plasma_manager.c index b439fcf35..34a8e474a 100644 --- a/src/plasma_manager.c +++ b/src/plasma_manager.c @@ -49,13 +49,6 @@ typedef struct { int64_t cursor; } conn_state; -typedef struct { - // Address of the manager. - struct sockaddr_in name; - // Is this manager connected? - int connected; -} manager_state; - typedef struct { // Name of the socket connecting to local plasma store. const char* store_socket_name; @@ -65,36 +58,17 @@ typedef struct { struct pollfd waiting[MAX_CONNECTIONS]; // Status of connections (both control and data). conn_state conn[MAX_CONNECTIONS]; - // Other plasma managers in the cluster. - manager_state managers[MAX_NUM_MANAGERS]; } plasma_manager_state; 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)); - memset(&s->managers, 0, sizeof(s->managers)); s->num_conn = 0; s->store_socket_name = store_socket_name; } #define h_addr h_addr_list[0] -// Add name info for another plasma manager from the cluster. -void add_manager(plasma_manager_state *s, int manager_id, char *ip_address, int port) { - assert(ip_address); - assert(s); - struct hostent *manager = gethostbyname(ip_address); - if (!manager) { - LOG_ERR("plasma manager %s not found", ip_address); - exit(-1); - } - s->managers[manager_id].connected = 1; - struct sockaddr_in *name = &s->managers[manager_id].name; - name->sin_family = AF_INET; - bcopy(manager->h_addr, &name->sin_addr.s_addr, manager->h_length); - name->sin_port = htons(port); -} - // Add connection for sending commands or data to another plasma manager // (returns the connection id). int add_conn(plasma_manager_state* s, int type, int fd, int events, plasma_buffer* buf) { @@ -125,7 +99,6 @@ void remove_conn(plasma_manager_state* s, int i) { // 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 manager_id = req->manager_id; int c = plasma_store_connect(state->store_socket_name); plasma_buffer buf = plasma_get(c, req->object_id); @@ -134,9 +107,24 @@ void initiate_transfer(plasma_manager_state* state, plasma_request* req) { LOG_ERR("could not create socket"); exit(-1); } - int r = connect(fd, (struct sockaddr*) &state->managers[manager_id].name, sizeof(state->managers[manager_id].name)); + + char ip_addr[16]; + snprintf(ip_addr, 32, "%d.%d.%d.%d", + req->addr[0], req->addr[1], + req->addr[2], req->addr[3]); + struct hostent *manager = gethostbyname(ip_addr); // TODO(pcm): cache this + if (!manager) { + LOG_ERR("plasma manager %s not found", ip_addr); + exit(-1); + } + struct sockaddr_in addr; + addr.sin_family = AF_INET; + bcopy(manager->h_addr, &addr.sin_addr.s_addr, manager->h_length); + addr.sin_port = htons(req->port); + + int r = connect(fd, (struct sockaddr*) &addr, sizeof(addr)); if (r < 0) { - LOG_ERR("could not establish connection to manager with id %d", manager_id); + LOG_ERR("could not establish connection to manager with id %s:%d", &ip_addr[0], req->port); exit(-1); } @@ -160,22 +148,9 @@ void setup_data_connection(int conn_idx, plasma_manager_state* state, plasma_req void process_command(int conn_idx, plasma_manager_state* state, plasma_request* req) { switch (req->type) { case PLASMA_TRANSFER: - LOG_INFO("transfering object to manager with id %d", req->manager_id); + LOG_INFO("transfering object to manager with port %d", req->port); initiate_transfer(state, req); break; - case PLASMA_REGISTER: { - char buff[16]; - snprintf(buff, 32, "%d.%d.%d.%d", - req->addr[0], req->addr[1], - req->addr[2], req->addr[3]); - if (req->manager_id >= MAX_NUM_MANAGERS) { - LOG_ERR("manager_id %d out of bounds", req->manager_id); - } else { - add_manager(state, req->manager_id, buff, req->port); - LOG_INFO("registering %s:%d with id %d", buff, req->port, req->manager_id); - } - } - break; case PLASMA_DATA: LOG_INFO("starting to stream data"); setup_data_connection(conn_idx, state, req); @@ -279,37 +254,7 @@ void event_loop(int sock, plasma_manager_state* state) { } } -// Register this plasma manager with the nameserver. -void register_with_nameserver(const char* nameserver_addr, int nameserver_port, - const char* manager_addr, int manager_port) { - int fd = socket(PF_INET, SOCK_STREAM, 0); - if (fd < 0) { - LOG_ERR("socket for nameserver connection could not be established"); - exit(-1); - } - struct hostent *host = gethostbyname(nameserver_addr); - if (!host) { - LOG_ERR("nameserver %s not found", nameserver_addr); - exit(-1); - } - struct sockaddr_in nameserver; - memset(&nameserver, 0, sizeof(struct sockaddr_in)); - nameserver.sin_family = AF_INET; - bcopy(host->h_addr, &nameserver.sin_addr.s_addr, host->h_length); - nameserver.sin_port = htons(nameserver_port); - if (connect(fd, (struct sockaddr*) &nameserver, sizeof(nameserver)) == -1) { - LOG_ERR("could not connect to nameserver %s:%d", nameserver_addr, nameserver_port); - exit(-1); - } - plasma_request req = { .type = PLASMA_REGISTER, .port = manager_port }; - // TODO(pcm): input validation - sscanf(manager_addr, "%" SCNu8 ".%" SCNu8 ".%" SCNu8 ".%" SCNu8, &req.addr[0], &req.addr[1], &req.addr[2], &req.addr[3]); - plasma_send(fd, &req); - close(fd); -} - -void start_server(const char *store_socket_name, const char* master_addr, - const char* nameserver_addr, int nameserver_port) { +void start_server(const char *store_socket_name, const char* master_addr, int port) { struct sockaddr_in name; int sock = socket(PF_INET, SOCK_STREAM, 0); if (sock < 0) { @@ -317,7 +262,7 @@ void start_server(const char *store_socket_name, const char* master_addr, exit(-1); } name.sin_family = AF_INET; - name.sin_port = 0; + name.sin_port = htons(port); name.sin_addr.s_addr = htonl(INADDR_ANY); int on = 1; // TODO(pcm): http://stackoverflow.com/q/1150635 @@ -330,17 +275,11 @@ void start_server(const char *store_socket_name, const char* master_addr, LOG_ERR("could not bind socket"); exit(-1); } - socklen_t len = sizeof(name); - if (getsockname(sock, (struct sockaddr*) &name, &len) == -1) { - LOG_ERR("getsockname failed"); - } else { - LOG_INFO("listening on port %d", ntohs(name.sin_port)); - } + LOG_INFO("listening on port %d", port); if (listen(sock, 5) == -1) { LOG_ERR("could not listen to socket"); exit(-1); } - register_with_nameserver(nameserver_addr, nameserver_port, master_addr, ntohs(name.sin_port)); plasma_manager_state state; init_manager_state(&state, store_socket_name); event_loop(sock, &state); @@ -349,22 +288,22 @@ void start_server(const char *store_socket_name, const char* master_addr, int main(int argc, char* argv[]) { // Socket name of the plasma store this manager is connected to. char *store_socket_name = NULL; - // IP address and port of the nameserver. - char *nameserver_addr_port = NULL; - // IP address this host can be reached at from the outside. + // IP address of this node char *master_addr = NULL; + // Port number the manager should use + int port; int c; - while ((c = getopt(argc, argv, "n:s:m:")) != -1) { + while ((c = getopt(argc, argv, "s:m:p:")) != -1) { switch (c) { case 's': store_socket_name = optarg; break; - case 'n': - nameserver_addr_port = optarg; - break; case 'm': master_addr = optarg; break; + case 'p': + port = atoi(optarg); + break; default: LOG_ERR("unknown option %c", c); exit(-1); @@ -378,13 +317,5 @@ int main(int argc, char* argv[]) { LOG_ERR("please specify ip address of the current host in the format 123.456.789.10 with -m switch"); exit(-1); } - // Parse nameserver address and port. - const char *format = "%15[0-9.]:%5[0-9]"; - char nameserver_addr[16] = { 0 }; - char nameserver_port[6] = { 0 }; - if (!nameserver_addr_port || sscanf(nameserver_addr_port, format, nameserver_addr, nameserver_port) != 2) { - LOG_ERR("need to specify nameserver address in the format 123.456.789.10:12345 with -n switch"); - exit(-1); - } - start_server(store_socket_name, master_addr, nameserver_addr, atoi(nameserver_port)); + start_server(store_socket_name, master_addr, port); } diff --git a/test/nameserver.py b/test/nameserver.py deleted file mode 100644 index 66fa6cf52..000000000 --- a/test/nameserver.py +++ /dev/null @@ -1,46 +0,0 @@ -import collections -import socket -import ctypes -import atexit - -import plasma - -DEFAULT_PORT = 16121 - -Connection = collections.namedtuple("Connection", ["address", "port"]) - -# list of IP addresses and ports of managers -object_managers = [] - -def send_addresses(conn, object_managers): - manager = plasma.PlasmaManager(conn.address, conn.port) - for (manager_id, object_manager) in enumerate(object_managers): - manager.register(manager_id, object_manager.address, object_manager.port) - -if __name__ == "__main__": - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.bind(("", DEFAULT_PORT)) - sock.listen(5) - - def cleanup(): - sock.shutdown(socket.SHUT_RDWR) - sock.close() - atexit.register(cleanup) - - while True: - client, address = sock.accept() - request = plasma.PlasmaRequest() - client.recv_into(request) - if request.type == plasma.PLASMA_REGISTER: - address = ".".join(map(str, request.addr[:])) - conn = Connection(address=address, port=request.port) - print "object manager " + str(conn) + " connected" - object_managers.append(conn) - for c in object_managers: - send_addresses(c, object_managers) - elif request.type == plasma.PLASMA_GET_MANAGER_PORT: - port = object_managers[request.manager_id].port - req = plasma.PlasmaRequest(type=plasma.PLASMA_RETURN_MANAGER_PORT, port=port) - client.send(buffer(req)[:]) - else: - raise Exception("This code should be unreachable.") diff --git a/test/test.py b/test/test.py index ba9789e81..b0284cb54 100644 --- a/test/test.py +++ b/test/test.py @@ -5,11 +5,12 @@ import sys import unittest import random import time +import tempfile import plasma def random_object_id(): - return "".join([chr(random.randint(0, 256)) for _ in range(20)]) + return "".join([chr(random.randint(0, 255)) for _ in range(20)]) class TestPlasmaClient(unittest.TestCase): @@ -64,9 +65,6 @@ class TestPlasmaClient(unittest.TestCase): class TestPlasmaManager(unittest.TestCase): def setUp(self): - # Start the nameserver. - nameserver_path = os.path.join(os.path.abspath(os.path.dirname(__file__)), "nameserver.py") - self.p1 = subprocess.Popen(["python", nameserver_path]) # 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"]) @@ -75,36 +73,19 @@ class TestPlasmaManager(unittest.TestCase): self.client1 = plasma.PlasmaClient("/tmp/store1") self.client2 = plasma.PlasmaClient("/tmp/store2") # Start two PlasmaManagers. - time.sleep(0.1) + 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, "-n", "127.0.0.1:16121", "-s", "/tmp/store1", "-m", "127.0.0.1"]) - self.p5 = subprocess.Popen([plasma_manager_executable, "-n", "127.0.0.1:16121", "-s", "/tmp/store2", "-m", "127.0.0.1"]) + 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)]) time.sleep(0.1) - # Connect to the nameserver. - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.connect(("127.0.0.1", 16121)) - # Get the port for the first PlasmaManager. - req = plasma.PlasmaRequest(type=plasma.PLASMA_GET_MANAGER_PORT, manager_id=0) - sock.send(buffer(req)[:]) - request = plasma.PlasmaRequest() - sock.recv_into(request) - port1 = request.port - time.sleep(0.1) - # Get the port for the second PlasmaManager. - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.connect(("127.0.0.1", 16121)) - req = plasma.PlasmaRequest(type=plasma.PLASMA_GET_MANAGER_PORT, manager_id=1) - sock.send(buffer(req)[:]) - request = plasma.PlasmaRequest() - sock.recv_into(request) - port2 = request.port # Connect two Python PlasmaManagers. - self.manager1 = plasma.PlasmaManager("127.0.0.1", port1) - self.manager2 = plasma.PlasmaManager("127.0.0.1", port2) + self.manager1 = plasma.PlasmaManager("127.0.0.1", self.port1) + self.manager2 = plasma.PlasmaManager("127.0.0.1", self.port2) + time.sleep(0.5) def tearDown(self): # Kill the nameserver, PlasmaStore and PlasmaManager processes. - self.p1.kill() self.p2.kill() self.p3.kill() self.p4.kill() @@ -120,11 +101,11 @@ class TestPlasmaManager(unittest.TestCase): # Seal the buffer. self.client1.seal(object_id1) # Transfer the buffer to the the other PlasmaStore. - self.manager1.transfer(1, object_id1) + self.manager1.transfer("127.0.0.1", self.port2, object_id1) # Compare the two buffers. self.assertEqual(self.client1.get(object_id1)[:], self.client2.get(object_id1)[:]) # Transfer the buffer again. - self.manager1.transfer(1, object_id1) + self.manager1.transfer("127.0.0.1", self.port2, object_id1) # Compare the two buffers. self.assertEqual(self.client1.get(object_id1)[:], self.client2.get(object_id1)[:]) # Create a new object id string. @@ -136,7 +117,7 @@ class TestPlasmaManager(unittest.TestCase): # Seal the buffer. self.client2.seal(object_id2) # Transfer the buffer to the the other PlasmaStore. - self.manager2.transfer(0, object_id2) + self.manager2.transfer("127.0.0.1", self.port1, object_id2) # Compare the two buffers. self.assertEqual(self.client1.get(object_id2)[:], self.client2.get(object_id2)[:])