Fix directory bulk lifetimes and recursive sfind filtering (2876eec2) · Commits · hpc / gekkofs

docs/sphinx/users/running.md

+13 −0

Changes for docs/sphinx/users/running.md: 13 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -246,6 +246,19 @@ GekkoFS allows to use external functions on your client code, via LD_PRELOAD.
		Source code needs to be compiled with -fPIC. We include a pfind io500 substitution,
		`examples/gfind/gfind.cpp` and a non-mpi version `examples/gfind/sfind.cpp`

		`sfind` scans all descendants through extended directory enumeration; it does
		not walk each subdirectory separately. `MATCHED X/Y` counts matching
		non-directory entries in `X` and all scanned entries (including directories)
		in `Y`. `--server-side` applies the same filters on the daemon, and `-C` returns
		counts without entry data.

		Name filters use substring regex matching against paths relative to the search
		root. `-name` converts `` to `.`; it is not a full POSIX `find -name` glob
		implementation. For example, `-name '*file_5'` also matches `file_50`.
		Use `-regex '(^\|/)file_5$'` to match only entries whose basename is `file_5`,
		including descendants. Failed scans return a nonzero exit status; partial
		counts from such runs are not valid results.

		### Data distributors

		The data distribution can be selected at compilation time, we have 2 distributors available:

examples/gfind/sfind.cpp

+15 −5

