diff --git a/src/ray/object_manager/connection_pool.cc b/src/ray/object_manager/connection_pool.cc deleted file mode 100644 index 6b029cbe8..000000000 --- a/src/ray/object_manager/connection_pool.cc +++ /dev/null @@ -1,171 +0,0 @@ -#include "ray/object_manager/connection_pool.h" - -namespace ray { - -ConnectionPool::ConnectionPool() {} - -void ConnectionPool::RegisterReceiver(ConnectionType type, const ClientID &client_id, - std::shared_ptr &conn) { - std::unique_lock guard(connection_mutex); - switch (type) { - case ConnectionType::MESSAGE: { - Add(message_receive_connections_, client_id, conn); - } break; - case ConnectionType::TRANSFER: { - Add(transfer_receive_connections_, client_id, conn); - } break; - } -} - -void ConnectionPool::RemoveReceiver(std::shared_ptr conn) { - std::unique_lock guard(connection_mutex); - const ClientID client_id = conn->GetClientId(); - if (message_receive_connections_.count(client_id) != 0) { - Remove(message_receive_connections_, client_id, conn); - } - if (transfer_receive_connections_.count(client_id) != 0) { - Remove(transfer_receive_connections_, client_id, conn); - } -} - -void ConnectionPool::RegisterSender(ConnectionType type, const ClientID &client_id, - std::shared_ptr &conn) { - std::unique_lock guard(connection_mutex); - SenderMapType &conn_map = (type == ConnectionType::MESSAGE) - ? message_send_connections_ - : transfer_send_connections_; - Add(conn_map, client_id, conn); - // Don't add to available connections. It will become available once it is released. -} - -void ConnectionPool::RemoveSender(const std::shared_ptr &conn) { - std::unique_lock guard(connection_mutex); - const ClientID client_id = conn->GetClientId(); - if (message_send_connections_.count(client_id) != 0) { - Remove(message_send_connections_, client_id, conn); - } - if (transfer_send_connections_.count(client_id) != 0) { - Remove(transfer_send_connections_, client_id, conn); - } -} - -void ConnectionPool::GetSender(ConnectionType type, const ClientID &client_id, - std::shared_ptr *conn) { - std::unique_lock guard(connection_mutex); - SenderMapType &avail_conn_map = (type == ConnectionType::MESSAGE) - ? available_message_send_connections_ - : available_transfer_send_connections_; - if (Count(avail_conn_map, client_id) > 0) { - *conn = Borrow(avail_conn_map, client_id); - } else { - *conn = nullptr; - } -} - -void ConnectionPool::ReleaseSender(ConnectionType type, - std::shared_ptr &conn) { - std::unique_lock guard(connection_mutex); - SenderMapType &conn_map = (type == ConnectionType::MESSAGE) - ? available_message_send_connections_ - : available_transfer_send_connections_; - Return(conn_map, conn->GetClientId(), conn); -} - -void ConnectionPool::Add(ReceiverMapType &conn_map, const ClientID &client_id, - std::shared_ptr conn) { - conn_map[client_id].push_back(std::move(conn)); -} - -void ConnectionPool::Add(SenderMapType &conn_map, const ClientID &client_id, - std::shared_ptr conn) { - conn_map[client_id].push_back(std::move(conn)); -} - -void ConnectionPool::Remove(ReceiverMapType &conn_map, const ClientID &client_id, - std::shared_ptr &conn) { - auto it = conn_map.find(client_id); - if (it == conn_map.end()) { - return; - } - auto &connections = it->second; - int64_t pos = - std::find(connections.begin(), connections.end(), conn) - connections.begin(); - if (pos >= static_cast(connections.size())) { - return; - } - connections.erase(connections.begin() + pos); -} - -void ConnectionPool::Remove(SenderMapType &conn_map, const ClientID &client_id, - const std::shared_ptr &conn) { - auto it = conn_map.find(client_id); - if (it == conn_map.end()) { - return; - } - auto &connections = it->second; - int64_t pos = - std::find(connections.begin(), connections.end(), conn) - connections.begin(); - if (pos >= static_cast(connections.size())) { - return; - } - connections.erase(connections.begin() + pos); -} - -uint64_t ConnectionPool::Count(SenderMapType &conn_map, const ClientID &client_id) { - auto it = conn_map.find(client_id); - if (it == conn_map.end()) { - return 0; - } - return it->second.size(); -} - -std::shared_ptr ConnectionPool::Borrow(SenderMapType &conn_map, - const ClientID &client_id) { - std::shared_ptr conn = std::move(conn_map[client_id].back()); - conn_map[client_id].pop_back(); - return conn; -} - -void ConnectionPool::Return(SenderMapType &conn_map, const ClientID &client_id, - std::shared_ptr conn) { - conn_map[client_id].push_back(std::move(conn)); -} - -std::string ConnectionPool::DebugString() const { - std::stringstream result; - result << "ConnectionPool:"; - result << "\n- num message send connections: " << message_send_connections_.size(); - result << "\n- num transfer send connections: " << transfer_send_connections_.size(); - result << "\n- num avail message send connections: " - << available_transfer_send_connections_.size(); - result << "\n- num avail transfer send connections: " - << available_transfer_send_connections_.size(); - result << "\n- num message receive connections: " - << message_receive_connections_.size(); - result << "\n- num transfer receive connections: " - << transfer_receive_connections_.size(); - return result.str(); -} - -void ConnectionPool::RecordMetrics() const { - stats::ConnectionPoolStats().Record( - message_send_connections_.size(), - {{stats::ValueTypeKey, "num_message_send_connections"}}); - stats::ConnectionPoolStats().Record( - transfer_send_connections_.size(), - {{stats::ValueTypeKey, "num_transfer_send_connections"}}); - stats::ConnectionPoolStats().Record( - available_transfer_send_connections_.size(), - {{stats::ValueTypeKey, "num_avail_message_send_connections"}}); - stats::ConnectionPoolStats().Record( - available_transfer_send_connections_.size(), - {{stats::ValueTypeKey, "num_avail_transfer_send_connections"}}); - stats::ConnectionPoolStats().Record( - message_receive_connections_.size(), - {{stats::ValueTypeKey, "num_message_receive_connections"}}); - stats::ConnectionPoolStats().Record( - transfer_receive_connections_.size(), - {{stats::ValueTypeKey, "num_transfer_receive_connections"}}); -} - -} // namespace ray diff --git a/src/ray/object_manager/connection_pool.h b/src/ray/object_manager/connection_pool.h deleted file mode 100644 index 820cf591e..000000000 --- a/src/ray/object_manager/connection_pool.h +++ /dev/null @@ -1,155 +0,0 @@ -#ifndef RAY_OBJECT_MANAGER_CONNECTION_POOL_H -#define RAY_OBJECT_MANAGER_CONNECTION_POOL_H - -#include -#include -#include -#include -#include -#include - -#include -#include -#include - -#include "ray/common/id.h" -#include "ray/common/status.h" - -#include -#include "ray/object_manager/format/object_manager_generated.h" -#include "ray/object_manager/object_directory.h" -#include "ray/object_manager/object_manager_client_connection.h" -#include "ray/stats/stats.h" - -namespace asio = boost::asio; - -namespace ray { - -class ConnectionPool { - public: - /// Callbacks for GetSender. - using SuccessCallback = std::function)>; - using FailureCallback = std::function; - - /// Connection type to distinguish between message and transfer connections. - enum class ConnectionType : int { MESSAGE = 0, TRANSFER }; - - /// Connection pool for all connections needed by the ObjectManager. - ConnectionPool(); - - /// Register a receiver connection. - /// - /// \param type The type of connection. - /// \param client_id The ClientID of the remote object manager. - /// \param conn The actual connection. - void RegisterReceiver(ConnectionType type, const ClientID &client_id, - std::shared_ptr &conn); - - /// Remove a receiver connection. - /// - /// \param conn The actual connection. - void RemoveReceiver(std::shared_ptr conn); - - /// Register a receiver connection. - /// - /// \param type The type of connection. - /// \param client_id The ClientID of the remote object manager. - /// \param conn The actual connection. - void RegisterSender(ConnectionType type, const ClientID &client_id, - std::shared_ptr &conn); - - /// Remove a sender connection. - /// - /// \param conn The actual connection. - void RemoveSender(const std::shared_ptr &conn); - - /// Get a sender connection from the connection pool. - /// The connection must be released or removed when the operation for which the - /// connection was obtained is completed. If the connection pool is empty, the - /// connection pointer passed in is set to a null pointer. - /// - /// \param[in] type The type of connection. - /// \param[in] client_id The ClientID of the remote object manager. - /// \param[out] conn An empty pointer to a shared pointer. - /// \return Void. - void GetSender(ConnectionType type, const ClientID &client_id, - std::shared_ptr *conn); - - /// Releases a sender connection, allowing it to be used by another operation. - /// - /// \param type The type of connection. - /// \param conn The actual connection. - /// \return Void. - void ReleaseSender(ConnectionType type, std::shared_ptr &conn); - - // TODO(hme): Implement with error handling. - /// Remove a sender connection. This is invoked if the connection is no longer - /// usable. - /// - /// \param type The type of connection. - /// \param conn The actual connection. - /// \return Status of invoking this method. - ray::Status RemoveSender(ConnectionType type, std::shared_ptr conn); - - /// Returns debug string for class. - /// - /// \return string. - std::string DebugString() const; - - /// Record metrics. - void RecordMetrics() const; - - /// This object cannot be copied for thread-safety. - RAY_DISALLOW_COPY_AND_ASSIGN(ConnectionPool); - - private: - /// A container type that maps ClientID to a connection type. - using SenderMapType = - std::unordered_map>>; - using ReceiverMapType = - std::unordered_map>>; - - /// Adds a receiver for ClientID to the given map. - void Add(ReceiverMapType &conn_map, const ClientID &client_id, - std::shared_ptr conn); - - /// Adds a sender for ClientID to the given map. - void Add(SenderMapType &conn_map, const ClientID &client_id, - std::shared_ptr conn); - - /// Removes the given receiver for ClientID from the given map. - void Remove(ReceiverMapType &conn_map, const ClientID &client_id, - std::shared_ptr &conn); - - /// Removes the given sender for ClientID from the given map. - void Remove(SenderMapType &conn_map, const ClientID &client_id, - const std::shared_ptr &conn); - - /// Returns the count of sender connections to ClientID. - uint64_t Count(SenderMapType &conn_map, const ClientID &client_id); - - /// Removes a sender connection to ClientID from the pool of available connections. - /// This method assumes conn_map has available connections to ClientID. - std::shared_ptr Borrow(SenderMapType &conn_map, - const ClientID &client_id); - - /// Returns a sender connection to ClientID to the pool of available connections. - void Return(SenderMapType &conn_map, const ClientID &client_id, - std::shared_ptr conn); - - // TODO(hme): Optimize with separate mutex per collection. - std::mutex connection_mutex; - - SenderMapType message_send_connections_; - SenderMapType transfer_send_connections_; - SenderMapType available_message_send_connections_; - SenderMapType available_transfer_send_connections_; - - ReceiverMapType message_receive_connections_; - ReceiverMapType transfer_receive_connections_; -}; - -} // namespace ray - -#endif // RAY_OBJECT_MANAGER_CONNECTION_POOL_H diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc index e25ba99c7..67a830a32 100644 --- a/src/ray/object_manager/object_manager.cc +++ b/src/ray/object_manager/object_manager.cc @@ -17,9 +17,8 @@ ObjectManager::ObjectManager(asio::io_service &main_service, store_notification_(main_service, config_.store_socket_name), buffer_pool_(config_.store_socket_name, config_.object_chunk_size), rpc_work_(rpc_service_), - connection_pool_(), gen_(std::chrono::high_resolution_clock::now().time_since_epoch().count()), - object_manager_server_("object_manager", config_.object_manager_port), + object_manager_server_("ObjectManager", config_.object_manager_port), object_manager_service_(rpc_service_, *this), client_call_manager_(main_service) { RAY_CHECK(config_.rpc_service_threads_number > 0); @@ -831,7 +830,6 @@ std::string ObjectManager::DebugString() const { result << "\n" << object_directory_->DebugString(); result << "\n" << store_notification_.DebugString(); result << "\n" << buffer_pool_.DebugString(); - result << "\n" << connection_pool_.DebugString(); return result.str(); } @@ -847,7 +845,6 @@ void ObjectManager::RecordMetrics() const { {{stats::ValueTypeKey, "num_pull_requests"}}); stats::ObjectManagerStats().Record(profile_events_.size(), {{stats::ValueTypeKey, "num_profile_events"}}); - connection_pool_.RecordMetrics(); } } // namespace ray diff --git a/src/ray/object_manager/object_manager.h b/src/ray/object_manager/object_manager.h index f40fd43b1..019b92bb9 100644 --- a/src/ray/object_manager/object_manager.h +++ b/src/ray/object_manager/object_manager.h @@ -16,15 +16,13 @@ #include "plasma/client.h" -#include "ray/common/client_connection.h" #include "ray/common/id.h" +#include "ray/common/ray_config.h" #include "ray/common/status.h" -#include "ray/object_manager/connection_pool.h" #include "ray/object_manager/format/object_manager_generated.h" #include "ray/object_manager/object_buffer_pool.h" #include "ray/object_manager/object_directory.h" -#include "ray/object_manager/object_manager_client_connection.h" #include "ray/object_manager/object_store_notification_manager.h" #include "ray/rpc/object_manager/object_manager_client.h" #include "ray/rpc/object_manager/object_manager_server.h" @@ -203,23 +201,6 @@ class ObjectManager : public ObjectManagerInterface, /// \return Void. void TryPull(const ObjectID &object_id); - /// Add a connection to a remote object manager. - /// This is invoked by an external server. - /// - /// \param conn The connection. - /// \return Status of whether the connection was successfully established. - void ProcessNewClient(TcpClientConnection &conn); - - /// Process messages sent from other nodes. We only establish - /// transfer connections using this method; all other transfer communication - /// is done separately. - /// - /// \param conn The connection. - /// \param message_type The message type. - /// \param message A pointer set to the beginning of the message. - void ProcessClientMessage(std::shared_ptr &conn, - int64_t message_type, const uint8_t *message); - /// Cancels all requests (Push/Pull) associated with the given ObjectID. This /// method is idempotent. /// @@ -277,7 +258,8 @@ class ObjectManager : public ObjectManagerInterface, }; struct WaitState { - WaitState(asio::io_service &service, int64_t timeout_ms, const WaitCallback &callback) + WaitState(boost::asio::io_service &service, int64_t timeout_ms, + const WaitCallback &callback) : timeout_ms(timeout_ms), timeout_timer(std::unique_ptr( new boost::asio::deadline_timer( @@ -337,20 +319,6 @@ class ObjectManager : public ObjectManagerInterface, /// Register object remove with directory. void NotifyDirectoryObjectDeleted(const ObjectID &object_id); - /// Part of an asynchronous sequence of Pull methods. - /// Uses an existing connection or creates a connection to ClientID. - /// Executes on main_service_ thread. - void PullEstablishConnection(const ObjectID &object_id, const ClientID &client_id); - - /// Asynchronously send a pull request via remote object manager connection. - /// Executes on main_service_ thread. - /// - /// \param object_id The ID of the object request. - /// \param conn The connection to the remote object manager. - /// \return Void. - void PullSendRequest(const ObjectID &object_id, - std::shared_ptr &conn); - /// This is used to notify the main thread that the sending of a chunk has /// completed. /// @@ -383,15 +351,6 @@ class ObjectManager : public ObjectManagerInterface, uint64_t chunk_index, double start_time_us, double end_time_us, ray::Status status); - /// Execute a receive on the receive_service_ thread pool. - ray::Status ExecuteReceiveObject(const ClientID &client_id, const ObjectID &object_id, - uint64_t data_size, uint64_t metadata_size, - uint64_t chunk_index, TcpClientConnection &conn); - - /// Handles freeing objects request. - void ReceiveFreeRequest(std::shared_ptr &conn, - const uint8_t *message); - /// Handle Push task timeout. void HandlePushTaskTimeout(const ObjectID &object_id, const ClientID &client_id); @@ -415,9 +374,6 @@ class ObjectManager : public ObjectManagerInterface, /// Data copy operations during request are done in this thread pool. std::vector rpc_threads_; - /// Connection pool for reusing outgoing connections to remote object managers. - ConnectionPool connection_pool_; - /// Mapping from locally available objects to information about those objects /// including when the object was last pushed to other object managers. std::unordered_map local_objects_; @@ -461,7 +417,7 @@ class ObjectManager : public ObjectManagerInterface, /// The client call manager used to deal with reply. rpc::ClientCallManager client_call_manager_; - /// clientID - object manager gRPC client. + /// Client id - object manager gRPC client. std::unordered_map> remote_object_manager_clients_; }; diff --git a/src/ray/object_manager/object_manager_client_connection.cc b/src/ray/object_manager/object_manager_client_connection.cc deleted file mode 100644 index dadfd72ce..000000000 --- a/src/ray/object_manager/object_manager_client_connection.cc +++ /dev/null @@ -1,28 +0,0 @@ -#include "ray/object_manager/object_manager_client_connection.h" - -namespace ray { - -uint64_t SenderConnection::id_counter_; - -std::shared_ptr SenderConnection::Create( - boost::asio::io_service &io_service, const ClientID &client_id, const std::string &ip, - uint16_t port) { - boost::asio::ip::tcp::socket socket(io_service); - Status status = TcpConnect(socket, ip, port); - if (status.ok()) { - std::shared_ptr conn = - TcpServerConnection::Create(std::move(socket)); - return std::make_shared(std::move(conn), client_id); - } else { - return nullptr; - } -}; - -SenderConnection::SenderConnection(std::shared_ptr conn, - const ClientID &client_id) - : conn_(conn) { - client_id_ = client_id; - connection_id_ = SenderConnection::id_counter_++; -}; - -} // namespace ray diff --git a/src/ray/object_manager/object_manager_client_connection.h b/src/ray/object_manager/object_manager_client_connection.h deleted file mode 100644 index c08b99748..000000000 --- a/src/ray/object_manager/object_manager_client_connection.h +++ /dev/null @@ -1,91 +0,0 @@ -#ifndef RAY_OBJECT_MANAGER_OBJECT_MANAGER_CLIENT_CONNECTION_H -#define RAY_OBJECT_MANAGER_OBJECT_MANAGER_CLIENT_CONNECTION_H - -#include -#include -#include - -#include -#include -#include -#include - -#include "ray/common/client_connection.h" -#include "ray/common/id.h" -#include "ray/common/ray_config.h" - -namespace ray { - -// TODO(ekl) this class can be replaced with a plain ClientConnection -class SenderConnection : public boost::enable_shared_from_this { - public: - /// Create a connection for sending data to other object managers. - /// - /// \param io_service The service to which the created socket should attach. - /// \param client_id The ClientID of the remote node. - /// \param ip The ip address of the remote node server. - /// \param port The port of the remote node server. - /// \return A connection to the remote object manager. This is null if the - /// connection was unsuccessful. - static std::shared_ptr Create(boost::asio::io_service &io_service, - const ClientID &client_id, - const std::string &ip, uint16_t port); - - /// \param socket A reference to the socket created by the static Create method. - /// \param client_id The ClientID of the remote node. - SenderConnection(std::shared_ptr conn, const ClientID &client_id); - - /// Write a message to the client. - /// - /// \param type The message type (e.g., a flatbuffer enum). - /// \param length The size in bytes of the message. - /// \param message A pointer to the message buffer. - /// \return Status. - ray::Status WriteMessage(int64_t type, uint64_t length, const uint8_t *message) { - return conn_->WriteMessage(type, length, message); - } - - /// Write a message to the client asynchronously. - /// - /// \param type The message type (e.g., a flatbuffer enum). - /// \param length The size in bytes of the message. - /// \param message A pointer to the message buffer. - /// \param handler A callback to run on write completion. - void WriteMessageAsync(int64_t type, int64_t length, const uint8_t *message, - const std::function &handler) { - conn_->WriteMessageAsync(type, length, message, handler); - } - - /// Write a buffer to this connection. - /// - /// \param buffer The buffer. - /// \return Status. - Status WriteBuffer(const std::vector &buffer) { - return conn_->WriteBuffer(buffer); - } - - /// Read a buffer from this connection. - /// - /// \param buffer The buffer. - /// \return Status. - Status ReadBuffer(const std::vector &buffer) { - return conn_->ReadBuffer(buffer); - } - - /// \return The ClientID of this connection. - const ClientID &GetClientId() { return client_id_; } - - private: - bool operator==(const SenderConnection &rhs) const { - return connection_id_ == rhs.connection_id_; - } - - static uint64_t id_counter_; - uint64_t connection_id_; - ClientID client_id_; - std::shared_ptr conn_; -}; - -} // namespace ray - -#endif // RAY_OBJECT_MANAGER_OBJECT_MANAGER_CLIENT_CONNECTION_H