指令引用了内存该内存不能为read 一直弹窗_【翻译】使用Rust测试ARM和X86内存模型

原文标题: The Story of Tail Call Optimizations in Rust 原文标题: Examining ARM vs X86 Memory Models with Rust
原文链接: https://www.nickwilcox.com/blog/arm_vs_x86_memory_model/
公众号： Rust碎碎念

苹果公司最近宣布，他们将要把笔记本和桌面电脑从Intel x86 CPU 迁移到自研的ARM架构的CPU。我认为是时候来看一下这两者之间那些会对使用Rust工作的系统程序员有影响的区别了。
ARM架构的CPU不同于X86 CPU的很重要的一点是它们的内存模型。这篇文章将会讨论什么是内存模型以及它是如何让代码在一种CPU架构上正确运行而在另一种CPU架构上引起竞争条件(race condition)。

内存模型

特定CPU上多个线程之间交互时对内存进行加载(load)和存储(store)的方式称为该架构的内存模型。
根据CPU的内存模型的不同，一个线程的多次写入操作可能会被另一个线程以不同的顺序可见。
进行多次读取操作的线程也是如此。一个正在进行多次读取操作的线程可能收到全局状态的“快照”，这些状态表示的时间顺序不同于事实上发生的顺序。
现代硬件需要这种灵活性从而能够最大化内存操作的吞吐量。每次CPU的更新换代就会提升CPU的时钟频率和核数，但是内存带宽一直在努力追赶保持同步。将数据从内存中取出进行操作通常是应用程序的性能瓶颈。
如果你从来没有写过多线程代码，或者仅仅使用高级同步原语，如std::sync::Mutex来完成任务，那你可能从来没有接触过内存模型的细节。这是因为，不管CPU的内存模型允许它执行什么样的重新排序，它总是对当前线程呈现出一致的内存视图。
如果我们看一下下面的代码片段，这段代码写入内存然后直接读取相同的内存，当我们进行读取时，我们总能按照预期读到58。我们永远不会从内存中读取过时的值。

pub unsafe fn read_after_write(u32_ptr: *mut u32) {u32_ptr.write_volatile(58);let u32_value = u32_ptr.read_volatile();println!("the value is {}", u32_value);
}

我之所以使用volatile操作是因为如果我使用普通的指针操作，编译器就会足够聪明地跳过内存读取而直接打印出58。Volatile操作阻止编译器重排序或跳过内存操作。但是，他们对硬件没有影响(或者说，编译器重排序相对于非易失性内存操作)。
一旦我们引入了多线程，我们就会面临这样一个事实：CPU可能对我们的内存操作重排序。
我们可以在多线程环境中测试下面的代码片段:

pub unsafe fn writer(u32_ptr_1: *mut u32, u32_ptr_2: *mut u32) {u32_ptr_1.write_volatile(58);u32_ptr_2.write_volatile(42);
}pub unsafe fn reader(u32_ptr_1: *mut u32, u32_ptr_2: *mut u32) -> (u32, u32) {(u32_ptr_1.read_volatile(), u32_ptr_2.read_volatile())
}

如果我们把两个指针指向的内容都初始化为0，然后每个函数放在不同的线程中运行，我们可以列出可能读取到的结果。我们知道，虽然没有同步机制，但是基于我们对单线程中代码的经验，我们可以想到可能的返回值是(0,0)，(58,0)，(58,42)。但是硬件对内存写操作的重排序可能会影响多线程，这意味着，还有第四种可能性(0,42)。
你可能认为，由于缺少同步机制，可能会产生更多的可能性。但是所有的硬件内存模型保证了原生字(word)对齐的加载(load)和存储(store)是原子性的(32位CPU的u32类型，64位CPU的u64类型)。如果我们把其中一个写入改为0xFFFF_FFFF，读取操作将永远只能看到旧值或新值。它将不会看到一个不完整的值，比如0xFFFF_0000。
当使用常规方式访问内存时，如果CPU的内存模型的细节被隐藏起来，当其影响到程序的正确性时，似乎我们就没有办法在多线程程序中对其进行控制。
幸运地是，Rust提供了如std::sync::atomic这样的模块，其中提供了能够满足我们控制需要的类型。我们使用这些类型来明确指定我们的代码所需要的内存序(memory order)要求。我们用性能换取正确性。我们对硬件执行内存操作的顺序进行了限制，取消了硬件希望执行的带宽优化。
当使用atomic模块进行工作的时候，我们不用担心各个CPU架构上的实际的内存模型。atomic模块工作在一个抽象的内存模型之上，对底层CPU并不知道。一旦我们在使用Rust内存模型时表明我们对加载(load)和存储(store)的需求，编译器就会将其映射到目标CPU的内存模型上。
我们对于每个操作的要求表现为我们想要在操作上允许(或拒绝)什么样的重排序。次序形成了一个层级，每一层对CPU进行了更多的限制。例如，Ordering::Relaxed意味着CPU可以自由执行任意的重排序。Ordering::Release意味着一个存储(store)操作只能在所有正在进行的存储完成结束之后才能完成。
让我们来看看，原子内存写操作相比较于常规写操作，实际上是怎么编译的。

