From e36de2dad1c7ded014692165cf5dc310cf6fb21a Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Sun, 23 Apr 2017 22:24:05 -0700 Subject: [PATCH] Convert object table to STL (#480) * convert object table to stl * temp commit * fix * comments * linting --- src/plasma/eviction_policy.cc | 4 +-- src/plasma/plasma.cc | 9 +++++ src/plasma/plasma.h | 14 +++++++- src/plasma/plasma_store.cc | 66 +++++++++++++---------------------- 4 files changed, 47 insertions(+), 46 deletions(-) diff --git a/src/plasma/eviction_policy.cc b/src/plasma/eviction_policy.cc index d82f843e6..da33ef447 100644 --- a/src/plasma/eviction_policy.cc +++ b/src/plasma/eviction_policy.cc @@ -122,9 +122,7 @@ int64_t EvictionState_choose_objects_to_evict( break; } /* Find the object table entry for this object. */ - object_table_entry *entry; - HASH_FIND(handle, plasma_store_info->objects, &element->object_id, - sizeof(element->object_id), entry); + object_table_entry *entry = plasma_store_info->objects[element->object_id]; /* Update the cumulative bytes and the number of objects so far. */ num_bytes += (entry->info.data_size + entry->info.metadata_size); num_objects += 1; diff --git a/src/plasma/plasma.cc b/src/plasma/plasma.cc index 83dce6ce1..820d47c78 100644 --- a/src/plasma/plasma.cc +++ b/src/plasma/plasma.cc @@ -39,3 +39,12 @@ uint8_t *create_object_info_buffer(ObjectInfoT *object_info) { memcpy(notification + sizeof(int64_t), fbb.GetBufferPointer(), fbb.GetSize()); return notification; } + +object_table_entry *get_object_table_entry(PlasmaStoreInfo *store_info, + ObjectID object_id) { + auto it = store_info->objects.find(object_id); + if (it == store_info->objects.end()) { + return NULL; + } + return it->second; +} diff --git a/src/plasma/plasma.h b/src/plasma/plasma.h index 3067a275b..34efd4060 100644 --- a/src/plasma/plasma.h +++ b/src/plasma/plasma.h @@ -121,12 +121,24 @@ typedef struct { /** The plasma store information that is exposed to the eviction policy. */ typedef struct { /** Objects that are in the Plasma store. */ - object_table_entry *objects; + std::unordered_map objects; /** The amount of memory (in bytes) that we allow to be allocated in the * store. */ int64_t memory_capacity; } PlasmaStoreInfo; +/** + * Get an entry from the object table and return NULL if the object_id + * is not present. + * + * @param store_info The PlasmaStoreInfo that contains the object table. + * @param object_id The object_id of the entry we are looking for. + * @return The entry associated with the object_id or NULL if the object_id + * is not present. + */ +object_table_entry *get_object_table_entry(PlasmaStoreInfo *store_info, + ObjectID object_id); + /** * Print a warning if the status is less than zero. This should be used to check * the success of messages sent to plasma clients. We print a warning instead of diff --git a/src/plasma/plasma_store.cc b/src/plasma/plasma_store.cc index a88d7a5b7..d5f273dec 100644 --- a/src/plasma/plasma_store.cc +++ b/src/plasma/plasma_store.cc @@ -146,10 +146,9 @@ GetRequest::GetRequest(Client *client, PlasmaStoreState::PlasmaStoreState(event_loop *loop, int64_t system_memory) : loop(loop), pending_notifications(NULL), - plasma_store_info((PlasmaStoreInfo *) malloc(sizeof(PlasmaStoreInfo))), + plasma_store_info(new PlasmaStoreInfo()), eviction_state(EvictionState_init()), builder(make_protocol_builder()) { - this->plasma_store_info->objects = NULL; this->plasma_store_info->memory_capacity = system_memory; utarray_new(this->input_buffer, &byte_icd); @@ -159,11 +158,9 @@ void PlasmaStoreState_free(PlasmaStoreState *state) { /* Here we only clean up objects that need to be cleaned * up to make the valgrind warnings go away. Objects that * are still reachable are not cleaned up. */ - object_table_entry *entry, *tmp; - HASH_ITER(handle, state->plasma_store_info->objects, entry, tmp) { - HASH_DELETE(handle, state->plasma_store_info->objects, entry); - utarray_free(entry->clients); - delete entry; + for (const auto &it : state->plasma_store_info->objects) { + utarray_free(it.second->clients); + delete it.second; } NotificationQueue *queue, *temp_queue; HASH_ITER(hh, state->pending_notifications, queue, temp_queue) { @@ -216,11 +213,7 @@ int create_object(Client *client_context, PlasmaObject *result) { LOG_DEBUG("creating object"); /* TODO(pcm): add ObjectID here */ PlasmaStoreState *plasma_state = client_context->plasma_state; - object_table_entry *entry; - /* TODO(swang): Return these error to the client instead of exiting. */ - HASH_FIND(handle, plasma_state->plasma_store_info->objects, &obj_id, - sizeof(obj_id), entry); - if (entry != NULL) { + if (plasma_state->plasma_store_info->objects.count(obj_id) != 0) { /* There is already an object with the same ID in the Plasma Store, so * ignore this requst. */ return PlasmaError_ObjectExists; @@ -258,7 +251,7 @@ int create_object(Client *client_context, get_malloc_mapinfo(pointer, &fd, &map_size, &offset); assert(fd != -1); - entry = new object_table_entry(); + object_table_entry *entry = new object_table_entry(); entry->object_id = obj_id; entry->info.object_id = std::string((char *) &obj_id.id[0], sizeof(obj_id)); entry->info.data_size = data_size; @@ -270,8 +263,7 @@ int create_object(Client *client_context, entry->offset = offset; entry->state = PLASMA_CREATED; utarray_new(entry->clients, &client_icd); - HASH_ADD(handle, plasma_state->plasma_store_info->objects, object_id, - sizeof(ObjectID), entry); + plasma_state->plasma_store_info->objects[obj_id] = entry; result->handle.store_fd = fd; result->handle.mmap_size = map_size; result->data_offset = offset; @@ -379,9 +371,8 @@ void update_object_get_requests(PlasmaStoreState *store_state, int num_requests = get_requests.size(); for (int i = 0; i < num_requests; ++i) { GetRequest *get_req = get_requests[index]; - object_table_entry *entry; - HASH_FIND(handle, store_state->plasma_store_info->objects, &obj_id, - sizeof(obj_id), entry); + object_table_entry *entry = + get_object_table_entry(store_state->plasma_store_info, obj_id); CHECK(entry != NULL); PlasmaObject_init(&get_req->objects[obj_id], entry); @@ -427,9 +418,8 @@ void process_get_request(Client *client_context, /* Check if this object is already present locally. If so, record that the * object is being used and mark it as accounted for. */ - object_table_entry *entry; - HASH_FIND(handle, plasma_state->plasma_store_info->objects, &obj_id, - sizeof(obj_id), entry); + object_table_entry *entry = + get_object_table_entry(plasma_state->plasma_store_info, obj_id); if (entry && entry->state == PLASMA_SEALED) { /* Update the get request to take into account the present object. */ PlasmaObject_init(&get_req->objects[obj_id], entry); @@ -491,9 +481,8 @@ int remove_client_from_object_clients(object_table_entry *entry, void release_object(Client *client_context, ObjectID object_id) { PlasmaStoreState *plasma_state = client_context->plasma_state; - object_table_entry *entry; - HASH_FIND(handle, plasma_state->plasma_store_info->objects, &object_id, - sizeof(object_id), entry); + object_table_entry *entry = + get_object_table_entry(plasma_state->plasma_store_info, object_id); CHECK(entry != NULL); /* Remove the client from the object's array of clients. */ CHECK(remove_client_from_object_clients(entry, client_context) == 1); @@ -502,9 +491,8 @@ void release_object(Client *client_context, ObjectID object_id) { /* Check if an object is present. */ int contains_object(Client *client_context, ObjectID object_id) { PlasmaStoreState *plasma_state = client_context->plasma_state; - object_table_entry *entry; - HASH_FIND(handle, plasma_state->plasma_store_info->objects, &object_id, - sizeof(object_id), entry); + object_table_entry *entry = + get_object_table_entry(plasma_state->plasma_store_info, object_id); return entry && (entry->state == PLASMA_SEALED) ? OBJECT_FOUND : OBJECT_NOT_FOUND; } @@ -515,9 +503,8 @@ void seal_object(Client *client_context, unsigned char digest[]) { LOG_DEBUG("sealing object"); // TODO(pcm): add ObjectID here PlasmaStoreState *plasma_state = client_context->plasma_state; - object_table_entry *entry; - HASH_FIND(handle, plasma_state->plasma_store_info->objects, &object_id, - sizeof(object_id), entry); + object_table_entry *entry = + get_object_table_entry(plasma_state->plasma_store_info, object_id); CHECK(entry != NULL); CHECK(entry->state == PLASMA_CREATED); /* Set the state of object to SEALED. */ @@ -535,9 +522,8 @@ void seal_object(Client *client_context, * be called on objects that are returned by the eviction policy to evict. */ void delete_object(PlasmaStoreState *plasma_state, ObjectID object_id) { LOG_DEBUG("deleting object"); - object_table_entry *entry; - HASH_FIND(handle, plasma_state->plasma_store_info->objects, &object_id, - sizeof(object_id), entry); + object_table_entry *entry = + get_object_table_entry(plasma_state->plasma_store_info, object_id); /* TODO(rkn): This should probably not fail, but should instead throw an * error. Maybe we should also support deleting objects that have been created * but not sealed. */ @@ -547,9 +533,9 @@ void delete_object(PlasmaStoreState *plasma_state, ObjectID object_id) { CHECKM(utarray_len(entry->clients) == 0, "To delete an object, there must be no clients currently using it."); uint8_t *pointer = entry->pointer; - HASH_DELETE(handle, plasma_state->plasma_store_info->objects, entry); dlfree(pointer); utarray_free(entry->clients); + plasma_state->plasma_store_info->objects.erase(object_id); delete entry; /* Inform all subscribers that the object has been deleted. */ ObjectInfoT notification; @@ -674,10 +660,8 @@ void subscribe_to_updates(Client *client_context, int conn) { HASH_ADD_INT(plasma_state->pending_notifications, subscriber_fd, queue); /* Push notifications to the new subscriber about existing objects. */ - object_table_entry *entry, *temp_entry; - HASH_ITER(handle, plasma_state->plasma_store_info->objects, entry, - temp_entry) { - push_notification(plasma_state, &entry->info); + for (const auto &entry : plasma_state->plasma_store_info->objects) { + push_notification(plasma_state, &entry.second->info); } send_notifications(plasma_state->loop, queue->subscriber_fd, plasma_state, 0); } @@ -779,10 +763,8 @@ void process_message(event_loop *loop, /* If this client was using any objects, remove it from the appropriate * lists. */ PlasmaStoreState *plasma_state = client_context->plasma_state; - object_table_entry *entry, *temp_entry; - HASH_ITER(handle, plasma_state->plasma_store_info->objects, entry, - temp_entry) { - remove_client_from_object_clients(entry, client_context); + for (const auto &entry : plasma_state->plasma_store_info->objects) { + remove_client_from_object_clients(entry.second, client_context); } /* Note, the store may still attempt to send a message to the disconnected * client (for example, when an object ID that the client was waiting for