bug fix (1397c1ff) · Commits · hpc / gekkofs

CHANGELOG.md

+1 −1

Changes for CHANGELOG.md: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -23,7 +23,7 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
		- Enable inline data for small files (`LIBGKFS_USE_INLINE_DATA`).
		- Create write optimization (`LIBGKFS_CREATE_WRITE_OPTIMIZATION`).
		- Read inline prefetch (`LIBGKFS_READ_INLINE_PREFETCH`).
		- Dirents compression (`LIBGKFS_USE_DIRENTS_COMPRESSION and GKFS_DAEMON_USE_DIRENTS_COMPRESSION`).
		- Dirents compression (`GKFS_USE_DIRENTS_COMPRESSION`).
		- Dirents buffer size control (`LIBGKFS_DIRENTS_BUFF_SIZE`).
		- New sfind filtering in the server side
		- Added new tests (and enabling failing ones) to increase coverage

src/client/gkfs_data.cpp

+6 −0

Changes for src/client/gkfs_data.cpp: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -92,6 +92,12 @@ ssize_t
		gkfs_do_write(gkfs::filemap::OpenFile& file, const char* buf, size_t count,
		off64_t offset, bool update_pos) {

		// Writes may immediately follow O_CREAT. Complete queued metadata creates
		// before accessing the file on the daemon.
		if(CTX->use_metadata_batch()) {
		CTX->flush_metadata_batches();
		}

		if(file.type() != gkfs::filemap::FileType::regular) {
		assert(file.type() == gkfs::filemap::FileType::directory);
		LOG(WARNING, "Cannot write to directory");

src/client/gkfs_metadata.cpp

+4 −0

Changes for src/client/gkfs_metadata.cpp: 4 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -1957,6 +1957,10 @@ gkfs_close(unsigned int fd) {
		if(file) {
		const auto path = file->path();

		if(CTX->use_metadata_batch()) {
		CTX->flush_metadata_batches();
		}

		if(CTX->use_async_write()) {
		CTX->wait_async_writes();
		}

src/client/preload.cpp

+6 −0

Changes for src/client/preload.cpp: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -503,8 +503,14 @@ init_environment() {
		} catch(...) {
		CTX->metadata_batch_threshold(64);
		}
		if(CTX->metadata_batch_threshold() == 0) {
		LOG(WARNING,
		"Metadata batching requested with threshold 0; disabling it");
		CTX->use_metadata_batch(false);
		} else {
		LOG(INFO, "Metadata batching enabled with threshold: {}",
		CTX->metadata_batch_threshold());
		}
		} else {
		CTX->use_metadata_batch(false);
		LOG(INFO, "Metadata batching disabled.");

src/client/preload_context.cpp

+37 −2

Changes for src/client/preload_context.cpp: 37 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -1084,8 +1084,29 @@ PreloadContext::flush_metadata_batches() {
		LOG(DEBUG, "{}() Flushing all {} batches...", __func__, batches.size());
		}

		const auto requeue = [&](uint64_t host_id,
		const std::vector<std::string>& paths,
		const std::vector<uint32_t>& modes, int err) {
		std::lock_guard<std::mutex> lock(metadata_batch_mutex_);
		auto& queue = metadata_batch_buffer_[host_id];
		queue.reserve(queue.size() + paths.size());
		queue.insert(queue.begin(), paths.size(),
		std::pair<std::string, mode_t>{});
		for(size_t i = 0; i < paths.size(); ++i) {
		queue[i] = {paths[i], modes[i]};
		}
		LOG(ERROR,
		"{}() Requeued {} metadata creates for host {} after error {}",
		__func__, paths.size(), host_id, err);
		};

		for(const auto& [host_id, data] : batches) {
		gkfs::rpc::forward_batch_create(host_id, data.first, data.second);
		const auto err = gkfs::rpc::forward_batch_create(host_id, data.first,
		data.second);
		if(err == EBUSY \|\| err == ENOTCONN \|\| err == EPROTO \|\|
		err == ETIMEDOUT) {
		requeue(host_id, data.first, data.second, err);
		}
		}
		}

		@@ -1110,7 +1131,21 @@ PreloadContext::flush_metadata_batch(uint64_t host_id) {
		if(!paths.empty()) {
		LOG(INFO, "{}() Flushing batch of size {} to host {}", __func__,
		paths.size(), host_id);
		gkfs::rpc::forward_batch_create(host_id, paths, modes);
		const auto err = gkfs::rpc::forward_batch_create(host_id, paths, modes);
		if(err == EBUSY \|\| err == ENOTCONN \|\| err == EPROTO \|\|
		err == ETIMEDOUT) {
		std::lock_guard<std::mutex> lock(metadata_batch_mutex_);
		auto& queue = metadata_batch_buffer_[host_id];
		queue.reserve(queue.size() + paths.size());
		queue.insert(queue.begin(), paths.size(),
		std::pair<std::string, mode_t>{});
		for(size_t i = 0; i < paths.size(); ++i) {
		queue[i] = {paths[i], modes[i]};
		}
		LOG(ERROR,
		"{}() Requeued {} metadata creates for host {} after error {}",
		__func__, paths.size(), host_id, err);
		}
		}
		}