diff --git a/src/common/state/redis.c b/src/common/state/redis.c index 951b660d4..a411466c7 100644 --- a/src/common/state/redis.c +++ b/src/common/state/redis.c @@ -364,7 +364,10 @@ void redis_object_table_get_entry(redisAsyncContext *c, int64_t manager_count = reply->elements; if (reply->type == REDIS_REPLY_ARRAY) { - const char **manager_vector = malloc(manager_count * sizeof(char *)); + const char **manager_vector = NULL; + if (manager_count > 0) { + manager_vector = malloc(manager_count * sizeof(char *)); + } for (int j = 0; j < reply->elements; ++j) { CHECK(reply->element[j]->type == REDIS_REPLY_STRING); memcpy(managers[j].id, reply->element[j]->str, sizeof(managers[j].id)); @@ -377,10 +380,13 @@ void redis_object_table_get_entry(redisAsyncContext *c, callback_data->user_context); /* remove timer */ destroy_timer_callback(callback_data->db_handle->loop, callback_data); - free(managers); + if (manager_count > 0) { + free(manager_vector); + } } else { LOG_FATAL("expected integer or string, received type %d", reply->type); } + free(managers); } void redis_object_table_subscribe_lookup(redisAsyncContext *c, diff --git a/src/common/test/db_tests.c b/src/common/test/db_tests.c index 5b818f152..8b4594e31 100644 --- a/src/common/test/db_tests.c +++ b/src/common/test/db_tests.c @@ -47,7 +47,6 @@ void lookup_done_callback(object_id object_id, received_port2) != 2) { CHECK(0); } - free(manager_vector); } /* Entry added to database successfully. */ diff --git a/src/common/test/object_table_tests.c b/src/common/test/object_table_tests.c index 9196e2ba6..365df0caa 100644 --- a/src/common/test/object_table_tests.c +++ b/src/common/test/object_table_tests.c @@ -290,7 +290,6 @@ void lookup_retry_done_callback(object_id object_id, void *context) { CHECK(context == (void *) lookup_retry_context); lookup_retry_succeeded = 1; - free(manager_vector); } void lookup_retry_fail_callback(unique_id id, diff --git a/src/plasma/lib/python/plasma.py b/src/plasma/lib/python/plasma.py index 7ff8c4200..71de34e46 100644 --- a/src/plasma/lib/python/plasma.py +++ b/src/plasma/lib/python/plasma.py @@ -181,6 +181,14 @@ class PlasmaClient(object): """ return libplasma.fetch(self.conn, object_ids) + def fetch2(self, object_ids): + """Fetch the objects with the given IDs from other plasma manager instances. + + Args: + object_ids (List[str]): A list of strings used to identify the objects. + """ + return libplasma.fetch2(self.conn, object_ids) + def wait(self, object_ids, timeout=PLASMA_WAIT_TIMEOUT, num_returns=1): """Wait until num_returns objects in object_ids are ready. diff --git a/src/plasma/plasma.h b/src/plasma/plasma.h index 25586a71f..dd8e1f05e 100644 --- a/src/plasma/plasma.h +++ b/src/plasma/plasma.h @@ -129,6 +129,8 @@ enum plasma_message_type { PLASMA_FETCH_REMOTE, /** Request a fetch of an object in another store. Blocking call. */ PLASMA_FETCH, + /** Request a fetch of an object in another store. Non-blocking call. */ + PLASMA_FETCH2, /** Request status of an object, i.e., whether the object is stored in the * local Plasma Store, in a remote Plasma Store, in transfer, or doesn't * exist in the system. */ diff --git a/src/plasma/plasma_client.c b/src/plasma/plasma_client.c index ff0d5db87..c1ba67d83 100644 --- a/src/plasma/plasma_client.c +++ b/src/plasma/plasma_client.c @@ -544,6 +544,18 @@ void plasma_fetch(plasma_connection *conn, } } +void plasma_fetch2(plasma_connection *conn, + int num_object_ids, + object_id object_ids[]) { + CHECK(conn != NULL); + CHECK(conn->manager_conn >= 0); + plasma_request *req = plasma_alloc_request(num_object_ids); + for (int i = 0; i < num_object_ids; ++i) { + req->object_requests[i].object_id = object_ids[i]; + } + CHECK(plasma_send_request(conn->manager_conn, PLASMA_FETCH2, req) >= 0); +} + int plasma_wait(plasma_connection *conn, int num_object_ids, object_id object_ids[], diff --git a/src/plasma/plasma_client.h b/src/plasma/plasma_client.h index 65dc2826d..605c10da2 100644 --- a/src/plasma/plasma_client.h +++ b/src/plasma/plasma_client.h @@ -190,6 +190,34 @@ void plasma_fetch(plasma_connection *conn, object_id object_ids[], int is_fetched[]); +/** + * Attempt to initiate the transfer of some objects from remote Plasma Stores. + * + * For an object that is available in the local Plasma Store, this method will + * not do anything. For an object that is not available locally, it will check + * if the object are already being fetched. If so, it will not do anything. If + * not, it will query the object table for a list of Plasma Managers that have + * the object. If that list is non-empty, it will attempt to initiate transfers + * from one of those Plasma Managers. If the list is empty, it will set a + * callback to initiate a transfer when the list becomes non-empty. + * + * TODO(rkn): Setting the callback for when the list becomes non-empty is not + * implemented. + * + * This function is non-blocking. + * + * This method is idempotent in the sense that it is ok to call it multiple + * times. + * + * @param conn The object containing the connection state. + * @param num_object_ids The number of object IDs fetch is being called on. + * @param object_ids The IDs of the objects that fetch is being called on. + * @return Void. + */ +void plasma_fetch2(plasma_connection *conn, + int num_object_ids, + object_id object_ids[]); + /** * Transfer local object to a different plasma manager. * diff --git a/src/plasma/plasma_extension.c b/src/plasma/plasma_extension.c index 441c7a146..434ef8140 100644 --- a/src/plasma/plasma_extension.c +++ b/src/plasma/plasma_extension.c @@ -170,6 +170,27 @@ PyObject *PyPlasma_fetch(PyObject *self, PyObject *args) { return success_list; } +PyObject *PyPlasma_fetch2(PyObject *self, PyObject *args) { + plasma_connection *conn; + PyObject *object_id_list; + if (!PyArg_ParseTuple(args, "O&O", PyObjectToPlasmaConnection, &conn, + &object_id_list)) { + return NULL; + } + if (!plasma_manager_is_connected(conn)) { + PyErr_SetString(PyExc_RuntimeError, "Not connected to the plasma manager"); + return NULL; + } + Py_ssize_t n = PyList_Size(object_id_list); + object_id *object_ids = malloc(sizeof(object_id) * n); + for (int i = 0; i < n; ++i) { + PyObjectToUniqueID(PyList_GetItem(object_id_list, i), &object_ids[i]); + } + plasma_fetch2(conn, (int) n, object_ids); + free(object_ids); + Py_RETURN_NONE; +} + PyObject *PyPlasma_wait(PyObject *self, PyObject *args) { plasma_connection *conn; PyObject *object_id_list; @@ -292,6 +313,8 @@ static PyMethodDef plasma_methods[] = { "Does the plasma store contain this plasma object?"}, {"fetch", PyPlasma_fetch, METH_VARARGS, "Fetch the object from another plasma manager instance."}, + {"fetch2", PyPlasma_fetch2, METH_VARARGS, + "Fetch the object from another plasma manager instance."}, {"wait", PyPlasma_wait, METH_VARARGS, "Wait until num_returns objects in object_ids are ready."}, {"evict", PyPlasma_evict, METH_VARARGS, diff --git a/src/plasma/plasma_manager.c b/src/plasma/plasma_manager.c index 3d12e9e9c..bdf64acaf 100644 --- a/src/plasma/plasma_manager.c +++ b/src/plasma/plasma_manager.c @@ -140,6 +140,31 @@ typedef struct { UT_hash_handle hh; } available_object; +typedef struct { + /** The ID of the object we are fetching or waiting for. */ + object_id object_id; + /** The plasma manager state. */ + plasma_manager_state *manager_state; + /** 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; + /** Pointer to the array containing the manager locations of this object. This + * struct owns and must free each entry. */ + char **manager_vector; + /** The number of manager locations in the array manager_vector. */ + int manager_count; + /** The next manager we should try to contact. This is set to an index in + * manager_vector in the retry handler, in case the current attempt fails to + * contact a manager. */ + int next_manager; + /** Handle for the uthash table in the manager state that keeps track of + * outstanding fetch requests. */ + UT_hash_handle hh; +} fetch_request2; + struct plasma_manager_state { /** Event loop. */ event_loop *loop; @@ -158,6 +183,9 @@ struct plasma_manager_state { * object id, value is a list of connections to the clients * who are blocking on a fetch of this object. */ client_object_request *fetch_requests; + /** Hash table of outstanding fetch requests. The key is the object ID. The + * value is the data needed to perform the fetch. */ + fetch_request2 *fetch_requests2; /** Initialize an empty hash map for the cache of local available object. */ available_object *local_available_objects; }; @@ -347,6 +375,24 @@ void remove_object_request(client_connection *client_conn, free_client_object_request(object_req); } +void remove_fetch_request(plasma_manager_state *manager_state, + fetch_request2 *fetch_req) { + /* Remove the fetch request from the table of fetch requests. */ + HASH_DELETE(hh, manager_state->fetch_requests2, fetch_req); + /* Remove the timer associated with this fetch request. */ + if (fetch_req->timer != -1) { + event_loop_remove_timer(manager_state->loop, fetch_req->timer); + } + /* Free the fetch request and everything in it. */ + for (int i = 0; i < fetch_req->manager_count; ++i) { + free(fetch_req->manager_vector[i]); + } + if (fetch_req->manager_vector != NULL) { + free(fetch_req->manager_vector); + } + free(fetch_req); +} + plasma_manager_state *init_plasma_manager_state(const char *store_socket_name, const char *manager_addr, int manager_port, @@ -358,6 +404,7 @@ plasma_manager_state *init_plasma_manager_state(const char *store_socket_name, plasma_connect(store_socket_name, NULL, PLASMA_DEFAULT_RELEASE_DELAY); state->manager_connections = NULL; state->fetch_requests = NULL; + state->fetch_requests2 = NULL; if (db_addr) { state->db = db_connect(db_addr, db_port, "plasma_manager", manager_addr, manager_port); @@ -402,6 +449,13 @@ void destroy_plasma_manager_state(plasma_manager_state *state) { } } + if (state->fetch_requests2 != NULL) { + fetch_request2 *fetch_req, *tmp; + HASH_ITER(hh, state->fetch_requests2, fetch_req, tmp) { + remove_fetch_request(fetch_req->manager_state, fetch_req); + } + } + plasma_disconnect(state->plasma_conn); event_loop_destroy(state->loop); free(state); @@ -565,9 +619,6 @@ void ignore_data_chunk(event_loop *loop, } free(buf->data); - if (buf->metadata) { - free(buf->metadata); - } free(buf); /* Switch to listening for requests from this socket, instead of reading * object data. */ @@ -699,12 +750,7 @@ void process_data_request(event_loop *loop, * for data and metadata, if needed. All memory associated with * buf/g_ignore_buf will be freed in ignore_data_chunkc(). */ conn->ignore_buffer = buf; - buf->data = (uint8_t *) malloc(buf->data_size); - if (buf->metadata_size > 0) { - buf->metadata = (uint8_t *) malloc(buf->metadata_size); - } else { - buf->metadata = NULL; - } + buf->data = (uint8_t *) malloc(buf->data_size + buf->metadata_size); event_loop_add_file(loop, client_sock, EVENT_LOOP_READ, ignore_data_chunk, conn); } @@ -743,6 +789,43 @@ void request_transfer_from(client_connection *client_conn, object_req->next_manager %= object_req->manager_count; } +void request_transfer_from2(plasma_manager_state *manager_state, + object_id object_id) { + fetch_request2 *fetch_req; + HASH_FIND(hh, manager_state->fetch_requests2, &object_id, sizeof(object_id), + fetch_req); + /* TODO(rkn): This probably can be NULL so we should remove this check, and + * instead return in the case where there is no fetch request. */ + CHECK(fetch_req != NULL); + + CHECK(fetch_req->manager_count > 0); + CHECK(fetch_req->next_manager >= 0 && + fetch_req->next_manager < fetch_req->manager_count); + char addr[16]; + int port; + parse_ip_addr_port(fetch_req->manager_vector[fetch_req->next_manager], addr, + &port); + + client_connection *manager_conn = + get_manager_connection(manager_state, addr, port); + plasma_request_buffer *transfer_request = + malloc(sizeof(plasma_request_buffer)); + transfer_request->type = PLASMA_TRANSFER; + transfer_request->object_id = fetch_req->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(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); + /* On the next attempt, try the next manager in manager_vector. */ + fetch_req->next_manager += 1; + fetch_req->next_manager %= fetch_req->manager_count; +} + int manager_timeout_handler(event_loop *loop, timer_id id, void *context) { client_object_request *object_req = context; client_connection *client_conn = object_req->client_conn; @@ -759,6 +842,28 @@ int manager_timeout_handler(event_loop *loop, timer_id id, void *context) { return EVENT_LOOP_TIMER_DONE; } +int manager_timeout_handler2(event_loop *loop, timer_id id, void *context) { + fetch_request2 *fetch_req = context; + plasma_manager_state *manager_state = fetch_req->manager_state; + LOG_DEBUG("Timer went off, %d tries left", fetch_req->num_retries); + if (fetch_req->num_retries > 0) { + request_transfer_from2(manager_state, fetch_req->object_id); + fetch_req->num_retries--; + return MANAGER_TIMEOUT; + } + /* TODO(rkn): This shouldn't be fatal. Instead, it should do nothing. */ + CHECK(0); + remove_fetch_request(manager_state, fetch_req); + return EVENT_LOOP_TIMER_DONE; +} + +bool is_object_local(plasma_manager_state *state, object_id object_id) { + available_object *entry; + HASH_FIND(hh, state->local_available_objects, &object_id, sizeof(object_id), + entry); + return entry != NULL; +} + /* TODO(swang): Consolidate transfer requests for same object * from different client IDs by passing in manager state, not * client context. */ @@ -779,7 +884,6 @@ void request_transfer(object_id object_id, 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 = plasma_make_reply(object_id); reply.object_status = PLASMA_OBJECT_NONEXISTENT; CHECK(plasma_send_reply(client_conn->fd, &reply) >= 0); @@ -801,7 +905,6 @@ void request_transfer(object_id object_id, strncpy(object_req->manager_vector[i], manager_vector[i], len); object_req->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_req->num_retries = NUM_RETRIES; @@ -811,11 +914,50 @@ void request_transfer(object_id object_id, request_transfer_from(client_conn, object_id); } -bool is_object_local(plasma_manager_state *state, object_id object_id) { - available_object *entry; - HASH_FIND(hh, state->local_available_objects, &object_id, sizeof(object_id), - entry); - return entry != NULL; +void request_transfer2(object_id object_id, + int manager_count, + const char *manager_vector[], + void *context) { + plasma_manager_state *manager_state = (plasma_manager_state *) context; + fetch_request2 *fetch_req; + HASH_FIND(hh, manager_state->fetch_requests2, &object_id, sizeof(object_id), + fetch_req); + + if (is_object_local(manager_state, object_id)) { + /* If the object is already here, then the fetch request should have been + * removed. */ + CHECK(fetch_req == NULL); + return; + } + + /* If the object is not present, then the fetch request should still be here. + * TODO(rkn): We actually have to remove this check to handle the rare + * scenario where the object is transferred here and then evicted before this + * callback gets called. */ + CHECK(fetch_req != NULL); + + if (manager_count == 0) { + /* TODO(rkn): Figure out what to do in this case. */ + remove_fetch_request(manager_state, fetch_req); + return; + } + /* Pick a different manager to request a transfer from on every attempt. */ + fetch_req->manager_count = manager_count; + fetch_req->manager_vector = malloc(manager_count * sizeof(char *)); + fetch_req->next_manager = 0; + memset(fetch_req->manager_vector, 0, manager_count * sizeof(char *)); + for (int i = 0; i < manager_count; ++i) { + int len = strlen(manager_vector[i]); + fetch_req->manager_vector[i] = malloc(len + 1); + strncpy(fetch_req->manager_vector[i], manager_vector[i], len); + fetch_req->manager_vector[i][len] = '\0'; + } + /* Wait for the object data for the default number of retries, which timeout + * after a default interval. */ + request_transfer_from2(manager_state, object_id); + fetch_req->num_retries = NUM_RETRIES; + fetch_req->timer = event_loop_add_timer(manager_state->loop, MANAGER_TIMEOUT, + manager_timeout_handler2, fetch_req); } void process_fetch_request(client_connection *client_conn, @@ -854,6 +996,56 @@ void process_fetch_requests(client_connection *client_conn, } } +void fatal_table_callback(object_id id, void *user_context, void *user_data) { + CHECK(0); +} + +void process_fetch_requests2(client_connection *client_conn, + int num_object_ids, + object_request object_requests[]) { + plasma_manager_state *manager_state = client_conn->manager_state; + for (int i = 0; i < num_object_ids; ++i) { + object_id obj_id = object_requests[i].object_id; + + /* Check if this object is already present locally. If so, do nothing. */ + if (is_object_local(manager_state, obj_id)) { + continue; + } + + /* Check if this object is already being fetched. If so, do nothing. */ + fetch_request2 *entry; + HASH_FIND(hh, manager_state->fetch_requests2, &obj_id, sizeof(obj_id), + entry); + if (entry != NULL) { + continue; + } + + /* Add an entry to the fetch requests data structure to indidate that the + * object is being fetched. */ + entry = malloc(sizeof(fetch_request2)); + entry->manager_state = manager_state; + entry->object_id = obj_id; + entry->timer = -1; + entry->manager_count = 0; + entry->manager_vector = NULL; + HASH_ADD(hh, manager_state->fetch_requests2, object_id, + sizeof(entry->object_id), entry); + + /* Get a list of Plasma Managers that have this object from the object + * table. If the list of Plasma Managers is non-empty, the callback should + * initiate a transfer. */ + /* TODO(rkn): Make sure this also handles the case where the list is + * initially empty. */ + retry_info retry; + memset(&retry, 0, sizeof(retry)); + retry.num_retries = NUM_RETRIES; + retry.timeout = MANAGER_TIMEOUT; + retry.fail_callback = fatal_table_callback; + object_table_lookup(manager_state->db, obj_id, &retry, request_transfer2, + manager_state); + } +} + void return_from_wait(client_connection *client_conn) { CHECK(client_conn->is_wait); /* TODO: check for wait1. */ @@ -1199,7 +1391,6 @@ int request_fetch_or_status(object_id object_id, /* If the object isn't on any managers, report a failure to the client. */ LOG_DEBUG("Object is on %d managers", manager_count); if (manager_count == 0) { - free(manager_vector); if (object_req) { remove_object_request(client_conn, object_req); } @@ -1221,7 +1412,6 @@ int request_fetch_or_status(object_id object_id, strncpy(object_req->manager_vector[i], manager_vector[i], len); object_req->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_req->num_retries = NUM_RETRIES; @@ -1320,6 +1510,12 @@ void process_object_notification(event_loop *loop, entry->object_id = obj_id; HASH_ADD(hh, state->local_available_objects, object_id, sizeof(object_id), entry); + /* If we were trying to fetch this object, finish up the fetch request. */ + fetch_request2 *fetch_req; + HASH_FIND(hh, state->fetch_requests2, &obj_id, sizeof(obj_id), fetch_req); + if (fetch_req != NULL) { + remove_fetch_request(state, fetch_req); + } /* Notify any clients who were waiting on a fetch to this object and tick * off objects we are waiting for. */ client_object_request *object_req, *next; @@ -1393,6 +1589,10 @@ void process_message(event_loop *loop, process_fetch_or_status_request(conn, req->object_requests[0].object_id, true); break; + case PLASMA_FETCH2: + LOG_DEBUG("Processing fetch remote"); + process_fetch_requests2(conn, req->num_object_ids, req->object_requests); + break; case PLASMA_WAIT: LOG_DEBUG("Processing wait"); process_wait_request(conn, req->num_object_ids, req->object_requests, diff --git a/src/plasma/test/manager_tests.c b/src/plasma/test/manager_tests.c index 8f03a6ba4..af384c1c8 100644 --- a/src/plasma/test/manager_tests.c +++ b/src/plasma/test/manager_tests.c @@ -158,6 +158,7 @@ TEST request_transfer_test(void) { utstring_printf(addr, "127.0.0.1:%d", remote_mock->port); manager_vector[0] = utstring_body(addr); request_transfer(oid, 1, manager_vector, local_mock->client_conn); + free(manager_vector); event_loop_add_timer(local_mock->loop, MANAGER_TIMEOUT, test_done_handler, local_mock->state); event_loop_run(local_mock->loop); @@ -203,6 +204,7 @@ TEST request_transfer_retry_test(void) { utstring_printf(addr1, "127.0.0.1:%d", remote_mock2->port); manager_vector[1] = utstring_body(addr1); request_transfer(oid, 2, manager_vector, local_mock->client_conn); + free(manager_vector); event_loop_add_timer(local_mock->loop, MANAGER_TIMEOUT * 2, test_done_handler, local_mock->state); event_loop_run(local_mock->loop); @@ -245,6 +247,7 @@ TEST request_transfer_timeout_test(void) { utstring_printf(addr, "127.0.0.1:%d", remote_mock->port); manager_vector[0] = utstring_body(addr); request_transfer(oid, 1, manager_vector, local_mock->client_conn); + free(manager_vector); event_loop_add_timer(local_mock->loop, MANAGER_TIMEOUT * (NUM_RETRIES + 2), test_done_handler, local_mock->state); event_loop_run(local_mock->loop); diff --git a/src/plasma/test/test.py b/src/plasma/test/test.py index 7e916f14c..523e6826a 100644 --- a/src/plasma/test/test.py +++ b/src/plasma/test/test.py @@ -37,13 +37,17 @@ def write_to_data_buffer(buff, length): 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() +def create_object_with_id(client, object_id, data_size, metadata_size, seal=True): 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 memory_buffer, metadata + +def create_object(client, data_size, metadata_size, seal=True): + object_id = random_object_id() + memory_buffer, metadata = create_object_with_id(client, object_id, data_size, metadata_size, seal=seal) return object_id, memory_buffer, metadata def assert_get_object_equal(unit_test, client1, client2, object_id, memory_buffer=None, metadata=None): @@ -350,6 +354,90 @@ class TestPlasmaManager(unittest.TestCase): object_id = random_object_id() self.assertRaises(Exception, lambda : self.client1.fetch([object_id, object_id])) + def test_fetch2(self): + if self.redis_process is None: + print("Cannot test fetch without a running redis instance.") + self.assertTrue(False) + for _ in range(10): + # Create an object. + object_id1, memory_buffer1, metadata1 = create_object(self.client1, 2000, 2000) + self.client1.fetch2([object_id1]) + self.assertEqual(self.client1.contains(object_id1), True) + self.assertEqual(self.client2.contains(object_id1), False) + # Fetch the object from the other plasma manager. + # TODO(rkn): Right now we must wait for the object table to be updated. + while not self.client2.contains(object_id1): + self.client2.fetch2([object_id1]) + # Compare the two buffers. + assert_get_object_equal(self, self.client1, self.client2, object_id1, + memory_buffer=memory_buffer1, metadata=metadata1) + + # Test that we can call fetch on object IDs that don't exist yet. + object_id2 = random_object_id() + self.client1.fetch2([object_id2]) + self.assertEqual(self.client1.contains(object_id2), False) + memory_buffer2, metadata2 = create_object_with_id(self.client2, object_id2, 2000, 2000) + # # Check that the object has been fetched. + # self.assertEqual(self.client1.contains(object_id2), True) + # Compare the two buffers. + # assert_get_object_equal(self, self.client1, self.client2, object_id2, + # memory_buffer=memory_buffer2, metadata=metadata2) + + # Test calling the same fetch request a bunch of times. + object_id3 = random_object_id() + self.assertEqual(self.client1.contains(object_id3), False) + self.assertEqual(self.client2.contains(object_id3), False) + for _ in range(10): + self.client1.fetch2([object_id3]) + self.client2.fetch2([object_id3]) + memory_buffer3, metadata3 = create_object_with_id(self.client1, object_id3, 2000, 2000) + for _ in range(10): + self.client1.fetch2([object_id3]) + self.client2.fetch2([object_id3]) + #TODO(rkn): Right now we must wait for the object table to be updated. + while not self.client2.contains(object_id3): + self.client2.fetch2([object_id3]) + assert_get_object_equal(self, self.client1, self.client2, object_id3, + memory_buffer=memory_buffer3, metadata=metadata3) + + def test_fetch2_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(rkn): Right now we must wait for the object table to be updated. + while (not self.client2.contains(object_id1)) or (not self.client2.contains(object_id2)): + self.client2.fetch2(object_ids) + # 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. + self.client1.fetch2(object_ids) + 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) + + # Check that we can call fetch with duplicated object IDs. + object_id3 = random_object_id() + self.client1.fetch2([object_id3, object_id3]) + object_id4, memory_buffer4, metadata4 = create_object(self.client1, 2000, 2000) + time.sleep(0.1) + # TODO(rkn): Right now we must wait for the object table to be updated. + while not self.client2.contains(object_id4): + self.client2.fetch2([object_id3, object_id3, object_id4, object_id4]) + assert_get_object_equal(self, self.client2, self.client1, object_id4, + memory_buffer=memory_buffer4, metadata=metadata4) + def test_wait(self): # Test timeout. obj_id0 = random_object_id()