nandi/jolt-native · stream.rs

⑂ cfd3e3677bed92e80cfe447469a249cfdc0b1401

stream.rs · 309 lines · 11.5 KBRust    Blame HistoryRaw
Lift freeq's AV media plane out of sleek   1
  sync::{
      atomic::{self, AtomicU64},
      Arc, Mutex,
  time::{Duration, Instant},
9use futures::executor::block_on;
10use pulseaudio::{protocol, AsPlaybackSource};
12use crate::{
  traits::StreamTrait, BackendSpecificError, BuildStreamError, Data, FrameCount,
  InputCallbackInfo, InputStreamTimestamp, OutputCallbackInfo, OutputStreamTimestamp,
  PlayStreamError, SampleFormat, StreamError, StreamInstant,
18const LATENCY_POLL_INTERVAL: Duration = Duration::from_millis(5);
20pub enum Stream {
  Playback(pulseaudio::PlaybackStream),
  Record(pulseaudio::RecordStream),
25impl StreamTrait for Stream {
  fn play(&self) -> Result<(), PlayStreamError> {
      match self {
          Stream::Playback(stream) => {
              block_on(stream.uncork()).map_err(Into::<BackendSpecificError>::into)?;
          Stream::Record(stream) => {
              block_on(stream.uncork()).map_err(Into::<BackendSpecificError>::into)?;
              block_on(stream.started()).map_err(Into::<BackendSpecificError>::into)?;
      Ok(())
  fn pause(&self) -> Result<(), crate::PauseStreamError> {
      let res = match self {
          Stream::Playback(stream) => block_on(stream.cork()),
          Stream::Record(stream) => block_on(stream.cork()),
      res.map_err(Into::<BackendSpecificError>::into)?;
      Ok(())
  fn buffer_size(&self) -> Option<FrameCount> {
      let (spec, bytes) = match self {
          Stream::Playback(s) => (
              s.sample_spec(),
              s.buffer_attr().minimum_request_length as usize,
          Stream::Record(s) => (s.sample_spec(), s.buffer_attr().fragment_size as usize),
      let frame_size = spec.channels as usize * spec.format.bytes_per_sample();
      if bytes > 0 {
          Some((bytes / frame_size) as _)
      } else {
          None
67impl Stream {
  pub fn new_playback<D, E>(
      client: pulseaudio::Client,
      params: protocol::PlaybackStreamParams,
      mut data_callback: D,
      error_callback: E,
  ) -> Result<Self, BuildStreamError>
74
      D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
      E: FnMut(StreamError) + Send + 'static,
      // Use a monotonic clock relative to stream creation for StreamInstants.
      let start = std::time::Instant::now();
      let current_latency_micros = Arc::new(AtomicU64::new(0));
      // Microseconds since stream creation at the time of the last latency poll, used
      // to interpolate the latency between polls.
      let last_poll_micros = Arc::new(AtomicU64::new(0));
      let latency_clone = current_latency_micros.clone();
      let poll_clone = last_poll_micros.clone();
      let sample_spec = params.sample_spec;
      let format: SampleFormat = sample_spec
          .format
          .try_into()
          .map_err(|_| BuildStreamError::StreamConfigNotSupported)?;
      // Silence for unsigned formats is the midpoint, not zero. Among
      // PulseAudio's supported formats, only U8 is unsigned and has a
      // single-byte repeatable silence representation (0x80). Multi-byte
      // unsigned formats (U16, U32, ...) are not currently supported.
      let silence_byte = if format == SampleFormat::U8 {
      } else {
      // Wrap the write callback to match the pulseaudio signature.
      let callback = move |buf: &mut [u8]| {
          let elapsed = Instant::now().saturating_duration_since(start);
          let elapsed_usec = elapsed.as_micros() as u64;
          // Interpolate the latency based on elapsed time since the last
          // poll: as audio plays, the DAC drains the buffer at a constant
          // rate, so the latency decreases linearly between polls.
          let stored_latency = latency_clone.load(atomic::Ordering::Relaxed);
          let poll_usec = poll_clone.load(atomic::Ordering::Relaxed);
          // Cap to one poll interval: the linear-drain assumption is only valid
          // for that window, and a stale poll_usec (e.g. after cork/uncork where
          // timing_info blocks) would otherwise saturate latency to zero.
          let elapsed_since_poll = elapsed_usec
              .saturating_sub(poll_usec)
              .min(LATENCY_POLL_INTERVAL.as_micros() as u64);
          let latency = stored_latency.saturating_sub(elapsed_since_poll);
          let playback_time = elapsed + Duration::from_micros(latency);
          let timestamp = OutputStreamTimestamp {
              callback: StreamInstant {
                  secs: elapsed.as_secs() as i64,
                  nanos: elapsed.subsec_nanos(),
              playback: StreamInstant {
                  secs: playback_time.as_secs() as i64,
                  nanos: playback_time.subsec_nanos(),
          // Preemptively fill the buffer with silence in case the user
          // callback doesn't fill it completely (cpal's API doesn't allow
          // short writes).
          buf.fill(silence_byte);
          let bps = sample_spec.format.bytes_per_sample();
          let n_samples = buf.len() / bps;
          // SAFETY: we calculated the number of samples based on
          // `sample_spec.format`, and `format` is directly derived from (and
          // equivalent to) `sample_spec.format`.
          let mut data = unsafe { Data::from_parts(buf.as_mut_ptr().cast(), n_samples, format) };
          data_callback(&mut data, &OutputCallbackInfo { timestamp });
          // We always consider the full buffer filled, because cpal's
          // user-facing API doesn't allow short writes.
          buf.len()
      let stream = block_on(client.create_playback_stream(params, callback.as_playback_source()))
          .map_err(Into::<BackendSpecificError>::into)?;
      // Share the error callback between the worker and latency threads so
      // both can surface errors to the user.
      let error_callback = Arc::new(Mutex::new(error_callback));
      // Spawn a thread to drive the stream future. It will exit automatically
      // when the stream is stopped by the user.
      let stream_clone = stream.clone();
      let error_callback_clone = error_callback.clone();
      std::thread::spawn(move || {
          if let Err(e) = block_on(stream_clone.play_all()) {
              error_callback_clone.lock().unwrap()(StreamError::from(BackendSpecificError {
                  description: e.to_string(),
      });
      // Spawn a thread to monitor the stream's latency in a loop. It will
      // exit automatically when the stream ends.
      let stream_clone = stream.clone();
      let latency_clone = current_latency_micros.clone();
      let poll_clone = last_poll_micros.clone();
      std::thread::spawn(move || loop {
          let timing_info = match block_on(stream_clone.timing_info()) {
              Ok(timing_info) => timing_info,
              Err(e) => {
                  error_callback.lock().unwrap()(StreamError::from(BackendSpecificError {
                      description: e.to_string(),
                  break;
          let poll_since_epoch =
              Instant::now().saturating_duration_since(start).as_micros() as u64;
          poll_clone.store(poll_since_epoch, atomic::Ordering::Relaxed);
          store_latency(
              &latency_clone,
              sample_spec,
              timing_info.sink_usec,
              timing_info.write_offset,
              timing_info.read_offset,
          std::thread::sleep(LATENCY_POLL_INTERVAL);
      });
      Ok(Self::Playback(stream))
  pub fn new_record<D, E>(
      client: pulseaudio::Client,
      params: protocol::RecordStreamParams,
      mut data_callback: D,
      mut error_callback: E,
  ) -> Result<Self, BuildStreamError>
214
      D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
      E: FnMut(StreamError) + Send + 'static,
      let start = Instant::now();
      let current_latency_micros = Arc::new(AtomicU64::new(0));
      let latency_clone = current_latency_micros.clone();
      let sample_spec = params.sample_spec;
      let format: SampleFormat = sample_spec
          .format
          .try_into()
          .map_err(|_| BuildStreamError::StreamConfigNotSupported)?;
      let callback = move |buf: &[u8]| {
          let elapsed = Instant::now().saturating_duration_since(start);
          let latency = latency_clone.load(atomic::Ordering::Relaxed);
          let capture_time = elapsed
              .checked_sub(Duration::from_micros(latency))
              .unwrap_or_default();
          let timestamp = InputStreamTimestamp {
              callback: StreamInstant {
                  secs: elapsed.as_secs() as i64,
                  nanos: elapsed.subsec_nanos(),
              capture: StreamInstant {
                  secs: capture_time.as_secs() as i64,
                  nanos: capture_time.subsec_nanos(),
          let bps = sample_spec.format.bytes_per_sample();
          let n_samples = buf.len() / bps;
          // SAFETY: we calculated the number of samples based on
          // `sample_spec.format`, and `format` is directly derived from (and
          // equivalent to) `sample_spec.format`. The pointer is cast from
          // *const to *mut, but cpal's Data type for input streams only
          // exposes shared references (&[T]), so no mutation occurs.
          let data = unsafe { Data::from_parts(buf.as_ptr() as *mut _, n_samples, format) };
          data_callback(&data, &InputCallbackInfo { timestamp });
      let stream = block_on(client.create_record_stream(params, callback))
          .map_err(Into::<BackendSpecificError>::into)?;
      // Spawn a thread to monitor the stream's latency in a loop. It will
      // exit automatically when the stream ends.
      let stream_clone = stream.clone();
      let latency_clone = current_latency_micros.clone();
      std::thread::spawn(move || loop {
          let timing_info = match block_on(stream_clone.timing_info()) {
              Ok(timing_info) => timing_info,
              Err(e) => {
                  error_callback(StreamError::from(BackendSpecificError {
                      description: e.to_string(),
                  break;
          store_latency(
              &latency_clone,
              sample_spec,
              timing_info.source_usec,
              timing_info.write_offset,
              timing_info.read_offset,
          std::thread::sleep(LATENCY_POLL_INTERVAL);
      });
      Ok(Self::Record(stream))
293fn store_latency(
  latency_micros: &AtomicU64,
  sample_spec: protocol::SampleSpec,
  device_latency_usec: u64,
  write_offset: i64,
  read_offset: i64,
  let offset = (write_offset - read_offset).max(0) as u64;
  let latency =
      Duration::from_micros(device_latency_usec) + sample_spec.bytes_to_duration(offset as usize);
  latency_micros.store(
      latency.as_micros().try_into().unwrap_or(u64::MAX),
      atomic::Ordering::Relaxed,

Lift freeq's AV media plane out of sleek 90f8b89 nandi 19d ago	1	use std::{
	2	sync::{
	3	atomic::{self, AtomicU64},
	4	Arc, Mutex,
	5	},
	6	time::{Duration, Instant},
	7	};
	8
	9	use futures::executor::block_on;
	10	use pulseaudio::{protocol, AsPlaybackSource};
	11
	12	use crate::{
	13	traits::StreamTrait, BackendSpecificError, BuildStreamError, Data, FrameCount,
	14	InputCallbackInfo, InputStreamTimestamp, OutputCallbackInfo, OutputStreamTimestamp,
	15	PlayStreamError, SampleFormat, StreamError, StreamInstant,
	16	};
	17
	18	const LATENCY_POLL_INTERVAL: Duration = Duration::from_millis(5);
	19
	20	pub enum Stream {
	21	Playback(pulseaudio::PlaybackStream),
	22	Record(pulseaudio::RecordStream),
	23	}
	24
	25	impl StreamTrait for Stream {
	26	fn play(&self) -> Result<(), PlayStreamError> {
	27	match self {
	28	Stream::Playback(stream) => {
	29	block_on(stream.uncork()).map_err(Into::<BackendSpecificError>::into)?;
	30	}
	31	Stream::Record(stream) => {
	32	block_on(stream.uncork()).map_err(Into::<BackendSpecificError>::into)?;
	33	block_on(stream.started()).map_err(Into::<BackendSpecificError>::into)?;
	34	}
	35	};
	36
	37	Ok(())
	38	}
	39
	40	fn pause(&self) -> Result<(), crate::PauseStreamError> {
	41	let res = match self {
	42	Stream::Playback(stream) => block_on(stream.cork()),
	43	Stream::Record(stream) => block_on(stream.cork()),
	44	};
	45
	46	res.map_err(Into::<BackendSpecificError>::into)?;
	47	Ok(())
	48	}
	49
	50	fn buffer_size(&self) -> Option<FrameCount> {
	51	let (spec, bytes) = match self {
	52	Stream::Playback(s) => (
	53	s.sample_spec(),
	54	s.buffer_attr().minimum_request_length as usize,
	55	),
	56	Stream::Record(s) => (s.sample_spec(), s.buffer_attr().fragment_size as usize),
	57	};
	58	let frame_size = spec.channels as usize * spec.format.bytes_per_sample();
	59	if bytes > 0 {
	60	Some((bytes / frame_size) as _)
	61	} else {
	62	None
	63	}
	64	}
65	}
66
67	impl Stream {
68	pub fn new_playback<D, E>(
69	client: pulseaudio::Client,
70	params: protocol::PlaybackStreamParams,
71	mut data_callback: D,
72	error_callback: E,
73	) -> Result<Self, BuildStreamError>
74	where
75	D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
76	E: FnMut(StreamError) + Send + 'static,
77	{
78	// Use a monotonic clock relative to stream creation for StreamInstants.
79	let start = std::time::Instant::now();
80
81	let current_latency_micros = Arc::new(AtomicU64::new(0));
82	// Microseconds since stream creation at the time of the last latency poll, used
83	// to interpolate the latency between polls.
84	let last_poll_micros = Arc::new(AtomicU64::new(0));
85	let latency_clone = current_latency_micros.clone();
86	let poll_clone = last_poll_micros.clone();
87	let sample_spec = params.sample_spec;
88
89	let format: SampleFormat = sample_spec
90	.format
91	.try_into()
92	.map_err(\|_\| BuildStreamError::StreamConfigNotSupported)?;
93
94	// Silence for unsigned formats is the midpoint, not zero. Among
95	// PulseAudio's supported formats, only U8 is unsigned and has a
96	// single-byte repeatable silence representation (0x80). Multi-byte
97	// unsigned formats (U16, U32, ...) are not currently supported.
98	let silence_byte = if format == SampleFormat::U8 {
99	0x80u8
100	} else {
101	0u8
102	};
103
104	// Wrap the write callback to match the pulseaudio signature.
105	let callback = move \|buf: &mut [u8]\| {
106	let elapsed = Instant::now().saturating_duration_since(start);
107	let elapsed_usec = elapsed.as_micros() as u64;
108
109	// Interpolate the latency based on elapsed time since the last
110	// poll: as audio plays, the DAC drains the buffer at a constant
111	// rate, so the latency decreases linearly between polls.
112	let stored_latency = latency_clone.load(atomic::Ordering::Relaxed);
113	let poll_usec = poll_clone.load(atomic::Ordering::Relaxed);
114	// Cap to one poll interval: the linear-drain assumption is only valid
115	// for that window, and a stale poll_usec (e.g. after cork/uncork where
116	// timing_info blocks) would otherwise saturate latency to zero.
117	let elapsed_since_poll = elapsed_usec
118	.saturating_sub(poll_usec)
119	.min(LATENCY_POLL_INTERVAL.as_micros() as u64);
120	let latency = stored_latency.saturating_sub(elapsed_since_poll);
121
122	let playback_time = elapsed + Duration::from_micros(latency);
123
124	let timestamp = OutputStreamTimestamp {
125	callback: StreamInstant {
126	secs: elapsed.as_secs() as i64,
127	nanos: elapsed.subsec_nanos(),
128	},
129	playback: StreamInstant {
130	secs: playback_time.as_secs() as i64,
131	nanos: playback_time.subsec_nanos(),
132	},
133	};
134
135	// Preemptively fill the buffer with silence in case the user
136	// callback doesn't fill it completely (cpal's API doesn't allow
137	// short writes).
138	buf.fill(silence_byte);
139
140	let bps = sample_spec.format.bytes_per_sample();
141	let n_samples = buf.len() / bps;
142
143	// SAFETY: we calculated the number of samples based on
144	// `sample_spec.format`, and `format` is directly derived from (and
145	// equivalent to) `sample_spec.format`.
146	let mut data = unsafe { Data::from_parts(buf.as_mut_ptr().cast(), n_samples, format) };
147
148	data_callback(&mut data, &OutputCallbackInfo { timestamp });
149
150	// We always consider the full buffer filled, because cpal's
151	// user-facing API doesn't allow short writes.
152	buf.len()
153	};
154
155	let stream = block_on(client.create_playback_stream(params, callback.as_playback_source()))
156	.map_err(Into::<BackendSpecificError>::into)?;
157
158	// Share the error callback between the worker and latency threads so
159	// both can surface errors to the user.
160	let error_callback = Arc::new(Mutex::new(error_callback));
161
162	// Spawn a thread to drive the stream future. It will exit automatically
163	// when the stream is stopped by the user.
164	let stream_clone = stream.clone();
165	let error_callback_clone = error_callback.clone();
166	std::thread::spawn(move \|\| {
167	if let Err(e) = block_on(stream_clone.play_all()) {
168	error_callback_clone.lock().unwrap()(StreamError::from(BackendSpecificError {
169	description: e.to_string(),
170	}));
171	}
172	});
173
174	// Spawn a thread to monitor the stream's latency in a loop. It will
175	// exit automatically when the stream ends.
176	let stream_clone = stream.clone();
177	let latency_clone = current_latency_micros.clone();
178	let poll_clone = last_poll_micros.clone();
179	std::thread::spawn(move \|\| loop {
180	let timing_info = match block_on(stream_clone.timing_info()) {
181	Ok(timing_info) => timing_info,
182	Err(e) => {
183	error_callback.lock().unwrap()(StreamError::from(BackendSpecificError {
184	description: e.to_string(),
185	}));
186	break;
187	}
188	};
189
190	let poll_since_epoch =
191	Instant::now().saturating_duration_since(start).as_micros() as u64;
192	poll_clone.store(poll_since_epoch, atomic::Ordering::Relaxed);
193
194	store_latency(
195	&latency_clone,
196	sample_spec,
197	timing_info.sink_usec,
198	timing_info.write_offset,
199	timing_info.read_offset,
200	);
201
202	std::thread::sleep(LATENCY_POLL_INTERVAL);
203	});
204
205	Ok(Self::Playback(stream))
206	}
207
208	pub fn new_record<D, E>(
209	client: pulseaudio::Client,
210	params: protocol::RecordStreamParams,
211	mut data_callback: D,
212	mut error_callback: E,
213	) -> Result<Self, BuildStreamError>
214	where
215	D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
216	E: FnMut(StreamError) + Send + 'static,
217	{
218	let start = Instant::now();
219
220	let current_latency_micros = Arc::new(AtomicU64::new(0));
221	let latency_clone = current_latency_micros.clone();
222	let sample_spec = params.sample_spec;
223
224	let format: SampleFormat = sample_spec
225	.format
226	.try_into()
227	.map_err(\|_\| BuildStreamError::StreamConfigNotSupported)?;
228
229	let callback = move \|buf: &[u8]\| {
230	let elapsed = Instant::now().saturating_duration_since(start);
231	let latency = latency_clone.load(atomic::Ordering::Relaxed);
232	let capture_time = elapsed
233	.checked_sub(Duration::from_micros(latency))
234	.unwrap_or_default();
235
236	let timestamp = InputStreamTimestamp {
237	callback: StreamInstant {
238	secs: elapsed.as_secs() as i64,
239	nanos: elapsed.subsec_nanos(),
240	},
241	capture: StreamInstant {
242	secs: capture_time.as_secs() as i64,
243	nanos: capture_time.subsec_nanos(),
244	},
245	};
246
247	let bps = sample_spec.format.bytes_per_sample();
248	let n_samples = buf.len() / bps;
249
250	// SAFETY: we calculated the number of samples based on
251	// `sample_spec.format`, and `format` is directly derived from (and
252	// equivalent to) `sample_spec.format`. The pointer is cast from
253	// const to mut, but cpal's Data type for input streams only
254	// exposes shared references (&[T]), so no mutation occurs.
255	let data = unsafe { Data::from_parts(buf.as_ptr() as *mut _, n_samples, format) };
256
257	data_callback(&data, &InputCallbackInfo { timestamp });
258	};
259
260	let stream = block_on(client.create_record_stream(params, callback))
261	.map_err(Into::<BackendSpecificError>::into)?;
262
263	// Spawn a thread to monitor the stream's latency in a loop. It will
264	// exit automatically when the stream ends.
265	let stream_clone = stream.clone();
266	let latency_clone = current_latency_micros.clone();
267	std::thread::spawn(move \|\| loop {
268	let timing_info = match block_on(stream_clone.timing_info()) {
269	Ok(timing_info) => timing_info,
270	Err(e) => {
271	error_callback(StreamError::from(BackendSpecificError {
272	description: e.to_string(),
273	}));
274	break;
275	}
276	};
277
278	store_latency(
279	&latency_clone,
280	sample_spec,
281	timing_info.source_usec,
282	timing_info.write_offset,
283	timing_info.read_offset,
284	);
285
286	std::thread::sleep(LATENCY_POLL_INTERVAL);
287	});
288
289	Ok(Self::Record(stream))
290	}
291	}
292
293	fn store_latency(
294	latency_micros: &AtomicU64,
295	sample_spec: protocol::SampleSpec,
296	device_latency_usec: u64,
297	write_offset: i64,
298	read_offset: i64,
299	) {
300	let offset = (write_offset - read_offset).max(0) as u64;
301
302	let latency =
303	Duration::from_micros(device_latency_usec) + sample_spec.bytes_to_duration(offset as usize);
304
305	latency_micros.store(
306	latency.as_micros().try_into().unwrap_or(u64::MAX),
307	atomic::Ordering::Relaxed,
308	);
309	}