diff --git a/src/plasma/plasma_manager.cc b/src/plasma/plasma_manager.cc index b47785f5f..75b374ccf 100644 --- a/src/plasma/plasma_manager.cc +++ b/src/plasma/plasma_manager.cc @@ -26,7 +26,6 @@ #include #include -#include "uthash.h" #include "common_protocol.h" #include "io.h" #include "net.h" @@ -207,7 +206,7 @@ struct PlasmaManagerState { /** Hash table of all contexts for active connections to * other plasma managers. These are used for writing data to * other plasma stores. */ - ClientConnection *manager_connections; + std::unordered_map manager_connections; DBHandle *db; /** Our address. */ const char *addr; @@ -264,9 +263,7 @@ struct ClientConnection { /** Fields specific to connections to plasma managers. Key that uniquely * identifies the plasma manager that we're connected to. We will use the * string
: as an identifier. */ - char *ip_addr_port; - /** Handle for the uthash table. */ - UT_hash_handle manager_hh; + std::string ip_addr_port; }; /** @@ -281,7 +278,7 @@ struct ClientConnection { */ ClientConnection *ClientConnection_init(PlasmaManagerState *state, int client_sock, - const char *client_key); + std::string const &client_key); /** * Destroys a plasma client and its connection. @@ -452,7 +449,6 @@ PlasmaManagerState *PlasmaManagerState_init(const char *store_socket_name, state->plasma_conn = new plasma::PlasmaClient(); ARROW_CHECK_OK(state->plasma_conn->Connect(store_socket_name, "", PLASMA_DEFAULT_RELEASE_DELAY)); - state->manager_connections = NULL; if (redis_primary_addr) { /* Get the manager port as a string. */ std::string manager_address_str = @@ -498,10 +494,13 @@ void PlasmaManagerState_free(PlasmaManagerState *state) { state->db = NULL; } - ClientConnection *manager_conn, *tmp_manager_conn; - HASH_ITER(manager_hh, state->manager_connections, manager_conn, - tmp_manager_conn) { - ClientConnection_free(manager_conn); + /* We have to be careful here because ClientConnection_free modifies + * state->manager_connections in place. */ + auto cc_it = state->manager_connections.begin(); + while (cc_it != state->manager_connections.end()) { + auto next_it = std::next(cc_it, 1); + ClientConnection_free(cc_it->second); + cc_it = next_it; } /* We have to be careful here because remove_fetch_request modifies @@ -718,16 +717,17 @@ ClientConnection *get_manager_connection(PlasmaManagerState *state, */ std::string ip_addr_port = std::string(ip_addr) + ":" + std::to_string(port); ClientConnection *manager_conn; - HASH_FIND(manager_hh, state->manager_connections, ip_addr_port.c_str(), - ip_addr_port.length(), manager_conn); - if (!manager_conn) { + auto cc_it = state->manager_connections.find(ip_addr_port); + if (cc_it == state->manager_connections.end()) { /* If we don't already have a connection to this manager, start one. */ int fd = connect_inet_sock(ip_addr, port); if (fd < 0) { return NULL; } - manager_conn = ClientConnection_init(state, fd, ip_addr_port.c_str()); + manager_conn = ClientConnection_init(state, fd, ip_addr_port); + } else { + manager_conn = cc_it->second; } return manager_conn; } @@ -1332,7 +1332,7 @@ void process_object_notification(event_loop *loop, * into two structs, one for workers and one for other plasma managers. */ ClientConnection *ClientConnection_init(PlasmaManagerState *state, int client_sock, - const char *client_key) { + std::string const &client_key) { /* Create a new data connection context per client. */ ClientConnection *conn = new ClientConnection(); conn->manager_state = state; @@ -1340,9 +1340,8 @@ ClientConnection *ClientConnection_init(PlasmaManagerState *state, conn->fd = client_sock; conn->num_return_objects = 0; - conn->ip_addr_port = strdup(client_key); - HASH_ADD_KEYPTR(manager_hh, conn->manager_state->manager_connections, - conn->ip_addr_port, strlen(conn->ip_addr_port), conn); + conn->ip_addr_port = client_key; + state->manager_connections[client_key] = conn; return conn; } @@ -1363,7 +1362,7 @@ ClientConnection *ClientConnection_listen(event_loop *loop, void ClientConnection_free(ClientConnection *client_conn) { PlasmaManagerState *state = client_conn->manager_state; - HASH_DELETE(manager_hh, state->manager_connections, client_conn); + state->manager_connections.erase(client_conn->ip_addr_port); client_conn->pending_object_transfers.clear(); @@ -1374,7 +1373,6 @@ void ClientConnection_free(ClientConnection *client_conn) { } /* Close the manager connection and free the remaining state. */ close(client_conn->fd); - free(client_conn->ip_addr_port); delete client_conn; } diff --git a/src/plasma/plasma_manager.h b/src/plasma/plasma_manager.h index 7eedc33a9..672c1edfb 100644 --- a/src/plasma/plasma_manager.h +++ b/src/plasma/plasma_manager.h @@ -2,8 +2,6 @@ #define PLASMA_MANAGER_H #include -#include "uthash.h" -#include "utarray.h" #ifndef RAY_NUM_RETRIES #define NUM_RETRIES 5