wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-09-09 11:32:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

Remove unused connection file in object manager (#5123)

Browse Source
This commit is contained in:
Joey Jiang
2019-07-08 10:59:36 +08:00
committed by Hao Chen
parent 893744b3be
commit 274233962f
6 changed files with 5 additions and 497 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/object_manager/connection_pool.cc
-171
View File
@@ -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<TcpClientConnection> &conn) {
std::unique_lock<std::mutex> 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<TcpClientConnection> conn) {
std::unique_lock<std::mutex> 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<SenderConnection> &conn) {
std::unique_lock<std::mutex> 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<SenderConnection> &conn) {
std::unique_lock<std::mutex> 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<SenderConnection> *conn) {
std::unique_lock<std::mutex> 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<SenderConnection> &conn) {
std::unique_lock<std::mutex> 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<TcpClientConnection> conn) {
conn_map[client_id].push_back(std::move(conn));
}
void ConnectionPool::Add(SenderMapType &conn_map, const ClientID &client_id,
std::shared_ptr<SenderConnection> conn) {
conn_map[client_id].push_back(std::move(conn));
}
void ConnectionPool::Remove(ReceiverMapType &conn_map, const ClientID &client_id,
std::shared_ptr<TcpClientConnection> &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<int64_t>(connections.size())) {
return;
}
connections.erase(connections.begin() + pos);
}
void ConnectionPool::Remove(SenderMapType &conn_map, const ClientID &client_id,
const std::shared_ptr<SenderConnection> &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<int64_t>(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<SenderConnection> ConnectionPool::Borrow(SenderMapType &conn_map,
const ClientID &client_id) {
std::shared_ptr<SenderConnection> 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<SenderConnection> 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
src/ray/object_manager/connection_pool.h
-155
View File
@@ -1,155 +0,0 @@
#ifndef RAY_OBJECT_MANAGER_CONNECTION_POOL_H
#define RAY_OBJECT_MANAGER_CONNECTION_POOL_H
#include <algorithm>
#include <cstdint>
#include <deque>
#include <map>
#include <memory>
#include <thread>
#include <boost/asio.hpp>
#include <boost/asio/error.hpp>
#include <boost/bind.hpp>
#include "ray/common/id.h"
#include "ray/common/status.h"
#include <mutex>
#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<void(std::shared_ptr<SenderConnection>)>;
using FailureCallback = std::function<void()>;
/// 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<TcpClientConnection> &conn);
/// Remove a receiver connection.
///
/// \param conn The actual connection.
void RemoveReceiver(std::shared_ptr<TcpClientConnection> 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<SenderConnection> &conn);
/// Remove a sender connection.
///
/// \param conn The actual connection.
void RemoveSender(const std::shared_ptr<SenderConnection> &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<SenderConnection> *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<SenderConnection> &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<SenderConnection> 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<ray::ClientID, std::vector<std::shared_ptr<SenderConnection>>>;
using ReceiverMapType =
std::unordered_map<ray::ClientID,
std::vector<std::shared_ptr<TcpClientConnection>>>;
/// Adds a receiver for ClientID to the given map.
void Add(ReceiverMapType &conn_map, const ClientID &client_id,
std::shared_ptr<TcpClientConnection> conn);
/// Adds a sender for ClientID to the given map.
void Add(SenderMapType &conn_map, const ClientID &client_id,
std::shared_ptr<SenderConnection> conn);
/// Removes the given receiver for ClientID from the given map.
void Remove(ReceiverMapType &conn_map, const ClientID &client_id,
std::shared_ptr<TcpClientConnection> &conn);
/// Removes the given sender for ClientID from the given map.
void Remove(SenderMapType &conn_map, const ClientID &client_id,
const std::shared_ptr<SenderConnection> &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<SenderConnection> 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<SenderConnection> 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
src/ray/object_manager/object_manager.cc
+1 -4
View File
@@ -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
src/ray/object_manager/object_manager.h
+4 -48
View File
@@ -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<TcpClientConnection> &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<boost::asio::deadline_timer>(
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<SenderConnection> &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<TcpClientConnection> &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<std::thread> 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<ObjectID, LocalObjectInfo> 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<ClientID, std::shared_ptr<rpc::ObjectManagerClient>>
remote_object_manager_clients_;
};
src/ray/object_manager/object_manager_client_connection.cc
-28
View File
@@ -1,28 +0,0 @@
#include "ray/object_manager/object_manager_client_connection.h"
namespace ray {
uint64_t SenderConnection::id_counter_;
std::shared_ptr<SenderConnection> 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<TcpServerConnection> conn =
TcpServerConnection::Create(std::move(socket));
return std::make_shared<SenderConnection>(std::move(conn), client_id);
} else {
return nullptr;
}
};
SenderConnection::SenderConnection(std::shared_ptr<TcpServerConnection> conn,
const ClientID &client_id)
: conn_(conn) {
client_id_ = client_id;
connection_id_ = SenderConnection::id_counter_++;
};
} // namespace ray
src/ray/object_manager/object_manager_client_connection.h
-91
View File
@@ -1,91 +0,0 @@
#ifndef RAY_OBJECT_MANAGER_OBJECT_MANAGER_CLIENT_CONNECTION_H
#define RAY_OBJECT_MANAGER_OBJECT_MANAGER_CLIENT_CONNECTION_H
#include <deque>
#include <memory>
#include <unordered_map>
#include <boost/asio.hpp>
#include <boost/asio/error.hpp>
#include <boost/bind.hpp>
#include <boost/enable_shared_from_this.hpp>
#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<SenderConnection> {
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<SenderConnection> 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<TcpServerConnection> 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<void(const ray::Status &)> &handler) {
conn_->WriteMessageAsync(type, length, message, handler);
}
/// Write a buffer to this connection.
///
/// \param buffer The buffer.
/// \return Status.
Status WriteBuffer(const std::vector<boost::asio::const_buffer> &buffer) {
return conn_->WriteBuffer(buffer);
}
/// Read a buffer from this connection.
///
/// \param buffer The buffer.
/// \return Status.
Status ReadBuffer(const std::vector<boost::asio::mutable_buffer> &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<TcpServerConnection> conn_;
};
} // namespace ray
#endif // RAY_OBJECT_MANAGER_OBJECT_MANAGER_CLIENT_CONNECTION_H