Changes for examples/gfind/sfind.cpp: 15 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -43,6 +43,7 @@
		#include <cstdint>
		#include <cstdio>
		#include <cstring>
		#include <dirent.h>
		#include <fstream>
		#include <getopt.h>
		#include <iostream>
		@@ -287,6 +288,7 @@ struct ThreadData {
		pthread_mutex_t* queue_mutex;
		atomic<unsigned long long>* total_found;
		atomic<unsigned long long>* total_checked;
		atomic<bool>* failed;
		};

		// The function each worker thread will execute
		@@ -349,6 +351,7 @@ worker_routine(void* arg) {


		if(n < 0) {
		data->failed->store(true);
		cerr << "Warning: Rank " << sfind_rank << " Thread "
		<< data->thread_id << " received error from server "
		<< server_id << " during filtered scan." << endl;
		@@ -372,8 +375,10 @@ worker_routine(void* arg) {
		total_bytes - bytes_processed;
		struct dirent_extended* temp = nullptr;
		string entry_name;
		if(!read_entry(ptr, remaining, temp, entry_name))
		if(!read_entry(ptr, remaining, temp, entry_name)) {
		data->failed->store(true);
		break;
		}

		local_found++;

		@@ -419,6 +424,7 @@ worker_routine(void* arg) {

		if(n <= 0) {
		if(n < 0) {
		data->failed->store(true);
		cerr << "Warning: Rank " << sfind_rank << " Thread "
		<< data->thread_id << " received error from server "
		<< server_id << endl;
		@@ -436,10 +442,12 @@ worker_routine(void* arg) {
		total_bytes - bytes_processed;
		struct dirent_extended* temp = nullptr;
		string entry_name;
		if(!read_entry(ptr, remaining, temp, entry_name))
		if(!read_entry(ptr, remaining, temp, entry_name)) {
		data->failed->store(true);
		break;
		}

		if(temp->d_type != 1) {
		if(temp->d_type != DT_DIR) {
		bool timeOK = data->opt->timestamp_file.empty() \|\|
		((uint64_t) temp->ctime >= runtime.ctime_min);
		bool sizeOK =
		@@ -472,6 +480,7 @@ int
		process_parallel_pthreads(const pfind_options_t* opt) {
		atomic<unsigned long long> global_found(0);
		atomic<unsigned long long> global_checked(0);
		atomic<bool> failed(false);
		runtime = {};

		if(!opt->timestamp_file.empty()) {
		@@ -549,7 +558,8 @@ process_parallel_pthreads(const pfind_options_t* opt) {
		&server_queue,
		&queue_mutex,
		&global_found,
		&global_checked};
		&global_checked,
		&failed};
		pthread_create(&threads[i], nullptr, worker_routine, &thread_args[i]);
		}

		@@ -573,7 +583,7 @@ process_parallel_pthreads(const pfind_options_t* opt) {
		output_file << "MATCHED " << global_found << "/" << global_checked << endl;
		output_file.close();

		return 0;
		return failed.load() ? 1 : 0;
		}

		int

include/client/rpc/utils.hpp

+27 −17

Changes for include/client/rpc/utils.hpp: 27 added lines, 17 removed lines.
Original line number	Diff line number	Diff line
		@@ -191,28 +191,25 @@ decompress_and_parse_entries(
		entries.reserve(static_cast<size_t>((end - p) / 32) + 1);

		while(p < end) {
		if(p + sizeof(unsigned char) > end) {
		LOG(ERROR, "{}() Unexpected end of buffer while parsing type",
		__func__);
		break;
		if(static_cast<size_t>(end - p) < sizeof(unsigned char)) {
		throw std::runtime_error("Truncated directory-entry type");
		}
		unsigned char type = 0;
		std::memcpy(&type, p, sizeof(unsigned char));
		p += sizeof(unsigned char);
		if(type > 2) {
		throw std::runtime_error("Invalid directory-entry type");
		}

		if(p + sizeof(size_t) > end) {
		LOG(ERROR, "{}() Unexpected end of buffer while parsing file_size",
		__func__);
		break;
		if(static_cast<size_t>(end - p) < sizeof(size_t)) {
		throw std::runtime_error("Truncated directory-entry size");
		}
		size_t file_size = 0;
		std::memcpy(&file_size, p, sizeof(size_t));
		p += sizeof(size_t);

		if(p + sizeof(time_t) > end) {
		LOG(ERROR, "{}() Unexpected end of buffer while parsing ctime",
		__func__);
		break;
		if(static_cast<size_t>(end - p) < sizeof(time_t)) {
		throw std::runtime_error("Truncated directory-entry ctime");
		}
		time_t ctime = 0;
		std::memcpy(&ctime, p, sizeof(time_t));
		@@ -221,17 +218,30 @@ decompress_and_parse_entries(
		// Name is null-terminated. We must check we don't go past 'end'.
		size_t name_len = strnlen(p, end - p);
		if(p + name_len >= end) {
		LOG(ERROR, "{}() Unexpected end of buffer while parsing name",
		__func__);
		break;
		throw std::runtime_error("Truncated directory-entry name");
		}

		std::string name(p, name_len);
		p += name_len + 1;

		if(!name.empty()) {
		entries.emplace_back(name, type, file_size, ctime);
		// Extended listings include relative descendant paths, not just
		// immediate basenames. Reject traversal, but retain nested entries.
		if(name.empty() \|\| name.front() == '/' \|\| name.back() == '/') {
		throw std::runtime_error("Invalid directory-entry name");
		}
		size_t component_start = 0;
		while(component_start < name.size()) {
		const auto separator = name.find('/', component_start);
		const auto component =
		name.substr(component_start, separator - component_start);
		if(component.empty() \|\| component == "." \|\| component == "..") {
		throw std::runtime_error("Invalid directory-entry path");
		}
		if(separator == std::string::npos)
		break;
		component_start = separator + 1;
		}
		entries.emplace_back(name, type, file_size, ctime);
		}

		return entries;

src/client/rpc/forward_metadata.cpp

+64 −18

Changes for src/client/rpc/forward_metadata.cpp: 64 added lines, 18 removed lines.
Original line number	Diff line number	Diff line
		@@ -527,27 +527,27 @@ decompress_and_parse_entries_standard(

		while(p < end) {
		if(p + sizeof(unsigned char) > end) {
		LOG(ERROR, "{}() Unexpected end of buffer while parsing type",
		__func__);
		break;
		throw std::runtime_error("Truncated directory-entry type");
		}
		unsigned char type = reinterpret_cast<const unsigned char>(p);
		p += sizeof(unsigned char);
		if(type > 2) {
		throw std::runtime_error("Invalid directory-entry type");
		}

		// Name is null-terminated. We must check we don't go past 'end'.
		size_t name_len = strnlen(p, end - p);
		if(p + name_len >= end) {
		LOG(ERROR, "{}() Unexpected end of buffer while parsing name",
		__func__);
		break;
		throw std::runtime_error("Truncated directory-entry name");
		}

		std::string name(p, name_len);
		p += name_len + 1;

		if(!name.empty()) {
		entries.emplace_back(name, type, 0, 0);
		if(name.empty() \|\| name.find('/') != std::string::npos) {
		throw std::runtime_error("Invalid directory-entry name");
		}
		entries.emplace_back(name, type, 0, 0);
		}

		return entries;
		@@ -558,8 +558,11 @@ forward_get_dirents(const string& path) {
		LOG(DEBUG, "{}() enter for path '{}'", __func__, path);
		auto const targets = CTX->distributor()->locate_directory_metadata();

		if(targets.empty()) {
		return make_pair(EIO, nullptr);
		}
		const std::size_t per_host_buff_size =
		gkfs::config::rpc::dirents_buff_size / targets.size();
		std::max<std::size_t>(1, CTX->dirents_buff_size() / targets.size());

		std::vector<std::vector<char>> buffers(targets.size());
		std::vector<thallium::bulk> exposed_buffers;
		@@ -634,6 +637,8 @@ forward_get_dirents(const string& path) {
		"{}() Buffer too small for host '{}'. Growing from {} to {} bytes and retrying.",
		__func__, targets[i], buffers[i].size(), required_size);

		// Release the registration before reallocating its storage.
		exposed_buffers[i] = thallium::bulk{};
		buffers[i].resize(required_size);
		try {
		std::vector<std::pair<void*, std::size_t>> segments = {
		@@ -671,15 +676,21 @@ forward_get_dirents(const string& path) {
		err = out.err;
		continue;
		}
		} else if(out.err != 0) {
		}
		if(out.err != 0) {
		LOG(ERROR, "Host reported error: {}", out.err);
		err = out.err;
		continue;
		} else {
		// Decompress and parse entries
		// The decompress function expects rpc_get_dirents_out_t
		// which matches the Thallium RPC output.
		auto entries = decompress_and_parse_entries_standard(
		out, buffers[i].data(), buffers[i].size());
		LOG(DEBUG,
		"{}() Host '{}' returned {} bytes and {} directory entries for '{}'",
		__func__, targets[i], out.dirents_size, entries.size(),
		path);
		for(auto& e : entries) {
		auto type = get<1>(e);
		gkfs::filemap::FileType ftype =
		@@ -693,6 +704,10 @@ forward_get_dirents(const string& path) {
		}
		}

		} catch(const std::runtime_error& ex) {
		LOG(ERROR, "Directory-entry response from host '{}' failed: {}",
		targets[i], ex.what());
		err = EIO;
		} catch(const std::exception& ex) {
		LOG(ERROR, "RPC wait failed: {}", ex.what());
		err = EBUSY;
		@@ -1079,12 +1094,10 @@ forward_get_dirents_single(const string& path, int server,
		root_path += '/';
		}

		thread_local std::vector<char> large_buffer;
		size_t buffer_size =
		std::max<std::size_t>(std::size_t{1}, CTX->dirents_buff_size());
		if(large_buffer.size() < buffer_size) {
		large_buffer.resize(buffer_size);
		}
		// Each listing owns its storage, including while an RPC is outstanding.
		std::vector<char> large_buffer(buffer_size);
		thallium::bulk exposed_buffer;
		bool expose_needed = true;

		@@ -1125,6 +1138,11 @@ forward_get_dirents_single(const string& path, int server,
		err = ENOBUFS;
		break;
		}
		// Both handles retain the registration. Release them
		// before resize can invalidate the registered address.
		in.bulk_handle = thallium::bulk{};
		exposed_buffer = thallium::bulk{};
		err = ENOBUFS;
		buffer_size = required_size;
		if(large_buffer.size() < buffer_size) {
		large_buffer.resize(buffer_size);
		@@ -1143,6 +1161,10 @@ forward_get_dirents_single(const string& path, int server,
		return make_pair(0, std::move(all_entries));
		}

		if(get<0>(current_entries.back()) <= start_key) {
		throw std::runtime_error(
		"Directory-entry pagination did not advance");
		}
		start_key = get<0>(current_entries.back());
		// We need to copy the entries to the complete list
		std::copy(current_entries.begin(), current_entries.end(),
		@@ -1154,6 +1176,11 @@ forward_get_dirents_single(const string& path, int server,
		__func__, start_key, current_entries.size(), buffer_size);
		break; // break attempt loop

		} catch(const std::runtime_error& ex) {
		LOG(ERROR, "{}() Invalid directory-entry response: '{}'",
		__func__, ex.what());
		err = EIO;
		break;
		} catch(const std::exception& ex) {
		LOG(ERROR, "{}() RPC failed: '{}'", __func__, ex.what());
		err = EBUSY;
		@@ -1177,6 +1204,13 @@ forward_get_dirents_filtered(const std::string& path, int server,
		int64_t filter_size, int64_t filter_ctime,
		bool count_only) {

		if(server < 0 \|\| static_cast<size_t>(server) >= CTX->hosts()->size()) {
		return std::make_tuple(
		EINVAL,
		std::vector<std::tuple<const std::string, unsigned char, size_t,
		time_t>>{},
		uint64_t{0}, uint64_t{0}, std::string{});
		}
		auto endp = CTX->hosts()->at(server);

		gkfs::rpc::rpc_get_dirents_filtered_in_t in;
		@@ -1191,7 +1225,8 @@ forward_get_dirents_filtered(const std::string& path, int server,
		// TODO makes sure the buffer size is sufficient
		// We shouldn't use the standard dirent size but the extended one
		// The extended one is bigger
		size_t buffer_size = gkfs::config::rpc::dirents_buff_size;
		size_t buffer_size =
		std::max<std::size_t>(std::size_t{1}, CTX->dirents_buff_size());

		std::vector<std::tuple<const std::string, unsigned char, size_t, time_t>>
		entries;
		@@ -1204,6 +1239,8 @@ forward_get_dirents_filtered(const std::string& path, int server,
		while(true) {
		std::unique_ptr<char[]> large_buffer;
		tl::bulk exposed_buffer;
		// The input bulk copy must also die before this iteration's storage.
		gkfs::rpc::rpc_get_dirents_filtered_in_t request_in = in;

		if(!count_only) {
		large_buffer = std::unique_ptr<char[]>(new char[buffer_size]);
		@@ -1220,20 +1257,24 @@ forward_get_dirents_filtered(const std::string& path, int server,
		}
		}

		in.bulk_handle = exposed_buffer;
		request_in.bulk_handle = exposed_buffer;

		auto get_dirents_rpc = gkfs::rpc::define_rpc(
		CTX->rpc_engine(), gkfs::rpc::tag::get_dirents_filtered);

		try {
		gkfs::rpc::rpc_get_dirents_filtered_out_t out =
		forward_with_timeout(get_dirents_rpc, endp, in);
		forward_with_timeout(get_dirents_rpc, endp, request_in);

		if(out.err == ENOBUFS) {
		size_t required_size = out.dirents_size;
		LOG(WARNING,
		"{}() Buffer too small. Server requested {} bytes. Retrying.",
		__func__, required_size);
		if(count_only \|\| required_size <= buffer_size) {
		err = ENOBUFS;
		break;
		}
		buffer_size = required_size;
		continue; // Retry with new buffer size
		} else if(out.err != 0) {
		@@ -1244,7 +1285,7 @@ forward_get_dirents_filtered(const std::string& path, int server,
		if(!count_only) {
		// reuse standard decompression for now as format is same
		entries = gkfs::rpc::decompress_and_parse_entries(
		out, large_buffer.get());
		out, large_buffer.get(), buffer_size);
		}

		last_scanned_key = out.last_scanned_key;
		@@ -1252,6 +1293,11 @@ forward_get_dirents_filtered(const std::string& path, int server,
		matched_count = out.matched_count;
		break;

		} catch(const std::runtime_error& e) {
		LOG(ERROR, "{}() Invalid filtered directory-entry response: '{}'",
		__func__, e.what());
		err = EIO;
		break;
		} catch(const std::exception& e) {
		LOG(ERROR, "{}() Failed to get filtered dirents: '{}'", __func__,
		e.what());

src/client/rpc/forward_metadata_proxy.cpp

+11 −0

Changes for src/client/rpc/forward_metadata_proxy.cpp: 11 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -232,6 +232,13 @@ forward_get_dirents_single_proxy_v2(const string& path, int server,
		LOG(WARNING,
		"{}() Buffer too small. Proxy requested {} bytes. Retrying.",
		__func__, required_size);
		if(required_size <= buffer_size) {
		err = ENOBUFS;
		break;
		}
		in.bulk_handle = thallium::bulk{};
		exposed_buffer = thallium::bulk{};
		err = ENOBUFS;
		buffer_size = required_size;
		large_buffer.resize(buffer_size);
		continue;
		@@ -252,6 +259,10 @@ forward_get_dirents_single_proxy_v2(const string& path, int server,
		return make_pair(0, std::move(all_entries));
		}

		if(get<0>(entries_vector.back()) <= start_key) {
		throw std::runtime_error(
		"Directory-entry pagination did not advance");
		}
		// We need to copy the entries to the complete list
		std::copy(entries_vector.begin(), entries_vector.end(),
		std::back_inserter(*all_entries));