use std::sync::atomic::*;pub unsafe fn test_write(shared_ptr: *mut u32) {*shared_ptr = 58;
}pub unsafe fn test_atomic_relaxed(shared_ptr: &AtomicU32) {shared_ptr.store(58, Ordering::Relaxed);
}pub unsafe fn test_atomic_release(shared_ptr: &AtomicU32) {shared_ptr.store(58, Ordering::Release);
}pub unsafe fn test_atomic_consistent(shared_ptr: &AtomicU32) {shared_ptr.store(58, Ordering::SeqCst);
}

如果我们看一下上面的代码生成的 X86 汇编，我们会看到前三个函数产生了相同的代码。直到更加严格的SeqCst次序，我们才得到一个生成的不同的指令集。

example::test_write:mov     dword ptr [rdi], 58retexample::test_atomic_relaxed:mov     dword ptr [rdi], 58retexample::test_atomic_release:mov     dword ptr [rdi], 58retexample::test_atomic_consistent:mov     eax, 58xchg    dword ptr [rdi], eaxret

前面两个次序，使用MOV(MOVe)指令把值写到内存。只有更严格的次序生成了不同的指令，XCHG(atomic eXCHanG)，来对一个原生指针进行写操作。
我们可以和生成的ARM汇编进行比较：

example::test_write:mov     w8, #58str     w8, [x0]retexample::test_atomic_relaxed:mov     w8, #58str     w8, [x0]retexample::test_atomic_release:mov     w8, #58stlr    w8, [x0]retexample::test_atomic_consistent:mov     w8, #58stlr    w8, [x0]ret

和之前相反，在我们达到release次序要求之后可以看到一些不同。原生指针和relax原子存储操作使用STR(STore Register)而release和sequential次序使用指令STLR(STore with reLease Register)。在这段汇编代码里，MOV指令把常量58移动到一个寄存器，它不是一个内存操作。
我们应该能够看出这里的风险，即对程序员的错误更加宽容。对我们而言，在抽象内存模型上写出错误的代码但是让它在某些CPU上产生正确的汇编代码并且正确工作也是有可能的。

使用Atomic写一个多线程程序

我们将要讨论的程序是构建于存储一个指针值是跨线程原子操作这一概念之上的。一个线程将要使用自己拥有的一个可变对象来执行某项任务。一旦它结束了那项任务，它将会以一个不可变的共享引用来发布该任务，使用一个原子指针写入工作完成的信号并且允许读线程使用数据。

仅X86模式下的实现

如果我们真的想要测试X86的内存模型有多么宽容(forgiving 译者注:这里暂未想到更合适的翻译 )，我们可以写一段跳过任意使用了std::sync::atomic模块的代码。我想强调的是，这不是你真正应该考虑做的事情。事实上，由于没有保证避免编译器对指令的重排序，所以这段代码有未定义行为(尽管如此，Rust1.44.1版编译器没有进行"重排序"，所以这段代码可以"工作")。这仅仅是个用作学习的小练习。

