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 char *format_options_str;
40 AVDictionary *format_options;
43 AVThreadMessageQueue *queue;
45 pthread_t writer_thread;
47 /* Return value of last write_trailer_call */
48 int write_trailer_ret;
50 /* Time to wait before next recovery attempt
51 * This can refer to the time in processed stream,
53 int64_t recovery_wait_time;
55 /* Maximal number of unsuccessful successive recovery attempts */
56 int max_recovery_attempts;
58 /* Whether to attempt recovery from failure */
61 /* If >0 stream time will be used when waiting
62 * for the recovery attempt instead of real time */
63 int recovery_wait_streamtime;
65 /* If >0 recovery will be attempted regardless of error code
66 * (except AVERROR_EXIT, so exit request is never ignored) */
67 int recover_any_error;
69 /* Whether to drop packets in case the queue is full. */
70 int drop_pkts_on_overflow;
72 /* Whether to wait for keyframe when recovering
73 * from failure or queue overflow */
74 int restart_with_keyframe;
76 pthread_mutex_t overflow_flag_lock;
77 int overflow_flag_lock_initialized;
78 /* Value > 0 signals queue overflow */
79 volatile uint8_t overflow_flag;
83 typedef struct FifoThreadContext {
86 /* Timestamp of last failure.
87 * This is either pts in case stream time is used,
88 * or microseconds as returned by av_getttime_relative() */
89 int64_t last_recovery_ts;
91 /* Number of current recovery process
92 * Value > 0 means we are in recovery process */
95 /* If > 0 all frames will be dropped until keyframe is received */
96 uint8_t drop_until_keyframe;
98 /* Value > 0 means that the previous write_header call was successful
99 * so finalization by calling write_trailer and ff_io_close must be done
100 * before exiting / reinitialization of underlying muxer */
101 uint8_t header_written;
104 typedef enum FifoMessageType {
110 typedef struct FifoMessage {
111 FifoMessageType type;
115 static int fifo_thread_write_header(FifoThreadContext *ctx)
117 AVFormatContext *avf = ctx->avf;
118 FifoContext *fifo = avf->priv_data;
119 AVFormatContext *avf2 = fifo->avf;
120 AVDictionary *format_options = NULL;
123 ret = av_dict_copy(&format_options, fifo->format_options, 0);
127 ret = ff_format_output_open(avf2, avf->filename, &format_options);
129 av_log(avf, AV_LOG_ERROR, "Error opening %s: %s\n", avf->filename,
134 for (i = 0;i < avf2->nb_streams; i++)
135 avf2->streams[i]->cur_dts = 0;
137 ret = avformat_write_header(avf2, &format_options);
139 ctx->header_written = 1;
141 // Check for options unrecognized by underlying muxer
142 if (format_options) {
143 AVDictionaryEntry *entry = NULL;
144 while ((entry = av_dict_get(format_options, "", entry, AV_DICT_IGNORE_SUFFIX)))
145 av_log(avf2, AV_LOG_ERROR, "Unknown option '%s'\n", entry->key);
146 ret = AVERROR(EINVAL);
150 av_dict_free(&format_options);
154 static int fifo_thread_flush_output(FifoThreadContext *ctx)
156 AVFormatContext *avf = ctx->avf;
157 FifoContext *fifo = avf->priv_data;
158 AVFormatContext *avf2 = fifo->avf;
160 return av_write_frame(avf2, NULL);
163 static int fifo_thread_write_packet(FifoThreadContext *ctx, AVPacket *pkt)
165 AVFormatContext *avf = ctx->avf;
166 FifoContext *fifo = avf->priv_data;
167 AVFormatContext *avf2 = fifo->avf;
168 AVRational src_tb, dst_tb;
171 if (ctx->drop_until_keyframe) {
172 if (pkt->flags & AV_PKT_FLAG_KEY) {
173 ctx->drop_until_keyframe = 0;
174 av_log(avf, AV_LOG_VERBOSE, "Keyframe received, recovering...\n");
176 av_log(avf, AV_LOG_VERBOSE, "Dropping non-keyframe packet\n");
177 av_packet_unref(pkt);
182 s_idx = pkt->stream_index;
183 src_tb = avf->streams[s_idx]->time_base;
184 dst_tb = avf2->streams[s_idx]->time_base;
185 av_packet_rescale_ts(pkt, src_tb, dst_tb);
187 ret = av_write_frame(avf2, pkt);
189 av_packet_unref(pkt);
193 static int fifo_thread_write_trailer(FifoThreadContext *ctx)
195 AVFormatContext *avf = ctx->avf;
196 FifoContext *fifo = avf->priv_data;
197 AVFormatContext *avf2 = fifo->avf;
200 if (!ctx->header_written)
203 ret = av_write_trailer(avf2);
204 ff_format_io_close(avf2, &avf2->pb);
209 static int fifo_thread_dispatch_message(FifoThreadContext *ctx, FifoMessage *msg)
211 int ret = AVERROR(EINVAL);
213 if (!ctx->header_written) {
214 ret = fifo_thread_write_header(ctx);
220 case FIFO_WRITE_HEADER:
221 av_assert0(ret >= 0);
223 case FIFO_WRITE_PACKET:
224 return fifo_thread_write_packet(ctx, &msg->pkt);
225 case FIFO_FLUSH_OUTPUT:
226 return fifo_thread_flush_output(ctx);
230 return AVERROR(EINVAL);
233 static int is_recoverable(const FifoContext *fifo, int err_no) {
234 if (!fifo->attempt_recovery)
237 if (fifo->recover_any_error)
238 return err_no != AVERROR_EXIT;
241 case AVERROR(EINVAL):
242 case AVERROR(ENOSYS):
245 case AVERROR_PATCHWELCOME:
252 static void free_message(void *msg)
254 FifoMessage *fifo_msg = msg;
256 if (fifo_msg->type == FIFO_WRITE_PACKET)
257 av_packet_unref(&fifo_msg->pkt);
260 static int fifo_thread_process_recovery_failure(FifoThreadContext *ctx, AVPacket *pkt,
263 AVFormatContext *avf = ctx->avf;
264 FifoContext *fifo = avf->priv_data;
267 av_log(avf, AV_LOG_INFO, "Recovery failed: %s\n",
270 if (fifo->recovery_wait_streamtime) {
271 if (pkt->pts == AV_NOPTS_VALUE)
272 av_log(avf, AV_LOG_WARNING, "Packet does not contain presentation"
273 " timestamp, recovery will be attempted immediately");
274 ctx->last_recovery_ts = pkt->pts;
276 ctx->last_recovery_ts = av_gettime_relative();
279 if (fifo->max_recovery_attempts &&
280 ctx->recovery_nr >= fifo->max_recovery_attempts) {
281 av_log(avf, AV_LOG_ERROR,
282 "Maximal number of %d recovery attempts reached.\n",
283 fifo->max_recovery_attempts);
286 ret = AVERROR(EAGAIN);
292 static int fifo_thread_attempt_recovery(FifoThreadContext *ctx, FifoMessage *msg, int err_no)
294 AVFormatContext *avf = ctx->avf;
295 FifoContext *fifo = avf->priv_data;
296 AVPacket *pkt = &msg->pkt;
297 int64_t time_since_recovery;
300 if (!is_recoverable(fifo, err_no)) {
305 if (ctx->header_written) {
306 fifo->write_trailer_ret = fifo_thread_write_trailer(ctx);
307 ctx->header_written = 0;
310 if (!ctx->recovery_nr) {
311 ctx->last_recovery_ts = fifo->recovery_wait_streamtime ?
314 if (fifo->recovery_wait_streamtime) {
315 if (ctx->last_recovery_ts == AV_NOPTS_VALUE) {
316 AVRational tb = avf->streams[pkt->stream_index]->time_base;
317 time_since_recovery = av_rescale_q(pkt->pts - ctx->last_recovery_ts,
320 /* Enforce recovery immediately */
321 time_since_recovery = fifo->recovery_wait_time;
324 time_since_recovery = av_gettime_relative() - ctx->last_recovery_ts;
327 if (time_since_recovery < fifo->recovery_wait_time)
328 return AVERROR(EAGAIN);
333 if (fifo->max_recovery_attempts) {
334 av_log(avf, AV_LOG_VERBOSE, "Recovery attempt #%d/%d\n",
335 ctx->recovery_nr, fifo->max_recovery_attempts);
337 av_log(avf, AV_LOG_VERBOSE, "Recovery attempt #%d\n",
341 if (fifo->restart_with_keyframe && fifo->drop_pkts_on_overflow)
342 ctx->drop_until_keyframe = 1;
344 ret = fifo_thread_dispatch_message(ctx, msg);
346 if (is_recoverable(fifo, ret)) {
347 return fifo_thread_process_recovery_failure(ctx, pkt, ret);
352 av_log(avf, AV_LOG_INFO, "Recovery successful\n");
353 ctx->recovery_nr = 0;
363 static int fifo_thread_recover(FifoThreadContext *ctx, FifoMessage *msg, int err_no)
365 AVFormatContext *avf = ctx->avf;
366 FifoContext *fifo = avf->priv_data;
370 if (!fifo->recovery_wait_streamtime && ctx->recovery_nr > 0) {
371 int64_t time_since_recovery = av_gettime_relative() - ctx->last_recovery_ts;
372 int64_t time_to_wait = FFMAX(0, fifo->recovery_wait_time - time_since_recovery);
374 av_usleep(FFMIN(10000, time_to_wait));
377 ret = fifo_thread_attempt_recovery(ctx, msg, err_no);
378 } while (ret == AVERROR(EAGAIN) && !fifo->drop_pkts_on_overflow);
380 if (ret == AVERROR(EAGAIN) && fifo->drop_pkts_on_overflow) {
381 if (msg->type == FIFO_WRITE_PACKET)
382 av_packet_unref(&msg->pkt);
389 static void *fifo_consumer_thread(void *data)
391 AVFormatContext *avf = data;
392 FifoContext *fifo = avf->priv_data;
393 AVThreadMessageQueue *queue = fifo->queue;
394 FifoMessage msg = {FIFO_WRITE_HEADER, {0}};
397 FifoThreadContext fifo_thread_ctx;
398 memset(&fifo_thread_ctx, 0, sizeof(FifoThreadContext));
399 fifo_thread_ctx.avf = avf;
402 uint8_t just_flushed = 0;
404 if (!fifo_thread_ctx.recovery_nr)
405 ret = fifo_thread_dispatch_message(&fifo_thread_ctx, &msg);
407 if (ret < 0 || fifo_thread_ctx.recovery_nr > 0) {
408 int rec_ret = fifo_thread_recover(&fifo_thread_ctx, &msg, ret);
410 av_thread_message_queue_set_err_send(queue, rec_ret);
415 /* If the queue is full at the moment when fifo_write_packet
416 * attempts to insert new message (packet) to the queue,
417 * it sets the fifo->overflow_flag to 1 and drops packet.
418 * Here in consumer thread, the flag is checked and if it is
419 * set, the queue is flushed and flag cleared. */
420 pthread_mutex_lock(&fifo->overflow_flag_lock);
421 if (fifo->overflow_flag) {
422 av_thread_message_flush(queue);
423 if (fifo->restart_with_keyframe)
424 fifo_thread_ctx.drop_until_keyframe = 1;
425 fifo->overflow_flag = 0;
428 pthread_mutex_unlock(&fifo->overflow_flag_lock);
431 av_log(avf, AV_LOG_INFO, "FIFO queue flushed\n");
433 ret = av_thread_message_queue_recv(queue, &msg, 0);
435 av_thread_message_queue_set_err_send(queue, ret);
440 fifo->write_trailer_ret = fifo_thread_write_trailer(&fifo_thread_ctx);
445 static int fifo_mux_init(AVFormatContext *avf, AVOutputFormat *oformat,
446 const char *filename)
448 FifoContext *fifo = avf->priv_data;
449 AVFormatContext *avf2;
452 ret = avformat_alloc_output_context2(&avf2, oformat, NULL, filename);
458 avf2->interrupt_callback = avf->interrupt_callback;
459 avf2->max_delay = avf->max_delay;
460 ret = av_dict_copy(&avf2->metadata, avf->metadata, 0);
463 avf2->opaque = avf->opaque;
464 avf2->io_close = avf->io_close;
465 avf2->io_open = avf->io_open;
466 avf2->flags = avf->flags;
468 for (i = 0; i < avf->nb_streams; ++i) {
469 AVStream *st = avformat_new_stream(avf2, NULL);
471 return AVERROR(ENOMEM);
473 ret = ff_stream_encode_params_copy(st, avf->streams[i]);
481 static int fifo_init(AVFormatContext *avf)
483 FifoContext *fifo = avf->priv_data;
484 AVOutputFormat *oformat;
487 if (fifo->recovery_wait_streamtime && !fifo->drop_pkts_on_overflow) {
488 av_log(avf, AV_LOG_ERROR, "recovery_wait_streamtime can be turned on"
489 " only when drop_pkts_on_overflow is also turned on\n");
490 return AVERROR(EINVAL);
493 if (fifo->format_options_str) {
494 ret = av_dict_parse_string(&fifo->format_options, fifo->format_options_str,
497 av_log(avf, AV_LOG_ERROR, "Could not parse format options list '%s'\n",
498 fifo->format_options_str);
503 oformat = av_guess_format(fifo->format, avf->filename, NULL);
505 ret = AVERROR_MUXER_NOT_FOUND;
509 ret = fifo_mux_init(avf, oformat, avf->filename);
513 ret = av_thread_message_queue_alloc(&fifo->queue, (unsigned) fifo->queue_size,
514 sizeof(FifoMessage));
518 av_thread_message_queue_set_free_func(fifo->queue, free_message);
520 ret = pthread_mutex_init(&fifo->overflow_flag_lock, NULL);
523 fifo->overflow_flag_lock_initialized = 1;
528 static int fifo_write_header(AVFormatContext *avf)
530 FifoContext * fifo = avf->priv_data;
533 ret = pthread_create(&fifo->writer_thread, NULL, fifo_consumer_thread, avf);
535 av_log(avf, AV_LOG_ERROR, "Failed to start thread: %s\n",
536 av_err2str(AVERROR(ret)));
543 static int fifo_write_packet(AVFormatContext *avf, AVPacket *pkt)
545 FifoContext *fifo = avf->priv_data;
546 FifoMessage msg = {.type = pkt ? FIFO_WRITE_PACKET : FIFO_FLUSH_OUTPUT};
550 av_init_packet(&msg.pkt);
551 ret = av_packet_ref(&msg.pkt,pkt);
556 ret = av_thread_message_queue_send(fifo->queue, &msg,
557 fifo->drop_pkts_on_overflow ?
558 AV_THREAD_MESSAGE_NONBLOCK : 0);
559 if (ret == AVERROR(EAGAIN)) {
560 uint8_t overflow_set = 0;
562 /* Queue is full, set fifo->overflow_flag to 1
563 * to let consumer thread know the queue should
565 pthread_mutex_lock(&fifo->overflow_flag_lock);
566 if (!fifo->overflow_flag)
567 fifo->overflow_flag = overflow_set = 1;
568 pthread_mutex_unlock(&fifo->overflow_flag_lock);
571 av_log(avf, AV_LOG_WARNING, "FIFO queue full\n");
574 } else if (ret < 0) {
581 av_packet_unref(&msg.pkt);
585 static int fifo_write_trailer(AVFormatContext *avf)
587 FifoContext *fifo= avf->priv_data;
590 av_thread_message_queue_set_err_recv(fifo->queue, AVERROR_EOF);
592 ret = pthread_join(fifo->writer_thread, NULL);
594 av_log(avf, AV_LOG_ERROR, "pthread join error: %s\n",
595 av_err2str(AVERROR(ret)));
599 ret = fifo->write_trailer_ret;
603 static void fifo_deinit(AVFormatContext *avf)
605 FifoContext *fifo = avf->priv_data;
607 av_dict_free(&fifo->format_options);
608 avformat_free_context(fifo->avf);
609 av_thread_message_queue_free(&fifo->queue);
610 if (fifo->overflow_flag_lock_initialized)
611 pthread_mutex_destroy(&fifo->overflow_flag_lock);
614 #define OFFSET(x) offsetof(FifoContext, x)
615 static const AVOption options[] = {
616 {"fifo_format", "Target muxer", OFFSET(format),
617 AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
619 {"queue_size", "Size of fifo queue", OFFSET(queue_size),
620 AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_QUEUE_SIZE}, 1, INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
622 {"format_opts", "Options to be passed to underlying muxer", OFFSET(format_options_str),
623 AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
625 {"drop_pkts_on_overflow", "Drop packets on fifo queue overflow not to block encoder", OFFSET(drop_pkts_on_overflow),
626 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
628 {"restart_with_keyframe", "Wait for keyframe when restarting output", OFFSET(restart_with_keyframe),
629 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
631 {"attempt_recovery", "Attempt recovery in case of failure", OFFSET(attempt_recovery),
632 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
634 {"max_recovery_attempts", "Maximal number of recovery attempts", OFFSET(max_recovery_attempts),
635 AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS}, 0, INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
637 {"recovery_wait_time", "Waiting time between recovery attempts", OFFSET(recovery_wait_time),
638 AV_OPT_TYPE_DURATION, {.i64 = FIFO_DEFAULT_RECOVERY_WAIT_TIME_USEC}, 0, INT64_MAX, AV_OPT_FLAG_ENCODING_PARAM},
640 {"recovery_wait_streamtime", "Use stream time instead of real time while waiting for recovery",
641 OFFSET(recovery_wait_streamtime), AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
643 {"recover_any_error", "Attempt recovery regardless of type of the error", OFFSET(recover_any_error),
644 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
649 static const AVClass fifo_muxer_class = {
650 .class_name = "Fifo muxer",
651 .item_name = av_default_item_name,
653 .version = LIBAVUTIL_VERSION_INT,
656 AVOutputFormat ff_fifo_muxer = {
658 .long_name = NULL_IF_CONFIG_SMALL("FIFO queue pseudo-muxer"),
659 .priv_data_size = sizeof(FifoContext),
661 .write_header = fifo_write_header,
662 .write_packet = fifo_write_packet,
663 .write_trailer = fifo_write_trailer,
664 .deinit = fifo_deinit,
665 .priv_class = &fifo_muxer_class,
666 .flags = AVFMT_NOFILE | AVFMT_ALLOW_FLUSH | AVFMT_TS_NEGATIVE,