Merge pull request #7 from pcmoritz/manager

Object store manager

Makefile

@@ -2,7 +2,7 @@ CC = gcc
 CFLAGS = -g -Wall
 BUILD = build
 
-all: $(BUILD)/plasma_store $(BUILD)/plasma_client.so $(BUILD)/example
+all: $(BUILD)/plasma_store $(BUILD)/plasma_manager $(BUILD)/plasma_client.so $(BUILD)/example
 
 clean:
 	rm -r $(BUILD)/*
@@ -10,6 +10,9 @@ clean:
 $(BUILD)/plasma_store: src/plasma_store.c src/plasma.h src/fling.h src/fling.c
 	$(CC) $(CFLAGS) --std=c99 -D_XOPEN_SOURCE=500 src/plasma_store.c src/fling.c -o $(BUILD)/plasma_store
 
+$(BUILD)/plasma_manager: src/plasma_manager.c src/plasma.h src/plasma_client.c src/fling.h src/fling.c
+	$(CC) $(CFLAGS) --std=c99 -D_XOPEN_SOURCE=500 src/plasma_manager.c src/plasma_client.c src/fling.c -o $(BUILD)/plasma_manager
+
 $(BUILD)/plasma_client.so: src/plasma_client.c src/fling.h src/fling.c
 	$(CC) $(CFLAGS) --std=c99 -D_XOPEN_SOURCE=500 src/plasma_client.c src/fling.c -fPIC -shared -o $(BUILD)/plasma_client.so
 

lib/python/plasma.py

@@ -9,12 +9,16 @@ ID = ctypes.c_ubyte * 20
 class PlasmaID(ctypes.Structure):
   _fields_ = [("plasma_id", ID)]
 
-# these must be in sync with plasma_request_type in plasma.h
+# these must be in sync with plasma_request_type in plasma.h (can we have a test for that?)
 PLASMA_CREATE = 0
+PLASMA_GET = 1
+PLASMA_SEAL = 2
+PLASMA_TRANSFER = 3
+PLASMA_DATA = 4
+PLASMA_REGISTER = 5
 
 class PlasmaRequest(ctypes.Structure):
   _fields_ = [("type", ctypes.c_int),
-              ("manager_id", ctypes.c_int),
               ("object_id", PlasmaID),
               ("size", ctypes.c_int64),
               ("addr", Addr),
@@ -32,7 +36,39 @@ def make_plasma_id(string):
   object_id = map(ord, string)
   return PlasmaID(plasma_id=ID(*object_id))
 
+class PlasmaManager(object):
+  """The PlasmaManager is used to manage a PlasmaStore.
+
+  There should be one PlasmaManager per PlasmaStore. The PlasmaManager is
+  responsible for interfacing with other PlasmaManagers in order to transfer
+  objects between PlasmaStores. This class sends commands to the C
+  implementation of the PlasmaManager using sockets.
+
+  Attributes:
+    sock: The socket used to communicate with the C implementation of the
+      PlasmaManager.
+  """
+
+  def __init__(self, addr, port):
+    """Initialize the PlasmaManager."""
+    self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    self.sock.connect((addr, port))
+
+  def transfer(self, addr, port, object_id):
+    """Transfer local object with id object_id to manager with id manager_id."""
+    req = PlasmaRequest(type=PLASMA_TRANSFER, object_id=make_plasma_id(object_id),
+                        addr=Addr(*map(int, addr.split("."))), port=port)
+    print "sending port", port
+    self.sock.send(buffer(req)[:])
+
 class PlasmaClient(object):
+  """The PlasmaClient is used to interface with a PlasmaStore.
+
+  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):
     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)
@@ -59,12 +95,36 @@ class PlasmaClient(object):
     self.sock = self.client.plasma_store_connect(socket_name)
 
   def create(self, object_id, size):
+    """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.
+    """
     buf = self.client.plasma_create(self.sock, make_plasma_id(object_id), size)
     return self.buffer_from_read_write_memory(buf.data, buf.size)
 
   def get(self, object_id):
+    """Create a buffer from the PlasmaStore based on object ID.
+
+    This method can only be called after the buffer has been sealed. The
+    retrieved buffer is immutable.
+
+    Args:
+      object_id (str): A string used to identify an object.
+    """
     buf = self.client.plasma_get(self.sock, make_plasma_id(object_id))
     return self.buffer_from_memory(buf.data, buf.size)
 
   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.sock, make_plasma_id(object_id))

src/plasma.h

@@ -6,6 +6,13 @@
 #include <errno.h>
 #include <string.h>
 
+#ifdef NDEBUG
+  #define LOG_DEBUG(M, ...)
+#else
+  #define LOG_DEBUG(M, ...) \
+    fprintf(stderr, "[DEBUG] (%s:%d) " M "\n", __FILE__, __LINE__, ##__VA_ARGS__)
+#endif
+
 #define LOG_ERR(M, ...) \
   fprintf(stderr, "[ERROR] (%s:%d: errno: %s) " M "\n", \
     __FILE__, __LINE__, errno == 0 ? "None" : strerror(errno), ##__VA_ARGS__)
@@ -24,18 +31,17 @@ typedef struct {
   unsigned char id[20];
 } plasma_id;
 
+// these values must be in sync with the ones in plasma.py (can we have a test for that?)
 enum plasma_request_type {
   PLASMA_CREATE, // create a new object
   PLASMA_GET, // get an object
   PLASMA_SEAL, // seal an object
   PLASMA_TRANSFER, // request transfer to another store
   PLASMA_DATA, // header for sending data
-  PLASMA_REGISTER // register a plasma manager
 };
 
 typedef struct {
   int type;
-  int manager_id;
   plasma_id object_id;
   int64_t size;
   uint8_t addr[4];
@@ -64,4 +70,6 @@ plasma_buffer plasma_create(int conn, plasma_id object_id, int64_t size);
 plasma_buffer plasma_get(int conn, plasma_id object_id);
 void plasma_seal(int fd, plasma_id object_id);
 
+void plasma_send(int fd, plasma_request *req);
+
 #endif

src/plasma_manager.c

@@ -0,0 +1,321 @@
+// 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. Also keeps a list of
+// other object managers.
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <sys/mman.h>
+#include <sys/types.h>
+#include <sys/ioctl.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 "plasma.h"
+
+#define MAX_CONNECTIONS 2048
+#define MAX_NUM_MANAGERS 1024
+
+enum conn_type {
+  // Connection to send commands to the manager.
+  CONN_CONTROL,
+  // Connection to send data to another manager.
+  CONN_WRITE_DATA,
+  // Connection to receive data from another manager.
+  CONN_READ_DATA
+};
+
+typedef struct {
+  // Unique identifier for the connection.
+  int id;
+  // Of type conn_type.
+  int type;
+  // Socket of the plasma store that is accessed for reading or writing data for
+  // this connection.
+  int store_conn;
+  // Buffer this connection is reading from or writing to.
+  plasma_buffer buf;
+  // Current position in the buffer.
+  int64_t cursor;
+} conn_state;
+
+typedef struct {
+  // Name of the socket connecting to local plasma store.
+  const char* store_socket_name;
+  // Number of connections.
+  int num_conn;
+  // For the "poll" system call.
+  struct pollfd waiting[MAX_CONNECTIONS];
+  // Status of connections (both control and data).
+  conn_state conn[MAX_CONNECTIONS];
+} plasma_manager_state;
+
+void init_manager_state(plasma_manager_state *s, const char* store_socket_name) {
+  memset(&s->waiting, 0, sizeof(s->waiting));
+  memset(&s->conn, 0, sizeof(s->conn));
+  s->num_conn = 0;
+  s->store_socket_name = store_socket_name;
+}
+
+#define h_addr h_addr_list[0]
+
+// Add connection for sending commands or data to another plasma manager
+// (returns the connection id).
+int add_conn(plasma_manager_state* s, int type, int fd, int events, plasma_buffer* buf) {
+  static int conn_id = 0;
+  s->waiting[s->num_conn].fd = fd;
+  s->waiting[s->num_conn].events = events;
+  s->conn[s->num_conn].id = conn_id;
+  s->conn[s->num_conn].type = type;
+  if (buf) {
+    s->conn[s->num_conn].buf = *buf;
+  }
+  s->conn[s->num_conn].cursor = 0;
+  s->num_conn += 1;
+  return conn_id++;
+}
+
+// Remove connection with index i by swapping it with the last element.
+void remove_conn(plasma_manager_state* s, int i) {
+  memcpy(&s->waiting[i], &s->waiting[s->num_conn-1], sizeof(struct pollfd));
+  memset(&s->waiting[s->num_conn-1], 0, sizeof(struct pollfd));
+  memcpy(&s->conn[i], &s->conn[s->num_conn-1], sizeof(conn_state));
+  memset(&s->conn[s->num_conn-1], 0, sizeof(conn_state));
+}
+
+#define BUFSIZE 4096
+
+// Start transfering data to another object store manager. This establishes
+// a connection to both the manager and the local object store and sends
+// the data header to the other object manager.
+void initiate_transfer(plasma_manager_state* state, plasma_request* req) {
+  int c = plasma_store_connect(state->store_socket_name);
+  plasma_buffer buf = plasma_get(c, req->object_id);
+
+  int fd = socket(PF_INET, SOCK_STREAM, 0);
+  if (fd < 0) {
+    LOG_ERR("could not create socket");
+    exit(-1);
+  }
+  
+  char ip_addr[16];
+  snprintf(ip_addr, 32, "%d.%d.%d.%d",
+                    req->addr[0], req->addr[1],
+                    req->addr[2], req->addr[3]);
+  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;
+  bcopy(manager->h_addr, &addr.sin_addr.s_addr, manager->h_length);
+  addr.sin_port = htons(req->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", &ip_addr[0], req->port);
+    exit(-1);
+  }
+
+  add_conn(state, CONN_WRITE_DATA, fd, POLLOUT, &buf);
+
+  plasma_request manager_req = { .type = PLASMA_DATA, .object_id = req->object_id, .size = buf.size };
+  LOG_INFO("filedescriptor is %d", fd);
+  plasma_send(fd, &manager_req);
+}
+
+void setup_data_connection(int conn_idx, plasma_manager_state* state, plasma_request* req) {
+  int store_conn = plasma_store_connect(state->store_socket_name);
+  state->conn[conn_idx].type = CONN_READ_DATA;
+  state->conn[conn_idx].store_conn = store_conn;
+  state->conn[conn_idx].buf = plasma_create(store_conn, req->object_id, req->size);
+  state->conn[conn_idx].cursor = 0;
+}
+
+// Handle a command request that came in through a socket (transfering data,
+// registering object managers, accepting incoming data).
+void process_command(int conn_idx, plasma_manager_state* state, plasma_request* req) {
+  switch (req->type) {
+  case PLASMA_TRANSFER:
+    LOG_INFO("transfering object to manager with port %d", req->port);
+    initiate_transfer(state, req);
+    break;
+  case PLASMA_DATA:
+    LOG_INFO("starting to stream data");
+    setup_data_connection(conn_idx, state, req);
+    break;
+  default:
+    LOG_ERR("invalid request %d", req->type);
+    exit(-1);
+  }
+}
+
+// Handle data or command event incoming on socket with index i.
+void read_from_socket(plasma_manager_state* state, int i, plasma_request* req) {
+  ssize_t r, s;
+  switch (state->conn[i].type) {
+    case CONN_CONTROL:
+      r = read(state->waiting[i].fd, req, sizeof(plasma_request));
+      if (r == 1) {
+        LOG_ERR("read error");
+      } else if (r == 0) {
+        LOG_INFO("connection with id %d disconnected", state->conn[i].id);
+        remove_conn(state, i);
+      } else {
+        process_command(i, state, req);
+      }
+      break;
+    case CONN_READ_DATA:
+      LOG_DEBUG("polled CONN_READ_DATA");
+      r = read(state->waiting[i].fd, state->conn[i].buf.data + state->conn[i].cursor, BUFSIZE);
+      if (r == -1) {
+        LOG_ERR("read error");
+      } else if (r == 0) {
+        LOG_INFO("end of file");
+      } else {
+        state->conn[i].cursor += r;
+      }
+      if (r == 0) {
+        close(state->waiting[i].fd);
+        state->waiting[i].fd = 0;
+        state->waiting[i].events = 0;
+        plasma_seal(state->conn[i].store_conn, state->conn[i].buf.object_id);
+      }
+      break;
+    case CONN_WRITE_DATA:
+      LOG_DEBUG("polled CONN_WRITE_DATA");
+      s = state->conn[i].buf.size - state->conn[i].cursor;
+      if (s > BUFSIZE)
+        s = BUFSIZE;
+      r = write(state->waiting[i].fd, state->conn[i].buf.data + state->conn[i].cursor, s);
+      if (r != s) {
+        if (r > 0) {
+          LOG_ERR("partial write on fd %d", state->waiting[i].fd);
+        } else {
+          LOG_ERR("write error");
+          exit(-1);
+        }
+      } else {
+        state->conn[i].cursor += r;
+      }
+      if (r == 0) {
+        close(state->waiting[i].fd);
+        state->waiting[i].fd = 0;
+        state->waiting[i].events = 0;
+      }
+      break;
+    default:
+      LOG_ERR("invalid connection type");
+      exit(-1);
+  }
+}
+
+// Main event loop of the plasma manager.
+void event_loop(int sock, plasma_manager_state* state) {
+  // Add listening socket.
+  add_conn(state, CONN_CONTROL, sock, POLLIN, NULL);
+  plasma_request req;
+  while (1) {
+    int num_ready = poll(state->waiting, state->num_conn, -1);
+    if (num_ready < 0) {
+      LOG_ERR("poll failed");
+      exit(-1);
+    }
+    for (int i = 0; i < state->num_conn; ++i) {
+      if (state->waiting[i].revents == 0)
+        continue;
+      if (state->waiting[i].fd == sock) {
+        // Handle new incoming connections.
+        int new_socket = accept(sock, NULL, NULL);
+        if (new_socket < 0) {
+          if (errno != EWOULDBLOCK) {
+            LOG_ERR("accept failed");
+            exit(-1);
+          }
+          break;
+        }
+        int conn_id = add_conn(state, CONN_CONTROL, new_socket, POLLIN, NULL);
+        LOG_INFO("new connection with id %d", conn_id);
+      } else {
+        read_from_socket(state, i, &req);
+      }
+    }
+  }
+}
+
+void start_server(const char *store_socket_name, const char* master_addr, int port) {
+  struct sockaddr_in name;
+  int sock = socket(PF_INET, SOCK_STREAM, 0);
+  if (sock < 0) {
+    LOG_ERR("could not create socket");
+    exit(-1);
+  }
+  name.sin_family = AF_INET;
+  name.sin_port = htons(port);
+  name.sin_addr.s_addr = htonl(INADDR_ANY);
+  int on = 1;
+  // TODO(pcm): http://stackoverflow.com/q/1150635
+  if (ioctl(sock, FIONBIO, (char*) &on) < 0) {
+    LOG_ERR("ioctl failed");
+    close(sock);
+    exit(-1);
+  }
+  if (bind(sock, (struct sockaddr*) &name, sizeof(name)) < 0) {
+    LOG_ERR("could not bind socket");
+    exit(-1);
+  }
+  LOG_INFO("listening on port %d", port);
+  if (listen(sock, 5) == -1) {
+    LOG_ERR("could not listen to socket");
+    exit(-1);
+  }
+  plasma_manager_state state;
+  init_manager_state(&state, store_socket_name);
+  event_loop(sock, &state);
+}
+
+int main(int argc, char* argv[]) {
+  // 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;
+  int c;
+  while ((c = getopt(argc, argv, "s:m:p:")) != -1) {
+    switch (c) {
+    case 's':
+      store_socket_name = optarg;
+      break;
+    case 'm':
+      master_addr = optarg;
+      break;
+    case 'p':
+      port = atoi(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);
+  }
+  start_server(store_socket_name, master_addr, port);
+}

src/plasma_store.c

@@ -28,9 +28,12 @@
 #define MAX_NUM_CLIENTS 2048
 
 typedef struct {
-  int num_clients; // number of clients connected
-  int client_id[MAX_NUM_CLIENTS]; // unique identifier for the clients
-  struct pollfd waiting[MAX_NUM_CLIENTS]; // data structure for polling
+  // Number of clients connected.
+  int num_clients;
+  // Unique identifier for the clients.
+  int client_id[MAX_NUM_CLIENTS];
+  // Data structure for polling.
+  struct pollfd waiting[MAX_NUM_CLIENTS];
 } plasma_store_state;
 
 void init_state(plasma_store_state* s) {
@@ -48,8 +51,8 @@ int add_client(plasma_store_state* s, int fd) {
   return curr_id++;
 }
 
-// remove the client at index i by swapping it with the
-// client at index num_clients-1 and zeroing the latter out
+// Remove the client at index i by swapping it with the
+// client at index num_clients-1 and zeroing the latter out.
 void remove_client(plasma_store_state* s, int i) {
   memcpy(&s->waiting[i], &s->waiting[s->num_clients-1], sizeof(struct pollfd));
   memset(&s->waiting[s->num_clients-1], 0, sizeof(struct pollfd));
@@ -59,27 +62,35 @@ void remove_client(plasma_store_state* s, int i) {
 }
 
 typedef struct {
-  plasma_id object_id; // object id of this object
-  plasma_object_info info; // object info like size, creation time and owner
-  int fd; // memory mapped file containing the object
-  UT_hash_handle handle; // handle for the uthash table
+  // Object id of this object.
+  plasma_id object_id;
+  // Object info like size, creation time and owner.
+  plasma_object_info info;
+  // Memory mapped file containing the object.
+  int fd;
+  // Handle for the uthash table.
+  UT_hash_handle handle;
 } object_table_entry;
 
 // objects that are still being written by their owner process
 object_table_entry* open_objects = NULL;
 
-// objects that have already been sealed by their owner process and
-// can now be shared with other processes
+// Objects that have already been sealed by their owner process and
+// can now be shared with other processes.
 object_table_entry* sealed_objects = NULL;
 
 typedef struct {
-  plasma_id object_id; // object id of this object
-  int num_waiting; // number of processes waiting for the object
-  int conn[MAX_NUM_CLIENTS]; // socket connections to waiting clients
-  UT_hash_handle handle; // handle for the uthash table
+  // Object id of this object.
+  plasma_id object_id;
+  // Number of processes waiting for the object.
+  int num_waiting;
+  // Socket connections to waiting clients.
+  int conn[MAX_NUM_CLIENTS];
+  // Handle for the uthash table.
+  UT_hash_handle handle;
 } object_notify_entry;
 
-// objects that processes are waiting for
+// Objects that processes are waiting for.
 object_notify_entry* objects_notify = NULL;
 
 // Create a buffer. This is creating a temporary file and then
@@ -107,7 +118,7 @@ int create_buffer(int64_t size) {
   return fd;
 }
 
-// create a new object buffer in the hash table
+// Create a new object buffer in the hash table.
 void create_object(int conn, plasma_request* req) {
   LOG_INFO("creating object"); // TODO(pcm): add object_id here
   int fd = create_buffer(req->size);
@@ -125,7 +136,7 @@ void create_object(int conn, plasma_request* req) {
   send_fd(conn, fd, (char*) &reply, sizeof(plasma_reply));
 }
 
-// get an object from the hash table
+// Get an object from the hash table.
 void get_object(int conn, plasma_request* req) {
   object_table_entry *entry;
   HASH_FIND(handle, sealed_objects, &req->object_id, sizeof(plasma_id), entry);
@@ -146,7 +157,7 @@ void get_object(int conn, plasma_request* req) {
   }
 }
 
-// seal an object that has been created in the hash table
+// Seal an object that has been created in the hash table.
 void seal_object(int conn, plasma_request* req) {
   LOG_INFO("sealing object"); // TODO(pcm): add object_id here
   object_table_entry *entry;
@@ -158,7 +169,7 @@ void seal_object(int conn, plasma_request* req) {
   int64_t size = entry->info.size;
   int fd = entry->fd;
   HASH_ADD(handle, sealed_objects, object_id, sizeof(plasma_id), entry);
-  // inform processes that the object is ready now
+  // Inform processes that the object is ready now.
   object_notify_entry* notify_entry;
   HASH_FIND(handle, objects_notify, &req->object_id, sizeof(plasma_id), notify_entry);
   if (!notify_entry) {
@@ -205,7 +216,7 @@ void event_loop(int socket) {
         continue;
       if (state.waiting[i].fd == socket) {
         while (1) {
-          // handle new incoming connections
+          // Handle new incoming connections.
           int new_socket = accept(socket, NULL, NULL);
           if (new_socket < 0) {
             if (errno != EWOULDBLOCK) {

test/test.py

@@ -1,11 +1,18 @@
 import os
+import socket
 import subprocess
 import sys
 import unittest
+import random
+import time
+import tempfile
 
 import plasma
 
-class TestPlasmaAPI(unittest.TestCase):
+def random_object_id():
+  return "".join([chr(random.randint(0, 255)) for _ in range(20)])
+
+class TestPlasmaClient(unittest.TestCase):
 
   def setUp(self):
     # Start Plasma.
@@ -15,12 +22,12 @@ class TestPlasmaAPI(unittest.TestCase):
     self.plasma_client = plasma.PlasmaClient("/tmp/store")
 
   def tearDown(self):
-    # Kill the plasma stoe process.
+    # Kill the plasma store process.
     self.p.kill()
 
   def test_create(self):
-    # Create an object string.
-    object_id = "id" + 18 * "x"
+    # 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)
@@ -34,8 +41,8 @@ class TestPlasmaAPI(unittest.TestCase):
       self.assertEqual(memory_buffer[i], chr(i % 256))
 
   def test_illegal_functionality(self):
-    # Create an object string.
-    object_id = "id" + 18 * "x"
+    # 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)
@@ -55,5 +62,73 @@ class TestPlasmaAPI(unittest.TestCase):
       memory_buffer[0] = chr(0)
     self.assertRaises(Exception, illegal_assignment)
 
+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")
+    self.p2 = subprocess.Popen([plasma_store_executable, "-s", "/tmp/store1"])
+    self.p3 = subprocess.Popen([plasma_store_executable, "-s", "/tmp/store2"])
+    # Connect two PlasmaClients.
+    self.client1 = plasma.PlasmaClient("/tmp/store1")
+    self.client2 = plasma.PlasmaClient("/tmp/store2")
+    # 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")
+    self.p4 = subprocess.Popen([plasma_manager_executable, "-s", "/tmp/store1", "-m", "127.0.0.1", "-p", str(self.port1)])
+    self.p5 = subprocess.Popen([plasma_manager_executable, "-s", "/tmp/store2", "-m", "127.0.0.1", "-p", str(self.port2)])
+    time.sleep(0.1)
+    # Connect two Python PlasmaManagers.
+    self.manager1 = plasma.PlasmaManager("127.0.0.1", self.port1)
+    self.manager2 = plasma.PlasmaManager("127.0.0.1", self.port2)
+    time.sleep(0.5)
+
+  def tearDown(self):
+    # Kill the nameserver, PlasmaStore and PlasmaManager processes.
+    self.p2.kill()
+    self.p3.kill()
+    self.p4.kill()
+    self.p5.kill()
+
+  def test_transfer(self):
+    # Create an object id string.
+    object_id1 = random_object_id()
+    # Create a new buffer and write to it.
+    memory_buffer = self.client1.create(object_id1, 20000)
+    for i in range(len(memory_buffer)):
+      memory_buffer[i] = chr(i % 10)
+    # Seal the buffer.
+    self.client1.seal(object_id1)
+    # Transfer the buffer to the the other PlasmaStore.
+    self.manager1.transfer("127.0.0.1", self.port2, object_id1)
+    # Compare the two buffers.
+    self.assertEqual(self.client1.get(object_id1)[:], self.client2.get(object_id1)[:])
+    # Transfer the buffer again.
+    self.manager1.transfer("127.0.0.1", self.port2, object_id1)
+    # Compare the two buffers.
+    self.assertEqual(self.client1.get(object_id1)[:], self.client2.get(object_id1)[:])
+    # Create a new object id string.
+    object_id2 = random_object_id()
+    # Create a new buffer and write to it.
+    memory_buffer = self.client2.create(object_id2, 20000)
+    for i in range(len(memory_buffer)):
+      memory_buffer[i] = chr(i % 10)
+    # Seal the buffer.
+    self.client2.seal(object_id2)
+    # Transfer the buffer to the the other PlasmaStore.
+    self.manager2.transfer("127.0.0.1", self.port1, object_id2)
+    # Compare the two buffers.
+    self.assertEqual(self.client1.get(object_id2)[:], self.client2.get(object_id2)[:])
+
+  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))
+
 if __name__ == "__main__":
   unittest.main(verbosity=2)