From 7b6622705e552b1dd5284b743974b66abe04954b Mon Sep 17 00:00:00 2001 From: Anton Khirnov Date: Wed, 3 Aug 2022 14:08:44 +0200 Subject: fftools/ffmpeg: move the input thread into its own file It will contain more demuxing-specific code in the future. --- fftools/Makefile | 1 + fftools/ffmpeg.c | 110 --------------------------------------- fftools/ffmpeg.h | 5 ++ fftools/ffmpeg_demux.c | 136 +++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 142 insertions(+), 110 deletions(-) create mode 100644 fftools/ffmpeg_demux.c diff --git a/fftools/Makefile b/fftools/Makefile index 6285e6eacb..5907f5c57e 100644 --- a/fftools/Makefile +++ b/fftools/Makefile @@ -10,6 +10,7 @@ ALLAVPROGS = $(AVBASENAMES:%=%$(PROGSSUF)$(EXESUF)) ALLAVPROGS_G = $(AVBASENAMES:%=%$(PROGSSUF)_g$(EXESUF)) OBJS-ffmpeg += \ + fftools/ffmpeg_demux.o \ fftools/ffmpeg_filter.o \ fftools/ffmpeg_hw.o \ fftools/ffmpeg_mux.o \ diff --git a/fftools/ffmpeg.c b/fftools/ffmpeg.c index 46de9ce40b..72cba3a525 100644 --- a/fftools/ffmpeg.c +++ b/fftools/ffmpeg.c @@ -161,8 +161,6 @@ static struct termios oldtty; static int restore_tty; #endif -static void free_input_threads(void); - /* sub2video hack: Convert subtitles to video with alpha to insert them in filter graphs. This is a temporary solution until libavfilter gets real subtitles support. @@ -3636,114 +3634,6 @@ static int check_keyboard_interaction(int64_t cur_time) return 0; } -static void *input_thread(void *arg) -{ - InputFile *f = arg; - AVPacket *pkt = f->pkt, *queue_pkt; - unsigned flags = f->non_blocking ? AV_THREAD_MESSAGE_NONBLOCK : 0; - int ret = 0; - - while (1) { - ret = av_read_frame(f->ctx, pkt); - - if (ret == AVERROR(EAGAIN)) { - av_usleep(10000); - continue; - } - if (ret < 0) { - av_thread_message_queue_set_err_recv(f->in_thread_queue, ret); - break; - } - queue_pkt = av_packet_alloc(); - if (!queue_pkt) { - av_packet_unref(pkt); - av_thread_message_queue_set_err_recv(f->in_thread_queue, AVERROR(ENOMEM)); - break; - } - av_packet_move_ref(queue_pkt, pkt); - ret = av_thread_message_queue_send(f->in_thread_queue, &queue_pkt, flags); - if (flags && ret == AVERROR(EAGAIN)) { - flags = 0; - ret = av_thread_message_queue_send(f->in_thread_queue, &queue_pkt, flags); - av_log(f->ctx, AV_LOG_WARNING, - "Thread message queue blocking; consider raising the " - "thread_queue_size option (current value: %d)\n", - f->thread_queue_size); - } - if (ret < 0) { - if (ret != AVERROR_EOF) - av_log(f->ctx, AV_LOG_ERROR, - "Unable to send packet to main thread: %s\n", - av_err2str(ret)); - av_packet_free(&queue_pkt); - av_thread_message_queue_set_err_recv(f->in_thread_queue, ret); - break; - } - } - - return NULL; -} - -static void free_input_thread(int i) -{ - InputFile *f = input_files[i]; - AVPacket *pkt; - - if (!f || !f->in_thread_queue) - return; - av_thread_message_queue_set_err_send(f->in_thread_queue, AVERROR_EOF); - while (av_thread_message_queue_recv(f->in_thread_queue, &pkt, 0) >= 0) - av_packet_free(&pkt); - - pthread_join(f->thread, NULL); - av_thread_message_queue_free(&f->in_thread_queue); -} - -static void free_input_threads(void) -{ - int i; - - for (i = 0; i < nb_input_files; i++) - free_input_thread(i); -} - -static int init_input_thread(int i) -{ - int ret; - InputFile *f = input_files[i]; - - if (f->thread_queue_size <= 0) - f->thread_queue_size = (nb_input_files > 1 ? 8 : 1); - - if (f->ctx->pb ? !f->ctx->pb->seekable : - strcmp(f->ctx->iformat->name, "lavfi")) - f->non_blocking = 1; - ret = av_thread_message_queue_alloc(&f->in_thread_queue, - f->thread_queue_size, sizeof(f->pkt)); - if (ret < 0) - return ret; - - if ((ret = pthread_create(&f->thread, NULL, input_thread, f))) { - av_log(NULL, AV_LOG_ERROR, "pthread_create failed: %s. Try to increase `ulimit -v` or decrease `ulimit -s`.\n", strerror(ret)); - av_thread_message_queue_free(&f->in_thread_queue); - return AVERROR(ret); - } - - return 0; -} - -static int init_input_threads(void) -{ - int i, ret; - - for (i = 0; i < nb_input_files; i++) { - ret = init_input_thread(i); - if (ret < 0) - return ret; - } - return 0; -} - static int get_input_packet(InputFile *f, AVPacket **pkt) { if (f->readrate || f->rate_emu) { diff --git a/fftools/ffmpeg.h b/fftools/ffmpeg.h index fc082bfbea..81356fd566 100644 --- a/fftools/ffmpeg.h +++ b/fftools/ffmpeg.h @@ -710,4 +710,9 @@ int64_t of_filesize(OutputFile *of); AVChapter * const * of_get_chapters(OutputFile *of, unsigned int *nb_chapters); +int init_input_threads(void); +int init_input_thread(int i); +void free_input_threads(void); +void free_input_thread(int i); + #endif /* FFTOOLS_FFMPEG_H */ diff --git a/fftools/ffmpeg_demux.c b/fftools/ffmpeg_demux.c new file mode 100644 index 0000000000..2629af5950 --- /dev/null +++ b/fftools/ffmpeg_demux.c @@ -0,0 +1,136 @@ +/* + * This file is part of FFmpeg. + * + * FFmpeg is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * FFmpeg is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with FFmpeg; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include "ffmpeg.h" + +#include "libavutil/error.h" +#include "libavutil/time.h" +#include "libavutil/thread.h" +#include "libavutil/threadmessage.h" + +#include "libavcodec/packet.h" + +#include "libavformat/avformat.h" + +static void *input_thread(void *arg) +{ + InputFile *f = arg; + AVPacket *pkt = f->pkt, *queue_pkt; + unsigned flags = f->non_blocking ? AV_THREAD_MESSAGE_NONBLOCK : 0; + int ret = 0; + + while (1) { + ret = av_read_frame(f->ctx, pkt); + + if (ret == AVERROR(EAGAIN)) { + av_usleep(10000); + continue; + } + if (ret < 0) { + av_thread_message_queue_set_err_recv(f->in_thread_queue, ret); + break; + } + queue_pkt = av_packet_alloc(); + if (!queue_pkt) { + av_packet_unref(pkt); + av_thread_message_queue_set_err_recv(f->in_thread_queue, AVERROR(ENOMEM)); + break; + } + av_packet_move_ref(queue_pkt, pkt); + ret = av_thread_message_queue_send(f->in_thread_queue, &queue_pkt, flags); + if (flags && ret == AVERROR(EAGAIN)) { + flags = 0; + ret = av_thread_message_queue_send(f->in_thread_queue, &queue_pkt, flags); + av_log(f->ctx, AV_LOG_WARNING, + "Thread message queue blocking; consider raising the " + "thread_queue_size option (current value: %d)\n", + f->thread_queue_size); + } + if (ret < 0) { + if (ret != AVERROR_EOF) + av_log(f->ctx, AV_LOG_ERROR, + "Unable to send packet to main thread: %s\n", + av_err2str(ret)); + av_packet_free(&queue_pkt); + av_thread_message_queue_set_err_recv(f->in_thread_queue, ret); + break; + } + } + + return NULL; +} + +void free_input_thread(int i) +{ + InputFile *f = input_files[i]; + AVPacket *pkt; + + if (!f || !f->in_thread_queue) + return; + av_thread_message_queue_set_err_send(f->in_thread_queue, AVERROR_EOF); + while (av_thread_message_queue_recv(f->in_thread_queue, &pkt, 0) >= 0) + av_packet_free(&pkt); + + pthread_join(f->thread, NULL); + av_thread_message_queue_free(&f->in_thread_queue); +} + +void free_input_threads(void) +{ + int i; + + for (i = 0; i < nb_input_files; i++) + free_input_thread(i); +} + +int init_input_thread(int i) +{ + int ret; + InputFile *f = input_files[i]; + + if (f->thread_queue_size <= 0) + f->thread_queue_size = (nb_input_files > 1 ? 8 : 1); + + if (f->ctx->pb ? !f->ctx->pb->seekable : + strcmp(f->ctx->iformat->name, "lavfi")) + f->non_blocking = 1; + ret = av_thread_message_queue_alloc(&f->in_thread_queue, + f->thread_queue_size, sizeof(f->pkt)); + if (ret < 0) + return ret; + + if ((ret = pthread_create(&f->thread, NULL, input_thread, f))) { + av_log(NULL, AV_LOG_ERROR, "pthread_create failed: %s. Try to increase `ulimit -v` or decrease `ulimit -s`.\n", strerror(ret)); + av_thread_message_queue_free(&f->in_thread_queue); + return AVERROR(ret); + } + + return 0; +} + +int init_input_threads(void) +{ + int i, ret; + + for (i = 0; i < nb_input_files; i++) { + ret = init_input_thread(i); + if (ret < 0) + return ret; + } + return 0; +} -- cgit v1.2.3