This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 8102b17d Support seqlock and use it for TaskGroup CPU time stat (#3541)
8102b17d is described below
commit 8102b17d59f39166f4c5b6e2fe98c2319eca9689
Author: Bright Chen <[email protected]>
AuthorDate: Wed Sep 16 22:37:12 2026 +0800
Support seqlock and use it for TaskGroup CPU time stat (#3541)
---
src/bthread/processor.h | 28 +--
src/bthread/task_group.cpp | 105 +---------
src/bthread/task_group.h | 117 +++++------
src/{bthread => butil}/processor.h | 39 +---
src/butil/synchronization/seqlock.h | 302 ++++++++++++++++++++++++++++
test/BUILD.bazel | 3 +-
test/CMakeLists.txt | 1 +
test/Makefile | 7 +-
test/seqlock_unittest.cpp | 391 ++++++++++++++++++++++++++++++++++++
9 files changed, 768 insertions(+), 225 deletions(-)
diff --git a/src/bthread/processor.h b/src/bthread/processor.h
index 06e75641..74be9e25 100644
--- a/src/bthread/processor.h
+++ b/src/bthread/processor.h
@@ -22,33 +22,7 @@
#ifndef BTHREAD_PROCESSOR_H
#define BTHREAD_PROCESSOR_H
-#include "butil/build_config.h"
-
-// Pause instruction to prevent excess processor bus usage, only works in GCC
-# ifndef cpu_relax
-#if defined(ARCH_CPU_ARM_FAMILY)
-# define cpu_relax() asm volatile("yield\n": : :"memory")
-#elif defined(ARCH_CPU_RISCV_FAMILY)
-// Use the pause hint (Zihintpause extension). Encoding 0x0100000F
-// (fence 0, 1) is a HINT on all RISC-V implementations: it never traps
-// and is ignored on CPUs without Zihintpause. On CPUs with Zihintpause
-// it provides a multi-cycle stall hint that reduces power and improves
-// resource fairness during spin-wait loops. Matches the Linux kernel's
-// RISC-V cpu_relax() behavior. .word is used instead of .insn or the
-// pause mnemonic for maximum assembler compatibility.
-# define cpu_relax() asm volatile(".word 0x0100000f\n": : :"memory")
-#elif defined(ARCH_CPU_LOONGARCH64_FAMILY)
-# define cpu_relax() asm volatile("nop\n": : :"memory");
-#else
-# define cpu_relax() asm volatile("pause\n": : :"memory")
-#endif
-# endif
-
-// Compile read-write barrier
-# ifndef barrier
-# define barrier() asm volatile("": : :"memory")
-# endif
-
+#include "butil/processor.h"
# define BT_LOOP_WHEN(expr, num_spins) \
do { \
diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp
index 5b02f8c9..0503fa4f 100644
--- a/src/bthread/task_group.cpp
+++ b/src/bthread/task_group.cpp
@@ -84,103 +84,6 @@ BAIDU_VOLATILE_THREAD_LOCAL(void*, tls_unique_user_ptr,
nullptr);
const TaskStatistics EMPTY_STAT = { 0, 0, 0 };
-AtomicInteger128::Value AtomicInteger128::load() const {
-#ifdef __x86_64__
- (void)_mutex;
- (void)_seq;
- __m128i value = _mm_load_si128(reinterpret_cast<const __m128i*>(&_value));
- return {value[0], value[1]};
-#elif defined(__ARM_NEON)
- (void)_mutex;
- (void)_seq;
- int64x2_t value = vld1q_s64(reinterpret_cast<const int64_t*>(&_value));
- return {value[0], value[1]};
-#elif defined(__riscv) && __riscv_xlen == 64
- (void)_mutex;
- // RISC-V: Seqlock-based atomic 128-bit load.
- int64_t v1, v2;
- uint64_t seq0, seq1;
- do {
- __asm__ volatile(
- "ld %0, %1\n\t"
- : "=r"(seq0)
- : "m"(_seq)
- : "memory"
- );
- if (seq0 & 1) continue;
- __asm__ volatile("fence r, rw\n\t" ::: "memory");
- __asm__ volatile(
- "ld %0, %2\n\t"
- "ld %1, %3\n\t"
- : "=r"(v1), "=r"(v2)
- : "m"(_value.v1), "m"(_value.v2)
- : "memory"
- );
- __asm__ volatile("fence r, rw\n\t" ::: "memory");
- __asm__ volatile(
- "ld %0, %1\n\t"
- : "=r"(seq1)
- : "m"(_seq)
- : "memory"
- );
- } while (seq0 != seq1);
- return {v1, v2};
-#else
- BAIDU_SCOPED_LOCK(const_cast<FastPthreadMutex&>(_mutex));
- return _value;
-#endif
-}
-
-void AtomicInteger128::store(Value value) {
-#ifdef __x86_64__
- (void)_seq;
- __m128i v = _mm_load_si128(reinterpret_cast<__m128i*>(&value));
- _mm_store_si128(reinterpret_cast<__m128i*>(&_value), v);
-#elif defined(__ARM_NEON)
- (void)_seq;
- int64x2_t v = vld1q_s64(reinterpret_cast<int64_t*>(&value));
- vst1q_s64(reinterpret_cast<int64_t*>(&_value), v);
-#elif defined(__riscv) && __riscv_xlen == 64
- (void)_mutex;
- // RISC-V: Seqlock-based atomic 128-bit store.
- uint64_t old_seq;
- __asm__ volatile(
- "ld %0, %1\n\t"
- : "=r"(old_seq)
- : "m"(_seq)
- : "memory"
- );
- uint64_t new_seq = old_seq + 1;
- __asm__ volatile(
- "fence w, w\n\t"
- "sd %1, %0\n\t"
- : "=m"(_seq)
- : "r"(new_seq)
- : "memory"
- );
- __asm__ volatile("fence w, w\n\t" ::: "memory");
- __asm__ volatile(
- "sd %2, %0\n\t"
- "sd %3, %1\n\t"
- : "=m"(_value.v1), "=m"(_value.v2)
- : "r"(value.v1), "r"(value.v2)
- : "memory"
- );
- __asm__ volatile("fence w, w\n\t" ::: "memory");
- new_seq++;
- __asm__ volatile(
- "sd %1, %0\n\t"
- : "=m"(_seq)
- : "r"(new_seq)
- : "memory"
- );
-#else
- BAIDU_SCOPED_LOCK(const_cast<FastPthreadMutex&>(_mutex));
- _value = value;
-#endif
-}
-
-
int TaskGroup::get_attr(bthread_t tid, bthread_attr_t* out) {
TaskMeta* const m = address_meta(tid);
if (m != nullptr) {
@@ -249,7 +152,9 @@ static double get_cumulated_cputime_from_this(void* arg) {
int64_t TaskGroup::cumulated_cputime_ns() const {
CPUTimeStat cpu_time_stat = _cpu_time_stat.load();
- // Add the elapsed time of running bthread.
+ // Add elapsed time only for a running non-main task. cpuwide_time_ns()
+ // advances while the worker is parked, so including the main task would
+ // count idle waiting as worker usage.
int64_t cumulated_cputime_ns = cpu_time_stat.cumulated_cputime_ns();
if (!cpu_time_stat.is_main_task()) {
cumulated_cputime_ns += butil::cpuwide_time_ns() -
cpu_time_stat.last_run_ns();
@@ -286,7 +191,7 @@ void TaskGroup::run_main_task() {
}
// Don't forget to add elapse of last wait_task.
current_task()->stat.cputime_ns +=
- butil::cpuwide_time_ns() - _cpu_time_stat.load_unsafe().last_run_ns();
+ butil::cpuwide_time_ns() -
_cpu_time_stat.load_for_writer().last_run_ns();
}
TaskGroup::TaskGroup(TaskControl* c)
@@ -840,7 +745,7 @@ void TaskGroup::sched_to(TaskGroup** pg, TaskMeta*
next_meta) {
TaskMeta* const cur_meta = g->_cur_meta;
int64_t now = butil::cpuwide_time_ns();
- CPUTimeStat cpu_time_stat = g->_cpu_time_stat.load_unsafe();
+ CPUTimeStat cpu_time_stat = g->_cpu_time_stat.load_for_writer();
int64_t elp_ns = now - cpu_time_stat.last_run_ns();
cur_meta->stat.cputime_ns += elp_ns;
// Update cpu_time_stat.
diff --git a/src/bthread/task_group.h b/src/bthread/task_group.h
index f48e02f5..bbd76ae7 100644
--- a/src/bthread/task_group.h
+++ b/src/bthread/task_group.h
@@ -22,12 +22,13 @@
#ifndef BTHREAD_TASK_GROUP_H
#define BTHREAD_TASK_GROUP_H
-#include "butil/time.h" // cpuwide_time_ns
+#include "butil/time.h"
+#include "butil/synchronization/seqlock.h"
#include "bthread/task_control.h"
-#include "bthread/task_meta.h" // bthread_t, TaskMeta
-#include "bthread/work_stealing_queue.h" // WorkStealingQueue
-#include "bthread/remote_task_queue.h" // RemoteTaskQueue
-#include "butil/resource_pool.h" // ResourceId
+#include "bthread/task_meta.h"
+#include "bthread/work_stealing_queue.h"
+#include "bthread/remote_task_queue.h"
+#include "butil/resource_pool.h"
#include "bthread/parking_lot.h"
#include "bthread/prime_offset.h"
@@ -48,37 +49,6 @@ private:
void* _value;
};
-// Refer to https://rigtorp.se/isatomic/, On the modern CPU microarchitectures
-// (Skylake and Zen 2) AVX/AVX2 128b/256b aligned loads and stores are atomic
-// even though Intel and AMD officially doesn’t guarantee this.
-// On X86, SSE instructions can ensure atomic loads and stores.
-// Starting from Armv8.4-A, neon can ensure atomic loads and stores.
-// Otherwise, use mutex to guarantee atomicity.
-class AtomicInteger128 {
-public:
- struct BAIDU_CACHELINE_ALIGNMENT Value {
- int64_t v1;
- int64_t v2;
- };
-
- AtomicInteger128() = default;
- explicit AtomicInteger128(Value value) : _value(value) {}
-
- Value load() const;
- Value load_unsafe() const {
- return _value;
- }
-
- void store(Value value);
-
-private:
- Value _value{};
- // Used to protect `_cpu_time_stat' on architectures without lock-free
128-bit atomics.
- FastPthreadMutex _mutex;
- // Sequence counter for RISC-V seqlock implementation.
- uint64_t _seq = 0;
-};
-
// Thread-local group of tasks.
// Notice that most methods involving context switching are static otherwise
// pointer `this' may change after wakeup. The **pg parameters in following
@@ -237,66 +207,83 @@ friend class TaskControl;
// Last scheduling time, task type and cumulated CPU time.
class CPUTimeStat {
- static constexpr int64_t LAST_SCHEDULING_TIME_MASK =
0x7FFFFFFFFFFFFFFFLL;
- static constexpr int64_t TASK_TYPE_MASK = 0x8000000000000000LL;
public:
- CPUTimeStat() : _last_run_ns_and_type(0), _cumulated_cputime_ns(0) {}
- CPUTimeStat(AtomicInteger128::Value value)
- : _last_run_ns_and_type(value.v1), _cumulated_cputime_ns(value.v2)
{}
-
- // Convert to AtomicInteger128::Value for atomic operations.
- explicit operator AtomicInteger128::Value() const {
- return {_last_run_ns_and_type, _cumulated_cputime_ns};
+ CPUTimeStat() : CPUTimeStat(0, 0, false) {}
+
+ CPUTimeStat(int64_t last_run_ns, int64_t cumulated_cputime_ns, bool
main_task)
+ : _cumulated_cputime_ns(cumulated_cputime_ns)
+ , _last_run_ns(last_run_ns)
+ , _main_task(main_task) {}
+
+ CPUTimeStat(const CPUTimeStat& other)
+ : CPUTimeStat(other.last_run_ns(),
+ other.cumulated_cputime_ns(),
+ other.is_main_task()) {}
+
+ CPUTimeStat& operator=(const CPUTimeStat& other) {
+ if (this != &other) {
+ _last_run_ns.store(other.last_run_ns(),
+ butil::memory_order_relaxed);
+ _cumulated_cputime_ns.store(other.cumulated_cputime_ns(),
+ butil::memory_order_relaxed);
+ _main_task.store(other.is_main_task(),
butil::memory_order_relaxed);
+ }
+ return *this;
}
void set_last_run_ns(int64_t last_run_ns, bool main_task) {
- _last_run_ns_and_type = (last_run_ns & LAST_SCHEDULING_TIME_MASK) |
- (static_cast<int64_t>(main_task) << 63);
+ _last_run_ns.store(last_run_ns, butil::memory_order_relaxed);
+ _main_task.store(main_task, butil::memory_order_relaxed);
}
int64_t last_run_ns() const {
- return _last_run_ns_and_type & LAST_SCHEDULING_TIME_MASK;
- }
- int64_t last_run_ns_and_type() const {
- return _last_run_ns_and_type;
+ return _last_run_ns.load(butil::memory_order_relaxed);
}
bool is_main_task() const {
- return _last_run_ns_and_type & TASK_TYPE_MASK;
+ return _main_task.load(butil::memory_order_relaxed);
}
void add_cumulated_cputime_ns(int64_t cputime_ns, bool main_task) {
if (main_task) {
return;
}
- _cumulated_cputime_ns += cputime_ns;
+
+ _cumulated_cputime_ns.store(cumulated_cputime_ns() + cputime_ns,
+ butil::memory_order_relaxed);
}
int64_t cumulated_cputime_ns() const {
- return _cumulated_cputime_ns;
+ return _cumulated_cputime_ns.load(butil::memory_order_relaxed);
}
private:
- // The higher bit for task type, main task is 1, otherwise 0.
- // Lowest 63 bits for last scheduling time.
- int64_t _last_run_ns_and_type;
- // Cumulated CPU time in nanoseconds.
- int64_t _cumulated_cputime_ns;
+ // Cumulated non-main-task elapsed time in nanoseconds.
+ butil::atomic<int64_t> _cumulated_cputime_ns;
+ butil::atomic<int64_t> _last_run_ns;
+ butil::atomic<bool> _main_task;
};
class AtomicCPUTimeStat {
public:
CPUTimeStat load() const {
- return _cpu_time_stat.load();
+ return _seqlock.load([&]() -> CPUTimeStat {
+ return _stat;
+ });
}
- CPUTimeStat load_unsafe() const {
- return _cpu_time_stat.load_unsafe();
+ // For the owning writer only, with no concurrent writes. Copies fields
+ // with relaxed atomic loads but skips sequence validation.
+ CPUTimeStat load_for_writer() const {
+ return _stat;
}
- void store(CPUTimeStat cpu_time_stat) {
- _cpu_time_stat.store(AtomicInteger128::Value(cpu_time_stat));
+ void store(const CPUTimeStat& stat) {
+ _seqlock.store([this, &stat]() {
+ _stat = stat;
+ });
}
private:
- AtomicInteger128 _cpu_time_stat;
+ CPUTimeStat _stat;
+ butil::Seqlock<> _seqlock;
};
// You shall use TaskControl::create_group to create new instance.
diff --git a/src/bthread/processor.h b/src/butil/processor.h
similarity index 55%
copy from src/bthread/processor.h
copy to src/butil/processor.h
index 06e75641..9219d316 100644
--- a/src/bthread/processor.h
+++ b/src/butil/processor.h
@@ -15,19 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-// bthread - An M:N threading library to make applications more concurrent.
-
-// Date: Fri Dec 5 13:40:57 CST 2014
-
-#ifndef BTHREAD_PROCESSOR_H
-#define BTHREAD_PROCESSOR_H
+#ifndef BUTIL_PROCESSOR_H
+#define BUTIL_PROCESSOR_H
#include "butil/build_config.h"
// Pause instruction to prevent excess processor bus usage, only works in GCC
-# ifndef cpu_relax
+#ifndef cpu_relax
#if defined(ARCH_CPU_ARM_FAMILY)
-# define cpu_relax() asm volatile("yield\n": : :"memory")
+#define cpu_relax() asm volatile("yield\n": : :"memory")
#elif defined(ARCH_CPU_RISCV_FAMILY)
// Use the pause hint (Zihintpause extension). Encoding 0x0100000F
// (fence 0, 1) is a HINT on all RISC-V implementations: it never traps
@@ -38,30 +34,15 @@
// pause mnemonic for maximum assembler compatibility.
# define cpu_relax() asm volatile(".word 0x0100000f\n": : :"memory")
#elif defined(ARCH_CPU_LOONGARCH64_FAMILY)
-# define cpu_relax() asm volatile("nop\n": : :"memory");
+# define cpu_relax() asm volatile("nop\n": : :"memory")
#else
# define cpu_relax() asm volatile("pause\n": : :"memory")
#endif
-# endif
+#endif // cpu_relax
// Compile read-write barrier
-# ifndef barrier
-# define barrier() asm volatile("": : :"memory")
-# endif
-
-
-# define BT_LOOP_WHEN(expr, num_spins) \
- do { \
- /*sched_yield may change errno*/ \
- const int saved_errno = errno; \
- for (int cnt = 0, saved_nspin = (num_spins); (expr); ++cnt) { \
- if (cnt < saved_nspin) { \
- cpu_relax(); \
- } else { \
- sched_yield(); \
- } \
- } \
- errno = saved_errno; \
- } while (0)
+#ifndef barrier
+#define barrier() asm volatile("": : :"memory")
+#endif // barrier
-#endif // BTHREAD_PROCESSOR_H
+#endif // BUTIL_PROCESSOR_H
\ No newline at end of file
diff --git a/src/butil/synchronization/seqlock.h
b/src/butil/synchronization/seqlock.h
new file mode 100644
index 00000000..3e91dcb5
--- /dev/null
+++ b/src/butil/synchronization/seqlock.h
@@ -0,0 +1,302 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef BUTIL_SYNCHRONIZATION_SEQLOCK_H
+#define BUTIL_SYNCHRONIZATION_SEQLOCK_H
+
+#include <stdint.h>
+#include <functional>
+#include <mutex>
+#include <type_traits>
+#include <utility>
+
+#include "butil/atomicops.h"
+#include "butil/compiler_specific.h"
+#include "butil/macros.h"
+#include "butil/processor.h"
+
+namespace butil {
+namespace internal {
+
+// Detects std::reference_wrapper<T>.
+template <typename T>
+struct IsReferenceWrapper : std::false_type {};
+template <typename T>
+struct IsReferenceWrapper<std::reference_wrapper<T>> : std::true_type {};
+
+// A sequence counter for consistent reads without acquiring a reader mutex
+// around caller-owned payloads. This is an implementation detail of
butil::Seqlock;
+// use butil::Seqlock instead.
+//
+// SeqCounter does not protect payload accesses from C++ data races. Callers
must
+// access shared payloads atomically (typically with relaxed ordering) or use
+// another mechanism that makes concurrent accesses valid.
+//
+//
+// Cache-line aligned so the sequence counter does not falsely share a line
+// with adjacent data (e.g. the writer mutex in Seqlock<Mutex> or a
neighbouring
+// payload), which would bounce the line between readers and writers.
+class BAIDU_CACHELINE_ALIGNMENT SeqCounter {
+public:
+ SeqCounter() : _seq(0) {}
+ DISALLOW_COPY_AND_ASSIGN(SeqCounter);
+
+ // Repeatedly invoke `load_payload` until it observes one consistent
version.
+ // `load_payload` may run several times when it races with a writer, so it
+ // must be cheap and side-effect free: only read the shared payload and
+ // return a copy, never mutate observable state. It must also access the
+ // shared payload without causing a C++ data race (relaxed atomics).
+ //
+ // The callback MUST return an owning value (a copy of the data), never a
+ // pointer, reference, or view (string_view, span, reference_wrapper, ...)
+ // into the payload. The consistency guarantee covers only the bytes copied
+ // out before the validating load: once load() returns, a later writer may
+ // mutate the payload, so any handle that still points into it no longer
+ // refers to a consistent snapshot.
+ template <typename Load>
+ typename std::decay<decltype(std::declval<Load&>()())>::type
+ load(Load&& load_payload) const {
+ typedef typename std::decay<decltype(std::declval<Load&>()())>::type
Result;
+ static_assert(!std::is_void<Result>::value,
+ "SeqCounter load callback must return a value");
+ static_assert(!std::is_pointer<Result>::value,
+ "SeqCounter load callback must return an owning value,
not "
+ "a pointer into the payload: the pointee can be mutated
by "
+ "a later writer, so it is not a consistent snapshot");
+ static_assert(!IsReferenceWrapper<Result>::value,
+ "SeqCounter load callback must return an owning value,
not "
+ "a std::reference_wrapper into the payload: the referent
"
+ "can be mutated by a later writer, so it is not a "
+ "consistent snapshot");
+
+ while (true) {
+ // Wait for any active writer, then read on an even sequence.
+ uint64_t seq;
+ while ((seq = _seq.load(memory_order_acquire)) & 1) {
+ cpu_relax();
+ }
+
+ Result result = load_payload();
+
+ // Keep the payload reads above before the sequence validation
+ // below; retry if a writer intervened.
+ atomic_thread_fence(memory_order_acquire);
+ if (_seq.load(memory_order_relaxed) == seq) {
+ return result;
+ }
+ }
+ }
+
+ // Invoke a payload writer inside one write section. Writers must be
serialized
+ // externally and must not nest write sections. The section is closed via
RAII
+ // even if `store_payload` throws, so an exception does not leave the
sequence
+ // permanently odd. A throwing writer may leave the payload partially
updated,
+ // so callers must ensure any partial state is still safe (non-crashing)
to read.
+ template <typename Store>
+ void store(Store&& store_payload) {
+ WriteGuard guard(*this);
+ store_payload();
+ }
+
+private:
+ // RAII scope for a write section.
+ class WriteGuard {
+ public:
+ explicit WriteGuard(SeqCounter& sc)
+ : _seq_counter(&sc)
+ , _next_seq(sc._seq.load(memory_order_relaxed) + 2) {
+ // Writers are serialized, so no atomic read-modify-write is
needed.
+ // Keep the next even sequence for publication when the guard
exits.
+ _seq_counter->_seq.store(_next_seq - 1, memory_order_relaxed);
+ // Order the odd-sequence store before the payload stores that
+ // follow. A release fence is a StoreStore (+LoadStore) barrier: it
+ // keeps any store after the fence (the payload writes) from being
+ // reordered ahead of stores before it (the odd sequence), so no
+ // reader can observe a payload write without also observing the
odd
+ // sequence. Equivalently, this release fence pairs -- through the
+ // payload atomics -- with the acquire fence in load(): if a reader
+ // reads a payload value published after this fence, that fence
+ // synchronizes-with the reader's acquire fence, so the reader is
+ // guaranteed to also see the odd sequence on its validating load
+ // and retry. The acquire half of an acq_rel fence would add
+ // nothing here (no prior load needs ordering), so release alone is
+ // sufficient and states the intent precisely.
+ atomic_thread_fence(memory_order_release);
+ }
+
+ DISALLOW_COPY_AND_ASSIGN(WriteGuard);
+
+ ~WriteGuard() {
+ // Publish the payload stores and leave the write section.
+ _seq_counter->_seq.store(_next_seq, memory_order_release);
+ }
+
+ private:
+ SeqCounter* _seq_counter;
+ uint64_t _next_seq;
+ };
+
+ atomic<uint64_t> _seq;
+};
+
+} // namespace internal
+
+// A sequence lock built on top of internal::SeqCounter.
+//
+// Seqlock<> : single-writer, no writer mutex. Backed directly by SeqCounter;
+// the caller must serialize writers with appropriate
synchronization
+// if ownership is transferred between threads.
+//
+// Seqlock<Mutex> : multi-writer. Owns a writer Mutex so that concurrent
+// store() calls are serialized automatically (this is the
+// Linux seqlock_t = SeqCounter + lock).
+//
+// In both forms readers do not acquire a mutex or prevent writers from
entering
+// a write section. Reads are NOT lock-free: they spin while the sequence is
odd
+// and retry if a write intervenes. A suspended writer can stall all readers,
and
+// continuous writes can starve readers; there is no bounded completion time.
+//
+// REQUIRED: write sections must not nest. A store() callback must not call
load()
+// or store() on the same Seqlock, or wait for work that needs to read or
write it.
+// In particular, a signal handler must not access the same Seqlock when it has
+// interrupted a writer: the interrupted write cannot finish until the handler
+// returns, so a reader in the handler would spin forever. Keep write sections
+// short and avoid blocking or yielding inside them.
+//
+// Example: publish a (x, y) pair atomically so readers never see a torn mix.
+//
+// // the caller-owned payload
+// struct Point {
+// butil::atomic<int64_t> x{0};
+// butil::atomic<int64_t> y{0};
+// };
+// Point point;
+// // single writer; use Seqlock<Mutex> if several threads may write.
+// butil::Seqlock<> seqlock;
+//
+// // Writer: the whole update is published as one consistent version.
+// void set(int64_t x, int64_t y) {
+// seqlock.store([&] {
+// point.x.store(x, butil::memory_order_relaxed);
+// point.y.store(y, butil::memory_order_relaxed);
+// });
+// }
+//
+// // Reader: load() retries internally until it copies out one consistent
+// // version, then returns whatever the callback returned.
+// std::pair<int64_t, int64_t> get() {
+// return seqlock.load([&] {
+// return std::make_pair(point.x.load(butil::memory_order_relaxed),
+// point.y.load(butil::memory_order_relaxed));
+// });
+// }
+//
+// REQUIRED: the payload accessed inside the load/store callbacks MUST be
+// atomic (e.g. butil::atomic fields, typically read/written with
+// memory_order_relaxed). This is not just a style preference -- the reader
+// deliberately reads the payload while a writer may be mutating it, so the
+// accesses are concurrent by design. Correctness relies on it in two ways:
+//
+// 1. Data race / UB. A non-atomic object read while another thread writes it
+// is a C++ data race, i.e. undefined behavior. The compiler is then free
+// to tear or fuse the access, invent extra reads, or hoist/sink it across
+// the fences below. The sequence-validation retry cannot rescue this: it
+// only tells you *whether* to retry, it cannot un-corrupt a value the
+// compiler already mangled, so you may return garbage even on a "clean"
+// (seq unchanged) read.
+// 2. Ordering. The release fence on the write side pairs with the acquire
fence
+// on the read side through the payload atomics (fence-fence
synchronization).
+// If the payload is not atomic that pairing does not hold, so "observe a
+// payload write => observe the odd sequence and retry" is no longer
guaranteed
+// and a reader can silently accept a stale or half-written snapshot.
+//
+// If you cannot make the payload atomic, do not use Seqlock -- use a Mutex or
+// an RWLock instead.
+//
+// REQUIRED: the load() callback must return an OWNING value (a copy of the
+// data), never a pointer, reference, or view (string_view, span,
+// reference_wrapper, ...) into the payload. load() only guarantees that the
+// bytes copied out before its validating load are consistent; after load()
+// returns a writer may mutate the payload again, so any handle still pointing
+// into it is no longer a consistent snapshot.
+//
+// Best fit: a small payload that is read far more often than written. The read
+// path copies the whole payload out and retries the copy whenever a write
races
+// it, so a large payload makes both the copy and the retries expensive. For a
+// large or heap-owning payload prefer a RCU/RWLock scheme instead.
+template <typename Mutex = void>
+class Seqlock;
+
+// Single-writer specialization: no mutex, delegates straight to SeqCounter.
+template <>
+class Seqlock<void> {
+public:
+ Seqlock() = default;
+ DISALLOW_COPY_AND_ASSIGN(Seqlock);
+
+ // Consistent read without acquiring a mutex; may spin/retry indefinitely.
+ template <typename Load>
+ typename std::decay<decltype(std::declval<Load&>()())>::type
+ load(Load&& load_payload) const {
+ return _seq.load(std::forward<Load>(load_payload));
+ }
+
+ // Single writer only: the caller MUST ensure there is no concurrent or
nested
+ // store(). The write section is closed via RAII even if store_payload
+ // throws.
+ template <typename Store>
+ void store(Store&& store_payload) {
+ _seq.store(std::forward<Store>(store_payload));
+ }
+
+private:
+ internal::SeqCounter _seq;
+};
+
+// Multi-writer specialization: SeqCounter + a writer Mutex.
+//
+// Mutex must be default-constructible and satisfy the C++ Lockable
+// requirements (lock()/unlock()), e.g. butil::Mutex.
+template <typename Mutex>
+class Seqlock {
+public:
+ Seqlock() = default;
+ DISALLOW_COPY_AND_ASSIGN(Seqlock);
+
+ // Consistent read without acquiring a mutex; may spin/retry indefinitely.
+ template <typename Load>
+ typename std::decay<decltype(std::declval<Load&>()())>::type
+ load(Load&& load_payload) const {
+ return _seq.load(std::forward<Load>(load_payload));
+ }
+
+ // Serialized write: acquires the mutex, then runs the payload writer in
+ // one write section.
+ template <typename Store>
+ void store(Store&& store_payload) {
+ std::lock_guard<Mutex> lk(_mutex);
+ _seq.store(std::forward<Store>(store_payload));
+ }
+
+private:
+ internal::SeqCounter _seq;
+ Mutex _mutex;
+};
+
+} // namespace butil
+
+#endif // BUTIL_SYNCHRONIZATION_SEQLOCK_H
diff --git a/test/BUILD.bazel b/test/BUILD.bazel
index 8706f63f..1aecc399 100644
--- a/test/BUILD.bazel
+++ b/test/BUILD.bazel
@@ -125,9 +125,10 @@ TEST_BUTIL_SOURCES = [
"recordio_unittest.cpp",
#"popen_unittest.cpp",
"bounded_queue_unittest.cc",
- "butil_unittest_main.cpp",
"scope_guard_unittest.cpp",
"optional_unittest.cpp",
+ "seqlock_unittest.cpp",
+ "butil_unittest_main.cpp",
] + select({
"@bazel_tools//tools/osx:darwin_x86_64": [],
"//conditions:default": [
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index 6aebe271..71baf85d 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -181,6 +181,7 @@ SET(TEST_BUTIL_SOURCES
${PROJECT_SOURCE_DIR}/test/scoped_locale.cc
${PROJECT_SOURCE_DIR}/test/scope_guard_unittest.cpp
${PROJECT_SOURCE_DIR}/test/optional_unittest.cpp
+ ${PROJECT_SOURCE_DIR}/test/seqlock_unittest.cpp
${PROJECT_SOURCE_DIR}/test/butil_unittest_main.cpp
)
diff --git a/test/Makefile b/test/Makefile
index f2348e7f..de4d566b 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -150,9 +150,10 @@ TEST_BUTIL_SOURCES = \
scoped_locale.cc \
popen_unittest.cpp \
bounded_queue_unittest.cc \
- butil_unittest_main.cpp \
- scope_guard_unittest.cpp \
- optional_unittest.cpp
+ scope_guard_unittest.cpp \
+ optional_unittest.cpp \
+ seqlock_unittest.cpp \
+ butil_unittest_main.cpp
ifeq ($(SYSTEM), Linux)
TEST_BUTIL_SOURCES += test_file_util_linux.cc \
diff --git a/test/seqlock_unittest.cpp b/test/seqlock_unittest.cpp
new file mode 100644
index 00000000..85ab4e9c
--- /dev/null
+++ b/test/seqlock_unittest.cpp
@@ -0,0 +1,391 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <unistd.h>
+#include <gtest/gtest.h>
+#include <pthread.h>
+#include <mutex>
+#include <cstdio>
+#include "butil/atomicops.h"
+#include "butil/time.h"
+#include "butil/synchronization/lock.h"
+#include "butil/synchronization/seqlock.h"
+
+namespace {
+
+// A multi-word payload. The seqlock's job is to make every reader observe a
+// snapshot in which all words are equal; a torn read would see a mix of an old
+// and a new value. Every field is a relaxed atomic, as Seqlock requires.
+static const int kWords = 8;
+struct Payload {
+ butil::atomic<uint64_t> w[kWords];
+
+ void relaxed_set(uint64_t v) {
+ for (auto& i : w) {
+ // Store word by word (not as one atomic group) so that, without
the
+ // seqlock, a concurrent reader could observe a torn value.
+ i.store(v, butil::memory_order_relaxed);
+ }
+ }
+ // Returns the first word and whether all words are equal to it.
+ uint64_t relaxed_get(bool* consistent) const {
+ uint64_t v0 = w[0].load(butil::memory_order_relaxed);
+ *consistent = true;
+ for (int i = 1; i < kWords; ++i) {
+ if (w[i].load(butil::memory_order_relaxed) != v0) {
+ *consistent = false;
+ }
+ }
+ return v0;
+ }
+};
+
+TEST(SeqlockTest, SingleThreadedReadWrite) {
+ butil::Seqlock<> seqlock;
+ Payload payload;
+ payload.relaxed_set(0);
+
+ for (uint64_t v = 1; v <= 1000; ++v) {
+ seqlock.store([&] { payload.relaxed_set(v); });
+ uint64_t got = seqlock.load([&] {
+ bool consistent = false;
+ uint64_t r = payload.relaxed_get(&consistent);
+ EXPECT_TRUE(consistent);
+ return r;
+ });
+ ASSERT_EQ(v, got);
+ }
+}
+
+TEST(SeqlockTest, LoadReturnsValueByType) {
+ butil::Seqlock<> seqlock;
+ butil::atomic<int> payload(0);
+ seqlock.store([&] {
+ payload.store(42, butil::memory_order_relaxed);
+ });
+ // load returns whatever the callback returns, by value.
+ int v = seqlock.load([&] {
+ return payload.load(butil::memory_order_relaxed);
+ });
+ ASSERT_EQ(42, v);
+ // A different return type also works.
+ std::pair<int, int> pr = seqlock.load([&] {
+ int v = payload.load(butil::memory_order_relaxed);
+ return std::make_pair(v, v + 1);
+ });
+ ASSERT_EQ(42, pr.first);
+ ASSERT_EQ(43, pr.second);
+}
+
+TEST(SeqlockTest, MutexSpecializationSingleThreaded) {
+ butil::Seqlock<butil::Mutex> seqlock;
+ Payload payload;
+ payload.relaxed_set(0);
+ for (uint64_t v = 1; v <= 1000; ++v) {
+ seqlock.store([&] { payload.relaxed_set(v); });
+ bool consistent = false;
+ uint64_t got = seqlock.load([&] {
+ bool c = false;
+ uint64_t r = payload.relaxed_get(&c);
+ consistent = c;
+ return r;
+ });
+ ASSERT_TRUE(consistent);
+ ASSERT_EQ(v, got);
+ }
+}
+
+// Single-threaded performance comparison: Seqlock vs Mutex.
+//
+// There is no contention here, so this measures the bare cost of the
+// synchronization itself:
+// * Seqlock<> : one relaxed counter load, two counter stores and a
+// release fence on the write path; two counter loads and
+// an acquire fence on the read path (no atomic RMW in
+// either path).
+// * Seqlock<Mutex> : the same counter work plus a lock/unlock pair on the
+// write path; the read path is identical to Seqlock<>.
+// * Mutex : a lock/unlock pair (atomic RMW + potential syscall on
+// contention) on BOTH the write and the read path.
+//
+// Timings are reported, not asserted: absolute numbers depend on the machine,
+// the compiler and the current load, so turning them into thresholds would
+// make the test flaky. The test only asserts that the measurements ran and
+// produced the expected values.
+static const int kPerfRounds = 1000000;
+
+// Accumulates every value read so the compiler cannot optimize the read loops
+// away as dead code.
+static butil::atomic<uint64_t> g_perf_sink(0);
+
+inline void perf_consume(uint64_t v) {
+ g_perf_sink.fetch_add(v, butil::memory_order_relaxed);
+}
+
+// A pure compiler barrier: it emits no instruction, but forbids the compiler
+// from moving memory accesses across it. Without one between the rounds below,
+// a write body made purely of relaxed stores is legally collapsible into its
+// last iteration -- which is exactly what happens to Seqlock<> once its
counter
+// increments are plain stores rather than fetch_add, and it measures 0 ns/op.
+// The mutex bodies resist that on their own; barriering all of them keeps the
+// comparison between the three honest.
+inline void perf_barrier() {
+ butil::atomic_signal_fence(butil::memory_order_seq_cst);
+}
+
+// Runs `body(round)` for `rounds` rounds and returns the elapsed nanoseconds.
+// A short warmup runs first so that page faults, branch predictor and cache
+// warmup are not charged to the measured loop.
+template <typename Body>
+int64_t TimeLoopNs(int rounds, Body&& body) {
+ int kWarmup = 1000;
+ for (int i = 1; i <= kWarmup; ++i) {
+ body(static_cast<uint64_t>(i));
+ perf_barrier();
+ }
+ int64_t start_ns = butil::cpuwide_time_ns();
+ for (int i = 1; i <= rounds; ++i) {
+ body(static_cast<uint64_t>(i));
+ perf_barrier();
+ }
+ return butil::cpuwide_time_ns() - start_ns;
+}
+
+inline double ns_per_op(int64_t elapsed_ns, int rounds) {
+ return static_cast<double>(elapsed_ns) / rounds;
+}
+
+TEST(SeqlockTest, SingleThreadedPerfVsMutex) {
+ // Seqlock<>: single-writer, no mutex.
+ butil::Seqlock<> seqlock;
+ Payload seqlock_payload;
+ seqlock_payload.relaxed_set(0);
+ int64_t seqlock_write_ns = TimeLoopNs(kPerfRounds, [&](uint64_t v) {
+ seqlock.store([&] { seqlock_payload.relaxed_set(v); });
+ // Make both the counter and payload observable after every write so
+ // the compiler cannot eliminate stores to these local test objects.
+ asm volatile("" : : "m"(seqlock), "m"(seqlock_payload) : "memory");
+ });
+ int64_t seqlock_read_ns = TimeLoopNs(kPerfRounds, [&](uint64_t) {
+ perf_consume(seqlock.load([&] {
+ bool consistent = false;
+ return seqlock_payload.relaxed_get(&consistent);
+ }));
+ });
+
+ // Seqlock<Mutex>: same read path, writes additionally take a mutex.
+ butil::Seqlock<butil::Mutex> mutex_seqlock;
+ Payload mutex_seqlock_payload;
+ mutex_seqlock_payload.relaxed_set(0);
+ int64_t mutex_seqlock_write_ns = TimeLoopNs(kPerfRounds, [&](uint64_t v) {
+ mutex_seqlock.store([&] { mutex_seqlock_payload.relaxed_set(v); });
+ asm volatile("" : : "m"(mutex_seqlock), "m"(mutex_seqlock_payload)
+ : "memory");
+ });
+ int64_t mutex_seqlock_read_ns = TimeLoopNs(kPerfRounds, [&](uint64_t) {
+ perf_consume(mutex_seqlock.load([&] {
+ bool consistent = false;
+ return mutex_seqlock_payload.relaxed_get(&consistent);
+ }));
+ });
+
+ // Plain Mutex baseline: both sides take the lock.
+ butil::Mutex mutex;
+ Payload mutex_payload;
+ mutex_payload.relaxed_set(0);
+ int64_t mutex_write_ns = TimeLoopNs(kPerfRounds, [&](uint64_t v) {
+ {
+ std::lock_guard<butil::Mutex> lk(mutex);
+ mutex_payload.relaxed_set(v);
+ }
+ asm volatile("" : : "m"(mutex), "m"(mutex_payload) : "memory");
+ });
+ int64_t mutex_read_ns = TimeLoopNs(kPerfRounds, [&](uint64_t) {
+ std::lock_guard<butil::Mutex> lk(mutex);
+ bool consistent = false;
+ perf_consume(mutex_payload.relaxed_get(&consistent));
+ });
+
+ printf("\n[seqlock perf] single thread, %d rounds, payload=%d words\n",
+ kPerfRounds, kWords);
+ printf("%-18s %14s %14s\n", "impl", "write(ns/op)", "read(ns/op)");
+ printf("%-18s %14.2f %14.2f\n", "Seqlock<>",
+ ns_per_op(seqlock_write_ns, kPerfRounds),
+ ns_per_op(seqlock_read_ns, kPerfRounds));
+ printf("%-18s %14.2f %14.2f\n", "Seqlock<Mutex>",
+ ns_per_op(mutex_seqlock_write_ns, kPerfRounds),
+ ns_per_op(mutex_seqlock_read_ns, kPerfRounds));
+ printf("%-18s %14.2f %14.2f\n", "Mutex",
+ ns_per_op(mutex_write_ns, kPerfRounds),
+ ns_per_op(mutex_read_ns, kPerfRounds));
+ printf("%-18s %13.2fx %13.2fx\n", "Mutex/Seqlock<>",
+ static_cast<double>(mutex_write_ns) / seqlock_write_ns,
+ static_cast<double>(mutex_read_ns) / seqlock_read_ns);
+
+ // The loops really ran and every implementation published the last value.
+ ASSERT_GT(seqlock_write_ns, 0);
+ ASSERT_GT(mutex_seqlock_write_ns, 0);
+ ASSERT_GT(mutex_write_ns, 0);
+ ASSERT_GT(seqlock_read_ns, 0);
+ ASSERT_GT(mutex_seqlock_read_ns, 0);
+ ASSERT_GT(mutex_read_ns, 0);
+ ASSERT_GT(g_perf_sink.load(butil::memory_order_relaxed), 0u);
+
+ bool consistent = false;
+ ASSERT_EQ(static_cast<uint64_t>(kPerfRounds),
+ seqlock_payload.relaxed_get(&consistent));
+ ASSERT_TRUE(consistent);
+ ASSERT_EQ(static_cast<uint64_t>(kPerfRounds),
+ mutex_seqlock_payload.relaxed_get(&consistent));
+ ASSERT_TRUE(consistent);
+ ASSERT_EQ(static_cast<uint64_t>(kPerfRounds),
+ mutex_payload.relaxed_get(&consistent));
+ ASSERT_TRUE(consistent);
+}
+
+// Concurrent consistency tests
+struct SharedState {
+ butil::Seqlock<>* single_writer_seqlock = nullptr; // single-writer lock
+ butil::Seqlock<butil::Mutex>* multi_writer_seqlock = nullptr; //
multi-writer lock
+ Payload payload;
+ butil::atomic<bool> stopped{false};
+ butil::atomic<uint64_t> version{0}; // source of the value written
+ butil::atomic<uint64_t> reads{0}; // total reads performed
+ butil::atomic<uint64_t> torn{0}; // inconsistent snapshots observed
+};
+
+void* SingleWriterThread(void* arg) {
+ SharedState* shared_state = static_cast<SharedState*>(arg);
+ while (!shared_state->stopped.load(butil::memory_order_relaxed)) {
+ uint64_t version = shared_state->version.fetch_add(1,
butil::memory_order_relaxed) + 1;
+ shared_state->single_writer_seqlock->store([&] {
+ shared_state->payload.relaxed_set(version);
+ });
+ }
+ return nullptr;
+}
+
+void* MultiWriterThread(void* arg) {
+ SharedState* shared_state = static_cast<SharedState*>(arg);
+ while (!shared_state->stopped.load(butil::memory_order_relaxed)) {
+ uint64_t version = shared_state->version.fetch_add(1,
butil::memory_order_relaxed) + 1;
+ shared_state->multi_writer_seqlock->store([&] {
+ shared_state->payload.relaxed_set(version);
+ });
+ }
+ return nullptr;
+}
+
+struct ReaderArg {
+ SharedState* shared_state;
+ bool multi_writer;
+};
+
+void* ReaderThread(void* arg) {
+ ReaderArg* reader_arg = static_cast<ReaderArg*>(arg);
+ SharedState* shared_state = reader_arg->shared_state;
+ uint64_t local_reads = 0;
+ uint64_t local_torn = 0;
+ while (!shared_state->stopped.load(butil::memory_order_relaxed)) {
+ bool consistent = false;
+ auto load_body = [&] {
+ bool c = false;
+ uint64_t r = shared_state->payload.relaxed_get(&c);
+ consistent = c;
+ return r;
+ };
+ if (reader_arg->multi_writer) {
+ shared_state->multi_writer_seqlock->load(load_body);
+ } else {
+ shared_state->single_writer_seqlock->load(load_body);
+ }
+ ++local_reads;
+ if (!consistent) {
+ ++local_torn;
+ }
+ }
+ shared_state->reads.fetch_add(local_reads, butil::memory_order_relaxed);
+ shared_state->torn.fetch_add(local_torn, butil::memory_order_relaxed);
+ return nullptr;
+}
+
+TEST(SeqlockTest, SingleWriterManyReaders) {
+ SharedState shared_state;
+ butil::Seqlock<> sl;
+ shared_state.single_writer_seqlock = &sl;
+ shared_state.payload.relaxed_set(0);
+
+ const int kReaders = 4;
+ pthread_t writer;
+ pthread_t readers[kReaders];
+ ReaderArg args[kReaders];
+
+ ASSERT_EQ(0, pthread_create(&writer, nullptr, SingleWriterThread,
&shared_state));
+ for (int i = 0; i < kReaders; ++i) {
+ args[i].shared_state = &shared_state;
+ args[i].multi_writer = false;
+ ASSERT_EQ(0, pthread_create(&readers[i], nullptr, ReaderThread,
&args[i]));
+ }
+
+ usleep(500 * 1000); // 0.5s of hammering
+ shared_state.stopped.store(true, butil::memory_order_relaxed);
+
+ pthread_join(writer, nullptr);
+ for (auto reader : readers) {
+ pthread_join(reader, nullptr);
+ }
+
+ ASSERT_GT(shared_state.reads.load(), 0u);
+ ASSERT_EQ(0u, shared_state.torn.load()) << "readers observed torn
snapshots";
+}
+
+TEST(SeqlockTest, MultiWriterManyReaders) {
+ SharedState shared_state;
+ butil::Seqlock<butil::Mutex> seqlock;
+ shared_state.multi_writer_seqlock = &seqlock;
+ shared_state.payload.relaxed_set(0);
+
+ const int kWriters = 3;
+ const int kReaders = 4;
+ pthread_t writers[kWriters];
+ pthread_t readers[kReaders];
+ ReaderArg args[kReaders];
+
+ for (auto& writer : writers) {
+ ASSERT_EQ(0, pthread_create(&writer, nullptr, MultiWriterThread,
&shared_state));
+ }
+ for (int i = 0; i < kReaders; ++i) {
+ args[i].shared_state = &shared_state;
+ args[i].multi_writer = true;
+ ASSERT_EQ(0, pthread_create(&readers[i], nullptr, ReaderThread,
&args[i]));
+ }
+
+ usleep(500 * 1000);
+ shared_state.stopped.store(true, butil::memory_order_relaxed);
+
+ for (auto writer : writers) {
+ pthread_join(writer, nullptr);
+ }
+ for (auto reader : readers) {
+ pthread_join(reader, nullptr);
+ }
+
+ ASSERT_GT(shared_state.reads.load(), 0u);
+ ASSERT_EQ(0u, shared_state.torn.load()) << "readers observed torn
snapshots";
+}
+
+} // namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
