From 420885c6627b15a2cd0658701a58819a296203a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20Fu=CC=88ller?= <5454068+florianfueller@users.noreply.github.com> Date: Thu, 22 Oct 2026 22:59:54 +0200 Subject: [PATCH 1/2] Fix data race in TimeDaemon shared memory handler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace the seqlock with reader-counted read-copy-update (RCU) slots closes #272 Signed-off-by: Florian Füller <5454068+florianfueller@users.noreply.github.com> --- .../src/ipc/core/shared_memory_handler.h | 127 ++++++++++++------ 1 file changed, 87 insertions(+), 40 deletions(-) diff --git a/score/time_daemon/src/ipc/core/shared_memory_handler.h b/score/time_daemon/src/ipc/core/shared_memory_handler.h index 2ba9a461e..b1a568061 100644 --- a/score/time_daemon/src/ipc/core/shared_memory_handler.h +++ b/score/time_daemon/src/ipc/core/shared_memory_handler.h @@ -18,12 +18,14 @@ #include "score/memory/shared/shared_memory_resource.h" #include "score/time_daemon/src/common/logging_contexts.h" +#include #include +#include #include #include #include #include -#include +#include #include namespace score @@ -34,6 +36,13 @@ namespace td /// /// \brief The class implements shared memory handling /// +/// \details One writer sends the latest DataType to any number of readers, also in other processes. It uses +/// read-copy-update (RCU) over a fixed number of slots: https://lwn.net/Articles/262464/ +/// - Send() writes into a slot that is not published and has no readers, then publishes it. +/// - Receive() adds 1 to the slot's reader count, checks that the slot is still published, copies the data +/// and removes 1 again. +/// Only one thread at a time, across all processes, may call Send(). +/// template class SharedMemoryHandler { @@ -73,16 +82,25 @@ class SharedMemoryHandler SharedMemoryHandler& operator=(SharedMemoryHandler&&) = delete; private: - static constexpr bool NoWriteInProgress(std::uint64_t seq) + /// \brief Assumed CPU cache-line size in bytes. + /// Used to keep data that different cores write on separate cache lines to reduce false sharing. + static constexpr std::size_t kCacheLineSize{64U}; + + /// \brief Number of copies of the data in shared memory. Send() needs a slot that is not published and has no + /// readers, so more slots let more readers be stuck in Receive() before Send() has to drop data. + static constexpr std::uint32_t kNumberOfSlots{10U}; + + /// \brief One copy of the data. Starts on a new cache line. + struct alignas(kCacheLineSize) Slot { - return (seq & 1U) == 0U; - } + DataType data_{}; + }; /// - /// \brief Common type to store mutex and specified data + /// \brief Everything in shared memory: the slots and the data that manages them /// \tparam DataType shall be trivially copyable data /// - struct SharedData + struct alignas(kCacheLineSize) SharedData { static_assert(std::is_trivially_copyable_v, "DataType must be trivially copyable to be stored in shared memory!"); @@ -90,15 +108,22 @@ class SharedMemoryHandler "DataType must be default constructible (required by DataType{} usage)!"); static_assert(std::is_standard_layout_v, "DataType should be standard-layout for robust shared memory/IPC usage!"); + static_assert(std::atomic::is_always_lock_free, + "Atomics shared between processes must be lock-free!"); - /// \brief seq_ seqlock counter: odd while a write is in progress, even otherwise. - std::atomic seq_{0U}; + /// \brief published_slot_ index of the slot that Receive() reads. Only Send() writes it. + std::atomic published_slot_{0U}; - /// \brief data_ specific data placed in share dmemory region. Note: It has to be trivially copyable with - // trivial simple types that are not allocated on heap! - DataType data_{}; + /// \brief readers_per_slot_ readers per slot, on its own cache line. Readers write nothing else. + alignas(kCacheLineSize) std::array, kNumberOfSlots> readers_per_slot_{}; + + /// \brief slots_ copies of the data. + std::array slots_{}; }; + /// \brief Bytes to reserve. Aligning SharedData can add up to alignof(SharedData) bytes of padding. + static constexpr std::size_t kSharedMemorySize{sizeof(SharedData) + alignof(SharedData)}; + const std::string shared_memory_path_; std::shared_ptr shared_memory_resource_; SharedMemoryHandler::SharedData* shared_memory_data_; @@ -116,7 +141,7 @@ bool SharedMemoryHandler::Init() [this](std::shared_ptr memory_resource) { shared_memory_data_ = memory_resource->construct(); }, - sizeof(SharedData), + kSharedMemorySize, {{permissions}, {}}); if (shared_memory_resource_ == nullptr) @@ -128,13 +153,23 @@ bool SharedMemoryHandler::Init() if ((shared_memory_data_ == nullptr) && (shared_memory_resource_ != nullptr)) { - // Shared memory was opened (not created) and now we need to map our struct there - // cast from void to T* is done by design as long as shared memory is agnostic to the type - // but clients only know, what is stored there. Also see the score::memory::shared::ManagedMemoryResource - // design for details. - shared_memory_data_ = static_cast(shared_memory_resource_->getUsableBaseAddress()); - score::mw::log::LogInfo(kIpcHandlerContext) - << "Shared memory object was found. Mapped data in it: " << shared_memory_data_; + // Shared memory was opened (not created) and now we need to map our struct there. + // The creator placed SharedData first, aligned to alignof(SharedData). Use the same alignment to find it. + void* address = shared_memory_resource_->getUsableBaseAddress(); + std::size_t space = shared_memory_resource_->GetUserAllocatedBytes(); + shared_memory_data_ = + static_cast(std::align(alignof(SharedData), sizeof(SharedData), address, space)); + + if (shared_memory_data_ == nullptr) + { + score::mw::log::LogError(kIpcHandlerContext) + << "Shared memory object " << shared_memory_path_ << " is too small for the expected data layout"; + } + else + { + score::mw::log::LogInfo(kIpcHandlerContext) + << "Shared memory object was found. Mapped data in it: " << shared_memory_data_; + } } return ((shared_memory_data_ != nullptr) && (shared_memory_resource_ != nullptr)); @@ -143,33 +178,37 @@ bool SharedMemoryHandler::Init() template std::optional SharedMemoryHandler::Receive() const { + std::optional read_data{}; + if (shared_memory_data_ != nullptr) { - DataType read_data{}; - for (std::uint8_t retry_cnt = 0U; retry_cnt < max_number_of_read_retries_; ++retry_cnt) { - const auto seq_before_read = shared_memory_data_->seq_.load(std::memory_order_acquire); - - if (NoWriteInProgress(seq_before_read)) + // The modulo keeps the index inside the array. + const auto slot = shared_memory_data_->published_slot_.load(std::memory_order_relaxed) % kNumberOfSlots; + auto& slot_readers = shared_memory_data_->readers_per_slot_[slot]; + + // Add 1 to the reader count, then check that the slot is still published. If it is, Send() does not + // write into it until we remove 1 again. + std::ignore = slot_readers.fetch_add(1U, std::memory_order_seq_cst); + if (shared_memory_data_->published_slot_.load(std::memory_order_seq_cst) == slot) { - // Copy the payload - read_data = shared_memory_data_->data_; - - // Unchanged counter: no write started or completed during the copy - if (shared_memory_data_->seq_.load(std::memory_order_relaxed) == seq_before_read) - { - return read_data; - } + read_data.emplace(shared_memory_data_->slots_[slot].data_); + + // Remove 1 from the reader count. Release makes sure the copy is done before Send() writes here again. + std::ignore = slot_readers.fetch_sub(1U, std::memory_order_release); + return read_data; } - std::this_thread::yield(); + + // Send() published another slot in the meantime. Remove 1 from the reader count and try the new slot. + std::ignore = slot_readers.fetch_sub(1U, std::memory_order_relaxed); } score::mw::log::LogError(kIpcHandlerContext) << "Read failed for number of retries: " << max_number_of_read_retries_; } - return std::nullopt; + return read_data; } template @@ -177,14 +216,22 @@ void SharedMemoryHandler::Send(const DataType& data) { if (shared_memory_data_ != nullptr) { - // Single writer: counter becomes odd to signal write in progress - const auto seq = shared_memory_data_->seq_.load(std::memory_order_relaxed); - shared_memory_data_->seq_.store(seq + 1U, std::memory_order_relaxed); + // Only Send() writes published_slot_, so this reads the value we wrote last time + const auto published_slot = shared_memory_data_->published_slot_.load(std::memory_order_relaxed); - shared_memory_data_->data_ = data; + // Start with the slot after the published one, which is usually the oldest. Never use the published slot. + for (std::uint32_t offset = 1U; offset < kNumberOfSlots; ++offset) + { + const auto candidate = (published_slot + offset) % kNumberOfSlots; + if (shared_memory_data_->readers_per_slot_[candidate].load(std::memory_order_seq_cst) == 0U) + { + shared_memory_data_->slots_[candidate].data_ = data; + shared_memory_data_->published_slot_.store(candidate, std::memory_order_seq_cst); + return; + } + } - // Even again: publishes the payload to readers - shared_memory_data_->seq_.store(seq + 2U, std::memory_order_release); + score::mw::log::LogError(kIpcHandlerContext) << "All slots are being read, data was not published"; } } From 790db5114367d8e441c6edd2007a489025f4bf9e Mon Sep 17 00:00:00 2001 From: Florian Fueller <5454068+florianfueller@users.noreply.github.com> Date: Thu, 29 Oct 2026 10:30:41 +0100 Subject: [PATCH 2/2] refactor: Replace shared-memory reader counts with per-slot seqlocks --- .../src/ipc/core/shared_memory_handler.h | 106 ++++++++---------- 1 file changed, 47 insertions(+), 59 deletions(-) diff --git a/score/time_daemon/src/ipc/core/shared_memory_handler.h b/score/time_daemon/src/ipc/core/shared_memory_handler.h index b1a568061..6b51d752f 100644 --- a/score/time_daemon/src/ipc/core/shared_memory_handler.h +++ b/score/time_daemon/src/ipc/core/shared_memory_handler.h @@ -25,7 +25,6 @@ #include #include #include -#include #include namespace score @@ -36,12 +35,8 @@ namespace td /// /// \brief The class implements shared memory handling /// -/// \details One writer sends the latest DataType to any number of readers, also in other processes. It uses -/// read-copy-update (RCU) over a fixed number of slots: https://lwn.net/Articles/262464/ -/// - Send() writes into a slot that is not published and has no readers, then publishes it. -/// - Receive() adds 1 to the slot's reader count, checks that the slot is still published, copies the data -/// and removes 1 again. -/// Only one thread at a time, across all processes, may call Send(). +/// \details Only one thread at a time, across all processes, may call Send(). Slots use per-slot seqlocks. +/// Payload accesses are not atomic yet; overlapping slot reuse and reads are not C++ data-race-free. /// template class SharedMemoryHandler @@ -62,12 +57,12 @@ class SharedMemoryHandler bool Init(); /// - /// \brief Safely read data from shared memory. + /// \brief Read a snapshot from shared memory, retrying if the slot changes. /// std::optional Receive() const; /// - /// \brief Safely write data to shared memory + /// \brief Write data to the next shared memory slot without waiting for readers. /// void Send(const DataType& data); @@ -82,25 +77,19 @@ class SharedMemoryHandler SharedMemoryHandler& operator=(SharedMemoryHandler&&) = delete; private: - /// \brief Assumed CPU cache-line size in bytes. - /// Used to keep data that different cores write on separate cache lines to reduce false sharing. - static constexpr std::size_t kCacheLineSize{64U}; + static constexpr bool NoWriteInProgress(std::uint64_t seq) + { + return (seq & 1U) == 0U; + } - /// \brief Number of copies of the data in shared memory. Send() needs a slot that is not published and has no - /// readers, so more slots let more readers be stuck in Receive() before Send() has to drop data. + static constexpr std::size_t kCacheLineSize{64U}; static constexpr std::uint32_t kNumberOfSlots{10U}; - /// \brief One copy of the data. Starts on a new cache line. - struct alignas(kCacheLineSize) Slot - { - DataType data_{}; - }; - /// - /// \brief Everything in shared memory: the slots and the data that manages them + /// \brief Common type to store seqlock and specified data /// \tparam DataType shall be trivially copyable data /// - struct alignas(kCacheLineSize) SharedData + struct alignas(kCacheLineSize) Slot { static_assert(std::is_trivially_copyable_v, "DataType must be trivially copyable to be stored in shared memory!"); @@ -111,13 +100,17 @@ class SharedMemoryHandler static_assert(std::atomic::is_always_lock_free, "Atomics shared between processes must be lock-free!"); - /// \brief published_slot_ index of the slot that Receive() reads. Only Send() writes it. - std::atomic published_slot_{0U}; + /// \brief seq_ seqlock counter: odd while a write is in progress, even otherwise. + std::atomic seq_{0U}; - /// \brief readers_per_slot_ readers per slot, on its own cache line. Readers write nothing else. - alignas(kCacheLineSize) std::array, kNumberOfSlots> readers_per_slot_{}; + /// \brief data_ specific data placed in share dmemory region. Note: It has to be trivially copyable with + // trivial simple types that are not allocated on heap! + DataType data_{}; + }; - /// \brief slots_ copies of the data. + struct alignas(kCacheLineSize) SharedData + { + std::atomic published_slot_{0U}; std::array slots_{}; }; @@ -178,37 +171,36 @@ bool SharedMemoryHandler::Init() template std::optional SharedMemoryHandler::Receive() const { - std::optional read_data{}; - if (shared_memory_data_ != nullptr) { + DataType read_data{}; + for (std::uint8_t retry_cnt = 0U; retry_cnt < max_number_of_read_retries_; ++retry_cnt) { - // The modulo keeps the index inside the array. - const auto slot = shared_memory_data_->published_slot_.load(std::memory_order_relaxed) % kNumberOfSlots; - auto& slot_readers = shared_memory_data_->readers_per_slot_[slot]; - - // Add 1 to the reader count, then check that the slot is still published. If it is, Send() does not - // write into it until we remove 1 again. - std::ignore = slot_readers.fetch_add(1U, std::memory_order_seq_cst); - if (shared_memory_data_->published_slot_.load(std::memory_order_seq_cst) == slot) - { - read_data.emplace(shared_memory_data_->slots_[slot].data_); + const auto slot_index = + shared_memory_data_->published_slot_.load(std::memory_order_seq_cst) % kNumberOfSlots; + const auto& slot = shared_memory_data_->slots_[slot_index]; + const auto seq_before_read = slot.seq_.load(std::memory_order_seq_cst); - // Remove 1 from the reader count. Release makes sure the copy is done before Send() writes here again. - std::ignore = slot_readers.fetch_sub(1U, std::memory_order_release); - return read_data; + if (NoWriteInProgress(seq_before_read)) + { + // Copy the payload + read_data = slot.data_; + + // Unchanged counter: no write started or completed during the copy + if ((slot.seq_.load(std::memory_order_seq_cst) == seq_before_read) && + (shared_memory_data_->published_slot_.load(std::memory_order_seq_cst) == slot_index)) + { + return read_data; + } } - - // Send() published another slot in the meantime. Remove 1 from the reader count and try the new slot. - std::ignore = slot_readers.fetch_sub(1U, std::memory_order_relaxed); } score::mw::log::LogError(kIpcHandlerContext) << "Read failed for number of retries: " << max_number_of_read_retries_; } - return read_data; + return std::nullopt; } template @@ -216,22 +208,18 @@ void SharedMemoryHandler::Send(const DataType& data) { if (shared_memory_data_ != nullptr) { - // Only Send() writes published_slot_, so this reads the value we wrote last time const auto published_slot = shared_memory_data_->published_slot_.load(std::memory_order_relaxed); + const auto next_slot = (published_slot + 1U) % kNumberOfSlots; + auto& slot = shared_memory_data_->slots_[next_slot]; - // Start with the slot after the published one, which is usually the oldest. Never use the published slot. - for (std::uint32_t offset = 1U; offset < kNumberOfSlots; ++offset) - { - const auto candidate = (published_slot + offset) % kNumberOfSlots; - if (shared_memory_data_->readers_per_slot_[candidate].load(std::memory_order_seq_cst) == 0U) - { - shared_memory_data_->slots_[candidate].data_ = data; - shared_memory_data_->published_slot_.store(candidate, std::memory_order_seq_cst); - return; - } - } + // Single writer: counter becomes odd to signal write in progress + const auto seq = slot.seq_.fetch_add(1U, std::memory_order_seq_cst); + + slot.data_ = data; - score::mw::log::LogError(kIpcHandlerContext) << "All slots are being read, data was not published"; + // Even again: publishes the payload to readers + slot.seq_.store(seq + 2U, std::memory_order_seq_cst); + shared_memory_data_->published_slot_.store(next_slot, std::memory_order_seq_cst); } }