X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=mux.cpp;h=cfe38b63a67041eb2a6f0f3b7cb0b1cec945f490;hb=d3e48df512d9476d3849227067792a3537bb094e;hp=1fd8e30ba3ff836a6b92521f96cda2b91a20f611;hpb=ee7da87b4aa284b7babd59dc21db925f7c384ce7;p=nageru diff --git a/mux.cpp b/mux.cpp index 1fd8e30..cfe38b6 100644 --- a/mux.cpp +++ b/mux.cpp @@ -23,6 +23,7 @@ extern "C" { #include "defs.h" #include "flags.h" +#include "metrics.h" #include "timebase.h" using namespace std; @@ -30,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; @@ -45,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) - : avctx(avctx), write_callback(write_callback) +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) { @@ -72,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; @@ -109,14 +112,32 @@ Mux::Mux(AVFormatContext *avctx, int width, int height, Codec video_codec, const fprintf(stderr, "avformat_write_header() failed\n"); exit(1); } + for (MuxMetrics *metric : metrics) { + metric->metric_written_bytes += avctx->pb->pos; + } // 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) { + metric->metric_written_bytes += avctx->pb->pos - old_pos; + } + if (!(avctx->oformat->flags & AVFMT_NOFILE) && !(avctx->flags & AVFMT_FLAG_CUSTOM_IO)) { avio_closep(&avctx->pb); @@ -124,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) { @@ -132,44 +153,56 @@ 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) { + metric->metric_video_bytes += pkt.size; + } else if (pkt.stream_index == 1) { + metric->metric_audio_bytes += pkt.size; + } else { + assert(false); + } + } + int64_t old_pos = avctx->pb->pos; if (av_interleaved_write_frame(avctx, const_cast(&pkt)) < 0) { fprintf(stderr, "av_interleaved_write_frame() failed\n"); exit(1); } avio_flush(avctx->pb); + 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() @@ -186,11 +219,53 @@ 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(); } - plugged_packets.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(); + } +} + +void MuxMetrics::init(const vector> &labels) +{ + vector> labels_video = labels; + labels_video.emplace_back("stream", "video"); + global_metrics.add("mux_stream_bytes", labels_video, &metric_video_bytes); + + vector> labels_audio = labels; + labels_audio.emplace_back("stream", "audio"); + global_metrics.add("mux_stream_bytes", labels_audio, &metric_audio_bytes); + + global_metrics.add("mux_written_bytes", labels, &metric_written_bytes); }