🐛 bug: yep sure enough I wasn't flushing the resampler properly.

I feel like I've done this _exact_ project 3 times now

.gitignore

@@ -1,3 +1,4 @@
 /target
 *.m4a
+*.ogg
 *.opus

Cargo.lock

@@ -24,6 +24,15 @@ dependencies = [
  "pkg-config",
 ]
 
+[[package]]
+name = "ansi_term"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d52a9bb7ec0cf484c551830a7ce27bd20d67eac647e1befb56b0be4ee39a55d2"
+dependencies = [
+ "winapi",
+]
+
 [[package]]
 name = "anyhow"
 version = "1.0.45"
@@ -407,6 +416,15 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "matchers"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8263075bb86c5a1b1427b5ae862e8889656f126e9f77c484496e8b47cf5c5558"
+dependencies = [
+ "regex-automata",
+]
+
 [[package]]
 name = "memchr"
 version = "2.4.1"
@@ -430,6 +448,8 @@ dependencies = [
  "byteorder",
  "cpal",
  "ffmpeg-next",
+ "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]
@@ -602,6 +622,12 @@ dependencies = [
  "cc",
 ]
 
+[[package]]
+name = "once_cell"
+version = "1.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "692fcb63b64b1758029e0a96ee63e049ce8c5948587f2f7208df04625e5f6b56"
+
 [[package]]
 name = "parking_lot"
 version = "0.11.2"
@@ -633,6 +659,12 @@ version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099"
 
+[[package]]
+name = "pin-project-lite"
+version = "0.2.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8d31d11c69a6b52a174b42bdc0c30e5e11670f90788b2c471c31c1d17d449443"
+
 [[package]]
 name = "pkg-config"
 version = "0.3.22"
@@ -694,6 +726,15 @@ dependencies = [
  "regex-syntax",
 ]
 
+[[package]]
+name = "regex-automata"
+version = "0.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132"
+dependencies = [
+ "regex-syntax",
+]
+
 [[package]]
 name = "regex-syntax"
 version = "0.6.25"
@@ -727,6 +768,15 @@ version = "1.0.130"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f12d06de37cf59146fbdecab66aa99f9fe4f78722e3607577a5375d66bd0c913"
 
+[[package]]
+name = "sharded-slab"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "900fba806f70c630b0a382d0d825e17a0f19fcd059a2ade1ff237bcddf446b31"
+dependencies = [
+ "lazy_static",
+]
+
 [[package]]
 name = "shlex"
 version = "0.1.1"
@@ -782,6 +832,15 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "thread_local"
+version = "1.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8018d24e04c95ac8790716a5987d0fec4f8b27249ffa0f7d33f1369bdfb88cbd"
+dependencies = [
+ "once_cell",
+]
+
 [[package]]
 name = "toml"
 version = "0.5.8"
@@ -791,6 +850,67 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "tracing"
+version = "0.1.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "375a639232caf30edfc78e8d89b2d4c375515393e7af7e16f01cd96917fb2105"
+dependencies = [
+ "cfg-if 1.0.0",
+ "pin-project-lite",
+ "tracing-attributes",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-attributes"
+version = "0.1.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f4f480b8f81512e825f337ad51e94c1eb5d3bbdf2b363dcd01e2b19a9ffe3f8e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "tracing-core"
+version = "0.1.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1f4ed65637b8390770814083d20756f87bfa2c21bf2f110babdc5438351746e4"
+dependencies = [
+ "lazy_static",
+]
+
+[[package]]
+name = "tracing-log"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6923477a48e41c1951f1999ef8bb5a3023eb723ceadafe78ffb65dc366761e3"
+dependencies = [
+ "lazy_static",
+ "log",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-subscriber"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "80a4ddde70311d8da398062ecf6fc2c309337de6b0f77d6c27aff8d53f6fca52"
+dependencies = [
+ "ansi_term",
+ "lazy_static",
+ "matchers",
+ "regex",
+ "sharded-slab",
+ "smallvec",
+ "thread_local",
+ "tracing",
+ "tracing-core",
+ "tracing-log",
+]
+
 [[package]]
 name = "unicode-xid"
 version = "0.2.2"

Cargo.toml

@@ -10,3 +10,5 @@ anyhow = "1.0.45"
 byteorder = "1.4.3"
 cpal = "0.13.4"
 ffmpeg-next = "4.4.0"
+tracing = "0.1.29"
+tracing-subscriber = { version = "0.3.1", features = ["env-filter"] }

src/decoder.rs

@@ -0,0 +1,168 @@
+use std::{
+	io::Cursor,
+};
+
+use anyhow::{
+	anyhow,
+	Result,
+};
+
+use byteorder::{
+	LittleEndian,
+	ReadBytesExt,
+};
+
+// This crate flitters between being very convenient and being a type labyrinth.
+
+use ffmpeg_next::{
+	decoder::Audio as DecodeContext,
+	format::context::Input as DemuxContext,
+	software::resampling::Context as ResamplingContext,
+	util::{
+		channel_layout::ChannelLayout,
+		format::sample::{
+			self,
+			Sample,
+		},
+	},
+};
+
+pub const SAMPLE_RATE: u32 = 48000;
+
+#[derive (Default)]
+pub struct PcmBuffers {
+	buffers: Vec <Vec <f32>>,
+	
+	// Always points into the first buffer, if any
+	consumer_cursor: usize,
+}
+
+impl PcmBuffers {
+	pub fn samples_available (&self) -> usize {
+		self.buffers.iter ().map (|b| b.len ()).sum::<usize> () - self.consumer_cursor
+	}
+	
+	#[warn(unused_must_use)]
+	pub fn consume_exact (&mut self, data: &mut [f32]) -> bool {
+		if data.len () > self.samples_available () {
+			return false;
+		}
+		
+		for x in data {
+			if self.consumer_cursor >= self.buffers [0].len () {
+				self.buffers.remove (0);
+				self.consumer_cursor = 0;
+			}
+			
+			*x = self.buffers [0][self.consumer_cursor];
+			self.consumer_cursor += 1;
+		}
+		
+		true
+	}
+	
+	pub fn produce (&mut self, new_buffer: Vec <f32>) {
+		self.buffers.push (new_buffer);
+	}
+	
+	pub fn produce_bytes (&mut self, new_buffer: &[u8]) {
+		let mut b = vec! [0.0f32; new_buffer.len () / 4];
+		let mut rdr = Cursor::new (new_buffer);
+		
+		rdr.read_f32_into::<LittleEndian> (&mut b).unwrap ();
+		
+		self.produce (b);
+	}
+}
+
+#[derive (Default)]
+pub struct SharedState {
+	pub pcm_buffers: PcmBuffers,
+	pub quit: bool,
+}
+
+pub struct Decoder {
+	pub input_ctx: DemuxContext,
+	best_stream_idx: usize,
+	pub decoder: DecodeContext,
+	pub resampler: ResamplingContext,
+}
+
+impl Decoder {
+	pub fn new (filename: &str) -> Result <Self> {
+		let input_ctx = ffmpeg_next::format::input (&filename)?;
+		let stream = input_ctx
+		.streams ()
+		.best (ffmpeg_next::media::Type::Audio)
+		.ok_or_else (|| anyhow! ("can't find good audio stream"))?;
+		let best_stream_idx = stream.index ();
+		
+		let decoder = stream.codec ().decoder ().audio ()?;
+		let resampler = decoder.resampler (
+			Sample::F32 (sample::Type::Packed),
+			ChannelLayout::STEREO,
+			48000,
+		)?;
+		
+		Ok (Self {
+			input_ctx,
+			best_stream_idx,
+			decoder,
+			resampler,
+		})
+	}
+	
+	pub fn fill_buffer (&mut self, pcm_buffers: &mut PcmBuffers) -> Result <bool> {
+		match self.resampler.delay () {
+			Some (x) if x.milliseconds > 500 => {
+				// tracing::trace! ("flushing resampler ({} ms)", x.milliseconds);
+				let mut frame_resampled = ffmpeg_next::util::frame::Audio::empty ();
+				frame_resampled.set_channel_layout (ChannelLayout::STEREO);
+				frame_resampled.set_format (Sample::F32 (sample::Type::Packed));
+				frame_resampled.set_rate (48000);
+				
+				if self.resampler.flush (&mut frame_resampled).is_ok () {
+					pcm_buffers.produce_bytes (frame_resampled.data (0));
+					return Ok (true);
+				}
+				else {
+					tracing::error! ("resampler flushed out a zero-length frame?");
+				}
+			},
+			_ => {},
+		}
+		
+		let mut frame_src = ffmpeg_next::util::frame::Audio::empty ();
+		if self.decoder.receive_frame (&mut frame_src).is_ok () {
+			//eprintln! ("decoder.receive_frame");
+			let mut frame_resampled = ffmpeg_next::util::frame::Audio::empty ();
+			self.resampler.run (&frame_src, &mut frame_resampled)?;
+			pcm_buffers.produce_bytes (frame_resampled.data (0));
+			return Ok (true);
+		}
+		
+		//eprintln! ("Decoder is dry, fetching a new packet...");
+		
+		while let Some ((stream, packet)) = self.input_ctx.packets ().next () {
+			if stream.index () == self.best_stream_idx {
+				//eprintln! ("decoder.send_packet");
+				self.decoder.send_packet (&packet)?;
+				return Ok (true);
+			}
+		}
+		
+		//eprintln! ("Decoder ran out of work");
+		
+		if self.resampler.delay ().is_some () {
+			tracing::trace! ("flushing resampler (out of compressed packets)");
+			let mut frame_resampled = ffmpeg_next::util::frame::Audio::empty ();
+			
+			if self.resampler.flush (&mut frame_resampled).is_ok () {
+				pcm_buffers.produce_bytes (frame_resampled.data (0));
+				return Ok (true);
+			}
+		}
+		
+		Ok (false)
+	}
+}

src/main.rs

@@ -1,128 +1,112 @@
 use std::{
-	io::Cursor,
 	sync::{
 		Arc,
 		Condvar,
 		Mutex,
 	},
-	thread,
+	thread::{
+		self,
+		sleep,
+	},
+	time::Duration,
 };
 
 use anyhow::{
 	anyhow,
-	//bail,
+	bail,
 	Result,
 };
 
-use byteorder::{
-	LittleEndian,
-	ReadBytesExt,
-};
-
 use cpal::traits::{
 	DeviceTrait,
 	HostTrait,
-	StreamTrait,
 };
 
-const SAMPLE_RATE: u32 = 48000;
-
-#[derive (Default)]
-struct PcmBuffers {
-	buffers: Vec <Vec <f32>>,
-	
-	// Always points into the first buffer, if any
-	consumer_cursor: usize,
-}
-
-impl PcmBuffers {
-	fn samples_available (&self) -> usize {
-		self.buffers.iter ().map (|b| b.len ()).sum::<usize> () - self.consumer_cursor
-	}
-	
-	#[warn(unused_must_use)]
-	fn consume_exact (&mut self, data: &mut [f32]) -> bool {
-		if data.len () > self.samples_available () {
-			return false;
-		}
-		
-		for x in data {
-			if self.consumer_cursor >= self.buffers [0].len () {
-				self.buffers.remove (0);
-				self.consumer_cursor = 0;
-			}
-			
-			*x = self.buffers [0][self.consumer_cursor];
-			self.consumer_cursor += 1;
-		}
-		
-		true
-	}
-	
-	fn produce (&mut self, new_buffer: Vec <f32>) {
-		self.buffers.push (new_buffer);
-	}
-	
-	fn produce_bytes (&mut self, new_buffer: &[u8]) {
-		let mut b = vec! [0.0f32; new_buffer.len () / 4];
-		let mut rdr = Cursor::new (new_buffer);
-		
-		rdr.read_f32_into::<LittleEndian> (&mut b).unwrap ();
-		
-		self.produce (b);
-	}
-}
+mod decoder;
 
 fn main () -> Result <()> {
-	let host = cpal::default_host ();
-	let device = host.default_output_device ().ok_or_else (|| anyhow! ("can't open cpal device"))?;
+	let args: Vec <_> = std::env::args ().collect ();
 	
-	let mut supported_configs_range = device.supported_output_configs ()?
-	.filter (|c| c.channels () == 2 && c.sample_format () == cpal::SampleFormat::F32);
+	match args.get (1).map (|s| &s[..]) {
+		None => bail! ("First argument must be a subcommand like `play`"),
+		Some ("debug") => cmd_debug (&args [1..]),
+		Some ("play") => cmd_play (&args [1..]),
+		Some (_) => bail! ("Unrecognized subcommand"),
+	}
+}
+
+fn cmd_debug (args: &[String]) -> Result <()> {
+	tracing_subscriber::fmt::init ();
 	
-	let config = supported_configs_range.next ()
-	.ok_or_else (|| anyhow! ("can't get stereo f32 audio output"))?
-	.with_sample_rate (cpal::SampleRate (SAMPLE_RATE))
-	.config ();
+	let filename = args.get (1)
+	.map (|s| s.to_string ())
+	.unwrap_or_else (|| "test-short.m4a".to_string ());
 	
-	#[derive (Default)]
-	struct DecoderState {
-		pcm_buffers: PcmBuffers,
-		quit: bool,
+	let mut decoder = decoder::Decoder::new (&filename)?;
+	let mut pcm_buffers = decoder::PcmBuffers::default ();
+	
+	tracing::debug! ("Constructed decoder");
+	
+	sleep (Duration::from_secs (3));
+	
+	tracing::debug! ("Decoding...");
+	
+	'decoding: loop {
+		while pcm_buffers.samples_available () < 48_000 {
+			if ! decoder.fill_buffer (&mut pcm_buffers)? {
+				tracing::info! ("Decoder finished");
+				break 'decoding;
+			}
+		}
+		
+		while pcm_buffers.samples_available () > 0 {
+			let mut buf = vec! [0.0f32; pcm_buffers.samples_available ()];
+			pcm_buffers.consume_exact (&mut buf);
+		}
 	}
 	
-	let pair = Arc::new ((Mutex::new (DecoderState::default ()), Condvar::new ()));
+	sleep (Duration::from_secs (3));
+	
+	tracing::debug! ("Dropping resampler...");
+	dbg! (decoder.resampler.delay ());
+	
+	drop (decoder.resampler);
+	sleep (Duration::from_secs (3));
+	
+	tracing::debug! ("Dropping decoder...");
+	
+	drop (decoder.decoder);
+	sleep (Duration::from_secs (3));
+	
+	tracing::debug! ("Dropping input_ctx...");
+	
+	drop (decoder.input_ctx);
+	sleep (Duration::from_secs (3));
+	
+	Ok (())
+}
+	
+fn cmd_play (args: &[String]) -> Result <()> {
+	tracing_subscriber::fmt::init ();
+	
+	let filename = args.get (1)
+	.map (|s| s.to_string ())
+	.unwrap_or_else (|| "test-short.m4a".to_string ());
+	
+	let pair = Arc::new ((Mutex::new (decoder::SharedState::default ()), Condvar::new ()));
 	let pair2 = Arc::clone (&pair);
 	let pair3 = Arc::clone (&pair);
 	
 	let thread_decoder = thread::spawn (move|| {
 		let (lock, cvar) = &*pair2;
 		
-		let mut input_ctx = ffmpeg_next::format::input (&"test.m4a")?;
-		let stream = input_ctx
-		.streams ()
-		.best (ffmpeg_next::media::Type::Audio)
-		.ok_or_else (|| anyhow! ("can't find good audio stream"))?;
-		let best_stream_idx = stream.index ();
-		
-		let mut decoder = stream.codec ().decoder ().audio ()?;
-		let mut resampler = decoder.resampler (
-			ffmpeg_next::util::format::sample::Sample::F32 (
-				ffmpeg_next::util::format::sample::Type::Packed,
-			),
-			ffmpeg_next::util::channel_layout::ChannelLayout::STEREO,
-			48000,
-		)?;
-		
-		let mut frame_src = ffmpeg_next::util::frame::Audio::empty ();
-		let mut frame_resampled = ffmpeg_next::util::frame::Audio::empty ();
-		let mut packets = input_ctx.packets ();
+		let mut decoder = decoder::Decoder::new (&filename)?;
 		
 		'decoder_thread: loop {
-			// eprintln! ("decode thread parking");
+			// tracing::trace! ("decode thread parking");
 			
 			let mut decoder_state = cvar.wait_while (lock.lock ().unwrap (), |decoder_state| {
-				decoder_state.pcm_buffers.samples_available () >= 24_000 &&
+				decoder_state.pcm_buffers.samples_available () >= 12_000 &&
 				! decoder_state.quit
 			}).unwrap ();
 			
@@ -134,85 +118,69 @@ fn main () -> Result <()> {
 			
 			let pcm_buffers = &mut decoder_state.pcm_buffers;
 			
-			'fill_buffer: while pcm_buffers.samples_available () < 48_000 {
-				//eprintln! ("Decoder is trying to work...");
-				
-				match resampler.delay () {
-					Some (x) if x.milliseconds > 500 => {
-						eprintln! ("flushing resampler");
-						if let Some (_) = resampler.flush (&mut frame_resampled)? {
-							pcm_buffers.produce_bytes (frame_resampled.data (0));
-							continue 'fill_buffer;
-						}
-					},
-					_ => {},
+			while pcm_buffers.samples_available () < 24_000 {
+				// tracing::trace! ("Decoder is trying to work...");
+				if ! decoder.fill_buffer (pcm_buffers)? {
+					tracing::info! ("Decoder thread is out of work, quitting");
+					break 'decoder_thread;
 				}
-				
-				if decoder.receive_frame (&mut frame_src).is_ok () {
-					//eprintln! ("decoder.receive_frame");
-					resampler.run (&frame_src, &mut frame_resampled)?;
-					pcm_buffers.produce_bytes (frame_resampled.data (0));
-					continue 'fill_buffer;
-				}
-				
-				//eprintln! ("Decoder is dry, fetching a new packet...");
-				
-				'get_packet: while let Some ((stream, packet)) = packets.next () {
-					if stream.index () == best_stream_idx {
-						//eprintln! ("decoder.send_packet");
-						decoder.send_packet (&packet)?;
-						continue 'fill_buffer;
-					}
-				}
-				
-				//eprintln! ("Decoder ran out of work");
-				
-				if resampler.delay ().is_some () {
-					eprintln! ("flushing resampler");
-					if let Some (_) = resampler.flush (&mut frame_resampled)? {
-						//eprintln! ("resampler.flush");
-						pcm_buffers.produce_bytes (frame_resampled.data (0));
-						continue 'fill_buffer;
-					}
-				}
-				
-				break 'fill_buffer;
 			}
 		}
 		
 		Ok::<_, anyhow::Error> (())
 	});
 	
-	{
-		let stream = device.build_output_stream (
-			&config,
-			move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
-				let (lock, cvar) = &*pair;
+	let host = cpal::default_host ();
+	let device = host.default_output_device ().ok_or_else (|| anyhow! ("can't open cpal device"))?;
+	
+	let mut supported_configs_range = device.supported_output_configs ()?
+	.filter (|c| c.channels () == 2 && c.sample_format () == cpal::SampleFormat::F32);
+	
+	let config = supported_configs_range.next ()
+	.ok_or_else (|| anyhow! ("can't get stereo f32 audio output"))?
+	.with_sample_rate (cpal::SampleRate (decoder::SAMPLE_RATE))
+	.config ();
+	
+	let pcm_quit = Arc::new ((Mutex::new (false), Condvar::new ()));
+	let pcm_quit2 = Arc::clone (&pcm_quit);
+	
+	let stream = device.build_output_stream (
+		&config,
+		move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
+			let (lock, cvar) = &*pair;
+			
+			let mut decoder_state = match lock.lock () {
+				Ok (x) => x,
+				Err (_) => return,
+			};
+			
+			let pcm_buffers = &mut decoder_state.pcm_buffers;
+			
+			if ! pcm_buffers.consume_exact (data) {
+				// tracing::warn! ("PCM buffer underflow");
 				
-				let mut decoder_state = match lock.lock () {
-					Ok (x) => x,
-					Err (_) => return,
-				};
-				
-				let pcm_buffers = &mut decoder_state.pcm_buffers;
-				
-				if ! pcm_buffers.consume_exact (data) {
-					eprintln! ("PCM buffer underflow");
+				for x in data {
+					*x = 0.0;
 				}
 				
+				let (lock, cvar) = &*pcm_quit;
+				let mut pcm_quit = lock.lock ().unwrap ();
+				*pcm_quit = true;
 				cvar.notify_one ();
-			},
-			move |err| {
-				// react to errors here.
-			},
-		);
-		
-		std::thread::sleep (std::time::Duration::from_millis (180_000));
-	}
+			}
+			
+			cvar.notify_one ();
+		},
+		move |_err| {
+			// react to errors here.
+		},
+	);
 	
-	eprintln! ("Joining decoder thread");
+	// sleep (std::time::Duration::from_secs (3 * 60 + 40));
 	
-	{
+	tracing::debug! ("Joining decoder thread...");
+	
+	if false {
 		let mut decoder_state = pair3.0.lock ().unwrap ();
 		decoder_state.quit = true;
 		pair3.1.notify_one ();
@@ -220,8 +188,16 @@ fn main () -> Result <()> {
 	
 	thread_decoder.join ().unwrap ()?;
 	
-	eprintln! ("Joined decoder thread");
+	tracing::debug! ("Joining PCM thread...");
 	
+	let (lock, cvar) = &*pcm_quit2;
+	let _ = cvar.wait (lock.lock ().unwrap ()).unwrap ();
+	
+	drop (stream);
+	
+	sleep (Duration::from_secs (1));
+	
+	tracing::info! ("Exiting cleanly.");
 	Ok (())
 }