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

Data structure updates to plasma manager (#937)

Browse Source 
* Implemented local_available_objects as an unordered set

* Implemented fetch_requests as an unordered map

* Fixed bug and changed fetch_requests from pointer to object

* free(PlasmaManagerState *) -> delete PlasmaManagerState *

* removed unnecessary newline

* Make local_available_objects not a pointer.

* Attempt to safely iterate over unordered_map and remove elements.
This commit is contained in:
Peter Schafhalter
2017-09-15 20:09:29 -07:00
committed by Robert Nishihara
parent 6601bb5f9e
commit 241612709e
1 changed files with 46 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/plasma/plasma_manager.cc
+46 -78
View File
@@ -22,6 +22,7 @@
/* C++ includes. */
#include <unordered_map>
#include <unordered_set>
#include <vector>
#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<ObjectID, FetchRequest *, UniqueIDHasher> 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<ObjectID, UniqueIDHasher> 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. */
}