[core] Pull Manager exponential backoff (#13024)

src/ray/object_manager/pull_manager.cc

@@ -111,6 +111,10 @@ void PullManager::TryPull(const ObjectID &object_id) {
 
   RAY_LOG(DEBUG) << "Sending pull request from " << self_node_id_ << " to " << node_id
                  << " of object " << object_id;
+  const auto time = get_time_();
+  auto &request = it->second;
+  auto retry_timeout_len = (pull_timeout_ms_ / 1000.) * (1UL << request.num_retries);
+  request.next_pull_time = time + retry_timeout_len;
   send_pull_request_(object_id, node_id);
 }
 
@@ -131,7 +135,8 @@ void PullManager::Tick() {
     const auto time = get_time_();
     if (time >= request.next_pull_time) {
       TryPull(object_id);
-      request.next_pull_time = time + pull_timeout_ms_ / 1000;
+      // Bound the retry time at 10 * 1024 seconds.
+      request.num_retries = std::min(request.num_retries + 1, 10);
     }
   }
 }

src/ray/object_manager/pull_manager.h

@@ -81,9 +81,10 @@ class PullManager {
   /// A helper structure for tracking information about each ongoing object pull.
   struct PullRequest {
     PullRequest(double first_retry_time)
-        : client_locations(), next_pull_time(first_retry_time) {}
+        : client_locations(), next_pull_time(first_retry_time), num_retries(0) {}
     std::vector<NodeID> client_locations;
     double next_pull_time;
+    uint8_t num_retries;
   };
 
   /// See the constructor's arguments.
@@ -92,7 +93,7 @@ class PullManager {
   const std::function<void(const ObjectID &, const NodeID &)> send_pull_request_;
   const RestoreSpilledObjectCallback restore_spilled_object_;
   const std::function<double()> get_time_;
-  int pull_timeout_ms_;
+  uint64_t pull_timeout_ms_;
 
   /// The objects that this object manager is currently trying to fetch from
   /// remote object managers.

src/ray/object_manager/test/pull_manager_test.cc

@@ -125,7 +125,7 @@ TEST_F(PullManagerTest, TestRetryTimer) {
   ASSERT_EQ(num_send_pull_request_calls_, 1);
   ASSERT_EQ(num_restore_spilled_object_calls_, 0);
 
-  for (; fake_time_ <= 127 * 10; fake_time_ += 0.1) {
+  for (; fake_time_ <= 127 * 10; fake_time_ += 1.) {
     pull_manager_.Tick();
   }
 
@@ -140,7 +140,7 @@ TEST_F(PullManagerTest, TestRetryTimer) {
   // OnLocationChange also doesn't count towards the retry timer.
   // To the casual observer, this may seem off-by-one, but this is due to floating point
   // error (0.1 + 0.1 ... 10k times > 10 == True)
-  ASSERT_EQ(num_send_pull_request_calls_, 127 * 2);
+  ASSERT_EQ(num_send_pull_request_calls_, 1 + 7 + 127);
   ASSERT_EQ(num_restore_spilled_object_calls_, 0);
 
   pull_manager_.CancelPull(obj1);