Noticeable delay between playing (streaming) different, in-memory inputs.

[Pranav-Badrinathan]

I am streaming my data to the discord bot, pre-encoded in Opus and with the correct DCA1 headers and metadata. This data can change in real time, and as such is sent to songbird in ~300 ms packets (each packets has multiple frames, all with correct size headers attached to them).

In 0.3.2 I had the streaming working seamlessly after some help at #192. However, now in 0.4.0, there seems to be additional delay between the playing of each of these packets. To rule out any possibility of this delay being caused by my implementation, I essentially queued up 10-20 of these packets and played them back, and it does have a noticeable delay between playing each of the 300ms packets.

Could this be due to using DCA1 as the use of DCA0 is now not supported? In 0.3.2, you could specify the Container and Codec yourself, so I never included any DCA1 specific information and had it "force" passthrough (?) on the data.

The delay is not from any of the in-between steps either. I get a byte array by stream, so I can .into() it into a Live::Raw(...) Input. Parsing it to make it playable with make_playable_async is not the pinch point, as testing with a bunch of pre-parsed packets leads to the same behaviour.

Is this known? And is there any way to remove this delay, something specific to streaming or similar?

// This function is called each time a packet is received.
pub async fn play_music(set: AudioSet) {
	use songbird::input::{
		Input,
		codecs::{CODEC_REGISTRY, PROBE}, 
	};

	// SONG: OnceCell<Arc<Songbird>>
	let sb = SONG.get().expect("Songbird not found!").clone();

	if let Some(h) = sb.get(set.guild_id) {
		let mut handler = h.lock().await;

		let audio: Input = set.audio_data.into();
		let audio: Input = audio.make_playable_async(&CODEC_REGISTRY, &PROBE).await
			.expect("Can't make audio playable!");

		let track_handle = handler.play_input(audio);

		while track_handle.get_info().await.unwrap().playing == songbird::tracks::PlayMode::Play {}
	}
}

[FelixMcFelix]

This shouldn't be down to DCA0 vs DCA1 -- parsing a small block of json is probably not expensive to pay once, as an audible gap here would exceed ~25--50ms, I'm guessing. You'd need an insanely large metadata block to blow past that in terms of cost.

I have to ask what you mean by packet: are you playing these as individual DCA files? I.e. n x {DCA header + 15 x 20ms frame}? If so then your tracks are almost certainly being handed off to an aux thread to be made ready -- this is cheap work, but will insert at least one tick between sending the track off and awaiting the response. Part of this is that songbird now treats all track initialisation lazily by default. You should be able to upgrade your Inputs before you play them to prevent this: https://docs.rs/songbird/latest/songbird/input/enum.Input.html#method.make_playable_async.

The ideal fix would be to write a custom Read implementation which emits the headers once. You would just dump frames into via an mpsc, and an empty channel read would emit a silent frame (I don't know if we export this definition, but it's in our constants.rs). Note that writing this correctly is a bit tricky! So I'd recommend you try the former first. :)

[FelixMcFelix]

Ah, sorry; I had missed that you were already making the tracks playable ahead-of-time. My bad for trying to write a reply while travelling! You do have the right idea there.

I suppose the next possible source for gaps would be the call's task moving between idle and active. I'm less certain how you could isolate that: with today's code, putting in a dummy track that is always paused should keep a mixer task active, so we could rule that out/in as the source. Keep in mind that so long as you are playing these segments as separate tracks, there's always going to be a risk that you mistime it and end up a gap in playback if there's no live track during a tick.

[Pranav-Badrinathan]

putting in a dummy track that is always paused should keep a mixer task active

If it works, this should be a good fix for now I guess.

Keep in mind that so long as you are playing these segments as separate tracks, there's always going to be a risk that you mistime it and end up a gap in playback

Yeah unfortunately I don't think this can be fixed going down this route 😅, just need to make sure the timing is as good as possible I suppose.

I feel like I'm using the library wrong lol, having to fight it and use it in unintended ways to get my work done (streaming packets in this case). Perhaps a reorder of my project is in order to make it utilize the tools as intended is necessary.

[Pranav-Badrinathan]

The ideal fix would be to write a custom Read implementation which emits the headers once. You would just dump frames into via an mpsc, and an empty channel read would emit a silent frame.

For this, could you point me to someplace in the source where the audio is "sent", something I could send the frames from the mspc to?
Closest I could find was here:

songbird/src/driver/scheduler/live.rs

Lines 252 to 279 in 087e5f2

	// Send all.
	self.start_of_work = Some(Instant::now());
	for (i, (packet_len, mixer)) in self
	.packet_lens
	.iter_mut()
	.zip(self.tasks.iter_mut())
	.enumerate()
	{
	let (block, inner) = get_memory_indices(i);
	let packet = &mut self.packets[block][inner..];
	if *packet_len > 0 {
	let res = mixer.send_packet(&packet[..*packet_len]);
	rebuild_if_err(mixer, res, &mut self.to_cull, i);
	}
	#[cfg(test)]
	if *packet_len == 0 {
	mixer.test_signal_empty_tick();
	}
	advance_rtp_counters(packet);
	}
	for (i, mixer) in self.tasks.iter_mut().enumerate() {
	let res = mixer
	.audio_commands_events()
	.and_then(|()| mixer.check_and_send_keepalive(self.start_of_work));
	rebuild_if_err(mixer, res, &mut self.to_cull, i);
	}

