visa_rs/async_io/
mod.rs

1//! Asynchronous reads and writes for a VISA instrument.
2//!
3//! Use [`AsyncInstrument::async_read`] to read a response, [`AsyncInstrument::async_read_exact`]
4//! for a fixed byte count, and [`AsyncInstrument::async_write`] to write data.
5//! These futures wait for transfer completion; dropping one requests cancellation.
6//!
7//! Tests with NI-VISA 23.5 indicate that async I/O on macOS may block until timeout.
8
9// VISA retains buffer pointers until IO completion, so transfers need owned storage.
10// Cancellation alone does not prove the driver has stopped accessing that storage.
11// Completions may arrive before submission returns a job ID; JobMap handles both orders.
12
13mod callback;
14mod future;
15
16pub use future::{AsyncRead, AsyncWrite};
17
18use callback::{Completed, JobMap};
19
20use crate::{
21    enums::{attribute, event, status::CompletionCode},
22    session::AsRawSs,
23    wrap_raw_error_in_unsafe, Instrument, JobID, Result,
24};
25#[cfg(feature = "tokio")]
26use std::task::Waker;
27use std::{
28    task::{Context, Poll},
29    time::Duration,
30};
31use visa_sys as vs;
32
33/// An [`Instrument`] wired up for asynchronous transfers.
34pub struct AsyncInstrument {
35    // Field order matters: close the session before dropping the callback pack.
36    // Unfinished buffers are leaked if completion was never reported; the entire
37    // pack is leaked if uninstall fails.
38    pub(super) instr: Instrument,
39    callback: Option<Box<JobMap>>,
40}
41
42impl From<AsyncInstrument> for Instrument {
43    fn from(mut async_instr: AsyncInstrument) -> Self {
44        // The session outlives this conversion, so the handler has to come off before
45        // the callback pack is freed, otherwise visa would keep a dangling `user_data`.
46        async_instr.uninstall_handler();
47        let async_instr = std::mem::ManuallyDrop::new(async_instr);
48        // SAFETY: `async_instr` is never dropped, so reading both fields out of it moves
49        // them exactly once. The handler is uninstalled or its pack deliberately leaked,
50        // which is all `Drop` does.
51        let instr = unsafe { std::ptr::read(&async_instr.instr) };
52        let callback = unsafe { std::ptr::read(&async_instr.callback) };
53        drop(callback);
54        instr
55    }
56}
57
58impl AsyncInstrument {
59    /// Wraps `instr` for asynchronous use.
60    ///
61    /// Enables termination-character and END detection. These settings remain in
62    /// effect when converted back into an [`Instrument`].
63    pub fn new(instr: Instrument) -> Result<Self> {
64        use crate::enums::attribute::HasAttribute;
65        instr.set_attr(attribute::AttrTermcharEn::VI_TRUE)?;
66        instr.set_attr(attribute::AttrSuppressEndEn::VI_FALSE)?;
67        let callback = Box::new(JobMap::new());
68        wrap_raw_error_in_unsafe!(vs::viInstallHandler(
69            instr.as_raw_ss(),
70            event::EventKind::EventIoCompletion as _,
71            Some(JobMap::call_in_c),
72            &*callback as *const JobMap as *mut std::ffi::c_void,
73        ))?;
74        // From this point on, even an enable failure must uninstall the handler
75        // before releasing its user-data allocation.
76        let async_instr = Self {
77            instr,
78            callback: Some(callback),
79        };
80        async_instr.instr.enable_event(
81            event::EventKind::EventIoCompletion,
82            event::Mechanism::Handler,
83        )?;
84        Ok(async_instr)
85    }
86
87    fn callback(&self) -> &JobMap {
88        self.callback
89            .as_deref()
90            .expect("callback present until teardown")
91    }
92
93    pub fn instrument(&self) -> &Instrument {
94        &self.instr
95    }
96
97    /// Reads one complete response, resolving to the bytes received.
98    ///
99    /// Stops at the termination character or END indicator. `timeout` applies to each
100    /// underlying read, not the total time to receive the response.
101    ///
102    /// For binary block data, whose bytes may legitimately include the termination
103    /// character, use [`Self::async_read_exact`] instead.
104    ///
105    /// Starting another operation while one is in progress may return
106    /// [`ErrorInProgress`](crate::enums::status::ErrorCode::ErrorInProgress).
107    pub fn async_read(&self, timeout: Duration) -> AsyncRead<'_> {
108        AsyncRead::new(self, None, timeout)
109    }
110
111    /// Reads exactly `len` bytes, resolving to them.
112    ///
113    /// Disables termination-character detection for binary data. Short reads are
114    /// continued until `len` bytes arrive; a zero-byte read before then returns
115    /// [`ErrorIo`](crate::enums::status::ErrorCode::ErrorIo).
116    /// `timeout` applies to each underlying read, not the total operation.
117    /// Read settings persist on the session. VISA errors and timeouts are returned immediately.
118    pub fn async_read_exact(&self, len: usize, timeout: Duration) -> AsyncRead<'_> {
119        AsyncRead::new(self, Some(len), timeout)
120    }
121
122    /// Writes `buf`, waiting for completion and returning the number of bytes written.
123    ///
124    /// Accepts inputs convertible to `Vec<u8>`, including byte slices, `Vec<u8>`, and
125    /// `String`. Check the returned count for a short write.
126    pub fn async_write(&self, buf: impl Into<Vec<u8>>) -> AsyncWrite<'_> {
127        AsyncWrite::new(self, buf.into())
128    }
129
130    // These session attributes persist after the transfer and affect concurrent operations.
131    pub(super) fn prepare_read(&self, timeout: Duration, termchar: bool) -> Result<()> {
132        use crate::enums::attribute::HasAttribute;
133        let ms = timeout.as_millis().min(vs::VI_TMO_INFINITE as u128) as _;
134        let tmo = attribute::AttrTmoValue::new_checked(ms).ok_or(crate::Error(
135            crate::enums::status::ErrorCode::ErrorNsupAttrState,
136        ))?;
137        self.instr.set_attr(tmo)?;
138        self.instr.set_attr(if termchar {
139            attribute::AttrTermcharEn::VI_TRUE
140        } else {
141            attribute::AttrTermcharEn::VI_FALSE
142        })?;
143        Ok(())
144    }
145
146    // Give the callback pack ownership until completion. Submission failure drops buf.
147    #[cfg(feature = "tokio")]
148    pub(crate) fn start_read_id(&self, mut buf: Vec<u8>, waker: &Waker) -> Result<JobID> {
149        let (job_id, completion) = unsafe { self.instr.visa_read_async(buf.as_mut())? };
150        self.callback().register(job_id, waker, buf);
151        note_sync_completion(completion, job_id);
152        Ok(job_id)
153    }
154
155    // Transfer buffer ownership as in start_read_id.
156    #[cfg(feature = "tokio")]
157    pub(crate) fn start_write_id(&self, buf: Vec<u8>, waker: &Waker) -> Result<JobID> {
158        let (job_id, completion) = unsafe { self.instr.visa_write_async(&buf)? };
159        self.callback().register(job_id, waker, buf);
160        note_sync_completion(completion, job_id);
161        Ok(job_id)
162    }
163
164    pub(crate) fn poll_job(
165        &self,
166        job_id: JobID,
167        cx: &Context<'_>,
168    ) -> Poll<(Result<Completed>, Vec<u8>)> {
169        self.callback()
170            .poll_job(job_id, cx)
171            .map(|(result, buf)| (result, buf.unwrap_or_default()))
172    }
173
174    // Keep the buffer until completion: viTerminate only requests cancellation.
175    pub(crate) fn cancel_job(&self, job_id: JobID) {
176        self.callback().cancel(job_id);
177        if let Err(e) = wrap_raw_error_in_unsafe!(vs::viTerminate(
178            self.instr.as_raw_ss(),
179            vs::VI_NULL as _,
180            job_id.0
181        )) {
182            log::warn!("terminating unfinished async io: {}", e)
183        };
184    }
185    // Remove VISA's user_data reference before freeing the pack or returning the session.
186    fn uninstall_handler(&mut self) {
187        let Some(callback) = self.callback.as_deref() else {
188            log::error!("cannot uninstall IO completion handler: callback state is missing");
189            return;
190        };
191        if let Err(e) = self.instr.disable_event(
192            event::EventKind::EventIoCompletion,
193            event::Mechanism::Handler,
194        ) {
195            log::warn!("error disabling io completion event: {}", e)
196        };
197        // Derived from a shared reference on purpose: visa only compares this pointer
198        // against the installed entry, and a `&mut` here would both assert uniqueness
199        // against an in-flight callback and invalidate the pointer visa still holds.
200        let user_data = callback as *const JobMap as *mut std::ffi::c_void;
201        if let Err(e) = wrap_raw_error_in_unsafe!(vs::viUninstallHandler(
202            self.instr.as_raw_ss(),
203            event::EventKind::EventIoCompletion as _,
204            Some(JobMap::call_in_c),
205            user_data,
206        )) {
207            // VISA may still hold user_data. Leak before logging so even a panicking
208            // logger cannot cause us to release memory still reachable by the driver.
209            if let Some(callback) = self.callback.take() {
210                let _ = Box::leak(callback);
211            }
212            log::warn!(
213                "failed to uninstall IO completion handler: {}; leaked callback state to keep VISA's user-data pointer valid",
214                e
215            );
216        };
217    }
218}
219
220impl Drop for AsyncInstrument {
221    fn drop(&mut self) {
222        self.uninstall_handler();
223    }
224}
225
226// VI_SUCCESS_SYNC may invoke the handler before submission returns. The completion
227// count still comes from the event; JobMap parks early results for registration.
228fn note_sync_completion(completion: CompletionCode, job_id: JobID) {
229    if completion == CompletionCode::SuccessSync {
230        log::trace!("async io for job {} completed synchronously", job_id.0);
231    }
232}
visa_rs/async_io/mod.rs

visa_rs/async_io/
mod.rs
