Support 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, 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)
@@ -821,7 +726,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.

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
@@ -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.

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