From db03cd080a4b38d76b85d3acf88a1bac0e84e734 Mon Sep 17 00:00:00 2001
From: "Steinar H. Gunderson" <sgunderson@bigfoot.com>
Date: Wed, 4 May 2016 20:19:37 +0200
Subject: [PATCH 1/1] Add back manual interleaving, but this time in Mux.

We need to interleave ourselves because we flush; otherwise, audio will
typically start 100 ms or so into the stream, and everything becomes bad.
Thankfully, we have only two streams to interleave, so we can choose
something quite simple, and this is still not bothered by the deadlocks
that arose from doing it in QuickSyncEncoder. (Hopefully we can soon find
keyframe boundaries without flushing, but it requires changes in avformat.)

Also fixes an issue where plugged packets would cause the boundaries
to fall in the wrong places.
---
 mux.cpp | 56 ++++++++++++++++++++++++++++++++++++++++++--------------
 mux.h   | 15 +++++++++++++++
 2 files changed, 57 insertions(+), 14 deletions(-)

diff --git a/mux.cpp b/mux.cpp
index 326c73f..aec124b 100644
--- a/mux.cpp
+++ b/mux.cpp
@@ -118,26 +118,57 @@ void Mux::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts)
 		assert(false);
 	}
 
-	if (keyframe_signal_receiver) {
-		if (pkt.flags & AV_PKT_FLAG_KEY) {
-			av_write_frame(avctx, nullptr);
-			keyframe_signal_receiver->signal_keyframe();
-		}
-	}
-
 	{
 		lock_guard<mutex> lock(mu);
 		if (plug_count > 0) {
 			plugged_packets.push_back(av_packet_clone(&pkt_copy));
-		} else if (av_interleaved_write_frame(avctx, &pkt_copy) < 0) {
-			fprintf(stderr, "av_interleaved_write_frame() failed\n");
-			exit(1);
+		} else {
+			add_interleaved_packet(pkt_copy);
 		}
 	}
 
 	av_packet_unref(&pkt_copy);
 }
 
+void Mux::add_interleaved_packet(const AVPacket &pkt)
+{
+	if (waiting_packets.empty() || waiting_packets.front()->stream_index == pkt.stream_index) {
+		// We could still get packets of the other type with earlier pts/dts,
+		// so we'll have to queue and wait.
+		waiting_packets.push(av_packet_clone(const_cast<AVPacket *>(&pkt)));
+		return;
+	}
+
+	// Flush all the queued packets that are supposed to go before this.
+	PacketBefore before(avctx);
+	while (!waiting_packets.empty() && !before(&pkt, waiting_packets.front())) {
+		AVPacket *queued_pkt = waiting_packets.front();
+		waiting_packets.pop();
+		write_packet_with_signal(*queued_pkt);
+		av_packet_unref(queued_pkt);
+	}
+
+	if (waiting_packets.empty()) {
+		waiting_packets.push(av_packet_clone(const_cast<AVPacket *>(&pkt)));
+	} else {
+		write_packet_with_signal(pkt);
+	}
+}
+
+void Mux::write_packet_with_signal(const AVPacket &pkt)
+{
+	if (keyframe_signal_receiver) {
+		if (pkt.flags & AV_PKT_FLAG_KEY) {
+			av_write_frame(avctx, nullptr);
+			keyframe_signal_receiver->signal_keyframe();
+		}
+	}
+	if (av_write_frame(avctx, const_cast<AVPacket *>(&pkt)) < 0) {
+		fprintf(stderr, "av_interleaved_write_frame() failed\n");
+		exit(1);
+	}
+}
+
 void Mux::plug()
 {
 	lock_guard<mutex> lock(mu);
@@ -155,10 +186,7 @@ void Mux::unplug()
 	sort(plugged_packets.begin(), plugged_packets.end(), PacketBefore(avctx));
 
 	for (AVPacket *pkt : plugged_packets) {
-		if (av_interleaved_write_frame(avctx, pkt) < 0) {
-			fprintf(stderr, "av_interleaved_write_frame() failed\n");
-			exit(1);
-		}
+		add_interleaved_packet(*pkt);
 		av_packet_free(&pkt);
 	}
 	plugged_packets.clear();
diff --git a/mux.h b/mux.h
index 47855f7..45eab34 100644
--- a/mux.h
+++ b/mux.h
@@ -10,6 +10,7 @@ extern "C" {
 }
 
 #include <mutex>
+#include <queue>
 #include <vector>
 
 class KeyFrameSignalReceiver {
@@ -42,11 +43,25 @@ public:
 	void unplug();
 
 private:
+	void add_interleaved_packet(const AVPacket &pkt);  // Must be called with <mu> held.
+	void write_packet_with_signal(const AVPacket &pkt);  // Must be called with <mu> held.
+
 	std::mutex mu;
 	AVFormatContext *avctx;  // Protected by <mu>.
 	int plug_count = 0;  // Protected by <mu>.
 	std::vector<AVPacket *> plugged_packets;  // Protected by <mu>.
 
+	// We need to do our own interleaving since we do explicit flushes
+	// before each keyframe. This queue contains every packet that we
+	// couldn't send yet, in add order. Essentially, we can't send a packet
+	// before we know we cannot receive an earlier (dts-wise) packet
+	// from another stream. This means that this queue will either contain
+	// video packets only or audio packets only, and as soon as a packet
+	// of the other type comes in, we can empty the flush the queue up
+	// to that point.
+	// Protected by <mu>.
+	std::queue<AVPacket *> waiting_packets;
+
 	AVStream *avstream_video, *avstream_audio;
 	KeyFrameSignalReceiver *keyframe_signal_receiver;
 };
-- 
2.39.2