From ff91c607b1fcc553d02dbdaa7962aa35ac9edc47 Mon Sep 17 00:00:00 2001 From: chenBright Date: Fri, 14 Aug 2026 23:40:06 +0800 Subject: [PATCH] Support seqlock and use it for TaskGroup CPU time stat --- src/bthread/processor.h | 28 +-- src/bthread/task_group.cpp | 97 ---------- src/bthread/task_group.h | 92 +++++---- src/butil/processor.h | 48 +++++ src/butil/synchronization/seqlock.h | 287 ++++++++++++++++++++++++++++ test/BUILD.bazel | 1 + test/CMakeLists.txt | 1 + test/Makefile | 3 +- test/seqlock_unittest.cpp | 240 +++++++++++++++++++++++ 9 files changed, 624 insertions(+), 173 deletions(-) create mode 100644 src/butil/processor.h create mode 100644 src/butil/synchronization/seqlock.h create mode 100644 test/seqlock_unittest.cpp diff --git a/src/bthread/processor.h b/src/bthread/processor.h index 06e75641e3..74be9e25ef 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 777d7514a3..5bcbdf9b79 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, NULL); 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(&_value)); - return {value[0], value[1]}; -#elif defined(__ARM_NEON) - (void)_mutex; - (void)_seq; - int64x2_t value = vld1q_s64(reinterpret_cast(&_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(_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(&value)); - vst1q_s64(reinterpret_cast(&_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(_mutex)); - _value = value; -#endif -} - - int TaskGroup::get_attr(bthread_t tid, bthread_attr_t* out) { TaskMeta* const m = address_meta(tid); if (m != NULL) { diff --git a/src/bthread/task_group.h b/src/bthread/task_group.h index c21e06ba39..556823ba66 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 @@ class ExitException : public std::exception { 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 @@ -240,14 +210,11 @@ friend class TaskControl; 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) {} + CPUTimeStat() : CPUTimeStat(0, 0) {} - // Convert to AtomicInteger128::Value for atomic operations. - explicit operator AtomicInteger128::Value() const { - return {_last_run_ns_and_type, _cumulated_cputime_ns}; - } + CPUTimeStat(int64_t last_run_ns_and_type, int64_t cumulated_cputime_ns) + : _last_run_ns_and_type(last_run_ns_and_type) + , _cumulated_cputime_ns(cumulated_cputime_ns) {} 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) | @@ -259,6 +226,10 @@ friend class TaskControl; int64_t last_run_ns_and_type() const { return _last_run_ns_and_type; } + int64_t last_run_ns_and_type_atomic_load() const { + return ((butil::atomic*)&_last_run_ns_and_type) + ->load(butil::memory_order_relaxed); + } bool is_main_task() const { return _last_run_ns_and_type & TASK_TYPE_MASK; @@ -273,6 +244,26 @@ friend class TaskControl; int64_t cumulated_cputime_ns() const { return _cumulated_cputime_ns; } + int64_t cumulated_cputime_ns_atomic_load() const { + return ((butil::atomic*)&_cumulated_cputime_ns) + ->load(butil::memory_order_relaxed); + } + + CPUTimeStat atomic_load() const { + return { + ((butil::atomic*)&_last_run_ns_and_type) + ->load(butil::memory_order_relaxed), + ((butil::atomic*)&_cumulated_cputime_ns) + ->load(butil::memory_order_relaxed) + }; + } + + void atomic_store(CPUTimeStat stat) { + ((butil::atomic*)&_last_run_ns_and_type) + ->store(stat._last_run_ns_and_type, butil::memory_order_relaxed); + ((butil::atomic*)&_cumulated_cputime_ns) + ->store(stat._cumulated_cputime_ns, butil::memory_order_relaxed); + } private: // The higher bit for task type, main task is 1, otherwise 0. @@ -285,18 +276,23 @@ friend class TaskControl; class AtomicCPUTimeStat { public: CPUTimeStat load() const { - return _cpu_time_stat.load(); + return _seqlock.load([&]() -> CPUTimeStat { + return _stat.atomic_load(); + }); } CPUTimeStat load_unsafe() const { - return _cpu_time_stat.load_unsafe(); + return _stat; } - void store(CPUTimeStat cpu_time_stat) { - _cpu_time_stat.store(AtomicInteger128::Value(cpu_time_stat)); + void store(CPUTimeStat stat) { + _seqlock.store([this, stat]() { + _stat.atomic_store(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/butil/processor.h b/src/butil/processor.h new file mode 100644 index 0000000000..8020a0533c --- /dev/null +++ b/src/butil/processor.h @@ -0,0 +1,48 @@ +// 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_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 +#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 // cpu_relax + +// Compile read-write barrier +#ifndef barrier +#define barrier() asm volatile("": : :"memory") +#endif // barrier + +#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 0000000000..d02f52aab7 --- /dev/null +++ b/src/butil/synchronization/seqlock.h @@ -0,0 +1,287 @@ +// 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 +#include +#include +#include +#include + +#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. +template +struct IsReferenceWrapper : std::false_type {}; +template +struct IsReferenceWrapper> : std::true_type {}; + +// A sequence counter for implementing lock-free, consistent reads 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 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 std::decay()())>::type + load(Load&& load_payload) const { + typedef typename std::decay()())>::type Result; + static_assert(!std::is_void::value, + "SeqCounter load callback must return a value"); + static_assert(!std::is_pointer::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::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(butil::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. + butil::atomic_thread_fence(butil::memory_order_acquire); + if (_seq.load(butil::memory_order_relaxed) == seq) { + return result; + } + } + } + + // Invoke a payload writer inside one write section. Writers must be serialized + // externally. The section is closed via RAII even if `store_payload` throws an + // exception, so readers are guaranteed to make progress. Note that a throwing + // writer may leave the payload partially updated, so callers must ensure any + // partial state is still safe (non-crashing) to read. + template + 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) { + _seq_counter->_seq.fetch_add(1, butil::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. + butil::atomic_thread_fence(butil::memory_order_release); + } + + DISALLOW_COPY_AND_ASSIGN(WriteGuard); + + ~WriteGuard() { + // Publish the payload stores and leave the write section. + _seq_counter->_seq.fetch_add(1, butil::memory_order_release); + } + + private: + SeqCounter* _seq_counter; + }; + + butil::atomic _seq; +}; + +} // namespace internal + +// A sequence lock built on top of internal::SeqCounter. +// +// Seqlock<> : single-writer, lock-free. Backed directly by SeqCounter; +// the caller must guarantee at most one writer at a time +// (concurrent store() is undefined behavior). +// +// Seqlock : 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 reads are lock-free and never block writers. +// +// Example: publish a (x, y) pair atomically so readers never see a torn mix. +// +// // the caller-owned payload +// struct Point { +// butil::atomic x{0}; +// butil::atomic y{0}; +// }; +// Point point; +// // single writer; use Seqlock 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 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 +class Seqlock; + +// Single-writer specialization: no mutex, delegates straight to SeqCounter. +template <> +class Seqlock { +public: + Seqlock() = default; + DISALLOW_COPY_AND_ASSIGN(Seqlock); + + // Lock-free consistent read. + template + typename std::decay()())>::type + load(Load&& load_payload) const { + return _seq.load(std::forward(load_payload)); + } + + // Single writer only: the caller MUST ensure there is no concurrent + // store(). The write section is closed via RAII even if store_payload + // throws. + template + void store(Store&& store_payload) { + _seq.store(std::forward(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 +class Seqlock { +public: + Seqlock() = default; + DISALLOW_COPY_AND_ASSIGN(Seqlock); + + // Lock-free consistent read. + template + typename std::decay()())>::type + load(Load&& load_payload) const { + return _seq.load(std::forward(load_payload)); + } + + // Serialized write: acquires the mutex, then runs the payload writer in + // one write section. + template + void store(Store&& store_payload) { + std::lock_guard lk(_mutex); + _seq.store(std::forward(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 27d63aed81..9428d053df 100644 --- a/test/BUILD.bazel +++ b/test/BUILD.bazel @@ -128,6 +128,7 @@ TEST_BUTIL_SOURCES = [ "butil_unittest_main.cpp", "scope_guard_unittest.cpp", "optional_unittest.cpp", + "seqlock_unittest.cpp", ] + select({ "@bazel_tools//tools/osx:darwin_x86_64": [], "//conditions:default": [ diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 6aebe271f4..71baf85df4 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 136b6f6803..1efe754197 100644 --- a/test/Makefile +++ b/test/Makefile @@ -152,7 +152,8 @@ TEST_BUTIL_SOURCES = \ bounded_queue_unittest.cc \ butil_unittest_main.cpp \ scope_guard_unittest.cpp \ - optional_unittest.cpp + optional_unittest.cpp \ + seqlock_unittest.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 0000000000..aec14a9165 --- /dev/null +++ b/test/seqlock_unittest.cpp @@ -0,0 +1,240 @@ +// 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 +#include +#include +#include "butil/atomicops.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 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 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 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 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); + } +} + +// Concurrent consistency tests +struct SharedState { + butil::Seqlock<>* single_writer_seqlock = NULL; // single-writer lock + butil::Seqlock* multi_writer_seqlock = NULL; // multi-writer lock + Payload payload; + butil::atomic stopped{false}; + butil::atomic version{0}; // source of the value written + butil::atomic reads{0}; // total reads performed + butil::atomic torn{0}; // inconsistent snapshots observed +}; + +void* SingleWriterThread(void* arg) { + SharedState* shared_state = static_cast(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 NULL; +} + +void* MultiWriterThread(void* arg) { + SharedState* shared_state = static_cast(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 NULL; +} + +struct ReaderArg { + SharedState* shared_state; + bool multi_writer; +}; + +void* ReaderThread(void* arg) { + ReaderArg* reader_arg = static_cast(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 NULL; +} + +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, NULL, 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], NULL, ReaderThread, &args[i])); + } + + usleep(500 * 1000); // 0.5s of hammering + shared_state.stopped.store(true, butil::memory_order_relaxed); + + pthread_join(writer, NULL); + for (auto reader : readers) { + pthread_join(reader, NULL); + } + + 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 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, NULL, 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], NULL, ReaderThread, &args[i])); + } + + usleep(500 * 1000); + shared_state.stopped.store(true, butil::memory_order_relaxed); + + for (auto writer : writers) { + pthread_join(writer, NULL); + } + for (auto reader : readers) { + pthread_join(reader, NULL); + } + + ASSERT_GT(shared_state.reads.load(), 0u); + ASSERT_EQ(0u, shared_state.torn.load()) << "readers observed torn snapshots"; +} + +} // namespace