asyn_rs/request.rs
1//! Request types for the port actor.
2
3use std::sync::Arc;
4use std::sync::atomic::{AtomicU8, Ordering as AtomicOrdering};
5use std::time::SystemTime;
6
7use crate::error::AsynStatus;
8use crate::param::ParamValue;
9
10/// A param value to set directly in the store (no writeInt32/on_param_change).
11/// Mirrors C ADCore's setIntegerParam/setDoubleParam.
12///
13/// The value is a [`ParamValue`], not one variant per type: this carrier used to
14/// enumerate its own subset of the parameter types (Int32/Float64/Octet/
15/// Int32Array/Float64Array/UInt32Digital), so a driver thread pushing any other
16/// supported type — `Int64`, `Int8Array`, `Int16Array`, `Int64Array`,
17/// `Float32Array`, `Enum`, `GenericPointer` — simply had no variant to put it in.
18/// Carrying the store's own value type means the two type sets cannot drift
19/// apart again: the actor applies it through the single [`crate::param::ParamList::set_value`]
20/// dispatch, whose exhaustive match makes a new `ParamValue` variant a compile
21/// error rather than an update that never arrives.
22#[derive(Debug, Clone)]
23pub enum ParamSetValue {
24 /// Set a parameter of any type (C `setIntegerParam` / `setDoubleParam` /
25 /// `setStringParam` / `doCallbacksXxxArray` …).
26 Value {
27 reason: usize,
28 addr: i32,
29 value: ParamValue,
30 },
31 /// asynUInt32Digital masked set — C `setUIntDigitalParam(reason, value,
32 /// mask, interruptMask)`, whose write mask and forced-callback mask a plain
33 /// [`Self::Value`] has no room for.
34 UInt32Digital {
35 reason: usize,
36 addr: i32,
37 value: u32,
38 mask: u32,
39 /// Bits to force into the I/O Intr callback mask even when the
40 /// stored value is unchanged (C `setUIntDigitalParam(..,
41 /// interruptMask)`); `0` for a plain value set.
42 interrupt_mask: u32,
43 },
44 /// C `setParamStatus` / `setParamAlarmStatus` / `setParamAlarmSeverity`
45 /// from a background thread — the alarm-push half of the C pattern
46 /// `lock(); setParamStatus(..); callParamCallbacks(); unlock()`. The
47 /// value is untouched; a status/alarm transition alone marks the param
48 /// changed so the flush delivers it (paramVal.cpp:71-104), and the
49 /// devEpics fill-in maps a non-Success `status` to the EPICS alarm
50 /// (asynDisconnected → COMM/INVALID) when `alarm_status`/`alarm_severity`
51 /// are left 0 (asynEpicsUtils.c:238-265).
52 Status {
53 reason: usize,
54 addr: i32,
55 status: AsynStatus,
56 alarm_status: u16,
57 alarm_severity: u16,
58 },
59}
60
61impl ParamSetValue {
62 /// Set `reason` at `addr` to `value` — any parameter type.
63 pub fn new(reason: usize, addr: i32, value: ParamValue) -> Self {
64 Self::Value {
65 reason,
66 addr,
67 value,
68 }
69 }
70
71 /// C `setUIntDigitalParam`: store `value & mask`, and force `interrupt_mask`
72 /// bits into the callback mask even when the stored value is unchanged.
73 pub fn uint32_digital(
74 reason: usize,
75 addr: i32,
76 value: u32,
77 mask: u32,
78 interrupt_mask: u32,
79 ) -> Self {
80 Self::UInt32Digital {
81 reason,
82 addr,
83 value,
84 mask,
85 interrupt_mask,
86 }
87 }
88
89 /// C `setParamStatus(list, index, status)` plus the alarm pair: set the
90 /// transport status and EPICS alarm without touching the value. Pass
91 /// `alarm_status`/`alarm_severity` 0 to let the record-side fill-in map
92 /// `status` itself (the C-normal path).
93 pub fn status(
94 reason: usize,
95 addr: i32,
96 status: AsynStatus,
97 alarm_status: u16,
98 alarm_severity: u16,
99 ) -> Self {
100 Self::Status {
101 reason,
102 addr,
103 status,
104 alarm_status,
105 alarm_severity,
106 }
107 }
108
109 /// The address list this update lands in — the list whose changed flags a
110 /// later `callParamCallbacks` has to consume for it to reach a record.
111 pub fn addr(&self) -> i32 {
112 match self {
113 Self::Value { addr, .. }
114 | Self::UInt32Digital { addr, .. }
115 | Self::Status { addr, .. } => *addr,
116 }
117 }
118}
119
120/// Operation the worker thread will dispatch to the port driver.
121#[derive(Debug, Clone)]
122pub enum RequestOp {
123 OctetWrite {
124 data: Vec<u8>,
125 },
126 OctetRead {
127 buf_size: usize,
128 },
129 OctetWriteRead {
130 data: Vec<u8>,
131 buf_size: usize,
132 /// Whether to drain the driver's input buffer before the write.
133 /// `true` = `asynOctetSyncIO::writeRead` (flush → write → read), the
134 /// StreamDevice/asynRecord pattern that discards stale warm-line bytes.
135 /// `false` = `devAsynOctet` raw write-then-read (no flush) — the
136 /// command-response dset (`callbackSiCmdResponse`) returns whatever the
137 /// device sends, including bytes already in the buffer.
138 flush: bool,
139 },
140 /// Binary octet write: writes `data` raw with the driver's output EOS
141 /// temporarily suppressed. C parity: asynRecord binary output
142 /// (`asynRecord.c:1528-1541`) saves the current output EOS, sets it to
143 /// NULL for the write, and restores it. The actor performs the
144 /// save/clear/restore atomically under its serial ownership so the EOS
145 /// is restored on every exit path.
146 OctetWriteBinary {
147 data: Vec<u8>,
148 },
149 /// Binary octet read: reads with the driver's input EOS temporarily
150 /// suppressed. C parity: asynRecord binary input
151 /// (`asynRecord.c:1564-1577`) saves the current input EOS, sets it to
152 /// NULL for the read, and restores it. The actor brackets the read so
153 /// the EOS is restored on every exit path.
154 OctetReadBinary {
155 buf_size: usize,
156 },
157 Int32Write {
158 value: i32,
159 },
160 Int32Read,
161 Int64Write {
162 value: i64,
163 },
164 Int64Read,
165 Float64Write {
166 value: f64,
167 },
168 Float64Read,
169 UInt32DigitalWrite {
170 value: u32,
171 mask: u32,
172 },
173 UInt32DigitalRead {
174 mask: u32,
175 },
176 Flush,
177 /// Connect to the port (bypass enabled/connected checks).
178 Connect,
179 /// Disconnect from the port (bypass enabled/connected checks).
180 Disconnect,
181 /// Permanently shut down a `ASYN_DESTRUCTIBLE` port. C parity:
182 /// `asynManager.c::shutdownPort` (lines 2251-2308). Marks the
183 /// port defunct so every subsequent request short-circuits;
184 /// idempotent; broadcasts `AsynException::Shutdown`.
185 ShutdownPort,
186 /// Connect a specific device address (multi-device ports).
187 ConnectAddr,
188 /// Disconnect a specific device address (multi-device ports).
189 DisconnectAddr,
190 /// Enable a specific device address (multi-device ports).
191 EnableAddr,
192 /// Disable a specific device address (multi-device ports).
193 DisableAddr,
194 /// Enable / disable the entire port. C parity:
195 /// `pasynManager->enable(pasynUser, enable)`
196 /// (`asynManager.c::enable`, fired by asynRecord `ENBL` writes
197 /// at `asynRecord.c:484-486`).
198 SetEnable {
199 yes: bool,
200 },
201 /// Enable / disable auto-connect for the port. C parity:
202 /// `pasynManager->autoConnect(pasynUser, autoConnect)`
203 /// (`asynManager.c::autoConnect`, fired by asynRecord `AUCT`
204 /// writes at `asynRecord.c:481-482`). `asynExceptionAutoConnect`
205 /// is emitted unconditionally on every call.
206 SetAutoConnect {
207 yes: bool,
208 },
209 /// Enable / disable auto-connect for ONE device of a multi-device port —
210 /// the `user.addr` variant of [`RequestOp::SetAutoConnect`], symmetric with
211 /// [`RequestOp::EnableAddr`]. C reaches both through a single
212 /// `pasynManager->autoConnect`, which picks device-vs-port state via
213 /// `findDpCommon` (asynManager.c:496-509, 2314); the caller of the shell
214 /// command `asynAutoConnect portName addr yesNo` is what supplies the addr.
215 SetAutoConnectAddr {
216 yes: bool,
217 },
218 /// Query int32 bounds (low, high).
219 GetBoundsInt32,
220 /// Query int64 bounds (low, high).
221 GetBoundsInt64,
222 /// Query whether the port is currently enabled. C parity:
223 /// `pasynManager->isEnabled` (`asynManager.c`).
224 GetEnable,
225 /// Query whether auto-connect is enabled for the port. C parity:
226 /// `pasynManager->isAutoConnect` (`asynManager.c`).
227 GetAutoConnect,
228 /// Block the port: only this user's requests will be dequeued until unblocked.
229 BlockProcess,
230 /// Unblock the port.
231 UnblockProcess,
232 /// Resolve a record's bind request to a parameter reason index. Carries the
233 /// full [`DrvUserRequest`](crate::port::DrvUserRequest) — drvInfo, asyn `addr`, and the record's asyn
234 /// interface — so an on-demand driver can create the parameter with the type
235 /// the record will read it as.
236 DrvUserCreate(crate::port::DrvUserRequest),
237 /// Read an enum value (index + string choices).
238 EnumRead,
239 /// Write an enum index.
240 EnumWrite {
241 index: usize,
242 },
243 /// Read an i32 array.
244 Int32ArrayRead {
245 max_elements: usize,
246 },
247 /// Write an i32 array.
248 Int32ArrayWrite {
249 data: Vec<i32>,
250 },
251 /// Read an f64 array.
252 Float64ArrayRead {
253 max_elements: usize,
254 },
255 /// Write an f64 array.
256 Float64ArrayWrite {
257 data: Vec<f64>,
258 },
259 /// Read an i8 array.
260 Int8ArrayRead {
261 max_elements: usize,
262 },
263 /// Write an i8 array.
264 Int8ArrayWrite {
265 data: Vec<i8>,
266 },
267 /// Read an i16 array.
268 Int16ArrayRead {
269 max_elements: usize,
270 },
271 /// Write an i16 array.
272 Int16ArrayWrite {
273 data: Vec<i16>,
274 },
275 /// Read an i64 array.
276 Int64ArrayRead {
277 max_elements: usize,
278 },
279 /// Write an i64 array.
280 Int64ArrayWrite {
281 data: Vec<i64>,
282 },
283 /// Read an f32 array.
284 Float32ArrayRead {
285 max_elements: usize,
286 },
287 /// Write an f32 array.
288 Float32ArrayWrite {
289 data: Vec<f32>,
290 },
291 /// Set params directly in the store (like C setIntegerParam/setDoubleParam)
292 /// and then fire interrupt notifications (callParamCallbacks).
293 /// Does NOT trigger writeInt32/on_param_change — avoids re-entrancy.
294 CallParamCallbacks {
295 addr: i32,
296 /// Param updates to apply before firing callbacks.
297 /// Empty = just fire callbacks for previously changed params.
298 updates: Vec<ParamSetValue>,
299 },
300 /// Get a port/driver option by key.
301 GetOption {
302 key: String,
303 },
304 /// Set a port/driver option by key.
305 SetOption {
306 key: String,
307 value: String,
308 },
309 /// Print a driver report (matches C `asynManager->report` /
310 /// iocsh `asynReport`). The actor calls
311 /// [`crate::port::PortDriver::report`] which writes to stderr
312 /// at the requested verbosity. Carried by the actor so the
313 /// driver is observed from its own thread (consistent with C
314 /// asyn's `pport->lock` invariant for `report`).
315 Report {
316 level: i32,
317 },
318 /// Set the port's input EOS bytes — C `pasynOctet->setInputEos`.
319 /// Drives the same `PortDriver::set_input_eos(&[u8])` hook the EOS
320 /// interpose layer reads, so asynRecord IEOS writes survive a
321 /// round trip through the actor (previously routed through the
322 /// generic option HashMap which no driver consumes).
323 SetInputEos {
324 eos: Vec<u8>,
325 },
326 /// Set the port's output EOS bytes — C `pasynOctet->setOutputEos`.
327 SetOutputEos {
328 eos: Vec<u8>,
329 },
330 /// Read back the port's input EOS bytes — C `pasynOctet->getInputEos`.
331 /// asynRecord's `getEos` (asynRecord.c:1985-2026) calls it after every
332 /// IEOS/OEOS put so the record shows what the driver actually holds, not
333 /// what was requested. Returns the bytes in [`RequestResult::data`].
334 GetInputEos,
335 /// Read back the port's output EOS bytes — C `pasynOctet->getOutputEos`.
336 GetOutputEos,
337 /// Query whether the port's *transport* is connected. C parity:
338 /// `pasynManager->isConnected` — the state the driver publishes through
339 /// `exceptionConnect`/`exceptionDisconnect`, not "is a record bound to
340 /// this port". `asynRecord` reads it in `monitorStatus` (asynRecord.c:
341 /// 1089-1093) to refresh CNCT, and gates its `callbackConnect` on it
342 /// (:858-888) so a CNCT put never re-connects an already-connected port.
343 GetConnected,
344 /// Install the echo interpose on top of the port's octet stack. C parity:
345 /// `asynInterposeEcho(portName, addr)`
346 /// (`asynInterposeEcho.c:165-190`), the iocsh command a startup script
347 /// runs *after* the port is configured.
348 ///
349 /// It is a request rather than a direct `install_interpose` because the actor
350 /// owns the driver once the port is registered — the same reason
351 /// `SetOption` / `SetInputEos` are requests. Installing from the shell
352 /// thread would race every in-flight transfer.
353 PushEchoInterpose,
354 /// Install the delay interpose on top of the port's octet stack. C parity:
355 /// `asynInterposeDelay(portName, addr, delay)`
356 /// (`asynInterposeDelay.c:176-215`), registered with iocsh at
357 /// `asynInterposeDelay.c:221-234`. Same actor-ownership reason as
358 /// [`RequestOp::PushEchoInterpose`].
359 PushDelayInterpose {
360 delay: std::time::Duration,
361 },
362 /// Install the EOS interpose on the addressed device's octet stack. C
363 /// parity: `asynInterposeEosConfig(portName, addr, processEosIn,
364 /// processEosOut)` (`asynInterposeEos.c:84-140`), registered with iocsh at
365 /// :393-410. The two flags select which half of the layer is live.
366 PushEosInterpose {
367 process_in: bool,
368 process_out: bool,
369 },
370 /// Set (or clear) the port's time-stamp source by NAME. C parity:
371 /// `asynRegisterTimeStampSource(portName, functionName)` /
372 /// `asynUnregisterTimeStampSource(portName)` (asynShellCommands.c:1181-1223)
373 /// — C resolves the name through `registryFunctionFind` and hands the
374 /// function to `pasynManager->registerTimeStampSource`. The NAME travels,
375 /// not the function: that is what makes it resolvable on the far side of a
376 /// remote port, exactly as C resolves it in the IOC's own registry.
377 /// `None` = unregister (back to the driver's default clock).
378 SetTimeStampSource {
379 name: Option<String>,
380 },
381 /// Install the flush-timeout interpose on the addressed device's octet
382 /// stack. C parity: `asynInterposeFlushConfig(portName, addr, timeout)`
383 /// (`asynInterposeFlush.c:66-91`); C's shell argument is in milliseconds
384 /// and `<= 0` means 1 ms (:78-79), so the conversion happens at the shell
385 /// and the op carries a real duration.
386 PushFlushInterpose {
387 flush_timeout: std::time::Duration,
388 },
389 /// Send a GPIB universal command byte — C `asynGpib::universalCmd`
390 /// (asynGpib.c:480-484). asynRecord's UCMD dispatch
391 /// (`gpibUniversalCmd`, asynRecord.c:1638-1679).
392 GpibUniversalCmd {
393 cmd: u8,
394 },
395 /// Send a GPIB addressed-command frame — C `asynGpib::addressedCmd`
396 /// (asynGpib.c:472-478). asynRecord's ACMD dispatch builds the frame
397 /// (`gpibAddressedCmd`, asynRecord.c:1681-1756).
398 GpibAddressedCmd {
399 data: Vec<u8>,
400 },
401 /// Assert Interface Clear — C `asynGpib::ifc` (asynGpib.c:486-490).
402 GpibIfc,
403 /// Set the Remote Enable line — C `asynGpib::ren` (asynGpib.c:492-496).
404 GpibRen {
405 enable: bool,
406 },
407}
408
409/// Result returned by the worker after executing a request.
410#[derive(Debug)]
411pub struct RequestResult {
412 pub status: AsynStatus,
413 pub message: String,
414 pub nbytes: usize,
415 pub data: Option<Vec<u8>>,
416 pub int_val: Option<i32>,
417 pub int64_val: Option<i64>,
418 pub float_val: Option<f64>,
419 pub uint_val: Option<u32>,
420 /// Reason index (from DrvUserCreate).
421 pub reason: Option<usize>,
422 /// Per-record octet length cap (from DrvUserCreate; C `modbusDrvUser_t.len`).
423 /// `None` when the drvInfo carried no cap.
424 pub max_octet_len: Option<usize>,
425 /// Enum index (from EnumRead).
426 pub enum_index: Option<usize>,
427 /// Driver enum string/value/severity table (from EnumRead). C asyn
428 /// device support reads this via `asynEnum->read` and pushes it onto
429 /// the record's state fields (ZRST/ZRVL/ZRSV…, ZNAM/ONAM…) at init —
430 /// see `devAsynInt32.c::initCommon` (297-324) / `setEnums` (415-435).
431 pub enum_entries: Option<Arc<[crate::param::EnumEntry]>>,
432 /// i32 array data (from Int32ArrayRead).
433 pub int32_array: Option<Vec<i32>>,
434 /// f64 array data (from Float64ArrayRead).
435 pub float64_array: Option<Vec<f64>>,
436 /// i8 array data (from Int8ArrayRead).
437 pub int8_array: Option<Vec<i8>>,
438 /// i16 array data (from Int16ArrayRead).
439 pub int16_array: Option<Vec<i16>>,
440 /// i64 array data (from Int64ArrayRead).
441 pub int64_array: Option<Vec<i64>>,
442 /// f32 array data (from Float32ArrayRead).
443 pub float32_array: Option<Vec<f32>>,
444 /// Alarm status from the driver param store (populated on reads).
445 pub alarm_status: u16,
446 /// Alarm severity from the driver param store (populated on reads).
447 pub alarm_severity: u16,
448 /// Timestamp from the driver param store (populated on reads).
449 pub timestamp: Option<SystemTime>,
450 /// Device read auxiliary status (C `pasynUser->auxStatus`), populated on
451 /// reads from the param store alongside the value. Distinct from
452 /// [`Self::status`] (the request/op outcome that drives an `Err`/Error
453 /// reply): a read OP can succeed and return a value while `aux_status`
454 /// flags that value invalid. Device support gates the value store on this —
455 /// C `processAi` stores the value only when `result.status == asynSuccess`
456 /// and otherwise returns -1 keeping the prior value (devAsynInt32.c:848-855)
457 /// — the same way the I/O Intr ring gates on `CachedInterrupt.aux_status`.
458 pub aux_status: AsynStatus,
459 /// Option value string (from GetOption).
460 pub option_value: Option<String>,
461 /// Int64 bounds (from GetBoundsInt32/Int64).
462 pub bounds: Option<(i64, i64)>,
463 /// End-of-message reason flags from an octet read.
464 ///
465 /// C parity: `asynOctet::read` returns `nbytes` together with
466 /// `int *eomReason` (`interfaces/asynOctet.h:38-40`). The flags
467 /// `ASYN_EOM_CNT | ASYN_EOM_EOS | ASYN_EOM_END` mirror
468 /// [`crate::interpose::EomReason`]. Stored as `u32` so the
469 /// request layer stays bitflag-crate-free; converters live on
470 /// `EomReason::from_bits_truncate`.
471 pub eom_reason: u32,
472}
473
474impl RequestResult {
475 fn base() -> Self {
476 Self {
477 status: AsynStatus::Success,
478 message: String::new(),
479 nbytes: 0,
480 data: None,
481 int_val: None,
482 int64_val: None,
483 float_val: None,
484 uint_val: None,
485 reason: None,
486 max_octet_len: None,
487 enum_index: None,
488 enum_entries: None,
489 int32_array: None,
490 float64_array: None,
491 int8_array: None,
492 int16_array: None,
493 int64_array: None,
494 float32_array: None,
495 alarm_status: 0,
496 alarm_severity: 0,
497 timestamp: None,
498 aux_status: AsynStatus::Success,
499 option_value: None,
500 bounds: None,
501 eom_reason: 0,
502 }
503 }
504
505 pub fn write_ok() -> Self {
506 Self::base()
507 }
508
509 /// Octet write result carrying the number of bytes transferred
510 /// (C `asynOctet::write`'s `*nbytesTransfered`). Used by
511 /// `PortHandle::write_octet` / `SyncIO::write_octet` to report how
512 /// many bytes the driver actually wrote on success.
513 pub fn write_n(nbytes: usize) -> Self {
514 Self {
515 nbytes,
516 ..Self::base()
517 }
518 }
519
520 pub fn octet_read(buf: Vec<u8>, nbytes: usize) -> Self {
521 Self {
522 nbytes,
523 data: Some(buf),
524 ..Self::base()
525 }
526 }
527
528 /// Variant of [`Self::octet_read`] that carries the
529 /// end-of-message reason flags returned by
530 /// [`crate::port::PortDriver::io_read_octet_eom`]. The raw `u32`
531 /// is decoded with `EomReason::from_bits_truncate` on the
532 /// consumer side.
533 pub fn octet_read_eom(buf: Vec<u8>, nbytes: usize, eom_reason: u32) -> Self {
534 Self {
535 nbytes,
536 data: Some(buf),
537 eom_reason,
538 ..Self::base()
539 }
540 }
541
542 pub fn int32_read(value: i32) -> Self {
543 Self {
544 int_val: Some(value),
545 ..Self::base()
546 }
547 }
548
549 pub fn int64_read(value: i64) -> Self {
550 Self {
551 int64_val: Some(value),
552 ..Self::base()
553 }
554 }
555
556 pub fn float64_read(value: f64) -> Self {
557 Self {
558 float_val: Some(value),
559 ..Self::base()
560 }
561 }
562
563 pub fn uint32_read(value: u32) -> Self {
564 Self {
565 uint_val: Some(value),
566 ..Self::base()
567 }
568 }
569
570 pub fn drv_user_create(reason: usize, max_octet_len: Option<usize>) -> Self {
571 Self {
572 reason: Some(reason),
573 max_octet_len,
574 ..Self::base()
575 }
576 }
577
578 pub fn enum_read(index: usize) -> Self {
579 Self {
580 enum_index: Some(index),
581 ..Self::base()
582 }
583 }
584
585 /// [`Self::enum_read`] carrying the driver's full enum table so the
586 /// device-support init path can propagate it to the record's state
587 /// fields (C `setEnums`). The index is the current selection.
588 pub fn enum_read_with_entries(index: usize, entries: Arc<[crate::param::EnumEntry]>) -> Self {
589 Self {
590 enum_index: Some(index),
591 enum_entries: Some(entries),
592 ..Self::base()
593 }
594 }
595
596 pub fn int32_array_read(data: Vec<i32>) -> Self {
597 Self {
598 int32_array: Some(data),
599 ..Self::base()
600 }
601 }
602
603 pub fn float64_array_read(data: Vec<f64>) -> Self {
604 Self {
605 float64_array: Some(data),
606 ..Self::base()
607 }
608 }
609
610 pub fn int8_array_read(data: Vec<i8>) -> Self {
611 Self {
612 int8_array: Some(data),
613 ..Self::base()
614 }
615 }
616
617 pub fn int16_array_read(data: Vec<i16>) -> Self {
618 Self {
619 int16_array: Some(data),
620 ..Self::base()
621 }
622 }
623
624 pub fn int64_array_read(data: Vec<i64>) -> Self {
625 Self {
626 int64_array: Some(data),
627 ..Self::base()
628 }
629 }
630
631 pub fn float32_array_read(data: Vec<f32>) -> Self {
632 Self {
633 float32_array: Some(data),
634 ..Self::base()
635 }
636 }
637
638 pub fn option_read(value: String) -> Self {
639 Self {
640 option_value: Some(value),
641 ..Self::base()
642 }
643 }
644
645 pub fn bounds_read(low: i64, high: i64) -> Self {
646 Self {
647 bounds: Some((low, high)),
648 ..Self::base()
649 }
650 }
651
652 /// Attach alarm/timestamp metadata to this result.
653 pub fn with_alarm(
654 mut self,
655 alarm_status: u16,
656 alarm_severity: u16,
657 timestamp: Option<SystemTime>,
658 ) -> Self {
659 self.alarm_status = alarm_status;
660 self.alarm_severity = alarm_severity;
661 self.timestamp = timestamp;
662 self
663 }
664}
665
666/// Lifecycle of a queued request, mirroring C `asynManager` queue/callback
667/// state so that `AQR` cancellation reproduces the `cancelRequest` `wasQueued`
668/// split (asynManager.c:1632-1692) by construction rather than by a runtime
669/// guard.
670///
671/// `cancelRequest` removes the request and reports `wasQueued==1` ONLY while it
672/// is still on the queue (asynManager.c:1661-1668); once the port thread has
673/// dequeued it and is running the callback (`callbackActive`) or it has already
674/// finished, `wasQueued==0` and the I/O runs to completion and is reported
675/// normally (asynManager.c:1645-1659). `Queued` is the only state a cancel can
676/// win from; the executor's `Queued -> Running` transition closes that window.
677///
678/// The queue-wait timeout (C `queueTimeoutCallback`, asynManager.c:647-700) is
679/// the second way a request can leave the queue without running, and it obeys
680/// the same rule: the timer callback returns immediately when `!isQueued`
681/// (:655-661), so a request the port thread has already dequeued always
682/// completes. `TimedOut` is therefore a sibling of `Cancelled` — a terminal
683/// state reachable only from `Queued` — and the two together are the complete
684/// set of "this request never ran" outcomes. Which one won is what tells the
685/// caller *which* C callback to report ("I/O request canceled" vs "process
686/// queueRequest timeout"), so they are distinct states rather than one flag.
687const STATE_QUEUED: u8 = 0;
688const STATE_RUNNING: u8 = 1;
689const STATE_DONE: u8 = 2;
690const STATE_CANCELLED: u8 = 3;
691const STATE_TIMED_OUT: u8 = 4;
692
693/// Token tracking the queue/execution lifecycle of an off-thread request.
694///
695/// The state machine makes the C `wasQueued` semantics hold by construction:
696/// `cancel()` succeeds only from `Queued`, the executor claims the request with
697/// `begin_running()` (refused once cancelled) and releases it with `finish()`,
698/// so a cancel that arrives after execution started cannot transition the token
699/// and is a no-op — the I/O completes and applies normally.
700#[derive(Clone, Debug)]
701pub struct CancelToken(pub Arc<AtomicU8>);
702
703impl CancelToken {
704 pub fn new() -> Self {
705 Self(Arc::new(AtomicU8::new(STATE_QUEUED)))
706 }
707
708 /// `AQR` / C `cancelRequest`: cancel the request iff it is still queued.
709 ///
710 /// Returns the C `wasQueued` flag — `true` when the request was removed
711 /// from the queue (the caller must report "I/O request canceled",
712 /// asynRecord.c:397-404); `false` when it had already been dequeued and was
713 /// running or had completed, in which case the I/O runs to completion and
714 /// reports normally (asynManager.c:1645-1659).
715 pub fn cancel(&self) -> bool {
716 self.0
717 .compare_exchange(
718 STATE_QUEUED,
719 STATE_CANCELLED,
720 AtomicOrdering::AcqRel,
721 AtomicOrdering::Acquire,
722 )
723 .is_ok()
724 }
725
726 /// C `queueTimeoutCallback` (asynManager.c:647-700): the queue-wait deadline
727 /// expired. Removes the request from the queue iff it is still queued —
728 /// C's `if(!puserPvt->isQueued) { ...; return; }` guard (:655-661) — and
729 /// returns whether it won.
730 ///
731 /// `false` means the port thread had already dequeued the request: the timer
732 /// fired too late, the I/O runs to completion and reports normally, and the
733 /// caller must keep waiting for it. This is the same `isQueued` gate
734 /// [`Self::cancel`] answers for `AQR`, so a cancel and a timeout racing the
735 /// same request cannot both win.
736 pub fn time_out_if_queued(&self) -> bool {
737 self.0
738 .compare_exchange(
739 STATE_QUEUED,
740 STATE_TIMED_OUT,
741 AtomicOrdering::AcqRel,
742 AtomicOrdering::Acquire,
743 )
744 .is_ok()
745 }
746
747 /// Executor at dequeue: claim the request for execution (C dequeue under
748 /// `asynManagerLock`, asynManager.c:1661-1666 is the cancel counterpart).
749 ///
750 /// Returns `false` iff the request left the queue without running — it was
751 /// cancelled (`AQR`) or its queue-wait deadline expired — in which case the
752 /// executor must drop it and report that outcome. Otherwise the token enters
753 /// `Running`. A multi-phase plan re-claims the same token for its next phase
754 /// from `Done`, so this transitions from either `Queued` or `Done`;
755 /// `Cancelled` and `TimedOut` are terminal.
756 pub fn begin_running(&self) -> bool {
757 let mut cur = self.0.load(AtomicOrdering::Acquire);
758 loop {
759 if cur == STATE_CANCELLED || cur == STATE_TIMED_OUT {
760 return false;
761 }
762 match self.0.compare_exchange_weak(
763 cur,
764 STATE_RUNNING,
765 AtomicOrdering::AcqRel,
766 AtomicOrdering::Acquire,
767 ) {
768 Ok(_) => return true,
769 Err(actual) => cur = actual,
770 }
771 }
772 }
773
774 /// Executor at completion: mark the running request finished so a later
775 /// cancel is a no-op (the C `wasQueued==0` window). Idempotent and a no-op
776 /// from any state other than `Running`.
777 pub fn finish(&self) {
778 let _ = self.0.compare_exchange(
779 STATE_RUNNING,
780 STATE_DONE,
781 AtomicOrdering::AcqRel,
782 AtomicOrdering::Acquire,
783 );
784 }
785
786 /// True iff the request was cancelled while still queued — the C
787 /// `wasQueued==true` outcome. A cancel that lost the race (the executor had
788 /// already begun running) leaves the state `Running`/`Done`, so this stays
789 /// `false` and the completed I/O applies normally.
790 pub fn is_cancelled(&self) -> bool {
791 self.0.load(AtomicOrdering::Acquire) == STATE_CANCELLED
792 }
793
794 /// True iff the request was removed from the queue by its queue-wait
795 /// deadline — the C `queueTimeoutCallback` outcome. Mutually exclusive with
796 /// [`Self::is_cancelled`]: both transition out of `Queued`, so exactly one
797 /// can win.
798 pub fn is_timed_out(&self) -> bool {
799 self.0.load(AtomicOrdering::Acquire) == STATE_TIMED_OUT
800 }
801}
802
803impl Default for CancelToken {
804 fn default() -> Self {
805 Self::new()
806 }
807}
808
809#[cfg(test)]
810mod tests {
811 use super::*;
812
813 #[test]
814 fn cancel_succeeds_only_while_queued() {
815 // C `wasQueued==1`: a still-queued request is cancelled and removed.
816 let token = CancelToken::new();
817 assert!(!token.is_cancelled());
818 assert!(token.cancel(), "a queued request reports wasQueued==true");
819 assert!(token.is_cancelled());
820 // The executor then refuses to run it (it was removed from the queue).
821 assert!(
822 !token.begin_running(),
823 "a cancelled request is not claimed for execution"
824 );
825 }
826
827 #[test]
828 fn cancel_after_begin_running_is_noop() {
829 // C `wasQueued==0` while `callbackActive`: the I/O runs to completion.
830 let token = CancelToken::new();
831 assert!(
832 token.begin_running(),
833 "the executor claims a queued request"
834 );
835 assert!(
836 !token.cancel(),
837 "a cancel during execution reports wasQueued==false"
838 );
839 assert!(
840 !token.is_cancelled(),
841 "the running I/O is not treated as cancelled"
842 );
843 token.finish();
844 assert!(!token.is_cancelled(), "the completed I/O applies normally");
845 }
846
847 #[test]
848 fn cancel_after_finish_is_noop() {
849 // C `wasQueued==0` after the callback finished: nothing to cancel.
850 let token = CancelToken::new();
851 assert!(token.begin_running());
852 token.finish();
853 assert!(
854 !token.cancel(),
855 "a cancel after completion reports wasQueued==false"
856 );
857 assert!(!token.is_cancelled());
858 }
859
860 #[test]
861 fn begin_running_reclaims_token_for_next_phase() {
862 // A WriteRead plan threads one token through two phases; the read phase
863 // re-claims the token the write phase finished.
864 let token = CancelToken::new();
865 assert!(token.begin_running(), "write phase claims the queued token");
866 token.finish();
867 assert!(
868 token.begin_running(),
869 "read phase re-claims the finished token"
870 );
871 token.finish();
872 assert!(!token.is_cancelled());
873 }
874
875 #[test]
876 fn cancel_is_terminal_across_phases() {
877 // Once cancelled while queued, no later phase may run.
878 let token = CancelToken::new();
879 assert!(token.cancel());
880 assert!(!token.begin_running(), "cancelled is terminal");
881 assert!(token.is_cancelled());
882 }
883}