From patchwork Thu Nov 23 19:15:06 2023 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Anton Khirnov X-Patchwork-Id: 44775 Delivered-To: ffmpegpatchwork2@gmail.com Received: by 2002:a05:6a20:bca6:b0:181:818d:5e7f with SMTP id fx38csp802680pzb; Thu, 23 Nov 2023 11:20:34 -0800 (PST) X-Google-Smtp-Source: AGHT+IHTqd8cstBIcY+RpFGEJgeHLxiC0Ygb0XdecYelsrt2SWatNP/HDV0CBus0Axd0Kn1VL7Cv X-Received: by 2002:a05:6402:254d:b0:548:5671:b6c5 with SMTP id l13-20020a056402254d00b005485671b6c5mr189341edb.0.1700767234591; Thu, 23 Nov 2023 11:20:34 -0800 (PST) ARC-Seal: i=1; a=rsa-sha256; t=1700767234; cv=none; d=google.com; s=arc-20160816; b=E+OzjCBpMh2xeqdb3/X64P6aD+wm7mofw825wj06QghHljdKoKuIKjP1qf8yEi+/Yr 6DfUeGIArkzWdO+Nfpnnb/puPWxyVKiwYjQqXSkSWfnK2z1bmsUTfnkUitBNvb4KGDc5 wAdK07NqoOYpCbpyeTiXfJ0VXl7MadHMW/NRXmuAKm2jFaZm2cNCe+a4a3oe6WYxmsxY Gh0119ArSi6Bo28qLZ+V4I0/QcFaSH95l6sT0kGNMWhzrzOV6uhqtEZ5X8m3WfZxvfIh lKAYOWq6axQjcnpCsnh12RtLSOXwuII31vI2WlR8B8CsG+gZaQRnEJ+FpYcpooHb3ecV dQ9g== ARC-Message-Signature: i=1; a=rsa-sha256; c=relaxed/relaxed; d=google.com; s=arc-20160816; h=sender:errors-to:content-transfer-encoding:reply-to:list-subscribe :list-help:list-post:list-archive:list-unsubscribe:list-id :precedence:subject:mime-version:references:in-reply-to:message-id :date:to:from:delivered-to; bh=auJA7s8M4VBmKanPCst4P1HW2/G+X3lwQ8/XnRa64KM=; fh=YOA8vD9MJZuwZ71F/05pj6KdCjf6jQRmzLS+CATXUQk=; b=DjW6nG4m7N1MK7WfZf+cpO0SCEPuUGsOCLnUKQaqDsuc/VybGIzkSOy17heuXzu7In RSC8bgbBpmH3FsOTsVVK9tuIa0LnL3MhixqgwJbSmPBZrRvHPlXpHwTnNYLIDmIHhyKc eG1o4HgWQypXIBT/X6UEYbCiiJD2a4oht1tzOPctWDXNjzWWByiMjWv74Ik0qQ2bWkG4 HuHUFYK6VMX3c2VAAB6JvmRTtxflxj9RuUT17yqZG+NlptEmWfbPhb3caRbZao9LkPjh 1mZbskHF65w0tUiXqCBk56U0S6jNCYvtAkDXoH7ZgsGnXc8FfBX6ey+hKcSJPMFVJ9OB Mrvg== ARC-Authentication-Results: i=1; mx.google.com; spf=pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) smtp.mailfrom=ffmpeg-devel-bounces@ffmpeg.org Return-Path: Received: from ffbox0-bg.mplayerhq.hu (ffbox0-bg.ffmpeg.org. [79.124.17.100]) by mx.google.com with ESMTP id v16-20020a50a450000000b0053e00942bd9si873393edb.336.2023.11.23.11.20.34; Thu, 23 Nov 2023 11:20:34 -0800 (PST) Received-SPF: pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) client-ip=79.124.17.100; Authentication-Results: mx.google.com; spf=pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) smtp.mailfrom=ffmpeg-devel-bounces@ffmpeg.org Received: from [127.0.1.1] (localhost [127.0.0.1]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTP id C05D668CF21; Thu, 23 Nov 2023 21:19:05 +0200 (EET) X-Original-To: ffmpeg-devel@ffmpeg.org Delivered-To: ffmpeg-devel@ffmpeg.org Received: from mail1.khirnov.net (quelana.khirnov.net [94.230.150.81]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTPS id A85BC68CF02 for ; Thu, 23 Nov 2023 21:18:54 +0200 (EET) Received: from localhost (mail1.khirnov.net [IPv6:::1]) by mail1.khirnov.net (Postfix) with ESMTP id 8560114E7 for ; Thu, 23 Nov 2023 20:18:19 +0100 (CET) Received: from mail1.khirnov.net ([IPv6:::1]) by localhost (mail1.khirnov.net [IPv6:::1]) (amavis, port 10024) with ESMTP id wlWkb-7nwS4O for ; Thu, 23 Nov 2023 20:18:18 +0100 (CET) Received: from libav.khirnov.net (libav.khirnov.net [IPv6:2a00:c500:561:201::7]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange X25519 server-signature RSA-PSS (2048 bits) server-digest SHA256 client-signature RSA-PSS (2048 bits) client-digest SHA256) (Client CN "libav.khirnov.net", Issuer "smtp.khirnov.net SMTP CA" (verified OK)) by mail1.khirnov.net (Postfix) with ESMTPS id 859FD1CAF for ; Thu, 23 Nov 2023 20:18:15 +0100 (CET) Received: from libav.khirnov.net (libav.khirnov.net [IPv6:::1]) by libav.khirnov.net (Postfix) with ESMTP id 799793A05C4 for ; Thu, 23 Nov 2023 20:18:15 +0100 (CET) From: Anton Khirnov To: ffmpeg-devel@ffmpeg.org Date: Thu, 23 Nov 2023 20:15:06 +0100 Message-ID: <20231123191524.11296-13-anton@khirnov.net> X-Mailer: git-send-email 2.42.0 In-Reply-To: <20231123191524.11296-2-anton@khirnov.net> References: <20231123191524.11296-2-anton@khirnov.net> MIME-Version: 1.0 Subject: [FFmpeg-devel] [PATCH 11/13] fftools/ffmpeg_enc: move encoding to a separate thread X-BeenThere: ffmpeg-devel@ffmpeg.org X-Mailman-Version: 2.1.29 Precedence: list List-Id: FFmpeg development discussions and patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Reply-To: FFmpeg development discussions and patches Errors-To: ffmpeg-devel-bounces@ffmpeg.org Sender: "ffmpeg-devel" X-TUID: d6LeVjs0d4gw As for the analogous decoding change, this is only a preparatory step to a fully threaded architecture and does not yet make encoding truly parallel. The main thread will currently submit a frame and wait until it has been fully processed by the encoder before moving on. That will change in future commits after filters are moved to threads and a thread-aware scheduler is added. This code suffers from a known issue - if an encoder with a sync queue receives EOF it will terminate after processing everything it currently has, even though the sync queue might still be triggered by other threads. That will be fixed in following commits. --- fftools/ffmpeg_enc.c | 360 ++++++++++++++++++++++++++++++++++++++----- 1 file changed, 320 insertions(+), 40 deletions(-) diff --git a/fftools/ffmpeg_enc.c b/fftools/ffmpeg_enc.c index fa4539664f..46c21fc0e4 100644 --- a/fftools/ffmpeg_enc.c +++ b/fftools/ffmpeg_enc.c @@ -20,6 +20,8 @@ #include #include "ffmpeg.h" +#include "ffmpeg_utils.h" +#include "thread_queue.h" #include "libavutil/avassert.h" #include "libavutil/avstring.h" @@ -43,6 +45,7 @@ struct Encoder { // packet for receiving encoded output AVPacket *pkt; + AVFrame *sub_frame; // combined size of all the packets received from the encoder uint64_t data_size; @@ -51,8 +54,48 @@ struct Encoder { uint64_t packets_encoded; int opened; + int finished; + + pthread_t thread; + /** + * Queue for sending frames from the main thread to + * the encoder thread. + */ + ThreadQueue *queue_in; + /** + * Queue for sending encoded packets from the encoder thread + * to the main thread. + * + * An empty packet is sent to signal that a previously sent + * frame has been fully processed. + */ + ThreadQueue *queue_out; }; +// data that is local to the decoder thread and not visible outside of it +typedef struct EncoderThread { + AVFrame *frame; + AVPacket *pkt; +} EncoderThread; + +static int enc_thread_stop(Encoder *e) +{ + void *ret; + + if (!e->queue_in) + return 0; + + tq_send_finish(e->queue_in, 0); + tq_receive_finish(e->queue_out, 0); + + pthread_join(e->thread, &ret); + + tq_free(&e->queue_in); + tq_free(&e->queue_out); + + return (int)(intptr_t)ret; +} + void enc_free(Encoder **penc) { Encoder *enc = *penc; @@ -60,7 +103,10 @@ void enc_free(Encoder **penc) if (!enc) return; + enc_thread_stop(enc); + av_frame_free(&enc->sq_frame); + av_frame_free(&enc->sub_frame); av_packet_free(&enc->pkt); @@ -77,6 +123,12 @@ int enc_alloc(Encoder **penc, const AVCodec *codec) if (!enc) return AVERROR(ENOMEM); + if (codec->type == AVMEDIA_TYPE_SUBTITLE) { + enc->sub_frame = av_frame_alloc(); + if (!enc->sub_frame) + goto fail; + } + enc->pkt = av_packet_alloc(); if (!enc->pkt) goto fail; @@ -165,6 +217,52 @@ static int set_encoder_id(OutputFile *of, OutputStream *ost) return 0; } +static void *encoder_thread(void *arg); + +static int enc_thread_start(OutputStream *ost) +{ + Encoder *e = ost->enc; + ObjPool *op; + int ret = 0; + + op = objpool_alloc_frames(); + if (!op) + return AVERROR(ENOMEM); + + e->queue_in = tq_alloc(1, 1, op, frame_move); + if (!e->queue_in) { + objpool_free(&op); + return AVERROR(ENOMEM); + } + + op = objpool_alloc_packets(); + if (!op) + goto fail; + + e->queue_out = tq_alloc(1, 4, op, pkt_move); + if (!e->queue_out) { + objpool_free(&op); + goto fail; + } + + ret = pthread_create(&e->thread, NULL, encoder_thread, ost); + if (ret) { + ret = AVERROR(ret); + av_log(ost, AV_LOG_ERROR, "pthread_create() failed: %s\n", + av_err2str(ret)); + goto fail; + } + + return 0; +fail: + if (ret >= 0) + ret = AVERROR(ENOMEM); + + tq_free(&e->queue_in); + tq_free(&e->queue_out); + return ret; +} + int enc_open(OutputStream *ost, const AVFrame *frame) { InputStream *ist = ost->ist; @@ -373,6 +471,13 @@ int enc_open(OutputStream *ost, const AVFrame *frame) if (ost->st->time_base.num <= 0 || ost->st->time_base.den <= 0) ost->st->time_base = av_add_q(ost->enc_ctx->time_base, (AVRational){0, 1}); + ret = enc_thread_start(ost); + if (ret < 0) { + av_log(ost, AV_LOG_ERROR, "Error starting encoder thread: %s\n", + av_err2str(ret)); + return ret; + } + ret = of_stream_init(of, ost); if (ret < 0) return ret; @@ -386,19 +491,18 @@ static int check_recording_time(OutputStream *ost, int64_t ts, AVRational tb) if (of->recording_time != INT64_MAX && av_compare_ts(ts, tb, of->recording_time, AV_TIME_BASE_Q) >= 0) { - close_output_stream(ost); return 0; } return 1; } -int enc_subtitle(OutputFile *of, OutputStream *ost, const AVSubtitle *sub) +static int do_subtitle_out(OutputFile *of, OutputStream *ost, const AVSubtitle *sub, + AVPacket *pkt) { Encoder *e = ost->enc; int subtitle_out_max_size = 1024 * 1024; int subtitle_out_size, nb, i, ret; AVCodecContext *enc; - AVPacket *pkt = e->pkt; int64_t pts; if (sub->pts == AV_NOPTS_VALUE) { @@ -429,7 +533,7 @@ int enc_subtitle(OutputFile *of, OutputStream *ost, const AVSubtitle *sub) AVSubtitle local_sub = *sub; if (!check_recording_time(ost, pts, AV_TIME_BASE_Q)) - return 0; + return AVERROR_EOF; ret = av_new_packet(pkt, subtitle_out_max_size); if (ret < 0) @@ -470,9 +574,11 @@ int enc_subtitle(OutputFile *of, OutputStream *ost, const AVSubtitle *sub) } pkt->dts = pkt->pts; - ret = of_output_packet(of, ost, pkt); - if (ret < 0) + ret = tq_send(e->queue_out, 0, pkt); + if (ret < 0) { + av_packet_unref(pkt); return ret; + } } return 0; @@ -610,11 +716,11 @@ static int update_video_stats(OutputStream *ost, const AVPacket *pkt, int write_ return 0; } -static int encode_frame(OutputFile *of, OutputStream *ost, AVFrame *frame) +static int encode_frame(OutputFile *of, OutputStream *ost, AVFrame *frame, + AVPacket *pkt) { Encoder *e = ost->enc; AVCodecContext *enc = ost->enc_ctx; - AVPacket *pkt = e->pkt; const char *type_desc = av_get_media_type_string(enc->codec_type); const char *action = frame ? "encode" : "flush"; int ret; @@ -664,11 +770,9 @@ static int encode_frame(OutputFile *of, OutputStream *ost, AVFrame *frame) if (ret == AVERROR(EAGAIN)) { av_assert0(frame); // should never happen during flushing return 0; - } else if (ret == AVERROR_EOF) { - ret = of_output_packet(of, ost, NULL); - return ret < 0 ? ret : AVERROR_EOF; } else if (ret < 0) { - av_log(ost, AV_LOG_ERROR, "%s encoding failed\n", type_desc); + if (ret != AVERROR_EOF) + av_log(ost, AV_LOG_ERROR, "%s encoding failed\n", type_desc); return ret; } @@ -703,22 +807,24 @@ static int encode_frame(OutputFile *of, OutputStream *ost, AVFrame *frame) e->packets_encoded++; - ret = of_output_packet(of, ost, pkt); - if (ret < 0) + ret = tq_send(e->queue_out, 0, pkt); + if (ret < 0) { + av_packet_unref(pkt); return ret; + } } av_assert0(0); } static int submit_encode_frame(OutputFile *of, OutputStream *ost, - AVFrame *frame) + AVFrame *frame, AVPacket *pkt) { Encoder *e = ost->enc; int ret; if (ost->sq_idx_encode < 0) - return encode_frame(of, ost, frame); + return encode_frame(of, ost, frame, pkt); if (frame) { ret = av_frame_ref(e->sq_frame, frame); @@ -747,22 +853,18 @@ static int submit_encode_frame(OutputFile *of, OutputStream *ost, return (ret == AVERROR(EAGAIN)) ? 0 : ret; } - ret = encode_frame(of, ost, enc_frame); + ret = encode_frame(of, ost, enc_frame, pkt); if (enc_frame) av_frame_unref(enc_frame); - if (ret < 0) { - if (ret == AVERROR_EOF) - close_output_stream(ost); + if (ret < 0) return ret; - } } } static int do_audio_out(OutputFile *of, OutputStream *ost, - AVFrame *frame) + AVFrame *frame, AVPacket *pkt) { AVCodecContext *enc = ost->enc_ctx; - int ret; if (!(enc->codec->capabilities & AV_CODEC_CAP_PARAM_CHANGE) && enc->ch_layout.nb_channels != frame->ch_layout.nb_channels) { @@ -772,10 +874,9 @@ static int do_audio_out(OutputFile *of, OutputStream *ost, } if (!check_recording_time(ost, frame->pts, frame->time_base)) - return 0; + return AVERROR_EOF; - ret = submit_encode_frame(of, ost, frame); - return (ret < 0 && ret != AVERROR_EOF) ? ret : 0; + return submit_encode_frame(of, ost, frame, pkt); } static enum AVPictureType forced_kf_apply(void *logctx, KeyframeForceCtx *kf, @@ -825,13 +926,13 @@ force_keyframe: } /* May modify/reset frame */ -static int do_video_out(OutputFile *of, OutputStream *ost, AVFrame *in_picture) +static int do_video_out(OutputFile *of, OutputStream *ost, + AVFrame *in_picture, AVPacket *pkt) { - int ret; AVCodecContext *enc = ost->enc_ctx; if (!check_recording_time(ost, in_picture->pts, ost->enc_ctx->time_base)) - return 0; + return AVERROR_EOF; in_picture->quality = enc->global_quality; in_picture->pict_type = forced_kf_apply(ost, &ost->kf, enc->time_base, in_picture); @@ -843,26 +944,203 @@ static int do_video_out(OutputFile *of, OutputStream *ost, AVFrame *in_picture) } #endif - ret = submit_encode_frame(of, ost, in_picture); - return (ret == AVERROR_EOF) ? 0 : ret; + return submit_encode_frame(of, ost, in_picture, pkt); +} + +static int frame_encode(OutputStream *ost, AVFrame *frame, AVPacket *pkt) +{ + OutputFile *of = output_files[ost->file_index]; + enum AVMediaType type = ost->type; + + if (type == AVMEDIA_TYPE_SUBTITLE) { + // no flushing for subtitles + return frame ? + do_subtitle_out(of, ost, (AVSubtitle*)frame->buf[0]->data, pkt) : 0; + } + + if (frame) { + return (type == AVMEDIA_TYPE_VIDEO) ? do_video_out(of, ost, frame, pkt) : + do_audio_out(of, ost, frame, pkt); + } + + return submit_encode_frame(of, ost, NULL, pkt); +} + +static void enc_thread_set_name(const OutputStream *ost) +{ + char name[16]; + snprintf(name, sizeof(name), "enc%d:%d:%s", ost->file_index, ost->index, + ost->enc_ctx->codec->name); + ff_thread_setname(name); +} + +static void enc_thread_uninit(EncoderThread *et) +{ + av_packet_free(&et->pkt); + av_frame_free(&et->frame); + + memset(et, 0, sizeof(*et)); +} + +static int enc_thread_init(EncoderThread *et) +{ + memset(et, 0, sizeof(*et)); + + et->frame = av_frame_alloc(); + if (!et->frame) + goto fail; + + et->pkt = av_packet_alloc(); + if (!et->pkt) + goto fail; + + return 0; + +fail: + enc_thread_uninit(et); + return AVERROR(ENOMEM); +} + +static void *encoder_thread(void *arg) +{ + OutputStream *ost = arg; + OutputFile *of = output_files[ost->file_index]; + Encoder *e = ost->enc; + EncoderThread et; + int ret = 0, input_status = 0; + + ret = enc_thread_init(&et); + if (ret < 0) + goto finish; + + enc_thread_set_name(ost); + + while (!input_status) { + int dummy; + + input_status = tq_receive(e->queue_in, &dummy, et.frame); + if (input_status < 0) + av_log(ost, AV_LOG_VERBOSE, "Encoder thread received EOF\n"); + + ret = frame_encode(ost, input_status >= 0 ? et.frame : NULL, et.pkt); + + av_packet_unref(et.pkt); + av_frame_unref(et.frame); + + if (ret < 0) { + if (ret == AVERROR_EOF) + av_log(ost, AV_LOG_VERBOSE, "Encoder returned EOF, finishing\n"); + else + av_log(ost, AV_LOG_ERROR, "Error encoding a frame: %s\n", + av_err2str(ret)); + break; + } + + // signal to the consumer thread that the frame was encoded + ret = tq_send(e->queue_out, 0, et.pkt); + if (ret < 0) { + if (ret != AVERROR_EOF) + av_log(ost, AV_LOG_ERROR, + "Error communicating with the main thread\n"); + break; + } + } + + // EOF is normal thread termination + if (ret == AVERROR_EOF) + ret = 0; + +finish: + if (ost->sq_idx_encode >= 0) + sq_send(of->sq_encode, ost->sq_idx_encode, SQFRAME(NULL)); + + tq_receive_finish(e->queue_in, 0); + tq_send_finish (e->queue_out, 0); + + enc_thread_uninit(&et); + + av_log(ost, AV_LOG_VERBOSE, "Terminating encoder thread\n"); + + return (void*)(intptr_t)ret; } int enc_frame(OutputStream *ost, AVFrame *frame) { OutputFile *of = output_files[ost->file_index]; - int ret; + Encoder *e = ost->enc; + int ret, thread_ret; ret = enc_open(ost, frame); if (ret < 0) return ret; - return ost->enc_ctx->codec_type == AVMEDIA_TYPE_VIDEO ? - do_video_out(of, ost, frame) : do_audio_out(of, ost, frame); + if (!e->queue_in) + return AVERROR_EOF; + + // send the frame/EOF to the encoder thread + if (frame) { + ret = tq_send(e->queue_in, 0, frame); + if (ret < 0) + goto finish; + } else + tq_send_finish(e->queue_in, 0); + + // retrieve all encoded data for the frame + while (1) { + int dummy; + + ret = tq_receive(e->queue_out, &dummy, e->pkt); + if (ret < 0) + break; + + // frame fully encoded + if (!e->pkt->data && !e->pkt->side_data_elems) + return 0; + + // process the encoded packet + ret = of_output_packet(of, ost, e->pkt); + if (ret < 0) + goto finish; + } + +finish: + thread_ret = enc_thread_stop(e); + if (thread_ret < 0) { + av_log(ost, AV_LOG_ERROR, "Encoder thread returned error: %s\n", + av_err2str(thread_ret)); + ret = err_merge(ret, thread_ret); + } + + if (ret < 0 && ret != AVERROR_EOF) + return ret; + + // signal EOF to the muxer + return of_output_packet(of, ost, NULL); +} + +int enc_subtitle(OutputFile *of, OutputStream *ost, const AVSubtitle *sub) +{ + Encoder *e = ost->enc; + AVFrame *f = e->sub_frame; + int ret; + + // XXX the queue for transferring data to the encoder thread runs + // on AVFrames, so we wrap AVSubtitle in an AVBufferRef and put + // that inside the frame + // eventually, subtitles should be switched to use AVFrames natively + ret = subtitle_wrap_frame(f, sub, 1); + if (ret < 0) + return ret; + + ret = enc_frame(ost, f); + av_frame_unref(f); + + return ret; } int enc_flush(void) { - int ret; + int ret = 0; for (OutputStream *ost = ost_iter(NULL); ost; ost = ost_iter(ost)) { OutputFile *of = output_files[ost->file_index]; @@ -873,16 +1151,18 @@ int enc_flush(void) for (OutputStream *ost = ost_iter(NULL); ost; ost = ost_iter(ost)) { Encoder *e = ost->enc; AVCodecContext *enc = ost->enc_ctx; - OutputFile *of = output_files[ost->file_index]; + int err; if (!enc || !e->opened || (enc->codec_type != AVMEDIA_TYPE_VIDEO && enc->codec_type != AVMEDIA_TYPE_AUDIO)) continue; - ret = submit_encode_frame(of, ost, NULL); - if (ret != AVERROR_EOF) - return ret; + err = enc_frame(ost, NULL); + if (err != AVERROR_EOF && ret < 0) + ret = err_merge(ret, err); + + av_assert0(!e->queue_in); } - return 0; + return ret; }