MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Rust 原子加载与存储操作的协同工作

2024-06-234.0k 阅读

Rust 原子加载与存储操作基础

在 Rust 中，原子类型提供了一种线程安全的方式来在多个线程间共享数据。原子加载（load）和存储（store）操作是这些原子类型的核心操作，它们协同工作以确保数据的一致性和线程安全。

原子类型定义在 std::sync::atomic 模块中。例如，AtomicI32 是一个可以原子操作的 32 位整数类型。

原子存储操作

原子存储操作将一个值写入到原子类型的实例中。这个操作是原子的，意味着它不会被其他线程的操作打断。在 Rust 中，原子存储操作通过 store 方法实现。

use std::sync::atomic::{AtomicI32, Ordering};

fn main() {
    let atomic_num = AtomicI32::new(0);
    atomic_num.store(42, Ordering::SeqCst);
}

在上述代码中，我们创建了一个 AtomicI32 实例 atomic_num 并初始化为 0。然后，我们使用 store 方法将值 42 存储到 atomic_num 中。Ordering::SeqCst 是一个内存排序参数，它保证了这个存储操作在全局顺序上是可见的。

原子加载操作

原子加载操作从原子类型的实例中读取一个值。同样，这个操作也是原子的，以确保读取到的值是一致的。在 Rust 中，原子加载操作通过 load 方法实现。

use std::sync::atomic::{AtomicI32, Ordering};

fn main() {
    let atomic_num = AtomicI32::new(42);
    let value = atomic_num.load(Ordering::SeqCst);
    println!("Loaded value: {}", value);
}

这里，我们先创建一个值为 42 的 AtomicI32 实例 atomic_num。然后使用 load 方法读取其值，并打印出来。同样，Ordering::SeqCst 确保了加载操作的顺序一致性。

内存排序与原子操作协同

内存排序是原子加载和存储操作协同工作的关键。Rust 中的 Ordering 枚举定义了不同的内存排序规则，这些规则决定了原子操作如何与其他内存操作交互。

顺序一致性（SeqCst）

Ordering::SeqCst 提供了最强的内存排序保证。所有使用 SeqCst 的原子操作在全局顺序上是一致的。这意味着所有线程都能看到这些操作以相同的顺序发生。

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

fn main() {
    let atomic_num = Arc::new(AtomicI32::new(0));
    let atomic_num_clone = atomic_num.clone();

    let handle = thread::spawn(move || {
        atomic_num_clone.store(42, Ordering::SeqCst);
    });

    handle.join().unwrap();
    let value = atomic_num.load(Ordering::SeqCst);
    println!("Loaded value: {}", value);
}

在这个例子中，主线程创建了一个 AtomicI32 实例，并将其克隆后传递给一个新线程。新线程使用 SeqCst 存储值 42，主线程使用 SeqCst 加载值。由于 SeqCst 的保证，主线程一定能加载到 42。

释放 - 获得（Release - Acquire）排序

Ordering::Release 用于存储操作，Ordering::Acquire 用于加载操作，它们共同提供了一种较弱但更高效的内存排序保证。使用 Release 存储操作的线程保证所有之前的写操作对其他使用 Acquire 加载操作的线程可见。

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

fn main() {
    let atomic_num = Arc::new(AtomicI32::new(0));
    let atomic_num_clone = atomic_num.clone();

    let data = Arc::new(Mutex::new(vec![1, 2, 3]));
    let data_clone = data.clone();

    let handle = thread::spawn(move || {
        let mut data_inner = data_clone.lock().unwrap();
        data_inner.push(4);
        atomic_num_clone.store(42, Ordering::Release);
    });

    atomic_num.load(Ordering::Acquire);
    let data_inner = data.lock().unwrap();
    println!("Data: {:?}", data_inner);

    handle.join().unwrap();
}

在这个例子中，新线程先修改了一个 Mutex 保护的向量，然后使用 Release 存储操作更新 AtomicI32。主线程先使用 Acquire 加载 AtomicI32，然后访问 Mutex 保护的向量。由于 Release - Acquire 排序，主线程能看到新线程对向量的修改。

Relaxed 排序

Ordering::Relaxed 提供了最弱的内存排序保证。它只保证原子操作本身的原子性，不提供任何跨线程的内存可见性保证。这种排序在只需要保证原子性而不需要严格内存顺序的场景下非常有用，例如原子计数器。

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

