diff --git a/src/plasma/plasma_manager.cc b/src/plasma/plasma_manager.cc index 69a205399..55439f2f1 100644 --- a/src/plasma/plasma_manager.cc +++ b/src/plasma/plasma_manager.cc @@ -22,6 +22,7 @@ /* C++ includes. */ #include +#include #include #include "uthash.h" @@ -139,14 +140,6 @@ void process_data_request(event_loop *loop, int64_t metadata_size, ClientConnection *conn); -/** Entry of the hashtable of objects that are available locally. */ -typedef struct { - /** Object id of this object. */ - ObjectID object_id; - /** Handle for the uthash table. */ - UT_hash_handle hh; -} AvailableObject; - typedef struct { /** The ID of the object we are fetching or waiting for. */ ObjectID object_id; @@ -159,9 +152,6 @@ typedef struct { * manager_vector in the retry handler, in case the current attempt fails to * contact a manager. */ int next_manager; - /** Handle for the uthash table in the manager state that keeps track of - * outstanding fetch requests. */ - UT_hash_handle hh; } FetchRequest; /** @@ -240,17 +230,18 @@ struct PlasmaManagerState { const char *addr; /** Our port. */ int port; - /** Hash table of outstanding fetch requests. The key is the object ID. The + /** Unordered map of outstanding fetch requests. The key is the object ID. The * value is the data needed to perform the fetch. */ - FetchRequest *fetch_requests; + std::unordered_map fetch_requests; /** A hash table mapping object IDs to a vector of the wait requests that * are waiting for the object to arrive locally. */ ObjectWaitRequests *object_wait_requests_local; /** A hash table mapping object IDs to a vector of the wait requests that * are waiting for the object to be available somewhere in the system. */ ObjectWaitRequests *object_wait_requests_remote; - /** Initialize an empty hash map for the cache of local available object. */ - AvailableObject *local_available_objects; + /** Initialize an empty unordered set for the cache of local available object. + */ + std::unordered_set local_available_objects; /** The time (in milliseconds since the Unix epoch) when the most recent * heartbeat was sent. */ int64_t previous_heartbeat_time; @@ -501,10 +492,17 @@ FetchRequest *create_fetch_request(PlasmaManagerState *manager_state, return fetch_req; } +/** + * Remove a fetch request from the table of fetch requests. + * + * @param manager_state The state of the manager. + * @param fetch_req The fetch request to remove. + * @return Void. + */ void remove_fetch_request(PlasmaManagerState *manager_state, FetchRequest *fetch_req) { /* Remove the fetch request from the table of fetch requests. */ - HASH_DELETE(hh, manager_state->fetch_requests, fetch_req); + manager_state->fetch_requests.erase(fetch_req->object_id); /* Free the fetch request and everything in it. */ for (int i = 0; i < fetch_req->manager_count; ++i) { free(fetch_req->manager_vector[i]); @@ -521,14 +519,12 @@ PlasmaManagerState *PlasmaManagerState_init(const char *store_socket_name, int manager_port, const char *redis_primary_addr, int redis_primary_port) { - PlasmaManagerState *state = - (PlasmaManagerState *) malloc(sizeof(PlasmaManagerState)); + PlasmaManagerState *state = new PlasmaManagerState(); state->loop = event_loop_create(); state->plasma_conn = new plasma::PlasmaClient(); ARROW_CHECK_OK(state->plasma_conn->Connect(store_socket_name, "", PLASMA_DEFAULT_RELEASE_DELAY)); state->manager_connections = NULL; - state->fetch_requests = NULL; state->object_wait_requests_local = NULL; state->object_wait_requests_remote = NULL; if (redis_primary_addr) { @@ -558,8 +554,6 @@ PlasmaManagerState *PlasmaManagerState_init(const char *store_socket_name, } state->addr = manager_addr; state->port = manager_port; - /* Initialize an empty hash map for the cache of local available objects. */ - state->local_available_objects = NULL; /* Subscribe to notifications about sealed objects. */ int plasma_fd; ARROW_CHECK_OK(state->plasma_conn->Subscribe(&plasma_fd)); @@ -586,17 +580,13 @@ void PlasmaManagerState_free(PlasmaManagerState *state) { ClientConnection_free(manager_conn); } - if (state->fetch_requests != NULL) { - FetchRequest *fetch_req, *tmp; - HASH_ITER(hh, state->fetch_requests, fetch_req, tmp) { - remove_fetch_request(state, fetch_req); - } - } - - AvailableObject *entry, *tmp_object_entry; - HASH_ITER(hh, state->local_available_objects, entry, tmp_object_entry) { - HASH_DELETE(hh, state->local_available_objects, entry); - free(entry); + /* We have to be careful here because remove_fetch_request modifies + * state->fetch_requests in place. */ + auto it = state->fetch_requests.begin(); + while (it != state->fetch_requests.end()) { + auto next_it = std::next(it, 1); + remove_fetch_request(state, it->second); + it = next_it; } ObjectWaitRequests *wait_reqs, *tmp_wait_reqs; @@ -612,7 +602,7 @@ void PlasmaManagerState_free(PlasmaManagerState *state) { ARROW_CHECK_OK(state->plasma_conn->Disconnect()); delete state->plasma_conn; event_loop_destroy(state->loop); - free(state); + delete state; } event_loop *get_event_loop(PlasmaManagerState *state) { @@ -1008,7 +998,7 @@ int fetch_timeout_handler(event_loop *loop, timer_id id, void *context) { /* Allocate a vector of object IDs to resend requests for location * notifications. */ int num_object_ids_to_request = 0; - int num_object_ids = HASH_COUNT(manager_state->fetch_requests); + int num_object_ids = manager_state->fetch_requests.size(); /* This is allocating more space than necessary, but we do not know the exact * number of object IDs to request notifications for yet. */ ObjectID *object_ids_to_request = @@ -1016,8 +1006,9 @@ int fetch_timeout_handler(event_loop *loop, timer_id id, void *context) { /* Loop over the fetch requests and reissue requests for objects whose * locations we know. */ - FetchRequest *fetch_req, *tmp; - HASH_ITER(hh, manager_state->fetch_requests, fetch_req, tmp) { + for (auto it = manager_state->fetch_requests.begin(); + it != manager_state->fetch_requests.end(); it++) { + FetchRequest *fetch_req = it->second; if (fetch_req->manager_count > 0) { request_transfer_from(manager_state, fetch_req); /* If we've tried all of the managers that we know about for this object, @@ -1046,10 +1037,7 @@ int fetch_timeout_handler(event_loop *loop, timer_id id, void *context) { } bool is_object_local(PlasmaManagerState *state, ObjectID object_id) { - AvailableObject *entry; - HASH_FIND(hh, state->local_available_objects, &object_id, sizeof(object_id), - entry); - return entry != NULL; + return state->local_available_objects.count(object_id) > 0; } void request_transfer(ObjectID object_id, @@ -1060,16 +1048,15 @@ void request_transfer(ObjectID object_id, /* This callback is called from object_table_subscribe, which guarantees that * the manager vector contains at least one element. */ CHECK(manager_count >= 1); - FetchRequest *fetch_req; - HASH_FIND(hh, manager_state->fetch_requests, &object_id, sizeof(object_id), - fetch_req); + auto it = manager_state->fetch_requests.find(object_id); if (is_object_local(manager_state, object_id)) { /* If the object is already here, then the fetch request should have been * removed. */ - CHECK(fetch_req == NULL); + CHECK(it == manager_state->fetch_requests.end()); return; } + FetchRequest *fetch_req = it->second; /* If the object is not present, then the fetch request should still be here. * TODO(rkn): We actually have to remove this check to handle the rare @@ -1107,15 +1094,12 @@ void call_request_transfer(ObjectID object_id, const char *manager_vector[], void *context) { PlasmaManagerState *manager_state = (PlasmaManagerState *) context; - FetchRequest *fetch_req; /* Check that there isn't already a fetch request for this object. */ - HASH_FIND(hh, manager_state->fetch_requests, &object_id, sizeof(object_id), - fetch_req); - CHECK(fetch_req == NULL); + auto it = manager_state->fetch_requests.find(object_id); + CHECK(it == manager_state->fetch_requests.end()); /* Create a fetch request. */ - fetch_req = create_fetch_request(manager_state, object_id); - HASH_ADD(hh, manager_state->fetch_requests, object_id, - sizeof(fetch_req->object_id), fetch_req); + FetchRequest *fetch_req = create_fetch_request(manager_state, object_id); + manager_state->fetch_requests[object_id] = fetch_req; request_transfer(object_id, manager_count, manager_vector, context); } @@ -1147,10 +1131,8 @@ void object_table_subscribe_callback(ObjectID object_id, void *context) { PlasmaManagerState *manager_state = (PlasmaManagerState *) context; /* Run the callback for fetch requests if there is a fetch request. */ - FetchRequest *fetch_req; - HASH_FIND(hh, manager_state->fetch_requests, &object_id, sizeof(object_id), - fetch_req); - if (fetch_req != NULL) { + auto it = manager_state->fetch_requests.find(object_id); + if (it != manager_state->fetch_requests.end()) { request_transfer(object_id, manager_count, manager_vector, context); } /* Run the callback for wait requests. */ @@ -1177,18 +1159,15 @@ void process_fetch_requests(ClientConnection *client_conn, } /* Check if this object is already being fetched. If so, do nothing. */ - FetchRequest *entry; - HASH_FIND(hh, manager_state->fetch_requests, &obj_id, sizeof(obj_id), - entry); - if (entry != NULL) { + auto it = manager_state->fetch_requests.find(obj_id); + if (it != manager_state->fetch_requests.end()) { continue; } /* Add an entry to the fetch requests data structure to indidate that the * object is being fetched. */ - entry = create_fetch_request(manager_state, obj_id); - HASH_ADD(hh, manager_state->fetch_requests, object_id, - sizeof(entry->object_id), entry); + FetchRequest *entry = create_fetch_request(manager_state, obj_id); + manager_state->fetch_requests[obj_id] = entry; /* Add this object ID to the list of object IDs to request notifications for * from the object table. */ object_ids_to_request[num_object_ids_to_request] = obj_id; @@ -1362,13 +1341,7 @@ void process_status_request(ClientConnection *client_conn, void process_delete_object_notification(PlasmaManagerState *state, ObjectID object_id) { - AvailableObject *entry; - HASH_FIND(hh, state->local_available_objects, &object_id, sizeof(object_id), - entry); - if (entry != NULL) { - HASH_DELETE(hh, state->local_available_objects, entry); - free(entry); - } + state->local_available_objects.erase(object_id); /* Remove this object from the (redis) object table. */ if (state->db) { @@ -1425,10 +1398,7 @@ void process_add_object_notification(PlasmaManagerState *state, int64_t data_size, int64_t metadata_size, unsigned char *digest) { - AvailableObject *entry = (AvailableObject *) malloc(sizeof(AvailableObject)); - entry->object_id = object_id; - HASH_ADD(hh, state->local_available_objects, object_id, sizeof(ObjectID), - entry); + state->local_available_objects.insert(object_id); /* Add this object to the (redis) object table. */ if (state->db) { @@ -1438,11 +1408,9 @@ void process_add_object_notification(PlasmaManagerState *state, } /* If we were trying to fetch this object, finish up the fetch request. */ - FetchRequest *fetch_req; - HASH_FIND(hh, state->fetch_requests, &object_id, sizeof(object_id), - fetch_req); - if (fetch_req != NULL) { - remove_fetch_request(state, fetch_req); + auto it = state->fetch_requests.find(object_id); + if (it != state->fetch_requests.end()) { + remove_fetch_request(state, it->second); /* TODO(rkn): We also really should unsubscribe from the object table. */ }