feat(client): implement general metadata local-bypass and clean up delayed coherence (b444aa9e) · Commits · hpc / gekkofs

include/client/preload_context.hpp

+0 −48

Changes for include/client/preload_context.hpp: 0 added lines, 48 removed lines.
Original line number	Diff line number	Diff line
		@@ -173,8 +173,6 @@ struct StatCacheEntry {
		std::chrono::steady_clock::time_point timestamp;
		};

		struct ShmStatCacheRegion;

		/**
		* Singleton class of the client context with all relevant global data
		*/
		@@ -202,10 +200,6 @@ private:
		mutable std::mutex stat_cache_mutex_;
		mutable std::unordered_map<std::string, StatCacheEntry> stat_cache_;

		void
		init_shm_stat_cache();


		std::string cwd_;
		std::vector<std::string> mountdir_components_;
		std::string mountdir_;
		@@ -252,12 +246,6 @@ private:
		std::shared_ptr<thallium::engine> rpc_engine_;
		std::shared_ptr<thallium::engine> ipc_engine_;

		bool use_metadata_batch_{false};
		size_t metadata_batch_threshold_{64};
		std::unordered_map<uint64_t, std::vector<std::pair<std::string, mode_t>>>
		metadata_batch_buffer_;
		mutable std::mutex metadata_batch_mutex_;

		bool use_async_write_{false};
		std::queue<WriteTask> async_write_queue_;
		std::vector<std::future<std::pair<int, long>>> async_write_futures_;
		@@ -291,10 +279,7 @@ private:

		std::string ofi_interface_;
		std::chrono::milliseconds rpc_timeout_{0};
		ShmStatCacheRegion* shm_stat_cache_{nullptr};
		bool shm_stat_cache_enabled_{false};
		bool negative_stat_cache_{false};
		std::unordered_map<std::string, uint64_t> metadata_batch_paths_;


		public:
		@@ -364,14 +349,6 @@ public:
		return stat_cache_ttl_ms_;
		}

		void
		shm_stat_cache_enabled(bool enabled) noexcept;

		bool
		shm_stat_cache_enabled() const noexcept {
		return shm_stat_cache_enabled_;
		}

		void
		negative_stat_cache(bool enabled) noexcept {
		negative_stat_cache_ = enabled;
		@@ -645,31 +622,6 @@ public:
		void
		rpc_timeout(std::chrono::milliseconds timeout);

		bool
		use_metadata_batch() const;

		void
		use_metadata_batch(bool use_metadata_batch);

		size_t
		metadata_batch_threshold() const;

		void
		metadata_batch_threshold(size_t threshold);

		void
		flush_metadata_batches();

		void
		flush_metadata_batch(uint64_t host_id);

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

		void
		add_metadata_batch_entry(uint64_t host_id, const std::string& path,
		mode_t mode);

		bool
		use_async_write() const;

include/client/rpc/forward_data.hpp

+20 −0

Changes for include/client/rpc/forward_data.hpp: 20 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -76,6 +76,26 @@ forward_truncate(const std::string& path, size_t current_size, size_t new_size,
		std::pair<int, ChunkStat>
		forward_get_chunk_stat();

		std::string
		get_local_chunk_dir(const std::string& rootdir, const std::string& file_path);

		std::string
		get_local_chunk_path(const std::string& rootdir, const std::string& file_path,
		uint64_t chunk_id);

		ssize_t
		write_local_chunk(const std::string& rootdir, const std::string& file_path,
		uint64_t chunk_id, const char* buf, size_t size,
		off64_t offset);

		ssize_t
		read_local_chunk(const std::string& rootdir, const std::string& file_path,
		uint64_t chunk_id, char* buf, size_t size, off64_t offset);

		void
		destroy_local_chunk_space(const std::string& rootdir,
		const std::string& file_path);

		} // namespace gkfs::rpc

		#endif // GEKKOFS_CLIENT_FORWARD_DATA_HPP

include/common/env.hpp

+0 −1

Changes for include/common/env.hpp: 0 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -87,7 +87,6 @@ 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");
		static constexpr auto SHM_STAT_CACHE = ADD_PREFIX("SHM_STAT_CACHE");
		static constexpr auto NEGATIVE_STAT_CACHE = ADD_PREFIX("NEGATIVE_STAT_CACHE");

		} // namespace gkfs::env

