Add butil::Seqlock and use it for TaskGroup CPU time stat

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 {                                                                \

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<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 != NULL) {

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<int64_t>*)&_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<int64_t>*)&_cumulated_cputime_ns)
+                ->load(butil::memory_order_relaxed);
+        }
+
+        CPUTimeStat atomic_load() const {
+            return {
+                ((butil::atomic<int64_t>*)&_last_run_ns_and_type)
+                    ->load(butil::memory_order_relaxed),
+                ((butil::atomic<int64_t>*)&_cumulated_cputime_ns)
+                    ->load(butil::memory_order_relaxed)
+            };
+        }
+
+        void atomic_store(CPUTimeStat stat) {
+            ((butil::atomic<int64_t>*)&_last_run_ns_and_type)
+                ->store(stat._last_run_ns_and_type, butil::memory_order_relaxed);
+            ((butil::atomic<int64_t>*)&_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.

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_