Non-blocking fetch implementation. (#83)

* Non-blocking fetch implementation.

* Make fetch tests more robust to timing issues.

* Bug fix when ignoring transferred objects.

* Fix.

* Documentation fixes.

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,

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. */

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,

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.
 

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. */

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[],

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.
  *

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,

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,

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);

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()