Lock System for cargo (!233) · Merge requests · hpc / gekkofs

include/client/env.hpp

+4 −0

Changes for include/client/env.hpp: 4 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -69,6 +69,10 @@ static constexpr auto METRICS_IP_PORT = ADD_PREFIX("METRICS_IP_PORT");
		#endif

		static constexpr auto PROTECT_FD = ADD_PREFIX("PROTECT_FD");
		static constexpr auto PROTECT_FILES_GENERATOR =
		ADD_PREFIX("PROTECT_FILES_GENERATOR");
		static constexpr auto PROTECT_FILES_CONSUMER =
		ADD_PREFIX("PROTECT_FILES_CONSUMER");
		static constexpr auto NUM_REPL = ADD_PREFIX("NUM_REPL");
		static constexpr auto PROXY_PID_FILE = ADD_PREFIX("PROXY_PID_FILE");
		namespace cache {

include/client/preload_context.hpp

+16 −0

Changes for include/client/preload_context.hpp: 16 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -140,6 +140,9 @@ private:
		int replicas_;

		bool protect_fds_{false};
		bool protect_files_generator_{false};
		bool protect_files_consumer_{false};


		std::shared_ptr<gkfs::messagepack::ClientMetrics> write_metrics_;
		std::shared_ptr<gkfs::messagepack::ClientMetrics> read_metrics_;
		@@ -322,6 +325,19 @@ public:
		void
		protect_fds(bool protect);

		bool
		protect_files_generator() const;

		void
		protect_files_generator(bool protect);

		bool
		protect_files_consumer() const;

		void
		protect_files_consumer(bool protect);


		const std::shared_ptr<gkfs::messagepack::ClientMetrics>
		write_metrics();

src/client/gkfs_functions.cpp

+109 −5

Changes for src/client/gkfs_functions.cpp: 109 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -48,6 +48,8 @@
		#include <client/rpc/forward_data_proxy.hpp>
		#include <client/open_dir.hpp>
		#include <client/cache.hpp>
		#include <string>
		#include <string_view>

		#include <common/path_util.hpp>
		#ifdef GKFS_ENABLE_CLIENT_METRICS
		@@ -116,6 +118,65 @@ check_parent_dir(const std::string& path) {

		namespace gkfs::syscall {

		/**
		* @brief generate_lock_file
		* @param path
		* @param increase
		*
		* Creates, if it does not exist, a lock file, path+".lockgekko", empty
		* If increase is true, increase the size of the file +1
		* if increase is false, decrease the size of the file -1
		* If size == 0, delete the file
		* Using calls : forward_create, forward_stat, forward_remove, forward_decr_size
		* and forward_update_metadentry_size Proxy not supported
		*/
		void
		generate_lock_file(const std::string& path, bool increase) {
		auto lock_path = path + ".lockgekko";
		if(increase) {
		auto md = gkfs::utils::get_metadata(lock_path);
		if(!md) {
		gkfs::rpc::forward_create(lock_path, 0777 \| S_IFREG, 0);
		}
		gkfs::rpc::forward_update_metadentry_size(lock_path, 1, 0, false, 0);
		} else {
		auto md = gkfs::utils::get_metadata(lock_path);
		if(md) {
		if(md.value().size() == 1) {
		LOG(DEBUG, "Deleting Lock file {}", lock_path);
		gkfs::rpc::forward_remove(lock_path, false, 0);
		} else {
		gkfs::rpc::forward_decr_size(lock_path, md.value().size() - 1,
		0);
		}
		}
		}
		}

		/**
		* @brief generate_lock_file
		* @param path
		*
		* Test if the lock file exists, if it exists, wait 0.5 second and check again
		* (max 80 times) Using calls : forward_stat
		*/
		void
		test_lock_file(const std::string& path) {
		auto lock_path = path + ".lockgekko";
		auto md = gkfs::utils::get_metadata(lock_path);
		if(md) {
		LOG(DEBUG, "Lock file exists {} --> {}", lock_path, md->size());
		for(int i = 0; i < 80; i++) {
		if(!md) {
		break;
		}
		std::this_thread::sleep_for(std::chrono::milliseconds(500));
		md = gkfs::utils::get_metadata(lock_path);
		}
		}
		}


		/**
		* gkfs wrapper for open() system calls
		* errno may be set
		@@ -168,9 +229,14 @@ gkfs_open(const std::string& path, mode_t mode, int flags) {
		return -1;
		}
		} else {
		// file was successfully created. Add to filemap
		return CTX->file_map()->add(
		auto fd = CTX->file_map()->add(
		std::make_shared<gkfs::filemap::OpenFile>(path, flags));
		// CREATE_MODE
		if(CTX->protect_files_generator()) {
		generate_lock_file(path, true);
		}
		// file was successfully created. Add to filemap
		return fd;
		}
		} else {
		auto md_ = gkfs::utils::get_metadata(path);
		@@ -228,7 +294,7 @@ gkfs_open(const std::string& path, mode_t mode, int flags) {
		return -1;
		}
		}

		// RENAMED OR SYMLINK NOT PROTECTED
		return CTX->file_map()->add(
		std::make_shared<gkfs::filemap::OpenFile>(new_path, flags));
		}
		@@ -248,9 +314,19 @@ gkfs_open(const std::string& path, mode_t mode, int flags) {
		return -1;
		}
		}

		return CTX->file_map()->add(
		// NORMAL OPEN
		auto fd = CTX->file_map()->add(
		std::make_shared<gkfs::filemap::OpenFile>(path, flags));


		if(CTX->protect_files_consumer()) {
		test_lock_file(path);
		}

		if(CTX->protect_files_generator()) {
		generate_lock_file(path, true);
		}
		return fd;
		}

		/**
		@@ -1498,6 +1574,14 @@ gkfs_getdents(unsigned int fd, struct linux_dirent* dirp, unsigned int count) {
		while(pos < open_dir->size()) {
		// get dentry fir current position
		auto de = open_dir->getdent(pos);
		if(CTX->protect_files_consumer() or CTX->protect_files_generator()) {
		// if de.name ends with lockgekko jump to the next file
		if(de.name().size() >= 10 &&
		de.name().substr(de.name().size() - 10) == ".lockgekko") {
		pos++;
		continue;
		}
		}
		/*
		* Calculate the total dentry size within the kernel struct
		* `linux_dirent` depending on the file name size. The size is then
		@@ -1567,6 +1651,14 @@ gkfs_getdents64(unsigned int fd, struct linux_dirent64* dirp,
		struct linux_dirent64* current_dirp = nullptr;
		while(pos < open_dir->size()) {
		auto de = open_dir->getdent(pos);
		if(CTX->protect_files_consumer() or CTX->protect_files_generator()) {
		// if de.name ends with lockgekko jump to the next file
		if(de.name().size() >= 10 &&
		de.name().substr(de.name().size() - 10) == ".lockgekko") {
		pos++;
		continue;
		}
		}
		/*
		* Calculate the total dentry size within the kernel struct
		* `linux_dirent` depending on the file name size. The size is then
		@@ -1662,6 +1754,11 @@ gkfs_close(unsigned int fd) {
		CTX->file_map()->get(fd)->path());
		}
		}

		if(CTX->protect_files_generator()) {
		auto path = CTX->file_map()->get(fd)->path();
		generate_lock_file(path, false);
		}
		// No call to the daemon is required
		CTX->file_map()->remove(fd);
		return 0;
		@@ -1791,6 +1888,13 @@ gkfs_get_file_list(const std::string& path) {

		while(pos < open_dir->size()) {
		auto de = open_dir->getdent(pos++);
		if(CTX->protect_files_consumer() or CTX->protect_files_generator()) {
		// if de.name ends with lockgekko jump to the next file
		if(de.name().size() >= 10 &&
		de.name().substr(de.name().size() - 10) == ".lockgekko") {
		continue;
		}
		}
		file_list.push_back(de.name());
		}
		return file_list;

src/client/gkfs_libc.cpp

+14 −28

Changes for src/client/gkfs_libc.cpp: 14 added lines, 28 removed lines.
Original line number	Diff line number	Diff line
		@@ -68,7 +68,6 @@ std::atomic<bool> initializing{false};
		// Define a debug macro, can be easily disabled

		#define GKFS_TRACE
		// #define PATH_SHORTCUT

		#ifdef GKFS_DEBUG_BUILD
		#ifdef GKFS_TRACE
		@@ -106,7 +105,6 @@ initializeGekko() {
		void
		log_arguments(const char* symbol) {
		DEBUG_INFO("[BYPASS] {}", symbol);
		// printf("[BYPASS] %s\n",symbol);
		}

		// Variadic case: 1+ arguments
		@@ -117,7 +115,6 @@ log_arguments(const char* symbol, Args&&... args) {
		ss << "[BYPASS] Calling " << symbol << " with arguments: ";
		((ss << "[" << typeid(Args).name() << "] " << args << " "), ...);
		DEBUG_INFO("{}", ss.str());
		// printf("%s\n",ss.str().c_str());
		}

		// Variadic case: 1+ arguments
		@@ -128,7 +125,6 @@ log_argumentsx(const char* symbol, Args&&... args) {
		ss << "[BYPASS-ERROR] Calling " << symbol << " with arguments: ";
		((ss << "[" << typeid(Args).name() << "] " << args << " "), ...);
		DEBUG_INFO("{}", ss.str());
		// printf("%s\n",ss.str().c_str());
		}

		/**
		@@ -219,8 +215,6 @@ DLSYM_WRAPPER(int, openat, (int fd, const char* path, int flags, mode_t mode),
		(fd, path, flags, mode), "openat")
		DLSYM_WRAPPER(int, open64, (const char* path, int flags, mode_t mode),
		(path, flags, mode), "open64")
		// DLSYM_WRAPPER(int, __open_2, (char* path, int flags), (path, flags),
		// "__open")
		DLSYM_WRAPPER(int, close, (int fd), (fd), "close")
		DLSYM_WRAPPER(int, close_range,
		(unsigned int low, unsigned int high, int flags),
		@@ -383,7 +377,7 @@ DLSYM_WRAPPER(int, chmod, (char* path, mode_t mode), (path, mode), "chmod")
		DLSYM_WRAPPER(int, fchmod, (int fd, mode_t mode), (fd, mode), "fchmod")
		DLSYM_WRAPPER(int, chown, (char* path, uid_t owner, gid_t group),
		(path, owner, group), "chown")
		// DLSYM_WRAPPER(int, fcntl, (int fd, int cmd, ...), (fd, cmd, ...), "fcntl")

		static int (*real_fcntl)(int fd, int cmd, ...) = nullptr;

		DLSYM_WRAPPER(int, access, (const char* path, int mode), (path, mode), "access")
		@@ -559,16 +553,7 @@ enum class PathStatus { External, Internal, Error };
		PathStatus
		resolve_gkfs_path(int dirfd, const char* path, std::string& resolved,
		int flags = 0, bool resolve_last_link = true) {
		// if path does not start with CTX->mountdir() just return external
		#ifdef PATH_SHORTCUT
		auto res = std::mismatch(CTX->mountdir().begin(), CTX->mountdir().end(),
		std::string(path).begin());

		if(res.first != CTX->mountdir().end() and dirfd == AT_FDCWD) {
		return PathStatus::External;
		}

		#endif
		const auto status = CTX->relativize_fd_path(dirfd, path, resolved, flags,
		resolve_last_link);

		@@ -814,7 +799,7 @@ creat(const char* path, mode_t mode) {
		int
		ftruncate(int fd, off_t length) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		std::string path_str = CTX->file_map()->get(fd)->path();
		DEBUG_INFO("[GKFS] {}", path_str);
		return gkfs::syscall::gkfs_truncate(path_str, length);
		@@ -1027,7 +1012,7 @@ __lxstat64(int ver, const char* path, struct stat64* buf) {
		int
		__fxstat(int ver, int fd, struct stat* buf) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());
		return gkfs::syscall::gkfs_stat(CTX->file_map()->get(fd)->path(), buf,
		true, true);
		@@ -1046,7 +1031,7 @@ __fxstatat(int ver, int dfd, const char* path, struct stat* buf, int flags) {
		int
		__fxstat64(int ver, int fd, struct stat64* buf) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		struct stat st;
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());
		int res = gkfs::syscall::gkfs_stat(CTX->file_map()->get(fd)->path(),
		@@ -1061,7 +1046,7 @@ __fxstat64(int ver, int fd, struct stat64* buf) {
		int
		fstat64(int fd, struct stat64* buf) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		struct stat st;
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());
		int res = gkfs::syscall::gkfs_stat(CTX->file_map()->get(fd)->path(),
		@@ -1076,7 +1061,7 @@ fstat64(int fd, struct stat64* buf) {
		int
		fstat(int fd, struct stat* buf) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());

		// The fd could be a renamed file, thus when doing gkfs_stat we will get
		@@ -1280,6 +1265,7 @@ dup(int fd) {
		int
		dup2(int fd, int fd2) {
		initializeGekko();
		if(CTX->interception_enabled()) {
		if(is_gkfs_fd(fd) && is_gkfs_fd(fd2)) {
		DEBUG_INFO("[GKFS] DUP2 G{} --> G{}", fd, fd2);
		return gkfs::syscall::gkfs_dup2(fd, fd2);
		@@ -1290,7 +1276,7 @@ dup2(int fd, int fd2) {
		DEBUG_INFO("[GKFS-NON] DUP2 G{} --> {}", fd, fd2);
		return gkfs::syscall::gkfs_dup2(fd, fd2);
		}

		}
		// GKFS_OPERATION(dup2, fd, fd2);
		GKFS_FALLBACK(dup2, fd, fd2);
		}
		@@ -1380,7 +1366,7 @@ fcntl(int fd, int cmd, ...) // TODO
		va_end(myargs);

		// is from gekkofs?
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		int ret = -1;
		switch(cmd) {

		@@ -2011,7 +1997,7 @@ fflush(FILE* stream) {
		void*
		mmap(void* addr, size_t length, int prot, int flags, int fd, off_t offset) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());
		// create a malloc of length
		void* ptr = malloc(length);
		@@ -2078,7 +2064,7 @@ munmap(void* addr, size_t length) {
		int
		fchown(int fd, uid_t owner, gid_t group) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());
		return 0;
		}
		@@ -2089,7 +2075,7 @@ fchown(int fd, uid_t owner, gid_t group) {
		int
		fchmod(int fd, mode_t mode) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());
		return 0;
		}
		@@ -2100,7 +2086,7 @@ fchmod(int fd, mode_t mode) {
		int
		futimes(int fd, const struct timeval tv[2]) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());
		return 0;
		}
		@@ -2139,7 +2125,7 @@ listxattr(const char* path, char* list, size_t size) {
		ssize_t
		flistxattr(int fd, char* list, size_t size) {
		initializeGekko();
		if(is_gkfs_fd(fd)) {
		if(CTX->interception_enabled() && is_gkfs_fd(fd)) {
		DEBUG_INFO("[GKFS] {}", CTX->file_map()->get(fd)->path());
		return 0;
		}

src/client/hooks.cpp

+2 −2

Changes for src/client/hooks.cpp: 2 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -722,7 +722,8 @@ hook_chdir(const char* path) {
		// path falls in our namespace
		auto md = gkfs::utils::get_metadata(rel_path);
		if(!md) {
		LOG(ERROR, "{}() path {} / {} errno {}", __func__, path, rel_path, errno);
		LOG(ERROR, "{}() path {} / {} errno {}", __func__, path, rel_path,
		errno);
		return -errno;
		}

		@@ -929,7 +930,6 @@ hook_fcntl(unsigned int fd, unsigned int cmd, unsigned long arg) {
		return gsl::narrow_cast<int>(
		syscall_no_intercept_wrapper(SYS_fcntl, fd, cmd, arg));


		case F_SETLK:
		LOG(ERROR, "{}() F_SETLK on fd (on underlying fd) {}", __func__,
		fd);

Lock System for cargo

Merge request reports