More updates

c_glib/plasma-glib/client.cpp

@@ -41,8 +41,7 @@ G_BEGIN_DECLS
  *
  * #GPlasmaClientCreateOptions is a class for customizing object creation.
  *
- * #GPlasmaClient is a class for an interface with a plasma store
- * and a plasma manager.
+ * #GPlasmaClient is a class for an interface with a plasma store.
  *
  * Since: 0.12.0
  */

cpp/apidoc/tutorials/tensor_to_py.md

@@ -105,7 +105,7 @@ The `inputs` variable will be a list of Object IDs in their raw byte string form
 import pyarrow as pa
 import pyarrow.plasma as plasma
 
-plasma_client = plasma.connect('/tmp/plasma', '', 0)
+plasma_client = plasma.connect('/tmp/plasma')
 
 # inputs: a list of object ids
 inputs = [20 * b'1']

cpp/src/plasma/client.h

@@ -49,8 +49,7 @@ class ARROW_EXPORT PlasmaClient {
   PlasmaClient();
   ~PlasmaClient();
 
-  /// Connect to the local plasma store and plasma manager. Return
-  /// the resulting connection.
+  /// Connect to the local plasma store. Return the resulting connection.
   ///
   /// \param store_socket_name The name of the UNIX domain socket to use to
   ///        connect to the Plasma store.

cpp/src/plasma/format/plasma.fbs

@@ -231,10 +231,6 @@ table PlasmaDeleteReply {
   errors: [PlasmaError];
 }
 
-// PlasmaContains is a subset of PlasmaStatus which does not
-// involve the plasma manager, only the store. We should consider
-// unifying them in the future and deprecating PlasmaContains.
-
 table PlasmaContainsRequest {
   // ID of the object we are querying.
   object_id: string;

cpp/src/plasma/test/client_tests.cc

@@ -187,7 +187,6 @@ TEST_F(TestPlasmaStore, DeleteTest) {
   ARROW_CHECK_OK(client_.Contains(object_id, &has_object));
   ASSERT_TRUE(has_object);
 
-  // Avoid race condition of Plasma Manager waiting for notification.
   ARROW_CHECK_OK(client_.Release(object_id));
   // object_id is marked as to-be-deleted, when it is not in use, it will be deleted.
   ARROW_CHECK_OK(client_.Contains(object_id, &has_object));
@@ -251,7 +250,6 @@ TEST_F(TestPlasmaStore, ContainsTest) {
   // First create object.
   std::vector<uint8_t> data(100, 0);
   CreateObject(client_, object_id, {42}, data);
-  // Avoid race condition of Plasma Manager waiting for notification.
   std::vector<ObjectBuffer> object_buffers;
   ARROW_CHECK_OK(client_.Get({object_id}, -1, &object_buffers));
   ARROW_CHECK_OK(client_.Contains(object_id, &has_object));

java/plasma/src/main/java/org/apache/arrow/plasma/ObjectStoreLink.java

@@ -79,16 +79,6 @@ public interface ObjectStoreLink {
    */
   List<ObjectStoreData> get(byte[][] objectIds, int timeoutMs);
 
-  /**
-   * Wait until <tt>numReturns</tt> objects in <tt>objectIds</tt> are ready.
-   *
-   * @param objectIds List of object IDs to wait for.
-   * @param timeoutMs Return to the caller after <tt>timeoutMs</tt> milliseconds.
-   * @param numReturns We are waiting for this number of objects to be ready.
-   * @return List of object IDs that are ready
-   */
-  List<byte[]> wait(byte[][] objectIds, int timeoutMs, int numReturns);
-
   /**
    * Compute the hash of an object in the object store.
    *
@@ -98,23 +88,6 @@ public interface ObjectStoreLink {
    */
   byte[] hash(byte[] objectId);
 
-  /**
-   * Fetch the object with the given ID from other plasma manager instances.
-   *
-   * @param objectId The object ID used to identify the object.
-   */
-  default void fetch(byte[] objectId) {
-    byte[][] objectIds = {objectId};
-    fetch(objectIds);
-  }
-
-  /**
-   * Fetch the objects with the given IDs from other plasma manager instances.
-   *
-   * @param objectIds List of object IDs used to identify the objects.
-   */
-  void fetch(byte[][] objectIds);
-
   /**
    * Evict some objects to recover given count of bytes.
    *

python/benchmarks/plasma.py

@@ -32,7 +32,7 @@ class SimplePlasmaThroughput(object):
         self.plasma_store_ctx = plasma.start_plasma_store(
             plasma_store_memory=10**9)
         plasma_store_name, p = self.plasma_store_ctx.__enter__()
-        self.plasma_client = plasma.connect(plasma_store_name, "", 64)
+        self.plasma_client = plasma.connect(plasma_store_name)
 
         self.data = np.random.randn(size // 8)
 
@@ -52,7 +52,7 @@ class SimplePlasmaLatency(object):
         self.plasma_store_ctx = plasma.start_plasma_store(
             plasma_store_memory=10**9)
         plasma_store_name, p = self.plasma_store_ctx.__enter__()
-        self.plasma_client = plasma.connect(plasma_store_name, "", 64)
+        self.plasma_client = plasma.connect(plasma_store_name)
 
     def teardown(self):
         self.plasma_store_ctx.__exit__(None, None, None)

python/examples/plasma/sorting/sort_df.py

@@ -49,7 +49,7 @@ column_to_sort = column_names[0]
 # Connect to clients
 def connect():
     global client
-    client = plasma.connect('/tmp/store', '', 0)
+    client = plasma.connect('/tmp/store')
     np.random.seed(int(time.time() * 10e7) % 10000000)
 
 

python/pyarrow/tensorflow/plasma_op.cc

@@ -71,13 +71,10 @@ class TensorToPlasmaOp : public tf::AsyncOpKernel {
   explicit TensorToPlasmaOp(tf::OpKernelConstruction* context) : tf::AsyncOpKernel(context) {
     OP_REQUIRES_OK(context, context->GetAttr("plasma_store_socket_name",
                                              &plasma_store_socket_name_));
-    OP_REQUIRES_OK(context, context->GetAttr("plasma_manager_socket_name",
-                                             &plasma_manager_socket_name_));
     tf::mutex_lock lock(mu_);
     if (!connected_) {
       VLOG(1) << "Connecting to Plasma...";
-      ARROW_CHECK_OK(client_.Connect(plasma_store_socket_name_,
-                                     plasma_manager_socket_name_));
+      ARROW_CHECK_OK(client_.Connect(plasma_store_socket_name_));
       VLOG(1) << "Connected!";
       connected_ = true;
     }
@@ -226,7 +223,6 @@ class TensorToPlasmaOp : public tf::AsyncOpKernel {
 
  private:
   std::string plasma_store_socket_name_;
-  std::string plasma_manager_socket_name_;
 
   tf::mutex mu_;
   bool connected_ = false;
@@ -243,13 +239,10 @@ class PlasmaToTensorOp : public tf::AsyncOpKernel {
   explicit PlasmaToTensorOp(tf::OpKernelConstruction* context) : tf::AsyncOpKernel(context) {
     OP_REQUIRES_OK(context, context->GetAttr("plasma_store_socket_name",
                                              &plasma_store_socket_name_));
-    OP_REQUIRES_OK(context, context->GetAttr("plasma_manager_socket_name",
-                                             &plasma_manager_socket_name_));
     tf::mutex_lock lock(mu_);
     if (!connected_) {
       VLOG(1) << "Connecting to Plasma...";
-      ARROW_CHECK_OK(client_.Connect(plasma_store_socket_name_,
-                                     plasma_manager_socket_name_));
+      ARROW_CHECK_OK(client_.Connect(plasma_store_socket_name_));
       VLOG(1) << "Connected!";
       connected_ = true;
     }
@@ -364,7 +357,6 @@ class PlasmaToTensorOp : public tf::AsyncOpKernel {
 
  private:
   std::string plasma_store_socket_name_;
-  std::string plasma_manager_socket_name_;
 
   tf::mutex mu_;
   bool connected_ = false;
@@ -375,8 +367,7 @@ REGISTER_OP("TensorToPlasma")
     .Input("input_tensor: dtypes")
     .Input("plasma_object_id: string")
     .Attr("dtypes: list(type)")
-    .Attr("plasma_store_socket_name: string")
-    .Attr("plasma_manager_socket_name: string");
+    .Attr("plasma_store_socket_name: string");
 
 REGISTER_KERNEL_BUILDER(Name("TensorToPlasma").Device(tf::DEVICE_CPU),
                         TensorToPlasmaOp<CPUDevice>);
@@ -389,8 +380,7 @@ REGISTER_OP("PlasmaToTensor")
     .Input("plasma_object_id: string")
     .Output("tensor: dtype")
     .Attr("dtype: type")
-    .Attr("plasma_store_socket_name: string")
-    .Attr("plasma_manager_socket_name: string");
+    .Attr("plasma_store_socket_name: string");
 
 REGISTER_KERNEL_BUILDER(Name("PlasmaToTensor").Device(tf::DEVICE_CPU),
                         PlasmaToTensorOp<CPUDevice>);

python/pyarrow/tests/test_plasma.py

@@ -121,8 +121,8 @@ class TestPlasmaClient(object):
             use_one_memory_mapped_file=use_one_memory_mapped_file)
         self.plasma_store_name, self.p = self.plasma_store_ctx.__enter__()
         # Connect to Plasma.
-        self.plasma_client = plasma.connect(self.plasma_store_name, "")
-        self.plasma_client2 = plasma.connect(self.plasma_store_name, "")
+        self.plasma_client = plasma.connect(self.plasma_store_name)
+        self.plasma_client2 = plasma.connect(self.plasma_store_name)
 
     def teardown_method(self, test_method):
         try:
@@ -147,7 +147,7 @@ class TestPlasmaClient(object):
         import pyarrow.plasma as plasma
         # ARROW-1264
         with pytest.raises(IOError):
-            plasma.connect('unknown-store-name', '', 0, 1)
+            plasma.connect('unknown-store-name', num_retries=1)
 
     def test_create(self):
         # Create an object id string.
@@ -860,7 +860,7 @@ class TestPlasmaClient(object):
         object_id = random_object_id()
 
         def client_blocked_in_get(plasma_store_name):
-            client = plasma.connect(self.plasma_store_name, "", 0)
+            client = plasma.connect(self.plasma_store_name)
             # Try to get an object ID that doesn't exist. This should block.
             client.get([object_id])
 
@@ -889,7 +889,7 @@ class TestPlasmaClient(object):
         object_ids = [random_object_id() for _ in range(10)]
 
         def client_get_multiple(plasma_store_name):
-            client = plasma.connect(self.plasma_store_name, "", 0)
+            client = plasma.connect(self.plasma_store_name)
             # Try to get an object ID that doesn't exist. This should block.
             client.get(object_ids)
 
@@ -948,7 +948,7 @@ def test_use_huge_pages():
             plasma_store_memory=2*10**9,
             plasma_directory="/mnt/hugepages",
             use_hugepages=True) as (plasma_store_name, p):
-        plasma_client = plasma.connect(plasma_store_name, "")
+        plasma_client = plasma.connect(plasma_store_name)
         create_object(plasma_client, 10**8)
 
 
@@ -962,7 +962,7 @@ def test_plasma_client_sharing():
     with plasma.start_plasma_store(
             plasma_store_memory=DEFAULT_PLASMA_STORE_MEMORY) \
             as (plasma_store_name, p):
-        plasma_client = plasma.connect(plasma_store_name, "")
+        plasma_client = plasma.connect(plasma_store_name)
         object_id = plasma_client.put(np.zeros(3))
         buf = plasma_client.get(object_id)
         del plasma_client
@@ -977,7 +977,7 @@ def test_plasma_list():
     with plasma.start_plasma_store(
             plasma_store_memory=DEFAULT_PLASMA_STORE_MEMORY) \
             as (plasma_store_name, p):
-        plasma_client = plasma.connect(plasma_store_name, "", 0)
+        plasma_client = plasma.connect(plasma_store_name)
 
         # Test sizes
         u, _, _ = create_object(plasma_client, 11, metadata_size=7, seal=False)

python/pyarrow/tests/test_plasma_tf_op.py

@@ -37,15 +37,13 @@ def run_tensorflow_test_with_dtype(tf, plasma, plasma_store_name,
         return plasma.tf_plasma_op.tensor_to_plasma(
             [data_tensor, ones_tensor],
             object_id,
-            plasma_store_socket_name=plasma_store_name,
-            plasma_manager_socket_name="")
+            plasma_store_socket_name=plasma_store_name)
 
     def FromPlasma():
         return plasma.tf_plasma_op.plasma_to_tensor(
             object_id,
             dtype=tf.as_dtype(dtype),
-            plasma_store_socket_name=plasma_store_name,
-            plasma_manager_socket_name="")
+            plasma_store_socket_name=plasma_store_name)
 
     with tf.device(FORCE_DEVICE):
         to_plasma = ToPlasma()
@@ -94,7 +92,7 @@ def test_plasma_tf_op(use_gpu=False):
         pytest.skip("TensorFlow Op not found")
 
     with plasma.start_plasma_store(10**8) as (plasma_store_name, p):
-        client = plasma.connect(plasma_store_name, "")
+        client = plasma.connect(plasma_store_name)
         for dtype in [np.float32, np.float64,
                       np.int8, np.int16, np.int32, np.int64]:
             run_tensorflow_test_with_dtype(tf, plasma, plasma_store_name,