src/client/gkfs_data.cpp

+26 −6

Changes for src/client/gkfs_data.cpp: 26 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -93,12 +93,6 @@ ssize_t
		gkfs_do_write(gkfs::filemap::OpenFile& file, const char* buf, size_t count,
		off64_t offset, bool update_pos) {

		// Writes may immediately follow O_CREAT. Complete queued metadata creates
		// for this path before accessing the file on the daemon.
		if(CTX->use_metadata_batch()) {
		CTX->flush_metadata_batch_if_needed(file.path());
		}

		if(file.type() != gkfs::filemap::FileType::regular) {
		assert(file.type() == gkfs::filemap::FileType::directory);
		LOG(WARNING, "Cannot write to directory");
		@@ -140,6 +134,17 @@ gkfs_do_write(gkfs::filemap::OpenFile& file, const char* buf, size_t count,
		std::string(buf, count), result.failed_copies);
		}
		gkfs::rpc::record_replica_write(result);
		const bool can_bypass = (num_replicas == 0) &&
		CTX->local_bypass() &&
		CTX->has_local_host() &&
		CTX->fs_conf() &&
		!CTX->fs_conf()->rootdir.empty();
		if(can_bypass && targets.size() == 1 &&
		targets[0].host == CTX->local_host_id()) {
		gkfs::rpc::write_local_chunk(
		CTX->fs_conf()->rootdir, *path, 0,
		static_cast<const char*>(buf), count, offset);
		}
		gkfs::metadata::Metadata md{
		static_cast<mode_t>(file.mode() \| S_IFREG)};
		md.size(static_cast<size_t>(offset) + count);
		@@ -212,6 +217,21 @@ gkfs_do_write(gkfs::filemap::OpenFile& file, const char* buf, size_t count,
		file.path(), buf, offset, count, is_append, num_replicas);
		err = ret_inline.first;
		if(err == 0) {
		const bool can_bypass = (num_replicas == 0) &&
		CTX->local_bypass() &&
		CTX->has_local_host() &&
		CTX->fs_conf() &&
		!CTX->fs_conf()->rootdir.empty();
		if(can_bypass) {
		const auto target = gkfs::rpc::locate_metadata_replica(
		*CTX->distributor(), file.path(), 0);
		if(target.host == CTX->local_host_id()) {
		gkfs::rpc::write_local_chunk(
		CTX->fs_conf()->rootdir, file.path(), 0,
		static_cast<const char*>(buf), count,
		is_append ? ret_inline.second : offset);
		}
		}
		if(update_pos)
		file.pos(ret_inline.second + count);
		return count;

src/client/gkfs_metadata.cpp

+0 −14

Changes for src/client/gkfs_metadata.cpp: 0 added lines, 14 removed lines.
Original line number	Diff line number	Diff line
		@@ -560,10 +560,6 @@ gkfs_remove(const std::string& path, bool is_rename_stub) {
		uint32_t mode = 0;
		std::string target_path;

		if(CTX->use_metadata_batch()) {
		CTX->flush_metadata_batch_if_needed(path);
		}

		if(gkfs::config::proxy::fwd_remove && CTX->use_proxy()) {
		err = gkfs::rpc::forward_remove_proxy(path, false);
		} else {
		@@ -1311,9 +1307,6 @@ gkfs_dup2(const int oldfd, const int newfd) {
		*/
		int
		gkfs_opendir(const std::string& path) {
		if(CTX->use_metadata_batch()) {
		CTX->flush_metadata_batches();
		}
		auto md = gkfs::utils::get_metadata(path);
		if(!md) {
		return -1;
		@@ -1526,9 +1519,6 @@ gkfs_opendir(const std::string& path) {
		int
		gkfs_rmdir(const std::string& path) {
		int err;
		if(CTX->use_metadata_batch()) {
		CTX->flush_metadata_batches();
		}
		// check that directory is empty if a strict dir hierarchy is
		// enforced
		if(gkfs::config::metadata::create_check_parents) {
		@@ -2105,10 +2095,6 @@ gkfs_close(unsigned int fd) {
		if(file) {
		const auto path = file->path();

		if(CTX->use_metadata_batch()) {
		CTX->flush_metadata_batches();
		}

		if(CTX->use_async_write()) {
		CTX->wait_async_writes();
		}