Fix infinite retry in Push function. (#2133)

src/common/state/ray_config.h

@@ -96,6 +96,10 @@ class RayConfig {
     return object_manager_max_receives_;
   }
 
+  int object_manager_max_push_retries() const {
+    return object_manager_max_push_retries_;
+  }
+
   uint64_t object_manager_default_chunk_size() const {
     return object_manager_default_chunk_size_;
   }
@@ -134,6 +138,7 @@ class RayConfig {
         object_manager_pull_timeout_ms_(20),
         object_manager_max_sends_(2),
         object_manager_max_receives_(2),
+        object_manager_max_push_retries_(1000),
         object_manager_default_chunk_size_(100000000) {}
 
   ~RayConfig() {}
@@ -228,6 +233,9 @@ class RayConfig {
   /// Maximum number of concurrent receives allowed by the object manager.
   int object_manager_max_receives_;
 
+  /// Maximum push retries allowed by the object manager.
+  int object_manager_max_push_retries_;
+
   /// Default chunk size for multi-chunk transfers to use in the object manager.
   /// In the object manager, no single thread is permitted to transfer more
   /// data than what is specified by the chunk size.

src/ray/common/client_connection.h

@@ -78,8 +78,8 @@ class ClientConnection : public ServerConnection<T>,
  public:
   /// Allocate a new node client connection.
   ///
-  /// \param ClientManager A reference to the manager that will process a
-  /// message from this client.
+  /// \param new_client_handler A reference to the client handler.
+  /// \param message_handler A reference to the message handler.
   /// \param socket The client socket.
   /// \return std::shared_ptr<ClientConnection>.
   static std::shared_ptr<ClientConnection<T>> Create(

src/ray/gcs/redis_context.h

@@ -63,7 +63,7 @@ class RedisContext {
   /// \param length The length of the data to be added, if data is provided.
   /// \param prefix
   /// \param pubsub_channel
-  /// \param callback_index
+  /// \param redisCallback The Redis callback function.
   /// \param log_length The RAY.TABLE_APPEND command takes in an optional index
   ///        at which the data must be appended. For all other commands, set to
   ///        -1 for unused. If set, then data must be provided.

src/ray/id.h

@@ -63,7 +63,7 @@ const TaskID FinishTaskId(const TaskID &task_id);
 /// Compute the object ID of an object returned by the task.
 ///
 /// \param task_id The task ID of the task that created the object.
-/// \param put_index What number return value this object is in the task.
+/// \param return_index What number return value this object is in the task.
 /// \return The computed object ID.
 const ObjectID ComputeReturnId(const TaskID &task_id, int64_t return_index);
 

src/ray/object_manager/object_manager.cc

@@ -23,6 +23,7 @@ ObjectManager::ObjectManager(asio::io_service &main_service,
       connection_pool_() {
   RAY_CHECK(config_.max_sends > 0);
   RAY_CHECK(config_.max_receives > 0);
+  RAY_CHECK(config_.max_push_retries > 0);
   main_service_ = &main_service;
   store_notification_.SubscribeObjAdded(
       [this](const ObjectInfoT &object_info) { NotifyDirectoryObjectAdd(object_info); });
@@ -46,6 +47,7 @@ ObjectManager::ObjectManager(asio::io_service &main_service,
       connection_pool_() {
   RAY_CHECK(config_.max_sends > 0);
   RAY_CHECK(config_.max_receives > 0);
+  RAY_CHECK(config_.max_push_retries > 0);
   // TODO(hme) Client ID is never set with this constructor.
   main_service_ = &main_service;
   store_notification_.SubscribeObjAdded(
@@ -194,11 +196,19 @@ ray::Status ObjectManager::PullSendRequest(const ObjectID &object_id,
   return ray::Status::OK();
 }
 
-ray::Status ObjectManager::Push(const ObjectID &object_id, const ClientID &client_id) {
+ray::Status ObjectManager::Push(const ObjectID &object_id, const ClientID &client_id,
+                                int retry) {
   if (local_objects_.count(object_id) == 0) {
-    // TODO(hme): Do not retry indefinitely...
-    main_service_->post(
-        [this, object_id, client_id]() { RAY_CHECK_OK(Push(object_id, client_id)); });
+    if (retry < 0) {
+      retry = config_.max_push_retries;
+    } else if (retry == 0) {
+      RAY_LOG(ERROR) << "Invalid Push request ObjectID: " << object_id
+                     << " after retrying " << config_.max_push_retries << " times.";
+      return ray::Status::OK();
+    }
+    main_service_->post([this, object_id, client_id, retry]() {
+      RAY_CHECK_OK(Push(object_id, client_id, retry - 1));
+    });
     return ray::Status::OK();
   }
 

src/ray/object_manager/object_manager.h

@@ -39,8 +39,10 @@ struct ObjectManagerConfig {
   int max_receives;
   /// Object chunk size, in bytes
   uint64_t object_chunk_size;
-  // TODO(hme): Implement num retries (to avoid infinite retries).
+  /// The stored socked name.
   std::string store_socket_name;
+  /// Maximun number of push retries.
+  int max_push_retries;
 };
 
 class ObjectManagerInterface {
@@ -97,8 +99,9 @@ class ObjectManager : public ObjectManagerInterface {
   ///
   /// \param object_id The object's object id.
   /// \param client_id The remote node's client id.
+  /// \param retry The count down retries, -1 means the default maximum retries.
   /// \return Status of whether the push request successfully initiated.
-  ray::Status Push(const ObjectID &object_id, const ClientID &client_id);
+  ray::Status Push(const ObjectID &object_id, const ClientID &client_id, int retry = -1);
 
   /// Pull an object from ClientID. Returns UniqueID asociated with
   /// an invocation of this method.

src/ray/object_manager/test/object_manager_stress_test.cc

@@ -124,6 +124,7 @@ class TestObjectManagerBase : public ::testing::Test {
     int max_sends = 2;
     int max_receives = 2;
     uint64_t object_chunk_size = static_cast<uint64_t>(std::pow(10, 3));
+    int max_push_retries = 1000;
 
     // start first server
     gcs_client_1 = std::shared_ptr<gcs::AsyncGcsClient>(new gcs::AsyncGcsClient());
@@ -133,6 +134,7 @@ class TestObjectManagerBase : public ::testing::Test {
     om_config_1.max_sends = max_sends;
     om_config_1.max_receives = max_receives;
     om_config_1.object_chunk_size = object_chunk_size;
+    om_config_1.max_push_retries = max_push_retries;
     server1.reset(new MockServer(main_service, om_config_1, gcs_client_1));
 
     // start second server
@@ -143,6 +145,7 @@ class TestObjectManagerBase : public ::testing::Test {
     om_config_2.max_sends = max_sends;
     om_config_2.max_receives = max_receives;
     om_config_2.object_chunk_size = object_chunk_size;
+    om_config_2.max_push_retries = max_push_retries;
     server2.reset(new MockServer(main_service, om_config_2, gcs_client_2));
 
     // connect to stores.

src/ray/object_manager/test/object_manager_test.cc

@@ -114,6 +114,7 @@ class TestObjectManager : public ::testing::Test {
     int max_sends = 2;
     int max_receives = 2;
     uint64_t object_chunk_size = static_cast<uint64_t>(std::pow(10, 3));
+    int max_push_retries = 1000;
 
     // start first server
     gcs_client_1 = std::shared_ptr<gcs::AsyncGcsClient>(new gcs::AsyncGcsClient());
@@ -123,6 +124,7 @@ class TestObjectManager : public ::testing::Test {
     om_config_1.max_sends = max_sends;
     om_config_1.max_receives = max_receives;
     om_config_1.object_chunk_size = object_chunk_size;
+    om_config_1.max_push_retries = max_push_retries;
     server1.reset(new MockServer(main_service, om_config_1, gcs_client_1));
 
     // start second server
@@ -133,6 +135,7 @@ class TestObjectManager : public ::testing::Test {
     om_config_2.max_sends = max_sends;
     om_config_2.max_receives = max_receives;
     om_config_2.object_chunk_size = object_chunk_size;
+    om_config_2.max_push_retries = max_push_retries;
     server2.reset(new MockServer(main_service, om_config_2, gcs_client_2));
 
     // connect to stores.

src/ray/raylet/main.cc

@@ -54,6 +54,8 @@ int main(int argc, char *argv[]) {
   object_manager_config.max_sends = RayConfig::instance().object_manager_max_sends();
   object_manager_config.max_receives =
       RayConfig::instance().object_manager_max_receives();
+  object_manager_config.max_push_retries =
+      RayConfig::instance().object_manager_max_push_retries();
   object_manager_config.object_chunk_size =
       RayConfig::instance().object_manager_default_chunk_size();
 

src/ray/raylet/object_manager_integration_test.cc

@@ -55,6 +55,7 @@ class TestObjectManagerBase : public ::testing::Test {
     gcs_client_1 = std::shared_ptr<gcs::AsyncGcsClient>(new gcs::AsyncGcsClient());
     ObjectManagerConfig om_config_1;
     om_config_1.store_socket_name = store_sock_1;
+    om_config_1.max_push_retries = 1000;
     server1.reset(new ray::raylet::Raylet(
         main_service, "raylet_1", "0.0.0.0", "127.0.0.1", 6379,
         GetNodeManagerConfig("raylet_1", store_sock_1), om_config_1, gcs_client_1));
@@ -63,6 +64,7 @@ class TestObjectManagerBase : public ::testing::Test {
     gcs_client_2 = std::shared_ptr<gcs::AsyncGcsClient>(new gcs::AsyncGcsClient());
     ObjectManagerConfig om_config_2;
     om_config_2.store_socket_name = store_sock_2;
+    om_config_2.max_push_retries = 1000;
     server2.reset(new ray::raylet::Raylet(
         main_service, "raylet_2", "0.0.0.0", "127.0.0.1", 6379,
         GetNodeManagerConfig("raylet_2", store_sock_2), om_config_2, gcs_client_2));