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

Stephanie's plasma refactor (#31)

Browse Source 
* Add Ray common as a submodule

* Convert to Ray common event loop

* Hide plasma manager state

* Interface changes

* Minor fixes: change LOG_INFO calls to LOG_DEBUG, comments, lint

* Turn off DEBUG by default and make Travis happy

* Allow processes time to clean up during Python tests

* Debugging travis...

* Plasma managers have long-lived connections per manager, not per object

* fix valgrind invalid read and cleanup

* make valgrind happy

* update store API

* put in place manager API

* fixed race condition while sending commands to plasma manager and store -- path sent by Phillip

* clang-format

* Revert "fixed race condition while sending commands to plasma manager and store -- path sent by Phillip"

This reverts commit 79e0f6e6d84f2a309b53155955b65c26c75af071.

* Use reliable socket read/writes from Ray common

* Merge data_connection and plasma_manager_connection structs

* small updates

* restore tests
This commit is contained in:
Philipp Moritz
2016-10-03 18:29:18 -07:00
committed by Robert Nishihara
parent eb71c2e84a
commit eabfa9ab6f
16 changed files with 708 additions and 1975 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitmodules
+3
View File
@@ -0,0 +1,3 @@
[submodule "common"]
path = common
url = https://github.com/ray-project/common.git
Makefile
+20 -9
View File
@@ -1,20 +1,31 @@
CC = gcc
CFLAGS = -g -Wall --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -I.
CFLAGS = -g -Wall --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -I. -Icommon -Icommon/thirdparty
BUILD = build
all: $(BUILD)/plasma_store $(BUILD)/plasma_manager $(BUILD)/plasma_client.so $(BUILD)/example
debug: FORCE
debug: CFLAGS += -DDEBUG=1
debug: all
clean:
cd common; make clean
rm -r $(BUILD)/*
$(BUILD)/plasma_store: src/plasma_store.c src/plasma.h src/event_loop.h src/event_loop.c src/fling.h src/fling.c src/malloc.c src/malloc.h thirdparty/dlmalloc.c
$(CC) $(CFLAGS) src/plasma_store.c src/event_loop.c src/fling.c src/malloc.c -o $(BUILD)/plasma_store
$(BUILD)/plasma_store: src/plasma_store.c src/plasma.h src/fling.h src/fling.c src/malloc.c src/malloc.h thirdparty/dlmalloc.c common
$(CC) $(CFLAGS) src/plasma_store.c src/fling.c src/malloc.c common/build/libcommon.a -o $(BUILD)/plasma_store
$(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_manager: src/plasma_manager.c src/plasma.h src/plasma_client.c src/fling.h src/fling.c common
$(CC) $(CFLAGS) src/plasma_manager.c src/plasma_client.c src/fling.c common/build/libcommon.a -o $(BUILD)/plasma_manager
$(BUILD)/plasma_client.so: src/plasma_client.c src/fling.h src/fling.c
$(CC) $(CFLAGS) src/plasma_client.c src/fling.c -fPIC -shared -o $(BUILD)/plasma_client.so
$(BUILD)/plasma_client.so: src/plasma_client.c src/fling.h src/fling.c common
$(CC) $(CFLAGS) src/plasma_client.c src/fling.c common/build/libcommon.a -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) src/plasma_client.c src/example.c src/fling.c -o $(BUILD)/example
$(BUILD)/example: src/plasma_client.c src/plasma.h src/example.c src/fling.h src/fling.c common
$(CC) $(CFLAGS) src/plasma_client.c src/example.c src/fling.c common/build/libcommon.a -o $(BUILD)/example
common: FORCE
git submodule update --init --recursive
cd common; make
FORCE:
common
Submodule
+1
Submodule common added at f4037ad19f
src/event_loop.c
-97
View File
@@ -1,97 +0,0 @@
#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
-43
View File
@@ -1,43 +0,0 @@
#ifndef EVENT_LOOP_H
#define EVENT_LOOP_H
#include <poll.h>
#include <string.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
+7 -6
View File
@@ -13,13 +13,14 @@
#include <assert.h>
#include "plasma.h"
#include "plasma_client.h"
int main(int argc, char *argv[]) {
int conn = -1;
plasma_store_conn *conn = NULL;
int64_t size;
uint8_t *data;
int c;
plasma_id id = {{255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
object_id id = {{255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
255, 255, 255, 255, 255, 255, 255, 255, 255, 255}};
while ((c = getopt(argc, argv, "s:cfg")) != -1) {
switch (c) {
@@ -27,11 +28,11 @@ int main(int argc, char *argv[]) {
conn = plasma_store_connect(optarg);
break;
case 'c':
assert(conn != -1);
assert(conn != NULL);
plasma_create(conn, id, 100, NULL, 0, &data);
break;
case 'f':
assert(conn != -1);
assert(conn != NULL);
plasma_seal(conn, id);
break;
case 'g':
@@ -41,6 +42,6 @@ int main(int argc, char *argv[]) {
abort();
}
}
assert(conn != -1);
close(conn);
assert(conn != NULL);
plasma_store_disconnect(conn);
}
src/malloc.c
+1
View File
@@ -6,6 +6,7 @@
#include <sys/mman.h>
#include <unistd.h>
#include "common.h"
#include "plasma.h"
#include "uthash.h"
src/plasma.h
+30 -94
View File
@@ -7,34 +7,7 @@
#include <stddef.h>
#include <string.h>
#include "uthash.h"
#ifdef NDEBUG
#define LOG_DEBUG(M, ...)
#else
#define LOG_DEBUG(M, ...) \
fprintf(stderr, "[DEBUG] (%s:%d) " M "\n", __FILE__, __LINE__, ##__VA_ARGS__)
#endif
#ifdef PLASMA_LOGGIN_ON
#define LOG_INFO(M, ...) \
fprintf(stderr, "[INFO] (%s:%d) " M "\n", __FILE__, __LINE__, ##__VA_ARGS__)
#else
#define LOG_INFO(M, ...)
#endif
#define LOG_ERR(M, ...) \
fprintf(stderr, "[ERROR] (%s:%d: errno: %s) " M "\n", __FILE__, __LINE__, \
errno == 0 ? "None" : strerror(errno), ##__VA_ARGS__)
#define PLASMA_CHECK(CONDITION, M, ...) \
do { \
if (!(CONDITION)) { \
fprintf(stderr, "[FATAL] (%s:%d " #CONDITION ") \n" M, __FILE__, \
__LINE__); \
exit(-1); \
} \
} while (0)
#include "common.h"
typedef struct {
int64_t data_size;
@@ -43,12 +16,34 @@ typedef struct {
int64_t construct_duration;
} plasma_object_info;
/** Represents an object ID hash, can hold a full SHA1 hash. */
typedef struct { unsigned char id[20]; } plasma_id;
/* Handle to access memory mapped file and map it into client address space */
typedef struct {
/** The file descriptor of the memory mapped file in the store. It is used
* as a unique identifier of the file in the client to look up the
* corresponding file descriptor on the client's side. */
int store_fd;
/** The size in bytes of the memory mapped file. */
int64_t mmap_size;
} object_handle;
enum plasma_request_type {
typedef struct {
/** Handle for memory mapped file the object is stored in. */
object_handle handle;
/** The offset in bytes in the memory mapped file of the data. */
ptrdiff_t data_offset;
/** The offset in bytes in the memory mapped file of the metadata. */
ptrdiff_t metadata_offset;
/** The size in bytes of the data. */
int64_t data_size;
/** The size in bytes of the metadata. */
int64_t metadata_size;
} plasma_object;
enum object_status { OBJECT_NOT_FOUND = 0, OBJECT_FOUND = 1 };
enum plasma_message_type {
/** Create a new object. */
PLASMA_CREATE,
PLASMA_CREATE = 128,
/** Get an object. */
PLASMA_GET,
/** Check if an object is present. */
@@ -64,10 +59,8 @@ enum plasma_request_type {
};
typedef struct {
/** The type of the request. */
int type;
/** The ID of the object that the request is about. */
plasma_id object_id;
object_id object_id;
/** The size of the object's data. */
int64_t data_size;
/** The size of the object's metadata. */
@@ -81,68 +74,11 @@ typedef struct {
} plasma_request;
typedef struct {
/** The offset in bytes in the memory mapped file of the data. */
ptrdiff_t data_offset;
/** The offset in bytes in the memory mapped file of the metadata. */
ptrdiff_t metadata_offset;
/** The size in bytes of the memory mapped file. */
int64_t map_size;
/** The size in bytesof the data. */
int64_t data_size;
/** The size in bytes of the metadata. */
int64_t metadata_size;
/** The object that is returned with this reply. */
plasma_object object;
/** This is used only to respond to requests of type PLASMA_CONTAINS. It is 1
* if the object is present and 0 otherwise. Used for plasma_contains. */
int has_object;
/** The file descriptor of the memory mapped file in the store. */
int store_fd_val;
} plasma_reply;
typedef struct {
plasma_id object_id;
uint8_t *data;
int64_t data_size;
uint8_t *metadata;
int64_t metadata_size;
int writable;
} plasma_buffer;
typedef struct {
/** Key that uniquely identifies the memory mapped file. In practice, we
* take the numerical value of the file descriptor in the object store. */
int key;
/** The result of mmap for this file descriptor. */
uint8_t *pointer;
/** Handle for the uthash table. */
UT_hash_handle hh;
} client_mmap_table_entry;
/** Information about a connection between a Plasma Client and Plasma Store.
* This is used to avoid mapping the same files into memory multiple times. */
typedef struct {
/** File descriptor of the Unix domain socket that connects to the store. */
int conn;
/** Table of dlmalloc buffer files that have been memory mapped so far. */
client_mmap_table_entry *mmap_table;
} plasma_store_conn;
/**
* This is used by the Plasma Client to send a request to the Plasma Store or
* the Plasma Manager.
*
* @param conn The file descriptor to use to send the request.
* @param req The address of the request to send.
* @return Void.
*/
void plasma_send_request(int conn, plasma_request *req);
/**
* This is used by the Plasma Store to send a reply to the Plasma Client.
*
* @param conn The file descriptor to use to send the reply.
* @param req The address of the reply to send.
* @return Void.
*/
void plasma_send_reply(int conn, plasma_reply *req);
#endif
src/plasma_client.c
+71 -52
View File
@@ -12,16 +12,35 @@
#include <netinet/in.h>
#include <netdb.h>
#include "common.h"
#include "io.h"
#include "plasma.h"
#include "plasma_client.h"
#include "fling.h"
#include "uthash.h"
void plasma_send_request(int fd, plasma_request *req) {
typedef struct {
/** Key that uniquely identifies the memory mapped file. In practice, we
* take the numerical value of the file descriptor in the object store. */
int key;
/** The result of mmap for this file descriptor. */
uint8_t *pointer;
/** Handle for the uthash table. */
UT_hash_handle hh;
} client_mmap_table_entry;
/** Information about a connection between a Plasma Client and Plasma Store.
* This is used to avoid mapping the same files into memory multiple times. */
struct plasma_store_conn {
/** File descriptor of the Unix domain socket that connects to the store. */
int conn;
/** Table of dlmalloc buffer files that have been memory mapped so far. */
client_mmap_table_entry *mmap_table;
};
void plasma_send_request(int fd, int type, plasma_request *req) {
int req_count = sizeof(plasma_request);
if (write(fd, req, req_count) != req_count) {
LOG_ERR("write error, fd = %d", fd);
exit(-1);
}
write_message(fd, type, req_count, (uint8_t *) req);
}
/* If the file descriptor fd has been mmapped in this client process before,
@@ -53,97 +72,93 @@ uint8_t *lookup_or_mmap(plasma_store_conn *conn,
}
void plasma_create(plasma_store_conn *conn,
plasma_id object_id,
object_id object_id,
int64_t data_size,
uint8_t *metadata,
int64_t metadata_size,
uint8_t **data) {
LOG_INFO(
"called plasma_create on conn %d with size %d and metadata size "
"%d" PRId64,
conn, size, metadata_size);
plasma_request req = {.type = PLASMA_CREATE,
.object_id = object_id,
LOG_DEBUG("called plasma_create on conn %d with size %" PRId64
" and metadata size "
"%" PRId64,
conn->conn, data_size, metadata_size);
plasma_request req = {.object_id = object_id,
.data_size = data_size,
.metadata_size = metadata_size};
plasma_send_request(conn->conn, &req);
plasma_send_request(conn->conn, PLASMA_CREATE, &req);
plasma_reply reply;
int fd = recv_fd(conn->conn, (char *) &reply, sizeof(plasma_reply));
assert(reply.data_size == data_size);
assert(reply.metadata_size == metadata_size);
plasma_object *object = &reply.object;
CHECK(object->data_size == data_size);
CHECK(object->metadata_size == metadata_size);
/* The metadata should come right after the data. */
assert(reply.metadata_offset == reply.data_offset + data_size);
*data = lookup_or_mmap(conn, fd, reply.store_fd_val, reply.map_size) +
reply.data_offset;
CHECK(object->metadata_offset == object->data_offset + data_size);
*data = lookup_or_mmap(conn, fd, object->handle.store_fd,
object->handle.mmap_size) +
object->data_offset;
/* If plasma_create is being called from a transfer, then we will not copy the
* metadata here. The metadata will be written along with the data streamed
* from the transfer. */
if (metadata != NULL) {
/* Copy the metadata to the buffer. */
memcpy(*data + reply.data_size, metadata, metadata_size);
memcpy(*data + object->data_size, metadata, metadata_size);
}
}
/* This method is used to get both the data and the metadata. */
void plasma_get(plasma_store_conn *conn,
plasma_id object_id,
object_id object_id,
int64_t *size,
uint8_t **data,
int64_t *metadata_size,
uint8_t **metadata) {
plasma_request req = {.type = PLASMA_GET, .object_id = object_id};
plasma_send_request(conn->conn, &req);
plasma_request req = {.object_id = object_id};
plasma_send_request(conn->conn, PLASMA_GET, &req);
plasma_reply reply;
int fd = recv_fd(conn->conn, (char *) &reply, sizeof(plasma_reply));
*data = lookup_or_mmap(conn, fd, reply.store_fd_val, reply.map_size) +
reply.data_offset;
*size = reply.data_size;
plasma_object *object = &reply.object;
*data = lookup_or_mmap(conn, fd, object->handle.store_fd,
object->handle.mmap_size) +
object->data_offset;
*size = object->data_size;
/* If requested, return the metadata as well. */
if (metadata != NULL) {
*metadata = *data + reply.data_size;
*metadata_size = reply.metadata_size;
*metadata = *data + object->data_size;
*metadata_size = object->metadata_size;
}
}
/* This method is used to query whether the plasma store contains an object. */
void plasma_contains(plasma_store_conn *conn,
plasma_id object_id,
object_id object_id,
int *has_object) {
plasma_request req = {.type = PLASMA_CONTAINS, .object_id = object_id};
plasma_send_request(conn->conn, &req);
plasma_request req = {.object_id = object_id};
plasma_send_request(conn->conn, PLASMA_CONTAINS, &req);
plasma_reply reply;
int r = read(conn->conn, &reply, sizeof(plasma_reply));
PLASMA_CHECK(r != -1, "read error");
PLASMA_CHECK(r != 0, "connection disconnected");
CHECKM(r != -1, "read error");
CHECKM(r != 0, "connection disconnected");
*has_object = reply.has_object;
}
void plasma_seal(plasma_store_conn *conn, plasma_id object_id) {
plasma_request req = {.type = PLASMA_SEAL, .object_id = object_id};
plasma_send_request(conn->conn, &req);
void plasma_seal(plasma_store_conn *conn, object_id object_id) {
plasma_request req = {.object_id = object_id};
plasma_send_request(conn->conn, PLASMA_SEAL, &req);
}
void plasma_delete(plasma_store_conn *conn, plasma_id object_id) {
plasma_request req = {.type = PLASMA_DELETE, .object_id = object_id};
plasma_send_request(conn->conn, &req);
void plasma_delete(plasma_store_conn *conn, object_id object_id) {
plasma_request req = {.object_id = object_id};
plasma_send_request(conn->conn, PLASMA_DELETE, &req);
}
plasma_store_conn *plasma_store_connect(const char *socket_name) {
assert(socket_name);
struct sockaddr_un addr;
int fd;
if ((fd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
LOG_ERR("socket error");
exit(-1);
}
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.
*/
int fd = -1;
int connected_successfully = 0;
for (int num_attempts = 0; num_attempts < 50; ++num_attempts) {
if (connect(fd, (struct sockaddr *) &addr, sizeof(addr)) == 0) {
fd = connect_ipc_sock(socket_name);
if (fd >= 0) {
connected_successfully = 1;
break;
}
@@ -162,6 +177,11 @@ plasma_store_conn *plasma_store_connect(const char *socket_name) {
return result;
}
void plasma_store_disconnect(plasma_store_conn *conn) {
close(conn->conn);
free(conn);
}
#define h_addr h_addr_list[0]
int plasma_manager_connect(const char *ip_addr, int port) {
@@ -196,14 +216,13 @@ int plasma_manager_connect(const char *ip_addr, int port) {
void plasma_transfer(int manager,
const char *addr,
int port,
plasma_id object_id) {
plasma_request req = {
.type = PLASMA_TRANSFER, .object_id = object_id, .port = port};
object_id object_id) {
plasma_request req = {.object_id = object_id, .port = port};
char *end = NULL;
for (int i = 0; i < 4; ++i) {
req.addr[i] = strtol(end ? end : addr, &end, 10);
/* skip the '.' */
end += 1;
}
plasma_send_request(manager, &req);
plasma_send_request(manager, PLASMA_TRANSFER, &req);
}
src/plasma_client.h
+26 -5
View File
@@ -1,6 +1,19 @@
#ifndef PLASMA_CLIENT_H
#define PLASMA_CLIENT_H
typedef struct plasma_store_conn plasma_store_conn;
/**
* This is used by the Plasma Client to send a request to the Plasma Store or
* the Plasma Manager.
*
* @param conn The file descriptor to use to send the request.
* @param type The type of request.
* @param req The address of the request to send.
* @return Void.
*/
void plasma_send_request(int conn, int type, plasma_request *req);
/**
* Connect to the local plasma store UNIX domain socket with path socket_name
* and return the resulting connection.
@@ -11,6 +24,14 @@
*/
plasma_store_conn *plasma_store_connect(const char *socket_name);
/**
* Disconnect from the local plasma store.
*
* @param conn The connection to the local plasma store.
* @return Void.
*/
void plasma_store_disconnect(plasma_store_conn *conn);
/**
* Connect to a possibly remote Plasma Manager.
*
@@ -36,7 +57,7 @@ int plasma_manager_connect(const char *addr, int port);
* @return Void.
*/
void plasma_create(plasma_store_conn *conn,
plasma_id object_id,
object_id object_id,
int64_t size,
uint8_t *metadata,
int64_t metadata_size,
@@ -58,7 +79,7 @@ void plasma_create(plasma_store_conn *conn,
* @return Void.
*/
void plasma_get(plasma_store_conn *conn,
plasma_id object_id,
object_id object_id,
int64_t *size,
uint8_t **data,
int64_t *metadata_size,
@@ -77,7 +98,7 @@ void plasma_get(plasma_store_conn *conn,
* @return Void.
*/
void plasma_contains(plasma_store_conn *conn,
plasma_id object_id,
object_id object_id,
int *has_object);
/**
@@ -88,7 +109,7 @@ void plasma_contains(plasma_store_conn *conn,
* @param object_id The ID of the object to seal.
* @return Void.
*/
void plasma_seal(plasma_store_conn *conn, plasma_id object_id);
void plasma_seal(plasma_store_conn *conn, object_id object_id);
/**
* Delete an object from the object store. This currently assumes that the
@@ -101,6 +122,6 @@ void plasma_seal(plasma_store_conn *conn, plasma_id object_id);
* @param object_id The ID of the object to delete.
* @return Void.
*/
void plasma_delete(plasma_store_conn *conn, plasma_id object_id);
void plasma_delete(plasma_store_conn *conn, object_id object_id);
#endif
src/plasma_manager.c
+267 -171
View File
@@ -20,197 +20,290 @@
#include <netinet/in.h>
#include <netdb.h>
#include "uthash.h"
#include "utlist.h"
#include "utstring.h"
#include "common.h"
#include "io.h"
#include "event_loop.h"
#include "plasma.h"
#include "plasma_client.h"
#include "plasma_manager.h"
typedef struct {
/* Connection to local plasma store. */
plasma_store_conn *conn;
/* Event loop. */
event_loop *loop;
/** Connection to the local plasma store for reading or writing data. */
plasma_store_conn *store_conn;
/** Hash table of all contexts for active connections to other plasma
* managers. These are used for writing data to other plasma stores. */
client_connection *manager_connections;
} 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->conn = plasma_store_connect(store_socket_name);
LOG_INFO("Connected to object store %s", store_socket_name);
}
typedef struct plasma_buffer plasma_buffer;
/* Start transfering data to another object store manager. This establishes
* a connection to the remote manager and sends the data header to the other
* object manager. */
void initiate_transfer(plasma_manager_state *s, plasma_request *req) {
/* Buffer for reading and writing data between plasma managers. */
struct plasma_buffer {
object_id object_id;
uint8_t *data;
int64_t data_size;
uint8_t *metadata;
int64_t metadata_size;
plasma_get(s->conn, req->object_id, &data_size, &data, &metadata_size,
&metadata);
assert(metadata == data + data_size);
plasma_buffer buf = {.object_id = req->object_id,
.data = data, /* We treat this as a pointer to the
concatenated data and metadata. */
.data_size = data_size,
.metadata_size = metadata_size,
.writable = 0};
char ip_addr[32];
snprintf(ip_addr, 32, "%d.%d.%d.%d", req->addr[0], req->addr[1], req->addr[2],
req->addr[3]);
int writable;
/* Pointer to the next buffer that we will write to this plasma manager. This
* field is only used if we're transferring data to another plasma manager,
* not if we are receiving data. */
plasma_buffer *next;
};
int fd = plasma_manager_connect(&ip_addr[0], req->port);
data_connection conn = {.type = DATA_CONNECTION_WRITE,
.store_conn = s->conn->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,
.data_size = buf.data_size,
.metadata_size = buf.metadata_size};
plasma_send_request(fd, &manager_req);
}
/* Context for a client connection to another plasma manager. */
struct client_connection {
/* Current state for this plasma manager. This is shared between all client
* connections to the plasma manager. */
plasma_manager_state *manager_state;
/* Current position in the buffer. */
int64_t cursor;
/* Buffer that this connection is reading from. If this is a connection to
* write data to another plasma store, then it is a linked list of buffers to
* write. */
plasma_buffer *transfer_queue;
/* File descriptor for the socket connected to the other plasma manager. */
int fd;
/* Following fields are used only for connections to plasma managers. */
/* Key that uniquely identifies the plasma manager that we're connected to.
* We will use the string <address>:<port> as an identifier. */
char *ip_addr_port;
/** Handle for the uthash table. */
UT_hash_handle hh;
};
/* 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) {
plasma_buffer buf = {.object_id = req->object_id,
.data_size = req->data_size,
.metadata_size = req->metadata_size,
.writable = 1};
plasma_create(s->conn, req->object_id, req->data_size, NULL,
req->metadata_size, &buf.data);
data_connection conn = {.type = DATA_CONNECTION_READ,
.store_conn = s->conn->conn,
.buf = buf,
.cursor = 0};
event_loop_set_connection(s->loop, index, &conn);
plasma_manager_state *init_plasma_manager_state(const char *store_socket_name) {
plasma_manager_state *state = malloc(sizeof(plasma_manager_state));
state->store_conn = plasma_store_connect(store_socket_name);
state->manager_connections = NULL;
return state;
}
/* 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);
initiate_transfer(state, req);
break;
case PLASMA_DATA:
LOG_INFO("starting to stream data");
start_reading_data(id, state, req);
break;
default:
LOG_ERR("invalid request %d", req->type);
exit(-1);
}
}
void process_message(event_loop *loop,
int client_sock,
void *context,
int events);
/* 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) {
void write_object_chunk(event_loop *loop,
int data_sock,
void *context,
int events) {
client_connection *conn = (client_connection *) context;
if (conn->transfer_queue == NULL) {
/* If there are no objects to transfer, temporarily remove this connection
* from the event loop. It will be reawoken when we receive another
* PLASMA_TRANSFER request. */
event_loop_remove_file(loop, conn->fd);
return;
}
LOG_DEBUG("Writing data");
ssize_t r, s;
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 id %" PRId64 " disconnected", index);
event_loop_detach(state->loop, index, 1);
} else {
process_command(index, state, req);
}
break;
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 {
conn->cursor += r;
}
if (r == 0) {
LOG_DEBUG("reading on channel %" PRId64 " finished", index);
plasma_seal(state->conn, conn->buf.object_id);
event_loop_detach(state->loop, index, 1);
}
break;
case DATA_CONNECTION_WRITE:
LOG_DEBUG("polled DATA_CONNECTION_WRITE");
s = conn->buf.data_size + conn->buf.metadata_size - conn->cursor;
if (s > BUFSIZE)
s = BUFSIZE;
r = write(waiting->fd, conn->buf.data + conn->cursor, s);
if (r != s) {
if (r > 0) {
LOG_ERR("partial write on fd %d", waiting->fd);
} else {
LOG_ERR("write error");
exit(-1);
}
} else {
conn->cursor += r;
}
if (r == 0) {
LOG_DEBUG("writing on channel %" PRId64 " finished", index);
event_loop_detach(state->loop, index, 1);
}
break;
default:
LOG_ERR("invalid connection type");
exit(-1);
plasma_buffer *buf = conn->transfer_queue;
if (conn->cursor == 0) {
/* If the cursor is zero, we haven't sent any requests for this object yet,
* so send the initial PLASMA_DATA request. */
plasma_request manager_req = {.object_id = buf->object_id,
.data_size = buf->data_size,
.metadata_size = buf->metadata_size};
plasma_send_request(conn->fd, PLASMA_DATA, &manager_req);
}
}
/* 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 = event_loop_poll(s->loop);
if (num_ready < 0) {
LOG_ERR("poll failed");
/* Try to write one BUFSIZE at a time. */
s = buf->data_size + buf->metadata_size - conn->cursor;
if (s > BUFSIZE)
s = BUFSIZE;
r = write(conn->fd, buf->data + conn->cursor, s);
if (r != s) {
if (r > 0) {
LOG_ERR("partial write on fd %d", conn->fd);
} else {
LOG_ERR("write error");
exit(-1);
}
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 (waiting->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;
}
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(s, waiting, i, &req);
}
}
} else {
conn->cursor += r;
}
if (r == 0) {
/* If we've finished writing this buffer, move on to the next transfer
* request and reset the cursor to zero. */
LOG_DEBUG("writing on channel %d finished", data_sock);
conn->cursor = 0;
LL_DELETE(conn->transfer_queue, buf);
free(buf);
}
}
void read_object_chunk(event_loop *loop,
int data_sock,
void *context,
int events) {
LOG_DEBUG("Reading data");
ssize_t r, s;
client_connection *conn = (client_connection *) context;
plasma_buffer *buf = conn->transfer_queue;
CHECK(buf != NULL);
/* Try to read one BUFSIZE at a time. */
s = buf->data_size + buf->metadata_size - conn->cursor;
if (s > BUFSIZE) {
s = BUFSIZE;
}
r = read(data_sock, buf->data + conn->cursor, s);
if (r == -1) {
LOG_ERR("read error");
} else if (r == 0) {
LOG_DEBUG("end of file");
} else {
conn->cursor += r;
}
if (conn->cursor == buf->data_size + buf->metadata_size) {
LOG_DEBUG("reading on channel %d finished", data_sock);
plasma_seal(conn->manager_state->store_conn, buf->object_id);
LL_DELETE(conn->transfer_queue, buf);
free(buf);
/* Switch to listening for requests from this socket, instead of reading
* data. */
event_loop_remove_file(loop, data_sock);
event_loop_add_file(loop, data_sock, EVENT_LOOP_READ, process_message,
conn);
}
return;
}
void start_writing_data(event_loop *loop,
object_id object_id,
uint8_t addr[4],
int port,
client_connection *conn) {
uint8_t *data;
int64_t data_size;
uint8_t *metadata;
int64_t metadata_size;
plasma_get(conn->manager_state->store_conn, object_id, &data_size, &data,
&metadata_size, &metadata);
assert(metadata == data + data_size);
plasma_buffer *buf = malloc(sizeof(plasma_buffer));
buf->object_id = object_id;
buf->data = data; /* We treat this as a pointer to the
concatenated data and metadata. */
buf->data_size = data_size;
buf->metadata_size = metadata_size;
buf->writable = 0;
/* Look to see if we already have a connection to this plasma manager. */
UT_string *ip_addr;
UT_string *ip_addr_port;
utstring_new(ip_addr);
utstring_new(ip_addr_port);
utstring_printf(ip_addr, "%d.%d.%d.%d", addr[0], addr[1], addr[2], addr[3]);
utstring_printf(ip_addr_port, "%s:%d", utstring_body(ip_addr), port);
client_connection *manager_conn;
HASH_FIND_STR(conn->manager_state->manager_connections,
utstring_body(ip_addr_port), manager_conn);
if (!manager_conn) {
/* If we don't already have a connection to this manager, start one. */
manager_conn = malloc(sizeof(client_connection));
manager_conn->fd = plasma_manager_connect(utstring_body(ip_addr), port);
manager_conn->manager_state = conn->manager_state;
manager_conn->transfer_queue = NULL;
manager_conn->cursor = 0;
manager_conn->ip_addr_port = strdup(utstring_body(ip_addr_port));
HASH_ADD_KEYPTR(hh, manager_conn->manager_state->manager_connections,
manager_conn->ip_addr_port,
strlen(manager_conn->ip_addr_port), manager_conn);
}
utstring_free(ip_addr_port);
utstring_free(ip_addr);
if (manager_conn->transfer_queue == NULL) {
/* If we already have a connection to this manager and its inactive,
* (re)register it with the event loop again. */
event_loop_add_file(loop, manager_conn->fd, EVENT_LOOP_WRITE,
write_object_chunk, manager_conn);
}
/* Add this transfer request to this connection's transfer queue. */
LL_APPEND(manager_conn->transfer_queue, buf);
}
void start_reading_data(event_loop *loop,
int client_sock,
object_id object_id,
int64_t data_size,
int64_t metadata_size,
client_connection *conn) {
plasma_buffer *buf = malloc(sizeof(plasma_buffer));
buf->object_id = object_id;
buf->data_size = data_size;
buf->metadata_size = metadata_size;
buf->writable = 1;
plasma_create(conn->manager_state->store_conn, object_id, data_size, NULL,
metadata_size, &(buf->data));
LL_APPEND(conn->transfer_queue, buf);
conn->cursor = 0;
/* Switch to reading the data from this socket, instead of listening for
* other requests. */
event_loop_remove_file(loop, client_sock);
event_loop_add_file(loop, client_sock, EVENT_LOOP_READ, read_object_chunk,
conn);
}
void process_message(event_loop *loop,
int client_sock,
void *context,
int events) {
client_connection *conn = (client_connection *) context;
int64_t type;
int64_t length;
plasma_request *req;
read_message(client_sock, &type, &length, (uint8_t **) &req);
switch (type) {
case PLASMA_TRANSFER:
LOG_DEBUG("transfering object to manager with port %d", req->port);
start_writing_data(loop, req->object_id, req->addr, req->port, conn);
break;
case PLASMA_DATA:
LOG_DEBUG("starting to stream data");
start_reading_data(loop, client_sock, req->object_id, req->data_size,
req->metadata_size, conn);
break;
case DISCONNECT_CLIENT: {
LOG_INFO("Disconnecting client on fd %d", client_sock);
event_loop_remove_file(loop, client_sock);
close(client_sock);
free(conn);
} break;
default:
LOG_ERR("invalid request %" PRId64, type);
exit(-1);
}
free(req);
}
void new_client_connection(event_loop *loop,
int listener_sock,
void *context,
int events) {
int new_socket = accept_client(listener_sock);
/* Create a new data connection context per client. */
client_connection *conn = malloc(sizeof(client_connection));
conn->manager_state = (plasma_manager_state *) context;
conn->transfer_queue = NULL;
event_loop_add_file(loop, new_socket, EVENT_LOOP_READ, process_message, conn);
LOG_DEBUG("new connection with fd %d", new_socket);
}
void start_server(const char *store_socket_name,
@@ -227,24 +320,27 @@ void start_server(const char *store_socket_name,
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) {
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) {
if (bind(sock, (struct sockaddr *) &name, sizeof(name)) < 0) {
LOG_ERR("could not bind socket");
exit(-1);
}
LOG_INFO("listening on port %d", port);
LOG_DEBUG("listening on port %d", port);
if (listen(sock, 5) == -1) {
LOG_ERR("could not listen to socket");
exit(-1);
}
plasma_manager_state state;
init_plasma_manager(&state, store_socket_name);
run_event_loop(sock, &state);
event_loop *loop = event_loop_create();
plasma_manager_state *state = init_plasma_manager_state(store_socket_name);
event_loop_add_file(loop, sock, EVENT_LOOP_READ, new_client_connection,
state);
event_loop_run(loop);
}
int main(int argc, char *argv[]) {
src/plasma_manager.h
+86 -24
View File
@@ -4,30 +4,92 @@
#include <poll.h>
#include "utarray.h"
typedef struct client_connection client_connection;
/**
* Start transfering data to another object store manager.
*
* @param loop This is the event loop of the plasma manager.
* @param object_id The object_id of the object we will be sending.
* @param addr The IP address of the plasma manager we are sending the object
* to.
* @param port The port of the plasma manager we are sending the object to.
* @param conn The client_connection to the other plasma manager.
*
* This establishes a connection to the remote manager and sends the data
* header to the other object manager.
*/
void start_writing_data(event_loop *loop,
object_id object_id,
uint8_t addr[4],
int port,
client_connection *conn);
/**
* Start reading data from another object manager.
*
* @param loop This is the event loop of the plasma manager.
* @param client_sock The connection to the other plasma manager.
* @param object_id The object_id of the object we will be reading.
* @param data_size Size of the object.
* @param metadata_size Size of the metadata.
* @param conn The client_connection to the other plasma 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(event_loop *loop,
int client_sock,
object_id object_id,
int64_t data_size,
int64_t metadata_size,
client_connection *conn);
/**
* Read the next chunk of the object in transit from the plasma manager
* that is connected to the connection with index "conn_index". Once all data
* has been read, the socket switches to listening for the next request.
*
* @param loop This is the event loop of the plasma manager.
* @param data_sock The connection to the other plasma manager.
* @param context The client_connection to the other plasma manager.
*
*/
void read_object_chunk(event_loop *loop,
int data_sock,
void *context,
int events);
/**
* Write the next chunk of the object currently transfered to the plasma manager
* that is connected to the socket "data_sock". If no data has been sent yet,
* the initial handshake to transfer the object size is performed.
*
* @param loop This is the event loop of the plasma manager.
* @param data_sock This is the socket the other plasma manager is listening on.
* @param context The client_connection to the other plasma manager, contains a
* queue of objects that will be sent.
*/
void write_object_chunk(event_loop *loop,
int data_sock,
void *context,
int events);
/**
* Register a new client connection with the plasma manager. A client can
* either be a worker or another plasma manager.
*
* @param loop This is the event loop of the plasma manager.
* @param listener_socket The socket the plasma manager is listening on.
* @param context The plasma manager state.
*
*/
void new_client_connection(event_loop *loop,
int listener_sock,
void *context,
int events);
/* The buffer size in bytes. Data will get transfered in multiples of this */
#define BUFSIZE 4096
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 data_connection_type. */
int type;
/* 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. */
plasma_buffer buf;
/* Current position in the buffer. */
int64_t cursor;
} data_connection;
#endif
#endif /* PLASMA_MANAGER_H */
src/plasma_store.c
+147 -162
View File
@@ -21,23 +21,22 @@
#include <limits.h>
#include <poll.h>
#include "common.h"
#include "event_loop.h"
#include "io.h"
#include "uthash.h"
#include "utarray.h"
#include "fling.h"
#include "malloc.h"
#include "plasma.h"
#include "event_loop.h"
#define MAX_NUM_CLIENTS 100
void *dlmalloc(size_t);
void dlfree(void *);
void* dlmalloc(size_t);
void dlfree(void*);
typedef struct {
/* Event loop for the plasma store. */
event_loop* loop;
} plasma_store_state;
void plasma_send_reply(int fd, plasma_reply* reply) {
/**
* This is used by the Plasma Store to send a reply to the Plasma Client.
*/
void plasma_send_reply(int fd, plasma_reply *reply) {
int reply_count = sizeof(plasma_reply);
if (write(fd, reply, reply_count) != reply_count) {
LOG_ERR("write error, fd = %d", fd);
@@ -45,14 +44,9 @@ void plasma_send_reply(int fd, plasma_reply* reply) {
}
}
void init_state(plasma_store_state* s) {
s->loop = malloc(sizeof(event_loop));
event_loop_init(s->loop);
}
typedef struct {
/* Object id of this object. */
plasma_id object_id;
object_id object_id;
/* Object info like size, creation time and owner. */
plasma_object_info info;
/* Memory mapped file containing the object. */
@@ -64,39 +58,41 @@ typedef struct {
/* Handle for the uthash table. */
UT_hash_handle handle;
/* Pointer to the object data. Needed to free the object. */
uint8_t* pointer;
uint8_t *pointer;
} object_table_entry;
/* Objects that are still being written by their owner process. */
object_table_entry* open_objects = NULL;
object_table_entry *open_objects = NULL;
/* Objects that have already been sealed by their owner process and
* can now be shared with other processes. */
object_table_entry* sealed_objects = NULL;
object_table_entry *sealed_objects = NULL;
typedef struct {
/* 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];
object_id object_id;
/* Socket connections of waiting clients. */
UT_array *conns;
/* Handle for the uthash table. */
UT_hash_handle handle;
} object_notify_entry;
/* Objects that processes are waiting for. */
object_notify_entry* objects_notify = NULL;
object_notify_entry *objects_notify = NULL;
/* 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 */
plasma_object create_object(int conn,
object_id object_id,
int64_t data_size,
int64_t metadata_size,
plasma_object *result) {
LOG_DEBUG("creating object"); /* TODO(pcm): add object_id here */
object_table_entry* entry;
HASH_FIND(handle, open_objects, &req->object_id, sizeof(plasma_id), entry);
PLASMA_CHECK(entry == NULL, "Cannot create object twice.");
object_table_entry *entry;
HASH_FIND(handle, open_objects, &object_id, sizeof(object_id), entry);
CHECKM(entry == NULL, "Cannot create object twice.");
uint8_t* pointer = dlmalloc(req->data_size + req->metadata_size);
uint8_t *pointer = dlmalloc(data_size + metadata_size);
int fd;
int64_t map_size;
ptrdiff_t offset;
@@ -104,197 +100,186 @@ void create_object(int conn, plasma_request* req) {
assert(fd != -1);
entry = malloc(sizeof(object_table_entry));
memcpy(&entry->object_id, &req->object_id, 20);
entry->info.data_size = req->data_size;
entry->info.metadata_size = req->metadata_size;
memcpy(&entry->object_id, &object_id, 20);
entry->info.data_size = data_size;
entry->info.metadata_size = metadata_size;
entry->pointer = pointer;
/* TODO(pcm): set the other fields */
entry->fd = fd;
entry->map_size = map_size;
entry->offset = offset;
HASH_ADD(handle, open_objects, object_id, sizeof(plasma_id), entry);
plasma_reply reply;
memset(&reply, 0, sizeof(reply));
reply.data_offset = offset;
reply.metadata_offset = offset + req->data_size;
reply.map_size = map_size;
reply.data_size = req->data_size;
reply.metadata_size = req->metadata_size;
reply.store_fd_val = fd;
send_fd(conn, fd, (char*) &reply, sizeof(reply));
HASH_ADD(handle, open_objects, object_id, sizeof(object_id), entry);
object_handle handle = {.store_fd = fd, .mmap_size = map_size};
result->handle = handle;
result->data_offset = offset;
result->metadata_offset = offset + data_size;
result->data_size = data_size;
result->metadata_size = metadata_size;
}
/* 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);
int get_object(int conn, object_id object_id, plasma_object *result) {
object_table_entry *entry;
HASH_FIND(handle, sealed_objects, &object_id, sizeof(object_id), entry);
if (entry) {
plasma_reply reply;
memset(&reply, 0, sizeof(plasma_reply));
reply.data_offset = entry->offset;
reply.map_size = entry->map_size;
reply.data_size = entry->info.data_size;
reply.metadata_size = entry->info.metadata_size;
reply.store_fd_val = entry->fd;
send_fd(conn, entry->fd, (char*) &reply, sizeof(plasma_reply));
object_handle handle = {.store_fd = entry->fd,
.mmap_size = entry->map_size};
result->handle = handle;
result->data_offset = entry->offset;
result->metadata_offset = entry->offset + entry->info.data_size;
result->data_size = entry->info.data_size;
result->metadata_size = entry->info.metadata_size;
return OBJECT_FOUND;
} else {
object_notify_entry* notify_entry;
LOG_INFO("object not in hash table of sealed objects");
HASH_FIND(handle, objects_notify, &req->object_id, sizeof(plasma_id),
object_notify_entry *notify_entry;
LOG_DEBUG("object not in hash table of sealed objects");
HASH_FIND(handle, objects_notify, &object_id, sizeof(object_id),
notify_entry);
if (!notify_entry) {
notify_entry = malloc(sizeof(object_notify_entry));
memset(notify_entry, 0, sizeof(object_notify_entry));
notify_entry->num_waiting = 0;
memcpy(&notify_entry->object_id, &req->object_id, 20);
HASH_ADD(handle, objects_notify, object_id, sizeof(plasma_id),
utarray_new(notify_entry->conns, &ut_int_icd);
memcpy(&notify_entry->object_id, &object_id, 20);
HASH_ADD(handle, objects_notify, object_id, sizeof(object_id),
notify_entry);
}
PLASMA_CHECK(notify_entry->num_waiting < MAX_NUM_CLIENTS - 1,
"This exceeds the maximum number of clients.");
notify_entry->conn[notify_entry->num_waiting] = conn;
notify_entry->num_waiting += 1;
utarray_push_back(notify_entry->conns, &conn);
}
return OBJECT_NOT_FOUND;
}
/* Check if an object is present. */
void check_if_object_present(int conn, plasma_request* req) {
object_table_entry* entry;
HASH_FIND(handle, sealed_objects, &req->object_id, sizeof(plasma_id), entry);
plasma_reply reply;
memset(&reply, 0, sizeof(plasma_reply));
reply.has_object = entry ? 1 : 0;
plasma_send_reply(conn, &reply);
int contains_object(int conn, object_id object_id) {
object_table_entry *entry;
HASH_FIND(handle, sealed_objects, &object_id, sizeof(object_id), entry);
return entry ? OBJECT_FOUND : OBJECT_NOT_FOUND;
}
/* 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);
void seal_object(int conn,
object_id object_id,
UT_array **conns,
plasma_object *result) {
LOG_DEBUG("sealing object"); // TODO(pcm): add object_id here
object_table_entry *entry;
HASH_FIND(handle, open_objects, &object_id, sizeof(object_id), entry);
if (!entry) {
return; /* TODO(pcm): return error */
}
int fd = entry->fd;
HASH_DELETE(handle, open_objects, entry);
HASH_ADD(handle, sealed_objects, object_id, sizeof(plasma_id), entry);
HASH_ADD(handle, sealed_objects, object_id, sizeof(object_id), entry);
/* Inform processes that the object is ready now. */
object_notify_entry* notify_entry;
HASH_FIND(handle, objects_notify, &req->object_id, sizeof(plasma_id),
object_notify_entry *notify_entry;
HASH_FIND(handle, objects_notify, &object_id, sizeof(object_id),
notify_entry);
if (!notify_entry) {
*conns = NULL;
return;
}
plasma_reply reply = {.data_offset = entry->offset,
.map_size = entry->map_size,
.data_size = entry->info.data_size,
.metadata_size = entry->info.metadata_size,
.store_fd_val = fd};
for (int i = 0; i < notify_entry->num_waiting; ++i) {
send_fd(notify_entry->conn[i], entry->fd, (char*) &reply,
sizeof(plasma_reply));
}
object_handle handle = {.store_fd = entry->fd, .mmap_size = entry->map_size};
result->handle = handle;
result->data_offset = entry->offset;
result->metadata_offset = entry->offset + entry->info.data_size;
result->data_size = entry->info.data_size;
result->metadata_size = entry->info.metadata_size;
HASH_DELETE(handle, objects_notify, notify_entry);
*conns = notify_entry->conns;
free(notify_entry);
}
/* Delete an object that has been created in the hash table. */
void delete_object(int conn, plasma_request* req) {
LOG_INFO("deleting object"); // TODO(rkn): add object_id here
object_table_entry* entry;
HASH_FIND(handle, sealed_objects, &req->object_id, sizeof(plasma_id), entry);
void delete_object(int conn, object_id object_id) {
LOG_DEBUG("deleting object"); // TODO(rkn): add object_id here
object_table_entry *entry;
HASH_FIND(handle, sealed_objects, &object_id, sizeof(object_id), entry);
/* TODO(rkn): This should probably not fail, but should instead throw an
* error. Maybe we should also support deleting objects that have been created
* but not sealed. */
PLASMA_CHECK(entry != NULL, "To delete an object it must have been sealed.");
uint8_t* pointer = entry->pointer;
CHECKM(entry != NULL, "To delete an object it must have been sealed.");
uint8_t *pointer = entry->pointer;
HASH_DELETE(handle, sealed_objects, entry);
dlfree(pointer);
}
void process_event(int conn, plasma_request* req) {
switch (req->type) {
void process_message(event_loop *loop,
int client_sock,
void *context,
int events) {
int64_t type;
int64_t length;
plasma_request *req;
read_message(client_sock, &type, &length, (uint8_t **) &req);
plasma_reply reply;
memset(&reply, 0, sizeof(reply));
UT_array *conns;
switch (type) {
case PLASMA_CREATE:
create_object(conn, req);
create_object(client_sock, req->object_id, req->data_size,
req->metadata_size, &reply.object);
send_fd(client_sock, reply.object.handle.store_fd, (char *) &reply,
sizeof(reply));
break;
case PLASMA_GET:
get_object(conn, req);
if (get_object(client_sock, req->object_id, &reply.object) ==
OBJECT_FOUND) {
send_fd(client_sock, reply.object.handle.store_fd, (char *) &reply,
sizeof(reply));
}
break;
case PLASMA_CONTAINS:
check_if_object_present(conn, req);
if (contains_object(client_sock, req->object_id) == OBJECT_FOUND) {
reply.has_object = 1;
}
plasma_send_reply(client_sock, &reply);
break;
case PLASMA_SEAL:
seal_object(conn, req);
seal_object(client_sock, req->object_id, &conns, &reply.object);
if (conns) {
for (int *c = (int *) utarray_front(conns); c != NULL;
c = (int *) utarray_next(conns, c)) {
send_fd(*c, reply.object.handle.store_fd, (char *) &reply,
sizeof(reply));
}
utarray_free(conns);
}
break;
case PLASMA_DELETE:
delete_object(conn, req);
delete_object(client_sock, req->object_id);
break;
case DISCONNECT_CLIENT: {
LOG_DEBUG("Disconnecting client on fd %d", client_sock);
event_loop_remove_file(loop, client_sock);
} break;
default:
LOG_ERR("invalid request %d", req->type);
exit(-1);
/* This code should be unreachable. */
CHECK(0);
}
free(req);
}
void run_event_loop(int socket) {
plasma_store_state state;
init_state(&state);
event_loop_attach(state.loop, 0, NULL, socket, POLLIN);
plasma_request req;
while (1) {
int num_ready = event_loop_poll(state.loop);
if (num_ready < 0) {
LOG_ERR("poll failed");
exit(-1);
}
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 (waiting->fd == socket) {
/* Handle new incoming connections. */
int new_socket = accept(socket, NULL, NULL);
event_loop_attach(state.loop, 0, NULL, new_socket, POLLIN);
LOG_INFO("adding new client");
} else {
int r = read(waiting->fd, &req, sizeof(plasma_request));
if (r == -1) {
LOG_ERR("read error");
continue;
} else if (r == 0) {
LOG_INFO("connection %d disconnected", i);
event_loop_detach(state.loop, i, 1);
} else {
process_event(waiting->fd, &req);
}
}
}
}
void new_client_connection(event_loop *loop,
int listener_sock,
void *context,
int events) {
int new_socket = accept_client(listener_sock);
event_loop_add_file(loop, new_socket, EVENT_LOOP_READ, process_message,
context);
LOG_DEBUG("new connection with fd %d", new_socket);
}
void start_server(char* socket_name) {
int fd = socket(AF_UNIX, SOCK_STREAM, 0);
if (fd == -1) {
LOG_ERR("socket error");
exit(-1);
}
int on = 1;
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char*) &on, sizeof(on)) < 0) {
LOG_ERR("setsockopt failed");
close(fd);
exit(-1);
}
struct sockaddr_un addr;
memset(&addr, 0, sizeof(addr));
addr.sun_family = AF_UNIX;
strncpy(addr.sun_path, socket_name, sizeof(addr.sun_path) - 1);
unlink(socket_name);
bind(fd, (struct sockaddr*) &addr, sizeof(addr));
listen(fd, 5);
run_event_loop(fd);
void start_server(char *socket_name) {
int socket = bind_ipc_sock(socket_name);
CHECK(socket >= 0);
event_loop *loop = event_loop_create();
event_loop_add_file(loop, socket, EVENT_LOOP_READ, new_client_connection,
NULL);
event_loop_run(loop);
}
int main(int argc, char* argv[]) {
char* socket_name = NULL;
int main(int argc, char *argv[]) {
char *socket_name = NULL;
int c;
while ((c = getopt(argc, argv, "s:")) != -1) {
switch (c) {
@@ -309,6 +294,6 @@ int main(int argc, char* argv[]) {
LOG_ERR("please specify socket for incoming connections with -s switch");
exit(-1);
}
LOG_INFO("starting server listening on %s", socket_name);
LOG_DEBUG("starting server listening on %s", socket_name);
start_server(socket_name);
}
src/plasma_store.h
+49
View File
@@ -0,0 +1,49 @@
#ifndef PLASMA_STORE_H
#define PLASMA_STORE_H
#include "plasma.h"
/**
* Create a new object:
*
* @param object_id Object ID of the object to be created.
* @param data_size Size in bytes of the object to be created.
* @param metadata_size Size in bytes of the object metadata.
*/
void create_object(int conn,
object_id object_id,
int64_t data_size,
int64_t metadata_size,
plasma_object *result);
/**
* Get an object:
*
* @param object_id Object ID of the object to be gotten.
*
* Returns the status of the object (object_status in plasma.h).
*/
int get_object(int conn, object_id object_id, plasma_object *result);
/**
* Seal an object:
*
* @param object_id Object ID of the object to be sealed.
* @param conns Returns the connection that are waiting for this object.
The caller is responsible for destroying this array.
*
* Should notify all the sockets waiting for the object.
*/
plasma_object seal_object(int conn,
object_id object_id,
UT_array **conns,
plasma_object *result);
/**
* Check if the plasma store contains an object:
*
* @param object_id Object ID that will be checked.
*/
int contains_object(int conn, object_id object_id);
#endif /* PLASMA_STORE_H */
src/utarray.h
-238
View File
@@ -1,238 +0,0 @@
/*
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 */
src/uthash.h
-1074
View File
File diff suppressed because it is too large Load Diff