From 10e067e5e571eb696a39840ae08630e9d064d5e3 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Tue, 27 Dec 2016 19:51:26 -0800 Subject: [PATCH] Delay releasing a maximum number of bytes in the plasma client. (#160) * Send message from plasma client to get plasma store capacity. * Release objects from plasma client if they are too large. * Use doubly-linked list instead of ring buffer for plasma client release history. * Address comments. * Fix problem with slicing PlasmaBuffer objects. * Fix crash in plasma manager during transfer. * Formatting. * Make plasma client cache larger and make caching test not throw exceptions on Travis. --- src/common/Makefile | 4 +- src/photon/Makefile | 16 +++- src/photon/photon_scheduler.c | 3 +- src/photon/test/photon_tests.c | 16 ++-- src/plasma/Makefile | 6 ++ src/plasma/eviction_policy.c | 11 +-- src/plasma/eviction_policy.h | 2 +- src/plasma/format/plasma.fbs | 15 ++++ src/plasma/plasma.h | 3 + src/plasma/plasma/plasma.py | 22 +++++ src/plasma/plasma_client.c | 151 ++++++++++++++++++++++++++------ src/plasma/plasma_client.h | 2 + src/plasma/plasma_manager.c | 15 +++- src/plasma/plasma_protocol.c | 28 ++++++ src/plasma/plasma_protocol.h | 12 +++ src/plasma/plasma_store.c | 8 +- src/plasma/test/manager_tests.c | 15 ++-- src/plasma/test/test.py | 19 ++-- test/microbenchmarks.py | 12 ++- 19 files changed, 291 insertions(+), 69 deletions(-) diff --git a/src/common/Makefile b/src/common/Makefile index 33de5d268..0d90afbcd 100644 --- a/src/common/Makefile +++ b/src/common/Makefile @@ -44,7 +44,7 @@ redismodule: test: CFLAGS += -DRAY_COMMON_LOG_LEVEL=4 test: hiredis redis redismodule $(BUILD)/common_tests $(BUILD)/task_table_tests $(BUILD)/object_table_tests $(BUILD)/db_tests $(BUILD)/io_tests $(BUILD)/task_tests $(BUILD)/redis_tests FORCE - ./thirdparty/redis/src/redis-server --loadmodule ./redis_module/ray_redis_module.so & + ./thirdparty/redis/src/redis-server --loglevel warning --loadmodule ./redis_module/ray_redis_module.so & sleep 1s ./build/common_tests ./build/db_tests @@ -57,7 +57,7 @@ test: hiredis redis redismodule $(BUILD)/common_tests $(BUILD)/task_table_tests python ./redis_module/runtest.py valgrind: test - ./thirdparty/redis/src/redis-server --loadmodule redis_module/ray_redis_module.so & + ./thirdparty/redis/src/redis-server --loglevel warning --loadmodule redis_module/ray_redis_module.so & sleep 1s valgrind --leak-check=full --error-exitcode=1 ./build/common_tests valgrind --leak-check=full --error-exitcode=1 ./build/db_tests diff --git a/src/photon/Makefile b/src/photon/Makefile index 9be039ab5..162c00c66 100644 --- a/src/photon/Makefile +++ b/src/photon/Makefile @@ -25,11 +25,19 @@ clean: # Set the request timeout low and logging level at FATAL for testing purposes. test: CFLAGS += -DRAY_TIMEOUT=50 -DRAY_COMMON_LOG_LEVEL=4 test: $(BUILD)/photon_tests FORCE - ../common/thirdparty/redis/src/redis-server --loadmodule ../common/redis_module/ray_redis_module.so & - sleep 0.5s && ./build/photon_tests && ../common/thirdparty/redis/src/redis-cli shutdown + ../common/thirdparty/redis/src/redis-server --loglevel warning --loadmodule ../common/redis_module/ray_redis_module.so & + ../plasma/build/plasma_store -s /tmp/plasma_store_socket_1 -m 100000000 & + sleep 0.5s + ./build/photon_tests + ../common/thirdparty/redis/src/redis-cli shutdown + killall plasma_store valgrind: test - ../common/thirdparty/redis/src/redis-server --loadmodule ../common/redis_module/ray_redis_module.so & - sleep 0.5s && valgrind --leak-check=full --show-leak-kinds=all --error-exitcode=1 ./build/photon_tests && ../common/thirdparty/redis/src/redis-cli shutdown + ../common/thirdparty/redis/src/redis-server --loglevel warning --loadmodule ../common/redis_module/ray_redis_module.so & + ../plasma/build/plasma_store -s /tmp/plasma_store_socket_1 -m 100000000 & + sleep 0.5s + valgrind --leak-check=full --show-leak-kinds=all --error-exitcode=1 ./build/photon_tests + ../common/thirdparty/redis/src/redis-cli shutdown + killall plasma_store FORCE: diff --git a/src/photon/photon_scheduler.c b/src/photon/photon_scheduler.c index b47019e79..328ce0943 100644 --- a/src/photon/photon_scheduler.c +++ b/src/photon/photon_scheduler.c @@ -63,8 +63,7 @@ local_scheduler_state *init_local_scheduler( } else { state->db = NULL; } - /* Connect to Plasma. This method will retry if Plasma hasn't started yet. - * Pass in a NULL manager address and port. */ + /* Connect to Plasma. This method will retry if Plasma hasn't started yet. */ state->plasma_conn = plasma_connect(plasma_store_socket_name, plasma_manager_socket_name, PLASMA_DEFAULT_RELEASE_DELAY); diff --git a/src/photon/test/photon_tests.c b/src/photon/test/photon_tests.c index fa5ce87b6..833a83017 100644 --- a/src/photon/test/photon_tests.c +++ b/src/photon/test/photon_tests.c @@ -22,7 +22,8 @@ SUITE(photon_tests); -const char *plasma_socket_name_format = "/tmp/plasma_socket_%d"; +const char *plasma_store_socket_name = "/tmp/plasma_store_socket_1"; +const char *plasma_manager_socket_name_format = "/tmp/plasma_manager_socket_%d"; const char *photon_socket_name_format = "/tmp/photon_socket_%d"; int64_t timeout_handler(event_loop *loop, int64_t id, void *context) { @@ -50,24 +51,23 @@ photon_mock *init_photon_mock() { memset(mock, 0, sizeof(photon_mock)); mock->loop = event_loop_create(); /* Bind to the Photon port and initialize the Photon scheduler. */ + /* TODO(rkn): Why are we reusing mock->plasma_fd for both the store and the + * manager? */ UT_string *plasma_manager_socket_name = - bind_ipc_sock_retry(plasma_socket_name_format, &mock->plasma_fd); - UT_string *plasma_store_socket_name = - bind_ipc_sock_retry(plasma_socket_name_format, &mock->plasma_fd); + bind_ipc_sock_retry(plasma_manager_socket_name_format, &mock->plasma_fd); + mock->plasma_fd = socket_connect_retry(plasma_store_socket_name, 5, 100); UT_string *photon_socket_name = bind_ipc_sock_retry(photon_socket_name_format, &mock->photon_fd); CHECK(mock->plasma_fd >= 0 && mock->photon_fd >= 0); mock->photon_state = init_local_scheduler( "127.0.0.1", mock->loop, redis_addr, redis_port, - utstring_body(photon_socket_name), - utstring_body(plasma_manager_socket_name), - utstring_body(plasma_store_socket_name), NULL, false); + utstring_body(photon_socket_name), plasma_store_socket_name, + utstring_body(plasma_manager_socket_name), NULL, false); /* Connect a Photon client. */ mock->conn = photon_connect(utstring_body(photon_socket_name)); new_client_connection(mock->loop, mock->photon_fd, (void *) mock->photon_state, 0); utstring_free(plasma_manager_socket_name); - utstring_free(plasma_store_socket_name); utstring_free(photon_socket_name); return mock; } diff --git a/src/plasma/Makefile b/src/plasma/Makefile index 1d326ed4a..725df02a0 100644 --- a/src/plasma/Makefile +++ b/src/plasma/Makefile @@ -46,7 +46,10 @@ common: FORCE test: CFLAGS += -DRAY_TIMEOUT=50 -DRAY_COMMON_LOG_LEVEL=4 # First, build and run all the unit tests. test: $(BUILD)/manager_tests $(BUILD)/client_tests $(BUILD)/serialization_tests FORCE + ./build/plasma_store -s /tmp/plasma_store_socket_1 -m 0 & + sleep 1 ./build/manager_tests + killall plasma_store ./build/serialization_tests ./test/run_client_tests.sh cd ../common; make redis @@ -54,7 +57,10 @@ test: $(BUILD)/manager_tests $(BUILD)/client_tests $(BUILD)/serialization_tests test: all valgrind: test + ./build/plasma_store -s /tmp/plasma_store_socket_1 -m 0 & + sleep 1 valgrind --leak-check=full --error-exitcode=1 ./build/manager_tests + killall plasma_store valgrind --leak-check=full --error-exitcode=1 ./build/serialization_tests FORCE: diff --git a/src/plasma/eviction_policy.c b/src/plasma/eviction_policy.c index fc2533387..deb8e4043 100644 --- a/src/plasma/eviction_policy.c +++ b/src/plasma/eviction_policy.c @@ -27,9 +27,6 @@ typedef struct { /** The part of the Plasma state that is maintained by the eviction policy. */ struct eviction_state { - /** The amount of memory (in bytes) that we allow to be allocated in the - * store. */ - int64_t memory_capacity; /** The amount of memory (in bytes) currently being used. */ int64_t memory_used; /** A doubly-linked list of the released objects in order from least recently @@ -44,10 +41,8 @@ struct eviction_state { * released_objects type. */ UT_icd released_objects_entry_icd = {sizeof(object_id), NULL, NULL, NULL}; -eviction_state *make_eviction_state(int64_t system_memory) { +eviction_state *make_eviction_state(void) { eviction_state *state = malloc(sizeof(eviction_state)); - /* Find the amount of available memory on the machine. */ - state->memory_capacity = system_memory; state->memory_used = 0; state->released_objects = NULL; state->released_object_table = NULL; @@ -166,11 +161,11 @@ void require_space(eviction_state *eviction_state, object_id **objects_to_evict) { /* Check if there is enough space to create the object. */ int64_t required_space = - eviction_state->memory_used + size - eviction_state->memory_capacity; + eviction_state->memory_used + size - plasma_store_info->memory_capacity; if (required_space > 0) { /* Try to free up at least as much space as we need right now but ideally * up to 20% of the total capacity. */ - int64_t space_to_free = MAX(size, eviction_state->memory_capacity / 5); + int64_t space_to_free = MAX(size, plasma_store_info->memory_capacity / 5); LOG_DEBUG("not enough space to create this object, so evicting objects"); /* Choose some objects to evict, and update the return pointers. */ int64_t num_bytes_evicted = choose_objects_to_evict( diff --git a/src/plasma/eviction_policy.h b/src/plasma/eviction_policy.h index 88412c80f..cc6497271 100644 --- a/src/plasma/eviction_policy.h +++ b/src/plasma/eviction_policy.h @@ -20,7 +20,7 @@ typedef struct eviction_state eviction_state; * store. * @return The internal state of the eviction policy. */ -eviction_state *make_eviction_state(int64_t system_memory); +eviction_state *make_eviction_state(void); /** * Free the eviction policy state. diff --git a/src/plasma/format/plasma.fbs b/src/plasma/format/plasma.fbs index 5b4402154..1e7120550 100644 --- a/src/plasma/format/plasma.fbs +++ b/src/plasma/format/plasma.fbs @@ -25,6 +25,9 @@ enum MessageType:int { // See if the store contains an object (will be deprecated). PlasmaContainsRequest, PlasmaContainsReply, + // Get information for a newly connecting client. + PlasmaConnectRequest, + PlasmaConnectReply, // Make room for new objects in the plasma store. PlasmaEvictRequest, PlasmaEvictReply, @@ -197,6 +200,18 @@ table PlasmaContainsReply { has_object: int; } +// PlasmaConnect is used by a plasma client the first time it connects with the +// store. This is not really necessary, but is used to get some information +// about the store such as its memory capacity. + +table PlasmaConnectRequest { +} + +table PlasmaConnectReply { + // The memory capacity of the store. + memory_capacity: long; +} + table PlasmaEvictRequest { // Number of bytes that shall be freed. num_bytes: ulong; diff --git a/src/plasma/plasma.h b/src/plasma/plasma.h index 51e7682f4..0ef7b2084 100644 --- a/src/plasma/plasma.h +++ b/src/plasma/plasma.h @@ -112,6 +112,9 @@ typedef struct { typedef struct { /** Objects that are in the Plasma store. */ object_table_entry *objects; + /** The amount of memory (in bytes) that we allow to be allocated in the + * store. */ + int64_t memory_capacity; } plasma_store_info; typedef struct { diff --git a/src/plasma/plasma/plasma.py b/src/plasma/plasma/plasma.py index 853cecb50..f65b088cc 100644 --- a/src/plasma/plasma/plasma.py +++ b/src/plasma/plasma/plasma.py @@ -43,6 +43,11 @@ class PlasmaBuffer(object): def __getitem__(self, index): """Read from the PlasmaBuffer as if it were just a regular buffer.""" + # We currently don't allow slicing plasma buffers. We should handle this + # better, but it requires some care because the slice may be backed by the + # same memory in the object store, but the original plasma buffer may go out + # of scope causing the memory to no longer be accessible. + assert not isinstance(index, slice) value = self.buffer[index] if sys.version_info >= (3, 0) and not isinstance(index, slice): value = chr(value) @@ -53,6 +58,11 @@ class PlasmaBuffer(object): This should fail because the buffer should be read only. """ + # We currently don't allow slicing plasma buffers. We should handle this + # better, but it requires some care because the slice may be backed by the + # same memory in the object store, but the original plasma buffer may go out + # of scope causing the memory to no longer be accessible. + assert not isinstance(index, slice) if sys.version_info >= (3, 0) and not isinstance(index, slice): value = ord(value) self.buffer[index] = value @@ -61,6 +71,18 @@ class PlasmaBuffer(object): """Return the length of the buffer.""" return len(self.buffer) +def buffers_equal(buff1, buff2): + """Compare two buffers. These buffers may be PlasmaBuffer objects. + + This method should only be used in the tests. We implement a special helper + method for doing this because doing comparisons by slicing is much faster, but + we don't want to expose slicing of PlasmaBuffer objects because it currently + is not safe. + """ + buff1_to_compare = buff1.buffer if isinstance(buff1, PlasmaBuffer) else buff1 + buff2_to_compare = buff2.buffer if isinstance(buff2, PlasmaBuffer) else buff2 + return buff1_to_compare[:] == buff2_to_compare[:] + class PlasmaClient(object): """The PlasmaClient is used to interface with a plasma store and a plasma manager. diff --git a/src/plasma/plasma_client.c b/src/plasma/plasma_client.c index 246a6724d..ada0d7df0 100644 --- a/src/plasma/plasma_client.c +++ b/src/plasma/plasma_client.c @@ -27,7 +27,7 @@ #include "plasma_client.h" #include "fling.h" #include "uthash.h" -#include "utringbuffer.h" +#include "utlist.h" #include "sha256.h" #define XXH_STATIC_LINKING_ONLY @@ -83,6 +83,17 @@ typedef struct { int release_delay; } plasma_client_config; +/** An element representing a pending release call in a doubly-linked list. This + * is used to implement the delayed release mechanism. */ +typedef struct pending_release { + /** The object_id of the released object. */ + object_id object_id; + /** Needed for the doubly-linked list macros. */ + struct pending_release *prev; + /** Needed for the doubly-linked list macros. */ + struct pending_release *next; +} pending_release; + /** 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 { @@ -103,13 +114,29 @@ struct plasma_connection { /** A hash table of the object IDs that are currently being used by this * client. */ object_in_use_entry *objects_in_use; - /** Object IDs of the last few release calls. This is used to delay releasing - * objects to see if they can be reused by subsequent tasks so we do not - * unneccessarily invalidate cpu caches. TODO(pcm): replace this with a - * proper lru cache of size sizeof(L3 cache). */ - UT_ringbuffer *release_history; + /** Object IDs of the last few release calls. This is a doubly-linked list and + * is used to delay releasing objects to see if they can be reused by + * subsequent tasks so we do not unneccessarily invalidate cpu caches. + * TODO(pcm): replace this with a proper lru cache using the size of the L3 + * cache. */ + pending_release *release_history; + /** This is the first element in the release_history doubly-linked list. It is + * an implementation detail used so we can pop from the front of the list. */ + pending_release *release_history_first_entry; + /** The length of the release_history doubly-linked list. This is an + * implementation detail. */ + int release_history_length; + /** The number of bytes in the combined objects that are held in the release + * history doubly-linked list. If this is too large then the client starts + * releasing objects. */ + int64_t in_use_object_bytes; /** Configuration options for the plasma client. */ plasma_client_config config; + /** The amount of memory available to the Plasma store. The client needs this + * information to make sure that it does not delay in releasing so much + * memory that the store is unable to evict enough objects to free up space. + */ + int64_t store_capacity; }; /* If the file descriptor fd has been mmapped in this client process before, @@ -176,6 +203,9 @@ void increment_object_count(plasma_connection *conn, HASH_FIND_INT(conn->mmap_table, &object->handle.store_fd, entry); CHECK(entry != NULL); CHECK(entry->count >= 0); + /* Update the in_use_object_bytes. */ + conn->in_use_object_bytes += + (object_entry->object.data_size + object_entry->object.metadata_size); entry->count += 1; } else { CHECK(object_entry->count > 0); @@ -228,6 +258,11 @@ bool plasma_create(plasma_connection *conn, * client is using. A call to plasma_release is required to decrement this * count. Cache the reference to the object. */ increment_object_count(conn, obj_id, &object, false); + /* We increment the count a second time (and the corresponding decrement will + * happen in a plasma_release call in plasma_seal) so even if the buffer + * returned by plasma_create goes out of scope, the object does not get + * released before the call to plasma_seal happens. */ + increment_object_count(conn, obj_id, &object, false); return true; } @@ -282,6 +317,17 @@ void plasma_get(plasma_connection *conn, increment_object_count(conn, obj_id, object, true); } +/** + * This is a helper method for implementing plasma_release. We maintain a buffer + * of release calls and only perform them once the buffer becomes full (as + * judged by the aggregate sizes of the objects). There may be multiple release + * calls for the same object ID in the buffer. In this case, the first release + * calls will not do anything. The client will only send a message to the store + * releasing the object when the client is truly done with the object. + * + * @param conn The plasma connection. + * @param object_id The object ID to attempt to release. + */ void plasma_perform_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 @@ -313,6 +359,10 @@ void plasma_perform_release(plasma_connection *conn, object_id object_id) { /* Tell the store that the client no longer needs the object. */ CHECK(plasma_send_ReleaseRequest(conn->store_conn, conn->builder, object_id) >= 0); + /* Update the in_use_object_bytes. */ + conn->in_use_object_bytes -= + (object_entry->object.data_size + object_entry->object.metadata_size); + DCHECK(conn->in_use_object_bytes >= 0); /* Remove the entry from the hash table of objects currently in use. */ HASH_DELETE(hh, conn->objects_in_use, object_entry); free(object_entry); @@ -320,20 +370,40 @@ void plasma_perform_release(plasma_connection *conn, object_id object_id) { } void plasma_release(plasma_connection *conn, object_id obj_id) { - /* If no ringbuffer is used, don't delay the release. */ - if (conn->config.release_delay == 0) { - plasma_perform_release(conn, obj_id); - } else if (!utringbuffer_full(conn->release_history)) { - /* Delay the release by storing new releases into a ringbuffer and only - * popping them off and actually releasing if the buffer is full. This is - * so consecutive tasks don't release and map again objects and invalidate - * the cpu cache this way. */ - utringbuffer_push_back(conn->release_history, &obj_id); - } else { - object_id object_id_to_release = - *(object_id *) utringbuffer_front(conn->release_history); - utringbuffer_push_back(conn->release_history, &obj_id); - plasma_perform_release(conn, object_id_to_release); + /* Add the new object to the release history. The corresponding call to free + * will occur in plasma_perform_release or in plasma_disconnect. */ + pending_release *pending_release_entry = malloc(sizeof(pending_release)); + pending_release_entry->object_id = obj_id; + DL_APPEND(conn->release_history, pending_release_entry); + conn->release_history_length += 1; + /* If the doubly-linked list was previously empty, update the pointer to the + * first element of the doubly-linked list. */ + if (conn->release_history_first_entry == NULL) { + conn->release_history_first_entry = pending_release_entry; + } + /* If there are too many bytes in use by the client or if there are too many + * pending release calls, and there are at least some pending release calls in + * the release_history list, then release some objects. TODO(rkn): Checking + * that conn->release_history_first_entry != NULL may be relying on an + * implementation detail that may or may not work as expected. */ + while ((conn->in_use_object_bytes > + MIN(L3_CACHE_SIZE_BYTES, conn->store_capacity / 100) || + conn->release_history_length > conn->config.release_delay) && + conn->release_history_length > 0) { + DCHECK(conn->release_history_first_entry != NULL); + /* Perform a release for the object ID for the first pending release. */ + plasma_perform_release(conn, conn->release_history_first_entry->object_id); + /* Remove the first entry from the doubly-linked list. */ + pending_release *new_first_entry = conn->release_history_first_entry->next; + DL_DELETE(conn->release_history, conn->release_history_first_entry); + free(conn->release_history_first_entry); + /* Update the first element in the doubly linked list. */ + conn->release_history_first_entry = new_first_entry; + conn->release_history_length -= 1; + DCHECK(conn->release_history_length >= 0); + } + if (conn->release_history_length == 0) { + DCHECK(conn->release_history_first_entry == NULL); } } @@ -402,6 +472,11 @@ void plasma_seal(plasma_connection *conn, object_id object_id) { CHECK(plasma_compute_object_hash(conn, object_id, &digest[0])); CHECK(plasma_send_SealRequest(conn->store_conn, conn->builder, object_id, &digest[0]) >= 0); + /* We call plasma_release to decrement the number of instances of this object + * that are currently being used by this client. The corresponding increment + * happened in plasma_create and was used to ensure that the object was not + * released before the call to plasma_seal. */ + plasma_release(conn, object_id); } void plasma_delete(plasma_connection *conn, object_id object_id) { @@ -482,17 +557,43 @@ plasma_connection *plasma_connect(const char *store_socket_name, result->mmap_table = NULL; result->objects_in_use = NULL; result->config.release_delay = release_delay; - utringbuffer_new(result->release_history, release_delay, &object_id_icd); + /* Initialize the release history doubly-linked list to NULL and also + * initialize other implementation details of the release history. */ + result->release_history = NULL; + result->release_history_first_entry = NULL; + result->release_history_length = 0; + result->in_use_object_bytes = 0; + /* Send a ConnectRequest to the store to get its memory capacity. */ + plasma_send_ConnectRequest(result->store_conn, result->builder); + uint8_t *reply_data = + plasma_receive(result->store_conn, MessageType_PlasmaConnectReply); + plasma_read_ConnectReply(reply_data, &result->store_capacity); + free(reply_data); return result; } void plasma_disconnect(plasma_connection *conn) { - object_id *id = NULL; - while ((id = (object_id *) utringbuffer_next(conn->release_history, id))) { - plasma_perform_release(conn, *id); + /* Perform the pending release calls to flush out the queue so that the counts + * in the objects_in_use table are accurate. */ + pending_release *element, *temp; + DL_FOREACH_SAFE(conn->release_history, element, temp) { + plasma_perform_release(conn, element->object_id); + DL_DELETE(conn->release_history, element); + free(element); } + /* Loop over the objects in use table and release all remaining objects. */ + object_in_use_entry *current_entry, *temp_entry; + HASH_ITER(hh, conn->objects_in_use, current_entry, temp_entry) { + object_id object_id_to_release = current_entry->object_id; + int count = current_entry->count; + for (int i = 0; i < count; ++i) { + plasma_perform_release(conn, object_id_to_release); + } + } + /* Check that we've successfully released everything. */ + CHECKM(conn->in_use_object_bytes == 0, "conn->in_use_object_bytes = %" PRId64, + conn->in_use_object_bytes); free_protocol_builder(conn->builder); - utringbuffer_free(conn->release_history); close(conn->store_conn); if (conn->manager_conn >= 0) { close(conn->manager_conn); diff --git a/src/plasma/plasma_client.h b/src/plasma/plasma_client.h index c440efdbc..0e446cd5f 100644 --- a/src/plasma/plasma_client.h +++ b/src/plasma/plasma_client.h @@ -7,6 +7,8 @@ #include "plasma.h" #define PLASMA_DEFAULT_RELEASE_DELAY 64 +/* Use 100MB as an overestimate of the L3 cache size. */ +#define L3_CACHE_SIZE_BYTES 100000000 typedef struct plasma_connection plasma_connection; diff --git a/src/plasma/plasma_manager.c b/src/plasma/plasma_manager.c index c29386966..33b25a650 100644 --- a/src/plasma/plasma_manager.c +++ b/src/plasma/plasma_manager.c @@ -728,9 +728,20 @@ void process_transfer_request(event_loop *loop, * forever if we don't end up sealing this object. */ /* The corresponding call to plasma_release will happen in * write_object_chunk. */ + /* TODO(rkn): The manager currently will block here if the object is not + * present in the store. This is completely unacceptable. The manager should + * do a non-blocking get call on the store, and if the object isn't there then + * perhaps the manager should initiate the transfer when it receives a + * notification from the store that the object is present. */ int has_obj; - plasma_contains(conn->manager_state->plasma_conn, object_id, &has_obj); - DCHECK(has_obj); + int counter = 0; + do { + plasma_contains(conn->manager_state->plasma_conn, object_id, &has_obj); + if (counter > 0) { + LOG_WARN("Blocking in the plasma manager."); + } + counter += 1; + } while (!has_obj); plasma_get(conn->manager_state->plasma_conn, object_id, &data_size, &data, &metadata_size, &metadata); assert(metadata == data + data_size); diff --git a/src/plasma/plasma_protocol.c b/src/plasma/plasma_protocol.c index d15630569..1c96fea9a 100644 --- a/src/plasma/plasma_protocol.c +++ b/src/plasma/plasma_protocol.c @@ -341,6 +341,34 @@ void plasma_read_ContainsReply(uint8_t *data, *has_object = PlasmaContainsReply_has_object(rep); } +/* Plasma connect message. */ + +int plasma_send_ConnectRequest(int sock, protocol_builder *B) { + PlasmaConnectRequest_start_as_root(B); + PlasmaConnectRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaConnectRequest); +} + +void plasma_read_ConnectRequest(uint8_t *data) { + DCHECK(data); + PlasmaConnectRequest_table_t req = PlasmaConnectRequest_as_root(data); +} + +int plasma_send_ConnectReply(int sock, + protocol_builder *B, + int64_t memory_capacity) { + PlasmaConnectReply_start_as_root(B); + PlasmaConnectReply_memory_capacity_add(B, memory_capacity); + PlasmaConnectReply_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaConnectReply); +} + +void plasma_read_ConnectReply(uint8_t *data, int64_t *memory_capacity) { + DCHECK(data); + PlasmaConnectReply_table_t rep = PlasmaConnectReply_as_root(data); + *memory_capacity = PlasmaConnectReply_memory_capacity(rep); +} + /* Plasma evict message. */ int plasma_send_EvictRequest(int sock, protocol_builder *B, int64_t num_bytes) { diff --git a/src/plasma/plasma_protocol.h b/src/plasma/plasma_protocol.h index b474c16a6..5bdbad9d8 100644 --- a/src/plasma/plasma_protocol.h +++ b/src/plasma/plasma_protocol.h @@ -187,6 +187,18 @@ void plasma_read_ContainsReply(uint8_t *data, object_id *object_id, int *has_object); +/* Plasma Connect message functions. */ + +int plasma_send_ConnectRequest(int sock, protocol_builder *B); + +void plasma_read_ConnectRequest(uint8_t *data); + +int plasma_send_ConnectReply(int sock, + protocol_builder *B, + int64_t memory_capacity); + +void plasma_read_ConnectReply(uint8_t *data, int64_t *memory_capacity); + /* Plasma Evict message functions (no reply so far). */ int plasma_send_EvictRequest(int sock, protocol_builder *B, int64_t num_bytes); diff --git a/src/plasma/plasma_store.c b/src/plasma/plasma_store.c index 04e806606..ccddacfd7 100644 --- a/src/plasma/plasma_store.c +++ b/src/plasma/plasma_store.c @@ -100,8 +100,9 @@ plasma_store_state *init_plasma_store(event_loop *loop, int64_t system_memory) { /* Initialize the plasma store info. */ state->plasma_store_info = malloc(sizeof(plasma_store_info)); state->plasma_store_info->objects = NULL; + state->plasma_store_info->memory_capacity = system_memory; /* Initialize the eviction state. */ - state->eviction_state = make_eviction_state(system_memory); + state->eviction_state = make_eviction_state(); utarray_new(state->input_buffer, &byte_icd); state->builder = make_protocol_builder(); return state; @@ -577,6 +578,11 @@ void process_message(event_loop *loop, case MessageType_PlasmaSubscribeRequest: subscribe_to_updates(client_context, client_sock); break; + case MessageType_PlasmaConnectRequest: + CHECK(plasma_send_ConnectReply(client_sock, state->builder, + state->plasma_store_info->memory_capacity) >= + 0); + break; case DISCONNECT_CLIENT: { LOG_DEBUG("Disconnecting client on fd %d", client_sock); event_loop_remove_file(loop, client_sock); diff --git a/src/plasma/test/manager_tests.c b/src/plasma/test/manager_tests.c index 04aecfd37..1e34473cb 100644 --- a/src/plasma/test/manager_tests.c +++ b/src/plasma/test/manager_tests.c @@ -19,7 +19,8 @@ SUITE(plasma_manager_tests); -const char *plasma_socket_name_format = "/tmp/plasma_socket_%d"; +const char *plasma_store_socket_name = "/tmp/plasma_store_socket_1"; +const char *plasma_manager_socket_name_format = "/tmp/plasma_manager_socket_%d"; const char *manager_addr = "127.0.0.1"; object_id oid; @@ -59,14 +60,13 @@ plasma_mock *init_plasma_mock(plasma_mock *remote_mock) { plasma_mock *mock = malloc(sizeof(plasma_mock)); /* Start listening on all the ports and initiate the local plasma manager. */ mock->port = bind_inet_sock_retry(&mock->manager_remote_fd); - UT_string *store_socket_name = - bind_ipc_sock_retry(plasma_socket_name_format, &mock->local_store); - UT_string *manager_socket_name = - bind_ipc_sock_retry(plasma_socket_name_format, &mock->manager_local_fd); + mock->local_store = socket_connect_retry(plasma_store_socket_name, 5, 100); + UT_string *manager_socket_name = bind_ipc_sock_retry( + plasma_manager_socket_name_format, &mock->manager_local_fd); CHECK(mock->manager_local_fd >= 0 && mock->local_store >= 0); - mock->state = init_plasma_manager_state(utstring_body(store_socket_name), + mock->state = init_plasma_manager_state(plasma_store_socket_name, utstring_body(manager_socket_name), manager_addr, mock->port, NULL, 0); mock->loop = get_event_loop(mock->state); @@ -84,12 +84,11 @@ plasma_mock *init_plasma_mock(plasma_mock *remote_mock) { } /* Connect a new client to the local plasma manager and mock a request to an * object. */ - mock->plasma_conn = plasma_connect(utstring_body(store_socket_name), + mock->plasma_conn = plasma_connect(plasma_store_socket_name, utstring_body(manager_socket_name), 0); wait_for_pollin(mock->manager_local_fd); mock->client_conn = new_client_connection(mock->loop, mock->manager_local_fd, mock->state, 0); - utstring_free(store_socket_name); utstring_free(manager_socket_name); return mock; } diff --git a/src/plasma/test/test.py b/src/plasma/test/test.py index 13631e611..2513413ad 100644 --- a/src/plasma/test/test.py +++ b/src/plasma/test/test.py @@ -22,13 +22,22 @@ USE_VALGRIND = False PLASMA_STORE_MEMORY = 1000000000 def assert_get_object_equal(unit_test, client1, client2, object_id, memory_buffer=None, metadata=None): + client1_buff = client1.get(object_id) + client2_buff = client2.get(object_id) + client1_metadata = client1.get_metadata(object_id) + client2_metadata = client2.get_metadata(object_id) + unit_test.assertEqual(len(client1_buff), len(client2_buff)) + unit_test.assertEqual(len(client1_metadata), len(client2_metadata)) + # Check that the buffers from the two clients are the same. + unit_test.assertTrue(plasma.buffers_equal(client1_buff, client2_buff)) + # Check that the metadata buffers from the two clients are the same. + unit_test.assertTrue(plasma.buffers_equal(client1_metadata, client2_metadata)) + # If a reference buffer was provided, check that it is the same as well. if memory_buffer is not None: - unit_test.assertEqual(memory_buffer[:], client2.get(object_id)[:]) + unit_test.assertTrue(plasma.buffers_equal(memory_buffer, client1_buff)) + # If reference metadata was provided, check that it is the same as well. 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)[:]) + unit_test.assertTrue(plasma.buffers_equal(metadata, client1_metadata)) class TestPlasmaClient(unittest.TestCase): diff --git a/test/microbenchmarks.py b/test/microbenchmarks.py index 061fb46da..974803591 100644 --- a/test/microbenchmarks.py +++ b/test/microbenchmarks.py @@ -2,6 +2,7 @@ from __future__ import absolute_import from __future__ import division from __future__ import print_function +import os import unittest import ray import sys @@ -89,8 +90,8 @@ class MicroBenchmarkTest(unittest.TestCase): def testCache(self): ray.init(start_ray_local=True, num_workers=1) - A = np.random.rand(5000, 5000) - v = np.random.rand(5000) + A = np.random.rand(1, 1000000) + v = np.random.rand(1000000) A_id = ray.put(A) v_id = ray.put(v) a = time.time() @@ -101,7 +102,12 @@ class MicroBenchmarkTest(unittest.TestCase): for i in range(100): ray.get(A_id).dot(ray.get(v_id)) d = time.time() - c - self.assertLess(d, 1.5 * b) + + if d > 1.5 * b: + if os.getenv("TRAVIS") is None: + raise Exception("The caching test was too slow. d = {}, b = {}".format(d, b)) + else: + print("WARNING: The caching test was too slow. d = {}, b = {}".format(d, b)) ray.worker.cleanup()