Fix CI replication, inline reads, and metadata semantics (fe32e11e) · Commits · hpc / gekkofs

src/client/gkfs_metadata.cpp

+9 −0

Changes for src/client/gkfs_metadata.cpp: 9 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -77,6 +77,7 @@ extern "C" {
		#include <sys/statvfs.h>
		#include <linux/stat.h>
		#include <sys/mman.h>
		#include <sys/syscall.h>
		}

		using namespace std;
		@@ -208,12 +209,20 @@ namespace {

		constexpr unsigned int max_symlink_depth = 40;

		mode_t
		creation_mode(mode_t mode) {
		const auto current_umask = static_cast<mode_t>(::syscall(SYS_umask, 0));
		::syscall(SYS_umask, current_umask);
		return mode & ~(current_umask & 0777);
		}

		int
		gkfs_open_impl(const std::string& path, mode_t mode, int flags,
		unsigned int symlink_depth) {

		LOG(DEBUG, "{}() called with path: \"{}\", mode: {}, flags: {}", __func__,
		path, mode, flags);
		mode = creation_mode(mode);
		// metadata object filled during create or stat
		gkfs::metadata::Metadata md{};
		if(flags & O_CREAT) {

src/client/preload_util.cpp

+2 −1

Changes for src/client/preload_util.cpp: 2 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -298,7 +298,8 @@ get_metadata(const string& path, bool follow_links, bool include_inline) {
		std::string inline_data;
		int err{};
		// Use file metadata from dentry cache if available
		if(CTX->use_dentry_cache() && CTX->dentry_cache()) {
		if(CTX->use_dentry_cache() && CTX->dentry_cache() &&
		!CTX->fs_conf()->mtime_state && !CTX->fs_conf()->ctime_state) {
		// get parent and filename path to retrieve the cache entry
		std::filesystem::path p(path);
		auto parent = p.parent_path().string();

src/client/rpc/forward_data.cpp

+22 −1

Changes for src/client/rpc/forward_data.cpp: 22 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -53,6 +53,7 @@
		#include <vector>
		#include <map>
		#include <iostream>
		#include <chrono>

		using namespace std;

		@@ -183,6 +184,9 @@ forward_write(const string& path, const void* buf, const off64_t offset,
		std::set<uint64_t> chnk_end_target{};

		std::unordered_map<uint64_t, std::vector<uint8_t>> write_ops_vect;
		const auto unavailable_hosts = num_copies > 0
		? CTX->unavailable_hosts()
		: std::set<gkfs::rpc::host_t>{};

		// If num_copies is 0, we do the normal write operation. Otherwise
		// we process all the replicas.
		@@ -197,6 +201,12 @@ forward_write(const string& path, const void* buf, const off64_t offset,
		continue;
		}
		const auto target = replica.host;
		if(unavailable_hosts.count(target) != 0) {
		LOG(WARNING,
		"{}() skipping unavailable replica host {} for '{}'",
		__func__, target, path);
		continue;
		}

		write_ops_vect.try_emplace(target, ((chnk_total + 7) / 8));
		gkfs::rpc::set_bitset(write_ops_vect[target], chnk_id - chnk_start);
		@@ -279,8 +289,15 @@ forward_write(const string& path, const void* buf, const off64_t offset,
		in.bulk_handle = bulk_handle;

		try {
		if(num_copies > 0 && CTX->rpc_timeout().count() == 0) {
		waiters.push_back(
		write_rpc.on(CTX->hosts().at(target))
		.timed_async(std::chrono::milliseconds(100),
		in));
		} else {
		waiters.push_back(forward_async_with_timeout(
		write_rpc, CTX->hosts().at(target), in));
		}
		waiter_targets.push_back(target);
		} catch(const std::exception& ex) {
		LOG(ERROR, "Failed to send RPC to host {}: {}", target, ex.what());
		@@ -334,7 +351,11 @@ forward_write(const string& path, const void* buf, const off64_t offset,
		}
		}

		if(submission_failed) {
		if(submission_failed && num_copies == 0) {
		err = EIO;
		}

		if(num_copies > 0 && waiters.empty()) {
		err = EIO;
		}

src/client/rpc/forward_metadata.cpp

+47 −7

Changes for src/client/rpc/forward_metadata.cpp: 47 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		@@ -39,6 +39,7 @@
		#include <zstd.h>
		#include <cstdint>
		#include <cstring>
		#include <chrono>


		using namespace std;
		@@ -57,6 +58,13 @@ forward_create(const std::string& path, const mode_t mode, const int copy) {
		gkfs::rpc::locate_metadata_replica(*CTX->distributor(), path, copy);
		auto host_id = target.host;

		if(CTX->get_replicas() > 0 &&
		CTX->unavailable_hosts().count(host_id) != 0) {
		LOG(WARNING, "{}() skipping unavailable host {} for '{}'", __func__,
		host_id, path);
		return EHOSTUNREACH;
		}

		if(CTX->use_metadata_batch()) {
		CTX->add_metadata_batch_entry(host_id, path, mode);
		return 0;
		@@ -73,11 +81,23 @@ forward_create(const std::string& path, const mode_t mode, const int copy) {
		in.path = path;
		in.mode = mode;

		auto out = gkfs::rpc::forward_call<gkfs::rpc::rpc_err_out_t>(
		CTX->rpc_engine(), endp, gkfs::rpc::tag::create, in, __func__,
		path);

		try {
		gkfs::rpc::rpc_err_out_t out;
		if(CTX->get_replicas() > 0 && CTX->rpc_timeout().count() == 0) {
		auto create_rpc = CTX->rpc_engine()->define(gkfs::rpc::tag::create);
		out = create_rpc.on(endp).timed(std::chrono::milliseconds(100), in);
		} else {
		out = gkfs::rpc::forward_call<gkfs::rpc::rpc_err_out_t>(
		CTX->rpc_engine(), endp, gkfs::rpc::tag::create, in,
		__func__, path);
		}
		return out.err;
		} catch(const std::exception& ex) {
		LOG(ERROR, "{}() create RPC for '{}' copy {} failed: {}", __func__,
		path, copy, ex.what());
		CTX->mark_host_failure(host_id);
		return EBUSY;
		}
		}

		int
		@@ -815,11 +835,13 @@ forward_read_inline(const std::string& path, void* buf, off64_t offset,
		gkfs::rpc::rpc_read_inline_out_t out =
		forward_with_timeout(read_rpc, endp, in);

		if(out.err == 0 && !out.data.empty()) {
		if(out.err == 0) {
		CTX->mark_host_success(target.host);
		gkfs::rpc::record_replica_read(target.copy);
		const auto len = std::min(read_size, out.data.size());
		if(len > 0) {
		std::memcpy(buf, out.data.data(), len);
		}
		return {0, static_cast<ssize_t>(len)};
		}
		if(out.err == EAGAIN) {
		@@ -901,7 +923,17 @@ forward_update_metadentry_size(const string& path, const size_t size,

		const auto replica_targets = gkfs::rpc::locate_metadata_replicas(
		*CTX->distributor(), path, num_copies);
		const auto unavailable_hosts = num_copies > 0
		? CTX->unavailable_hosts()
		: std::set<gkfs::rpc::host_t>{};
		for(const auto& replica : replica_targets) {
		if(unavailable_hosts.count(replica.host) != 0) {
		LOG(WARNING,
		"{}() skipping unavailable host {} for '{}' metadata size",
		__func__, replica.host, path);
		copy_errors.emplace_back(replica.copy, EHOSTUNREACH);
		continue;
		}
		auto endp = CTX->hosts().at(replica.host);

		gkfs::rpc::rpc_update_metadentry_size_in_t in;
		@@ -912,8 +944,16 @@ forward_update_metadentry_size(const string& path, const size_t size,
		in.clear_inline = clear_inline_flag;

		try {
		waiters.push_back({replica.copy, forward_async_with_timeout(
		update_rpc, endp, in)});
		if(num_copies > 0 && CTX->rpc_timeout().count() == 0) {
		waiters.push_back(
		{replica.copy,
		update_rpc.on(endp).timed_async(
		std::chrono::milliseconds(100), in)});
		} else {
		waiters.push_back(
		{replica.copy,
		forward_async_with_timeout(update_rpc, endp, in)});
		}
		} catch(const std::exception& ex) {
		LOG(ERROR, "{}() posting rpc for path '{}' replica {} failed: {}",
		__func__, path, replica.copy, ex.what());

src/daemon/backend/metadata/rocksdb_backend.cpp

+13 −6

Changes for src/daemon/backend/metadata/rocksdb_backend.cpp: 13 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -66,7 +66,7 @@ namespace gkfs::metadata {
		// Fast-path parsing: extracts only the first few mandatory integer fields
		bool
		parse_metadata_fast(std::string_view val, mode_t& mode, size_t& size,
		time_t& ctime, blkcnt_t& blocks) {
		time_t& mtime, time_t& ctime, blkcnt_t& blocks) {
		auto next_field = [](std::string_view& s, auto& val) -> bool {
		size_t pos = s.find('\|');
		std::string_view field =
		@@ -104,9 +104,10 @@ parse_metadata_fast(std::string_view val, mode_t& mode, size_t& size,
		return false;
		}
		if constexpr(gkfs::config::metadata::use_mtime) {
		long long dummy;
		if(!next_field(s, dummy))
		long long parsed_mtime = 0;
		if(!next_field(s, parsed_mtime))
		return false;
		mtime = parsed_mtime;
		}
		if constexpr(gkfs::config::metadata::use_ctime) {
		long long parsed_ctime = 0;
		@@ -487,14 +488,16 @@ iterate_dirents(rocksdb::DB* db, const std::string& root_path,

		mode_t mode = 0;
		size_t size = 0;
		time_t mtime = 0;
		time_t ctime = 0;
		blkcnt_t blocks = 0;

		std::string_view val_view(it->value().data(), it->value().size());
		if(!parse_metadata_fast(val_view, mode, size, ctime, blocks)) {
		if(!parse_metadata_fast(val_view, mode, size, mtime, ctime, blocks)) {
		Metadata md(it->value().ToString());
		mode = md.mode();
		size = md.size();
		mtime = md.mtime();
		ctime = md.ctime();
		blocks = md.blocks();
		}
		@@ -546,14 +549,16 @@ iterate_dirents(rocksdb::DB* db, const std::string& root_path,

		mode_t mode = 0;
		size_t size = 0;
		time_t mtime = 0;
		time_t ctime = 0;
		blkcnt_t blocks = 0;

		std::string_view val_view(it->value().data(), it->value().size());
		if(!parse_metadata_fast(val_view, mode, size, ctime, blocks)) {
		if(!parse_metadata_fast(val_view, mode, size, mtime, ctime, blocks)) {
		Metadata md(it->value().ToString());
		mode = md.mode();
		size = md.size();
		mtime = md.mtime();
		ctime = md.ctime();
		blocks = md.blocks();
		}
		@@ -674,14 +679,16 @@ RocksDBBackend::get_all_dirents_extended_impl(const std::string& dir,

		mode_t mode = 0;
		size_t size = 0;
		time_t mtime = 0;
		time_t ctime = 0;
		blkcnt_t blocks = 0;

		std::string_view val_view(it->value().data(), it->value().size());
		if(!parse_metadata_fast(val_view, mode, size, ctime, blocks)) {
		if(!parse_metadata_fast(val_view, mode, size, mtime, ctime, blocks)) {
		Metadata md(it->value().ToString());
		mode = md.mode();
		size = md.size();
		mtime = md.mtime();
		ctime = md.ctime();
		blocks = md.blocks();
		}