But this would not allow for I just can't find anywhere in the code that would allow for this... A function that simply just sends whatever is sent its way, if it exists. I think the one linked above does this, but it does it for an instance of the Live struct, and I don't know if I can create one with all the parameters it needs. Please correct me, I am definitely wrong.

EDIT: Maybe there is some way to gain access to a particular Live instance, on which I can call add_packet_block and send the data that way?

songbird/src/driver/scheduler/live.rs

Lines 531 to 543 in 087e5f2

	fn add_packet_block(&mut self) {
	let n_packets = if let Some(limit) = self.config.strategy.task_limit() {
	let (block, inner) = get_memory_indices_unscaled(limit);
	if self.packets.len() < block || inner == 0 {
	PACKETS_PER_BLOCK
	} else {
	inner
	}
	} else {
	PACKETS_PER_BLOCK
	};
	self.packets.push(packet_block(n_packets));
	}

[FelixMcFelix]

You're perhaps looking too far ahead -- you should be looking for where the packet data is built/encrypted/... in the mixer task, and then placed into self.packets. The important part is in the mixer task:

songbird/src/driver/tasks/mixer/mod.rs

Lines 743 to 868 in 087e5f2

	fn mix_tracks(&mut self, packet: &mut [u8]) -> MixType {
	// Get a slice of bytes to write in data for Opus packet passthrough.
	let mut rtp = MutableRtpPacket::new(packet).expect(
	"FATAL: Too few bytes in self.packet for RTP header.\
	(Blame: VOICE_PACKET_MAX?)",
	);
	let payload = rtp.payload_mut();
	let opus_frame = &mut payload[TAG_SIZE..];
	// Opus frame passthrough.
	// This requires that we have only one PLAYING track, who has volume 1.0, and an
	// Opus codec type (verified later in mix_symph_indiv).
	//
	// We *could* cache the number of live tracks separately, but that makes this
	// quite fragile given all the ways a user can alter the PlayMode.
	let mut num_live = 0;
	let mut last_live_vol = 1.0;
	for track in &self.tracks {
	if track.playing.is_playing() {
	num_live += 1;
	last_live_vol = track.volume;
	}
	}
	let do_passthrough = num_live == 1 && (last_live_vol - 1.0).abs() < f32::EPSILON;
	let mut len = 0;
	for (i, track) in self.tracks.iter_mut().enumerate() {
	let vol = track.volume;
	// This specifically tries to get tracks who are "preparing",
	// so that event handlers and the like can all be fired without
	// the track being in a `Play` state.
	if !track.should_check_input() {
	continue;
	}
	let should_play = track.playing.is_playing();
	let input = track.get_or_ready_input(
	i,
	&self.interconnect,
	&self.thread_pool,
	&self.config,
	self.prevent_events,
	);
	let (input, mix_state) = match input {
	Ok(i) => i,
	Err(InputReadyingError::Waiting) => continue,
	Err(InputReadyingError::NeedsSeek(req)) => {
	track.seek(
	i,
	req,
	&self.interconnect,
	&self.thread_pool,
	&self.config,
	self.prevent_events,
	);
	continue;
	},
	// TODO: allow for retry in given time.
	Err(e) => {
	if let Some(fail) = e.as_user() {
	track.playing = PlayMode::Errored(fail);
	}
	continue;
	},
	};
	// Now that we have dealt with potential errors in preparing tracks,
	// only do any mixing if the track is to be played!
	if !should_play {
	continue;
	}
	let (mix_type, status) = mix_logic::mix_symph_indiv(
	&mut self.symph_mix,
	&mut self.resample_scratch,
	input,
	mix_state,
	vol,
	do_passthrough.then_some(&mut *opus_frame),
	);
	let return_here = if let MixType::MixedPcm(pcm_len) = mix_type {
	len = len.max(pcm_len);
	false
	} else {
	if mix_state.passthrough == Passthrough::Inactive {
	input.decoder.reset();
	}
	mix_state.passthrough = Passthrough::Active;
	true
	};
	// FIXME: allow Ended to trigger a seek/loop/revisit in the same mix cycle?
	// Would this be possible with special-casing to mark some inputs as fast
	// to recreate? Probably not doable in the general case.
	match status {
	MixStatus::Live => track.step_frame(),
	MixStatus::Errored(e) =>
	track.playing = PlayMode::Errored(PlayError::Decode(e.into())),
	MixStatus::Ended if track.do_loop() => {
	drop(self.track_handles[i].seek(Duration::default()));
	if !self.prevent_events {
	// position update is sent out later, when the seek concludes.
	drop(self.interconnect.events.send(EventMessage::ChangeState(
	i,
	TrackStateChange::Loops(track.loops, false),
	)));
	}
	},
	MixStatus::Ended => {
	track.end();
	},
	}
	// This needs to happen here due to borrow checker shenanigans.
	if return_here {
	return mix_type;
	}
	}
	MixType::MixedPcm(len)
	}
	}

