Rust 原理丨聊一聊 Rust 的 Atomic 和内存顺序

fn main() {
    let num_done = AtomicUsize::new(0);

    let main_thread = thread::current();

    thread::scope(|s| {
        s.spawn(|| {
            for i in 0..100 {
                sleep(Duration::from_millis(10));
                num_done.store(i + 1, std::sync::atomic::Ordering::Relaxed);  // store 存储
                main_thread.unpark();
            }
        });

        loop {
            let n = num_done.load(std::sync::atomic::Ordering::Relaxed); // load 获取
            if n == 100 {
                break;
            }
            println!("Working... {n}/100 done");
            thread::park_timeout(Duration::from_millis(1));
        }
    });
    println!("Done!");
}

fn main() {
    let num_done = &AtomicUsize::new(0);

    thread::scope(|s| {
        s.spawn(|| {
            for _ in 0..100 {
                num_done.fetch_add(1, std::sync::atomic::Ordering::Relaxed); // 使用 fetch_add 进行加 1
            }
        });

        loop {
            let n = num_done.load(std::sync::atomic::Ordering::Relaxed);
            if n == 100 {
                break;
            }
            println!("Working... {n}/100 done");
        }
    });
    println!("Done!");
}

fn allocate_new_id() -> u32 {
    static NEXT_ID: AtomicU32 = AtomicU32::new(0);
    let mut id = NEXT_ID.load(std::sync::atomic::Ordering::Relaxed);
    loop {
        assert!(id < 1000, "Too many IDs!");
        match NEXT_ID.compare_exchange(  // 只有 id 没有发生变化，才允许进行加 1
            id,
            id + 1,
            std::sync::atomic::Ordering::Relaxed,
            std::sync::atomic::Ordering::Relaxed,
        ) {
            Ok(_) => return id,
            Err(v) => id = v,
        }
    }
}

CPU 发出 LOCK 信号
   └── 激活处理器的 LOCK# 引脚
      └── 获得总线的独占访问权
          └── 执行原子操作
              └── 释放 LOCK 信号
                  └── 其他处理器可以访问内存

M (Modified)：已修改
E (Exclusive)：独占
S (Shared)：共享
I (Invalid)：无效

操作流程：

1. 检查数据是否在缓存中
2. 如果在，将状态改为 Exclusive
3. 执行原子操作
4. 通知其他 CPU 使其缓存失效

; 原子加法操作
lock add dword ptr [memory], 1

; 比较并交换
lock cmpxchg dword ptr [memory], eax

use std::sync::atomic::{AtomicI64, Ordering};
use std::thread;

static ATOMIC: AtomicI64 = AtomicI64::new(0);

fn main() {
    let t1 = thread::spawn(|| {
        ATOMIC.store(10086, Ordering::Release);
    });

    let t2 = thread::spawn(|| {
        let val = ATOMIC.load(Ordering::Acquire);
        println!("{val}");
    });

    t1.join().unwrap();
    t2.join().unwrap();
}

rustc -O --emit asm src/main.rs

__ZN3std3sys9backtrace28__rust_begin_short_backtrace17h750d7a3a9c81fc67E:
	.cfi_startproc
Lloh8:
	adrp	x8, __ZN4main6ATOMIC17hd0b0dbf92e477148E.0@PAGE
Lloh9:
	add	x8, x8, __ZN4main6ATOMIC17hd0b0dbf92e477148E.0@PAGEOFF
	mov	w9, #10086 ; 将值 10086 移入寄存器
	stlr	x9, [x8] ; Store-Release 指令，原子地存储值
	; InlineAsm Start
	; InlineAsm End
	ret
	.loh AdrpAdd	Lloh8, Lloh9
	.cfi_endproc

