如何在多线程环境下实现u128类型的原子加法操作？

背景 #

在多线程环境中运行程序，需要每秒25-50次将64位值添加到全局u128类型计数器中。

初始实现的漏洞分析 #

你给出的初始AtomicU128实现存在严重线程安全问题：
当多个线程同时触发lsv溢出时，每个线程都会独立执行msv.fetch_add(1)，导致msv被重复累加。比如lsv初始为u64::MAX，两个线程同时调用fetch_add(1)，每个线程都会判断出溢出并给msv加1，但实际上只需要加1就能正确表示128位溢出结果，最终msv会变成2而非1，导致计数器值完全错误。
此外，该实现没有保证lsv和msv的更新是原子操作，线程间的操作可能交叉，导致整体状态不一致。

用Mutex包裹时是否需要AtomicU64？ #

不需要。
Mutex的核心作用是保证同一时间只有一个线程能访问被包裹的数据，此时内部使用普通u128就足够安全。原子类型（AtomicU64）是用于无锁线程同步场景的，在已经用Mutex做互斥的情况下，原子类型的特性完全用不上，反而会带来不必要的开销。

更优的u128安全加法实现方式 #

根据你的操作频率（每秒25-50次，属于低频率），以下几种方案均可选择：

方案1：简化Mutex封装 #

你现有的Mutex实现已经足够安全，还可以进一步简化，甚至无需自定义结构体，直接使用Arc<Mutex<u128>>：

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let counter = Arc::new(Mutex::new(u64::MAX as u128));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter_clone = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut inner = counter_clone.lock().unwrap();
            *inner += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Final value: {}", *counter.lock().unwrap());
}

该方案实现简单，完全基于标准库，对于低频率操作来说性能完全够用。

方案2：无锁CAS实现AtomicU128 #

如果想要无锁实现（虽然对你的频率来说必要性不大），可以通过CAS循环保证128位加法的原子性：

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::thread;

#[derive(Debug)]
pub struct AtomicU128 {
    low: AtomicU64,
    high: AtomicU64,
}

impl AtomicU128 {
    pub fn new(value: u128) -> Self {
        AtomicU128 {
            low: AtomicU64::new((value & u64::MAX as u128) as u64),
            high: AtomicU64::new((value >> 64) as u64),
        }
    }

    pub fn fetch_add(&self, value: u64) -> u128 {
        loop {
            // 读取当前高低位的一致快照
            let current_high = self.high.load(Ordering::SeqCst);
            let current_low = self.low.load(Ordering::SeqCst);
            let current = (current_high as u128) << 64 | current_low as u128;

            // 计算新值
            let new = current + value as u128;
            let new_low = (new & u64::MAX as u128) as u64;
            let new_high = (new >> 64) as u64;

            // CAS更新低位，失败则重试
            if self.low.compare_exchange_weak(
                current_low,
                new_low,
                Ordering::SeqCst,
                Ordering::Relaxed,
            ).is_ok() {
                // 高位有变化时再CAS更新
                if current_high != new_high {
                    self.high.compare_exchange(
                        current_high,
                        new_high,
                        Ordering::SeqCst,
                        Ordering::SeqCst,
                    ).unwrap();
                }
                return current;
            }
        }
    }

    pub fn load(&self) -> u128 {
        let high = self.high.load(Ordering::SeqCst);
        let low = self.low.load(Ordering::SeqCst);
        (high as u128) << 64 | low as u128
    }
}

fn main() {
    let counter = Arc::new(AtomicU128::new(u64::MAX as u128));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter_clone = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            counter_clone.fetch_add(1);
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Final value: {}", counter.load());
    println!("{:?}", counter);
}

该实现通过CAS循环保证了128位加法的原子性，避免了Mutex的开销，但实现相对复杂。对于你的低频率操作，Mutex方案已经完全够用。

总结 #

• 你的操作频率极低，Mutex方案是最简单且足够安全的选择，无需过度追求无锁实现。
• 若确实需要无锁，上述CAS循环实现可保证线程安全。

内容的提问来源于stack exchange，提问作者Jamie