laser_dac/presentation/
session.rs

1//! FrameSession and FrameSessionConfig — public frame-mode API.
2
3use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
4use std::sync::mpsc;
5use std::sync::{Arc, Mutex};
6use std::thread::JoinHandle;
7use std::time::{Duration, Instant};
8
9use crate::backend::BackendKind;
10use crate::config::StreamConfig;
11use crate::device::{DacInfo, OutputModel};
12use crate::error::{Error, Result};
13use crate::reconnect::ReconnectPolicy;
14use crate::stream::{ControlMsg, RunExit, StreamControl};
15
16use super::driver::{self, DriverInputs, SourceOwned};
17use super::engine::PresentationEngine;
18use super::slice_pipeline::SlicePipeline;
19use super::{default_transition, Frame, OutputResetReason, TransitionFn};
20
21// =============================================================================
22// FrameSessionConfig
23// =============================================================================
24
25/// Configuration for a frame-mode streaming session.
26pub struct FrameSessionConfig {
27    /// Points per second output rate.
28    pub pps: u32,
29    /// Transition function for blanking between frames.
30    pub transition_fn: TransitionFn,
31    /// Duration of forced blanking after arming (default: 1ms).
32    pub startup_blank: std::time::Duration,
33    /// Number of points to shift RGB relative to XY (0 = disabled).
34    ///
35    /// Delays color channels relative to XY coordinates, compensating for
36    /// the difference in galvo mirror and laser modulation response times.
37    /// Applied at composition time. Set to 0 to disable.
38    pub color_delay_points: usize,
39    /// Reconnection configuration (default: disabled).
40    ///
41    /// Set via [`with_reconnect`](Self::with_reconnect) to enable automatic
42    /// reconnection when the device disconnects.
43    pub reconnect: Option<crate::config::ReconnectConfig>,
44    /// Policy for what to output when the stream is idle (disarmed).
45    ///
46    /// Controls scanner behavior when disarmed. Default: [`Blank`](crate::config::IdlePolicy::Blank)
47    /// (park at origin with laser off). Use [`Park`](crate::config::IdlePolicy::Park) to park at a
48    /// specific position.
49    pub idle_policy: crate::config::IdlePolicy,
50    /// Optional hook for processing the final presented output.
51    pub output_filter: Option<Box<dyn super::OutputFilter>>,
52}
53
54impl FrameSessionConfig {
55    const DEFAULT_COLOR_DELAY: std::time::Duration = std::time::Duration::from_micros(150);
56
57    /// Create a new config with the given PPS and default transition.
58    pub fn new(pps: u32) -> Self {
59        let color_delay_points =
60            (Self::DEFAULT_COLOR_DELAY.as_secs_f64() * pps as f64).ceil() as usize;
61        Self {
62            pps,
63            transition_fn: default_transition(pps),
64            startup_blank: std::time::Duration::from_millis(1),
65            color_delay_points,
66            idle_policy: crate::config::IdlePolicy::default(),
67            reconnect: None,
68            output_filter: None,
69        }
70    }
71
72    /// Set the transition function (builder pattern).
73    pub fn with_transition_fn(mut self, f: TransitionFn) -> Self {
74        self.transition_fn = f;
75        self
76    }
77
78    /// Set the startup blank duration (builder pattern).
79    pub fn with_startup_blank(mut self, duration: std::time::Duration) -> Self {
80        self.startup_blank = duration;
81        self
82    }
83
84    /// Set the color delay in points (builder pattern).
85    pub fn with_color_delay_points(mut self, n: usize) -> Self {
86        self.color_delay_points = n;
87        self
88    }
89
90    /// Enable automatic reconnection (builder pattern).
91    ///
92    /// Requires the device to have been opened via [`open_device`](crate::open_device).
93    pub fn with_reconnect(mut self, config: crate::config::ReconnectConfig) -> Self {
94        self.reconnect = Some(config);
95        self
96    }
97
98    /// Set the idle policy (builder pattern).
99    ///
100    /// Controls scanner behavior when disarmed. See [`crate::config::IdlePolicy`].
101    pub fn with_idle_policy(mut self, policy: crate::config::IdlePolicy) -> Self {
102        self.idle_policy = policy;
103        self
104    }
105
106    /// Install a final-output filter (builder pattern).
107    pub fn with_output_filter(mut self, filter: Box<dyn super::OutputFilter>) -> Self {
108        self.output_filter = Some(filter);
109        self
110    }
111}
112
113// =============================================================================
114// FrameSession
115// =============================================================================
116
117/// Read-only liveness and connectivity metrics for a [`FrameSession`].
118#[derive(Clone)]
119pub struct FrameSessionMetrics {
120    inner: Arc<FrameSessionMetricsInner>,
121}
122
123struct FrameSessionMetricsInner {
124    connected: AtomicBool,
125    origin: Instant,
126    last_loop_activity_nanos: AtomicU64,
127    last_write_success_nanos: AtomicU64,
128}
129
130impl FrameSessionMetrics {
131    pub(crate) fn new(connected: bool) -> Self {
132        let metrics = Self {
133            inner: Arc::new(FrameSessionMetricsInner {
134                connected: AtomicBool::new(connected),
135                origin: Instant::now(),
136                last_loop_activity_nanos: AtomicU64::new(0),
137                last_write_success_nanos: AtomicU64::new(0),
138            }),
139        };
140        metrics.mark_loop_activity();
141        metrics
142    }
143
144    /// Returns whether the session currently has a connected backend.
145    pub fn connected(&self) -> bool {
146        self.inner.connected.load(Ordering::SeqCst)
147    }
148
149    /// Returns the last time the scheduler thread showed progress.
150    pub fn last_loop_activity(&self) -> Option<Instant> {
151        self.instant_from_nanos(self.inner.last_loop_activity_nanos.load(Ordering::SeqCst))
152    }
153
154    /// Returns the last time a backend write completed successfully.
155    pub fn last_write_success(&self) -> Option<Instant> {
156        self.instant_from_nanos(self.inner.last_write_success_nanos.load(Ordering::SeqCst))
157    }
158
159    fn instant_from_nanos(&self, nanos: u64) -> Option<Instant> {
160        if nanos == 0 {
161            None
162        } else {
163            self.inner.origin.checked_add(Duration::from_nanos(nanos))
164        }
165    }
166
167    fn now_nanos(&self) -> u64 {
168        (self.inner.origin.elapsed().as_nanos().min(u64::MAX as u128) as u64).max(1)
169    }
170
171    pub(super) fn mark_loop_activity(&self) {
172        self.inner
173            .last_loop_activity_nanos
174            .store(self.now_nanos(), Ordering::SeqCst);
175    }
176
177    pub(super) fn mark_write_success(&self) {
178        let now = self.now_nanos();
179        self.inner
180            .last_loop_activity_nanos
181            .store(now, Ordering::SeqCst);
182        self.inner
183            .last_write_success_nanos
184            .store(now, Ordering::SeqCst);
185    }
186
187    pub(super) fn set_connected(&self, connected: bool) {
188        self.inner.connected.store(connected, Ordering::SeqCst);
189        self.mark_loop_activity();
190    }
191}
192
193struct MetricsDisconnectGuard(FrameSessionMetrics);
194
195impl Drop for MetricsDisconnectGuard {
196    fn drop(&mut self) {
197        self.0.set_connected(false);
198    }
199}
200
201/// A frame-mode streaming session.
202///
203/// Owns a scheduler thread that reads frames from a channel and writes them
204/// to the DAC backend using the appropriate strategy (FIFO or frame-swap).
205///
206/// # Example
207///
208/// ```ignore
209/// use laser_dac::{open_device, FrameSessionConfig, Frame, LaserPoint};
210///
211/// let device = open_device("my-device")?;
212/// let config = FrameSessionConfig::new(30_000);
213/// let (session, _info) = device.start_frame_session(config)?;
214///
215/// session.control().arm()?;
216/// session.send_frame(Frame::new(vec![
217///     LaserPoint::new(0.0, 0.0, 65535, 0, 0, 65535),
218/// ]));
219/// ```
220pub struct FrameSession {
221    control: StreamControl,
222    thread: Option<JoinHandle<Result<RunExit>>>,
223    frame_slot: Arc<Mutex<Option<Frame>>>,
224    metrics: FrameSessionMetrics,
225}
226
227impl FrameSession {
228    /// Start a frame session on the given backend.
229    pub(crate) fn start(
230        mut backend: BackendKind,
231        config: FrameSessionConfig,
232        reconnect_policy: Option<ReconnectPolicy>,
233    ) -> Result<Self> {
234        if !backend.is_connected() {
235            backend.connect()?;
236        }
237
238        let (control_tx, control_rx) = mpsc::channel();
239        let initial_color_delay = if config.color_delay_points > 0 {
240            Duration::from_secs_f64(config.color_delay_points as f64 / config.pps as f64)
241        } else {
242            Duration::ZERO
243        };
244        let control = StreamControl::new(control_tx, initial_color_delay, config.pps);
245        let frame_slot: Arc<Mutex<Option<Frame>>> = Arc::new(Mutex::new(None));
246        let metrics = FrameSessionMetrics::new(backend.is_connected());
247
248        let control_clone = control.clone();
249        let slot_clone = frame_slot.clone();
250        let metrics_clone = metrics.clone();
251
252        // Named for diagnosability in profilers / thread dumps.
253        let thread = std::thread::Builder::new()
254            .name("laser-frame-scheduler".to_string())
255            .spawn(move || {
256                // Elevate this thread's scheduling priority so pacing sleeps are
257                // less likely to be preempted under system load. Best-effort:
258                // many systems disallow raising priority without privileges
259                // (e.g. Linux without CAP_SYS_NICE), so we log and continue at
260                // default priority rather than fail the session.
261                if let Err(e) = thread_priority::set_current_thread_priority(
262                    thread_priority::ThreadPriority::Max,
263                ) {
264                    log::warn!("laser-frame-scheduler: could not raise thread priority: {e:?}");
265                }
266                let _disconnect_guard = MetricsDisconnectGuard(metrics_clone.clone());
267                Self::run_loop(
268                    backend,
269                    config,
270                    control_clone,
271                    control_rx,
272                    slot_clone,
273                    metrics_clone,
274                    reconnect_policy,
275                )
276            })
277            .map_err(Error::backend)?;
278
279        Ok(Self {
280            control,
281            thread: Some(thread),
282            frame_slot,
283            metrics,
284        })
285    }
286
287    /// Returns a control handle for arm/disarm/stop.
288    pub fn control(&self) -> StreamControl {
289        self.control.clone()
290    }
291
292    /// Submit a frame for display. Latest-wins: overwrites any unconsumed
293    /// pending frame immediately, with no buffering or memory growth.
294    pub fn send_frame(&self, frame: Frame) {
295        *self.frame_slot.lock().unwrap() = Some(frame);
296    }
297
298    /// Returns true if the scheduler thread has finished.
299    pub fn is_finished(&self) -> bool {
300        self.thread.as_ref().is_some_and(|h| h.is_finished())
301    }
302
303    /// Returns a metrics handle for observing scheduler liveness.
304    pub fn metrics(&self) -> FrameSessionMetrics {
305        self.metrics.clone()
306    }
307
308    /// Wait for the session thread to finish and return the exit reason.
309    pub fn join(mut self) -> Result<RunExit> {
310        if let Some(handle) = self.thread.take() {
311            handle
312                .join()
313                .unwrap_or(Err(Error::disconnected("thread panicked")))
314        } else {
315            Ok(RunExit::Stopped)
316        }
317    }
318
319    // =========================================================================
320    // Driver loop
321    // =========================================================================
322
323    fn run_loop(
324        mut backend: BackendKind,
325        config: FrameSessionConfig,
326        control: StreamControl,
327        control_rx: mpsc::Receiver<ControlMsg>,
328        frame_slot: Arc<Mutex<Option<Frame>>>,
329        metrics: FrameSessionMetrics,
330        reconnect_policy: Option<ReconnectPolicy>,
331    ) -> Result<RunExit> {
332        let FrameSessionConfig {
333            pps: _,
334            transition_fn,
335            startup_blank,
336            color_delay_points,
337            idle_policy,
338            output_filter,
339            reconnect: _,
340        } = config;
341
342        let mut engine = PresentationEngine::new(transition_fn);
343        if backend.is_frame_swap() {
344            engine.set_frame_capacity(backend.frame_capacity());
345        }
346
347        // Per-OutputModel initial buffer sizing: FIFO bounds by max_points_per_chunk;
348        // frame-swap bounds by frame_capacity (max_points_per_chunk is meaningless there).
349        let initial_buf_capacity = match backend.caps().output_model {
350            OutputModel::UsbFrameSwap => backend.frame_capacity().unwrap_or(0),
351            OutputModel::NetworkFifo | OutputModel::UdpTimed | OutputModel::BlockingFifo => {
352                backend.caps().max_points_per_chunk
353            }
354        };
355        let mut pipeline = SlicePipeline::with_startup_blank(
356            engine,
357            color_delay_points,
358            output_filter,
359            idle_policy,
360            initial_buf_capacity,
361            startup_blank,
362        );
363        pipeline.reset_output_filter(OutputResetReason::SessionStart);
364
365        let expected_frame_swap = backend.is_frame_swap();
366        let source: SourceOwned = if expected_frame_swap {
367            SourceOwned::Frame(Box::new(pipeline))
368        } else {
369            SourceOwned::Fifo(Box::new(pipeline))
370        };
371
372        let validator = Self::reconnect_validator(reconnect_policy.as_ref());
373        if !backend.is_connected() {
374            backend.connect()?;
375        }
376
377        let target_buffer = target_buffer_for_backend(&backend);
378
379        driver::run(DriverInputs {
380            backend,
381            source,
382            control,
383            control_rx,
384            metrics,
385            reconnect_policy,
386            validator,
387            error_sink: Box::new(|_e: Error| { /* frame-mode swallows non-fatal errors */ }),
388            target_buffer,
389            drain_timeout: Duration::ZERO,
390            pending_frame: Some(frame_slot),
391            clock: DriverInputs::system_clock(),
392        })
393    }
394
395    fn reconnect_validator(policy: Option<&ReconnectPolicy>) -> driver::ReconnectValidator {
396        let target_id = policy
397            .map(|p| p.target.device_id.clone())
398            .unwrap_or_default();
399        Box::new(move |info: &DacInfo, _backend: &BackendKind, pps: u32| {
400            if pps < info.caps.pps_min || pps > info.caps.pps_max {
401                log::error!(
402                    "'{}' PPS {} outside new device range [{}, {}]",
403                    target_id,
404                    pps,
405                    info.caps.pps_min,
406                    info.caps.pps_max
407                );
408                return Err(RunExit::Disconnected);
409            }
410            Ok(())
411        })
412    }
413}
414
415impl Drop for FrameSession {
416    fn drop(&mut self) {
417        let _ = self.control.stop();
418        if let Some(handle) = self.thread.take() {
419            let _ = handle.join();
420        }
421    }
422}
423
424pub(super) fn target_buffer_for_backend(backend: &BackendKind) -> Duration {
425    // Delegate to the shared policy so the frame path and the stream path's
426    // `apply_backend_buffer_defaults` can never disagree on the cushion.
427    StreamConfig::default_target_buffer_for(&backend.dac_type(), &backend.caps().output_model)
428}
laser_dac/presentation/session.rs

laser_dac/presentation/
session.rs
