fix(ci,client): disable shm by default, fix ASan leak, and fix compatibility_corpus env/timeouts (a3b64fba) · Commits · hpc / gekkofs

include/client/preload_context.hpp

+20 −0

Changes for include/client/preload_context.hpp: 20 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -292,6 +292,8 @@ 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_;


		@@ -362,6 +364,24 @@ 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;
		}

		bool
		negative_stat_cache() const noexcept {
		return negative_stat_cache_;
		}

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

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
		@@ -87,6 +87,8 @@ 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

scripts/compatibility_corpus.py

+40 −3

Changes for scripts/compatibility_corpus.py: 40 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -39,10 +39,19 @@ def snapshot_tree(root, env=None, client=None, errors=None):
		pending = [Path(root)]
		while pending:
		directory = pending.pop()
		try:
		result = subprocess.run(
		[str(client), "readdir", str(directory)], env=env,
		capture_output=True, text=True, check=False,
		capture_output=True, text=True, check=False, timeout=30,
		)
		except subprocess.TimeoutExpired:
		if errors is not None:
		errors.append({
		"command": [str(client), "readdir", str(directory)],
		"returncode": -1,
		"error": "timed out after 30s",
		})
		continue
		if result.returncode != 0:
		if errors is not None:
		errors.append({
		@@ -81,10 +90,19 @@ def snapshot_tree(root, env=None, client=None, errors=None):
		path = directory / name
		dirent_type = dirent.get("d_type")
		if dirent_type == 10: # DT_LNK
		try:
		target = subprocess.run(
		["readlink", str(path)], env=env, capture_output=True,
		text=True, check=False,
		text=True, check=False, timeout=30,
		)
		except subprocess.TimeoutExpired:
		if errors is not None:
		errors.append({
		"command": ["readlink", str(path)],
		"returncode": -1,
		"error": "timed out after 30s",
		})
		continue
		if target.returncode != 0:
		if errors is not None:
		errors.append({
		@@ -289,8 +307,9 @@ class Daemon:
		self.process = None


		def run_command(command, cwd, env):
		def run_command(command, cwd, env, timeout=60):
		started = time.monotonic()
		try:
		result = subprocess.run(
		command,
		cwd=cwd,
		@@ -300,6 +319,7 @@ def run_command(command, cwd, env):
		encoding="utf-8",
		errors="replace",
		check=False,
		timeout=timeout,
		)
		return {
		"command": command,
		@@ -308,6 +328,14 @@ def run_command(command, cwd, env):
		"stdout": result.stdout[-4000:],
		"stderr": result.stderr[-4000:],
		}
		except subprocess.TimeoutExpired as exc:
		return {
		"command": command,
		"returncode": -1,
		"elapsed_seconds": round(time.monotonic() - started, 6),
		"stdout": (exc.stdout or "")[-4000:] if isinstance(exc.stdout, str) else "",
		"stderr": (exc.stderr or "")[-4000:] if isinstance(exc.stderr, str) else "Command timed out after {}s".format(timeout),
		}


		def workload_commands(name, mount, outside):
		@@ -461,7 +489,16 @@ def run_mode(mode, args):
		*library_dirs, client_env.get("LD_LIBRARY_PATH", "")
		]))
		client_env["LD_PRELOAD"] = str(library.resolve())
		client_env["LIBGKFS_HOSTS_FILE"] = str(hostfile)
		client_env["GKFS_HOSTS_FILE"] = str(hostfile)
		client_env["LIBGKFS_MOUNTDIR"] = str(mount)
		client_env["GKFS_MOUNTDIR"] = str(mount)
		client_env["LIBGKFS_CLEAN_RPC_SHUTDOWN"] = "1"
		client_env["LIBGKFS_SHM_STAT_CACHE"] = "0"
		daemon_env = os.environ.copy()
		daemon_env["LD_LIBRARY_PATH"] = os.pathsep.join(filter(None, [
		*library_dirs, daemon_env.get("LD_LIBRARY_PATH", "")
		]))
		daemon = Daemon(daemon_binary, root, mount, hostfile, daemon_env)
		try:
		daemon.start()

src/client/preload.cpp

+21 −0

Changes for src/client/preload.cpp: 21 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -833,6 +833,27 @@ init_preload() {
		CTX->stat_cache_ttl_ms(stat_ttl);
		LOG(INFO, "Short-TTL stat cache: {} ms", CTX->stat_cache_ttl_ms());

		bool shm_stat_cache =
		gkfs::env::get_var_bool(gkfs::env::SHM_STAT_CACHE, false);
		if(!shm_stat_cache && gkfs::env::var_is_set("LIBGKFS_SHM_STAT_CACHE")) {
		shm_stat_cache =
		gkfs::env::get_var_bool("LIBGKFS_SHM_STAT_CACHE", false);
		}
		CTX->shm_stat_cache_enabled(shm_stat_cache);
		LOG(INFO, "Shared memory stat cache (/dev/shm): {}",
		CTX->shm_stat_cache_enabled() ? "ON" : "OFF");

		bool neg_stat_cache =
		gkfs::env::get_var_bool(gkfs::env::NEGATIVE_STAT_CACHE, false);
		if(!neg_stat_cache &&
		gkfs::env::var_is_set("LIBGKFS_NEGATIVE_STAT_CACHE")) {
		neg_stat_cache =
		gkfs::env::get_var_bool("LIBGKFS_NEGATIVE_STAT_CACHE", false);
		}
		CTX->negative_stat_cache(neg_stat_cache);
		LOG(INFO, "Negative stat cache (ENOENT): {}",
		CTX->negative_stat_cache() ? "ON" : "OFF");

		CTX->enable_interception();
		#ifndef BYPASS_SYSCALL
		gkfs::preload::start_interception();

src/client/preload_context.cpp

+50 −15

Changes for src/client/preload_context.cpp: 50 added lines, 15 removed lines.
Original line number	Diff line number	Diff line
		@@ -206,26 +206,48 @@ PreloadContext::~PreloadContext() {
		// only place its reload watcher is guaranteed to get joined.
		stop_reload_watcher_thread();

		#if defined(__SANITIZE_ADDRESS__) \|\| defined(__SANITIZE_LEAK__)
		constexpr bool under_sanitizer = true;
		#elif defined(__has_feature)
		#if __has_feature(address_sanitizer) \|\| __has_feature(leak_sanitizer)
		constexpr bool under_sanitizer = true;
		#else
		constexpr bool under_sanitizer = false;
		#endif
		#else
		constexpr bool under_sanitizer = false;
		#endif

		const char* clean_rpc_shutdown = std::getenv("LIBGKFS_CLEAN_RPC_SHUTDOWN");
		if(clean_rpc_shutdown == nullptr \|\|
		(std::strcmp(clean_rpc_shutdown, "1") != 0 &&
		std::strcmp(clean_rpc_shutdown, "ON") != 0 &&
		std::strcmp(clean_rpc_shutdown, "true") != 0)) {
		const bool asan_env = std::getenv("ASAN_OPTIONS") != nullptr \|\|
		std::getenv("LSAN_OPTIONS") != nullptr;
		const bool clean_requested = clean_rpc_shutdown != nullptr &&
		(std::strcmp(clean_rpc_shutdown, "1") == 0 \|\|
		std::strcmp(clean_rpc_shutdown, "ON") == 0 \|\|
		std::strcmp(clean_rpc_shutdown, "true") == 0);

		if(!under_sanitizer && !asan_env && !clean_requested) {
		// PreloadContext is a function-local static whose destructor runs via
		// __cxa_atexit BEFORE .fini_array (destroy_preload). Retain a heap
		// reference to the RPC/IPC engines and endpoint snapshot so member
		// destruction does not invoke margo_finalize() (~200 ms futex waits).
		if(routing_ && !routing_->hosts.empty()) {
		auto* leaked_routing =
		new std::shared_ptr<const RoutingSnapshot>(routing_);
		(void) leaked_routing;
		}
		if(rpc_engine_) {
		auto* leaked_rpc_engine =
		new std::shared_ptr<thallium::engine>(rpc_engine_);
		(void) leaked_rpc_engine;
		}
		if(ipc_engine_) {
		auto* leaked_ipc_engine =
		new std::shared_ptr<thallium::engine>(ipc_engine_);
		(void) leaked_routing;
		(void) leaked_rpc_engine;
		(void) leaked_ipc_engine;
		}
		}
		}

		void
		PreloadContext::init_logging() {
		@@ -446,7 +468,25 @@ PreloadContext::stat_cache_ttl_ms(uint64_t ttl_ms) noexcept {
		shm_stat_cache_ = nullptr;
		}
		stat_cache_.clear();
		} else if(shm_stat_cache_ == nullptr && !mountdir_.empty()) {
		} else if(shm_stat_cache_enabled_ && shm_stat_cache_ == nullptr &&
		!mountdir_.empty()) {
		init_shm_stat_cache();
		}
		}

		void
		PreloadContext::shm_stat_cache_enabled(bool enabled) noexcept {
		shm_stat_cache_enabled_ = enabled;
		std::lock_guard<std::mutex> lock(stat_cache_mutex_);
		if(!enabled) {
		if(shm_stat_cache_ != nullptr) {
		::syscall_no_intercept(
		SYS_munmap, reinterpret_cast<long>(shm_stat_cache_),
		static_cast<long>(sizeof(ShmStatCacheRegion)));
		shm_stat_cache_ = nullptr;
		}
		} else if(shm_stat_cache_ == nullptr && !mountdir_.empty() &&
		stat_cache_ttl_ms_ > 0) {
		init_shm_stat_cache();
		}
		}
		@@ -459,13 +499,8 @@ PreloadContext::init_shm_stat_cache() {
		static_cast<long>(sizeof(ShmStatCacheRegion)));
		shm_stat_cache_ = nullptr;
		}
		if(mountdir_.empty() \|\| stat_cache_ttl_ms_ == 0) {
		return;
		}
		const char* shm_env = std::getenv("LIBGKFS_SHM_STAT_CACHE");
		if(shm_env != nullptr &&
		(std::strcmp(shm_env, "0") == 0 \|\| std::strcmp(shm_env, "OFF") == 0 \|\|
		std::strcmp(shm_env, "false") == 0)) {
		if(!shm_stat_cache_enabled_ \|\| mountdir_.empty() \|\|
		stat_cache_ttl_ms_ == 0) {
		return;
		}
		const auto hostfile_path = gkfs::env::get_var(gkfs::env::HOSTS_FILE,
		@@ -858,7 +893,7 @@ PreloadContext::put_stat_cache(const std::string& path, const std::string& attr,

		void
		PreloadContext::put_negative_stat_cache(const std::string& path) {
		if(stat_cache_ttl_ms_ == 0 \|\| path.empty()) {
		if(!negative_stat_cache_ \|\| stat_cache_ttl_ms_ == 0 \|\| path.empty()) {
		return;
		}
		if(path.size() >= 10 &&