From patchwork Sat Nov 4 07:56:28 2023 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Anton Khirnov X-Patchwork-Id: 44520 Delivered-To: ffmpegpatchwork2@gmail.com Received: by 2002:a05:6a20:3aa6:b0:181:818d:5e7f with SMTP id d38csp336861pzh; Sat, 4 Nov 2023 02:24:48 -0700 (PDT) X-Google-Smtp-Source: AGHT+IHq/OPYpnNZdHi80MVSUW9kbAo4gxyXL79U98SZnVS8MxdRq5wicY7TVCkDmSjdbMlSflYv X-Received: by 2002:a17:907:25cd:b0:9c7:4d98:981f with SMTP id ae13-20020a17090725cd00b009c74d98981fmr8267410ejc.33.1699089888026; Sat, 04 Nov 2023 02:24:48 -0700 (PDT) ARC-Seal: i=1; a=rsa-sha256; t=1699089888; cv=none; d=google.com; s=arc-20160816; b=Ydu8eX+Df3Lv/I+N8KL9RbcK/ra1+N119HCfALoDd2k0RiQLQGsepMbEHY19swNzSb gNe32oxeeJjB6O5wd9Oc4t+Hi7TneUVjat70DdU0hcPzQawFx8Nbr2fmb4HRPjIoJ0HN CRzpmF4JbdvzaBF45f9UJOnr0e2pFZH/YNM0pub2VRppBYhZCnPi8W5r+yx3wzrZRjOp grvPffvizeN8nNoV7Msc/V/7k1PANacfrSQa0yvmQRDcxN+wuXdaLc5/x+SMAdvzX/n8 BcHBYjy25H/ljZUGrC69AV1qmtF0+PQfmvEVN+fYWiR7Cl0HLyU0PBwI+q1rdLwgnCHR Q+pQ== 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=9HHrHtQRwGuz1tNinAK7AoVUU9/ySzqbYIXIXP5cee8=; fh=YOA8vD9MJZuwZ71F/05pj6KdCjf6jQRmzLS+CATXUQk=; b=HF6d6XDQ+05ZkHWVqUnFvf47idnHiT13gZF578ClTQtY3NbWa0xUOgQ+IwImEX5BRM FmTjas0USPdk+BVgu2yy8WngpJweEI5Le3f+pMxh29fczHvGAjSLsgy8l5TyzfguCzUx UU+kpMAdGzJLkdJPUZ5WrMOiZ14k5mNqc4AotrgKigMLb4IlqomD2qDOx4FVuH+LHsp5 sAQXGJ04OaQQudUkEUKSqPCQkO5QPlaCfkcKycjOFbNsOzWcg53SUSQhjz+Yq7n2gVQJ 2iy28Xy4xGYb3vIeEiIWRfd9/DIOWPjNljmdCOOhF4NyiUNKjbMgGBCH355Bg+KQFGTV hxgA== 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 s1-20020a17090699c100b00992a0f83dfcsi2054142ejn.471.2023.11.04.02.24.47; Sat, 04 Nov 2023 02:24:48 -0700 (PDT) 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 8359A68CEB7; Sat, 4 Nov 2023 11:22:14 +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 63E9B68CE46 for ; Sat, 4 Nov 2023 11:21:55 +0200 (EET) Received: from localhost (mail1.khirnov.net [IPv6:::1]) by mail1.khirnov.net (Postfix) with ESMTP id 98B6B1322 for ; Sat, 4 Nov 2023 10:21:52 +0100 (CET) Received: from mail1.khirnov.net ([IPv6:::1]) by localhost (mail1.khirnov.net [IPv6:::1]) (amavis, port 10024) with ESMTP id UdPgeOZPiF-o for ; Sat, 4 Nov 2023 10:21:52 +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 C56AE13B0 for ; Sat, 4 Nov 2023 10:21:47 +0100 (CET) Received: from libav.khirnov.net (libav.khirnov.net [IPv6:::1]) by libav.khirnov.net (Postfix) with ESMTP id 01B583A15AC for ; Sat, 4 Nov 2023 10:21:41 +0100 (CET) From: Anton Khirnov To: ffmpeg-devel@ffmpeg.org Date: Sat, 4 Nov 2023 08:56:28 +0100 Message-ID: <20231104092125.10213-20-anton@khirnov.net> X-Mailer: git-send-email 2.42.0 In-Reply-To: <20231104092125.10213-1-anton@khirnov.net> References: <20231104092125.10213-1-anton@khirnov.net> MIME-Version: 1.0 Subject: [FFmpeg-devel] [PATCH 19/24] fftools/ffmpeg_demux: convert to the scheduler 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: z9IALlBuE7rm --- fftools/ffmpeg.c | 12 +- fftools/ffmpeg.h | 21 +--- fftools/ffmpeg_demux.c | 268 ++++++++++++++++++++--------------------- 3 files changed, 134 insertions(+), 167 deletions(-) diff --git a/fftools/ffmpeg.c b/fftools/ffmpeg.c index 1a58bf98cf..611ac4621d 100644 --- a/fftools/ffmpeg.c +++ b/fftools/ffmpeg.c @@ -791,16 +791,6 @@ static int process_input_packet(InputStream *ist, const AVPacket *pkt, int no_eo dts_est = pd->dts_est; } - if (f->recording_time != INT64_MAX) { - int64_t start_time = 0; - if (copy_ts) { - start_time += f->start_time != AV_NOPTS_VALUE ? f->start_time : 0; - start_time += start_at_zero ? 0 : f->start_time_effective; - } - if (dts_est >= f->recording_time + start_time) - pkt = NULL; - } - for (int oidx = 0; oidx < ist->nb_outputs; oidx++) { OutputStream *ost = ist->outputs[oidx]; if (ost->enc || (!pkt && no_eof)) @@ -1029,7 +1019,7 @@ static int process_input(int file_index, AVPacket *pkt) InputStream *ist; int ret, i; - ret = ifile_get_packet(ifile, pkt); + ret = 0; if (ret == 1) { /* the input file is looped: flush the decoders */ diff --git a/fftools/ffmpeg.h b/fftools/ffmpeg.h index 5833f85ab5..73b3e54fb0 100644 --- a/fftools/ffmpeg.h +++ b/fftools/ffmpeg.h @@ -89,6 +89,10 @@ enum FrameOpaque { FRAME_OPAQUE_SEND_COMMAND, }; +enum PacketOpaque { + PKT_OPAQUE_SUB_HEARTBEAT = 1, +}; + typedef struct HWDevice { const char *name; enum AVHWDeviceType type; @@ -424,11 +428,6 @@ typedef struct InputFile { float readrate; int accurate_seek; - - /* when looping the input file, this queue is used by decoders to report - * the last frame timestamp back to the demuxer thread */ - AVThreadMessageQueue *audio_ts_queue; - int audio_ts_queue_size; } InputFile; enum forced_keyframes_const { @@ -842,18 +841,6 @@ int64_t of_filesize(OutputFile *of); int ifile_open(const OptionsContext *o, const char *filename, Scheduler *sch); void ifile_close(InputFile **f); -/** - * Get next input packet from the demuxer. - * - * @param pkt the packet is written here when this function returns 0 - * @return - * - 0 when a packet has been read successfully - * - 1 when stream end was reached, but the stream is looped; - * caller should flush decoders and read from this demuxer again - * - a negative error code on failure - */ -int ifile_get_packet(InputFile *f, AVPacket *pkt); - int ist_output_add(InputStream *ist, OutputStream *ost); int ist_filter_add(InputStream *ist, InputFilter *ifilter, int is_simple); diff --git a/fftools/ffmpeg_demux.c b/fftools/ffmpeg_demux.c index 2234dbe076..91cd7a1125 100644 --- a/fftools/ffmpeg_demux.c +++ b/fftools/ffmpeg_demux.c @@ -22,8 +22,6 @@ #include "ffmpeg.h" #include "ffmpeg_sched.h" #include "ffmpeg_utils.h" -#include "objpool.h" -#include "thread_queue.h" #include "libavutil/avassert.h" #include "libavutil/avstring.h" @@ -35,7 +33,6 @@ #include "libavutil/pixdesc.h" #include "libavutil/time.h" #include "libavutil/timestamp.h" -#include "libavutil/thread.h" #include "libavcodec/packet.h" @@ -66,7 +63,11 @@ typedef struct DemuxStream { double ts_scale; + // scheduler returned EOF for this stream + int finished; + int streamcopy_needed; + int have_sub2video; int wrap_correction_done; int saw_first_ts; @@ -101,6 +102,7 @@ typedef struct Demuxer { /* number of times input stream should be looped */ int loop; + int have_audio_dec; /* duration of the looped segment of the input file */ Timestamp duration; /* pts with the smallest/largest values ever seen */ @@ -113,11 +115,12 @@ typedef struct Demuxer { double readrate_initial_burst; Scheduler *sch; - ThreadQueue *thread_queue; - int thread_queue_size; - pthread_t thread; + + AVPacket *pkt_heartbeat; int read_started; + int nb_streams_used; + int nb_streams_finished; } Demuxer; static DemuxStream *ds_from_ist(InputStream *ist) @@ -153,7 +156,7 @@ static void report_new_stream(Demuxer *d, const AVPacket *pkt) d->nb_streams_warn = pkt->stream_index + 1; } -static int seek_to_start(Demuxer *d) +static int seek_to_start(Demuxer *d, Timestamp end_pts) { InputFile *ifile = &d->f; AVFormatContext *is = ifile->ctx; @@ -163,21 +166,10 @@ static int seek_to_start(Demuxer *d) if (ret < 0) return ret; - if (ifile->audio_ts_queue_size) { - int got_ts = 0; - - while (got_ts < ifile->audio_ts_queue_size) { - Timestamp ts; - ret = av_thread_message_queue_recv(ifile->audio_ts_queue, &ts, 0); - if (ret < 0) - return ret; - got_ts++; - - if (d->max_pts.ts == AV_NOPTS_VALUE || - av_compare_ts(d->max_pts.ts, d->max_pts.tb, ts.ts, ts.tb) < 0) - d->max_pts = ts; - } - } + if (end_pts.ts != AV_NOPTS_VALUE && + (d->max_pts.ts == AV_NOPTS_VALUE || + av_compare_ts(d->max_pts.ts, d->max_pts.tb, end_pts.ts, end_pts.tb) < 0)) + d->max_pts = end_pts; if (d->max_pts.ts != AV_NOPTS_VALUE) { int64_t min_pts = d->min_pts.ts == AV_NOPTS_VALUE ? 0 : d->min_pts.ts; @@ -404,7 +396,7 @@ static int ts_fixup(Demuxer *d, AVPacket *pkt) duration = av_rescale_q(d->duration.ts, d->duration.tb, pkt->time_base); if (pkt->pts != AV_NOPTS_VALUE) { // audio decoders take precedence for estimating total file duration - int64_t pkt_duration = ifile->audio_ts_queue_size ? 0 : pkt->duration; + int64_t pkt_duration = d->have_audio_dec ? 0 : pkt->duration; pkt->pts += duration; @@ -440,7 +432,7 @@ static int ts_fixup(Demuxer *d, AVPacket *pkt) return 0; } -static int input_packet_process(Demuxer *d, AVPacket *pkt) +static int input_packet_process(Demuxer *d, AVPacket *pkt, unsigned *send_flags) { InputFile *f = &d->f; InputStream *ist = f->streams[pkt->stream_index]; @@ -451,6 +443,16 @@ static int input_packet_process(Demuxer *d, AVPacket *pkt) if (ret < 0) return ret; + if (f->recording_time != INT64_MAX) { + int64_t start_time = 0; + if (copy_ts) { + start_time += f->start_time != AV_NOPTS_VALUE ? f->start_time : 0; + start_time += start_at_zero ? 0 : f->start_time_effective; + } + if (ds->dts >= f->recording_time + start_time) + *send_flags |= DEMUX_SEND_STREAMCOPY_EOF; + } + ds->data_size += pkt->size; ds->nb_packets++; @@ -465,6 +467,8 @@ static int input_packet_process(Demuxer *d, AVPacket *pkt) av_ts2timestr(input_files[ist->file_index]->ts_offset, &AV_TIME_BASE_Q)); } + pkt->stream_index = ds->sch_idx_stream; + return 0; } @@ -488,6 +492,65 @@ static void readrate_sleep(Demuxer *d) } } +static int do_send(Demuxer *d, DemuxStream *ds, AVPacket *pkt, unsigned flags, + const char *pkt_desc) +{ + int ret; + + ret = sch_demux_send(d->sch, d->f.index, pkt, flags); + if (ret == AVERROR_EOF) { + av_packet_unref(pkt); + + av_log(ds, AV_LOG_VERBOSE, "All consumers of this stream are done\n"); + ds->finished = 1; + + if (++d->nb_streams_finished == d->nb_streams_used) { + av_log(d, AV_LOG_VERBOSE, "All consumers are done\n"); + return AVERROR_EOF; + } + } else if (ret < 0) { + if (ret != AVERROR_EXIT) + av_log(d, AV_LOG_ERROR, + "Unable to send %s packet to consumers: %s\n", + pkt_desc, av_err2str(ret)); + return ret; + } + + return 0; +} + +static int demux_send(Demuxer *d, DemuxStream *ds, AVPacket *pkt, unsigned flags) +{ + InputFile *f = &d->f; + int ret; + + // send heartbeat for sub2video streams + if (d->pkt_heartbeat && pkt->pts != AV_NOPTS_VALUE) { + for (int i = 0; i < f->nb_streams; i++) { + DemuxStream *ds1 = ds_from_ist(f->streams[i]); + + if (ds1->finished || !ds1->have_sub2video) + continue; + + d->pkt_heartbeat->pts = pkt->pts; + d->pkt_heartbeat->time_base = pkt->time_base; + d->pkt_heartbeat->stream_index = ds1->sch_idx_stream; + d->pkt_heartbeat->opaque = (void*)(intptr_t)PKT_OPAQUE_SUB_HEARTBEAT; + + ret = do_send(d, ds1, d->pkt_heartbeat, 0, "heartbeat"); + if (ret < 0) + return ret; + } + } + + ret = do_send(d, ds, pkt, flags, "demuxed"); + if (ret < 0) + return ret; + + + return 0; +} + static void discard_unused_programs(InputFile *ifile) { for (int j = 0; j < ifile->ctx->nb_programs; j++) { @@ -527,9 +590,13 @@ static void *input_thread(void *arg) discard_unused_programs(f); + d->read_started = 1; d->wallclock_start = av_gettime_relative(); while (1) { + DemuxStream *ds; + unsigned send_flags = 0; + ret = av_read_frame(f->ctx, pkt); if (ret == AVERROR(EAGAIN)) { @@ -538,11 +605,13 @@ static void *input_thread(void *arg) } if (ret < 0) { if (d->loop) { - /* signal looping to the consumer thread */ + /* signal looping to our consumers */ pkt->stream_index = -1; - ret = tq_send(d->thread_queue, 0, pkt); + + ret = sch_demux_send(d->sch, f->index, pkt, 0); if (ret >= 0) - ret = seek_to_start(d); + ret = seek_to_start(d, (Timestamp){ .ts = pkt->pts, + .tb = pkt->time_base }); if (ret >= 0) continue; @@ -551,9 +620,11 @@ static void *input_thread(void *arg) if (ret == AVERROR_EOF) av_log(d, AV_LOG_VERBOSE, "EOF while reading input\n"); - else + else { av_log(d, AV_LOG_ERROR, "Error during demuxing: %s\n", av_err2str(ret)); + ret = exit_on_error ? ret : 0; + } break; } @@ -565,8 +636,9 @@ static void *input_thread(void *arg) /* the following test is needed in case new streams appear dynamically in stream : we ignore them */ - if (pkt->stream_index >= f->nb_streams || - f->streams[pkt->stream_index]->discard) { + ds = pkt->stream_index < f->nb_streams ? + ds_from_ist(f->streams[pkt->stream_index]) : NULL; + if (!ds || ds->ist.discard || ds->finished) { report_new_stream(d, pkt); av_packet_unref(pkt); continue; @@ -583,122 +655,26 @@ static void *input_thread(void *arg) } } - ret = input_packet_process(d, pkt); + ret = input_packet_process(d, pkt, &send_flags); if (ret < 0) break; if (f->readrate) readrate_sleep(d); - ret = tq_send(d->thread_queue, 0, pkt); - if (ret < 0) { - if (ret != AVERROR_EOF) - av_log(f, AV_LOG_ERROR, - "Unable to send packet to main thread: %s\n", - av_err2str(ret)); + ret = demux_send(d, ds, pkt, send_flags); + if (ret < 0) break; - } } + // EOF/EXIT is normal termination + if (ret == AVERROR_EOF || ret == AVERROR_EXIT) + ret = 0; + finish: - av_assert0(ret < 0); - tq_send_finish(d->thread_queue, 0); - av_packet_free(&pkt); - av_log(d, AV_LOG_VERBOSE, "Terminating demuxer thread\n"); - - return NULL; -} - -static void thread_stop(Demuxer *d) -{ - InputFile *f = &d->f; - - if (!d->thread_queue) - return; - - tq_receive_finish(d->thread_queue, 0); - - pthread_join(d->thread, NULL); - - tq_free(&d->thread_queue); - - av_thread_message_queue_free(&f->audio_ts_queue); -} - -static int thread_start(Demuxer *d) -{ - int ret; - InputFile *f = &d->f; - ObjPool *op; - - if (d->thread_queue_size <= 0) - d->thread_queue_size = (nb_input_files > 1 ? 8 : 1); - - op = objpool_alloc_packets(); - if (!op) - return AVERROR(ENOMEM); - - d->thread_queue = tq_alloc(1, d->thread_queue_size, op, pkt_move); - if (!d->thread_queue) { - objpool_free(&op); - return AVERROR(ENOMEM); - } - - if (d->loop) { - int nb_audio_dec = 0; - - for (int i = 0; i < f->nb_streams; i++) { - InputStream *ist = f->streams[i]; - nb_audio_dec += !!(ist->decoding_needed && - ist->st->codecpar->codec_type == AVMEDIA_TYPE_AUDIO); - } - - if (nb_audio_dec) { - ret = av_thread_message_queue_alloc(&f->audio_ts_queue, - nb_audio_dec, sizeof(Timestamp)); - if (ret < 0) - goto fail; - f->audio_ts_queue_size = nb_audio_dec; - } - } - - if ((ret = pthread_create(&d->thread, NULL, input_thread, d))) { - av_log(d, AV_LOG_ERROR, "pthread_create failed: %s. Try to increase `ulimit -v` or decrease `ulimit -s`.\n", strerror(ret)); - ret = AVERROR(ret); - goto fail; - } - - d->read_started = 1; - - return 0; -fail: - tq_free(&d->thread_queue); - return ret; -} - -int ifile_get_packet(InputFile *f, AVPacket *pkt) -{ - Demuxer *d = demuxer_from_ifile(f); - int ret, dummy; - - if (!d->thread_queue) { - ret = thread_start(d); - if (ret < 0) - return ret; - } - - ret = tq_receive(d->thread_queue, &dummy, pkt); - if (ret < 0) - return ret; - - if (pkt->stream_index == -1) { - av_assert0(!pkt->data && !pkt->side_data_elems); - return 1; - } - - return 0; + return (void*)(intptr_t)ret; } static void demux_final_stats(Demuxer *d) @@ -769,8 +745,6 @@ void ifile_close(InputFile **pf) if (!f) return; - thread_stop(d); - if (d->read_started) demux_final_stats(d); @@ -780,6 +754,8 @@ void ifile_close(InputFile **pf) avformat_close_input(&f->ctx); + av_packet_free(&d->pkt_heartbeat); + av_freep(pf); } @@ -802,7 +778,11 @@ static int ist_use(InputStream *ist, int decoding_needed) ds->sch_idx_stream = ret; } - ist->discard = 0; + if (ist->discard) { + ist->discard = 0; + d->nb_streams_used++; + } + ist->st->discard = ist->user_set_discard; ist->decoding_needed |= decoding_needed; ds->streamcopy_needed |= !decoding_needed; @@ -823,6 +803,8 @@ static int ist_use(InputStream *ist, int decoding_needed) ret = dec_open(ist, d->sch, ds->sch_idx_dec); if (ret < 0) return ret; + + d->have_audio_dec |= is_audio; } return 0; @@ -848,6 +830,7 @@ int ist_output_add(InputStream *ist, OutputStream *ost) int ist_filter_add(InputStream *ist, InputFilter *ifilter, int is_simple) { + Demuxer *d = demuxer_from_ifile(input_files[ist->file_index]); DemuxStream *ds = ds_from_ist(ist); int ret; @@ -866,6 +849,15 @@ int ist_filter_add(InputStream *ist, InputFilter *ifilter, int is_simple) if (ret < 0) return ret; + if (ist->dec_ctx->codec_type == AVMEDIA_TYPE_SUBTITLE) { + if (!d->pkt_heartbeat) { + d->pkt_heartbeat = av_packet_alloc(); + if (!d->pkt_heartbeat) + return AVERROR(ENOMEM); + } + ds->have_sub2video = 1; + } + return ds->sch_idx_dec; } @@ -1607,8 +1599,6 @@ int ifile_open(const OptionsContext *o, const char *filename, Scheduler *sch) "since neither -readrate nor -re were given\n"); } - d->thread_queue_size = o->thread_queue_size; - /* Add all the streams from the given input file to the demuxer */ for (int i = 0; i < ic->nb_streams; i++) { ret = ist_add(o, d, ic->streams[i]);