Revert "Enable TryCreateImmediately to use the fallback allocation" (#16542)

This reverts commit 41cf2e3d50.

python/ray/tests/test_object_manager.py

@@ -51,12 +51,8 @@ def test_object_transfer_during_oom(ray_start_cluster_head):
     local_ref = ray.put(np.random.rand(5 * 1024 * 1024))
     remote_ref = put.remote()
 
-    if ray.worker.global_worker.core_worker.plasma_unlimited():
-        ray.get(remote_ref, timeout=10)
-    else:
-        with pytest.raises(GetTimeoutError):
-            # Sadly, the test cannot work in this mode.
-            ray.get(remote_ref, timeout=1)
+    with pytest.raises(GetTimeoutError):
+        ray.get(remote_ref, timeout=1)
     del local_ref
     ray.get(remote_ref)
 
@@ -286,13 +282,8 @@ def test_pull_request_retry(shutdown_only):
 
         remote_ref = put.remote()
 
-        ready, _ = ray.wait([remote_ref], timeout=10)
-
-        if ray.worker.global_worker.core_worker.plasma_unlimited():
-            # Sadly, the test cannot work in this mode.
-            assert len(ready) == 1
-        else:
-            assert len(ready) == 0
+        ready, _ = ray.wait([remote_ref], timeout=1)
+        assert len(ready) == 0
 
         del local_ref
 

src/ray/object_manager/object_manager.cc

@@ -939,9 +939,9 @@ std::string ObjectManager::DebugString() const {
   result << "\n- num buffered profile events: " << profile_events_.size();
   result << "\n- num chunks received total: " << num_chunks_received_total_;
   result << "\n- num chunks received total failed: " << num_chunks_received_total_failed_;
-  result << "\n- num chunks received cancelled: " << num_chunks_received_cancelled_;
-  result << "\n- num chunks received thrashed: " << num_chunks_received_thrashed_;
-  result << "\n- num chunks received, plasma error: "
+  result << "\n  - num chunks received cancelled: " << num_chunks_received_cancelled_;
+  result << "\n  - num chunks received thrashed: " << num_chunks_received_thrashed_;
+  result << "\n  - num chunks received, plasma error : "
          << num_chunks_received_failed_due_to_plasma_;
   result << "\nEvent stats:" << rpc_service_.StatsString();
   result << "\n" << push_manager_->DebugString();

src/ray/object_manager/plasma/create_request_queue.cc

@@ -63,12 +63,6 @@ std::pair<PlasmaObject, PlasmaError> CreateRequestQueue::TryRequestImmediately(
     const CreateObjectCallback &create_callback, size_t object_size) {
   PlasmaObject result = {};
 
-  // Immediately fulfill it using the fallback allocator.
-  if (RayConfig::instance().plasma_unlimited()) {
-    PlasmaError error = create_callback(&result, /*fallback_allocator=*/true);
-    return {result, error};
-  }
-
   if (!queue_.empty()) {
     // There are other requests queued. Return an out-of-memory error
     // immediately because this request cannot be served.

src/ray/object_manager/plasma/plasma_allocator.cc

@@ -66,11 +66,9 @@ void *PlasmaAllocator::DiskMemalignUnlimited(size_t alignment, size_t bytes) {
   if (!mem) {
     return nullptr;
   }
+  RAY_CHECK(IsOutsideInitialAllocation(mem));
   allocated_ += bytes;
-  // The allocation was servicable using the initial region, no need to fallback.
-  if (IsOutsideInitialAllocation(mem)) {
-    fallback_allocated_ += bytes;
-  }
+  fallback_allocated_ += bytes;
   return mem;
 }
 

src/ray/object_manager/test/create_request_queue_test.cc

@@ -475,13 +475,8 @@ TEST_F(CreateRequestQueueTest, TestTryRequestImmediately) {
   // Request would block.
   auto req_id = queue_.AddRequest(ObjectID::Nil(), client, request, 1234);
   result = queue_.TryRequestImmediately(ObjectID::Nil(), client, request, 1234);
-  if (RayConfig::instance().plasma_unlimited()) {
-    result = queue_.TryRequestImmediately(ObjectID::Nil(), client, request, 1234);
-    ASSERT_EQ(result.first.data_size, 1234);
-  } else {
-    ASSERT_EQ(result.first.data_size, 0);
-    ASSERT_EQ(result.second, PlasmaError::OutOfMemory);
-  }
+  ASSERT_EQ(result.first.data_size, 0);
+  ASSERT_EQ(result.second, PlasmaError::OutOfMemory);
   ASSERT_TRUE(queue_.ProcessRequests().ok());
 
   // Queue is empty again, request can be fulfilled.

src/ray/raylet/node_manager.cc

@@ -459,7 +459,6 @@ ray::Status NodeManager::RegisterGcs() {
     periodical_runner_.RunFnPeriodically(
         [this] {
           RAY_LOG(INFO) << "Event stats:\n\n" << io_service_.StatsString() << "\n\n";
-          RAY_LOG(INFO) << DebugString() << "\n\n";
         },
         event_stats_print_interval_ms,
         "NodeManager.deadline_timer.print_event_loop_stats");