From a0247ffe5502268045afd744104303b3107e3fcd Mon Sep 17 00:00:00 2001 From: "Siyuan (Ryans) Zhuang" Date: Sat, 6 Jun 2020 22:11:37 -0700 Subject: [PATCH] Build plasma store as a library (#8817) * build plasma store as a library * remove unused headers * windows support --- BUILD.bazel | 7 +- src/ray/plasma/store.cc | 219 ++++++++++++++--------------------- src/ray/plasma/store.h | 19 +++ src/ray/plasma/store_exec.cc | 42 +++++++ 4 files changed, 152 insertions(+), 135 deletions(-) create mode 100644 src/ray/plasma/store_exec.cc diff --git a/BUILD.bazel b/BUILD.bazel index 46d27bdd2..27af8796f 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -374,7 +374,7 @@ cc_library( ) cc_library( - name = "plasma_lib", + name = "plasma_store_server_lib", srcs = [ "src/ray/plasma/dlmalloc.cc", "src/ray/plasma/events.cc", @@ -382,6 +382,7 @@ cc_library( "src/ray/plasma/external_store.cc", "src/ray/plasma/plasma_allocator.cc", "src/ray/plasma/quota_aware_policy.cc", + "src/ray/plasma/store.cc", ], hdrs = [ "src/ray/plasma/events.h", @@ -406,12 +407,12 @@ cc_library( cc_binary( name = "plasma_store_server", srcs = [ - "src/ray/plasma/store.cc", + "src/ray/plasma/store_exec.cc", ], copts = PLASMA_COPTS, visibility = ["//visibility:public"], deps = [ - ":plasma_lib", + ":plasma_store_server_lib", ":platform_shims", ], ) diff --git a/src/ray/plasma/store.cc b/src/ray/plasma/store.cc index c11085b52..f0f090e19 100644 --- a/src/ray/plasma/store.cc +++ b/src/ray/plasma/store.cc @@ -30,17 +30,13 @@ #include #include -#include #include #include #include #include #include -#include -#include #include #include -#include #include #include @@ -1138,125 +1134,37 @@ Status PlasmaStore::ProcessMessage(Client* client) { return Status::OK(); } -class PlasmaStoreRunner { - public: - PlasmaStoreRunner() {} - - void Start(char* socket_name, std::string directory, bool hugepages_enabled, - std::shared_ptr external_store) { - // Create the event loop. - loop_.reset(new EventLoop); - store_.reset(new PlasmaStore(loop_.get(), directory, hugepages_enabled, socket_name, - external_store)); - plasma_config = store_->GetPlasmaStoreInfo(); - - // We are using a single memory-mapped file by mallocing and freeing a single - // large amount of space up front. According to the documentation, - // dlmalloc might need up to 128*sizeof(size_t) bytes for internal - // bookkeeping. - void* pointer = plasma::PlasmaAllocator::Memalign( - kBlockSize, PlasmaAllocator::GetFootprintLimit() - 256 * sizeof(size_t)); - ARROW_CHECK(pointer != nullptr); - // This will unmap the file, but the next one created will be as large - // as this one (this is an implementation detail of dlmalloc). - plasma::PlasmaAllocator::Free( - pointer, PlasmaAllocator::GetFootprintLimit() - 256 * sizeof(size_t)); - - int socket = ConnectOrListenIpcSock(socket_name, true); - // TODO(pcm): Check return value. - ARROW_CHECK(socket >= 0); - - loop_->AddFileEvent(socket, kEventLoopRead, [this, socket](int events) { - this->store_->ConnectClient(socket); - }); - loop_->Start(); - } - - void Stop() { loop_->Stop(); } - - void Shutdown() { - loop_->Shutdown(); - loop_ = nullptr; - store_ = nullptr; - } - - private: - std::unique_ptr loop_; - std::unique_ptr store_; -}; - -static std::unique_ptr g_runner = nullptr; - void HandleSignal(int signal) { if (signal == SIGTERM) { ARROW_LOG(INFO) << "SIGTERM Signal received, closing Plasma Server..."; - if (g_runner != nullptr) { - g_runner->Stop(); - } - } -} - -void StartServer(char* socket_name, std::string plasma_directory, bool hugepages_enabled, - std::shared_ptr external_store) { -#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 - // to a client that has already died, the store could die. - signal(SIGPIPE, SIG_IGN); +#ifdef _WIN32 + ExitThread(0); +#else + pthread_exit(0); #endif - - g_runner.reset(new PlasmaStoreRunner()); - signal(SIGTERM, HandleSignal); - g_runner->Start(socket_name, plasma_directory, hugepages_enabled, external_store); + } } -} // namespace plasma - -int main(int argc, char* argv[]) { - ArrowLog::StartArrowLog(argv[0], ArrowLogLevel::ARROW_INFO); +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) { + ArrowLog::StartArrowLog("plasma_store", ArrowLogLevel::ARROW_INFO); ArrowLog::InstallFailureSignalHandler(); - char* socket_name = nullptr; - // Directory where plasma memory mapped files are stored. - std::string plasma_directory; - std::string external_store_endpoint; - bool hugepages_enabled = false; - int64_t system_memory = -1; - int c; - while ((c = getopt(argc, argv, "s:m:d:e:h")) != -1) { - switch (c) { - case 'd': - plasma_directory = std::string(optarg); - break; - case 'e': - external_store_endpoint = std::string(optarg); - break; - case 'h': - hugepages_enabled = true; - break; - case 's': - socket_name = optarg; - break; - case 'm': { - char extra; - int scanned = sscanf(optarg, "%" SCNd64 "%c", &system_memory, &extra); - ARROW_CHECK(scanned == 1); - // Set system memory capacity - plasma::PlasmaAllocator::SetFootprintLimit(static_cast(system_memory)); - ARROW_LOG(INFO) << "Allowing the Plasma store to use up to " - << static_cast(system_memory) / 1000000000 - << "GB of memory."; - break; - } - default: - exit(-1); - } - } - // Sanity check command line options. - if (!socket_name) { + + // Sanity check. + if (socket_name.empty()) { ARROW_LOG(FATAL) << "please specify socket for incoming connections with -s switch"; } + socket_name_ = socket_name; if (system_memory == -1) { ARROW_LOG(FATAL) << "please specify the amount of system memory with -m switch"; } + // Set system memory capacity + plasma::PlasmaAllocator::SetFootprintLimit(static_cast(system_memory)); + ARROW_LOG(INFO) << "Allowing the Plasma store to use up to " + << static_cast(system_memory) / 1000000000 + << "GB of memory."; if (hugepages_enabled && plasma_directory.empty()) { ARROW_LOG(FATAL) << "if you want to use hugepages, please specify path to huge pages " "filesystem with -d"; @@ -1298,33 +1206,80 @@ int main(int argc, char* argv[]) { plasma::SetMallocGranularity(1024 * 1024 * 1024); // 1 GB } #endif - // Get external store - std::shared_ptr external_store{nullptr}; - if (!external_store_endpoint.empty()) { - std::string name; - ARROW_CHECK_OK( - plasma::ExternalStores::ExtractStoreName(external_store_endpoint, &name)); - external_store = plasma::ExternalStores::GetStore(name); - if (external_store == nullptr) { - ARROW_LOG(FATAL) << "No such external store \"" << name << "\""; - return -1; - } - ARROW_LOG(DEBUG) << "connecting to external store..."; - ARROW_CHECK_OK(external_store->Connect(external_store_endpoint)); - } - ARROW_LOG(DEBUG) << "starting server listening on " << socket_name; + system_memory_ = system_memory; + plasma_directory_ = plasma_directory; +} + +void PlasmaStoreRunner::Start() { #ifdef _WINSOCKAPI_ WSADATA wsadata; WSAStartup(MAKEWORD(2, 2), &wsadata); #endif - plasma::StartServer(socket_name, plasma_directory, hugepages_enabled, external_store); - plasma::g_runner->Shutdown(); - plasma::g_runner = nullptr; + // Get external store + std::shared_ptr external_store{nullptr}; + if (!external_store_endpoint_.empty()) { + std::string name; + ARROW_CHECK_OK( + plasma::ExternalStores::ExtractStoreName(external_store_endpoint_, &name)); + external_store = plasma::ExternalStores::GetStore(name); + if (external_store == nullptr) { + ARROW_LOG(FATAL) << "No such external store \"" << name << "\""; + } + ARROW_LOG(DEBUG) << "connecting to external store..."; + ARROW_CHECK_OK(external_store->Connect(external_store_endpoint_)); + } + ARROW_LOG(DEBUG) << "starting server listening on " << socket_name_; - ArrowLog::UninstallSignalAction(); - ArrowLog::ShutDownArrowLog(); + // Ignore SIGPIPE signals. If we don't do this, then when we attempt to write + // to a client that has already died, the store could die. +#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 + // to a client that has already died, the store could die. + signal(SIGPIPE, SIG_IGN); +#endif + signal(SIGTERM, HandleSignal); + + // Create the event loop. + loop_.reset(new EventLoop); + store_.reset(new PlasmaStore(loop_.get(), plasma_directory_, hugepages_enabled_, + socket_name_, external_store)); + plasma_config = store_->GetPlasmaStoreInfo(); + + // We are using a single memory-mapped file by mallocing and freeing a single + // large amount of space up front. According to the documentation, + // dlmalloc might need up to 128*sizeof(size_t) bytes for internal + // bookkeeping. + void* pointer = plasma::PlasmaAllocator::Memalign( + kBlockSize, PlasmaAllocator::GetFootprintLimit() - 256 * sizeof(size_t)); + ARROW_CHECK(pointer != nullptr); + // This will unmap the file, but the next one created will be as large + // as this one (this is an implementation detail of dlmalloc). + plasma::PlasmaAllocator::Free( + pointer, PlasmaAllocator::GetFootprintLimit() - 256 * sizeof(size_t)); + + int socket = ConnectOrListenIpcSock(socket_name_, true); + // TODO(pcm): Check return value. + ARROW_CHECK(socket >= 0); + + loop_->AddFileEvent(socket, kEventLoopRead, [this, socket](int events) { + this->store_->ConnectClient(socket); + }); + loop_->Start(); + + Shutdown(); #ifdef _WINSOCKAPI_ WSACleanup(); #endif - return 0; + ArrowLog::UninstallSignalAction(); + ArrowLog::ShutDownArrowLog(); } + +void PlasmaStoreRunner::Stop() { loop_->Stop(); } + +void PlasmaStoreRunner::Shutdown() { + loop_->Shutdown(); + loop_ = nullptr; + store_ = nullptr; +} + +} // namespace plasma diff --git a/src/ray/plasma/store.h b/src/ray/plasma/store.h index 7c3b5fbe3..9ce620c14 100644 --- a/src/ray/plasma/store.h +++ b/src/ray/plasma/store.h @@ -245,6 +245,25 @@ class PlasmaStore { #endif }; +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); + void Start(); + void Stop(); + void Shutdown(); + + private: + std::string socket_name_; + int64_t system_memory_; + bool hugepages_enabled_; + std::string plasma_directory_; + std::string external_store_endpoint_; + std::unique_ptr loop_; + std::unique_ptr store_; +}; + } // namespace plasma #endif // PLASMA_STORE_H diff --git a/src/ray/plasma/store_exec.cc b/src/ray/plasma/store_exec.cc new file mode 100644 index 000000000..d4a876186 --- /dev/null +++ b/src/ray/plasma/store_exec.cc @@ -0,0 +1,42 @@ +#include "plasma/store.h" +#include +#include + +int main(int argc, char* argv[]) { + std::string socket_name; + // Directory where plasma memory mapped files are stored. + std::string plasma_directory; + std::string external_store_endpoint; + bool hugepages_enabled = false; + int64_t system_memory = -1; + int c; + while ((c = getopt(argc, argv, "s:m:d:e:h")) != -1) { + switch (c) { + case 'd': + plasma_directory = std::string(optarg); + break; + case 'e': + external_store_endpoint = std::string(optarg); + break; + case 'h': + hugepages_enabled = true; + break; + case 's': + socket_name = std::string(optarg); + break; + case 'm': { + char extra; + int scanned = sscanf(optarg, "%" SCNd64 "%c", &system_memory, &extra); + ARROW_CHECK(scanned == 1); + break; + } + default: + exit(-1); + } + } + + plasma::PlasmaStoreRunner runner( + socket_name, system_memory, hugepages_enabled, plasma_directory, external_store_endpoint); + runner.Start(); + return 0; +}