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

Build plasma store as a library (#8817)

Browse Source 
* build plasma store as a library

* remove unused headers

* windows support
This commit is contained in:
Siyuan (Ryans) Zhuang
2020-06-06 22:11:37 -07:00
committed by GitHub
parent 5d124489a9
commit a0247ffe55
4 changed files with 152 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
BUILD.bazel
+4 -3
View File
@@ -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",
],
)
src/ray/plasma/store.cc
+87 -132
View File
@@ -30,17 +30,13 @@
#include <assert.h>
#include <fcntl.h>
#include <getopt.h>
#include <limits.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/statvfs.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>
#include <ctime>
@@ -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<ExternalStore> 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<EventLoop> loop_;
std::unique_ptr<PlasmaStore> store_;
};
static std::unique_ptr<PlasmaStoreRunner> 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<ExternalStore> 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<size_t>(system_memory));
ARROW_LOG(INFO) << "Allowing the Plasma store to use up to "
<< static_cast<double>(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<size_t>(system_memory));
ARROW_LOG(INFO) << "Allowing the Plasma store to use up to "
<< static_cast<double>(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<plasma::ExternalStore> 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<plasma::ExternalStore> 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
src/ray/plasma/store.h
+19
View File
@@ -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<EventLoop> loop_;
std::unique_ptr<PlasmaStore> store_;
};
} // namespace plasma
#endif // PLASMA_STORE_H
src/ray/plasma/store_exec.cc
+42
View File
@@ -0,0 +1,42 @@
#include "plasma/store.h"
#include <getopt.h>
#include <stdio.h>
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;
}