and then in mix_symph_indiv:

songbird/src/driver/tasks/mixer/mix_logic.rs

Lines 69 to 92 in 087e5f2

	// Opus packet passthrough special case.
	if codec_type == CODEC_TYPE_OPUS && local_state.passthrough != Passthrough::Block {
	if let Some(slot) = opus_slot.as_mut() {
	let sample_ct = buf
	.try_into()
	.and_then(|buf| audiopus::packet::nb_samples(buf, SAMPLE_RATE));
	// We don't actually block passthrough until a few violations are
	// seen. The main one is that most Opus tracks end on a sub-20ms
	// frame, particularly on Youtube.
	// However, a frame that's bigger than the target buffer is an instant block.
	let buf_size_fatal = buf.len() <= slot.len();
	if match sample_ct {
	Ok(MONO_FRAME_SIZE) => true,
	_ => !local_state.record_and_check_passthrough_strike_final(buf_size_fatal),
	} {
	slot.write_all(buf)
	.expect("Bounds check performed, and failure will block passthrough.");
	return (MixType::Passthrough(buf.len()), MixStatus::Live);
	}
	}
	}

You can see that a Live mixer/call is allocated a packet buffer, which is then passed up into the mixer logic. If the conditions are right (Opus codec, one track, unchanged volume), the mixer will instead copy the inner frame data directly into that packet buffer, which is eventually sent. However, we need to encrypt that packet even if it's 'passthrough', which is why you can't just write straight into the packet buffer as a user and expect it to work. There is an overview included in the codefence in the scheduler section of the architecture document There isn't a mechanism to 'just pass frames in' today, you need to build an impl Read that symphonia can parse and use here. As a sketch for DCA1, which is far simpler to hack together than an Ogg stream:

struct OpusStream {
  // actual 20ms opus frames in these Vecs, not the 300ms files you use today.
  rx: mpsc::Receiver<Vec<u8>>,
  current_frame: Option<Vec<u8>>,
  chunk_pos: usize,
  pos: usize,
}

// prebuild this, fill in len, etc. use a lazy static?
const BASIC_DCA1: &[u8] = b"DCA1\0\0\0\0{\"dca\":{...}}";

impl std::io::Read for OpusStream {
  fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
    todo!();
    // if pos < BASIC_DCA1.len()
    //   fill buf from there and update pos + chunk_pos.
    // else if no current frame
    //   fetch new frame from rx
    //   if channel closed, return Ok(0)
    //   else if channel empty, current_frame = Some(SILENT_FRAME)
    //   else if read ok, current_frame = Some(rxd_frame)
    //   chunk_pos = 0
    //
    // (current_frame now guaranteed Some(frame))
    // if chunk_pos < 2
    //   fill buf from (current_frame.len() as u16).to_le_bytes()
    //   update pos, chunk_pos
    // else
    //   fill buf from current_frame
    //   update pos, chunk_pos
    //   unset current_frame if done (chunk_pos == current_frame.len() + 2)
  }
}

wrapped in a ReadOnlySource. Just make sure to only return Ok(0) on channel close or the file will be considered finished, and to take care around partial writes -- there are no guarantees on the length of buf on an implementor of Read. This is rough, but I hope it gets across the intuition -- you write out the header only once, and then you output the bytes of each frame and its length as you would if you were building a straight Vec<u8>.

EDIT: Maybe there is some way to gain access to a particular Live instance, on which I can call add_packet_block and send the data that way?

I don't think this is really the abstraction we're going for, and given that Live/Idle state is handled by the library then it's fraught to do so. add_packet_block probably does not do what you think. There is undoubtedly a better way to build this into songbird, but that adds more maintenance surface area for something that's in principle doable with fairly low overhead.

[Pranav-Badrinathan]

Wow thanks for the detailed explanation.

You're perhaps looking too far ahead

Yeah, all the code was pub(crate) at best where I was looking.

you should be looking for where the packet data is built/encrypted/... in the mixer task ... However, we need to encrypt that packet even if it's 'passthrough', which is why you can't just write straight into the packet buffer as a user and expect it to work.

Ah okay I see. Data encryption is something I failed to consider, it makes much more sense now.

...you need to build an impl Read that symphonia can parse and use here.

I'll look into this then. Thanks for the sketch, I was quite lost on what to do when you mentioned this in the first reply, and having an idea of what is going on helps a lot.

[Pranav-Badrinathan]

I didn't quite get how to integrate your suggestion into songbird to actually use the OpusStream struct and its Read, but after careful thought over these last few days, I think I may understand what to do (probably not, my confidence is not high)?

In your example above, OpusStream would need to implement symphonia's MediaSource (via ReadOnlySource), which requires Read and Seek. Then I can do something like Input::Live(LiveInput::Raw(AudioStream {input: opus_stream, hint})), call promote and and play_input on it, and the stream should be live. Then, it is just a matter of sending in frames via the mspc like you mention above.