Lloh11:
	add	x8, x8, __ZN4main6ATOMIC17hd0b0dbf92e477148E.0@PAGEOFF
	ldapr	x8, [x8] ; ; Load-Acquire 指令，原子地加载值
	str	x8, [sp, #8]

package main

import (
	"sync/atomic"
)

func main() {
	data := atomic.Int64{}
	go func() {
		data.Store(10086)
	}()

	go func() {
		a := data.Load()
		println(a)
	}()
}

go build -gcflags=-S -ldflags=-w main.go 2> assembly.txt

	0x0008 00008 (/Users/wangjiahan/go/go1.23.2/src/sync/atomic/type.go:109)	MOVD	$10086, R1
	0x000c 00012 (/Users/wangjiahan/go/go1.23.2/src/sync/atomic/type.go:109)	STLR	R1, (R0)

	0x001c 00028 (/Users/wangjiahan/goStudy/go-atomic/main.go:14)	HINT	$0
	0x0020 00032 (/Users/wangjiahan/go/go1.23.2/src/sync/atomic/type.go:106)	LDAR	(R0), R0
	0x0024 00036 (/Users/wangjiahan/go/go1.23.2/src/sync/atomic/type.go:106)	MOVD	R0, main..autotmp_6-8(SP)

fn f(a: &mut i32, b: &mut i32) {
  *a += 1;
  *b += 1;
  *a += 1;
}

fn f(a: &mut i32, b: &mut i32) {
  *a += 2;
  *b += 1;
}

static X: AtomicI32 = AtomicI32::new(0);

fn main() {
    X.store(1, Relaxed);
    let t = thread::spawn(f);
    X.store(2, Relaxed);
    t.join().unwrap();
    X.store(3, Relaxed);
}

fn f() {
    let x = X.load(Relaxed);
    assert!(x == 1 || x == 2);
}

static X: AtomicI32 = AtomicI32::new(0);

fn a() {
    X.fetch_add(5, Relaxed);
    X.fetch_add(10, Relaxed);
}

fn b() {
    let a = X.load(Relaxed);
    let b = X.load(Relaxed);
    let c = X.load(Relaxed);
    let d = X.load(Relaxed);
    println!("{a} {b} {c} {d}");  // 这个输出不一定
}
fn main() {
    thread::scope(|s| {
        s.spawn(a);
        s.spawn(b);
    });

    println!("{:?}", X.load(Relaxed)); // 最终结果一定是 15
}

static X: AtomicI32 = AtomicI32::new(0);

fn a1() {
    X.fetch_add(5, Relaxed);
}

fn a2() {
    X.fetch_add(10, Relaxed);
}

fn b() {
    let a = X.load(Relaxed);
    let b = X.load(Relaxed);
    let c = X.load(Relaxed);
    let d = X.load(Relaxed);
    println!("{a} {b} {c} {d}");  // 这个输出不一定
}
fn main() {
    thread::scope(|s| {
        s.spawn(a1);
      	s.apawn(a2);
        s.spawn(b);
    });

    println!("{:?}", X.load(Relaxed)); // 最终结果一定是 15
}

static DATA: AtomicI32 = AtomicI32::new(0);
static READY: AtomicBool = AtomicBool::new(false);

fn main() {
    thread::scope(|s| {
        // 线程 A - 写入者
        s.spawn(|| {
            DATA.store(123, Ordering::Relaxed);     // ① 准备数据
            READY.store(true, Ordering::Relaxed);   // ② 发出数据就绪信号
        });

        // 线程 B - 读取者
        s.spawn(|| {
            while !READY.load(Ordering::Relaxed) {  // ③ 等待数据就绪信号
                thread::yield_now();
            }
            assert_eq!(DATA.load(Ordering::Relaxed), 123); // ④ 获取数据，这里断言一定成功吗？
        });
    });
}

DATA.store(123, Ordering::Relaxed);     // 准备数据
READY.store(true, Ordering::Relaxed);   // 发出数据就绪信号

static DATA: AtomicI32 = AtomicI32::new(0);
static READY: AtomicBool = AtomicBool::new(false);

fn main() {
    thread::scope(|s| {
        s.spawn(|| {
            DATA.store(123, Ordering::Relaxed);
            READY.store(true, Ordering::Release);   // 这里改为 release
        });

        s.spawn(|| {
            while !READY.load(Ordering::Acquire) {  // 这里改为 acquire
                thread::yield_now();
            }
            assert_eq!(DATA.load(Ordering::Relaxed), 123); // 必定成功
        });
    });
}

static mut DATA: u64 = 0;
static READY: AtomicBool = AtomicBool::new(false);

fn main() {
    thread::spawn(|| {
        // Safety: 此时没有其他线程访问 DATA，
        // 因为我们还没有设置 READY 标志
        unsafe { DATA = 123 };
        READY.store(true, Release); // 在这个存储操作之前的所有内存操作 ..
    });
    while !READY.load(Acquire) { // .. 在这个加载操作返回 true 后都是可见的
        thread::sleep(Duration::from_millis(100));
        println!("waiting...");
    }
    // Safety: 没有线程会修改 DATA，因为 READY 已经被设置
    println!("{}", unsafe { DATA });
}

use std::sync::atomic::Ordering::SeqCst;

static A: AtomicBool = AtomicBool::new(false);
static B: AtomicBool = AtomicBool::new(false);

static mut S: String = String::new();

fn main() {
    let a = thread::spawn(|| {
        A.store(true, SeqCst);
        if !B.load(SeqCst) {
            unsafe { S.push('!') };
        }
    });

    let b = thread::spawn(|| {
        B.store(true, SeqCst);
        if !A.load(SeqCst) {
            unsafe { S.push('!') };
        }
    });

    a.join().unwrap();
    b.join().unwrap();
}

; x86/x64
MFENCE  ; 全屏障
LFENCE  ; 读屏障
SFENCE  ; 写屏障

; ARM
DMB     ; 数据内存屏障
DSB     ; 数据同步屏障
ISB     ; 指令同步屏障

// Release 写入会插入 Store Barrier
atomic.store(42, Ordering::Release);  // 编译器会在此处插入 Store Barrier

// Acquire 读取会插入 Load Barrier
let x = atomic.load(Ordering::Acquire);  // 编译器会在此处插入 Load Barrier

// SeqCst 操作会插入 Full Barrier
atomic.store(42, Ordering::SeqCst);  // 编译器会在此处插入 Full Barrier

package main

import (
	"sync/atomic"
)

func main() {
	data := atomic.Int64{}
	data.Add(1)
	data.And(2)
	data.Or(3)
	data.Swap(4)
	data.Store(5)
	data.Load()
	data.CompareAndSwap(6, 7)
}

// The load and store operations, implemented by the LoadT and StoreT
// functions, are the atomic equivalents of "return *addr" and
// "*addr = val".
//
// In the terminology of [the Go memory model], if the effect of
// an atomic operation A is observed by atomic operation B,
// then A “synchronizes before” B.
// Additionally, all the atomic operations executed in a program
// behave as though executed in some sequentially consistent order.
// This definition provides the same semantics as
// C++'s sequentially consistent atomics and Java's volatile variables.
//
// [the Go memory model]: https://go.dev/ref/mem

If you must read the rest of this document to understand the behavior of your program, you are being too clever.
Don't be clever.

Share memory by communicating; don't communicate by sharing memory.