pub struct SynchronisedSum {shared: UnsafeCell<*const u32>,samples: usize,
}impl SynchronisedSum {pub fn new(samples: usize) -> Self {assert!(samples < (u32::MAX as usize));Self {shared: UnsafeCell::new(std::ptr::null()),samples,}}pub fn generate(&self) {// do work on data this thread ownslet data: Box<[u32]> = (0..self.samples as u32).collect();// publish to other threadslet shared_ptr = self.shared.get();unsafe {shared_ptr.write_volatile(data.as_ptr());}std::mem::forget(data);}pub fn calculate(&self, expected_sum: u32) {loop {            // check if the work has been published yetlet shared_ptr = self.shared.get();let data_ptr = unsafe { shared_ptr.read_volatile() };if !data_ptr.is_null() {// the data is now accessible by multiple threads, treat it as an immutable reference.let data = unsafe { std::slice::from_raw_parts(data_ptr, self.samples) };let mut sum = 0;for i in (0..self.samples).rev() {sum += data[i];}// did we access the data we expected?assert_eq!(sum, expected_sum);break;}}}
}

计算数组之和的函数从执行一个循环开始，这个循环里会读取共享指针的值。因为我们已知的原子存储保证所以read_volatile()只返回null或者一个指向u32slice的指针。我们不断地进行循环直到生成线程结束并且发布它的工作。一旦它被发布，我们就能读取到它并且计算元素的和。

测试代码

作为一个简单的测试，我们将要同时运行两个线程，一个用来生成值另一个用来计算总和。两个线程执行完各自的工作之后都会退出，我们通过使用join来等待它们退出。

pub fn main() {print_arch();for i in 0..10_000 {let sum_generate = Arc::new(SynchronisedSum::new(512));let sum_calculate = Arc::clone(&sum_generate);let calculate_thread = thread::spawn(move || {sum_calculate.calculate(130816);});thread::sleep(std::time::Duration::from_millis(1));let generate_thread = thread::spawn(move || {sum_generate.generate();});calculate_thread.join().expect(&format!("iteration {} failed", i));generate_thread.join().unwrap();}println!("all iterations passed");
}

如果我在一个Intel的CPU上运行测试，我会得到下面的结果:

running on x86_64
all iterations passed

如果我在一个具有两个核的ARM CPU上运行测试，我会得到:

running on aarch64
thread '<unnamed>' panicked at 'assertion failed: `(left == right)`left: `122824`,right: `130816`', srcmain.rs:45:17
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
thread 'main' panicked at 'iteration 35 failed: Any', srcmain.rs:128:9

X86处理器能够成功运行10000次测试，但是ARM处理器在第35次运行失败了。

哪里出问题了？

在我们执行最后的写入共享指针将其发布给其他线程之前，我们模式的正常运作要求我们正在进行的“工作(work)”在内存中处于正确的状态。
ARM的内存模型不同于X86内存模型的地方在于ARM CPU将会对写入操作进行重排序，而X86不会。所以，计算线程能够看到一个非空(non-null)的指针并且在slice还没被写入之前就开始从其中读取值。
对于我们程序中的大多数内存操作，我们想要给CPU足够的自由来重新整理操作从而使性能最大化。我们只想要指定最小的必要性约束来确保正确性。
至于我们的generate函数，我们想要slice中的值以任意能够带来最快速度的顺序写入内存。但是，所有的写入必须在我们把值写入共享指针之前完成。
在calculate函数上正好相反。我们有一个要求，从slice内存中读取的值至少和共享指针中的值来自相同的时间点。
尽管在对共享指针的读取完成之前不会执行这些指令，但我们需要确保不会从过期的缓存中得到这些值。

正确的版本

为了确保我们代码的正确性，对共享指针的写入必须使用release次序，并且由于calculate的读取顺序要求，我们使用acquire次序。
我们对数据的初始化以及计算总和的代码都没有改变，我们想给CPU足够的自由以最高效的方式来运行。

struct SynchronisedSumFixed {shared: AtomicPtr<u32>,samples: usize,
}impl SynchronisedSumFixed {fn new(samples: usize) -> Self {assert!(samples < (u32::MAX as usize));Self {shared: AtomicPtr::new(std::ptr::null_mut()),samples,}}fn generate(&self) {// do work on data this thread ownslet mut data: Box<[u32]> = (0..self.samples as u32).collect();// publish (aka release) this data to other threadsself.shared.store(data.as_mut_ptr(), Ordering::Release);std::mem::forget(data);}fn calculate(&self, expected_sum: u32) {loop {let data_ptr = self.shared.load(Ordering::Acquire);// when the pointer is non null we have safely acquired a reference to the global dataif !data_ptr.is_null() {let data = unsafe { std::slice::from_raw_parts(data_ptr, self.samples) };let mut sum = 0;for i in (0..self.samples).rev() {sum += data[i];}assert_eq!(sum, expected_sum);break;}}}
}

如果我们在ARM CPU上运行使用了AtomicPtr<u32>更新后的版本，我们会得到:

running on aarch64
all iterations passed

次序的选择

在跨多个CPU进行工作的时候，使用atomic模块仍然需要注意。正如我们看到的X86和ARM汇编代码的输出，如果我们在store上使用Ordering::Relaxed来替换Ordering::Release，我们能回退到一个在x86上正确运行但是在ARM上会失败的版本。使用AtomicPtr尤其需要在最终访问指针指向的值的时候避免未定义行为。