Retry legacy directory RPCs after ENOBUFS (e9f60961) · Commits · hpc / gekkofs

src/client/rpc/forward_metadata.cpp

+52 −9

Changes for src/client/rpc/forward_metadata.cpp: 52 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		@@ -556,20 +556,19 @@ forward_get_dirents(const string& path) {
		LOG(DEBUG, "{}() enter for path '{}'", __func__, path);
		auto const targets = CTX->distributor()->locate_directory_metadata();

		auto large_buffer = std::unique_ptr<char[]>(
		new char[gkfs::config::rpc::dirents_buff_size]);
		const std::size_t per_host_buff_size =
		gkfs::config::rpc::dirents_buff_size / targets.size();

		std::vector<std::vector<char>> buffers(targets.size());
		std::vector<thallium::bulk> exposed_buffers;
		exposed_buffers.reserve(targets.size());

		// Create bulk handles
		for(std::size_t i = 0; i < targets.size(); ++i) {
		void* buf_ptr = large_buffer.get() + (i * per_host_buff_size);
		buffers[i].resize(per_host_buff_size);
		try {
		std::vector<std::pair<void*, std::size_t>> segments = {
		std::make_pair(buf_ptr, per_host_buff_size)};
		std::make_pair(buffers[i].data(), buffers[i].size())};
		exposed_buffers.emplace_back(CTX->rpc_engine()->expose(
		segments, thallium::bulk_mode::read_write));
		} catch(const std::exception& e) {
		@@ -620,21 +619,65 @@ forward_get_dirents(const string& path) {
		gkfs::rpc::rpc_get_dirents_out_t out = waiters[i].wait(); // blocks

		if(out.err == ENOBUFS) {
		LOG(WARNING,
		"{}() Buffer too small for host '{}'. Required: {}. Retrying logic not implemented.",
		const auto required_size = out.dirents_size;
		if(required_size <= buffers[i].size()) {
		LOG(ERROR,
		"{}() Host '{}' reported invalid directory-entry buffer size {} (current {})",
		__func__, targets[i], required_size, buffers[i].size());
		err = ENOBUFS;
		continue;
		}

		LOG(DEBUG,
		"{}() Buffer too small for host '{}'. Growing from {} to {} bytes and retrying.",
		__func__, targets[i], buffers[i].size(), required_size);

		buffers[i].resize(required_size);
		try {
		std::vector<std::pair<void*, std::size_t>> segments = {
		std::make_pair(buffers[i].data(),
		buffers[i].size())};
		exposed_buffers[i] = CTX->rpc_engine()->expose(
		segments, thallium::bulk_mode::read_write);

		gkfs::rpc::rpc_get_dirents_in_t retry_in;
		retry_in.path = root_path;
		retry_in.start_key = "";
		retry_in.bulk_handle = exposed_buffers[i];
		auto retry_waiter = forward_async_with_timeout(
		get_dirents_rpc, CTX->hosts().at(targets[i]),
		retry_in);
		out = retry_waiter.wait();
		} catch(const std::exception& ex) {
		LOG(ERROR,
		"{}() Retry of directory-entry RPC for host '{}' failed: {}",
		__func__, targets[i], ex.what());
		err = EBUSY;
		continue;
		}

		if(out.err == ENOBUFS) {
		LOG(ERROR,
		"{}() Directory-entry buffer for host '{}' still too small after retry (required {})",
		__func__, targets[i], out.dirents_size);
		err = ENOBUFS;
		continue;
		}
		if(out.err != 0) {
		LOG(ERROR, "Host reported error after buffer retry: {}",
		out.err);
		err = out.err;
		continue;
		}
		} else if(out.err != 0) {
		LOG(ERROR, "Host reported error: {}", out.err);
		err = out.err;
		} else {
		void* base_ptr = large_buffer.get() + (i * per_host_buff_size);

		// 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, base_ptr, per_host_buff_size);
		out, buffers[i].data(), buffers[i].size());
		for(auto& e : entries) {
		auto type = get<1>(e);
		gkfs::filemap::FileType ftype =