stateful_callback/
stateful_callback.rs

1// Drive acquisition from a callback that mutates shared outer state, instead
2// of a polling loop.
3//
4// openDAQ invokes a reader's "data available" procedure from its own scheduler
5// thread(s) as packets arrive.  A Rust closure handed to openDAQ must therefore
6// be `Send + Sync`, and any state it touches has to be shared safely -- an
7// `Arc` around atomics or a `Mutex`, never a plain captured `&mut`, an `Rc`, or
8// a `RefCell` (those wouldn't even compile here, precisely because the callback
9// can run on another thread).
10//
11// The callback below tallies -- into outer state shared with `main` -- how many
12// times openDAQ signalled that new samples were ready, and which threads it was
13// called from.  The main thread reads the samples and, afterwards, the
14// accumulated state back.
15
16use std::collections::BTreeSet;
17use std::sync::atomic::{AtomicUsize, Ordering};
18use std::sync::{Arc, Mutex};
19use std::time::Duration;
20
21use opendaq::{Channel, Instance, Procedure, StreamReader};
22
23fn main() -> opendaq::Result<()> {
24    let instance = Instance::new()?;
25    let device = instance.add_device("daqref://device0")?.expect("device");
26    let channel = instance
27        .find_component("Dev/RefDev0/IO/AI/RefCh0")?
28        .expect("reference channel not found")
29        .cast::<Channel>()?;
30    device.set_property_value("GlobalSampleRate", 1000)?;
31    let signal = &channel.signals()?[0];
32
33    let reader = StreamReader::<f64>::new(signal)?;
34
35    // The outer state the callback mutates, shared with the main thread: an
36    // atomic counter of notifications, and the set of thread ids we were
37    // called from.  Both live behind `Arc` so the closure and `main` share
38    // one instance.
39    let notifications = Arc::new(AtomicUsize::new(0));
40    let caller_threads = Arc::new(Mutex::new(BTreeSet::<String>::new()));
41
42    // Clone the Arc handles into the closure.  It captures only these (not the
43    // reader), so there is no ownership cycle, and it is `move` + Send + Sync.
44    let notifications_cb = Arc::clone(&notifications);
45    let caller_threads_cb = Arc::clone(&caller_threads);
46    let on_ready = Procedure::from_fn(move |_args| {
47        notifications_cb.fetch_add(1, Ordering::Relaxed);
48        let thread = format!("{:?}", std::thread::current().id());
49        caller_threads_cb.lock().unwrap().insert(thread);
50        Ok(())
51    })?;
52    reader.set_on_data_available(&on_ready)?;
53
54    // Let the simulator stream for a moment.  The callback fires in the
55    // background as packets arrive; here on the main thread we drain the
56    // samples and keep a running total to show acquisition really is flowing.
57    let mut total_samples = 0usize;
58    let mut running_sum = 0.0f64;
59    for _ in 0..10 {
60        let samples = reader.read(1000, 200)?;
61        total_samples += samples.sample_count();
62        running_sum += samples.iter().sum::<f64>();
63        std::thread::sleep(Duration::from_millis(50));
64    }
65
66    // Read the accumulated outer state back on the main thread.
67    let fired = notifications.load(Ordering::Relaxed);
68    let caller_threads = caller_threads.lock().unwrap();
69    let main_thread = format!("{:?}", std::thread::current().id());
70    let mean = if total_samples > 0 {
71        running_sum / total_samples as f64
72    } else {
73        0.0
74    };
75
76    println!("Read {total_samples} samples (mean {mean:.4}).");
77    println!("The data-ready callback fired {fired} time(s) while we read.");
78    println!("Main thread: {main_thread}");
79    println!("Callback ran on:  {caller_threads:?}");
80    if caller_threads.iter().any(|t| *t != main_thread) {
81        println!(
82            "The callback ran on an openDAQ scheduler thread -- which is why its \
83             closure must be Send + Sync and its state shared through Arc/atomics."
84        );
85    } else {
86        println!(
87            "openDAQ happened to notify us on the calling thread this run, but it is \
88             free to use its own worker threads, so the Send + Sync / shared-state \
89             contract still applies."
90        );
91    }
92
93    Ok(())
94}
stateful_callback/stateful_callback.rs

stateful_callback/
stateful_callback.rs
