From cf9952a0281b1a249b773ec9dbf9e26121ed8b14 Mon Sep 17 00:00:00 2001 From: "Siyuan (Ryans) Zhuang" Date: Thu, 24 Dec 2020 17:30:06 -0800 Subject: [PATCH] [Core] Remote outdated external store (#13080) * remove outdated external store --- BUILD.bazel | 3 - src/ray/object_manager/object_manager.cc | 2 +- src/ray/object_manager/plasma/client.h | 5 +- .../object_manager/plasma/external_store.cc | 63 --------- .../object_manager/plasma/external_store.h | 120 ------------------ src/ray/object_manager/plasma/store.cc | 79 ++---------- src/ray/object_manager/plasma/store.h | 8 +- src/ray/object_manager/plasma/store_runner.cc | 22 +--- src/ray/object_manager/plasma/store_runner.h | 4 +- src/ray/plasma/store_exec.cc | 9 +- 10 files changed, 24 insertions(+), 291 deletions(-) delete mode 100644 src/ray/object_manager/plasma/external_store.cc delete mode 100644 src/ray/object_manager/plasma/external_store.h diff --git a/BUILD.bazel b/BUILD.bazel index 669d11a67..0d4007a6f 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -248,7 +248,6 @@ cc_library( "src/ray/object_manager/plasma/client.h", "src/ray/object_manager/plasma/common.h", "src/ray/object_manager/plasma/compat.h", - "src/ray/object_manager/plasma/external_store.h", "src/ray/object_manager/plasma/connection.h", "src/ray/object_manager/plasma/malloc.h", "src/ray/object_manager/plasma/plasma.h", @@ -287,7 +286,6 @@ cc_library( "src/ray/object_manager/plasma/create_request_queue.cc", "src/ray/object_manager/plasma/dlmalloc.cc", "src/ray/object_manager/plasma/eviction_policy.cc", - "src/ray/object_manager/plasma/external_store.cc", "src/ray/object_manager/plasma/plasma_allocator.cc", "src/ray/object_manager/plasma/quota_aware_policy.cc", "src/ray/object_manager/plasma/store.cc", @@ -297,7 +295,6 @@ cc_library( "src/ray/object_manager/common.h", "src/ray/object_manager/plasma/create_request_queue.h", "src/ray/object_manager/plasma/eviction_policy.h", - "src/ray/object_manager/plasma/external_store.h", "src/ray/object_manager/plasma/plasma_allocator.h", "src/ray/object_manager/plasma/quota_aware_policy.h", "src/ray/object_manager/plasma/store.h", diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc index 8ae4d723e..f3620d4c9 100644 --- a/src/ray/object_manager/object_manager.cc +++ b/src/ray/object_manager/object_manager.cc @@ -32,7 +32,7 @@ ObjectStoreRunner::ObjectStoreRunner(const ObjectManagerConfig &config, if (config.object_store_memory > 0) { plasma::plasma_store_runner.reset(new plasma::PlasmaStoreRunner( config.store_socket_name, config.object_store_memory, config.huge_pages, - config.plasma_directory, "")); + config.plasma_directory)); // Initialize object store. store_thread_ = std::thread(&plasma::PlasmaStoreRunner::Start, plasma::plasma_store_runner.get(), diff --git a/src/ray/object_manager/plasma/client.h b/src/ray/object_manager/plasma/client.h index 4c1d2745a..8c5e607b8 100644 --- a/src/ray/object_manager/plasma/client.h +++ b/src/ray/object_manager/plasma/client.h @@ -45,8 +45,7 @@ struct ObjectBuffer { int device_num; }; -// TODO(suquark): Maybe we should not export plasma later? -class RAY_EXPORT PlasmaClient { +class PlasmaClient { public: PlasmaClient(); ~PlasmaClient(); @@ -279,7 +278,7 @@ class RAY_EXPORT PlasmaClient { bool IsInUse(const ObjectID &object_id); - class RAY_NO_EXPORT Impl; + class Impl; std::shared_ptr impl_; }; diff --git a/src/ray/object_manager/plasma/external_store.cc b/src/ray/object_manager/plasma/external_store.cc deleted file mode 100644 index 87df6cda3..000000000 --- a/src/ray/object_manager/plasma/external_store.cc +++ /dev/null @@ -1,63 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -#include -#include - -#include "arrow/util/memory.h" - -#include "ray/object_manager/plasma/external_store.h" - -namespace plasma { - -Status ExternalStores::ExtractStoreName(const std::string &endpoint, - std::string *store_name) { - size_t off = endpoint.find_first_of(':'); - if (off == std::string::npos) { - return Status::Invalid("Malformed endpoint " + endpoint); - } - *store_name = endpoint.substr(0, off); - return Status::OK(); -} - -void ExternalStores::RegisterStore(const std::string &store_name, - std::shared_ptr store) { - Stores().insert({store_name, store}); -} - -void ExternalStores::DeregisterStore(const std::string &store_name) { - auto it = Stores().find(store_name); - if (it == Stores().end()) { - return; - } - Stores().erase(it); -} - -std::shared_ptr ExternalStores::GetStore(const std::string &store_name) { - auto it = Stores().find(store_name); - if (it == Stores().end()) { - return nullptr; - } - return it->second; -} - -ExternalStores::StoreMap &ExternalStores::Stores() { - static auto *external_stores = new StoreMap(); - return *external_stores; -} - -} // namespace plasma diff --git a/src/ray/object_manager/plasma/external_store.h b/src/ray/object_manager/plasma/external_store.h deleted file mode 100644 index 2c1934a73..000000000 --- a/src/ray/object_manager/plasma/external_store.h +++ /dev/null @@ -1,120 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -#pragma once - -#include -#include -#include -#include - -#include "ray/object_manager/plasma/client.h" - -namespace plasma { - -// ==== The external store ==== -// -// This file contains declaration for all functions that need to be implemented -// for an external storage service so that objects evicted from Plasma store -// can be written to it. - -class ExternalStore { - public: - /// Default constructor. - ExternalStore() = default; - - /// Virtual destructor. - virtual ~ExternalStore() = default; - - /// Connect to the local plasma store. Return the resulting connection. - /// - /// \param endpoint The name of the endpoint to connect to the external - /// storage service. While the formatting of the endpoint name is - /// specific to the implementation of the external store, it always - /// starts with {store-name}://, where {store-name} is the name of the - /// external store. - /// - /// \return The return status. - virtual Status Connect(const std::string &endpoint) = 0; - - /// This method will be called whenever an object in the Plasma store needs - /// to be evicted to the external store. - /// - /// This API is experimental and might change in the future. - /// - /// \param ids The IDs of the objects to put. - /// \param data The object data to put. - /// \return The return status. - virtual Status Put(const std::vector &ids, - const std::vector> &data) = 0; - - /// This method will be called whenever an evicted object in the external - /// store store needs to be accessed. - /// - /// This API is experimental and might change in the future. - /// - /// \param ids The IDs of the objects to get. - /// \param buffers List of buffers the data should be written to. - /// \return The return status. - virtual Status Get(const std::vector &ids, - std::vector> buffers) = 0; -}; - -class ExternalStores { - public: - typedef std::unordered_map> StoreMap; - /// Extracts the external store name from the external store endpoint. - /// - /// \param endpoint The endpoint for the external store. - /// \param[out] store_name The name of the external store. - /// \return The return status. - static Status ExtractStoreName(const std::string &endpoint, std::string *store_name); - - /// Register a new external store. - /// - /// \param store_name Name of the new external store. - /// \param store The new external store object. - static void RegisterStore(const std::string &store_name, - std::shared_ptr store); - - /// Remove an external store from the registry. - /// - /// \param store_name Name of the external store to remove. - static void DeregisterStore(const std::string &store_name); - - /// Obtain the external store given its name. - /// - /// \param store_name Name of the external store. - /// \return The external store object. - static std::shared_ptr GetStore(const std::string &store_name); - - private: - /// Obtain mapping between external store names and store instances. - /// - /// \return Mapping between external store names and store instances. - static StoreMap &Stores(); -}; - -#define REGISTER_EXTERNAL_STORE(name, store) \ - class store##Class { \ - public: \ - store##Class() { ExternalStores::RegisterStore(name, std::make_shared()); } \ - ~store##Class() { ExternalStores::DeregisterStore(name); } \ - }; \ - store##Class singleton_##store = store##Class() - -} // namespace plasma diff --git a/src/ray/object_manager/plasma/store.cc b/src/ray/object_manager/plasma/store.cc index ded4f3628..7fe41faa3 100644 --- a/src/ray/object_manager/plasma/store.cc +++ b/src/ray/object_manager/plasma/store.cc @@ -112,7 +112,6 @@ GetRequest::GetRequest(boost::asio::io_service &io_context, PlasmaStore::PlasmaStore(boost::asio::io_service &main_service, std::string directory, bool hugepages_enabled, const std::string &socket_name, - std::shared_ptr external_store, uint32_t delay_on_oom_ms, ray::SpillObjectsCallback spill_objects_callback, std::function object_store_full_callback) @@ -121,7 +120,6 @@ PlasmaStore::PlasmaStore(boost::asio::io_service &main_service, std::string dire acceptor_(main_service, ParseUrlEndpoint(socket_name)), socket_(main_service), eviction_policy_(&store_info_, PlasmaAllocator::GetFootprintLimit()), - external_store_(external_store), spill_objects_callback_(spill_objects_callback), delay_on_oom_ms_(delay_on_oom_ms), usage_log_interval_ns_(RayConfig::instance().object_store_usage_log_interval_s() * @@ -467,8 +465,6 @@ void PlasmaStore::ProcessGetRequest(const std::shared_ptr &client, int64_t timeout_ms) { // Create a get request for this object. auto get_req = new GetRequest(io_context_, client, object_ids); - std::vector evicted_ids; - std::vector evicted_entries; for (auto object_id : object_ids) { // Check if this object is already present // locally. If so, record that the object is being used and mark it as accounted for. @@ -489,12 +485,12 @@ void PlasmaStore::ProcessGetRequest(const std::shared_ptr &client, /*evict=*/true, &entry->fd, &entry->map_size, &entry->offset, client, false, &error); if (entry->pointer) { + // TODO(suquark): Not sure if this old behavior is still compatible + // with our current object spilling mechanics. entry->state = ObjectState::PLASMA_CREATED; entry->create_time = std::time(nullptr); eviction_policy_.ObjectCreated(object_id, client.get(), false); AddToClientObjectIds(object_id, store_info_.objects[object_id].get(), client); - evicted_ids.push_back(object_id); - evicted_entries.push_back(entry); } else { // We are out of memory and cannot allocate memory for this object. // Change the state of the object back to PLASMA_EVICTED so some @@ -511,31 +507,6 @@ void PlasmaStore::ProcessGetRequest(const std::shared_ptr &client, } } - if (!evicted_ids.empty()) { - std::vector> buffers; - for (size_t i = 0; i < evicted_ids.size(); ++i) { - RAY_CHECK(evicted_entries[i]->pointer != nullptr); - buffers.emplace_back(new arrow::MutableBuffer(evicted_entries[i]->pointer, - evicted_entries[i]->data_size)); - } - if (external_store_->Get(evicted_ids, buffers).ok()) { - for (size_t i = 0; i < evicted_ids.size(); ++i) { - evicted_entries[i]->state = ObjectState::PLASMA_SEALED; - evicted_entries[i]->construct_duration = - std::time(nullptr) - evicted_entries[i]->create_time; - PlasmaObject_init(&get_req->objects[evicted_ids[i]], evicted_entries[i]); - get_req->num_satisfied += 1; - } - } else { - // We tried to get the objects from the external store, but could not get them. - // Set the state of these objects back to PLASMA_EVICTED so some other request - // can try again. - for (size_t i = 0; i < evicted_ids.size(); ++i) { - evicted_entries[i]->state = ObjectState::PLASMA_EVICTED; - } - } - } - // If all of the objects are present already or if the timeout is 0, return to // the client. if (get_req->num_satisfied == get_req->num_objects_to_wait_for || timeout_ms == 0) { @@ -703,9 +674,6 @@ void PlasmaStore::EvictObjects(const std::vector &object_ids) { if (object_ids.size() == 0) { return; } - - std::vector> evicted_object_data; - std::vector evicted_entries; for (const auto &object_id : object_ids) { RAY_LOG(DEBUG) << "evicting object " << object_id.Hex(); auto entry = GetObjectTableEntry(&store_info_, object_id); @@ -718,37 +686,18 @@ void PlasmaStore::EvictObjects(const std::vector &object_ids) { RAY_CHECK(entry->ref_count == 0) << "To evict an object, there must be no clients currently using it."; - // If there is a backing external store, then mark object for eviction to - // external store, free the object data pointer and keep a placeholder - // entry in ObjectTable - if (external_store_) { - evicted_object_data.push_back(std::make_shared( - entry->pointer, entry->data_size + entry->metadata_size)); - evicted_entries.push_back(entry); - } else { - // Prepare the notification before deleting the object. - ObjectInfoT notification; - notification.object_id = object_id.Binary(); - notification.owner_raylet_id = entry->owner_raylet_id.Binary(); - notification.owner_ip_address = entry->owner_ip_address; - notification.owner_port = entry->owner_port; - notification.owner_worker_id = entry->owner_worker_id.Binary(); - notification.is_deletion = true; - // If there is no backing external store, just erase the object entry - // and send a deletion notification. - EraseFromObjectTable(object_id); - // Inform all subscribers that the object has been deleted. - PushNotification(¬ification); - } - } - - if (external_store_ && !object_ids.empty()) { - RAY_CHECK_OK(external_store_->Put(object_ids, evicted_object_data)); - for (auto entry : evicted_entries) { - PlasmaAllocator::Free(entry->pointer, entry->data_size + entry->metadata_size); - entry->pointer = nullptr; - entry->state = ObjectState::PLASMA_EVICTED; - } + // Prepare the notification before deleting the object. + ObjectInfoT notification; + notification.object_id = object_id.Binary(); + notification.owner_raylet_id = entry->owner_raylet_id.Binary(); + notification.owner_ip_address = entry->owner_ip_address; + notification.owner_port = entry->owner_port; + notification.owner_worker_id = entry->owner_worker_id.Binary(); + notification.is_deletion = true; + // Erase the object entry and send a deletion notification. + EraseFromObjectTable(object_id); + // Inform all subscribers that the object has been deleted. + PushNotification(¬ification); } } diff --git a/src/ray/object_manager/plasma/store.h b/src/ray/object_manager/plasma/store.h index 60efeb893..ec338d388 100644 --- a/src/ray/object_manager/plasma/store.h +++ b/src/ray/object_manager/plasma/store.h @@ -32,7 +32,6 @@ #include "ray/object_manager/plasma/common.h" #include "ray/object_manager/plasma/connection.h" #include "ray/object_manager/plasma/create_request_queue.h" -#include "ray/object_manager/plasma/external_store.h" #include "ray/object_manager/plasma/plasma.h" #include "ray/object_manager/plasma/protocol.h" #include "ray/object_manager/plasma/quota_aware_policy.h" @@ -55,8 +54,7 @@ class PlasmaStore { // TODO: PascalCase PlasmaStore methods. PlasmaStore(boost::asio::io_service &main_service, std::string directory, bool hugepages_enabled, const std::string &socket_name, - std::shared_ptr external_store, uint32_t delay_on_oom_ms, - ray::SpillObjectsCallback spill_objects_callback, + uint32_t delay_on_oom_ms, ray::SpillObjectsCallback spill_objects_callback, std::function object_store_full_callback); ~PlasmaStore(); @@ -275,10 +273,6 @@ class PlasmaStore { std::unordered_set deletion_cache_; - /// Manages worker threads for handling asynchronous/multi-threaded requests - /// for reading/writing data to/from external store. - std::shared_ptr external_store_; - std::shared_ptr notification_listener_; /// A callback to asynchronously spill objects when space is needed. The /// callback returns the amount of space still needed after the spilling is diff --git a/src/ray/object_manager/plasma/store_runner.cc b/src/ray/object_manager/plasma/store_runner.cc index 1fc6a0662..723213bac 100644 --- a/src/ray/object_manager/plasma/store_runner.cc +++ b/src/ray/object_manager/plasma/store_runner.cc @@ -14,10 +14,8 @@ namespace plasma { void SetMallocGranularity(int value); PlasmaStoreRunner::PlasmaStoreRunner(std::string socket_name, int64_t system_memory, - bool hugepages_enabled, std::string plasma_directory, - const std::string external_store_endpoint) - : hugepages_enabled_(hugepages_enabled), - external_store_endpoint_(external_store_endpoint) { + bool hugepages_enabled, std::string plasma_directory) + : hugepages_enabled_(hugepages_enabled) { // Sanity check. if (socket_name.empty()) { RAY_LOG(FATAL) << "please specify socket for incoming connections with -s switch"; @@ -77,25 +75,11 @@ PlasmaStoreRunner::PlasmaStoreRunner(std::string socket_name, int64_t system_mem void PlasmaStoreRunner::Start(ray::SpillObjectsCallback spill_objects_callback, std::function object_store_full_callback) { - // Get external store - std::shared_ptr external_store{nullptr}; - if (!external_store_endpoint_.empty()) { - std::string name; - RAY_CHECK_OK( - plasma::ExternalStores::ExtractStoreName(external_store_endpoint_, &name)); - external_store = plasma::ExternalStores::GetStore(name); - if (external_store == nullptr) { - RAY_LOG(FATAL) << "No such external store \"" << name << "\""; - } - RAY_LOG(DEBUG) << "connecting to external store..."; - RAY_CHECK_OK(external_store->Connect(external_store_endpoint_)); - } RAY_LOG(DEBUG) << "starting server listening on " << socket_name_; - { absl::MutexLock lock(&store_runner_mutex_); store_.reset(new PlasmaStore(main_service_, plasma_directory_, hugepages_enabled_, - socket_name_, external_store, + socket_name_, RayConfig::instance().object_store_full_delay_ms(), spill_objects_callback, object_store_full_callback)); plasma_config = store_->GetPlasmaStoreInfo(); diff --git a/src/ray/object_manager/plasma/store_runner.h b/src/ray/object_manager/plasma/store_runner.h index 07317c25d..3edd70350 100644 --- a/src/ray/object_manager/plasma/store_runner.h +++ b/src/ray/object_manager/plasma/store_runner.h @@ -13,8 +13,7 @@ namespace plasma { class PlasmaStoreRunner { public: PlasmaStoreRunner(std::string socket_name, int64_t system_memory, - bool hugepages_enabled, std::string plasma_directory, - const std::string external_store_endpoint); + bool hugepages_enabled, std::string plasma_directory); void Start(ray::SpillObjectsCallback spill_objects_callback = nullptr, std::function object_store_full_callback = nullptr); void Stop(); @@ -31,7 +30,6 @@ class PlasmaStoreRunner { int64_t system_memory_; bool hugepages_enabled_; std::string plasma_directory_; - std::string external_store_endpoint_; boost::asio::io_service main_service_; std::unique_ptr store_; std::shared_ptr listener_; diff --git a/src/ray/plasma/store_exec.cc b/src/ray/plasma/store_exec.cc index f1c140db6..fe6d3346a 100644 --- a/src/ray/plasma/store_exec.cc +++ b/src/ray/plasma/store_exec.cc @@ -15,9 +15,6 @@ // Command-line flags. DEFINE_string(d, SHM_DEFAULT_PATH, "directory where to create the memory-backed file"); -DEFINE_string(e, "", - "endpoint for external storage service, where objects " - "evicted from Plasma store can be written to, optional"); DEFINE_bool(h, false, "whether to enable hugepage support"); DEFINE_string(s, "", "socket name where the Plasma store will listen for requests, required"); @@ -50,7 +47,6 @@ int main(int argc, char *argv[]) { ray::RayLog::InstallFailureSignalHandler(); // Directory where plasma memory mapped files are stored. std::string plasma_directory = FLAGS_d; - std::string external_store_endpoint = FLAGS_e; bool hugepages_enabled = FLAGS_h; std::string socket_name = FLAGS_s; int64_t system_memory = FLAGS_m; @@ -70,9 +66,8 @@ int main(int argc, char *argv[]) { } RAY_CHECK(!plasma_directory.empty()); - plasma::plasma_store_runner.reset( - new plasma::PlasmaStoreRunner(socket_name, system_memory, hugepages_enabled, - plasma_directory, external_store_endpoint)); + plasma::plasma_store_runner.reset(new plasma::PlasmaStoreRunner( + socket_name, system_memory, hugepages_enabled, plasma_directory)); // Install signal handler before starting the eventloop. #ifndef _WIN32 // TODO(mehrdadn): Is there an equivalent of this we need for Windows? // Ignore SIGPIPE signals. If we don't do this, then when we attempt to write