perf(client,rpc): add lazy per-host endpoint lookup, short-TTL stat cache, and inline small dirents (60f02545) · Commits · hpc / gekkofs

include/client/preload_context.hpp

+75 −2

Changes for include/client/preload_context.hpp: 75 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -125,12 +125,53 @@ enum class RelativizeStatus { internal, external, fd_unknown, fd_not_a_dir };
		* distributor built against a hosts list from before/after it) without
		* blocking on RPC I/O for the swap itself.
		*/
		class HostEndpointTable {
		private:
		std::vector<std::string> uris_;
		mutable std::vector<thallium::endpoint> endpoints_;
		mutable std::unique_ptr<std::once_flag[]> once_flags_;

		public:
		HostEndpointTable() = default;

		explicit HostEndpointTable(std::vector<thallium::endpoint> endpoints)
		: endpoints_(std::move(endpoints)) {}

		explicit HostEndpointTable(std::vector<std::string> uris)
		: uris_(std::move(uris)),
		endpoints_(uris_.size()),
		once_flags_(std::make_unique<std::once_flag[]>(uris_.size())) {}

		HostEndpointTable(HostEndpointTable&&) noexcept = default;
		HostEndpointTable& operator=(HostEndpointTable&&) noexcept = default;

		std::size_t
		size() const noexcept {
		return endpoints_.size();
		}

		bool
		empty() const noexcept {
		return endpoints_.empty();
		}

		const thallium::endpoint&
		at(std::size_t id) const;
		};

		struct RoutingSnapshot {
		std::vector<thallium::endpoint> hosts;
		HostEndpointTable hosts;
		std::shared_ptr<gkfs::rpc::Distributor> distributor;
		uint64_t local_host_id{0};
		};

		struct StatCacheEntry {
		std::string attr;
		std::string inline_data;
		bool has_inline{false};
		std::chrono::steady_clock::time_point timestamp;
		};

		/**
		* Singleton class of the client context with all relevant global data
		*/
		@@ -154,6 +195,9 @@ private:
		bool has_local_host_{false};
		bool cas_enabled_{false};
		size_t cas_min_size_{16384};
		uint64_t stat_cache_ttl_ms_{500};
		mutable std::mutex stat_cache_mutex_;
		mutable std::unordered_map<std::string, StatCacheEntry> stat_cache_;


		std::string cwd_;
		@@ -281,7 +325,7 @@ public:
		const std::string&
		cwd() const;

		std::shared_ptr<const std::vector<thallium::endpoint>>
		std::shared_ptr<const HostEndpointTable>
		hosts() const;

		/**
		@@ -293,10 +337,39 @@ public:
		* from a different generation.
		*/
		void
		commit_routing(HostEndpointTable hosts,
		std::shared_ptr<gkfs::rpc::Distributor> distributor,
		uint64_t local_host_id);

		void
		commit_routing(std::vector<thallium::endpoint> hosts,
		std::shared_ptr<gkfs::rpc::Distributor> distributor,
		uint64_t local_host_id);

		void
		stat_cache_ttl_ms(uint64_t ttl_ms) noexcept {
		stat_cache_ttl_ms_ = ttl_ms;
		}

		uint64_t
		stat_cache_ttl_ms() const noexcept {
		return stat_cache_ttl_ms_;
		}

		bool
		get_stat_cache(const std::string& path, bool need_inline,
		std::string& attr_out, std::string& inline_out) const;

		void
		put_stat_cache(const std::string& path, const std::string& attr,
		const std::string& inline_data, bool has_inline);

		void
		invalidate_stat_cache(const std::string& path);

		void
		clear_stat_cache();

		void
		clear_hosts();

include/client/preload_util.hpp

+3 −2

Changes for include/client/preload_util.hpp: 3 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -135,8 +135,9 @@ read_hosts_file();
		* distributor built from the same (hosts, local_host_id) pair, so the three
		* never drift out of sync.
		*/
		std::pair<std::vector<thallium::endpoint>, uint64_t>
		connect_to_hosts(const std::vector<std::pair<std::string, std::string>>& hosts);
		std::pair<gkfs::preload::HostEndpointTable, uint64_t>
		connect_to_hosts(const std::vector<std::pair<std::string, std::string>>& hosts,
		bool allow_lazy = true);

		void
		check_for_proxy();

include/client/rpc/utils.hpp

+7 −0

Changes for include/client/rpc/utils.hpp: 7 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -140,6 +140,13 @@ decompress_and_parse_entries(
		if(out.dirents_size == 0) {
		return {};
		}
		if constexpr(std::is_same_v<std::decay_t<OutputOrErr>,
		gkfs::rpc::rpc_get_dirents_out_t>) {
		if(!out.inline_data.empty()) {
		compressed_buffer = out.inline_data.data();
		buffer_size = out.inline_data.size();
		}
		}
		if(out.dirents_size > buffer_size) {
		throw std::runtime_error(
		"Server returned dirents payload larger than exposed client buffer: " +

include/common/env.hpp

+2 −0

Changes for include/common/env.hpp: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -85,6 +85,8 @@ static constexpr auto ENABLE_TRACE = ADD_PREFIX("ENABLE_TRACE");
		static constexpr auto TRACE_SAMPLE_RATE = ADD_PREFIX("TRACE_SAMPLE_RATE");
		static constexpr auto ENABLE_CAS = ADD_PREFIX("ENABLE_CAS");
		static constexpr auto CAS_MIN_SIZE = ADD_PREFIX("CAS_MIN_SIZE");
		static constexpr auto LAZY_CONNECT = ADD_PREFIX("LAZY_CONNECT");
		static constexpr auto STAT_CACHE_TTL_MS = ADD_PREFIX("STAT_CACHE_TTL_MS");

		} // namespace gkfs::env

include/common/rpc/rpc_types_thallium.hpp

+8 −2

Changes for include/common/rpc/rpc_types_thallium.hpp: 8 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -363,23 +363,29 @@ struct rpc_get_dirents_in_t {
		std::string path;
		std::string start_key;
		uint64_t trace_id{gkfs::trace::next()};
		bool has_bulk{false};
		tl::bulk bulk_handle;

		template <class Archive>
		void
		serialize(Archive& ar) {
		ar(path, start_key, trace_id, bulk_handle);
		has_bulk = !bulk_handle.is_null();
		ar(path, start_key, trace_id, has_bulk);
		if(has_bulk) {
		ar(bulk_handle);
		}
		}
		};

		struct rpc_get_dirents_out_t {
		int32_t err;
		size_t dirents_size;
		std::string inline_data;

		template <class Archive>
		void
		serialize(Archive& ar) {
		ar(err, dirents_size);
		ar(err, dirents_size, inline_data);
		}
		};