async writes (!306) · Merge requests · hpc / gekkofs

include/client/env.hpp

+1 −0

Changes for include/client/env.hpp: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -98,6 +98,7 @@ static constexpr auto ENABLE_FORK = ADD_PREFIX("ENABLE_FORK");
		static constexpr auto METADATA_BATCH = ADD_PREFIX("METADATA_BATCH");
		static constexpr auto METADATA_BATCH_THRESHOLD =
		ADD_PREFIX("METADATA_BATCH_THRESHOLD");
		static constexpr auto ASYNC_WRITE = ADD_PREFIX("ASYNC_WRITE");

		} // namespace gkfs::env

+45 −0

Changes for include/client/preload_context.hpp: 45 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -42,6 +42,10 @@

		#include <map>
		#include <unordered_map>
		#include <queue>
		#include <future>
		#include <thread>
		#include <condition_variable>
		#include <thallium.hpp>
		#include <memory>
		#include <vector>
		@@ -52,6 +56,17 @@

		#include <bitset>

		namespace gkfs::preload {
		struct WriteTask {
		std::string path;
		std::vector<char> buf;
		off64_t offset;
		size_t count;
		int8_t num_replicas;
		std::shared_ptr<std::promise<std::pair<int, long>>> promise;
		};
		} // namespace gkfs::preload

		/* Forward declarations */
		namespace gkfs {
		namespace filemap {
		@@ -163,6 +178,14 @@ private:
		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_;
		std::mutex async_write_mutex_;
		std::condition_variable async_write_cv_;
		std::thread async_write_thread_;
		bool async_write_stop_{false};


		public:
		static PreloadContext*
		@@ -404,6 +427,28 @@ public:
		void
		add_metadata_batch_entry(uint64_t host_id, const std::string& path,
		mode_t mode);

		bool
		use_async_write() const;

		void
		use_async_write(bool use_async_write);

		void
		start_async_write_thread();

		void
		stop_async_write_thread();

		void
		async_write_worker();

		void
		enqueue_async_write(const std::string& path, const void* buf,
		off64_t offset, size_t count, int8_t num_copies);

		void
		wait_async_writes();
		};

		} // namespace preload

+19 −0

Changes for src/client/gkfs_data.cpp: 19 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -277,6 +277,17 @@ gkfs_do_write(gkfs::filemap::OpenFile& file, const char* buf, size_t count,
		}
		}

		if(CTX->use_async_write()) {
		CTX->enqueue_async_write(*path, buf, offset, count, 0);
		if(num_replicas > 0) {
		CTX->enqueue_async_write(*path, buf, offset, count, num_replicas);
		}
		if(update_pos) {
		file.pos(offset + count);
		}
		return count;
		}

		pair<int, long> ret_write;
		if(gkfs::config::proxy::fwd_io && CTX->use_proxy() &&
		count > gkfs::config::proxy::fwd_io_count_threshold) {
		@@ -592,6 +603,9 @@ gkfs_do_read(const gkfs::filemap::OpenFile& file, char* buf, size_t count,
		ssize_t
		gkfs_read_ws(const gkfs::filemap::OpenFile& file, char* buf, size_t count,
		off64_t offset) {
		if(CTX->use_async_write()) {
		CTX->wait_async_writes();
		}
		#ifdef GKFS_ENABLE_CLIENT_METRICS
		auto start_t = std::chrono::high_resolution_clock::now();
		auto read = gkfs_do_read(file, buf, count, offset);
		@@ -724,6 +738,11 @@ gkfs_fsync(unsigned int fd) {
		errno = EBADF;
		return -1;
		}

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

		// flush write size cache to be server consistent
		if(CTX->use_write_size_cache()) {
		auto err = CTX->write_size_cache()->flush(file->path(), true).first;

+4 −0

Changes for src/client/gkfs_metadata.cpp: 4 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -1910,6 +1910,10 @@ gkfs_close(unsigned int fd) {
		if(file) {
		const auto path = file->path();

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

		if(file->get_flag(gkfs::filemap::OpenFile_flags::creation_pending)) {
		gkfs_create(path, file->mode());
		file->set_flag(gkfs::filemap::OpenFile_flags::creation_pending,

+17 −0

Changes for src/client/preload.cpp: 17 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -362,6 +362,17 @@ init_environment() {
		LOG(INFO, "Metadata batching disabled.");
		}

		auto use_async_write =
		gkfs::env::get_var(gkfs::env::ASYNC_WRITE, "OFF") == "ON";
		if(use_async_write) {
		CTX->use_async_write(true);
		CTX->start_async_write_thread();
		LOG(INFO, "Client-side async write cache enabled.");
		} else {
		CTX->use_async_write(false);
		LOG(INFO, "Client-side async write cache disabled.");
		}

		LOG(INFO, "Environment initialization successful.");
		}

		@@ -530,6 +541,12 @@ destroy_preload() {
		LOG(INFO, "Flushing final metadata batches...");
		CTX->flush_metadata_batches();
		}

		if(CTX->use_async_write()) {
		LOG(INFO, "Flushing final async writes...");
		CTX->wait_async_writes();
		CTX->stop_async_write_thread();
		}
		auto forwarding_map_file = gkfs::env::get_var(
		gkfs::env::FORWARDING_MAP_FILE, gkfs::config::forwarding_file_path);
		if(!forwarding_map_file.empty()) {