asyn-rs

Rust port of EPICS asyn — an async device I/O framework for hardware drivers.

No C dependencies. Pure Rust. Integrates with epics-ca via the optional epics feature.

Repository: https://github.com/epics-rs/epics-rs

Overview

asyn-rs provides the same driver model as C asyn, but uses Rust's type system and async concurrency for safety and performance:

• PortDriver trait — implement read_*/write_* for your hardware
• ParamList — named parameter cache with change tracking, timestamps, and alarm status
• InterruptManager — dual async (broadcast) + sync (mpsc) callback delivery
• PortManager — registry of named port drivers
• AsynDeviceSupport — universal asyn device support factory bridging any asyn-rs driver to epics-ca records

What's New in v0.2

v0.2.0 — Actor Model + Typed Capabilities

Actor-based port driver execution — drivers are no longer accessed through Arc<Mutex<dyn PortDriver>>. Instead, each driver runs in its own actor thread with exclusive ownership:

• PortActor — owns the driver exclusively, dispatches requests via channel
• PortHandle — cloneable async handle with typed convenience methods (read_int32(), write_float64(), etc.)
• AsyncCompletionHandle — Future impl + wait_blocking() for sync callers

Adapter migration — AsynDeviceSupport now supports both legacy (Arc<Mutex>) and actor (PortHandle) backends via PortBackend enum. New drivers should use from_handle().

Typed capability system:

• InterfaceType enum with bidirectional string conversion (e.g. "asynInt32" ↔ InterfaceType::Int32)
• Capability enum for declaring driver capabilities at type level
• PortDriver::capabilities() / supports() default trait methods

Extended request types — RequestOp extended with DrvUserCreate, Enum, Int32Array, Float64Array. RequestResult gains alarm/timestamp metadata.

v0.2.1 — Protocol, Transport, Runtime

Pure-data protocol (src/protocol/) — serializable message types at all boundaries, no trait objects or closures:

All types derive serde::Serialize/Deserialize for future wire transport.

Pluggable transport (src/transport/) — RuntimeClient trait decouples callers from transport:

pub trait RuntimeClient: Send + Sync + Clone + 'static {
    fn request(&self, req: PortRequest) -> Pin<Box<dyn Future<Output = Result<PortReply, TransportError>> + Send + '_>>;
    fn request_blocking(&self, req: PortRequest) -> Result<PortReply, TransportError>;
    fn subscribe(&self, filter: EventFilter) -> ...;
}

• InProcessClient — zero-cost fast path, direct enum pass-through (no serialization)
• Future: UnixSocketClient for multi-process deployments

Runtime module (src/runtime/) — promoted actors with lifecycle management:

• PortRuntime — promoted PortActor with RuntimeEvent broadcast (Started/Stopped/Connected/Disconnected/Error) and graceful shutdown
• AxisRuntime — per-axis motor actor with event emission, poll loop, and I/O Intr notification
• Supervision — generic restart loop with configurable policy (max_restarts, restart_window)
• PortManager integration — register_port_runtime() auto-registers both runtime handle and legacy port handle for backwards compatibility

Criterion benchmarks (benches/throughput.rs):

• local_int32_read / local_float64_write / local_octet_roundtrip — legacy mutex path
• actor_int32_read — PortHandle via actor
• concurrent_32_producers — 32 threads on same port
• interrupt_event_throughput — 1k events broadcast delivery

Architecture

┌─────────────────────────────────────────────┐
│  EPICS Records (ai, ao, longin, ...)        │
│         ↕ DeviceSupport trait                │
│  ┌─────────────────────────────────┐        │
│  │  AsynDeviceSupport (adapter)    │ epics   │
│  │  - alarm/timestamp propagation  │ feature │
│  │  - I/O Intr scan bridging       │         │
│  └──────────┬──────────────────────┘        │
└─────────────┼───────────────────────────────┘
              ↕
┌─────────────────────────────────────────────┐
│  RuntimeClient trait (transport layer)       │
│  ├── InProcessClient (zero-cost fast path)  │
│  └── [UnixSocketClient] (future)            │
│         ↕ PortCommand / PortReply            │
│  ┌─────────────────────────────────┐        │
│  │  PortRuntime / PortActor        │        │
│  │  - exclusive driver ownership   │        │
│  │  - RuntimeEvent broadcast       │        │
│  │  - graceful shutdown            │        │
│  └──────────┬──────────────────────┘        │
└─────────────┼───────────────────────────────┘
              ↕