fn main() {
    let atomic_counter = AtomicI32::new(0);
    let mut handles = Vec::new();

    for _ in 0..10 {
        let atomic_counter_clone = atomic_counter.clone();
        let handle = thread::spawn(move || {
            for _ in 0..100 {
                atomic_counter_clone.fetch_add(1, Ordering::Relaxed);
            }
        });
        handles.push(handle);
    }

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

    let final_value = atomic_counter.load(Ordering::Relaxed);
    println!("Final counter value: {}", final_value);
}

在这个例子中，多个线程使用 Relaxed 排序的 fetch_add 操作对 AtomicI32 进行递增。由于 Relaxed 排序，不同线程对计数器的修改顺序在全局上是不确定的，但最终计数器的值是正确的，因为每个 fetch_add 操作是原子的。

原子加载与存储在多线程编程中的应用

共享状态管理

在多线程环境中，原子加载和存储操作常用于管理共享状态。例如，在一个多线程的服务器应用中，可能需要共享一个全局的计数器来统计请求数量。

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

fn main() {
    let request_counter = AtomicI32::new(0);
    let mut handles = Vec::new();

    for _ in 0..10 {
        let request_counter_clone = request_counter.clone();
        let handle = thread::spawn(move || {
            for _ in 0..100 {
                request_counter_clone.fetch_add(1, Ordering::Relaxed);
            }
        });
        handles.push(handle);
    }

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

    let total_requests = request_counter.load(Ordering::Relaxed);
    println!("Total requests: {}", total_requests);
}

在这个例子中，多个线程模拟处理请求，每次处理请求时使用 fetch_add（基于原子存储操作）递增计数器。最后，主线程使用原子加载操作获取总的请求数量。

线程同步

原子加载和存储操作还可以用于线程同步。例如，一个线程可能需要等待另一个线程完成某个任务后才能继续执行。

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

fn main() {
    let task_completed = AtomicBool::new(false);
    let task_completed_clone = task_completed.clone();

    let handle = thread::spawn(move || {
        // 模拟任务执行
        thread::sleep(std::time::Duration::from_secs(2));
        task_completed_clone.store(true, Ordering::Release);
    });

    while!task_completed.load(Ordering::Acquire) {
        thread::yield_now();
    }

    println!("Task is completed. Proceeding...");
    handle.join().unwrap();
}

在这个例子中，一个线程负责执行任务，完成后使用 Release 存储操作将 AtomicBool 设置为 true。主线程使用 Acquire 加载操作不断检查任务是否完成，直到任务完成才继续执行。

高级原子加载与存储操作协同

条件存储（Compare - And - Swap，CAS）

条件存储操作，通常称为 Compare - And - Swap（CAS），是一种原子操作，它将一个原子变量的值与一个预期值比较。如果相等，则将原子变量的值更新为一个新值。这个操作在多线程环境中用于实现无锁数据结构。

use std::sync::atomic::{AtomicI32, Ordering};

fn main() {
    let atomic_num = AtomicI32::new(42);
    let expected = 42;
    let new_value = 43;

    let result = atomic_num.compare_and_swap(expected, new_value, Ordering::SeqCst);
    if result == expected {
        println!("Value was successfully updated to {}", new_value);
    } else {
        println!("Value was not updated. Expected {}, but got {}", expected, result);
    }
}

在这个例子中，我们使用 compare_and_swap 方法尝试将 AtomicI32 的值从 42 更新为 43。如果当前值确实是 42（预期值），则更新成功并返回旧值 42；否则，返回当前值（非预期值）。

读 - 修改 - 写操作

原子类型还支持读 - 修改 - 写操作，如 fetch_add、fetch_sub 等。这些操作结合了原子加载和存储操作，在多线程环境中非常有用。

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

fn main() {
    let atomic_num = AtomicI32::new(0);
    let mut handles = Vec::new();

    for _ in 0..10 {
        let atomic_num_clone = atomic_num.clone();
        let handle = thread::spawn(move || {
            for _ in 0..100 {
                atomic_num_clone.fetch_add(1, Ordering::Relaxed);
            }
        });
        handles.push(handle);
    }

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

    let final_value = atomic_num.load(Ordering::Relaxed);
    println!("Final value: {}", final_value);
}

在这个例子中，多个线程使用 fetch_add 操作对 AtomicI32 进行递增。fetch_add 操作首先原子地加载当前值，增加 1，然后原子地存储新值。

原子加载与存储操作的性能考量

不同内存排序的性能差异

