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.

diff --git a/mux.cpp b/mux.cpp
index 326c73f69cf2e4be34004bb209359a0751b838c4..aec124b34aea7be5f5c5de7e9ec36a2eb86b0980 100644 (file)
--- 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 47855f728ae82b5d09d9a180c0035fc59d770027..45eab34a3d12390ed69e1367b798f5383cced118 100644 (file)
--- 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;
 };