3 * Copyright (c) 2016 Jan Sebechlebsky
5 * This file is part of FFmpeg.
7 * FFmpeg is free software; you can redistribute it and/or
8 * modify it under the terms of the GNU Lesser General Public License
9 * as published by the Free Software Foundation; either
10 * version 2.1 of the License, or (at your option) any later version.
12 * FFmpeg is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU Lesser General Public License for more details.
17 * You should have received a copy of the GNU Lesser General Public License
18 * along with FFmpeg; if not, write to the Free Software * Foundation, Inc.,
19 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
22 #include "libavutil/avassert.h"
23 #include "libavutil/opt.h"
24 #include "libavutil/time.h"
25 #include "libavutil/thread.h"
26 #include "libavutil/threadmessage.h"
30 #define FIFO_DEFAULT_QUEUE_SIZE 60
31 #define FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS 0
32 #define FIFO_DEFAULT_RECOVERY_WAIT_TIME_USEC 5000000 // 5 seconds
34 typedef struct FifoContext {
39 AVDictionary *format_options;
42 AVThreadMessageQueue *queue;
44 pthread_t writer_thread;
46 /* Return value of last write_trailer_call */
47 int write_trailer_ret;
49 /* Time to wait before next recovery attempt
50 * This can refer to the time in processed stream,
52 int64_t recovery_wait_time;
54 /* Maximal number of unsuccessful successive recovery attempts */
55 int max_recovery_attempts;
57 /* Whether to attempt recovery from failure */
60 /* If >0 stream time will be used when waiting
61 * for the recovery attempt instead of real time */
62 int recovery_wait_streamtime;
64 /* If >0 recovery will be attempted regardless of error code
65 * (except AVERROR_EXIT, so exit request is never ignored) */
66 int recover_any_error;
68 /* Whether to drop packets in case the queue is full. */
69 int drop_pkts_on_overflow;
71 /* Whether to wait for keyframe when recovering
72 * from failure or queue overflow */
73 int restart_with_keyframe;
75 pthread_mutex_t overflow_flag_lock;
76 int overflow_flag_lock_initialized;
77 /* Value > 0 signals queue overflow */
78 volatile uint8_t overflow_flag;
82 typedef struct FifoThreadContext {
85 /* Timestamp of last failure.
86 * This is either pts in case stream time is used,
87 * or microseconds as returned by av_getttime_relative() */
88 int64_t last_recovery_ts;
90 /* Number of current recovery process
91 * Value > 0 means we are in recovery process */
94 /* If > 0 all frames will be dropped until keyframe is received */
95 uint8_t drop_until_keyframe;
97 /* Value > 0 means that the previous write_header call was successful
98 * so finalization by calling write_trailer and ff_io_close must be done
99 * before exiting / reinitialization of underlying muxer */
100 uint8_t header_written;
103 typedef enum FifoMessageType {
109 typedef struct FifoMessage {
110 FifoMessageType type;
114 static int fifo_thread_write_header(FifoThreadContext *ctx)
116 AVFormatContext *avf = ctx->avf;
117 FifoContext *fifo = avf->priv_data;
118 AVFormatContext *avf2 = fifo->avf;
119 AVDictionary *format_options = NULL;
122 ret = av_dict_copy(&format_options, fifo->format_options, 0);
126 ret = ff_format_output_open(avf2, avf->url, &format_options);
128 av_log(avf, AV_LOG_ERROR, "Error opening %s: %s\n", avf->url,
133 for (i = 0;i < avf2->nb_streams; i++)
134 avf2->streams[i]->cur_dts = 0;
136 ret = avformat_write_header(avf2, &format_options);
138 ctx->header_written = 1;
140 // Check for options unrecognized by underlying muxer
141 if (format_options) {
142 AVDictionaryEntry *entry = NULL;
143 while ((entry = av_dict_get(format_options, "", entry, AV_DICT_IGNORE_SUFFIX)))
144 av_log(avf2, AV_LOG_ERROR, "Unknown option '%s'\n", entry->key);
145 ret = AVERROR(EINVAL);
149 av_dict_free(&format_options);
153 static int fifo_thread_flush_output(FifoThreadContext *ctx)
155 AVFormatContext *avf = ctx->avf;
156 FifoContext *fifo = avf->priv_data;
157 AVFormatContext *avf2 = fifo->avf;
159 return av_write_frame(avf2, NULL);
162 static int fifo_thread_write_packet(FifoThreadContext *ctx, AVPacket *pkt)
164 AVFormatContext *avf = ctx->avf;
165 FifoContext *fifo = avf->priv_data;
166 AVFormatContext *avf2 = fifo->avf;
167 AVRational src_tb, dst_tb;
170 if (ctx->drop_until_keyframe) {
171 if (pkt->flags & AV_PKT_FLAG_KEY) {
172 ctx->drop_until_keyframe = 0;
173 av_log(avf, AV_LOG_VERBOSE, "Keyframe received, recovering...\n");
175 av_log(avf, AV_LOG_VERBOSE, "Dropping non-keyframe packet\n");
176 av_packet_unref(pkt);
181 s_idx = pkt->stream_index;
182 src_tb = avf->streams[s_idx]->time_base;
183 dst_tb = avf2->streams[s_idx]->time_base;
184 av_packet_rescale_ts(pkt, src_tb, dst_tb);
186 ret = av_write_frame(avf2, pkt);
188 av_packet_unref(pkt);
192 static int fifo_thread_write_trailer(FifoThreadContext *ctx)
194 AVFormatContext *avf = ctx->avf;
195 FifoContext *fifo = avf->priv_data;
196 AVFormatContext *avf2 = fifo->avf;
199 if (!ctx->header_written)
202 ret = av_write_trailer(avf2);
203 ff_format_io_close(avf2, &avf2->pb);
208 static int fifo_thread_dispatch_message(FifoThreadContext *ctx, FifoMessage *msg)
210 int ret = AVERROR(EINVAL);
212 if (!ctx->header_written) {
213 ret = fifo_thread_write_header(ctx);
219 case FIFO_WRITE_HEADER:
220 av_assert0(ret >= 0);
222 case FIFO_WRITE_PACKET:
223 return fifo_thread_write_packet(ctx, &msg->pkt);
224 case FIFO_FLUSH_OUTPUT:
225 return fifo_thread_flush_output(ctx);
229 return AVERROR(EINVAL);
232 static int is_recoverable(const FifoContext *fifo, int err_no) {
233 if (!fifo->attempt_recovery)
236 if (fifo->recover_any_error)
237 return err_no != AVERROR_EXIT;
240 case AVERROR(EINVAL):
241 case AVERROR(ENOSYS):
244 case AVERROR_PATCHWELCOME:
251 static void free_message(void *msg)
253 FifoMessage *fifo_msg = msg;
255 if (fifo_msg->type == FIFO_WRITE_PACKET)
256 av_packet_unref(&fifo_msg->pkt);
259 static int fifo_thread_process_recovery_failure(FifoThreadContext *ctx, AVPacket *pkt,
262 AVFormatContext *avf = ctx->avf;
263 FifoContext *fifo = avf->priv_data;
266 av_log(avf, AV_LOG_INFO, "Recovery failed: %s\n",
269 if (fifo->recovery_wait_streamtime) {
270 if (pkt->pts == AV_NOPTS_VALUE)
271 av_log(avf, AV_LOG_WARNING, "Packet does not contain presentation"
272 " timestamp, recovery will be attempted immediately");
273 ctx->last_recovery_ts = pkt->pts;
275 ctx->last_recovery_ts = av_gettime_relative();
278 if (fifo->max_recovery_attempts &&
279 ctx->recovery_nr >= fifo->max_recovery_attempts) {
280 av_log(avf, AV_LOG_ERROR,
281 "Maximal number of %d recovery attempts reached.\n",
282 fifo->max_recovery_attempts);
285 ret = AVERROR(EAGAIN);
291 static int fifo_thread_attempt_recovery(FifoThreadContext *ctx, FifoMessage *msg, int err_no)
293 AVFormatContext *avf = ctx->avf;
294 FifoContext *fifo = avf->priv_data;
295 AVPacket *pkt = &msg->pkt;
296 int64_t time_since_recovery;
299 if (!is_recoverable(fifo, err_no)) {
304 if (ctx->header_written) {
305 fifo->write_trailer_ret = fifo_thread_write_trailer(ctx);
306 ctx->header_written = 0;
309 if (!ctx->recovery_nr) {
310 ctx->last_recovery_ts = fifo->recovery_wait_streamtime ?
313 if (fifo->recovery_wait_streamtime) {
314 if (ctx->last_recovery_ts == AV_NOPTS_VALUE) {
315 AVRational tb = avf->streams[pkt->stream_index]->time_base;
316 time_since_recovery = av_rescale_q(pkt->pts - ctx->last_recovery_ts,
319 /* Enforce recovery immediately */
320 time_since_recovery = fifo->recovery_wait_time;
323 time_since_recovery = av_gettime_relative() - ctx->last_recovery_ts;
326 if (time_since_recovery < fifo->recovery_wait_time)
327 return AVERROR(EAGAIN);
332 if (fifo->max_recovery_attempts) {
333 av_log(avf, AV_LOG_VERBOSE, "Recovery attempt #%d/%d\n",
334 ctx->recovery_nr, fifo->max_recovery_attempts);
336 av_log(avf, AV_LOG_VERBOSE, "Recovery attempt #%d\n",
340 if (fifo->restart_with_keyframe && fifo->drop_pkts_on_overflow)
341 ctx->drop_until_keyframe = 1;
343 ret = fifo_thread_dispatch_message(ctx, msg);
345 if (is_recoverable(fifo, ret)) {
346 return fifo_thread_process_recovery_failure(ctx, pkt, ret);
351 av_log(avf, AV_LOG_INFO, "Recovery successful\n");
352 ctx->recovery_nr = 0;
362 static int fifo_thread_recover(FifoThreadContext *ctx, FifoMessage *msg, int err_no)
364 AVFormatContext *avf = ctx->avf;
365 FifoContext *fifo = avf->priv_data;
369 if (!fifo->recovery_wait_streamtime && ctx->recovery_nr > 0) {
370 int64_t time_since_recovery = av_gettime_relative() - ctx->last_recovery_ts;
371 int64_t time_to_wait = FFMAX(0, fifo->recovery_wait_time - time_since_recovery);
373 av_usleep(FFMIN(10000, time_to_wait));
376 ret = fifo_thread_attempt_recovery(ctx, msg, err_no);
377 } while (ret == AVERROR(EAGAIN) && !fifo->drop_pkts_on_overflow);
379 if (ret == AVERROR(EAGAIN) && fifo->drop_pkts_on_overflow) {
380 if (msg->type == FIFO_WRITE_PACKET)
381 av_packet_unref(&msg->pkt);
388 static void *fifo_consumer_thread(void *data)
390 AVFormatContext *avf = data;
391 FifoContext *fifo = avf->priv_data;
392 AVThreadMessageQueue *queue = fifo->queue;
393 FifoMessage msg = {FIFO_WRITE_HEADER, {0}};
396 FifoThreadContext fifo_thread_ctx;
397 memset(&fifo_thread_ctx, 0, sizeof(FifoThreadContext));
398 fifo_thread_ctx.avf = avf;
401 uint8_t just_flushed = 0;
403 if (!fifo_thread_ctx.recovery_nr)
404 ret = fifo_thread_dispatch_message(&fifo_thread_ctx, &msg);
406 if (ret < 0 || fifo_thread_ctx.recovery_nr > 0) {
407 int rec_ret = fifo_thread_recover(&fifo_thread_ctx, &msg, ret);
409 av_thread_message_queue_set_err_send(queue, rec_ret);
414 /* If the queue is full at the moment when fifo_write_packet
415 * attempts to insert new message (packet) to the queue,
416 * it sets the fifo->overflow_flag to 1 and drops packet.
417 * Here in consumer thread, the flag is checked and if it is
418 * set, the queue is flushed and flag cleared. */
419 pthread_mutex_lock(&fifo->overflow_flag_lock);
420 if (fifo->overflow_flag) {
421 av_thread_message_flush(queue);
422 if (fifo->restart_with_keyframe)
423 fifo_thread_ctx.drop_until_keyframe = 1;
424 fifo->overflow_flag = 0;
427 pthread_mutex_unlock(&fifo->overflow_flag_lock);
430 av_log(avf, AV_LOG_INFO, "FIFO queue flushed\n");
432 ret = av_thread_message_queue_recv(queue, &msg, 0);
434 av_thread_message_queue_set_err_send(queue, ret);
439 fifo->write_trailer_ret = fifo_thread_write_trailer(&fifo_thread_ctx);
444 static int fifo_mux_init(AVFormatContext *avf, ff_const59 AVOutputFormat *oformat,
445 const char *filename)
447 FifoContext *fifo = avf->priv_data;
448 AVFormatContext *avf2;
451 ret = avformat_alloc_output_context2(&avf2, oformat, NULL, filename);
457 avf2->interrupt_callback = avf->interrupt_callback;
458 avf2->max_delay = avf->max_delay;
459 ret = av_dict_copy(&avf2->metadata, avf->metadata, 0);
462 avf2->opaque = avf->opaque;
463 avf2->io_close = avf->io_close;
464 avf2->io_open = avf->io_open;
465 avf2->flags = avf->flags;
467 for (i = 0; i < avf->nb_streams; ++i) {
468 AVStream *st = avformat_new_stream(avf2, NULL);
470 return AVERROR(ENOMEM);
472 ret = ff_stream_encode_params_copy(st, avf->streams[i]);
480 static int fifo_init(AVFormatContext *avf)
482 FifoContext *fifo = avf->priv_data;
483 ff_const59 AVOutputFormat *oformat;
486 if (fifo->recovery_wait_streamtime && !fifo->drop_pkts_on_overflow) {
487 av_log(avf, AV_LOG_ERROR, "recovery_wait_streamtime can be turned on"
488 " only when drop_pkts_on_overflow is also turned on\n");
489 return AVERROR(EINVAL);
492 oformat = av_guess_format(fifo->format, avf->url, NULL);
494 ret = AVERROR_MUXER_NOT_FOUND;
498 ret = fifo_mux_init(avf, oformat, avf->url);
502 ret = av_thread_message_queue_alloc(&fifo->queue, (unsigned) fifo->queue_size,
503 sizeof(FifoMessage));
507 av_thread_message_queue_set_free_func(fifo->queue, free_message);
509 ret = pthread_mutex_init(&fifo->overflow_flag_lock, NULL);
512 fifo->overflow_flag_lock_initialized = 1;
517 static int fifo_write_header(AVFormatContext *avf)
519 FifoContext * fifo = avf->priv_data;
522 ret = pthread_create(&fifo->writer_thread, NULL, fifo_consumer_thread, avf);
524 av_log(avf, AV_LOG_ERROR, "Failed to start thread: %s\n",
525 av_err2str(AVERROR(ret)));
532 static int fifo_write_packet(AVFormatContext *avf, AVPacket *pkt)
534 FifoContext *fifo = avf->priv_data;
535 FifoMessage msg = {.type = pkt ? FIFO_WRITE_PACKET : FIFO_FLUSH_OUTPUT};
539 av_init_packet(&msg.pkt);
540 ret = av_packet_ref(&msg.pkt,pkt);
545 ret = av_thread_message_queue_send(fifo->queue, &msg,
546 fifo->drop_pkts_on_overflow ?
547 AV_THREAD_MESSAGE_NONBLOCK : 0);
548 if (ret == AVERROR(EAGAIN)) {
549 uint8_t overflow_set = 0;
551 /* Queue is full, set fifo->overflow_flag to 1
552 * to let consumer thread know the queue should
554 pthread_mutex_lock(&fifo->overflow_flag_lock);
555 if (!fifo->overflow_flag)
556 fifo->overflow_flag = overflow_set = 1;
557 pthread_mutex_unlock(&fifo->overflow_flag_lock);
560 av_log(avf, AV_LOG_WARNING, "FIFO queue full\n");
563 } else if (ret < 0) {
570 av_packet_unref(&msg.pkt);
574 static int fifo_write_trailer(AVFormatContext *avf)
576 FifoContext *fifo= avf->priv_data;
579 av_thread_message_queue_set_err_recv(fifo->queue, AVERROR_EOF);
581 ret = pthread_join(fifo->writer_thread, NULL);
583 av_log(avf, AV_LOG_ERROR, "pthread join error: %s\n",
584 av_err2str(AVERROR(ret)));
588 ret = fifo->write_trailer_ret;
592 static void fifo_deinit(AVFormatContext *avf)
594 FifoContext *fifo = avf->priv_data;
596 avformat_free_context(fifo->avf);
597 av_thread_message_queue_free(&fifo->queue);
598 if (fifo->overflow_flag_lock_initialized)
599 pthread_mutex_destroy(&fifo->overflow_flag_lock);
602 #define OFFSET(x) offsetof(FifoContext, x)
603 static const AVOption options[] = {
604 {"fifo_format", "Target muxer", OFFSET(format),
605 AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
607 {"queue_size", "Size of fifo queue", OFFSET(queue_size),
608 AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_QUEUE_SIZE}, 1, INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
610 {"format_opts", "Options to be passed to underlying muxer", OFFSET(format_options),
611 AV_OPT_TYPE_DICT, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
613 {"drop_pkts_on_overflow", "Drop packets on fifo queue overflow not to block encoder", OFFSET(drop_pkts_on_overflow),
614 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
616 {"restart_with_keyframe", "Wait for keyframe when restarting output", OFFSET(restart_with_keyframe),
617 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
619 {"attempt_recovery", "Attempt recovery in case of failure", OFFSET(attempt_recovery),
620 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
622 {"max_recovery_attempts", "Maximal number of recovery attempts", OFFSET(max_recovery_attempts),
623 AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS}, 0, INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
625 {"recovery_wait_time", "Waiting time between recovery attempts", OFFSET(recovery_wait_time),
626 AV_OPT_TYPE_DURATION, {.i64 = FIFO_DEFAULT_RECOVERY_WAIT_TIME_USEC}, 0, INT64_MAX, AV_OPT_FLAG_ENCODING_PARAM},
628 {"recovery_wait_streamtime", "Use stream time instead of real time while waiting for recovery",
629 OFFSET(recovery_wait_streamtime), AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
631 {"recover_any_error", "Attempt recovery regardless of type of the error", OFFSET(recover_any_error),
632 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
637 static const AVClass fifo_muxer_class = {
638 .class_name = "Fifo muxer",
639 .item_name = av_default_item_name,
641 .version = LIBAVUTIL_VERSION_INT,
644 AVOutputFormat ff_fifo_muxer = {
646 .long_name = NULL_IF_CONFIG_SMALL("FIFO queue pseudo-muxer"),
647 .priv_data_size = sizeof(FifoContext),
649 .write_header = fifo_write_header,
650 .write_packet = fifo_write_packet,
651 .write_trailer = fifo_write_trailer,
652 .deinit = fifo_deinit,
653 .priv_class = &fifo_muxer_class,
654 .flags = AVFMT_NOFILE | AVFMT_ALLOW_FLUSH | AVFMT_TS_NEGATIVE,