不同的内存排序规则对性能有显著影响。SeqCst 提供了最强的内存排序保证，但也是最昂贵的，因为它需要在全局范围内同步所有原子操作。Relaxed 排序提供了最弱的保证，因此性能最好，但仅适用于不需要跨线程内存可见性保证的场景。Release - Acquire 排序则在保证一定内存可见性的同时，提供了较好的性能。

原子操作的开销

原子操作本身比普通的非原子操作有更高的开销。这是因为原子操作需要硬件支持来保证其原子性和内存排序。在性能敏感的应用中，应尽量减少不必要的原子操作，仅在需要线程安全的地方使用原子类型和操作。

优化建议

  1. 使用合适的内存排序：根据应用的需求选择合适的内存排序规则。如果不需要严格的全局顺序，使用 Release - Acquire 或 Relaxed 排序可以提高性能。
  2. 批量操作：尽量将多个原子操作合并为一个操作。例如，使用 compare_and_swap 而不是单独的加载、比较和存储操作。
  3. 减少竞争：通过合理的设计减少多个线程对同一原子变量的竞争，例如使用锁分段或无锁数据结构。

与其他并发原语的协同

与 Mutex 的协同

虽然原子操作提供了一种轻量级的线程安全方式，但在某些情况下，它们需要与其他并发原语（如 Mutex）协同工作。Mutex 提供了互斥访问，而原子操作可以在 Mutex 保护的临界区内提供更细粒度的线程安全操作。

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

fn main() {
    let data = Arc::new(Mutex::new(vec![AtomicI32::new(0); 10]));
    let data_clone = data.clone();

    let handle = thread::spawn(move || {
        let mut data_inner = data_clone.lock().unwrap();
        for atomic_num in data_inner.iter_mut() {
            atomic_num.fetch_add(1, Ordering::Relaxed);
        }
    });

    let mut data_inner = data.lock().unwrap();
    for atomic_num in data_inner.iter() {
        println!("Value: {}", atomic_num.load(Ordering::Relaxed));
    }

    handle.join().unwrap();
}

在这个例子中，Mutex 保护一个包含多个 AtomicI32 的向量。线程在 Mutex 保护的临界区内对 AtomicI32 进行原子操作，这样既保证了向量整体的线程安全访问，又利用了原子操作的细粒度线程安全性。

与 Channel 的协同

在基于通道（std::sync::mpsc）的并发模型中，原子操作可以用于在通道的发送和接收端之间传递状态信息。

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{channel, Sender};
use std::thread;

fn main() {
    let (tx, rx): (Sender<bool>, _) = channel();
    let task_completed = AtomicBool::new(false);
    let task_completed_clone = task_completed.clone();

    let handle = thread::spawn(move || {
        // 模拟任务执行
        thread::sleep(std::time::Duration::from_secs(2));
        task_completed_clone.store(true, Ordering::Release);
        tx.send(true).unwrap();
    });

    while!task_completed.load(Ordering::Acquire) {
        thread::yield_now();
    }

    if rx.recv().unwrap() {
        println!("Task is completed.");
    }

    handle.join().unwrap();
}

在这个例子中，一个线程通过通道发送任务完成的信号，同时使用原子操作更新 AtomicBool 表示任务完成状态。接收端通过原子加载操作检查任务是否完成，并接收通道信号确认任务完成。

原子加载与存储操作的常见错误与陷阱

错误的内存排序使用

使用错误的内存排序规则可能导致程序出现难以调试的并发问题。例如，在需要严格顺序一致性的场景下使用 Relaxed 排序，可能导致线程看到不一致的数据。

忘记原子性

在多线程环境中，忘记对共享数据使用原子操作可能导致数据竞争和未定义行为。例如，直接在多个线程间共享非原子的整数类型并进行读写操作。

过度使用原子操作

虽然原子操作提供了线程安全，但过度使用它们会增加性能开销。在不需要线程安全的地方使用原子操作，或者在可以使用更粗粒度同步机制（如 Mutex）的地方使用原子操作，都是不必要的。

总结

Rust 的原子加载和存储操作是多线程编程中的重要工具，它们与内存排序规则紧密结合，为开发者提供了灵活且安全的并发编程方式。通过合理选择内存排序、优化原子操作的使用，并与其他并发原语协同工作，开发者可以构建高效且线程安全的应用程序。同时，要注意避免常见的错误和陷阱，以确保程序的正确性和性能。无论是简单的共享状态管理，还是复杂的无锁数据结构实现，原子加载和存储操作都扮演着关键的角色。