From 07318aa78e680a00e1b4bb9640629ac99a186845 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Metin=20=C3=87ak=C4=B1rcal=C4=B1?= Date: Wed, 2 Sep 2026 11:47:33 +0200 Subject: [PATCH 1/6] fix(FDB-723): add barrier for data writes --- src/fdb5/remote/client/ClientConnection.cc | 35 ++++++++++++++++++---- src/fdb5/remote/client/ClientConnection.h | 3 ++ 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/src/fdb5/remote/client/ClientConnection.cc b/src/fdb5/remote/client/ClientConnection.cc index cda5816f0..1c3515fef 100644 --- a/src/fdb5/remote/client/ClientConnection.cc +++ b/src/fdb5/remote/client/ClientConnection.cc @@ -54,15 +54,19 @@ class DataWriteRequest { public: - DataWriteRequest() : client_(nullptr), msg_(Message::None), id_(0), data_(Buffer(0)) {} + DataWriteRequest() = default; DataWriteRequest(Client* client, Message msg, uint32_t id, Buffer&& data) : client_(client), msg_(msg), id_(id), data_(std::move(data)) {} - Client* client_; - Message msg_; - uint32_t id_; - Buffer data_; + /// @param barrier: no payload, signals when all writes are finished + explicit DataWriteRequest(std::shared_ptr> barrier) : barrier_{std::move(barrier)} {} + + Client* client_{nullptr}; + Message msg_{Message::None}; + uint32_t id_{0}; + Buffer data_{Buffer(0)}; + std::shared_ptr> barrier_; }; //---------------------------------------------------------------------------------------------------------------------- @@ -270,7 +274,10 @@ void ClientConnection::dataWriteThreadLoop() { ASSERT(dataWriteQueue_); while (dataWriteQueue_->pop(element) != -1) { - + if (element.barrier_) { + element.barrier_->set_value(); // unblock the waiting thread + continue; + } dataWrite(element); } @@ -285,6 +292,22 @@ void ClientConnection::dataWriteThreadLoop() { // They will be released when flush() is called. } +void ClientConnection::flushDataWrites() { + std::shared_ptr> barrier; + std::future written; + { + std::lock_guard lock(dataWriteMutex_); + if (!dataWriteQueue_) { + return; + } + barrier = std::make_shared>(); + written = barrier->get_future(); + dataWriteQueue_->emplace(barrier); + } + // block the thread! + written.get(); +} + void ClientConnection::writeControlStartupMessage(const Configuration& config) { Buffer payload(4096); diff --git a/src/fdb5/remote/client/ClientConnection.h b/src/fdb5/remote/client/ClientConnection.h index 65328eec0..ee1208f46 100644 --- a/src/fdb5/remote/client/ClientConnection.h +++ b/src/fdb5/remote/client/ClientConnection.h @@ -56,6 +56,9 @@ class ClientConnection : protected Connection { void dataWrite(Client& client, Message msg, uint32_t requestID, PayloadList payloads = {}); + // blocks until all queued data writes are finished + void flushDataWrites(); + void add(Client& client); bool remove(uint32_t clientID); From 4bc1a7cbfdb369ab36740eacfcfcb6ec24296cb3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Metin=20=C3=87ak=C4=B1rcal=C4=B1?= Date: Wed, 2 Sep 2026 11:47:59 +0200 Subject: [PATCH 2/6] fix(FDB-723): flush data writes before blocking flush --- src/fdb5/remote/client/RemoteCatalogue.cc | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/fdb5/remote/client/RemoteCatalogue.cc b/src/fdb5/remote/client/RemoteCatalogue.cc index 7674e901f..948a3986b 100644 --- a/src/fdb5/remote/client/RemoteCatalogue.cc +++ b/src/fdb5/remote/client/RemoteCatalogue.cc @@ -149,6 +149,10 @@ void RemoteCatalogue::flush(size_t archivedFields) { LOG_DEBUG_LIB(LibFdb5) << " RemoteCatalogue::flush - flushing " << numLocations_ << " fields" << std::endl; + // Ensure the (field-location) "Blob" arrive before the blocking "Flush"! + // (may deadlock server on a single connection) + connection_->flushDataWrites(); + // The flush call is blocking controlWriteCheckResponse(Message::Flush, generateRequestID(), false, sendBuf, s.position()); From 9009c0e12e82335b9cdc0c5ea71f18dd0674417e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Metin=20=C3=87ak=C4=B1rcal=C4=B1?= Date: Wed, 2 Sep 2026 14:09:12 +0200 Subject: [PATCH 3/6] fix(FDB-723): pr comment --- src/fdb5/remote/client/ClientConnection.h | 2 +- src/fdb5/remote/client/RemoteCatalogue.cc | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/fdb5/remote/client/ClientConnection.h b/src/fdb5/remote/client/ClientConnection.h index ee1208f46..de448c614 100644 --- a/src/fdb5/remote/client/ClientConnection.h +++ b/src/fdb5/remote/client/ClientConnection.h @@ -56,7 +56,7 @@ class ClientConnection : protected Connection { void dataWrite(Client& client, Message msg, uint32_t requestID, PayloadList payloads = {}); - // blocks until all queued data writes are finished + // Blocks until the data-writer thread has processed all data writes queued before this call. void flushDataWrites(); void add(Client& client); diff --git a/src/fdb5/remote/client/RemoteCatalogue.cc b/src/fdb5/remote/client/RemoteCatalogue.cc index 948a3986b..8731ceee5 100644 --- a/src/fdb5/remote/client/RemoteCatalogue.cc +++ b/src/fdb5/remote/client/RemoteCatalogue.cc @@ -149,7 +149,7 @@ void RemoteCatalogue::flush(size_t archivedFields) { LOG_DEBUG_LIB(LibFdb5) << " RemoteCatalogue::flush - flushing " << numLocations_ << " fields" << std::endl; - // Ensure the (field-location) "Blob" arrive before the blocking "Flush"! + // Ensure the (field-location) "Blob"s arrive before the blocking "Flush"! // (may deadlock server on a single connection) connection_->flushDataWrites(); From 9021d3a02762972e0bd01ac74a0bda3026fcbbdc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Metin=20=C3=87ak=C4=B1rcal=C4=B1?= Date: Wed, 2 Sep 2026 14:25:49 +0200 Subject: [PATCH 4/6] fix(FDB-723): lock write queue --- src/fdb5/remote/client/ClientConnection.cc | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/fdb5/remote/client/ClientConnection.cc b/src/fdb5/remote/client/ClientConnection.cc index 1c3515fef..720b6b49c 100644 --- a/src/fdb5/remote/client/ClientConnection.cc +++ b/src/fdb5/remote/client/ClientConnection.cc @@ -117,8 +117,11 @@ bool ClientConnection::remove(uint32_t clientID) { } ClientConnection::~ClientConnection() { - if (dataWriteQueue_) { - dataWriteQueue_->close(); + { + std::lock_guard lock(dataWriteMutex_); + if (dataWriteQueue_) { + dataWriteQueue_->close(); + } } disconnect(); From 365778ed97387dcdc250ae7823e292440de7b3d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Metin=20=C3=87ak=C4=B1rcal=C4=B1?= Date: Wed, 2 Sep 2026 14:26:43 +0200 Subject: [PATCH 5/6] fix(FDB-723): don't throw in thread, add fail data barriers --- src/fdb5/remote/client/ClientConnection.cc | 50 ++++++++++++++++------ src/fdb5/remote/client/ClientConnection.h | 6 +++ 2 files changed, 44 insertions(+), 12 deletions(-) diff --git a/src/fdb5/remote/client/ClientConnection.cc b/src/fdb5/remote/client/ClientConnection.cc index 720b6b49c..bac045a3d 100644 --- a/src/fdb5/remote/client/ClientConnection.cc +++ b/src/fdb5/remote/client/ClientConnection.cc @@ -23,6 +23,7 @@ #include +#include #include #include #include @@ -273,44 +274,69 @@ void ClientConnection::dataWriteThreadLoop() { Timer timer; DataWriteRequest element; + auto* queue = dataWriteQueue_.get(); + ASSERT(queue); + try { - ASSERT(dataWriteQueue_); - while (dataWriteQueue_->pop(element) != -1) { + while (queue->pop(element) != -1) { if (element.barrier_) { - element.barrier_->set_value(); // unblock the waiting thread + std::lock_guard lock(dataWriteMutex_); + element.barrier_->set_value(); + dataBarriers_.erase(std::remove(dataBarriers_.begin(), dataBarriers_.end(), element.barrier_), + dataBarriers_.end()); continue; } dataWrite(element); } + std::lock_guard lock(dataWriteMutex_); dataWriteQueue_.reset(); } catch (...) { - dataWriteQueue_->interrupt(std::current_exception()); - throw; + // we must not rethrow in thread (that would std::terminate) + // propagate and unblock any waiting flushDataWrites() + queue->interrupt(std::current_exception()); + failDataBarriers(std::current_exception()); } - - // We are inside an async, so don't need to worry about exceptions escaping. - // They will be released when flush() is called. } void ClientConnection::flushDataWrites() { - std::shared_ptr> barrier; std::future written; { std::lock_guard lock(dataWriteMutex_); if (!dataWriteQueue_) { return; } - barrier = std::make_shared>(); + auto barrier = std::make_shared>(); written = barrier->get_future(); - dataWriteQueue_->emplace(barrier); + dataBarriers_.push_back(barrier); + try { + dataWriteQueue_->emplace(barrier); + } + catch (...) { + LOG_DEBUG_LIB(LibFdb5) << "flushDataWrites - failed to enqueue barrier!" << std::endl; + dataBarriers_.pop_back(); + throw; + } } - // block the thread! + // block until the writer reaches the barrier (or the connection fails) written.get(); } +void ClientConnection::failDataBarriers(const std::exception_ptr& eptr) { + std::lock_guard lock(dataWriteMutex_); + for (auto& barrier : dataBarriers_) { + try { + barrier->set_exception(eptr); + } + catch (...) { + Log::warning() << "failDataBarriers - barrier already satisfied." << std::endl; + } + } + dataBarriers_.clear(); +} + void ClientConnection::writeControlStartupMessage(const Configuration& config) { Buffer payload(4096); diff --git a/src/fdb5/remote/client/ClientConnection.h b/src/fdb5/remote/client/ClientConnection.h index de448c614..fb92efdc7 100644 --- a/src/fdb5/remote/client/ClientConnection.h +++ b/src/fdb5/remote/client/ClientConnection.h @@ -28,6 +28,7 @@ #include #include #include +#include namespace eckit { @@ -96,6 +97,9 @@ class ClientConnection : protected Connection { // do not hang forever once the connection is known to be dead. void failPendingRequests(const std::exception_ptr& eptr); + // fail any waiting flush data barriers + void failDataBarriers(const std::exception_ptr& eptr); + void dataWriteThreadLoop(); void closeConnection(); @@ -134,6 +138,8 @@ class ClientConnection : protected Connection { std::mutex dataWriteMutex_; std::unique_ptr> dataWriteQueue_; std::thread dataWriteThread_; + + std::vector>> dataBarriers_; }; //---------------------------------------------------------------------------------------------------------------------- From 7eccbe3b75361d5829d093275b6b7691b679b535 Mon Sep 17 00:00:00 2001 From: Emanuele <71709533+danovaro@users.noreply.github.com> Date: Thu, 3 Sep 2026 21:39:01 +0200 Subject: [PATCH 6/6] FDB-723 - improved comment wording Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/fdb5/remote/client/ClientConnection.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/fdb5/remote/client/ClientConnection.cc b/src/fdb5/remote/client/ClientConnection.cc index 1dd5f4173..28601e226 100644 --- a/src/fdb5/remote/client/ClientConnection.cc +++ b/src/fdb5/remote/client/ClientConnection.cc @@ -60,7 +60,7 @@ class DataWriteRequest { DataWriteRequest(Client* client, Message msg, uint32_t id, Buffer&& data) : client_(client), msg_(msg), id_(id), data_(std::move(data)) {} - /// @param barrier: no payload, signals when all writes are finished + /// @param barrier: no payload, signals when all queued writes before this barrier have been processed explicit DataWriteRequest(std::shared_ptr> barrier) : barrier_{std::move(barrier)} {} Client* client_{nullptr};