┌─────────────────────────────────────────────┐
│  PortDriver trait                            │
│  - read/write: Int32, Float64, Octet,       │
│    UInt32Digital, arrays                     │
│  - InterfaceType / Capability declarations  │
│                                              │
│  PortDriverBase                              │
│  ├── ParamList (cache + change tracking)     │
│  ├── InterruptManager (broadcast + mpsc)     │
│  └── options: HashMap<String, String>        │
└─────────────────────────────────────────────┘
              ↕
┌─────────────────────────────────────────────┐
│  Your Hardware Driver                        │
│  - Background async task polls device         │
│  - set_*_param() + call_param_callbacks()    │
│  - Default read_* returns cached values      │
└─────────────────────────────────────────────┘

Quick Start

Add to Cargo.toml:

[dependencies]
asyn-rs = { path = "../asyn-rs" }
# With EPICS integration:
# asyn-rs = { path = "../asyn-rs", features = ["epics"] }

Implementing a Driver

use asyn_rs::param::ParamType;
use asyn_rs::port::{PortDriver, PortDriverBase, PortFlags};
use asyn_rs::error::AsynResult;

struct TemperatureDriver {
    base: PortDriverBase,
    temp_idx: usize,
}

impl TemperatureDriver {
    fn new() -> Self {
        let mut base = PortDriverBase::new("tempPort", 1, PortFlags::default());
        let temp_idx = base.create_param("TEMPERATURE", ParamType::Float64).unwrap();
        Self { base, temp_idx }
    }

    /// Call from a background task to update the cached value.
    fn update_temperature(&mut self, value: f64) -> AsynResult<()> {
        self.base.set_float64_param(self.temp_idx, 0, value)?;
        self.base.call_param_callbacks(0)?;
        Ok(())
    }
}

impl PortDriver for TemperatureDriver {
    fn base(&self) -> &PortDriverBase { &self.base }
    fn base_mut(&mut self) -> &mut PortDriverBase { &mut self.base }
}

Registering with PortManager

use asyn_rs::manager::PortManager;

let manager = PortManager::new();
let port = manager.register_port(TemperatureDriver::new());

// Access from anywhere via Arc<RwLock<dyn PortDriver>>
let p = manager.find_port("tempPort").unwrap();

EPICS Integration

With the epics feature, use AsynDeviceSupport to bridge drivers to epics-ca records:

use asyn_rs::adapter::{AsynDeviceSupport, parse_asyn_link};

// In a DeviceSupport factory:
let link = parse_asyn_link("@asyn(tempPort, 0, 1.0) TEMPERATURE").unwrap();
let port = manager.find_port(&link.port_name).unwrap();
let adapter = AsynDeviceSupport::new(port, link, "asynFloat64");

The adapter handles:

• Parameter resolution via drvUserCreate
• Value read/write through the port driver's cache
• Alarm status/severity propagation from driver to record
• Timestamp propagation (driver-supplied or auto-generated)
• I/O Intr scan support (broadcast → per-record mpsc bridge)

Modules

Runtime Facade

asyn-rs re-exports async runtime primitives so driver authors never depend on tokio directly:

use asyn_rs::runtime::sync::{mpsc, Notify, Arc};    // channels, sync primitives
use asyn_rs::runtime::task::{spawn, sleep, interval}; // task utilities
use asyn_rs::runtime::select;                         // async multiplexing

For IOC binaries, use #[epics_base_rs::epics_main] instead of #[tokio::main], and #[epics_base_rs::epics_test] instead of #[tokio::test].

I/O Model

asyn-rs uses a cache-based model instead of C asyn's queue/block model:

1. A background task polls the hardware
2. Driver calls set_*_param() to update cached values
3. Driver calls call_param_callbacks() to notify subscribers
4. Default read_* methods return the cached value immediately

This means can_block is preserved for compatibility but has no runtime effect. For command/response hardware, drivers manage their own async task and request queue.

Testing

cargo test                    # Core tests (316)
cargo test --features epics   # With EPICS integration (326)
cargo bench                   # Criterion throughput benchmarks

License

EPICS Open License