X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=mux.cpp;h=cfe38b63a67041eb2a6f0f3b7cb0b1cec945f490;hb=d3e48df512d9476d3849227067792a3537bb094e;hp=d01c18038a893cfec7152d35d8e3b0f1131ed0fb;hpb=bc0683200d63d2fd0d49c0b998885c8b4d9db77e;p=nageru diff --git a/mux.cpp b/mux.cpp index d01c180..cfe38b6 100644 --- a/mux.cpp +++ b/mux.cpp @@ -31,7 +31,9 @@ using namespace std; struct PacketBefore { PacketBefore(const AVFormatContext *ctx) : ctx(ctx) {} - bool operator() (const AVPacket *a, const AVPacket *b) const { + bool operator() (const Mux::QueuedPacket &a_qp, const Mux::QueuedPacket &b_qp) const { + const AVPacket *a = a_qp.pkt; + const AVPacket *b = b_qp.pkt; int64_t a_dts = (a->dts == AV_NOPTS_VALUE ? a->pts : a->dts); int64_t b_dts = (b->dts == AV_NOPTS_VALUE ? b->pts : b->dts); AVRational a_timebase = ctx->streams[a->stream_index]->time_base; @@ -46,8 +48,8 @@ struct PacketBefore { const AVFormatContext * const ctx; }; -Mux::Mux(AVFormatContext *avctx, int width, int height, Codec video_codec, const string &video_extradata, const AVCodecParameters *audio_codecpar, int time_base, std::function write_callback, const vector &metrics) - : avctx(avctx), write_callback(write_callback), metrics(metrics) +Mux::Mux(AVFormatContext *avctx, int width, int height, Codec video_codec, const string &video_extradata, const AVCodecParameters *audio_codecpar, int time_base, std::function write_callback, WriteStrategy write_strategy, const vector &metrics) + : write_strategy(write_strategy), avctx(avctx), write_callback(write_callback), metrics(metrics) { avstream_video = avformat_new_stream(avctx, nullptr); if (avstream_video == nullptr) { @@ -73,7 +75,7 @@ Mux::Mux(AVFormatContext *avctx, int width, int height, Codec video_codec, const // Note that there's no way to change this per-frame as the H.264 stream // would like to be able to. avstream_video->codecpar->color_primaries = AVCOL_PRI_BT709; // RGB colorspace (inout_format.color_space). - avstream_video->codecpar->color_trc = AVCOL_TRC_UNSPECIFIED; // Gamma curve (inout_format.gamma_curve). + avstream_video->codecpar->color_trc = AVCOL_TRC_IEC61966_2_1; // Gamma curve (inout_format.gamma_curve). // YUV colorspace (output_ycbcr_format.luma_coefficients). if (global_flags.ycbcr_rec709_coefficients) { avstream_video->codecpar->color_space = AVCOL_SPC_BT709; @@ -116,10 +118,20 @@ Mux::Mux(AVFormatContext *avctx, int width, int height, Codec video_codec, const // Make sure the header is written before the constructor exits. avio_flush(avctx->pb); + + if (write_strategy == WRITE_BACKGROUND) { + writer_thread = thread(&Mux::thread_func, this); + } } Mux::~Mux() { + assert(plug_count == 0); + if (write_strategy == WRITE_BACKGROUND) { + writer_thread_should_quit = true; + packet_queue_ready.notify_all(); + writer_thread.join(); + } int64_t old_pos = avctx->pb->pos; av_write_trailer(avctx); for (MuxMetrics *metric : metrics) { @@ -133,7 +145,7 @@ Mux::~Mux() avformat_free_context(avctx); } -void Mux::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts) +void Mux::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts, AVRational timebase) { AVPacket pkt_copy; if (av_copy_packet(&pkt_copy, &pkt) < 0) { @@ -141,38 +153,33 @@ void Mux::add_packet(const AVPacket &pkt, int64_t pts, int64_t dts) exit(1); } if (pkt.stream_index == 0) { - pkt_copy.pts = av_rescale_q(pts, AVRational{1, TIMEBASE}, avstream_video->time_base); - pkt_copy.dts = av_rescale_q(dts, AVRational{1, TIMEBASE}, avstream_video->time_base); - pkt_copy.duration = av_rescale_q(pkt.duration, AVRational{1, TIMEBASE}, avstream_video->time_base); + pkt_copy.pts = av_rescale_q(pts, timebase, avstream_video->time_base); + pkt_copy.dts = av_rescale_q(dts, timebase, avstream_video->time_base); + pkt_copy.duration = av_rescale_q(pkt.duration, timebase, avstream_video->time_base); } else if (pkt.stream_index == 1) { - pkt_copy.pts = av_rescale_q(pts, AVRational{1, TIMEBASE}, avstream_audio->time_base); - pkt_copy.dts = av_rescale_q(dts, AVRational{1, TIMEBASE}, avstream_audio->time_base); - pkt_copy.duration = av_rescale_q(pkt.duration, AVRational{1, TIMEBASE}, avstream_audio->time_base); + pkt_copy.pts = av_rescale_q(pts, timebase, avstream_audio->time_base); + pkt_copy.dts = av_rescale_q(dts, timebase, avstream_audio->time_base); + pkt_copy.duration = av_rescale_q(pkt.duration, timebase, avstream_audio->time_base); } else { assert(false); } { lock_guard lock(mu); - if (plug_count > 0) { - plugged_packets.push_back(av_packet_clone(&pkt_copy)); + if (write_strategy == WriteStrategy::WRITE_BACKGROUND) { + packet_queue.push_back(QueuedPacket{ av_packet_clone(&pkt_copy), pts }); + if (plug_count == 0) packet_queue_ready.notify_all(); + } else if (plug_count > 0) { + packet_queue.push_back(QueuedPacket{ av_packet_clone(&pkt_copy), pts }); } else { - write_packet_or_die(pkt_copy); + write_packet_or_die(pkt_copy, pts); } } av_packet_unref(&pkt_copy); - - // Note: This will be wrong in the case of plugged packets, but that only happens - // for network streams, not for files, and write callbacks are only really relevant - // for files. (We don't want to do this from write_packet_or_die, as it only has - // the rescaled pts, which is unsuitable for callback.) - if (pkt.stream_index == 0 && write_callback != nullptr) { - write_callback(pts); - } } -void Mux::write_packet_or_die(const AVPacket &pkt) +void Mux::write_packet_or_die(const AVPacket &pkt, int64_t unscaled_pts) { for (MuxMetrics *metric : metrics) { if (pkt.stream_index == 0) { @@ -192,6 +199,10 @@ void Mux::write_packet_or_die(const AVPacket &pkt) for (MuxMetrics *metric : metrics) { metric->metric_written_bytes += avctx->pb->pos - old_pos; } + + if (pkt.stream_index == 0 && write_callback != nullptr) { + write_callback(unscaled_pts); + } } void Mux::plug() @@ -208,13 +219,42 @@ void Mux::unplug() } assert(plug_count >= 0); - sort(plugged_packets.begin(), plugged_packets.end(), PacketBefore(avctx)); + sort(packet_queue.begin(), packet_queue.end(), PacketBefore(avctx)); - for (AVPacket *pkt : plugged_packets) { - write_packet_or_die(*pkt); - av_packet_free(&pkt); + if (write_strategy == WRITE_BACKGROUND) { + packet_queue_ready.notify_all(); + } else { + for (QueuedPacket &qp : packet_queue) { + write_packet_or_die(*qp.pkt, qp.unscaled_pts); + av_packet_free(&qp.pkt); + } + packet_queue.clear(); + } +} + +void Mux::thread_func() +{ + unique_lock lock(mu); + for ( ;; ) { + packet_queue_ready.wait(lock, [this]() { + return writer_thread_should_quit || (!packet_queue.empty() && plug_count == 0); + }); + if (writer_thread_should_quit && packet_queue.empty()) { + // All done. + break; + } + + assert(!packet_queue.empty() && plug_count == 0); + vector packets; + swap(packets, packet_queue); + + lock.unlock(); + for (QueuedPacket &qp : packets) { + write_packet_or_die(*qp.pkt, qp.unscaled_pts); + av_packet_free(&qp.pkt); + } + lock.lock(); } - plugged_packets.clear(); } void MuxMetrics::init(const vector> &labels)