[core] Delete() should never remote objects from in-memory store (#7117)

build.sh

@@ -120,7 +120,7 @@ pushd "$BUILD_DIR"
 # The following line installs pyarrow from S3, these wheels have been
 # generated from https://github.com/ray-project/arrow-build from
 # the commit listed in the command.
-if [ -z "$SKIP_PYARROW_INSTALL" ]; then
+if [ -z "$SKIP_THIRDPARTY_INSTALL" ]; then
     "$PYTHON_EXECUTABLE" -m pip install -q \
         --target="$ROOT_DIR/python/ray/pyarrow_files" pyarrow==0.14.0.RAY \
         --find-links https://s3-us-west-2.amazonaws.com/arrow-wheels/3a11193d9530fe8ec7fdb98057f853b708f6f6ae/index.html
@@ -137,8 +137,10 @@ popd
 popd
 
 
-"$PYTHON_EXECUTABLE" -m pip install -q psutil setproctitle \
-        --target="$ROOT_DIR/python/ray/thirdparty_files"
+if [ -z "$SKIP_THIRDPARTY_INSTALL" ]; then
+    "$PYTHON_EXECUTABLE" -m pip install -q psutil setproctitle \
+            --target="$ROOT_DIR/python/ray/thirdparty_files"
+fi
 
 export PYTHON3_BIN_PATH="$PYTHON_EXECUTABLE"
 

python/ray/tests/test_advanced.py

@@ -21,6 +21,35 @@ from ray.test_utils import RayTestTimeoutException
 logger = logging.getLogger(__name__)
 
 
+# issue https://github.com/ray-project/ray/issues/7105
+def test_internal_free(shutdown_only):
+    ray.init(num_cpus=1)
+
+    @ray.remote
+    class Sampler:
+        def sample(self):
+            return [1, 2, 3, 4, 5]
+
+        def sample_big(self):
+            return np.zeros(1024 * 1024)
+
+    sampler = Sampler.remote()
+
+    # Free does not delete from in-memory store.
+    obj_id = sampler.sample.remote()
+    ray.get(obj_id)
+    ray.internal.free(obj_id)
+    assert ray.get(obj_id) == [1, 2, 3, 4, 5]
+
+    # Free deletes big objects from plasma store.
+    big_id = sampler.sample_big.remote()
+    ray.get(big_id)
+    ray.internal.free(big_id)
+    time.sleep(1)  # wait for delete RPC to propagate
+    with pytest.raises(Exception):
+        ray.get(big_id)
+
+
 def test_wait_iterables(ray_start_regular):
     @ray.remote
     def f(delay):

src/ray/core_worker/core_worker.cc

@@ -614,14 +614,16 @@ Status CoreWorker::Wait(const std::vector<ObjectID> &ids, int num_objects,
 
 Status CoreWorker::Delete(const std::vector<ObjectID> &object_ids, bool local_only,
                           bool delete_creating_tasks) {
-  absl::flat_hash_set<ObjectID> plasma_object_ids;
-  absl::flat_hash_set<ObjectID> memory_object_ids;
-  GroupObjectIdsByStoreProvider(object_ids, &plasma_object_ids, &memory_object_ids);
-
   // TODO(edoakes): what are the desired semantics for deleting from a non-owner?
   // Should we just delete locally or ping the owner and delete globally?
   reference_counter_->DeleteReferences(object_ids);
-  memory_store_->Delete(memory_object_ids, &plasma_object_ids);
+
+  // We only delete from plasma, which avoids hangs (issue #7105). In-memory
+  // objects are always handled by ref counting only.
+  absl::flat_hash_set<ObjectID> plasma_object_ids;
+  for (const auto &obj_id : object_ids) {
+    plasma_object_ids.insert(obj_id);
+  }
   RAY_RETURN_NOT_OK(plasma_store_provider_->Delete(plasma_object_ids, local_only,
                                                    delete_creating_tasks));
 

src/ray/core_worker/core_worker.h

@@ -269,7 +269,7 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
   Status Wait(const std::vector<ObjectID> &object_ids, const int num_objects,
               const int64_t timeout_ms, std::vector<bool> *results);
 
-  /// Delete a list of objects from the object store.
+  /// Delete a list of objects from the plasma object store.
   ///
   /// \param[in] object_ids IDs of the objects to delete.
   /// \param[in] local_only Whether only delete the objects in local node, or all nodes in