mirror of
https://github.com/wassname/ray.git
synced 2026-09-09 11:32:43 +08:00
Merge remote-tracking branch 'r1remote/moveout' into switch
This commit is contained in:
Robert Nishihara
@@ -0,0 +1,40 @@
|
||||
CC = gcc
|
||||
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 $(BUILD)/libplasma_client.a
|
||||
|
||||
debug: FORCE
|
||||
debug: CFLAGS += -DRAY_COMMON_DEBUG=1
|
||||
debug: all
|
||||
|
||||
clean:
|
||||
cd common; make clean
|
||||
rm -r $(BUILD)/*
|
||||
|
||||
$(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/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 common/thirdparty/hiredis/libhiredis.a -o $(BUILD)/plasma_manager
|
||||
|
||||
$(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)/libplasma_client.a: src/plasma_client.o src/fling.o
|
||||
ar rcs $@ $^
|
||||
|
||||
$(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
|
||||
|
||||
# Set the request timeout low for testing purposes.
|
||||
test: CFLAGS += -DRAY_TIMEOUT=50
|
||||
test: FORCE
|
||||
cd common; make redis
|
||||
test: all
|
||||
|
||||
FORCE:
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,47 @@
|
||||
/* A simple example on how to use the plasma store
|
||||
*
|
||||
* Can be called in the following way:
|
||||
*
|
||||
* cd build
|
||||
* ./plasma_store -s /tmp/plasma_socket
|
||||
* ./example -s /tmp/plasma_socket -g
|
||||
* ./example -s /tmp/plasma_socket -c -f */
|
||||
|
||||
#include <stdlib.h>
|
||||
#include <getopt.h>
|
||||
#include <unistd.h>
|
||||
#include <assert.h>
|
||||
|
||||
#include "plasma.h"
|
||||
#include "plasma_client.h"
|
||||
|
||||
int main(int argc, char *argv[]) {
|
||||
plasma_connection *conn = NULL;
|
||||
int64_t size;
|
||||
uint8_t *data;
|
||||
int c;
|
||||
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) {
|
||||
case 's':
|
||||
conn = plasma_connect(optarg, NULL, 0);
|
||||
break;
|
||||
case 'c':
|
||||
assert(conn != NULL);
|
||||
plasma_create(conn, id, 100, NULL, 0, &data);
|
||||
break;
|
||||
case 'f':
|
||||
assert(conn != NULL);
|
||||
plasma_seal(conn, id);
|
||||
break;
|
||||
case 'g':
|
||||
plasma_get(conn, id, &size, &data, NULL, NULL);
|
||||
break;
|
||||
default:
|
||||
abort();
|
||||
}
|
||||
}
|
||||
assert(conn != NULL);
|
||||
plasma_disconnect(conn);
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
#include "fling.h"
|
||||
|
||||
#include <string.h>
|
||||
|
||||
void init_msg(struct msghdr *msg,
|
||||
struct iovec *iov,
|
||||
char *buf,
|
||||
size_t buf_len) {
|
||||
iov->iov_base = buf;
|
||||
iov->iov_len = 1;
|
||||
|
||||
msg->msg_iov = iov;
|
||||
msg->msg_iovlen = 1;
|
||||
msg->msg_control = buf;
|
||||
msg->msg_controllen = buf_len;
|
||||
msg->msg_name = NULL;
|
||||
msg->msg_namelen = 0;
|
||||
}
|
||||
|
||||
int send_fd(int conn, int fd, const char *payload, int size) {
|
||||
struct msghdr msg;
|
||||
struct iovec iov;
|
||||
char buf[CMSG_SPACE(sizeof(int))];
|
||||
memset(&buf, 0, CMSG_SPACE(sizeof(int)));
|
||||
|
||||
init_msg(&msg, &iov, buf, sizeof(buf));
|
||||
|
||||
struct cmsghdr *header = CMSG_FIRSTHDR(&msg);
|
||||
header->cmsg_level = SOL_SOCKET;
|
||||
header->cmsg_type = SCM_RIGHTS;
|
||||
header->cmsg_len = CMSG_LEN(sizeof(int));
|
||||
*(int *) CMSG_DATA(header) = fd;
|
||||
|
||||
/* send file descriptor and payload */
|
||||
return sendmsg(conn, &msg, 0) != -1 && send(conn, payload, size, 0) == -1;
|
||||
}
|
||||
|
||||
int recv_fd(int conn, char *payload, int size) {
|
||||
struct msghdr msg;
|
||||
struct iovec iov;
|
||||
char buf[CMSG_SPACE(sizeof(int))];
|
||||
init_msg(&msg, &iov, buf, sizeof(buf));
|
||||
|
||||
if (recvmsg(conn, &msg, 0) == -1)
|
||||
return -1;
|
||||
|
||||
int found_fd = -1;
|
||||
int oh_noes = 0;
|
||||
for (struct cmsghdr *header = CMSG_FIRSTHDR(&msg); header != NULL;
|
||||
header = CMSG_NXTHDR(&msg, header))
|
||||
if (header->cmsg_level == SOL_SOCKET && header->cmsg_type == SCM_RIGHTS) {
|
||||
int count =
|
||||
(header->cmsg_len - (CMSG_DATA(header) - (unsigned char *) header)) /
|
||||
sizeof(int);
|
||||
for (int i = 0; i < count; ++i) {
|
||||
int fd = ((int *) CMSG_DATA(header))[i];
|
||||
if (found_fd == -1) {
|
||||
found_fd = fd;
|
||||
} else {
|
||||
close(fd);
|
||||
oh_noes = 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* The sender sent us more than one file descriptor. We've closed
|
||||
* them all to prevent fd leaks but notify the caller that we got
|
||||
* a bad message. */
|
||||
if (oh_noes) {
|
||||
close(found_fd);
|
||||
errno = EBADMSG;
|
||||
return -1;
|
||||
}
|
||||
|
||||
ssize_t len = recv(conn, payload, size, 0);
|
||||
if (len < 0) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
return found_fd;
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
/* FLING: Exchanging file descriptors over sockets
|
||||
*
|
||||
* This is a little library for sending file descriptors over a socket
|
||||
* between processes. The reason for doing that (as opposed to using
|
||||
* filenames to share the files) is so (a) no files remain in the
|
||||
* filesystem after all the processes terminate, (b) to make sure that
|
||||
* there are no name collisions and (c) to be able to control who has
|
||||
* access to the data.
|
||||
*
|
||||
* Most of the code is from https://github.com/sharvil/flingfd */
|
||||
|
||||
#include <unistd.h>
|
||||
#include <errno.h>
|
||||
#include <sys/types.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/un.h>
|
||||
|
||||
/* This is neccessary for Mac OS X, see http://www.apuebook.com/faqs2e.html
|
||||
* (10). */
|
||||
#if !defined(CMSG_SPACE) && !defined(CMSG_LEN)
|
||||
#define CMSG_SPACE(len) \
|
||||
(__DARWIN_ALIGN32(sizeof(struct cmsghdr)) + __DARWIN_ALIGN32(len))
|
||||
#define CMSG_LEN(len) (__DARWIN_ALIGN32(sizeof(struct cmsghdr)) + (len))
|
||||
#endif
|
||||
|
||||
void init_msg(struct msghdr *msg, struct iovec *iov, char *buf, size_t buf_len);
|
||||
|
||||
/* Send a file descriptor "fd" and a payload "payload" of size "size"
|
||||
* over the socket "conn". Return 0 on success. */
|
||||
int send_fd(int conn, int fd, const char *payload, int size);
|
||||
|
||||
/* Receive a file descriptor and a payload of size up to "size" from a
|
||||
* socket "conn". The payload will be written to "payload" and the file
|
||||
* descriptor will be returned. Returns -1 on failure. */
|
||||
int recv_fd(int conn, char *payload, int size);
|
||||
@@ -0,0 +1,246 @@
|
||||
import os
|
||||
import socket
|
||||
import ctypes
|
||||
import time
|
||||
|
||||
Addr = ctypes.c_ubyte * 4
|
||||
|
||||
PLASMA_ID_SIZE = 20
|
||||
ID = ctypes.c_ubyte * PLASMA_ID_SIZE
|
||||
|
||||
class PlasmaID(ctypes.Structure):
|
||||
_fields_ = [("plasma_id", ID)]
|
||||
|
||||
def make_plasma_id(string):
|
||||
if len(string) != PLASMA_ID_SIZE:
|
||||
raise Exception("PlasmaIDs must be {} characters long".format(PLASMA_ID_SIZE))
|
||||
object_id = map(ord, string)
|
||||
return PlasmaID(plasma_id=ID(*object_id))
|
||||
|
||||
class PlasmaBuffer(object):
|
||||
"""This is the type of objects returned by calls to get with a PlasmaClient.
|
||||
|
||||
We define our own class instead of directly returning a buffer object so that
|
||||
we can add a custom destructor which notifies Plasma that the object is no
|
||||
longer being used, so the memory in the Plasma store backing the object can
|
||||
potentially be freed.
|
||||
|
||||
Attributes:
|
||||
buffer (buffer): A buffer containing an object in the Plasma store.
|
||||
plasma_id (PlasmaID): The ID of the object in the buffer.
|
||||
plasma_client (PlasmaClient): The PlasmaClient that we use to communicate
|
||||
with the store and manager.
|
||||
"""
|
||||
def __init__(self, buff, plasma_id, plasma_client):
|
||||
"""Initialize a PlasmaBuffer."""
|
||||
self.buffer = buff
|
||||
self.plasma_id = plasma_id
|
||||
self.plasma_client = plasma_client
|
||||
|
||||
def __del__(self):
|
||||
"""Notify Plasma that the object is no longer needed."""
|
||||
self.plasma_client.client.plasma_release(self.plasma_client.plasma_conn, self.plasma_id)
|
||||
|
||||
def __getitem__(self, index):
|
||||
"""Read from the PlasmaBuffer as if it were just a regular buffer."""
|
||||
return self.buffer[index]
|
||||
|
||||
def __setitem__(self, index, value):
|
||||
"""Write to the PlasmaBuffer as if it were just a regular buffer.
|
||||
|
||||
This should fail because the buffer should be read only.
|
||||
"""
|
||||
self.buffer[index] = value
|
||||
|
||||
def __len__(self):
|
||||
"""Return the length of the buffer."""
|
||||
return len(self.buffer)
|
||||
|
||||
class PlasmaClient(object):
|
||||
"""The PlasmaClient is used to interface with a plasma store and a plasma manager.
|
||||
|
||||
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, addr=None, port=None):
|
||||
"""Initialize the PlasmaClient.
|
||||
|
||||
Args:
|
||||
socket_name (str): Name of the socket the plasma store is listening at.
|
||||
addr (str): IPv4 address of plasma manager attached to the plasma store.
|
||||
port (int): Port number of the plasma manager attached to the plasma store.
|
||||
"""
|
||||
if port is not None:
|
||||
if not isinstance(port, int):
|
||||
raise Exception("The 'port' argument must be an integer. The given argument has type {}.".format(type(port)))
|
||||
if not 0 < port < 65536:
|
||||
raise Exception("The 'port' argument must be greater than 0 and less than 65536. The given value is {}.".format(port))
|
||||
|
||||
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)
|
||||
|
||||
self.client.plasma_connect.restype = ctypes.c_void_p
|
||||
self.client.plasma_create.restype = None
|
||||
self.client.plasma_get.restype = None
|
||||
self.client.plasma_release.restype = None
|
||||
self.client.plasma_contains.restype = None
|
||||
self.client.plasma_seal.restype = None
|
||||
self.client.plasma_delete.restype = None
|
||||
self.client.plasma_subscribe.restype = ctypes.c_int
|
||||
|
||||
self.buffer_from_memory = ctypes.pythonapi.PyBuffer_FromMemory
|
||||
self.buffer_from_memory.argtypes = [ctypes.c_void_p, ctypes.c_int64]
|
||||
self.buffer_from_memory.restype = ctypes.py_object
|
||||
|
||||
self.buffer_from_read_write_memory = ctypes.pythonapi.PyBuffer_FromReadWriteMemory
|
||||
self.buffer_from_read_write_memory.argtypes = [ctypes.c_void_p, ctypes.c_int64]
|
||||
self.buffer_from_read_write_memory.restype = ctypes.py_object
|
||||
|
||||
if addr is not None and port is not None:
|
||||
self.has_manager_conn = True
|
||||
self.plasma_conn = ctypes.c_void_p(self.client.plasma_connect(socket_name, addr, port))
|
||||
else:
|
||||
self.has_manager_conn = False
|
||||
self.plasma_conn = ctypes.c_void_p(self.client.plasma_connect(socket_name, None, 0))
|
||||
|
||||
def create(self, object_id, size, metadata=None):
|
||||
"""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.
|
||||
metadata (buffer): An optional buffer encoding whatever metadata the user
|
||||
wishes to encode.
|
||||
"""
|
||||
# This is used to hold the address of the buffer.
|
||||
data = ctypes.c_void_p()
|
||||
# Turn the metadata into the right type.
|
||||
metadata = buffer("") if metadata is None else metadata
|
||||
metadata = (ctypes.c_ubyte * len(metadata)).from_buffer_copy(metadata)
|
||||
self.client.plasma_create(self.plasma_conn, make_plasma_id(object_id), size, ctypes.cast(metadata, ctypes.POINTER(ctypes.c_ubyte * len(metadata))), len(metadata), ctypes.byref(data))
|
||||
return PlasmaBuffer(self.buffer_from_read_write_memory(data, size), make_plasma_id(object_id), self)
|
||||
|
||||
def get(self, object_id):
|
||||
"""Create a buffer from the PlasmaStore based on object ID.
|
||||
|
||||
If the object has not been sealed yet, this call will block. The retrieved
|
||||
buffer is immutable.
|
||||
|
||||
Args:
|
||||
object_id (str): A string used to identify an object.
|
||||
"""
|
||||
size = ctypes.c_int64()
|
||||
data = ctypes.c_void_p()
|
||||
metadata_size = ctypes.c_int64()
|
||||
metadata = ctypes.c_void_p()
|
||||
self.client.plasma_get(self.plasma_conn, make_plasma_id(object_id), ctypes.byref(size), ctypes.byref(data), ctypes.byref(metadata_size), ctypes.byref(metadata))
|
||||
return PlasmaBuffer(self.buffer_from_memory(data, size), make_plasma_id(object_id), self)
|
||||
|
||||
def get_metadata(self, object_id):
|
||||
"""Create a buffer from the PlasmaStore based on object ID.
|
||||
|
||||
If the object has not been sealed yet, this call will block until the object
|
||||
has been sealed. The retrieved buffer is immutable.
|
||||
|
||||
Args:
|
||||
object_id (str): A string used to identify an object.
|
||||
"""
|
||||
size = ctypes.c_int64()
|
||||
data = ctypes.c_void_p()
|
||||
metadata_size = ctypes.c_int64()
|
||||
metadata = ctypes.c_void_p()
|
||||
self.client.plasma_get(self.plasma_conn, make_plasma_id(object_id), ctypes.byref(size), ctypes.byref(data), ctypes.byref(metadata_size), ctypes.byref(metadata))
|
||||
return PlasmaBuffer(self.buffer_from_memory(metadata, metadata_size), make_plasma_id(object_id), self)
|
||||
|
||||
def contains(self, object_id):
|
||||
"""Check if the object is present and has been sealed in the PlasmaStore.
|
||||
|
||||
Args:
|
||||
object_id (str): A string used to identify an object.
|
||||
"""
|
||||
has_object = ctypes.c_int()
|
||||
self.client.plasma_contains(self.plasma_conn, make_plasma_id(object_id), ctypes.byref(has_object))
|
||||
has_object = has_object.value
|
||||
if has_object == 1:
|
||||
return True
|
||||
elif has_object == 0:
|
||||
return False
|
||||
else:
|
||||
raise Exception("This code should be unreachable.")
|
||||
|
||||
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.plasma_conn, make_plasma_id(object_id))
|
||||
|
||||
def delete(self, object_id):
|
||||
"""Delete the buffer in the PlasmaStore for a particular object ID.
|
||||
|
||||
Once a buffer has been deleted, the buffer is no longer accessible.
|
||||
|
||||
Args:
|
||||
object_id (str): A string used to identify an object.
|
||||
"""
|
||||
self.client.plasma_delete(self.plasma_conn, make_plasma_id(object_id))
|
||||
|
||||
def transfer(self, addr, port, object_id):
|
||||
"""Transfer local object with id object_id to another plasma instance
|
||||
|
||||
Args:
|
||||
addr (str): IPv4 address of the plasma instance the object is sent to.
|
||||
port (int): Port number of the plasma instance the object is sent to.
|
||||
object_id (str): A string used to identify an object.
|
||||
"""
|
||||
if not self.has_manager_conn:
|
||||
raise Exception("Not connected to the plasma manager socket")
|
||||
self.client.plasma_transfer(self.plasma_conn, addr, port, make_plasma_id(object_id))
|
||||
|
||||
def fetch(self, object_ids):
|
||||
"""Fetch the object with id object_id from another plasma manager instance.
|
||||
|
||||
Args:
|
||||
object_id (str): A string used to identify an object.
|
||||
"""
|
||||
object_id_array = (len(object_ids) * PlasmaID)()
|
||||
for i, object_id in enumerate(object_ids):
|
||||
object_id_array[i] = make_plasma_id(object_id)
|
||||
success_array = (len(object_ids) * ctypes.c_int)()
|
||||
if not self.has_manager_conn:
|
||||
raise Exception("Not connected to the plasma manager socket")
|
||||
self.client.plasma_fetch(self.plasma_conn,
|
||||
object_id_array._length_,
|
||||
object_id_array,
|
||||
success_array);
|
||||
return [bool(success) for success in success_array]
|
||||
|
||||
def subscribe(self):
|
||||
"""Subscribe to notifications about sealed objects."""
|
||||
fd = self.client.plasma_subscribe(self.plasma_conn)
|
||||
self.notification_sock = socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
# Make the socket non-blocking.
|
||||
self.notification_sock.setblocking(0)
|
||||
|
||||
def get_next_notification(self):
|
||||
"""Get the next notification from the notification socket."""
|
||||
if not self.notification_sock:
|
||||
raise Exception("To get notifications, first call subscribe.")
|
||||
# Loop until we've read PLASMA_ID_SIZE bytes from the socket.
|
||||
while True:
|
||||
try:
|
||||
message_data = self.notification_sock.recv(PLASMA_ID_SIZE)
|
||||
except socket.error:
|
||||
time.sleep(0.001)
|
||||
else:
|
||||
assert len(message_data) == PLASMA_ID_SIZE
|
||||
break
|
||||
return message_data
|
||||
@@ -0,0 +1,140 @@
|
||||
#include <assert.h>
|
||||
#include <stdlib.h>
|
||||
#include <stddef.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <sys/mman.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "common.h"
|
||||
#include "plasma.h"
|
||||
#include "uthash.h"
|
||||
|
||||
void *fake_mmap(size_t);
|
||||
int fake_munmap(void *, size_t);
|
||||
|
||||
#define MMAP(s) fake_mmap(s)
|
||||
#define MUNMAP(a, s) fake_munmap(a, s)
|
||||
#define DIRECT_MMAP(s) fake_mmap(s)
|
||||
#define DIRECT_MUNMAP(a, s) fake_munmap(a, s)
|
||||
#define USE_DL_PREFIX
|
||||
#define HAVE_MORECORE 0
|
||||
#define DEFAULT_MMAP_THRESHOLD MAX_SIZE_T
|
||||
#define DEFAULT_GRANULARITY ((size_t) 128U * 1024U)
|
||||
|
||||
#include "thirdparty/dlmalloc.c"
|
||||
|
||||
#undef MMAP
|
||||
#undef MUNMAP
|
||||
#undef DIRECT_MMAP
|
||||
#undef DIRECT_MUNMAP
|
||||
#undef USE_DL_PREFIX
|
||||
#undef HAVE_MORECORE
|
||||
#undef DEFAULT_GRANULARITY
|
||||
|
||||
struct mmap_record {
|
||||
int fd;
|
||||
void *pointer;
|
||||
int64_t size;
|
||||
UT_hash_handle hh_fd;
|
||||
UT_hash_handle hh_pointer;
|
||||
};
|
||||
|
||||
/* TODO(rshin): Don't have two hash tables. */
|
||||
struct mmap_record *records_by_fd = NULL;
|
||||
struct mmap_record *records_by_pointer = NULL;
|
||||
|
||||
const int GRANULARITY_MULTIPLIER = 2;
|
||||
|
||||
/* Create a buffer. This is creating a temporary file and then
|
||||
* immediately unlinking it so we do not leave traces in the system. */
|
||||
int create_buffer(int64_t size) {
|
||||
static char template[] = "/tmp/plasmaXXXXXX";
|
||||
char file_name[32];
|
||||
strncpy(file_name, template, 32);
|
||||
int fd = mkstemp(file_name);
|
||||
if (fd < 0)
|
||||
return -1;
|
||||
FILE *file = fdopen(fd, "a+");
|
||||
if (!file) {
|
||||
close(fd);
|
||||
return -1;
|
||||
}
|
||||
if (unlink(file_name) != 0) {
|
||||
LOG_ERR("unlink error");
|
||||
return -1;
|
||||
}
|
||||
if (ftruncate(fd, (off_t) size) != 0) {
|
||||
LOG_ERR("ftruncate error");
|
||||
return -1;
|
||||
}
|
||||
return fd;
|
||||
}
|
||||
|
||||
void *fake_mmap(size_t size) {
|
||||
/* Add sizeof(size_t) so that the returned pointer is deliberately not
|
||||
* page-aligned. This ensures that the segments of memory returned by
|
||||
* fake_mmap are never contiguous. */
|
||||
size += sizeof(size_t);
|
||||
|
||||
int fd = create_buffer(size);
|
||||
void *pointer = mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
|
||||
if (pointer == MAP_FAILED) {
|
||||
return pointer;
|
||||
}
|
||||
|
||||
/* Increase dlmalloc's allocation granularity directly. */
|
||||
mparams.granularity *= GRANULARITY_MULTIPLIER;
|
||||
|
||||
struct mmap_record *record = malloc(sizeof(struct mmap_record));
|
||||
record->fd = fd;
|
||||
record->pointer = pointer;
|
||||
record->size = size;
|
||||
HASH_ADD(hh_fd, records_by_fd, fd, sizeof(fd), record);
|
||||
HASH_ADD(hh_pointer, records_by_pointer, pointer, sizeof(pointer), record);
|
||||
|
||||
/* We lie to dlmalloc about where mapped memory actually lives. */
|
||||
pointer += sizeof(size_t);
|
||||
LOG_DEBUG("%p = fake_mmap(%lu)", pointer, size);
|
||||
return pointer;
|
||||
}
|
||||
|
||||
int fake_munmap(void *addr, size_t size) {
|
||||
LOG_DEBUG("fake_munmap(%p, %lu)", addr, size);
|
||||
addr -= sizeof(size_t);
|
||||
size += sizeof(size_t);
|
||||
|
||||
struct mmap_record *record;
|
||||
|
||||
HASH_FIND(hh_pointer, records_by_pointer, &addr, sizeof(addr), record);
|
||||
if (record == NULL || record->size != size) {
|
||||
/* Reject requests to munmap that don't directly match previous
|
||||
* calls to mmap, to prevent dlmalloc from trimming. */
|
||||
return -1;
|
||||
}
|
||||
close(record->fd);
|
||||
|
||||
HASH_DELETE(hh_fd, records_by_fd, record);
|
||||
HASH_DELETE(hh_pointer, records_by_pointer, record);
|
||||
|
||||
return munmap(addr, size);
|
||||
}
|
||||
|
||||
void get_malloc_mapinfo(void *addr,
|
||||
int *fd,
|
||||
int64_t *map_size,
|
||||
ptrdiff_t *offset) {
|
||||
struct mmap_record *record;
|
||||
/* TODO(rshin): Implement a more efficient search through records_by_fd. */
|
||||
for (record = records_by_fd; record != NULL; record = record->hh_fd.next) {
|
||||
if (addr >= record->pointer && addr < record->pointer + record->size) {
|
||||
*fd = record->fd;
|
||||
*map_size = record->size;
|
||||
*offset = addr - record->pointer;
|
||||
return;
|
||||
}
|
||||
}
|
||||
*fd = -1;
|
||||
*map_size = 0;
|
||||
*offset = 0;
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
#ifndef MALLOC_H
|
||||
#define MALLOC_H
|
||||
|
||||
void get_malloc_mapinfo(void *addr,
|
||||
int *fd,
|
||||
int64_t *map_length,
|
||||
ptrdiff_t *offset);
|
||||
|
||||
#endif /* MALLOC_H */
|
||||
@@ -0,0 +1,96 @@
|
||||
#ifndef PLASMA_H
|
||||
#define PLASMA_H
|
||||
|
||||
#include <inttypes.h>
|
||||
#include <stdio.h>
|
||||
#include <errno.h>
|
||||
#include <stddef.h>
|
||||
#include <string.h>
|
||||
|
||||
#include "common.h"
|
||||
|
||||
typedef struct {
|
||||
int64_t data_size;
|
||||
int64_t metadata_size;
|
||||
int64_t create_time;
|
||||
int64_t construct_duration;
|
||||
} plasma_object_info;
|
||||
|
||||
/* 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;
|
||||
|
||||
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 = 128,
|
||||
/** Get an object. */
|
||||
PLASMA_GET,
|
||||
/** Tell the store that the client no longer needs an object. */
|
||||
PLASMA_RELEASE,
|
||||
/** Check if an object is present. */
|
||||
PLASMA_CONTAINS,
|
||||
/** Seal an object. */
|
||||
PLASMA_SEAL,
|
||||
/** Delete an object. */
|
||||
PLASMA_DELETE,
|
||||
/** Subscribe to notifications about sealed objects. */
|
||||
PLASMA_SUBSCRIBE,
|
||||
/** Request transfer to another store. */
|
||||
PLASMA_TRANSFER,
|
||||
/** Header for sending data. */
|
||||
PLASMA_DATA,
|
||||
/** Request a fetch of an object in another store. */
|
||||
PLASMA_FETCH,
|
||||
};
|
||||
|
||||
typedef struct {
|
||||
/** The size of the object's data. */
|
||||
int64_t data_size;
|
||||
/** The size of the object's metadata. */
|
||||
int64_t metadata_size;
|
||||
/** In a transfer request, this is the IP address of the Plasma Manager to
|
||||
* transfer the object to. */
|
||||
uint8_t addr[4];
|
||||
/** In a transfer request, this is the port of the Plasma Manager to transfer
|
||||
* the object to. */
|
||||
int port;
|
||||
/** The number of object IDs that will be included in this request. */
|
||||
int num_object_ids;
|
||||
/** The IDs of the objects that the request is about. */
|
||||
object_id object_ids[1];
|
||||
} plasma_request;
|
||||
|
||||
typedef struct {
|
||||
/** The object ID that this reply refers to. */
|
||||
object_id object_id;
|
||||
/** The object that is returned with this reply. */
|
||||
plasma_object object;
|
||||
/** This is used only to respond to requests of type
|
||||
* PLASMA_CONTAINS or PLASMA_FETCH. It is 1 if the object is
|
||||
* present and 0 otherwise. Used for plasma_contains and
|
||||
* plasma_fetch. */
|
||||
int has_object;
|
||||
} plasma_reply;
|
||||
|
||||
#endif
|
||||
@@ -0,0 +1,434 @@
|
||||
/* PLASMA CLIENT: Client library for using the plasma store and manager */
|
||||
|
||||
#include <assert.h>
|
||||
#include <fcntl.h>
|
||||
#include <stdlib.h>
|
||||
#include <stdio.h>
|
||||
#include <sys/types.h>
|
||||
#include <unistd.h>
|
||||
#include <sys/mman.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/un.h>
|
||||
#include <strings.h>
|
||||
#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"
|
||||
|
||||
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;
|
||||
/** The length of the memory-mapped file. */
|
||||
size_t length;
|
||||
/** The number of objects in this memory-mapped file that are currently being
|
||||
* used by the client. When this count reaches zeros, we unmap the file. */
|
||||
int count;
|
||||
/** Handle for the uthash table. */
|
||||
UT_hash_handle hh;
|
||||
} client_mmap_table_entry;
|
||||
|
||||
typedef struct {
|
||||
/** The ID of the object. This is used as the key in the hash table. */
|
||||
object_id object_id;
|
||||
/** The file descriptor of the memory-mapped file that contains the object. */
|
||||
int fd;
|
||||
/** A count of the number of times this client has called plasma_create or
|
||||
* plasma_get on this object ID minus the number of calls to plasma_release.
|
||||
* When this count reaches zero, we remove the entry from the objects_in_use
|
||||
* and decrement a count in the relevant client_mmap_table_entry. */
|
||||
int count;
|
||||
/** Handle for the uthash table. */
|
||||
UT_hash_handle hh;
|
||||
} object_in_use_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_connection {
|
||||
/** File descriptor of the Unix domain socket that connects to the store. */
|
||||
int store_conn;
|
||||
/** File descriptor of the Unix domain socket that connects to the manager. */
|
||||
int manager_conn;
|
||||
/** Table of dlmalloc buffer files that have been memory mapped so far. This
|
||||
* is a hash table mapping a file descriptor to a struct containing the
|
||||
* address of the corresponding memory-mapped file. */
|
||||
client_mmap_table_entry *mmap_table;
|
||||
/** A hash table of the object IDs that are currently being used by this
|
||||
* client. */
|
||||
object_in_use_entry *objects_in_use;
|
||||
};
|
||||
|
||||
int plasma_request_size(int num_object_ids) {
|
||||
int object_ids_size = (num_object_ids - 1) * sizeof(object_id);
|
||||
return sizeof(plasma_request) + object_ids_size;
|
||||
}
|
||||
|
||||
void plasma_send_request(int fd, int type, plasma_request *req) {
|
||||
int req_size = plasma_request_size(req->num_object_ids);
|
||||
int error = write_message(fd, type, req_size, (uint8_t *) req);
|
||||
/* TODO(swang): Actually handle the write error. */
|
||||
CHECK(!error);
|
||||
}
|
||||
|
||||
plasma_request make_plasma_request(object_id object_id) {
|
||||
plasma_request req = {.num_object_ids = 1, .object_ids = {object_id}};
|
||||
return req;
|
||||
}
|
||||
|
||||
plasma_request *make_plasma_multiple_request(int num_object_ids,
|
||||
object_id object_ids[]) {
|
||||
int req_size = plasma_request_size(num_object_ids);
|
||||
plasma_request *req = malloc(req_size);
|
||||
req->num_object_ids = num_object_ids;
|
||||
memcpy(&req->object_ids, object_ids, num_object_ids * sizeof(object_id));
|
||||
return req;
|
||||
}
|
||||
|
||||
/* If the file descriptor fd has been mmapped in this client process before,
|
||||
* return the pointer that was returned by mmap, otherwise mmap it and store the
|
||||
* pointer in a hash table. */
|
||||
uint8_t *lookup_or_mmap(plasma_connection *conn,
|
||||
int fd,
|
||||
int store_fd_val,
|
||||
int64_t map_size) {
|
||||
client_mmap_table_entry *entry;
|
||||
HASH_FIND_INT(conn->mmap_table, &store_fd_val, entry);
|
||||
if (entry) {
|
||||
close(fd);
|
||||
return entry->pointer;
|
||||
} else {
|
||||
uint8_t *result =
|
||||
mmap(NULL, map_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
|
||||
if (result == MAP_FAILED) {
|
||||
LOG_ERR("mmap failed");
|
||||
exit(-1);
|
||||
}
|
||||
close(fd);
|
||||
entry = malloc(sizeof(client_mmap_table_entry));
|
||||
entry->key = store_fd_val;
|
||||
entry->pointer = result;
|
||||
entry->length = map_size;
|
||||
entry->count = 0;
|
||||
HASH_ADD_INT(conn->mmap_table, key, entry);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
void increment_object_count(plasma_connection *conn,
|
||||
object_id object_id,
|
||||
int fd) {
|
||||
/* Increment the count of the object to track the fact that it is being used.
|
||||
* The corresponding decrement should happen in plasma_release. */
|
||||
object_in_use_entry *object_entry;
|
||||
HASH_FIND(hh, conn->objects_in_use, &object_id, sizeof(object_id),
|
||||
object_entry);
|
||||
if (object_entry == NULL) {
|
||||
/* Add this object ID to the hash table of object IDs in use. The
|
||||
* corresponding call to free happens in plasma_release. */
|
||||
object_entry = malloc(sizeof(object_in_use_entry));
|
||||
object_entry->object_id = object_id;
|
||||
object_entry->fd = fd;
|
||||
object_entry->count = 0;
|
||||
HASH_ADD(hh, conn->objects_in_use, object_id, sizeof(object_id),
|
||||
object_entry);
|
||||
/* Increment the count of the number of objects in the memory-mapped file
|
||||
* that are being used. The corresponding decrement should happen in
|
||||
* plasma_release. */
|
||||
client_mmap_table_entry *entry;
|
||||
HASH_FIND_INT(conn->mmap_table, &object_entry->fd, entry);
|
||||
CHECK(entry != NULL);
|
||||
CHECK(entry->count >= 0);
|
||||
entry->count += 1;
|
||||
} else {
|
||||
CHECK(object_entry->count > 0);
|
||||
}
|
||||
/* Increment the count of the number of instances of this object that are
|
||||
* being used by this client. The corresponding decrement should happen in
|
||||
* plasma_release. */
|
||||
object_entry->count += 1;
|
||||
}
|
||||
|
||||
void plasma_create(plasma_connection *conn,
|
||||
object_id object_id,
|
||||
int64_t data_size,
|
||||
uint8_t *metadata,
|
||||
int64_t metadata_size,
|
||||
uint8_t **data) {
|
||||
LOG_DEBUG("called plasma_create on conn %d with size %" PRId64
|
||||
" and metadata size "
|
||||
"%" PRId64,
|
||||
conn->store_conn, data_size, metadata_size);
|
||||
plasma_request req = make_plasma_request(object_id);
|
||||
req.data_size = data_size;
|
||||
req.metadata_size = metadata_size;
|
||||
plasma_send_request(conn->store_conn, PLASMA_CREATE, &req);
|
||||
plasma_reply reply;
|
||||
int fd = recv_fd(conn->store_conn, (char *) &reply, sizeof(plasma_reply));
|
||||
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. */
|
||||
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 + object->data_size, metadata, metadata_size);
|
||||
}
|
||||
/* Increment the count of the number of instances of this object that this
|
||||
* client is using. A call to plasma_release is required to decrement this
|
||||
* count. */
|
||||
increment_object_count(conn, object_id, object->handle.store_fd);
|
||||
}
|
||||
|
||||
/* This method is used to get both the data and the metadata. */
|
||||
void plasma_get(plasma_connection *conn,
|
||||
object_id object_id,
|
||||
int64_t *size,
|
||||
uint8_t **data,
|
||||
int64_t *metadata_size,
|
||||
uint8_t **metadata) {
|
||||
plasma_request req = make_plasma_request(object_id);
|
||||
plasma_send_request(conn->store_conn, PLASMA_GET, &req);
|
||||
plasma_reply reply;
|
||||
int fd = recv_fd(conn->store_conn, (char *) &reply, sizeof(plasma_reply));
|
||||
CHECKM(fd != -1, "recv not successful");
|
||||
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 + object->data_size;
|
||||
*metadata_size = object->metadata_size;
|
||||
}
|
||||
/* Increment the count of the number of instances of this object that this
|
||||
* client is using. A call to plasma_release is required to decrement this
|
||||
* count. */
|
||||
increment_object_count(conn, object_id, object->handle.store_fd);
|
||||
}
|
||||
|
||||
void plasma_release(plasma_connection *conn, object_id object_id) {
|
||||
/* Decrement the count of the number of instances of this object that are
|
||||
* being used by this client. The corresponding increment should have happened
|
||||
* in plasma_get. */
|
||||
object_in_use_entry *object_entry;
|
||||
HASH_FIND(hh, conn->objects_in_use, &object_id, sizeof(object_id),
|
||||
object_entry);
|
||||
CHECK(object_entry != NULL);
|
||||
object_entry->count -= 1;
|
||||
CHECK(object_entry->count >= 0);
|
||||
/* Check if the client is no longer using this object. */
|
||||
if (object_entry->count == 0) {
|
||||
/* Decrement the count of the number of objects in this memory-mapped file
|
||||
* that the client is using. The corresponding increment should have
|
||||
* happened in plasma_get. */
|
||||
client_mmap_table_entry *entry;
|
||||
HASH_FIND_INT(conn->mmap_table, &object_entry->fd, entry);
|
||||
CHECK(entry != NULL);
|
||||
entry->count -= 1;
|
||||
CHECK(entry->count >= 0);
|
||||
/* If none are being used then unmap the file. */
|
||||
if (entry->count == 0) {
|
||||
munmap(entry->pointer, entry->length);
|
||||
/* Remove the corresponding entry from the hash table. */
|
||||
HASH_DELETE(hh, conn->mmap_table, entry);
|
||||
free(entry);
|
||||
}
|
||||
/* Tell the store that the client no longer needs the object. */
|
||||
plasma_request req = make_plasma_request(object_id);
|
||||
plasma_send_request(conn->store_conn, PLASMA_RELEASE, &req);
|
||||
/* Remove the entry from the hash table of objects currently in use. */
|
||||
HASH_DELETE(hh, conn->objects_in_use, object_entry);
|
||||
free(object_entry);
|
||||
}
|
||||
}
|
||||
|
||||
/* This method is used to query whether the plasma store contains an object. */
|
||||
void plasma_contains(plasma_connection *conn,
|
||||
object_id object_id,
|
||||
int *has_object) {
|
||||
plasma_request req = make_plasma_request(object_id);
|
||||
plasma_send_request(conn->store_conn, PLASMA_CONTAINS, &req);
|
||||
plasma_reply reply;
|
||||
int r = read(conn->store_conn, &reply, sizeof(plasma_reply));
|
||||
CHECKM(r != -1, "read error");
|
||||
CHECKM(r != 0, "connection disconnected");
|
||||
*has_object = reply.has_object;
|
||||
}
|
||||
|
||||
void plasma_seal(plasma_connection *conn, object_id object_id) {
|
||||
plasma_request req = make_plasma_request(object_id);
|
||||
plasma_send_request(conn->store_conn, PLASMA_SEAL, &req);
|
||||
if (conn->manager_conn >= 0) {
|
||||
plasma_send_request(conn->manager_conn, PLASMA_SEAL, &req);
|
||||
}
|
||||
}
|
||||
|
||||
void plasma_delete(plasma_connection *conn, object_id object_id) {
|
||||
plasma_request req = make_plasma_request(object_id);
|
||||
plasma_send_request(conn->store_conn, PLASMA_DELETE, &req);
|
||||
}
|
||||
|
||||
int plasma_subscribe(plasma_connection *conn) {
|
||||
int fd[2];
|
||||
/* Create a non-blocking socket pair. This will only be used to send
|
||||
* notifications from the Plasma store to the client. */
|
||||
socketpair(AF_UNIX, SOCK_STREAM, 0, fd);
|
||||
/* Make the socket non-blocking. */
|
||||
int flags = fcntl(fd[1], F_GETFL, 0);
|
||||
CHECK(fcntl(fd[1], F_SETFL, flags | O_NONBLOCK) == 0);
|
||||
/* Tell the Plasma store about the subscription. */
|
||||
plasma_request req = {};
|
||||
plasma_send_request(conn->store_conn, PLASMA_SUBSCRIBE, &req);
|
||||
/* Send the file descriptor that the Plasma store should use to push
|
||||
* notifications about sealed objects to this client. We include a one byte
|
||||
* message because otherwise it seems to hang on Linux. */
|
||||
char dummy = '\0';
|
||||
send_fd(conn->store_conn, fd[1], &dummy, 1);
|
||||
/* Return the file descriptor that the client should use to read notifications
|
||||
* about sealed objects. */
|
||||
return fd[0];
|
||||
}
|
||||
|
||||
plasma_connection *plasma_connect(const char *store_socket_name,
|
||||
const char *manager_addr,
|
||||
int manager_port) {
|
||||
CHECK(store_socket_name);
|
||||
/* 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) {
|
||||
fd = connect_ipc_sock(store_socket_name);
|
||||
if (fd >= 0) {
|
||||
connected_successfully = 1;
|
||||
break;
|
||||
}
|
||||
/* Sleep for 100 milliseconds. */
|
||||
usleep(100000);
|
||||
}
|
||||
/* If we could not connect to the Plasma store, exit. */
|
||||
if (!connected_successfully) {
|
||||
LOG_ERR("could not connect to store %s", store_socket_name);
|
||||
exit(-1);
|
||||
}
|
||||
/* Initialize the store connection struct */
|
||||
plasma_connection *result = malloc(sizeof(plasma_connection));
|
||||
result->store_conn = fd;
|
||||
if (manager_addr != NULL) {
|
||||
result->manager_conn = plasma_manager_connect(manager_addr, manager_port);
|
||||
} else {
|
||||
result->manager_conn = -1;
|
||||
}
|
||||
result->mmap_table = NULL;
|
||||
result->objects_in_use = NULL;
|
||||
return result;
|
||||
}
|
||||
|
||||
void plasma_disconnect(plasma_connection *conn) {
|
||||
close(conn->store_conn);
|
||||
if (conn->manager_conn >= 0) {
|
||||
close(conn->manager_conn);
|
||||
}
|
||||
free(conn);
|
||||
}
|
||||
|
||||
#define h_addr h_addr_list[0]
|
||||
|
||||
/* TODO(swang): Return the error to the caller. */
|
||||
int plasma_manager_connect(const char *ip_addr, int port) {
|
||||
int fd = socket(PF_INET, SOCK_STREAM, 0);
|
||||
if (fd < 0) {
|
||||
LOG_ERR("could not create socket");
|
||||
exit(-1);
|
||||
}
|
||||
|
||||
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;
|
||||
memcpy(&addr.sin_addr.s_addr, manager->h_addr, manager->h_length);
|
||||
addr.sin_port = htons(port);
|
||||
|
||||
int r = connect(fd, (struct sockaddr *) &addr, sizeof(addr));
|
||||
if (r < 0) {
|
||||
LOG_ERR(
|
||||
"could not establish connection to manager with id %s:%d (probably ran "
|
||||
"out of ports)",
|
||||
&ip_addr[0], port);
|
||||
exit(-1);
|
||||
}
|
||||
return fd;
|
||||
}
|
||||
|
||||
void plasma_transfer(plasma_connection *conn,
|
||||
const char *addr,
|
||||
int port,
|
||||
object_id object_id) {
|
||||
plasma_request req = make_plasma_request(object_id);
|
||||
req.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(conn->manager_conn, PLASMA_TRANSFER, &req);
|
||||
}
|
||||
|
||||
void plasma_fetch(plasma_connection *conn,
|
||||
int num_object_ids,
|
||||
object_id object_ids[],
|
||||
int is_fetched[]) {
|
||||
CHECK(conn->manager_conn >= 0);
|
||||
plasma_request *req =
|
||||
make_plasma_multiple_request(num_object_ids, object_ids);
|
||||
LOG_DEBUG("Requesting fetch");
|
||||
plasma_send_request(conn->manager_conn, PLASMA_FETCH, req);
|
||||
free(req);
|
||||
|
||||
plasma_reply reply;
|
||||
int nbytes, success;
|
||||
for (int received = 0; received < num_object_ids; ++received) {
|
||||
nbytes = recv(conn->manager_conn, (uint8_t *) &reply, sizeof(reply),
|
||||
MSG_WAITALL);
|
||||
if (nbytes < 0) {
|
||||
LOG_ERR("Error while waiting for manager response in fetch");
|
||||
success = 0;
|
||||
} else if (nbytes == 0) {
|
||||
success = 0;
|
||||
} else {
|
||||
CHECK(nbytes == sizeof(reply));
|
||||
success = reply.has_object;
|
||||
}
|
||||
/* Update the correct index in is_fetched. */
|
||||
int i = 0;
|
||||
for (; i < num_object_ids; i++) {
|
||||
if (memcmp(&object_ids[i], &reply.object_id, sizeof(object_id)) == 0) {
|
||||
/* Check that this isn't a duplicate response. */
|
||||
CHECK(!is_fetched[i]);
|
||||
is_fetched[i] = success;
|
||||
break;
|
||||
}
|
||||
}
|
||||
CHECKM(i != num_object_ids,
|
||||
"Received unexpected object ID from manager during fetch.");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,198 @@
|
||||
#ifndef PLASMA_CLIENT_H
|
||||
#define PLASMA_CLIENT_H
|
||||
|
||||
#include "plasma.h"
|
||||
|
||||
typedef struct plasma_connection plasma_connection;
|
||||
|
||||
/**
|
||||
* 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 fd, int type, plasma_request *req);
|
||||
|
||||
/**
|
||||
* Create a plasma request to be sent with a single object ID.
|
||||
*
|
||||
* @param object_id The object ID to include in the request.
|
||||
* @return The plasma request.
|
||||
*/
|
||||
plasma_request make_plasma_request(object_id object_id);
|
||||
|
||||
/**
|
||||
* Create a plasma request to be sent with multiple object ID. Caller must free
|
||||
* the returned plasma request pointer.
|
||||
*
|
||||
* @param num_object_ids The number of object IDs to include in the request.
|
||||
* @param object_ids The array of object IDs to include in the request. It must
|
||||
* have length at least equal to num_object_ids.
|
||||
* @return A pointer to the newly created plasma request.
|
||||
*/
|
||||
plasma_request *make_plasma_multiple_request(int num_object_ids,
|
||||
object_id object_ids[]);
|
||||
|
||||
/**
|
||||
* Connect to the local plasma store and plasma manager. Return
|
||||
* the resulting connection.
|
||||
*
|
||||
* @param socket_name The name of the UNIX domain socket to use to connect to
|
||||
* the Plasma Store.
|
||||
* @param manager_addr The IP address of the plasma manager to connect to. If
|
||||
* this is NULL, then this function will not connect to a manager.
|
||||
* @param manager_port The port of the plasma manager to connect to. If
|
||||
* manager_addr is NULL, then this argument is unused.
|
||||
* @return The object containing the connection state.
|
||||
*/
|
||||
plasma_connection *plasma_connect(const char *store_socket_name,
|
||||
const char *manager_addr,
|
||||
int manager_port);
|
||||
|
||||
/**
|
||||
* Disconnect from the local plasma instance, including the local store and
|
||||
* manager.
|
||||
*
|
||||
* @param conn The connection to the local plasma store and plasma manager.
|
||||
* @return Void.
|
||||
*/
|
||||
void plasma_disconnect(plasma_connection *conn);
|
||||
|
||||
/**
|
||||
* Connect to a possibly remote Plasma Manager.
|
||||
*
|
||||
* @param addr The IP address of the Plasma Manager to connect to.
|
||||
* @param port The port of the Plasma Manager to connect to.
|
||||
* @return The file descriptor to use to send messages to the Plasma Manager.
|
||||
*/
|
||||
int plasma_manager_connect(const char *addr, int port);
|
||||
|
||||
/**
|
||||
* Create an object in the Plasma Store. Any metadata for this object must be
|
||||
* be passed in when the object is created.
|
||||
*
|
||||
* @param conn The object containing the connection state.
|
||||
* @param object_id The ID to use for the newly created object.
|
||||
* @param size The size in bytes of the space to be allocated for this object's
|
||||
data (this does not include space used for metadata).
|
||||
* @param metadata The object's metadata. If there is no metadata, this pointer
|
||||
should be NULL.
|
||||
* @param metadata_size The size in bytes of the metadata. If there is no
|
||||
metadata, this should be 0.
|
||||
* @param data The address of the newly created object will be written here.
|
||||
* @return Void.
|
||||
*/
|
||||
void plasma_create(plasma_connection *conn,
|
||||
object_id object_id,
|
||||
int64_t size,
|
||||
uint8_t *metadata,
|
||||
int64_t metadata_size,
|
||||
uint8_t **data);
|
||||
|
||||
/**
|
||||
* Get an object from the Plasma Store. This function will block until the
|
||||
* object has been created and sealed in the Plasma Store.
|
||||
*
|
||||
* @param conn The object containing the connection state.
|
||||
* @param object_id The ID of the object to get.
|
||||
* @param size The size in bytes of the retrieved object will be written at this
|
||||
address.
|
||||
* @param data The address of the object will be written at this address.
|
||||
* @param metadata_size The size in bytes of the object's metadata will be
|
||||
* written at this address.
|
||||
* @param metadata The address of the object's metadata will be written at this
|
||||
* address.
|
||||
* @return Void.
|
||||
*/
|
||||
void plasma_get(plasma_connection *conn,
|
||||
object_id object_id,
|
||||
int64_t *size,
|
||||
uint8_t **data,
|
||||
int64_t *metadata_size,
|
||||
uint8_t **metadata);
|
||||
|
||||
/**
|
||||
* Tell Plasma that the client no longer needs the object. This should be called
|
||||
* after plasma_get when the client is done with the object. After this call,
|
||||
* the address returned by plasma_get is no longer valid. This should be called
|
||||
* once for each call to plasma_get (with the same object ID).
|
||||
*
|
||||
* @param conn The object containing the connection state.
|
||||
* @param object_id The ID of the object that is no longer needed.
|
||||
* @return Void.
|
||||
*/
|
||||
void plasma_release(plasma_connection *conn, object_id object_id);
|
||||
|
||||
/**
|
||||
* Check if the object store contains a particular object and the object has
|
||||
* been sealed. The result will be stored in has_object.
|
||||
*
|
||||
* @todo: We may want to indicate if the object has been created but not sealed.
|
||||
*
|
||||
* @param conn The object containing the connection state.
|
||||
* @param object_id The ID of the object whose presence we are checking.
|
||||
* @param has_object The function will write 1 at this address if the object is
|
||||
* present and 0 if it is not present.
|
||||
* @return Void.
|
||||
*/
|
||||
void plasma_contains(plasma_connection *conn,
|
||||
object_id object_id,
|
||||
int *has_object);
|
||||
|
||||
/**
|
||||
* Seal an object in the object store. The object will be immutable after this
|
||||
* call.
|
||||
*
|
||||
* @param conn The object containing the connection state.
|
||||
* @param object_id The ID of the object to seal.
|
||||
* @return Void.
|
||||
*/
|
||||
void plasma_seal(plasma_connection *conn, object_id object_id);
|
||||
|
||||
/**
|
||||
* Delete an object from the object store. This currently assumes that the
|
||||
* object is present and has been sealed.
|
||||
*
|
||||
* @todo We may want to allow the deletion of objects that are not present or
|
||||
* haven't been sealed.
|
||||
*
|
||||
* @param conn The object containing the connection state.
|
||||
* @param object_id The ID of the object to delete.
|
||||
* @return Void.
|
||||
*/
|
||||
void plasma_delete(plasma_connection *conn, object_id object_id);
|
||||
|
||||
/**
|
||||
* Fetch objects from remote plasma stores that have the
|
||||
* objects stored.
|
||||
*
|
||||
* @param manager A file descriptor for the socket connection
|
||||
* to the local manager.
|
||||
* @param object_id_count The number of object IDs requested.
|
||||
* @param object_ids[] The vector of object IDs requested. Length must be at
|
||||
* least num_object_ids.
|
||||
* @param is_fetched[] The vector in which to return the success
|
||||
* of each object's fetch operation, in the same order as
|
||||
* object_ids. Length must be at least num_object_ids.
|
||||
* @return Void.
|
||||
*/
|
||||
void plasma_fetch(plasma_connection *conn,
|
||||
int num_object_ids,
|
||||
object_id object_ids[],
|
||||
int is_fetched[]);
|
||||
|
||||
/**
|
||||
* Subscribe to notifications when objects are sealed in the object store.
|
||||
* Whenever an object is sealed, a message will be written to the client socket
|
||||
* that is returned by this method.
|
||||
*
|
||||
* @param conn The object containing the connection state.
|
||||
* @return The file descriptor that the client should use to read notifications
|
||||
from the object store about sealed objects.
|
||||
*/
|
||||
int plasma_subscribe(plasma_connection *conn);
|
||||
|
||||
#endif
|
||||
@@ -0,0 +1,804 @@
|
||||
/* PLASMA MANAGER: Local to a node, connects to other managers to send and
|
||||
* receive objects from them
|
||||
*
|
||||
* The storage manager listens on its main listening port, and if a request for
|
||||
* transfering an object to another object store comes in, it ships the data
|
||||
* using a new connection to the target object manager. */
|
||||
|
||||
#include <fcntl.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <unistd.h>
|
||||
#include <signal.h>
|
||||
#include <stdlib.h>
|
||||
#include <sys/mman.h>
|
||||
#include <sys/types.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/un.h>
|
||||
#include <strings.h>
|
||||
#include <poll.h>
|
||||
#include <assert.h>
|
||||
#include <netinet/in.h>
|
||||
#include <netdb.h>
|
||||
|
||||
#include "uthash.h"
|
||||
#include "utlist.h"
|
||||
#include "utarray.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"
|
||||
#include "state/db.h"
|
||||
#include "state/object_table.h"
|
||||
|
||||
#define NUM_RETRIES 5
|
||||
|
||||
/* Timeouts are in milliseconds. */
|
||||
#ifndef RAY_TIMEOUT
|
||||
#define MANAGER_TIMEOUT 1000
|
||||
#else
|
||||
#define MANAGER_TIMEOUT RAY_TIMEOUT
|
||||
#endif
|
||||
|
||||
typedef struct client_object_connection client_object_connection;
|
||||
|
||||
typedef struct {
|
||||
/** Event loop. */
|
||||
event_loop *loop;
|
||||
/** Connection to the local plasma store for reading or writing data. */
|
||||
plasma_connection *plasma_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;
|
||||
db_handle *db;
|
||||
/** Our address. */
|
||||
uint8_t addr[4];
|
||||
/** Our port. */
|
||||
int port;
|
||||
/** Hash table of outstanding fetch requests. The key is
|
||||
* object id, value is a list of connections to the clients
|
||||
* who are blocking on a fetch of this object. */
|
||||
client_object_connection *fetch_connections;
|
||||
} plasma_manager_state;
|
||||
|
||||
plasma_manager_state *g_manager_state = NULL;
|
||||
|
||||
typedef struct plasma_request_buffer plasma_request_buffer;
|
||||
|
||||
/* Buffer for requests between plasma managers. */
|
||||
struct plasma_request_buffer {
|
||||
int type;
|
||||
object_id object_id;
|
||||
uint8_t *data;
|
||||
int64_t data_size;
|
||||
uint8_t *metadata;
|
||||
int64_t metadata_size;
|
||||
/* Pointer to the next buffer that we will write to this plasma manager. This
|
||||
* field is only used if we're pushing requests to another plasma manager,
|
||||
* not if we are receiving data. */
|
||||
plasma_request_buffer *next;
|
||||
};
|
||||
|
||||
/* The context for fetch and wait requests. These are per client, per object. */
|
||||
struct client_object_connection {
|
||||
/** The ID of the object we are fetching or waiting for. */
|
||||
object_id object_id;
|
||||
/** The client connection context, shared between other
|
||||
* client_object_connections for the same client. */
|
||||
client_connection *client_conn;
|
||||
/** The ID for the timer that will time out the current request to the state
|
||||
* database or another plasma manager. */
|
||||
int64_t timer;
|
||||
/** How many retries we have left for the request. Decremented on every
|
||||
* timeout. */
|
||||
int num_retries;
|
||||
/** Handle for a linked list. */
|
||||
client_object_connection *next;
|
||||
/** Pointer to the array containing the manager locations of
|
||||
* this object. */
|
||||
char **manager_vector;
|
||||
/** The number of manager locations in the array manager_vector. */
|
||||
int manager_count;
|
||||
/** Handle for the uthash table in the client connection
|
||||
* context that keeps track of active object connection
|
||||
* contexts. */
|
||||
UT_hash_handle active_hh;
|
||||
/** Handle for the uthash table in the manager state that
|
||||
* keeps track of outstanding fetch requests. */
|
||||
UT_hash_handle fetch_hh;
|
||||
};
|
||||
|
||||
/* 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. */
|
||||
/* TODO(swang): Split into two queues, data transfers and data requests. */
|
||||
plasma_request_buffer *transfer_queue;
|
||||
/** File descriptor for the socket connected to the other
|
||||
* plasma manager. */
|
||||
int fd;
|
||||
/** The objects that we are waiting for and their callback
|
||||
* contexts, for either a fetch or a wait operation. */
|
||||
client_object_connection *active_objects;
|
||||
/** The number of objects that we have left to return for
|
||||
* this fetch or wait operation. */
|
||||
int num_return_objects;
|
||||
/** Fields specific to 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;
|
||||
};
|
||||
|
||||
void free_client_object_connection(client_object_connection *object_conn) {
|
||||
for (int i = 0; i < object_conn->manager_count; ++i) {
|
||||
free(object_conn->manager_vector[i]);
|
||||
}
|
||||
free(object_conn->manager_vector);
|
||||
free(object_conn);
|
||||
}
|
||||
|
||||
int send_client_reply(client_connection *conn, plasma_reply *reply) {
|
||||
conn->num_return_objects--;
|
||||
CHECK(conn->num_return_objects >= 0);
|
||||
/* TODO(swang): Handle errors in write. */
|
||||
int n = write(conn->fd, (uint8_t *) reply, sizeof(plasma_reply));
|
||||
return (n != sizeof(plasma_reply));
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the context for the given object ID for the given client
|
||||
* connection, if there is one active.
|
||||
*
|
||||
* @param client_conn The client connection context.
|
||||
* @param object_id The object ID whose context we want.
|
||||
* @return A pointer to the active object context, or NULL if
|
||||
* there isn't one.
|
||||
*/
|
||||
client_object_connection *get_object_connection(client_connection *client_conn,
|
||||
object_id object_id) {
|
||||
client_object_connection *object_conn;
|
||||
HASH_FIND(active_hh, client_conn->active_objects, &object_id,
|
||||
sizeof(object_id), object_conn);
|
||||
return object_conn;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new context for the given object ID with the given
|
||||
* client connection and register it with the manager's
|
||||
* outstanding fetch or wait requests and the client
|
||||
* connection's active object contexts.
|
||||
*
|
||||
* @param client_conn The client connection context.
|
||||
* @param object_id The object ID whose context we want to
|
||||
* create.
|
||||
* @return A pointer to the newly created object context.
|
||||
*/
|
||||
client_object_connection *add_object_connection(client_connection *client_conn,
|
||||
object_id object_id) {
|
||||
/* TODO(swang): Support registration of wait operations. */
|
||||
/* Create a new context for this client connection and object. */
|
||||
client_object_connection *object_conn =
|
||||
malloc(sizeof(client_object_connection));
|
||||
if (!object_conn) {
|
||||
return NULL;
|
||||
}
|
||||
object_conn->object_id = object_id;
|
||||
object_conn->client_conn = client_conn;
|
||||
object_conn->manager_count = 0;
|
||||
object_conn->manager_vector = NULL;
|
||||
/* Register the object context with the client context. */
|
||||
HASH_ADD(active_hh, client_conn->active_objects, object_id, sizeof(object_id),
|
||||
object_conn);
|
||||
/* Register the object context with the manager state. */
|
||||
client_object_connection *fetch_connections;
|
||||
HASH_FIND(fetch_hh, client_conn->manager_state->fetch_connections, &object_id,
|
||||
sizeof(object_id), fetch_connections);
|
||||
LOG_DEBUG("Registering fd %d for fetch.", client_conn->fd);
|
||||
if (!fetch_connections) {
|
||||
fetch_connections = NULL;
|
||||
LL_APPEND(fetch_connections, object_conn);
|
||||
HASH_ADD(fetch_hh, client_conn->manager_state->fetch_connections, object_id,
|
||||
sizeof(object_id), fetch_connections);
|
||||
} else {
|
||||
LL_APPEND(fetch_connections, object_conn);
|
||||
}
|
||||
return object_conn;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up and free an active object context. Deregister it from the
|
||||
* associated client connection and from the manager state.
|
||||
*
|
||||
* @param client_conn The client connection context.
|
||||
* @param object_id The object ID whose context we want to delete.
|
||||
*/
|
||||
void remove_object_connection(client_connection *client_conn,
|
||||
client_object_connection *object_conn) {
|
||||
/* Deregister the object context with the client context. */
|
||||
HASH_DELETE(active_hh, client_conn->active_objects, object_conn);
|
||||
/* Deregister the object context with the manager state. */
|
||||
client_object_connection *object_conns;
|
||||
HASH_FIND(fetch_hh, client_conn->manager_state->fetch_connections,
|
||||
&(object_conn->object_id), sizeof(object_conn->object_id),
|
||||
object_conns);
|
||||
CHECK(object_conns);
|
||||
int len;
|
||||
client_object_connection *tmp;
|
||||
LL_COUNT(object_conns, tmp, len);
|
||||
if (len == 1) {
|
||||
HASH_DELETE(fetch_hh, client_conn->manager_state->fetch_connections,
|
||||
object_conns);
|
||||
}
|
||||
LL_DELETE(object_conns, object_conn);
|
||||
/* Free the object. */
|
||||
free_client_object_connection(object_conn);
|
||||
}
|
||||
|
||||
/* Helper function to parse a string of the form <IP address>:<port> into the
|
||||
* given ip_addr and port pointers. The ip_addr buffer must already be
|
||||
* allocated. */
|
||||
/* TODO(swang): Move this function to Ray common. */
|
||||
void parse_ip_addr_port(const char *ip_addr_port, char *ip_addr, int *port) {
|
||||
char port_str[6];
|
||||
int parsed = sscanf(ip_addr_port, "%15[0-9.]:%5[0-9]", ip_addr, port_str);
|
||||
CHECK(parsed == 2);
|
||||
*port = atoi(port_str);
|
||||
}
|
||||
|
||||
plasma_manager_state *init_plasma_manager_state(const char *store_socket_name,
|
||||
const char *manager_addr,
|
||||
int manager_port,
|
||||
const char *db_addr,
|
||||
int db_port) {
|
||||
plasma_manager_state *state = malloc(sizeof(plasma_manager_state));
|
||||
state->loop = event_loop_create();
|
||||
state->plasma_conn = plasma_connect(store_socket_name, NULL, 0);
|
||||
state->manager_connections = NULL;
|
||||
state->fetch_connections = NULL;
|
||||
if (db_addr) {
|
||||
state->db = db_connect(db_addr, db_port, "plasma_manager", manager_addr,
|
||||
manager_port);
|
||||
db_attach(state->db, state->loop);
|
||||
LOG_DEBUG("Connected to db at %s:%d, assigned client ID %d", db_addr,
|
||||
db_port, get_client_id(state->db));
|
||||
} else {
|
||||
state->db = NULL;
|
||||
LOG_DEBUG("No db connection specified");
|
||||
}
|
||||
sscanf(manager_addr, "%hhu.%hhu.%hhu.%hhu", &state->addr[0], &state->addr[1],
|
||||
&state->addr[2], &state->addr[3]);
|
||||
state->port = manager_port;
|
||||
return state;
|
||||
}
|
||||
|
||||
/* Handle a command request that came in through a socket (transfering data,
|
||||
* or accepting incoming data). */
|
||||
void process_message(event_loop *loop,
|
||||
int client_sock,
|
||||
void *context,
|
||||
int events);
|
||||
|
||||
void write_object_chunk(client_connection *conn, plasma_request_buffer *buf) {
|
||||
ssize_t r, s;
|
||||
/* 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);
|
||||
}
|
||||
} else {
|
||||
conn->cursor += r;
|
||||
}
|
||||
if (r == 0) {
|
||||
/* If we've finished writing this buffer, reset the cursor to zero. */
|
||||
LOG_DEBUG("writing on channel %d finished", conn->fd);
|
||||
conn->cursor = 0;
|
||||
/* We are done sending the object, so release it. The corresponding call to
|
||||
* plasma_get occurred in process_transfer_request. */
|
||||
plasma_release(conn->manager_state->plasma_conn, buf->object_id);
|
||||
}
|
||||
}
|
||||
|
||||
void send_queued_request(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;
|
||||
}
|
||||
|
||||
plasma_request_buffer *buf = conn->transfer_queue;
|
||||
plasma_request manager_req = make_plasma_request(buf->object_id);
|
||||
switch (buf->type) {
|
||||
case PLASMA_TRANSFER:
|
||||
LOG_DEBUG("Requesting transfer on DB client %d",
|
||||
get_client_id(conn->manager_state->db));
|
||||
memcpy(manager_req.addr, conn->manager_state->addr,
|
||||
sizeof(manager_req.addr));
|
||||
manager_req.port = conn->manager_state->port;
|
||||
plasma_send_request(conn->fd, buf->type, &manager_req);
|
||||
break;
|
||||
case PLASMA_DATA:
|
||||
LOG_DEBUG("Transferring object to manager");
|
||||
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. */
|
||||
manager_req.data_size = buf->data_size;
|
||||
manager_req.metadata_size = buf->metadata_size;
|
||||
plasma_send_request(conn->fd, PLASMA_DATA, &manager_req);
|
||||
}
|
||||
write_object_chunk(conn, buf);
|
||||
break;
|
||||
default:
|
||||
LOG_ERR("Buffered request has unknown type.");
|
||||
}
|
||||
|
||||
/* We are done sending this request. */
|
||||
if (conn->cursor == 0) {
|
||||
LL_DELETE(conn->transfer_queue, buf);
|
||||
free(buf);
|
||||
}
|
||||
}
|
||||
|
||||
void process_data_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_request_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) {
|
||||
/* If we haven't finished reading all the data for this object yet, we're
|
||||
* done for now. */
|
||||
return;
|
||||
}
|
||||
|
||||
/* Seal the object and release it. The release corresponds to the call to
|
||||
* plasma_create that occurred in process_data_request. */
|
||||
LOG_DEBUG("reading on channel %d finished", data_sock);
|
||||
plasma_seal(conn->manager_state->plasma_conn, buf->object_id);
|
||||
plasma_release(conn->manager_state->plasma_conn, buf->object_id);
|
||||
/* Notify any clients who were waiting on a fetch to this object. */
|
||||
client_object_connection *object_conn, *next;
|
||||
client_connection *client_conn;
|
||||
HASH_FIND(fetch_hh, conn->manager_state->fetch_connections, &(buf->object_id),
|
||||
sizeof(buf->object_id), object_conn);
|
||||
plasma_reply reply = {.object_id = buf->object_id, .has_object = 1};
|
||||
while (object_conn) {
|
||||
next = object_conn->next;
|
||||
client_conn = object_conn->client_conn;
|
||||
send_client_reply(client_conn, &reply);
|
||||
event_loop_remove_timer(client_conn->manager_state->loop,
|
||||
object_conn->timer);
|
||||
remove_object_connection(client_conn, object_conn);
|
||||
object_conn = next;
|
||||
}
|
||||
/* Remove the request buffer used for reading this object's data. */
|
||||
LL_DELETE(conn->transfer_queue, buf);
|
||||
free(buf);
|
||||
/* Switch to listening for requests from this socket, instead of reading
|
||||
* object data. */
|
||||
event_loop_remove_file(loop, data_sock);
|
||||
event_loop_add_file(loop, data_sock, EVENT_LOOP_READ, process_message, conn);
|
||||
}
|
||||
|
||||
client_connection *get_manager_connection(plasma_manager_state *state,
|
||||
const char *ip_addr,
|
||||
int port) {
|
||||
/* TODO(swang): Should probably check whether ip_addr and port belong to us.
|
||||
*/
|
||||
UT_string *ip_addr_port;
|
||||
utstring_new(ip_addr_port);
|
||||
utstring_printf(ip_addr_port, "%s:%d", ip_addr, port);
|
||||
client_connection *manager_conn;
|
||||
HASH_FIND_STR(state->manager_connections, utstring_body(ip_addr_port),
|
||||
manager_conn);
|
||||
LOG_DEBUG("Getting manager connection to %s on DB client %d",
|
||||
utstring_body(ip_addr_port), get_client_id(state->db));
|
||||
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(ip_addr, port);
|
||||
manager_conn->manager_state = 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);
|
||||
return manager_conn;
|
||||
}
|
||||
|
||||
void process_transfer_request(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;
|
||||
/* TODO(swang): A non-blocking plasma_get, or else we could block here
|
||||
* forever if we don't end up sealing this object. */
|
||||
/* The corresponding call to plasma_release will happen in
|
||||
* write_object_chunk. */
|
||||
plasma_get(conn->manager_state->plasma_conn, object_id, &data_size, &data,
|
||||
&metadata_size, &metadata);
|
||||
assert(metadata == data + data_size);
|
||||
plasma_request_buffer *buf = malloc(sizeof(plasma_request_buffer));
|
||||
buf->type = PLASMA_DATA;
|
||||
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;
|
||||
|
||||
UT_string *ip_addr;
|
||||
utstring_new(ip_addr);
|
||||
utstring_printf(ip_addr, "%d.%d.%d.%d", addr[0], addr[1], addr[2], addr[3]);
|
||||
client_connection *manager_conn =
|
||||
get_manager_connection(conn->manager_state, utstring_body(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,
|
||||
send_queued_request, manager_conn);
|
||||
}
|
||||
/* Add this transfer request to this connection's transfer queue. */
|
||||
LL_APPEND(manager_conn->transfer_queue, buf);
|
||||
}
|
||||
|
||||
void process_data_request(event_loop *loop,
|
||||
int client_sock,
|
||||
object_id object_id,
|
||||
int64_t data_size,
|
||||
int64_t metadata_size,
|
||||
client_connection *conn) {
|
||||
plasma_request_buffer *buf = malloc(sizeof(plasma_request_buffer));
|
||||
buf->object_id = object_id;
|
||||
buf->data_size = data_size;
|
||||
buf->metadata_size = metadata_size;
|
||||
|
||||
/* The corresponding call to plasma_release should happen in
|
||||
* process_data_chunk. */
|
||||
plasma_create(conn->manager_state->plasma_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, process_data_chunk,
|
||||
conn);
|
||||
}
|
||||
|
||||
/**
|
||||
* Request a transfer for the given object ID from the next manager believed to
|
||||
* have a copy. Adds the request for this object ID to the queue of outgoing
|
||||
* requests to the manager we want to try.
|
||||
*
|
||||
* @param client_conn The context for the connection to this client.
|
||||
* @param object_id The object ID we want to request a transfer of.
|
||||
* @returns Void.
|
||||
*/
|
||||
void request_transfer_from(client_connection *client_conn,
|
||||
object_id object_id) {
|
||||
client_object_connection *object_conn =
|
||||
get_object_connection(client_conn, object_id);
|
||||
CHECK(object_conn);
|
||||
CHECK(object_conn->manager_count > 0);
|
||||
char addr[16];
|
||||
int port;
|
||||
int i = object_conn->num_retries % object_conn->manager_count;
|
||||
parse_ip_addr_port(object_conn->manager_vector[i], addr, &port);
|
||||
|
||||
client_connection *manager_conn =
|
||||
get_manager_connection(client_conn->manager_state, addr, port);
|
||||
plasma_request_buffer *transfer_request =
|
||||
malloc(sizeof(plasma_request_buffer));
|
||||
transfer_request->type = PLASMA_TRANSFER;
|
||||
transfer_request->object_id = object_conn->object_id;
|
||||
|
||||
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. */
|
||||
event_loop_add_file(client_conn->manager_state->loop, manager_conn->fd,
|
||||
EVENT_LOOP_WRITE, send_queued_request, manager_conn);
|
||||
}
|
||||
/* Add this transfer request to this connection's transfer queue. */
|
||||
LL_APPEND(manager_conn->transfer_queue, transfer_request);
|
||||
}
|
||||
|
||||
int manager_timeout_handler(event_loop *loop, timer_id id, void *context) {
|
||||
client_object_connection *object_conn = context;
|
||||
client_connection *client_conn = object_conn->client_conn;
|
||||
LOG_DEBUG("Timer went off, %d tries left", object_conn->num_retries);
|
||||
if (object_conn->num_retries > 0) {
|
||||
request_transfer_from(client_conn, object_conn->object_id);
|
||||
object_conn->num_retries--;
|
||||
return MANAGER_TIMEOUT;
|
||||
}
|
||||
plasma_reply reply = {.object_id = object_conn->object_id, .has_object = 0};
|
||||
send_client_reply(client_conn, &reply);
|
||||
remove_object_connection(client_conn, object_conn);
|
||||
return AE_NOMORE;
|
||||
}
|
||||
|
||||
/**
|
||||
* Given an object ID and the managers it can be found on, start requesting a
|
||||
* transfer from the managers.
|
||||
*
|
||||
* @param object_id The object ID we want to request a transfer of.
|
||||
* @param manager_count The number of managers the object can be found on.
|
||||
* @param manager_vector A vector of the IP addresses of the managers that the
|
||||
* object can be found on.
|
||||
* @param context The context for the connection to this client.
|
||||
*
|
||||
* Initializes a new context for this client and object. Managers are tried in
|
||||
* order until we receive the data or we timeout and run out of retries.
|
||||
*/
|
||||
void request_transfer(object_id object_id,
|
||||
int manager_count,
|
||||
const char *manager_vector[],
|
||||
void *context) {
|
||||
client_connection *client_conn = (client_connection *) context;
|
||||
client_object_connection *object_conn =
|
||||
get_object_connection(client_conn, object_id);
|
||||
CHECK(object_conn);
|
||||
LOG_DEBUG("Object is on %d managers", manager_count);
|
||||
if (manager_count == 0) {
|
||||
/* TODO(swang): Instead of immediately counting this as a failure, maybe
|
||||
* register a Redis callback for changes to this object table entry. */
|
||||
free(manager_vector);
|
||||
plasma_reply reply = {.object_id = object_conn->object_id, .has_object = 0};
|
||||
send_client_reply(client_conn, &reply);
|
||||
remove_object_connection(client_conn, object_conn);
|
||||
return;
|
||||
}
|
||||
/* Pick a different manager to request a transfer from on every attempt. */
|
||||
object_conn->manager_count = manager_count;
|
||||
object_conn->manager_vector = malloc(manager_count * sizeof(char *));
|
||||
memset(object_conn->manager_vector, 0, manager_count * sizeof(char *));
|
||||
for (int i = 0; i < manager_count; ++i) {
|
||||
int len = strlen(manager_vector[i]);
|
||||
object_conn->manager_vector[i] = malloc(len + 1);
|
||||
strncpy(object_conn->manager_vector[i], manager_vector[i], len);
|
||||
object_conn->manager_vector[i][len] = '\0';
|
||||
}
|
||||
free(manager_vector);
|
||||
/* Wait for the object data for the default number of retries, which timeout
|
||||
* after a default interval. */
|
||||
object_conn->num_retries = NUM_RETRIES;
|
||||
object_conn->timer =
|
||||
event_loop_add_timer(client_conn->manager_state->loop, MANAGER_TIMEOUT,
|
||||
manager_timeout_handler, object_conn);
|
||||
request_transfer_from(client_conn, object_id);
|
||||
}
|
||||
|
||||
void process_fetch_request(client_connection *client_conn,
|
||||
object_id object_id) {
|
||||
plasma_reply reply = {.object_id = object_id};
|
||||
if (client_conn->manager_state->db == NULL) {
|
||||
reply.has_object = 0;
|
||||
send_client_reply(client_conn, &reply);
|
||||
return;
|
||||
}
|
||||
/* Return success immediately if we already have this object. */
|
||||
int is_local = 0;
|
||||
plasma_contains(client_conn->manager_state->plasma_conn, object_id,
|
||||
&is_local);
|
||||
if (is_local) {
|
||||
reply.has_object = 1;
|
||||
send_client_reply(client_conn, &reply);
|
||||
return;
|
||||
}
|
||||
/* Register the new context with the current client connection. */
|
||||
client_object_connection *object_conn =
|
||||
add_object_connection(client_conn, object_id);
|
||||
if (!object_conn) {
|
||||
LOG_DEBUG("Unable to allocate memory for object context.");
|
||||
reply.has_object = 0;
|
||||
send_client_reply(client_conn, &reply);
|
||||
}
|
||||
/* Request a transfer from a plasma manager that has this object. */
|
||||
object_table_lookup(client_conn->manager_state->db, object_id,
|
||||
request_transfer, client_conn);
|
||||
}
|
||||
|
||||
void process_fetch_requests(client_connection *client_conn,
|
||||
int num_object_ids,
|
||||
object_id object_ids[]) {
|
||||
for (int i = 0; i < num_object_ids; ++i) {
|
||||
client_conn->num_return_objects++;
|
||||
process_fetch_request(client_conn, object_ids[i]);
|
||||
}
|
||||
}
|
||||
|
||||
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:
|
||||
process_transfer_request(loop, req->object_ids[0], req->addr, req->port,
|
||||
conn);
|
||||
break;
|
||||
case PLASMA_DATA:
|
||||
LOG_DEBUG("Starting to stream data");
|
||||
process_data_request(loop, client_sock, req->object_ids[0], req->data_size,
|
||||
req->metadata_size, conn);
|
||||
break;
|
||||
case PLASMA_FETCH:
|
||||
LOG_DEBUG("Processing fetch");
|
||||
process_fetch_requests(conn, req->num_object_ids, req->object_ids);
|
||||
break;
|
||||
case PLASMA_SEAL:
|
||||
LOG_DEBUG("Publishing to object table from DB client %d.",
|
||||
get_client_id(conn->manager_state->db));
|
||||
object_table_add(conn->manager_state->db, req->object_ids[0]);
|
||||
break;
|
||||
case DISCONNECT_CLIENT: {
|
||||
LOG_INFO("Disconnecting client on fd %d", client_sock);
|
||||
/* TODO(swang): Check if this connection was to a plasma manager. If so,
|
||||
* delete it. */
|
||||
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;
|
||||
conn->fd = new_socket;
|
||||
conn->active_objects = NULL;
|
||||
conn->num_return_objects = 0;
|
||||
event_loop_add_file(loop, new_socket, EVENT_LOOP_READ, process_message, conn);
|
||||
LOG_DEBUG("New plasma manager connection with fd %d", new_socket);
|
||||
}
|
||||
|
||||
void start_server(const char *store_socket_name,
|
||||
const char *master_addr,
|
||||
int port,
|
||||
const char *db_addr,
|
||||
int db_port) {
|
||||
int sock = bind_inet_sock(port);
|
||||
CHECKM(sock >= 0, "Unable to bind to manager port");
|
||||
|
||||
g_manager_state = init_plasma_manager_state(store_socket_name, master_addr,
|
||||
port, db_addr, db_port);
|
||||
CHECK(g_manager_state);
|
||||
LOG_DEBUG("Started server connected to store %s, listening on port %d",
|
||||
store_socket_name, port);
|
||||
event_loop_add_file(g_manager_state->loop, sock, EVENT_LOOP_READ,
|
||||
new_client_connection, g_manager_state);
|
||||
event_loop_run(g_manager_state->loop);
|
||||
}
|
||||
|
||||
/* Report "success" to valgrind. */
|
||||
void signal_handler(int signal) {
|
||||
if (signal == SIGTERM) {
|
||||
if (g_manager_state) {
|
||||
db_disconnect(g_manager_state->db);
|
||||
}
|
||||
exit(0);
|
||||
}
|
||||
}
|
||||
|
||||
int main(int argc, char *argv[]) {
|
||||
signal(SIGTERM, signal_handler);
|
||||
/* Socket name of the plasma store this manager is connected to. */
|
||||
char *store_socket_name = NULL;
|
||||
/* IP address of this node. */
|
||||
char *master_addr = NULL;
|
||||
/* Port number the manager should use. */
|
||||
int port;
|
||||
/* IP address and port of state database. */
|
||||
char *db_host = NULL;
|
||||
int c;
|
||||
while ((c = getopt(argc, argv, "s:m:p:d:")) != -1) {
|
||||
switch (c) {
|
||||
case 's':
|
||||
store_socket_name = optarg;
|
||||
break;
|
||||
case 'm':
|
||||
master_addr = optarg;
|
||||
break;
|
||||
case 'p':
|
||||
port = atoi(optarg);
|
||||
break;
|
||||
case 'd':
|
||||
db_host = 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);
|
||||
}
|
||||
char db_addr[16];
|
||||
int db_port;
|
||||
if (db_host) {
|
||||
parse_ip_addr_port(db_host, db_addr, &db_port);
|
||||
start_server(store_socket_name, master_addr, port, db_addr, db_port);
|
||||
} else {
|
||||
start_server(store_socket_name, master_addr, port, NULL, 0);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
#ifndef PLASMA_MANAGER_H
|
||||
#define PLASMA_MANAGER_H
|
||||
|
||||
#include <poll.h>
|
||||
#include "utarray.h"
|
||||
|
||||
typedef struct client_connection client_connection;
|
||||
|
||||
/**
|
||||
* Process a request from another object store manager to transfer an object.
|
||||
*
|
||||
* @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.
|
||||
* @return Void.
|
||||
*
|
||||
* This establishes a connection to the remote manager if one doesn't already
|
||||
* exist, and queues up the request to transfer the data to the other object
|
||||
* manager.
|
||||
*/
|
||||
void process_transfer(event_loop *loop,
|
||||
object_id object_id,
|
||||
uint8_t addr[4],
|
||||
int port,
|
||||
client_connection *conn);
|
||||
|
||||
/**
|
||||
* Process a request from another object store manager to receive data.
|
||||
*
|
||||
* @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.
|
||||
* @return Void.
|
||||
*
|
||||
* Initializes the object we are going to write to in the local plasma store
|
||||
* and then switches the data socket to read the raw object bytes instead of
|
||||
* plasma requests.
|
||||
*/
|
||||
void process_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
|
||||
* connected to the given socket. Once all data for this object has been read,
|
||||
* the socket switches to listening for the next plasma 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.
|
||||
* @return Void.
|
||||
*/
|
||||
void process_data_chunk(event_loop *loop,
|
||||
int data_sock,
|
||||
void *context,
|
||||
int events);
|
||||
|
||||
/**
|
||||
* Process a fetch request. The fetch request tries:
|
||||
* 1) If there is no connection to the database, return faliure to the client.
|
||||
* 2) If the object is available locally, return success to the client.
|
||||
* 3) Query the database for plasma managers that the object might be on.
|
||||
* 4) Request a transfer from each of the managers that the object might be on
|
||||
* until we receive the data, or until we timeout.
|
||||
* 5) Returns success or failure to the client depending on whether we received
|
||||
* the data or not.
|
||||
*
|
||||
* @param client_conn The connection context for the client that made the
|
||||
* request.
|
||||
* @param object_id The object ID requested.
|
||||
* @return Void.
|
||||
*/
|
||||
void process_fetch_request(client_connection *client_conn, object_id object_id);
|
||||
|
||||
/**
|
||||
* Process a fetch request for multiple objects. The success of each object
|
||||
* will be written back individually to the socket connected to the client that
|
||||
* made the request in a plasma_reply. See documentation for
|
||||
* process_fetch_request for the sequence of operations per object.
|
||||
*
|
||||
* @param client_conn The connection context for the client that made the
|
||||
* request.
|
||||
* @param object_id_count The number of object IDs requested.
|
||||
* @param object_ids[] The vector of object IDs requested.
|
||||
* @return Void.
|
||||
*/
|
||||
void process_fetch_requests(client_connection *client_conn,
|
||||
int object_id_count,
|
||||
object_id object_ids[]);
|
||||
|
||||
/**
|
||||
* Send the next request queued for the other plasma manager connected to the
|
||||
* socket "data_sock". This could be a request to either write object data or
|
||||
* request object data. If the request is to write object data and 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.
|
||||
* @return Void.
|
||||
*/
|
||||
void send_queued_request(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.
|
||||
* @return Void.
|
||||
*/
|
||||
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
|
||||
|
||||
#endif /* PLASMA_MANAGER_H */
|
||||
@@ -0,0 +1,520 @@
|
||||
/* PLASMA STORE: This is a simple object store server process
|
||||
*
|
||||
* It accepts incoming client connections on a unix domain socket
|
||||
* (name passed in via the -s option of the executable) and uses a
|
||||
* single thread to serve the clients. Each client establishes a
|
||||
* connection and can create objects, wait for objects and seal
|
||||
* objects through that connection.
|
||||
*
|
||||
* It keeps a hash table that maps object_ids (which are 20 byte long,
|
||||
* just enough to store and SHA1 hash) to memory mapped files. */
|
||||
|
||||
#include <assert.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <unistd.h>
|
||||
#include <sys/ioctl.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/types.h>
|
||||
#include <sys/un.h>
|
||||
#include <getopt.h>
|
||||
#include <string.h>
|
||||
#include <signal.h>
|
||||
#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_store.h"
|
||||
|
||||
void *dlmalloc(size_t);
|
||||
void dlfree(void *);
|
||||
|
||||
/**
|
||||
* 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);
|
||||
exit(-1);
|
||||
}
|
||||
}
|
||||
|
||||
typedef struct {
|
||||
/* Object id of this object. */
|
||||
object_id object_id;
|
||||
/* Object info like size, creation time and owner. */
|
||||
plasma_object_info info;
|
||||
/* Memory mapped file containing the object. */
|
||||
int fd;
|
||||
/* Size of the underlying map. */
|
||||
int64_t map_size;
|
||||
/* Offset from the base of the mmap. */
|
||||
ptrdiff_t offset;
|
||||
/* Handle for the uthash table. */
|
||||
UT_hash_handle handle;
|
||||
/* Pointer to the object data. Needed to free the object. */
|
||||
uint8_t *pointer;
|
||||
/** An array of the clients that are currently using this object. */
|
||||
UT_array *clients;
|
||||
} object_table_entry;
|
||||
|
||||
typedef struct {
|
||||
/* Object id of this object. */
|
||||
object_id object_id;
|
||||
/* An array of the clients that are waiting to get this object. */
|
||||
UT_array *waiting_clients;
|
||||
/* Handle for the uthash table. */
|
||||
UT_hash_handle handle;
|
||||
} object_notify_entry;
|
||||
|
||||
/** Contains all information that is associated with a client. */
|
||||
struct client {
|
||||
/** The socket used to communicate with the client. */
|
||||
int sock;
|
||||
/** A pointer to the global plasma state. */
|
||||
plasma_store_state *plasma_state;
|
||||
};
|
||||
|
||||
/* This is used to define the array of clients used to define the
|
||||
* object_table_entry type. */
|
||||
UT_icd client_icd = {sizeof(client *), NULL, NULL, NULL};
|
||||
|
||||
/* This is used to define the array of object IDs used to define the
|
||||
* notification_queue type. */
|
||||
UT_icd object_table_entry_icd = {sizeof(object_id), NULL, NULL, NULL};
|
||||
|
||||
typedef struct {
|
||||
/** Client file descriptor. This is used as a key for the hash table. */
|
||||
int subscriber_fd;
|
||||
/** The object IDs to notify the client about. We notify the client about the
|
||||
* IDs in the order that the objects were sealed. */
|
||||
UT_array *object_ids;
|
||||
/** Handle for the uthash table. */
|
||||
UT_hash_handle hh;
|
||||
} notification_queue;
|
||||
|
||||
struct plasma_store_state {
|
||||
/* Event loop of the plasma store. */
|
||||
event_loop *loop;
|
||||
/* Objects that are still being written by their owner process. */
|
||||
object_table_entry *open_objects;
|
||||
/* Objects that have already been sealed by their owner process and
|
||||
* can now be shared with other processes. */
|
||||
object_table_entry *sealed_objects;
|
||||
/* Objects that processes are waiting for. */
|
||||
object_notify_entry *objects_notify;
|
||||
/** The pending notifications that have not been sent to subscribers because
|
||||
* the socket send buffers were full. This is a hash table from client file
|
||||
* descriptor to an array of object_ids to send to that client. */
|
||||
notification_queue *pending_notifications;
|
||||
};
|
||||
|
||||
plasma_store_state *init_plasma_store(event_loop *loop) {
|
||||
plasma_store_state *state = malloc(sizeof(plasma_store_state));
|
||||
state->loop = loop;
|
||||
state->open_objects = NULL;
|
||||
state->sealed_objects = NULL;
|
||||
state->objects_notify = NULL;
|
||||
state->pending_notifications = NULL;
|
||||
return state;
|
||||
}
|
||||
|
||||
/* If this client is not already using the object, add the client to the
|
||||
* object's list of clients, otherwise do nothing. */
|
||||
void add_client_to_object_clients(object_table_entry *entry,
|
||||
client *client_info) {
|
||||
/* Check if this client is already using the object. */
|
||||
for (int i = 0; i < utarray_len(entry->clients); ++i) {
|
||||
client **c = (client **) utarray_eltptr(entry->clients, i);
|
||||
if (*c == client_info) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
/* Add the client pointer to the list of clients using this object. */
|
||||
utarray_push_back(entry->clients, &client_info);
|
||||
}
|
||||
|
||||
/* Create a new object buffer in the hash table. */
|
||||
void create_object(client *client_context,
|
||||
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 */
|
||||
plasma_store_state *plasma_state = client_context->plasma_state;
|
||||
|
||||
object_table_entry *entry;
|
||||
/* TODO(swang): Return these error to the client instead of exiting. */
|
||||
HASH_FIND(handle, plasma_state->open_objects, &object_id, sizeof(object_id),
|
||||
entry);
|
||||
CHECKM(entry == NULL, "Cannot create object twice.");
|
||||
HASH_FIND(handle, plasma_state->sealed_objects, &object_id, sizeof(object_id),
|
||||
entry);
|
||||
CHECKM(entry == NULL, "Cannot create object twice.");
|
||||
|
||||
uint8_t *pointer = dlmalloc(data_size + metadata_size);
|
||||
int fd;
|
||||
int64_t map_size;
|
||||
ptrdiff_t offset;
|
||||
get_malloc_mapinfo(pointer, &fd, &map_size, &offset);
|
||||
assert(fd != -1);
|
||||
|
||||
entry = malloc(sizeof(object_table_entry));
|
||||
memcpy(&entry->object_id, &object_id, sizeof(object_id));
|
||||
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;
|
||||
utarray_new(entry->clients, &client_icd);
|
||||
HASH_ADD(handle, plasma_state->open_objects, object_id, sizeof(object_id),
|
||||
entry);
|
||||
result->handle.store_fd = fd;
|
||||
result->handle.mmap_size = map_size;
|
||||
result->data_offset = offset;
|
||||
result->metadata_offset = offset + data_size;
|
||||
result->data_size = data_size;
|
||||
result->metadata_size = metadata_size;
|
||||
/* Record that this client is using this object. */
|
||||
add_client_to_object_clients(entry, client_context);
|
||||
}
|
||||
|
||||
/* Get an object from the hash table. */
|
||||
int get_object(client *client_context,
|
||||
int conn,
|
||||
object_id object_id,
|
||||
plasma_object *result) {
|
||||
plasma_store_state *plasma_state = client_context->plasma_state;
|
||||
object_table_entry *entry;
|
||||
HASH_FIND(handle, plasma_state->sealed_objects, &object_id, sizeof(object_id),
|
||||
entry);
|
||||
if (entry) {
|
||||
result->handle.store_fd = entry->fd;
|
||||
result->handle.mmap_size = entry->map_size;
|
||||
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;
|
||||
/* If necessary, record that this client is using this object. In the case
|
||||
* where entry == NULL, this will be called from seal_object. */
|
||||
add_client_to_object_clients(entry, client_context);
|
||||
return OBJECT_FOUND;
|
||||
} else {
|
||||
object_notify_entry *notify_entry;
|
||||
LOG_DEBUG("object not in hash table of sealed objects");
|
||||
HASH_FIND(handle, plasma_state->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));
|
||||
utarray_new(notify_entry->waiting_clients, &client_icd);
|
||||
memcpy(¬ify_entry->object_id, &object_id, sizeof(object_id));
|
||||
HASH_ADD(handle, plasma_state->objects_notify, object_id,
|
||||
sizeof(object_id), notify_entry);
|
||||
}
|
||||
utarray_push_back(notify_entry->waiting_clients, &client_context);
|
||||
}
|
||||
return OBJECT_NOT_FOUND;
|
||||
}
|
||||
|
||||
int remove_client_from_object_clients(object_table_entry *entry,
|
||||
client *client_info) {
|
||||
/* Find the location of the client in the array. */
|
||||
for (int i = 0; i < utarray_len(entry->clients); ++i) {
|
||||
client **c = (client **) utarray_eltptr(entry->clients, i);
|
||||
if (*c == client_info) {
|
||||
/* Remove the client from the array. */
|
||||
utarray_erase(entry->clients, i, 1);
|
||||
/* Return 1 to indicate that the client was removed. */
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
/* Return 0 to indicate that the client was not removed. */
|
||||
return 0;
|
||||
}
|
||||
|
||||
void release_object(client *client_context, object_id object_id) {
|
||||
plasma_store_state *plasma_state = client_context->plasma_state;
|
||||
object_table_entry *open_entry;
|
||||
object_table_entry *sealed_entry;
|
||||
|
||||
HASH_FIND(handle, plasma_state->open_objects, &object_id, sizeof(object_id),
|
||||
open_entry);
|
||||
HASH_FIND(handle, plasma_state->sealed_objects, &object_id, sizeof(object_id),
|
||||
sealed_entry);
|
||||
/* Exactly one of open_entry and sealed_entry should be NULL. */
|
||||
CHECK((open_entry == NULL) != (sealed_entry == NULL));
|
||||
/* Remove the client from the object's array of clients. */
|
||||
if (open_entry != NULL) {
|
||||
CHECK(remove_client_from_object_clients(open_entry, client_context) == 1);
|
||||
} else {
|
||||
CHECK(remove_client_from_object_clients(sealed_entry, client_context) == 1);
|
||||
}
|
||||
}
|
||||
|
||||
/* Check if an object is present. */
|
||||
int contains_object(client *client_context, object_id object_id) {
|
||||
plasma_store_state *plasma_state = client_context->plasma_state;
|
||||
object_table_entry *entry;
|
||||
HASH_FIND(handle, plasma_state->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(client *client_context, object_id object_id) {
|
||||
LOG_DEBUG("sealing object"); // TODO(pcm): add object_id here
|
||||
plasma_store_state *plasma_state = client_context->plasma_state;
|
||||
object_table_entry *entry;
|
||||
HASH_FIND(handle, plasma_state->open_objects, &object_id, sizeof(object_id),
|
||||
entry);
|
||||
CHECK(entry != NULL);
|
||||
/* Move the object table entry from the table of open objects to the table of
|
||||
* sealed objects. */
|
||||
HASH_DELETE(handle, plasma_state->open_objects, entry);
|
||||
HASH_ADD(handle, plasma_state->sealed_objects, object_id, sizeof(object_id),
|
||||
entry);
|
||||
|
||||
/* Inform all subscribers that a new object has been sealed. */
|
||||
notification_queue *queue, *temp_queue;
|
||||
HASH_ITER(hh, plasma_state->pending_notifications, queue, temp_queue) {
|
||||
utarray_push_back(queue->object_ids, &object_id);
|
||||
send_notifications(plasma_state->loop, queue->subscriber_fd, plasma_state,
|
||||
0);
|
||||
}
|
||||
|
||||
/* Inform processes getting this object that the object is ready now. */
|
||||
object_notify_entry *notify_entry;
|
||||
HASH_FIND(handle, plasma_state->objects_notify, &object_id, sizeof(object_id),
|
||||
notify_entry);
|
||||
if (notify_entry) {
|
||||
plasma_reply reply;
|
||||
memset(&reply, 0, sizeof(reply));
|
||||
plasma_object *result = &reply.object;
|
||||
result->handle.store_fd = entry->fd;
|
||||
result->handle.mmap_size = entry->map_size;
|
||||
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, plasma_state->objects_notify, notify_entry);
|
||||
/* Send notifications to the clients that were waiting for this object. */
|
||||
for (int i = 0; i < utarray_len(notify_entry->waiting_clients); ++i) {
|
||||
client **c = (client **) utarray_eltptr(notify_entry->waiting_clients, i);
|
||||
send_fd((*c)->sock, reply.object.handle.store_fd, (char *) &reply,
|
||||
sizeof(reply));
|
||||
/* Record that the client is using this object. */
|
||||
add_client_to_object_clients(entry, *c);
|
||||
}
|
||||
utarray_free(notify_entry->waiting_clients);
|
||||
free(notify_entry);
|
||||
}
|
||||
}
|
||||
|
||||
/* Delete an object that has been created in the hash table. */
|
||||
void delete_object(client *client_context, object_id object_id) {
|
||||
LOG_DEBUG("deleting object"); // TODO(rkn): add object_id here
|
||||
plasma_store_state *plasma_state = client_context->plasma_state;
|
||||
object_table_entry *entry;
|
||||
HASH_FIND(handle, plasma_state->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. */
|
||||
CHECKM(entry != NULL, "To delete an object it must have been sealed.");
|
||||
CHECKM(utarray_len(entry->clients) == 0,
|
||||
"To delete an object, there must be no clients currently using it.");
|
||||
uint8_t *pointer = entry->pointer;
|
||||
HASH_DELETE(handle, plasma_state->sealed_objects, entry);
|
||||
dlfree(pointer);
|
||||
utarray_free(entry->clients);
|
||||
free(entry);
|
||||
}
|
||||
|
||||
/* Send more notifications to a subscriber. */
|
||||
void send_notifications(event_loop *loop,
|
||||
int client_sock,
|
||||
void *context,
|
||||
int events) {
|
||||
plasma_store_state *plasma_state = context;
|
||||
notification_queue *queue;
|
||||
HASH_FIND_INT(plasma_state->pending_notifications, &client_sock, queue);
|
||||
CHECK(queue != NULL);
|
||||
|
||||
int num_processed = 0;
|
||||
/* Loop over the array of pending notifications and send as many of them as
|
||||
* possible. */
|
||||
for (int i = 0; i < utarray_len(queue->object_ids); ++i) {
|
||||
object_id *obj_id = (object_id *) utarray_eltptr(queue->object_ids, i);
|
||||
/* Attempt to send a notification about this object ID. */
|
||||
int nbytes = send(client_sock, obj_id, sizeof(object_id), 0);
|
||||
if (nbytes >= 0) {
|
||||
CHECK(nbytes == sizeof(object_id));
|
||||
} else if (nbytes == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
|
||||
LOG_DEBUG(
|
||||
"The socket's send buffer is full, so we are caching this "
|
||||
"notification and will send it later.");
|
||||
break;
|
||||
} else {
|
||||
CHECKM(0, "This code should be unreachable.");
|
||||
}
|
||||
num_processed += 1;
|
||||
}
|
||||
/* Remove the sent notifications from the array. */
|
||||
utarray_erase(queue->object_ids, 0, num_processed);
|
||||
}
|
||||
|
||||
/* Subscribe to notifications about sealed objects. */
|
||||
void subscribe_to_updates(client *client_context, int conn) {
|
||||
LOG_DEBUG("subscribing to updates");
|
||||
plasma_store_state *plasma_state = client_context->plasma_state;
|
||||
char dummy;
|
||||
int fd = recv_fd(conn, &dummy, 1);
|
||||
CHECKM(HASH_CNT(handle, plasma_state->open_objects) == 0,
|
||||
"plasma_subscribe should be called before any objects are created.");
|
||||
CHECKM(HASH_CNT(handle, plasma_state->sealed_objects) == 0,
|
||||
"plasma_subscribe should be called before any objects are created.");
|
||||
/* Create a new array to buffer notifications that can't be sent to the
|
||||
* subscriber yet because the socket send buffer is full. TODO(rkn): the queue
|
||||
* never gets freed. */
|
||||
notification_queue *queue =
|
||||
(notification_queue *) malloc(sizeof(notification_queue));
|
||||
queue->subscriber_fd = fd;
|
||||
utarray_new(queue->object_ids, &object_table_entry_icd);
|
||||
HASH_ADD_INT(plasma_state->pending_notifications, subscriber_fd, queue);
|
||||
/* Add a callback to the event loop to send queued notifications whenever
|
||||
* there is room in the socket's send buffer. */
|
||||
event_loop_add_file(plasma_state->loop, fd, EVENT_LOOP_WRITE,
|
||||
send_notifications, plasma_state);
|
||||
}
|
||||
|
||||
void process_message(event_loop *loop,
|
||||
int client_sock,
|
||||
void *context,
|
||||
int events) {
|
||||
client *client_context = context;
|
||||
int64_t type;
|
||||
int64_t length;
|
||||
plasma_request *req;
|
||||
read_message(client_sock, &type, &length, (uint8_t **) &req);
|
||||
/* We're only sending a single object ID at a time for now. */
|
||||
plasma_reply reply;
|
||||
memset(&reply, 0, sizeof(reply));
|
||||
/* Process the different types of requests. */
|
||||
switch (type) {
|
||||
case PLASMA_CREATE:
|
||||
create_object(client_context, req->object_ids[0], 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:
|
||||
if (get_object(client_context, client_sock, req->object_ids[0],
|
||||
&reply.object) == OBJECT_FOUND) {
|
||||
send_fd(client_sock, reply.object.handle.store_fd, (char *) &reply,
|
||||
sizeof(reply));
|
||||
}
|
||||
break;
|
||||
case PLASMA_RELEASE:
|
||||
release_object(client_context, req->object_ids[0]);
|
||||
break;
|
||||
case PLASMA_CONTAINS:
|
||||
if (contains_object(client_context, req->object_ids[0]) == OBJECT_FOUND) {
|
||||
reply.has_object = 1;
|
||||
}
|
||||
plasma_send_reply(client_sock, &reply);
|
||||
break;
|
||||
case PLASMA_SEAL:
|
||||
seal_object(client_context, req->object_ids[0]);
|
||||
break;
|
||||
case PLASMA_DELETE:
|
||||
delete_object(client_context, req->object_ids[0]);
|
||||
break;
|
||||
case PLASMA_SUBSCRIBE:
|
||||
subscribe_to_updates(client_context, client_sock);
|
||||
break;
|
||||
case DISCONNECT_CLIENT: {
|
||||
LOG_DEBUG("Disconnecting client on fd %d", client_sock);
|
||||
event_loop_remove_file(loop, client_sock);
|
||||
/* If this client was using any objects, remove it from the appropriate
|
||||
* lists. */
|
||||
plasma_store_state *plasma_state = client_context->plasma_state;
|
||||
object_table_entry *entry, *temp_entry;
|
||||
HASH_ITER(handle, plasma_state->open_objects, entry, temp_entry) {
|
||||
remove_client_from_object_clients(entry, client_context);
|
||||
}
|
||||
HASH_ITER(handle, plasma_state->sealed_objects, entry, temp_entry) {
|
||||
remove_client_from_object_clients(entry, client_context);
|
||||
}
|
||||
} break;
|
||||
default:
|
||||
/* This code should be unreachable. */
|
||||
CHECK(0);
|
||||
}
|
||||
|
||||
free(req);
|
||||
}
|
||||
|
||||
void new_client_connection(event_loop *loop,
|
||||
int listener_sock,
|
||||
void *context,
|
||||
int events) {
|
||||
plasma_store_state *plasma_state = context;
|
||||
int new_socket = accept_client(listener_sock);
|
||||
/* Create a new client object. This will also be used as the context to use
|
||||
* for events on this client's socket. TODO(rkn): free this somewhere. */
|
||||
client *client_context = (client *) malloc(sizeof(client));
|
||||
client_context->sock = new_socket;
|
||||
client_context->plasma_state = plasma_state;
|
||||
/* Add a callback to handle events on this socket. */
|
||||
event_loop_add_file(loop, new_socket, EVENT_LOOP_READ, process_message,
|
||||
client_context);
|
||||
LOG_DEBUG("new connection with fd %d", new_socket);
|
||||
}
|
||||
|
||||
/* Report "success" to valgrind. */
|
||||
void signal_handler(int signal) {
|
||||
if (signal == SIGTERM) {
|
||||
exit(0);
|
||||
}
|
||||
}
|
||||
|
||||
void start_server(char *socket_name) {
|
||||
int socket = bind_ipc_sock(socket_name);
|
||||
CHECK(socket >= 0);
|
||||
event_loop *loop = event_loop_create();
|
||||
plasma_store_state *state = init_plasma_store(loop);
|
||||
event_loop_add_file(loop, socket, EVENT_LOOP_READ, new_client_connection,
|
||||
state);
|
||||
event_loop_run(loop);
|
||||
}
|
||||
|
||||
int main(int argc, char *argv[]) {
|
||||
signal(SIGTERM, signal_handler);
|
||||
char *socket_name = NULL;
|
||||
int c;
|
||||
while ((c = getopt(argc, argv, "s:")) != -1) {
|
||||
switch (c) {
|
||||
case 's':
|
||||
socket_name = optarg;
|
||||
break;
|
||||
default:
|
||||
exit(-1);
|
||||
}
|
||||
}
|
||||
if (!socket_name) {
|
||||
LOG_ERR("please specify socket for incoming connections with -s switch");
|
||||
exit(-1);
|
||||
}
|
||||
LOG_DEBUG("starting server listening on %s", socket_name);
|
||||
start_server(socket_name);
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
#ifndef PLASMA_STORE_H
|
||||
#define PLASMA_STORE_H
|
||||
|
||||
#include "plasma.h"
|
||||
|
||||
typedef struct client client;
|
||||
|
||||
typedef struct plasma_store_state plasma_store_state;
|
||||
|
||||
/**
|
||||
* Create a new object. The client must do a call to release_object to tell the
|
||||
* store when it is done with the object.
|
||||
*
|
||||
* @param client_context The context of the client making this request.
|
||||
* @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.
|
||||
* @return Void.
|
||||
*/
|
||||
void create_object(client *client_context,
|
||||
object_id object_id,
|
||||
int64_t data_size,
|
||||
int64_t metadata_size,
|
||||
plasma_object *result);
|
||||
|
||||
/**
|
||||
* Get an object. This method assumes that we currently have or will eventually
|
||||
* have this object sealed. If the object has not yet been sealed, the client
|
||||
* that requested the object will be notified when it is sealed.
|
||||
*
|
||||
* For each call to get_object, the client must do a call to release_object to
|
||||
* tell the store when it is done with the object.
|
||||
*
|
||||
* @param client_context The context of the client making this request.
|
||||
* @param conn The client connection that requests the object.
|
||||
* @param object_id Object ID of the object to be gotten.
|
||||
* @return The status of the object (object_status in plasma.h).
|
||||
*/
|
||||
int get_object(client *client_context,
|
||||
int conn,
|
||||
object_id object_id,
|
||||
plasma_object *result);
|
||||
|
||||
/**
|
||||
* Record the fact that a particular client is no longer using an object.
|
||||
*
|
||||
* @param client_context The context of the client making this request.
|
||||
* @param object_id The object ID of the object that is being released.
|
||||
* @param Void.
|
||||
*/
|
||||
void release_object(client *client_context, object_id object_id);
|
||||
|
||||
/**
|
||||
* Seal an object. The object is now immutable and can be accessed with get.
|
||||
*
|
||||
* @param client_context The context of the client making this request.
|
||||
* @param object_id Object ID of the object to be sealed.
|
||||
* @return Void.
|
||||
*/
|
||||
void seal_object(client *client_context, object_id object_id);
|
||||
|
||||
/**
|
||||
* Check if the plasma store contains an object:
|
||||
*
|
||||
* @param client_context The context of the client making this request.
|
||||
* @param object_id Object ID that will be checked.
|
||||
* @return OBJECT_FOUND if the object is in the store, OBJECT_NOT_FOUND if not
|
||||
*/
|
||||
int contains_object(client *client_context, object_id object_id);
|
||||
|
||||
/**
|
||||
* Delete an object from the plasma store:
|
||||
*
|
||||
* @param client_context The context of the client making this request.
|
||||
* @param object_id Object ID of the object to be deleted.
|
||||
* @return Void.
|
||||
*/
|
||||
void delete_object(client *client_context, object_id object_id);
|
||||
|
||||
/**
|
||||
* Send notifications about sealed objects to the subscribers. This is called
|
||||
* in seal_object. If the socket's send buffer is full, the notification will be
|
||||
* buffered, and this will be called again when the send buffer has room.
|
||||
*
|
||||
* @param loop The Plasma store event loop.
|
||||
* @param client_sock The socket of the client to send the notification to.
|
||||
* @param plasma_state The plasma store global state.
|
||||
* @param events This is needed for this function to have the signature of a
|
||||
callback.
|
||||
* @return Void.
|
||||
*/
|
||||
void send_notifications(event_loop *loop,
|
||||
int client_sock,
|
||||
void *plasma_state,
|
||||
int events);
|
||||
|
||||
#endif /* PLASMA_STORE_H */
|
||||
@@ -0,0 +1,396 @@
|
||||
from __future__ import print_function
|
||||
|
||||
import os
|
||||
import signal
|
||||
import socket
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
import random
|
||||
import time
|
||||
import tempfile
|
||||
|
||||
import plasma
|
||||
|
||||
USE_VALGRIND = False
|
||||
|
||||
def random_object_id():
|
||||
return "".join([chr(random.randint(0, 255)) for _ in range(plasma.PLASMA_ID_SIZE)])
|
||||
|
||||
def generate_metadata(length):
|
||||
metadata = length * ["\x00"]
|
||||
if length > 0:
|
||||
metadata[0] = chr(random.randint(0, 255))
|
||||
metadata[-1] = chr(random.randint(0, 255))
|
||||
for _ in range(100):
|
||||
metadata[random.randint(0, length - 1)] = chr(random.randint(0, 255))
|
||||
return buffer("".join(metadata))
|
||||
|
||||
def write_to_data_buffer(buff, length):
|
||||
if length > 0:
|
||||
buff[0] = chr(random.randint(0, 255))
|
||||
buff[-1] = chr(random.randint(0, 255))
|
||||
for _ in range(100):
|
||||
buff[random.randint(0, length - 1)] = chr(random.randint(0, 255))
|
||||
|
||||
def create_object(client, data_size, metadata_size, seal=True):
|
||||
object_id = random_object_id()
|
||||
metadata = generate_metadata(metadata_size)
|
||||
memory_buffer = client.create(object_id, data_size, metadata)
|
||||
write_to_data_buffer(memory_buffer, data_size)
|
||||
if seal:
|
||||
client.seal(object_id)
|
||||
return object_id, memory_buffer, metadata
|
||||
|
||||
def assert_get_object_equal(unit_test, client1, client2, object_id, memory_buffer=None, metadata=None):
|
||||
if memory_buffer is not None:
|
||||
unit_test.assertEqual(memory_buffer[:], client2.get(object_id)[:])
|
||||
if metadata is not None:
|
||||
unit_test.assertEqual(metadata[:], client2.get_metadata(object_id)[:])
|
||||
unit_test.assertEqual(client1.get(object_id)[:], client2.get(object_id)[:])
|
||||
unit_test.assertEqual(client1.get_metadata(object_id)[:],
|
||||
client2.get_metadata(object_id)[:])
|
||||
|
||||
class TestPlasmaClient(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Start Plasma.
|
||||
plasma_store_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_store")
|
||||
store_name = "/tmp/store{}".format(random.randint(0, 10000))
|
||||
command = [plasma_store_executable, "-s", store_name]
|
||||
if USE_VALGRIND:
|
||||
self.p = subprocess.Popen(["valgrind", "--track-origins=yes", "--leak-check=full"] + command)
|
||||
time.sleep(2.0)
|
||||
else:
|
||||
self.p = subprocess.Popen(command)
|
||||
# Connect to Plasma.
|
||||
self.plasma_client = plasma.PlasmaClient(store_name)
|
||||
|
||||
def tearDown(self):
|
||||
# Kill the plasma store process.
|
||||
if USE_VALGRIND:
|
||||
self.p.send_signal(signal.SIGTERM)
|
||||
self.p.wait()
|
||||
if self.p.returncode != 0:
|
||||
os._exit(-1)
|
||||
else:
|
||||
self.p.kill()
|
||||
|
||||
def test_create(self):
|
||||
# Create an object id string.
|
||||
object_id = random_object_id()
|
||||
# Create a new buffer and write to it.
|
||||
length = 50
|
||||
memory_buffer = self.plasma_client.create(object_id, length)
|
||||
for i in range(length):
|
||||
memory_buffer[i] = chr(i % 256)
|
||||
# Seal the object.
|
||||
self.plasma_client.seal(object_id)
|
||||
# Get the object.
|
||||
memory_buffer = self.plasma_client.get(object_id)
|
||||
for i in range(length):
|
||||
self.assertEqual(memory_buffer[i], chr(i % 256))
|
||||
|
||||
def test_create_with_metadata(self):
|
||||
for length in range(1000):
|
||||
# Create an object id string.
|
||||
object_id = random_object_id()
|
||||
# Create a random metadata string.
|
||||
metadata = generate_metadata(length)
|
||||
# Create a new buffer and write to it.
|
||||
memory_buffer = self.plasma_client.create(object_id, length, metadata)
|
||||
for i in range(length):
|
||||
memory_buffer[i] = chr(i % 256)
|
||||
# Seal the object.
|
||||
self.plasma_client.seal(object_id)
|
||||
# Get the object.
|
||||
memory_buffer = self.plasma_client.get(object_id)
|
||||
for i in range(length):
|
||||
self.assertEqual(memory_buffer[i], chr(i % 256))
|
||||
# Get the metadata.
|
||||
metadata_buffer = self.plasma_client.get_metadata(object_id)
|
||||
self.assertEqual(len(metadata), len(metadata_buffer))
|
||||
for i in range(len(metadata)):
|
||||
self.assertEqual(metadata[i], metadata_buffer[i])
|
||||
|
||||
def test_contains(self):
|
||||
fake_object_ids = [random_object_id() for _ in range(100)]
|
||||
real_object_ids = [random_object_id() for _ in range(100)]
|
||||
for object_id in real_object_ids:
|
||||
self.assertFalse(self.plasma_client.contains(object_id))
|
||||
memory_buffer = self.plasma_client.create(object_id, 100)
|
||||
self.plasma_client.seal(object_id)
|
||||
self.assertTrue(self.plasma_client.contains(object_id))
|
||||
for object_id in fake_object_ids:
|
||||
self.assertFalse(self.plasma_client.contains(object_id))
|
||||
for object_id in real_object_ids:
|
||||
self.assertTrue(self.plasma_client.contains(object_id))
|
||||
|
||||
# def test_individual_delete(self):
|
||||
# length = 100
|
||||
# # Create an object id string.
|
||||
# object_id = random_object_id()
|
||||
# # Create a random metadata string.
|
||||
# metadata = generate_metadata(100)
|
||||
# # Create a new buffer and write to it.
|
||||
# memory_buffer = self.plasma_client.create(object_id, length, metadata)
|
||||
# for i in range(length):
|
||||
# memory_buffer[i] = chr(i % 256)
|
||||
# # Seal the object.
|
||||
# self.plasma_client.seal(object_id)
|
||||
# # Check that the object is present.
|
||||
# self.assertTrue(self.plasma_client.contains(object_id))
|
||||
# # Delete the object.
|
||||
# self.plasma_client.delete(object_id)
|
||||
# # Make sure the object is no longer present.
|
||||
# self.assertFalse(self.plasma_client.contains(object_id))
|
||||
#
|
||||
# def test_delete(self):
|
||||
# # Create some objects.
|
||||
# object_ids = [random_object_id() for _ in range(100)]
|
||||
# for object_id in object_ids:
|
||||
# length = 100
|
||||
# # Create a random metadata string.
|
||||
# metadata = generate_metadata(100)
|
||||
# # Create a new buffer and write to it.
|
||||
# memory_buffer = self.plasma_client.create(object_id, length, metadata)
|
||||
# for i in range(length):
|
||||
# memory_buffer[i] = chr(i % 256)
|
||||
# # Seal the object.
|
||||
# self.plasma_client.seal(object_id)
|
||||
# # Check that the object is present.
|
||||
# self.assertTrue(self.plasma_client.contains(object_id))
|
||||
#
|
||||
# # Delete the objects and make sure they are no longer present.
|
||||
# for object_id in object_ids:
|
||||
# # Delete the object.
|
||||
# self.plasma_client.delete(object_id)
|
||||
# # Make sure the object is no longer present.
|
||||
# self.assertFalse(self.plasma_client.contains(object_id))
|
||||
|
||||
def test_illegal_functionality(self):
|
||||
# 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)
|
||||
# Make sure we cannot access memory out of bounds.
|
||||
self.assertRaises(Exception, lambda : memory_buffer[length])
|
||||
# Seal the object.
|
||||
self.plasma_client.seal(object_id)
|
||||
# This test is commented out because it currently fails.
|
||||
# # Make sure the object is ready only now.
|
||||
# def illegal_assignment():
|
||||
# memory_buffer[0] = chr(0)
|
||||
# self.assertRaises(Exception, illegal_assignment)
|
||||
# Get the object.
|
||||
memory_buffer = self.plasma_client.get(object_id)
|
||||
# Make sure the object is read only.
|
||||
def illegal_assignment():
|
||||
memory_buffer[0] = chr(0)
|
||||
self.assertRaises(Exception, illegal_assignment)
|
||||
|
||||
def test_subscribe(self):
|
||||
# Subscribe to notifications from the Plasma Store.
|
||||
sock = self.plasma_client.subscribe()
|
||||
for i in [1, 10, 100, 1000, 10000, 100000]:
|
||||
object_ids = [random_object_id() for _ in range(i)]
|
||||
for object_id in object_ids:
|
||||
# Create an object and seal it to trigger a notification.
|
||||
self.plasma_client.create(object_id, 1000)
|
||||
self.plasma_client.seal(object_id)
|
||||
# Check that we received notifications for all of the objects.
|
||||
for object_id in object_ids:
|
||||
message_data = self.plasma_client.get_next_notification()
|
||||
self.assertEqual(object_id, message_data)
|
||||
|
||||
class TestPlasmaManager(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Start two PlasmaStores.
|
||||
plasma_store_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_store")
|
||||
store_name1 = "/tmp/store{}".format(random.randint(0, 10000))
|
||||
store_name2 = "/tmp/store{}".format(random.randint(0, 10000))
|
||||
plasma_store_command1 = [plasma_store_executable, "-s", store_name1]
|
||||
plasma_store_command2 = [plasma_store_executable, "-s", store_name2]
|
||||
|
||||
if USE_VALGRIND:
|
||||
self.p2 = subprocess.Popen(["valgrind", "--track-origins=yes", "--leak-check=full", "--error-exitcode=1"] + plasma_store_command1)
|
||||
self.p3 = subprocess.Popen(["valgrind", "--track-origins=yes", "--leak-check=full", "--error-exitcode=1"] + plasma_store_command2)
|
||||
else:
|
||||
self.p2 = subprocess.Popen(plasma_store_command1)
|
||||
self.p3 = subprocess.Popen(plasma_store_command2)
|
||||
|
||||
# Start a Redis server.
|
||||
redis_path = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../common/thirdparty/redis-3.2.3/src/redis-server")
|
||||
self.redis_process = None
|
||||
manager_redis_args = []
|
||||
if os.path.exists(redis_path):
|
||||
redis_port = 6379
|
||||
with open(os.devnull, 'w') as FNULL:
|
||||
self.redis_process = subprocess.Popen([redis_path,
|
||||
"--port", str(redis_port)],
|
||||
stdout=FNULL)
|
||||
time.sleep(0.1)
|
||||
manager_redis_args = ["-d", "{addr}:{port}".format(addr="127.0.0.1",
|
||||
port=redis_port)]
|
||||
|
||||
# Start two PlasmaManagers.
|
||||
self.port1 = random.randint(10000, 50000)
|
||||
self.port2 = random.randint(10000, 50000)
|
||||
plasma_manager_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_manager")
|
||||
plasma_manager_command1 = [plasma_manager_executable,
|
||||
"-s", store_name1,
|
||||
"-m", "127.0.0.1",
|
||||
"-p", str(self.port1)] + manager_redis_args
|
||||
plasma_manager_command2 = [plasma_manager_executable,
|
||||
"-s", store_name2,
|
||||
"-m", "127.0.0.1",
|
||||
"-p", str(self.port2)] + manager_redis_args
|
||||
|
||||
if USE_VALGRIND:
|
||||
self.p4 = subprocess.Popen(["valgrind", "--track-origins=yes", "--leak-check=full", "--error-exitcode=1"] + plasma_manager_command1)
|
||||
self.p5 = subprocess.Popen(["valgrind", "--track-origins=yes", "--leak-check=full", "--error-exitcode=1"] + plasma_manager_command2)
|
||||
time.sleep(2.0)
|
||||
else:
|
||||
self.p4 = subprocess.Popen(plasma_manager_command1)
|
||||
self.p5 = subprocess.Popen(plasma_manager_command2)
|
||||
time.sleep(0.1)
|
||||
|
||||
# Connect two PlasmaClients.
|
||||
self.client1 = plasma.PlasmaClient(store_name1, "127.0.0.1", self.port1)
|
||||
self.client2 = plasma.PlasmaClient(store_name2, "127.0.0.1", self.port2)
|
||||
time.sleep(0.5)
|
||||
|
||||
def tearDown(self):
|
||||
# Kill the PlasmaStore and PlasmaManager processes.
|
||||
if USE_VALGRIND:
|
||||
self.p4.send_signal(signal.SIGTERM)
|
||||
self.p4.wait()
|
||||
self.p5.send_signal(signal.SIGTERM)
|
||||
self.p5.wait()
|
||||
self.p2.send_signal(signal.SIGTERM)
|
||||
self.p2.wait()
|
||||
self.p3.send_signal(signal.SIGTERM)
|
||||
self.p3.wait()
|
||||
if self.p2.returncode != 0 or self.p3.returncode != 0 or self.p4.returncode != 0 or self.p5.returncode != 0:
|
||||
print("aborting due to valgrind error")
|
||||
os._exit(-1)
|
||||
else:
|
||||
self.p2.kill()
|
||||
self.p3.kill()
|
||||
self.p4.kill()
|
||||
self.p5.kill()
|
||||
if self.redis_process:
|
||||
self.redis_process.kill()
|
||||
|
||||
def test_fetch(self):
|
||||
if self.redis_process is None:
|
||||
print("Cannot test fetch without a running redis instance.")
|
||||
self.assertTrue(False)
|
||||
for _ in range(100):
|
||||
# Create an object.
|
||||
object_id1, memory_buffer1, metadata1 = create_object(self.client1, 2000, 2000)
|
||||
# Fetch the object from the other plasma store.
|
||||
# TODO(swang): This line is a hack! It makes sure that the entry will be
|
||||
# in the object table once we call the fetch operation. Remove once
|
||||
# retries are implemented by Ray common.
|
||||
time.sleep(0.1)
|
||||
successes = self.client2.fetch([object_id1])
|
||||
self.assertEqual(successes, [True])
|
||||
# Compare the two buffers.
|
||||
assert_get_object_equal(self, self.client1, self.client2, object_id1,
|
||||
memory_buffer=memory_buffer1, metadata=metadata1)
|
||||
# Fetch in the other direction. These should return quickly because
|
||||
# client1 already has the object.
|
||||
successes = self.client1.fetch([object_id1])
|
||||
self.assertEqual(successes, [True])
|
||||
assert_get_object_equal(self, self.client2, self.client1, object_id1,
|
||||
memory_buffer=memory_buffer1, metadata=metadata1)
|
||||
|
||||
def test_fetch_multiple(self):
|
||||
if self.redis_process is None:
|
||||
print("Cannot test fetch without a running redis instance.")
|
||||
self.assertTrue(False)
|
||||
for _ in range(20):
|
||||
# Create two objects and a third fake one that doesn't exist.
|
||||
object_id1, memory_buffer1, metadata1 = create_object(self.client1, 2000, 2000)
|
||||
missing_object_id = random_object_id()
|
||||
object_id2, memory_buffer2, metadata2 = create_object(self.client1, 2000, 2000)
|
||||
object_ids = [object_id1, missing_object_id, object_id2]
|
||||
# Fetch the objects from the other plasma store. The second object ID
|
||||
# should timeout since it does not exist.
|
||||
# TODO(swang): This line is a hack! It makes sure that the entry will be
|
||||
# in the object table once we call the fetch operation. Remove once
|
||||
# retries are implemented by Ray common.
|
||||
time.sleep(0.1)
|
||||
successes = self.client2.fetch(object_ids)
|
||||
self.assertEqual(successes, [True, False, True])
|
||||
# Compare the buffers of the objects that do exist.
|
||||
assert_get_object_equal(self, self.client1, self.client2, object_id1,
|
||||
memory_buffer=memory_buffer1, metadata=metadata1)
|
||||
assert_get_object_equal(self, self.client1, self.client2, object_id2,
|
||||
memory_buffer=memory_buffer2, metadata=metadata2)
|
||||
# Fetch in the other direction. The fake object still does not exist.
|
||||
successes = self.client1.fetch(object_ids)
|
||||
self.assertEqual(successes, [True, False, True])
|
||||
assert_get_object_equal(self, self.client2, self.client1, object_id1,
|
||||
memory_buffer=memory_buffer1, metadata=metadata1)
|
||||
assert_get_object_equal(self, self.client2, self.client1, object_id2,
|
||||
memory_buffer=memory_buffer2, metadata=metadata2)
|
||||
|
||||
def test_transfer(self):
|
||||
for _ in range(100):
|
||||
# Create an object.
|
||||
object_id1, memory_buffer1, metadata1 = create_object(self.client1, 2000, 2000)
|
||||
# Transfer the buffer to the the other PlasmaStore.
|
||||
self.client1.transfer("127.0.0.1", self.port2, object_id1)
|
||||
# Compare the two buffers.
|
||||
assert_get_object_equal(self, self.client1, self.client2, object_id1,
|
||||
memory_buffer=memory_buffer1, metadata=metadata1)
|
||||
# # Transfer the buffer again.
|
||||
# self.client1.transfer("127.0.0.1", self.port2, object_id1)
|
||||
# # Compare the two buffers.
|
||||
# assert_get_object_equal(self, self.client1, self.client2, object_id1,
|
||||
# memory_buffer=memory_buffer1, metadata=metadata1)
|
||||
|
||||
# Create an object.
|
||||
object_id2, memory_buffer2, metadata2 = create_object(self.client2, 20000, 20000)
|
||||
# Transfer the buffer to the the other PlasmaStore.
|
||||
self.client2.transfer("127.0.0.1", self.port1, object_id2)
|
||||
# Compare the two buffers.
|
||||
assert_get_object_equal(self, self.client1, self.client2, object_id2,
|
||||
memory_buffer=memory_buffer2, metadata=metadata2)
|
||||
|
||||
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))
|
||||
|
||||
def test_stresstest(self):
|
||||
a = time.time()
|
||||
object_ids = []
|
||||
for i in range(10000): # TODO(pcm): increase this to 100000
|
||||
object_id = random_object_id()
|
||||
object_ids.append(object_id)
|
||||
self.client1.create(object_id, 1)
|
||||
self.client1.seal(object_id)
|
||||
for object_id in object_ids:
|
||||
self.client1.transfer("127.0.0.1", self.port2, object_id)
|
||||
b = time.time() - a
|
||||
|
||||
print("it took", b, "seconds to put and transfer the objects")
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) > 1:
|
||||
# pop the argument so we don't mess with unittest's own argument parser
|
||||
if sys.argv[-1] == "valgrind":
|
||||
arg = sys.argv.pop()
|
||||
USE_VALGRIND = True
|
||||
print("Using valgrind for tests")
|
||||
unittest.main(verbosity=2)
|
||||
Vendored
+6280
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user