dspk_stream/transport/
remote_signal.rs

1//! A recording on a server as a viewer sees it: [`RemoteSignal`] is a [`SignalBackend`] whose
2//! views are answered by the server (from its own pyramid), so a viewer draws a remote recording
3//! as it draws a local one, and nothing is copied whole. Calls block (as the trait says: viewers
4//! call them from worker threads) on the session's async runtime.
5
6use std::ops::Range;
7use std::sync::atomic::{AtomicBool, Ordering};
8use std::sync::{Arc, Mutex};
9use std::time::Duration;
10
11use dspk_core::{DspError, DspResult, RecordingInfo, RecordingSource};
12use dspk_view::{Envelope, OnProgress, Progress, SignalBackend, View};
13
14use super::client::Session;
15use super::server::{MAX_VIEW_COLUMNS, MAX_VIEW_VALUES};
16use crate::error::StreamError;
17
18/// How often an incomplete remote pyramid is asked about again.
19const PYRAMID_POLL: Duration = Duration::from_secs(1);
20/// Columns of the small view that asks whether the server's pyramid is complete.
21const PROBE_COLUMNS: usize = 64;
22
23fn dsp(e: StreamError) -> DspError {
24    DspError::Io(format!("remote: {e}"))
25}
26
27/// The recording a [`Session`] has open, as a [`RecordingSource`]: samples (scaled `f32`) read on
28/// the server as sample views, in pieces under its limits. For small reads (a window of
29/// waveforms, the value under the cursor); a whole recording belongs on the server.
30pub struct RemoteSource {
31    handle: tokio::runtime::Handle,
32    session: Arc<Session>,
33    info: RecordingInfo,
34}
35
36impl RemoteSource {
37    /// The recording `session` has open; calls run on `handle` (the runtime that drives the
38    /// session's connection).
39    pub fn new(session: Arc<Session>, handle: tokio::runtime::Handle) -> Result<Self, StreamError> {
40        let info = session.recording().cloned().ok_or_else(|| StreamError::protocol("no recording is open in this session"))?;
41        Ok(Self { handle, session, info })
42    }
43
44    /// Samples of `channels` channels one request reads: a sample view of c channels and n
45    /// samples is c × n values and n columns, both under the server's limits.
46    fn samples_per_request(channels: usize) -> u64 {
47        (MAX_VIEW_VALUES / channels.max(1) as u64).min(u64::from(MAX_VIEW_COLUMNS)).max(1)
48    }
49}
50
51impl RecordingSource for RemoteSource {
52    fn info(&self) -> &RecordingInfo {
53        &self.info
54    }
55
56    /// Samples of `channels`, channel-major, fetched as sample views and reassembled.
57    fn read(&self, channels: &[usize], samples: Range<u64>, out: &mut [f32]) -> DspResult<()> {
58        let n = (samples.end.saturating_sub(samples.start)) as usize;
59        if out.len() != channels.len() * n {
60            return Err(DspError::ShapeMismatch { expected: vec![channels.len(), n], actual: vec![out.len()] });
61        }
62        let step = Self::samples_per_request(channels.len());
63        let mut start = samples.start;
64        while start < samples.end {
65            let end = (start + step).min(samples.end);
66            let piece = (end - start) as usize;
67            let view = View { channels: channels.to_vec(), start, end, width: piece };
68            let values = match self.handle.block_on(self.session.view(&view)).map_err(dsp)? {
69                Envelope::Samples(values) => values,
70                Envelope::Columns { .. } => return Err(DspError::Io("remote: the server answered columns for a sample read".into())),
71            };
72            let offset = (start - samples.start) as usize;
73            for (c, row) in values.chunks_exact(piece).enumerate() {
74                out[c * n + offset..c * n + offset + piece].copy_from_slice(row);
75            }
76            start = end;
77        }
78        Ok(())
79    }
80}
81
82/// The recording a [`Session`] has open, behind [`SignalBackend`].
83pub struct RemoteSignal {
84    handle: tokio::runtime::Handle,
85    session: Arc<Session>,
86    source: Arc<RemoteSource>,
87    info: RecordingInfo,
88    /// Whether the server's pyramid answered complete views (until then views may hold NaN).
89    complete: Arc<AtomicBool>,
90    listeners: Arc<Mutex<Vec<OnProgress>>>,
91    polling: Arc<AtomicBool>,
92}
93
94impl RemoteSignal {
95    /// The recording `session` has open; calls run on `handle` (the runtime that drives the
96    /// session's connection).
97    pub fn new(session: Arc<Session>, handle: tokio::runtime::Handle) -> Result<Self, StreamError> {
98        let source = Arc::new(RemoteSource::new(session.clone(), handle.clone())?);
99        let info = source.info().clone();
100        Ok(Self {
101            handle,
102            session,
103            source,
104            info,
105            complete: Arc::new(AtomicBool::new(false)),
106            listeners: Arc::new(Mutex::new(Vec::new())),
107            polling: Arc::new(AtomicBool::new(false)),
108        })
109    }
110
111    /// The same recording as a [`RecordingSource`] (sample reads), sharing the session.
112    pub fn source(&self) -> Arc<RemoteSource> {
113        self.source.clone()
114    }
115
116    /// Asks the server again, every [`PYRAMID_POLL`], until its pyramid answers complete views,
117    /// telling the listeners each time (viewers then ask for their views again).
118    fn poll_until_complete(&self) {
119        if self.polling.swap(true, Ordering::Relaxed) {
120            return;
121        }
122        let (handle, session, complete, listeners, polling) = (self.handle.clone(), self.session.clone(), self.complete.clone(), self.listeners.clone(), self.polling.clone());
123        let total = self.info.samples;
124        std::thread::spawn(move || {
125            let probe = View { channels: vec![0], start: 0, end: total, width: PROBE_COLUMNS };
126            while !complete.load(Ordering::Relaxed) {
127                std::thread::sleep(PYRAMID_POLL);
128                let done = match handle.block_on(session.view(&probe)) {
129                    Ok(Envelope::Columns { complete: true, .. }) | Ok(Envelope::Samples(_)) => {
130                        complete.store(true, Ordering::Relaxed);
131                        total
132                    }
133                    Ok(_) => 0,
134                    // The session is gone: stop asking
135                    Err(_) => break,
136                };
137                for listener in listeners.lock().unwrap_or_else(|e| e.into_inner()).iter() {
138                    listener(Progress { filled: 0..0, done, total });
139                }
140            }
141            polling.store(false, Ordering::Relaxed);
142        });
143    }
144}
145
146impl SignalBackend for RemoteSignal {
147    fn info(&self) -> &RecordingInfo {
148        &self.info
149    }
150
151    fn view(&self, view: &View, out: &mut Envelope) -> DspResult<()> {
152        let envelope = self.handle.block_on(self.session.view(view)).map_err(dsp)?;
153        match &envelope {
154            Envelope::Columns { complete: false, .. } => {
155                self.complete.store(false, Ordering::Relaxed);
156                self.poll_until_complete();
157            }
158            Envelope::Columns { complete: true, .. } => self.complete.store(true, Ordering::Relaxed),
159            Envelope::Samples(_) => {}
160        }
161        *out = envelope;
162        Ok(())
163    }
164
165    /// Samples of `channels`, channel-major (through [`RemoteSource`]).
166    fn read(&self, channels: &[usize], samples: Range<u64>, out: &mut [f32]) -> DspResult<()> {
167        self.source.read(channels, samples, out)
168    }
169
170    /// The server builds its own pyramid: nothing to steer from here.
171    fn focus(&self, _sample: u64) {}
172
173    fn prepare(&self, _start: u64, _end: u64) -> DspResult<()> {
174        Ok(())
175    }
176
177    fn progress(&self) -> Progress {
178        let total = self.info.samples;
179        Progress { filled: 0..0, done: if self.complete.load(Ordering::Relaxed) { total } else { 0 }, total }
180    }
181
182    fn watch(&self, on_progress: OnProgress) {
183        self.listeners.lock().unwrap_or_else(|e| e.into_inner()).push(on_progress);
184    }
185}
dspk_stream/transport/remote_signal.rs

dspk_stream/transport/
remote_signal.rs
