From a8517cc82a35e72f085884f8c9e3f9c7fb63f178 Mon Sep 17 00:00:00 2001 From: Yuhong Guo Date: Fri, 25 May 2018 16:16:44 +0800 Subject: [PATCH] Fix infinite retry in Push function. (#2133) --- src/common/state/ray_config.h | 8 ++++++++ src/ray/common/client_connection.h | 4 ++-- src/ray/gcs/redis_context.h | 2 +- src/ray/id.h | 2 +- src/ray/object_manager/object_manager.cc | 18 ++++++++++++++---- src/ray/object_manager/object_manager.h | 7 +++++-- .../test/object_manager_stress_test.cc | 3 +++ .../object_manager/test/object_manager_test.cc | 3 +++ src/ray/raylet/main.cc | 2 ++ .../raylet/object_manager_integration_test.cc | 2 ++ 10 files changed, 41 insertions(+), 10 deletions(-) diff --git a/src/common/state/ray_config.h b/src/common/state/ray_config.h index 07f4c0d3c..957dcf96f 100644 --- a/src/common/state/ray_config.h +++ b/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. diff --git a/src/ray/common/client_connection.h b/src/ray/common/client_connection.h index 0f65e4327..a0c8b659c 100644 --- a/src/ray/common/client_connection.h +++ b/src/ray/common/client_connection.h @@ -78,8 +78,8 @@ class ClientConnection : public ServerConnection, 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. static std::shared_ptr> Create( diff --git a/src/ray/gcs/redis_context.h b/src/ray/gcs/redis_context.h index e4f90b074..14faab7ac 100644 --- a/src/ray/gcs/redis_context.h +++ b/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. diff --git a/src/ray/id.h b/src/ray/id.h index beb15fc5e..cf91a2835 100644 --- a/src/ray/id.h +++ b/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); diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc index af7ba1274..be5a98816 100644 --- a/src/ray/object_manager/object_manager.cc +++ b/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(); } diff --git a/src/ray/object_manager/object_manager.h b/src/ray/object_manager/object_manager.h index f5ea78a54..a77149abf 100644 --- a/src/ray/object_manager/object_manager.h +++ b/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. diff --git a/src/ray/object_manager/test/object_manager_stress_test.cc b/src/ray/object_manager/test/object_manager_stress_test.cc index f0931973b..c6075e6dc 100644 --- a/src/ray/object_manager/test/object_manager_stress_test.cc +++ b/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(std::pow(10, 3)); + int max_push_retries = 1000; // start first server gcs_client_1 = std::shared_ptr(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. diff --git a/src/ray/object_manager/test/object_manager_test.cc b/src/ray/object_manager/test/object_manager_test.cc index c1d83bc11..fb0d35cc6 100644 --- a/src/ray/object_manager/test/object_manager_test.cc +++ b/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(std::pow(10, 3)); + int max_push_retries = 1000; // start first server gcs_client_1 = std::shared_ptr(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. diff --git a/src/ray/raylet/main.cc b/src/ray/raylet/main.cc index 453df9dc5..68bd98480 100644 --- a/src/ray/raylet/main.cc +++ b/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(); diff --git a/src/ray/raylet/object_manager_integration_test.cc b/src/ray/raylet/object_manager_integration_test.cc index b7f2889ac..ec1586ce3 100644 --- a/src/ray/raylet/object_manager_integration_test.cc +++ b/src/ray/raylet/object_manager_integration_test.cc @@ -55,6 +55,7 @@ class TestObjectManagerBase : public ::testing::Test { gcs_client_1 = std::shared_ptr(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(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));