perf(client,distributor): fast child process launch, opendir pagination & guided directory routing (f266d6be) · Commits · hpc / gekkofs

include/common/rpc/distributor.hpp

+13 −0

Changes for include/common/rpc/distributor.hpp: 13 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -110,6 +110,13 @@ public:
		/** @brief Return hosts responsible for directory metadata. */
		virtual std::vector<host_t>
		locate_directory_metadata() const = 0;

		/** @brief Return hosts responsible for directory metadata for a specific path. */
		virtual std::vector<host_t>
		locate_directory_metadata(const std::string& path) const {
		(void) path;
		return locate_directory_metadata();
		}
		};


		@@ -243,6 +250,9 @@ private:
		std::unordered_map<std::string, std::pair<IntervalSet, unsigned int>>
		map_interval;
		std::vector<std::pair<std::string, host_t>> prefix_list; // Should not be very long
		std::unordered_map<std::string, host_t> prefix_map_;
		unsigned int prefix_depth_mask_{0};
		bool fast_prefix_map_usable_{true};
		bool
		init_guided();

		@@ -274,6 +284,9 @@ public:

		std::vector<host_t>
		locate_directory_metadata() const override;

		std::vector<host_t>
		locate_directory_metadata(const std::string& path) const override;
		};

		} // namespace gkfs::rpc

src/client/gkfs_metadata.cpp

+11 −6

