addr_monitor

This crate is for doing one thing and one thing only; safely monitoring an address for updates.

It exposes hardware monitors on x86_64 via x86_intr with fallbacks using spin loops.

Usage

There are two modules to use:

• hardware: exposes HardwareMonitor for directly using x86_64 userspace hardware monitors on Intel and AMD
• blocking: synchronization primitives built on top of the hardware monitor like CAS loops and conditional waits

To use them in code:

swap_on_write is nothing more than a CAS loop.

let value = Arc::new(AtomicI32::new(0));
let barrier = Arc::new(Barrier::new(WRITERS as usize));

let mut handles = Vec::new();

for new_value in 1..=WRITERS {
    let value = Arc::clone(&value);
    let barrier = Arc::clone(&barrier);

    handles.push(thread::spawn(move || {
        barrier.wait();

        let previous = swap_on_write(value.as_ref(), new_value);

        (new_value, previous)
    }));
}

let results: Vec<(i32, i32)> = handles
    .into_iter()
    .map(|handle| handle.join().unwrap())
    .collect();

let final_value = value.load(Ordering::Acquire);

assert!((1..=WRITERS).contains(&final_value));

assert_eq!(
    results.iter().filter(|(_, previous)| *previous == 0).count(),
    1
);

for (_, previous) in &results {
    assert!(
        *previous == 0 || (1..=WRITERS).contains(previous),
        "unexpected previous value: {previous}"
    );
}

where WRITERS is a constant i32.

cond_wait_eq and cond_wait_ne block the thread until the atomic value is equal to or not equal to the desired value, respectively.

// cond_wait_eq
let value = Arc::new(AtomicI32::new(42));
let watched = Arc::clone(&value);

let (started_tx, started_rx) = mpsc::channel();
let (finished_tx, finished_rx) = mpsc::channel();

let handle = thread::spawn(move || {
    started_tx.send(()).unwrap();

    cond_wait_eq(watched.as_ref(), 43);

    finished_tx.send(()).unwrap();
});

started_rx.recv().unwrap();

assert!(finished_rx.try_recv().is_err());

value.store(43, Ordering::Release);

finished_rx
    .recv_timeout(Duration::from_secs(1))
    .unwrap();

handle.join().unwrap();

// cond_wait_ne

let value = Arc::new(AtomicI32::new(42));
let watched = Arc::clone(&value);

let (started_tx, started_rx) = mpsc::channel();
let (finished_tx, finished_rx) = mpsc::channel();

let handle = thread::spawn(move || {
    started_tx.send(()).unwrap();

    cond_wait_ne(watched.as_ref(), 42);

    finished_tx.send(()).unwrap();
});

started_rx.recv().unwrap();

assert!(finished_rx.try_recv().is_err());

value.store(43, Ordering::Release);

finished_rx
    .recv_timeout(Duration::from_secs(1))
    .unwrap();

handle.join().unwrap();

block_on_write is another conditional wait but waits until the atomic value has changed from its starting value when it's called:

let value = Arc::new(AtomicI32::new(123));
let watched = Arc::clone(&value);

let handle = thread::spawn(move || {
    block_on_write(watched.as_ref());
});

thread::sleep(Duration::from_millis(50));
value.store(123, Ordering::Relaxed);

thread::sleep(Duration::from_millis(50));
assert!(!handle.is_finished());

value.store(124, Ordering::Relaxed);

handle.join().unwrap();