Changes for src/client/gkfs_metadata.cpp: 11 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -1314,12 +1314,14 @@ gkfs_opendir(const std::string& path) {
		// Collect and process results
		ret.second->add(".", gkfs::filemap::FileType::directory);
		ret.second->add("..", gkfs::filemap::FileType::directory);
		const auto targets = CTX->distributor()->locate_directory_metadata();
		const auto targets =
		CTX->distributor()->locate_directory_metadata(path);
		const auto target_count = targets.size();
		const bool use_proxy_dirents =
		gkfs::config::proxy::fwd_get_dirents_single && CTX->use_proxy();
		const size_t initial_buffer_size =
		std::max<size_t>(size_t{1}, CTX->dirents_buff_size());
		const size_t initial_buffer_size = std::clamp<size_t>(
		CTX->dirents_buff_size() / std::max<size_t>(size_t{1}, target_count),
		size_t{65536}, size_t{262144});

		std::vector<std::vector<char>> buffers(target_count);
		std::vector<thallium::bulk> exposed_buffers;
		@@ -1405,10 +1407,10 @@ gkfs_opendir(const std::string& path) {
		get_dirents_rpc.on(CTX->hosts()->at(targets[i]))
		.async(in));
		waiter_buffer_ids.push_back(i);
		waiter_servers.push_back(i);
		waiter_servers.push_back(targets[i]);
		} catch(const std::exception& ex) {
		LOG(ERROR, "{}() Failed to post dirents RPC to {}: {}",
		__func__, i, ex.what());
		__func__, targets[i], ex.what());
		ret.first = EBUSY;
		}
		}
		@@ -1451,7 +1453,10 @@ gkfs_opendir(const std::string& path) {
		out, buffers[buffer_id].data(),
		buffers[buffer_id].size());
		consume_entries(entries);
		if(!entries.empty()) {
		const bool may_be_truncated =
		gkfs::config::rpc::use_dirents_compression \|\|
		(out.dirents_size + 4096 > buffers[buffer_id].size());
		if(!entries.empty() && may_be_truncated) {
		const auto& last_key = std::get<0>(entries.back());
		auto remainder_res =
		use_proxy_dirents

src/client/intercept.cpp

+26 −0

Changes for src/client/intercept.cpp: 26 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -1404,6 +1404,32 @@ int
		internal_hook_guard_wrapper(long syscall_number, long arg0, long arg1,
		long arg2, long arg3, long arg4, long arg5,
		long* syscall_return_value) {
		if(gkfs::preload::is_initializing()) {
		const char* open_path = nullptr;
		if(syscall_number == SYS_openat) {
		open_path = reinterpret_cast<const char*>(arg1);
		}
		#ifdef SYS_open
		else if(syscall_number == SYS_open) {
		open_path = reinterpret_cast<const char*>(arg0);
		}
		#endif
		if(open_path != nullptr) {
		if(std::strcmp(open_path, "/proc/kallsyms") == 0) {
		const char* allow_kallsyms =
		std::getenv("LIBGKFS_ALLOW_KALLSYMS");
		if(allow_kallsyms == nullptr \|\| allow_kallsyms[0] == '\0' \|\|
		std::strcmp(allow_kallsyms, "0") == 0) {
		*syscall_return_value = -ENOENT;
		return gkfs::syscall::hooked;
		}
		}
		if( std::strstr(open_path, "libpsm3-fi.so") != nullptr ) {
		*syscall_return_value = -ENOENT;
		return gkfs::syscall::hooked;
		}
		}
		}
		if(!CTX->interception_enabled()) {
		if(syscall_number == SYS_chdir) {
		const char* path = reinterpret_cast<const char*>(arg0);

src/client/preload.cpp

+47 −9

Changes for src/client/preload.cpp: 47 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		@@ -661,6 +661,15 @@ init_preload() {

		LOG(DEBUG, "Current working directory: '{}'", CTX->cwd());
		LOG(DEBUG, "Number of replicas : '{}'", CTX->get_replicas());
		if(std::getenv("LIBGKFS_KEEP_FI_PROVIDER_PATH") == nullptr) {
		unsetenv("FI_PROVIDER_PATH");
		}
		const char* client_hg_init = std::getenv("LIBGKFS_CLIENT_HG_INIT_INFO");
		if(client_hg_init != nullptr && client_hg_init[0] != '\0') {
		setenv("HG_INIT_INFO", client_hg_init, 1);
		} else {
		setenv("HG_INIT_INFO", "request_post_init=64", 1);
		}
		gkfs::preload::init_environment();
		CTX->enable_interception();

		@@ -744,7 +753,10 @@ init_preload() {
		}
		return false;
		});
		if(CTX->get_replicas() > 0 \|\|
		!gkfs::env::get_var(gkfs::env::REPAIR_JOURNAL_PATH).empty()) {
		CTX->start_repair_worker();
		}
		std::atexit(quick_exit_handler);
		preload_initializing.store(false, std::memory_order_release);
		}
		@@ -810,6 +822,21 @@ destroy_preload() {
		rm->flush_msgpack();
		CTX->destroy_metrics();
		#endif
		if(CTX->interception_enabled()) {
		#ifndef BYPASS_SYSCALL
		gkfs::preload::stop_interception();
		#endif
		CTX->disable_interception();
		#ifndef BYPASS_SYSCALL
		LOG(DEBUG, "Syscall interception stopped");
		#endif
		}

		const char* clean_rpc_shutdown = std::getenv("LIBGKFS_CLEAN_RPC_SHUTDOWN");
		if(clean_rpc_shutdown != nullptr &&
		(std::strcmp(clean_rpc_shutdown, "1") == 0 \|\|
		std::strcmp(clean_rpc_shutdown, "ON") == 0 \|\|
		std::strcmp(clean_rpc_shutdown, "true") == 0)) {
		CTX->clear_hosts();
		LOG(DEBUG, "Peer information deleted");

		@@ -821,15 +848,26 @@ destroy_preload() {
		LOG(DEBUG, "Shutting down RPC subsystem");
		CTX->rpc_engine(nullptr);
		LOG(DEBUG, "RPC subsystem shut down");

		if(CTX->interception_enabled()) {
		#ifndef BYPASS_SYSCALL
		gkfs::preload::stop_interception();
		#endif
		CTX->disable_interception();
		#ifndef BYPASS_SYSCALL
		LOG(DEBUG, "Syscall interception stopped");
		#endif
		} else {
		// Intentionally leak RPC/IPC engines and endpoint handles at process
		// exit so ~PreloadContext() does not block for ~200 ms in
		// margo_finalize() futex waits right before exit_group().
		auto* leaked_hosts =
		new std::shared_ptr<const std::vector<thallium::endpoint>>(
		CTX->hosts());
		auto* leaked_rpc_engine =
		new std::shared_ptr<thallium::engine>(CTX->rpc_engine());
		auto* leaked_ipc_engine =
		new std::shared_ptr<thallium::engine>(CTX->ipc_engine());
		(void) leaked_hosts;
		(void) leaked_rpc_engine;
		(void) leaked_ipc_engine;
		CTX->clear_hosts();
		if(CTX->use_proxy()) {
		CTX->clear_proxy_host();
		CTX->ipc_engine(nullptr);
		}
		CTX->rpc_engine(nullptr);
		}

		LOG(INFO, "All subsystems shut down. Client shutdown complete.");

src/client/preload_context.cpp

+40 −2

Changes for src/client/preload_context.cpp: 40 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -146,6 +146,26 @@ PreloadContext::~PreloadContext() {
		// Safety net: fuse_client never calls destroy_preload(), so this is the
		// only place its reload watcher is guaranteed to get joined.
		stop_reload_watcher_thread();

		const char* clean_rpc_shutdown = std::getenv("LIBGKFS_CLEAN_RPC_SHUTDOWN");
		if(clean_rpc_shutdown == nullptr \|\|
		(std::strcmp(clean_rpc_shutdown, "1") != 0 &&
		std::strcmp(clean_rpc_shutdown, "ON") != 0 &&
		std::strcmp(clean_rpc_shutdown, "true") != 0)) {
		// PreloadContext is a function-local static whose destructor runs via
		// __cxa_atexit BEFORE .fini_array (destroy_preload). Retain a heap
		// reference to the RPC/IPC engines and endpoint snapshot so member
		// destruction does not invoke margo_finalize() (~200 ms futex waits).
		auto* leaked_routing =
		new std::shared_ptr<const RoutingSnapshot>(routing_);
		auto* leaked_rpc_engine =
		new std::shared_ptr<thallium::engine>(rpc_engine_);
		auto* leaked_ipc_engine =
		new std::shared_ptr<thallium::engine>(ipc_engine_);
		(void) leaked_routing;
		(void) leaked_rpc_engine;
		(void) leaked_ipc_engine;
		}
		}

		void
		@@ -226,6 +246,9 @@ PreloadContext::destroy_metrics() {
		bool
		PreloadContext::init_metrics() {
		#ifdef GKFS_ENABLE_CLIENT_METRICS
		if(!gkfs::env::var_is_set(gkfs::env::ENABLE_METRICS)) {
		return true;
		}
		auto flush_interval = std::stoi(gkfs::env::get_var(
		gkfs::env::METRICS_FLUSH_INTERVAL,
		std::to_string(gkfs::config::client_metrics::flush_interval)));
		@@ -1503,7 +1526,8 @@ PreloadContext::repair_worker() {
		{
		std::unique_lock<std::mutex> lock(repair_queue_mutex_);
		const auto now = std::chrono::steady_clock::now();
		if(repair_worker_stop_ && now >= repair_worker_deadline_) {
		if(repair_worker_stop_ &&
		(now >= repair_worker_deadline_ \|\| repair_queue_.size() == 0)) {
		break;
		}
		const auto wait_time =
		@@ -1516,12 +1540,16 @@ PreloadContext::repair_worker() {
		repair_queue_cv_.wait_for(lock, wait_time);

		if(repair_worker_stop_ &&
		std::chrono::steady_clock::now() >= repair_worker_deadline_) {
		(std::chrono::steady_clock::now() >= repair_worker_deadline_ \|\|
		repair_queue_.size() == 0)) {
		break;
		}

		executor = repair_executor_;
		if(!executor) {
		if(repair_worker_stop_) {
		break;
		}
		continue;
		}
		task = repair_queue_.pop_ready([](const auto& candidate) {
		@@ -1547,6 +1575,8 @@ PreloadContext::repair_worker() {
		in_flight_repairs_.push_back(*task);
		persist_repair_journal_locked();
		}
		} else if(repair_worker_stop_) {
		break;
		}
		}

		@@ -1589,6 +1619,14 @@ PreloadContext::repair_worker() {

		bool
		PreloadContext::enqueue_repair_task(gkfs::rpc::repair_task task) {
		if(repair_worker_threads_.empty() && !repair_worker_stop_) {
		load_repair_journal();
		repair_worker_threads_.reserve(repair_worker_count_);
		for(uint32_t i = 0; i < repair_worker_count_; ++i) {
		repair_worker_threads_.emplace_back(&PreloadContext::repair_worker,
		this);
		}
		}
		const auto before = repair_queue_.size();
		const auto accepted = repair_queue_.enqueue(std::move(task));
		if(accepted && repair_queue_.size() > before) {