git.sesse.net Git - ffmpeg/blob - libavformat/fifo.c

   1 /*
   2  * FIFO pseudo-muxer
   3  * Copyright (c) 2016 Jan Sebechlebsky
   4  *
   5  * This file is part of FFmpeg.
   6  *
   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.
  11  *
  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.
  16  *
  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
  20  */
  21
  22 #include "libavutil/opt.h"
  23 #include "libavutil/time.h"
  24 #include "libavutil/thread.h"
  25 #include "libavutil/threadmessage.h"
  26 #include "avformat.h"
  27 #include "internal.h"
  28
  29 #define FIFO_DEFAULT_QUEUE_SIZE              60
  30 #define FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS   0
  31 #define FIFO_DEFAULT_RECOVERY_WAIT_TIME_USEC 5000000 // 5 seconds
  32
  33 typedef struct FifoContext {
  34     const AVClass *class;
  35     AVFormatContext *avf;
  36
  37     char *format;
  38     char *format_options_str;
  39     AVDictionary *format_options;
  40
  41     int queue_size;
  42     AVThreadMessageQueue *queue;
  43
  44     pthread_t writer_thread;
  45
  46     /* Return value of last write_trailer_call */
  47     int write_trailer_ret;
  48
  49     /* Time to wait before next recovery attempt
  50      * This can refer to the time in processed stream,
  51      * or real time. */
  52     int64_t recovery_wait_time;
  53
  54     /* Maximal number of unsuccessful successive recovery attempts */
  55     int max_recovery_attempts;
  56
  57     /* Whether to attempt recovery from failure */
  58     int attempt_recovery;
  59
  60     /* If >0 stream time will be used when waiting
  61      * for the recovery attempt instead of real time */
  62     int recovery_wait_streamtime;
  63
  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;
  67
  68     /* Whether to drop packets in case the queue is full. */
  69     int drop_pkts_on_overflow;
  70
  71     /* Whether to wait for keyframe when recovering
  72      * from failure or queue overflow */
  73     int restart_with_keyframe;
  74
  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;
  79
  80 } FifoContext;
  81
  82 typedef struct FifoThreadContext {
  83     AVFormatContext *avf;
  84
  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;
  89
  90     /* Number of current recovery process
  91      * Value > 0 means we are in recovery process */
  92     int recovery_nr;
  93
  94     /* If > 0 all frames will be dropped until keyframe is received */
  95     uint8_t drop_until_keyframe;
  96
  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;
 101 } FifoThreadContext;
 102
 103 typedef enum FifoMessageType {
 104     FIFO_WRITE_HEADER,
 105     FIFO_WRITE_PACKET,
 106     FIFO_FLUSH_OUTPUT
 107 } FifoMessageType;
 108
 109 typedef struct FifoMessage {
 110     FifoMessageType type;
 111     AVPacket pkt;
 112 } FifoMessage;
 113
 114 static int fifo_thread_write_header(FifoThreadContext *ctx)
 115 {
 116     AVFormatContext *avf = ctx->avf;
 117     FifoContext *fifo = avf->priv_data;
 118     AVFormatContext *avf2 = fifo->avf;
 119     AVDictionary *format_options = NULL;
 120     int ret, i;
 121
 122     ret = av_dict_copy(&format_options, fifo->format_options, 0);
 123     if (ret < 0)
 124         return ret;
 125
 126     ret = ff_format_output_open(avf2, avf->filename, &format_options);
 127     if (ret < 0) {
 128         av_log(avf, AV_LOG_ERROR, "Error opening %s: %s\n", avf->filename,
 129                av_err2str(ret));
 130         goto end;
 131     }
 132
 133     for (i = 0;i < avf2->nb_streams; i++)
 134         avf2->streams[i]->cur_dts = 0;
 135
 136     ret = avformat_write_header(avf2, &format_options);
 137     if (!ret)
 138         ctx->header_written = 1;
 139
 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);
 146     }
 147
 148 end:
 149     av_dict_free(&format_options);
 150     return ret;
 151 }
 152
 153 static int fifo_thread_flush_output(FifoThreadContext *ctx)
 154 {
 155     AVFormatContext *avf = ctx->avf;
 156     FifoContext *fifo = avf->priv_data;
 157     AVFormatContext *avf2 = fifo->avf;
 158
 159     return av_write_frame(avf2, NULL);
 160 }
 161
 162 static int fifo_thread_write_packet(FifoThreadContext *ctx, AVPacket *pkt)
 163 {
 164     AVFormatContext *avf = ctx->avf;
 165     FifoContext *fifo = avf->priv_data;
 166     AVFormatContext *avf2 = fifo->avf;
 167     AVRational src_tb, dst_tb;
 168     int ret, s_idx;
 169
 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");
 174         } else {
 175             av_log(avf, AV_LOG_VERBOSE, "Dropping non-keyframe packet\n");
 176             av_packet_unref(pkt);
 177             return 0;
 178         }
 179     }
 180
 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);
 185
 186     ret = av_write_frame(avf2, pkt);
 187     if (ret >= 0)
 188         av_packet_unref(pkt);
 189     return ret;
 190 }
 191
 192 static int fifo_thread_write_trailer(FifoThreadContext *ctx)
 193 {
 194     AVFormatContext *avf = ctx->avf;
 195     FifoContext *fifo = avf->priv_data;
 196     AVFormatContext *avf2 = fifo->avf;
 197     int ret;
 198
 199     if (!ctx->header_written)
 200         return 0;
 201
 202     ret = av_write_trailer(avf2);
 203     ff_format_io_close(avf2, &avf2->pb);
 204
 205     return ret;
 206 }
 207
 208 static int fifo_thread_dispatch_message(FifoThreadContext *ctx, FifoMessage *msg)
 209 {
 210     int ret;
 211
 212     if (!ctx->header_written) {
 213         ret = fifo_thread_write_header(ctx);
 214         if (ret < 0)
 215             return ret;
 216     }
 217
 218     switch(msg->type) {
 219     case FIFO_WRITE_HEADER:
 220         return ret;
 221     case FIFO_WRITE_PACKET:
 222         return fifo_thread_write_packet(ctx, &msg->pkt);
 223     case FIFO_FLUSH_OUTPUT:
 224         return fifo_thread_flush_output(ctx);
 225     }
 226
 227     return AVERROR(EINVAL);
 228 }
 229
 230 static int is_recoverable(const FifoContext *fifo, int err_no) {
 231     if (!fifo->attempt_recovery)
 232         return 0;
 233
 234     if (fifo->recover_any_error)
 235         return err_no != AVERROR_EXIT;
 236
 237     switch (err_no) {
 238     case AVERROR(EINVAL):
 239     case AVERROR(ENOSYS):
 240     case AVERROR_EOF:
 241     case AVERROR_EXIT:
 242     case AVERROR_PATCHWELCOME:
 243         return 0;
 244     default:
 245         return 1;
 246     }
 247 }
 248
 249 static void free_message(void *msg)
 250 {
 251     FifoMessage *fifo_msg = msg;
 252
 253     if (fifo_msg->type == FIFO_WRITE_PACKET)
 254         av_packet_unref(&fifo_msg->pkt);
 255 }
 256
 257 static int fifo_thread_process_recovery_failure(FifoThreadContext *ctx, AVPacket *pkt,
 258                                                 int err_no)
 259 {
 260     AVFormatContext *avf = ctx->avf;
 261     FifoContext *fifo = avf->priv_data;
 262     int ret;
 263
 264     av_log(avf, AV_LOG_INFO, "Recovery failed: %s\n",
 265            av_err2str(err_no));
 266
 267     if (fifo->recovery_wait_streamtime) {
 268         if (pkt->pts == AV_NOPTS_VALUE)
 269             av_log(avf, AV_LOG_WARNING, "Packet does not contain presentation"
 270                    " timestamp, recovery will be attempted immediately");
 271         ctx->last_recovery_ts = pkt->pts;
 272     } else {
 273         ctx->last_recovery_ts = av_gettime_relative();
 274     }
 275
 276     if (fifo->max_recovery_attempts &&
 277         ctx->recovery_nr >= fifo->max_recovery_attempts) {
 278         av_log(avf, AV_LOG_ERROR,
 279                "Maximal number of %d recovery attempts reached.\n",
 280                fifo->max_recovery_attempts);
 281         ret = err_no;
 282     } else {
 283         ret = AVERROR(EAGAIN);
 284     }
 285
 286     return ret;
 287 }
 288
 289 static int fifo_thread_attempt_recovery(FifoThreadContext *ctx, FifoMessage *msg, int err_no)
 290 {
 291     AVFormatContext *avf = ctx->avf;
 292     FifoContext *fifo = avf->priv_data;
 293     AVPacket *pkt = &msg->pkt;
 294     int64_t time_since_recovery;
 295     int ret;
 296
 297     if (!is_recoverable(fifo, err_no)) {
 298         ret = err_no;
 299         goto fail;
 300     }
 301
 302     if (ctx->header_written) {
 303         fifo->write_trailer_ret = fifo_thread_write_trailer(ctx);
 304         ctx->header_written = 0;
 305     }
 306
 307     if (!ctx->recovery_nr) {
 308         ctx->last_recovery_ts = fifo->recovery_wait_streamtime ?
 309                                 AV_NOPTS_VALUE : 0;
 310     } else {
 311         if (fifo->recovery_wait_streamtime) {
 312             if (ctx->last_recovery_ts == AV_NOPTS_VALUE) {
 313                 AVRational tb = avf->streams[pkt->stream_index]->time_base;
 314                 time_since_recovery = av_rescale_q(pkt->pts - ctx->last_recovery_ts,
 315                                                    tb, AV_TIME_BASE_Q);
 316             } else {
 317                 /* Enforce recovery immediately */
 318                 time_since_recovery = fifo->recovery_wait_time;
 319             }
 320         } else {
 321             time_since_recovery = av_gettime_relative() - ctx->last_recovery_ts;
 322         }
 323
 324         if (time_since_recovery < fifo->recovery_wait_time)
 325             return AVERROR(EAGAIN);
 326     }
 327
 328     ctx->recovery_nr++;
 329
 330     if (fifo->max_recovery_attempts) {
 331         av_log(avf, AV_LOG_VERBOSE, "Recovery attempt #%d/%d\n",
 332                ctx->recovery_nr, fifo->max_recovery_attempts);
 333     } else {
 334         av_log(avf, AV_LOG_VERBOSE, "Recovery attempt #%d\n",
 335                ctx->recovery_nr);
 336     }
 337
 338     if (fifo->restart_with_keyframe && fifo->drop_pkts_on_overflow)
 339         ctx->drop_until_keyframe = 1;
 340
 341     ret = fifo_thread_dispatch_message(ctx, msg);
 342     if (ret < 0) {
 343         if (is_recoverable(fifo, ret)) {
 344             return fifo_thread_process_recovery_failure(ctx, pkt, ret);
 345         } else {
 346             goto fail;
 347         }
 348     } else {
 349         av_log(avf, AV_LOG_INFO, "Recovery successful\n");
 350         ctx->recovery_nr = 0;
 351     }
 352
 353     return 0;
 354
 355 fail:
 356     free_message(msg);
 357     return ret;
 358 }
 359
 360 static int fifo_thread_recover(FifoThreadContext *ctx, FifoMessage *msg, int err_no)
 361 {
 362     AVFormatContext *avf = ctx->avf;
 363     FifoContext *fifo = avf->priv_data;
 364     int ret;
 365
 366     do {
 367         if (!fifo->recovery_wait_streamtime && ctx->recovery_nr > 0) {
 368             int64_t time_since_recovery = av_gettime_relative() - ctx->last_recovery_ts;
 369             int64_t time_to_wait = FFMAX(0, fifo->recovery_wait_time - time_since_recovery);
 370             if (time_to_wait)
 371                 av_usleep(FFMIN(10000, time_to_wait));
 372         }
 373
 374         ret = fifo_thread_attempt_recovery(ctx, msg, err_no);
 375     } while (ret == AVERROR(EAGAIN) && !fifo->drop_pkts_on_overflow);
 376
 377     if (ret == AVERROR(EAGAIN) && fifo->drop_pkts_on_overflow) {
 378         if (msg->type == FIFO_WRITE_PACKET)
 379             av_packet_unref(&msg->pkt);
 380         ret = 0;
 381     }
 382
 383     return ret;
 384 }
 385
 386 static void *fifo_consumer_thread(void *data)
 387 {
 388     AVFormatContext *avf = data;
 389     FifoContext *fifo = avf->priv_data;
 390     AVThreadMessageQueue *queue = fifo->queue;
 391     FifoMessage msg = {FIFO_WRITE_HEADER, {0}};
 392     int ret;
 393
 394     FifoThreadContext fifo_thread_ctx;
 395     memset(&fifo_thread_ctx, 0, sizeof(FifoThreadContext));
 396     fifo_thread_ctx.avf = avf;
 397
 398     while (1) {
 399         uint8_t just_flushed = 0;
 400
 401         if (!fifo_thread_ctx.recovery_nr)
 402             ret = fifo_thread_dispatch_message(&fifo_thread_ctx, &msg);
 403
 404         if (ret < 0 || fifo_thread_ctx.recovery_nr > 0) {
 405             int rec_ret = fifo_thread_recover(&fifo_thread_ctx, &msg, ret);
 406             if (rec_ret < 0) {
 407                 av_thread_message_queue_set_err_send(queue, rec_ret);
 408                 break;
 409             }
 410         }
 411
 412         /* If the queue is full at the moment when fifo_write_packet
 413          * attempts to insert new message (packet) to the queue,
 414          * it sets the fifo->overflow_flag to 1 and drops packet.
 415          * Here in consumer thread, the flag is checked and if it is
 416          * set, the queue is flushed and flag cleared. */
 417         pthread_mutex_lock(&fifo->overflow_flag_lock);
 418         if (fifo->overflow_flag) {
 419             av_thread_message_flush(queue);
 420             if (fifo->restart_with_keyframe)
 421                 fifo_thread_ctx.drop_until_keyframe = 1;
 422             fifo->overflow_flag = 0;
 423             just_flushed = 1;
 424         }
 425         pthread_mutex_unlock(&fifo->overflow_flag_lock);
 426
 427         if (just_flushed)
 428             av_log(avf, AV_LOG_INFO, "FIFO queue flushed\n");
 429
 430         ret = av_thread_message_queue_recv(queue, &msg, 0);
 431         if (ret < 0) {
 432             av_thread_message_queue_set_err_send(queue, ret);
 433             break;
 434         }
 435     }
 436
 437     fifo->write_trailer_ret = fifo_thread_write_trailer(&fifo_thread_ctx);
 438
 439     return NULL;
 440 }
 441
 442 static int fifo_mux_init(AVFormatContext *avf, AVOutputFormat *oformat)
 443 {
 444     FifoContext *fifo = avf->priv_data;
 445     AVFormatContext *avf2;
 446     int ret = 0, i;
 447
 448     ret = avformat_alloc_output_context2(&avf2, oformat, NULL, NULL);
 449     if (ret < 0)
 450         return ret;
 451
 452     fifo->avf = avf2;
 453
 454     avf2->interrupt_callback = avf->interrupt_callback;
 455     avf2->max_delay = avf->max_delay;
 456     ret = av_dict_copy(&avf2->metadata, avf->metadata, 0);
 457     if (ret < 0)
 458         return ret;
 459     avf2->opaque = avf->opaque;
 460     avf2->io_close = avf->io_close;
 461     avf2->io_open = avf->io_open;
 462     avf2->flags = avf->flags;
 463
 464     for (i = 0; i < avf->nb_streams; ++i) {
 465         AVStream *st = avformat_new_stream(avf2, NULL);
 466         if (!st)
 467             return AVERROR(ENOMEM);
 468
 469         ret = ff_stream_encode_params_copy(st, avf->streams[i]);
 470         if (ret < 0)
 471             return ret;
 472     }
 473
 474     return 0;
 475 }
 476
 477 static int fifo_init(AVFormatContext *avf)
 478 {
 479     FifoContext *fifo = avf->priv_data;
 480     AVOutputFormat *oformat;
 481     int ret = 0;
 482
 483     if (fifo->recovery_wait_streamtime && !fifo->drop_pkts_on_overflow) {
 484         av_log(avf, AV_LOG_ERROR, "recovery_wait_streamtime can be turned on"
 485                " only when drop_pkts_on_overflow is also turned on\n");
 486         return AVERROR(EINVAL);
 487     }
 488
 489     if (fifo->format_options_str) {
 490         ret = av_dict_parse_string(&fifo->format_options, fifo->format_options_str,
 491                                    "=", ":", 0);
 492         if (ret < 0) {
 493             av_log(avf, AV_LOG_ERROR, "Could not parse format options list '%s'\n",
 494                    fifo->format_options_str);
 495             return ret;
 496         }
 497     }
 498
 499     oformat = av_guess_format(fifo->format, avf->filename, NULL);
 500     if (!oformat) {
 501         ret = AVERROR_MUXER_NOT_FOUND;
 502         return ret;
 503     }
 504
 505     ret = fifo_mux_init(avf, oformat);
 506     if (ret < 0)
 507         return ret;
 508
 509     ret = av_thread_message_queue_alloc(&fifo->queue, (unsigned) fifo->queue_size,
 510                                         sizeof(FifoMessage));
 511     if (ret < 0)
 512         return ret;
 513
 514     av_thread_message_queue_set_free_func(fifo->queue, free_message);
 515
 516     ret = pthread_mutex_init(&fifo->overflow_flag_lock, NULL);
 517     if (ret < 0)
 518         return AVERROR(ret);
 519     fifo->overflow_flag_lock_initialized = 1;
 520
 521     return 0;
 522 }
 523
 524 static int fifo_write_header(AVFormatContext *avf)
 525 {
 526     FifoContext * fifo = avf->priv_data;
 527     int ret;
 528
 529     ret = pthread_create(&fifo->writer_thread, NULL, fifo_consumer_thread, avf);
 530     if (ret) {
 531         av_log(avf, AV_LOG_ERROR, "Failed to start thread: %s\n",
 532                av_err2str(AVERROR(ret)));
 533         ret = AVERROR(ret);
 534     }
 535
 536     return ret;
 537 }
 538
 539 static int fifo_write_packet(AVFormatContext *avf, AVPacket *pkt)
 540 {
 541     FifoContext *fifo = avf->priv_data;
 542     FifoMessage msg = {.type = pkt ? FIFO_WRITE_PACKET : FIFO_FLUSH_OUTPUT};
 543     int ret;
 544
 545     if (pkt) {
 546         av_init_packet(&msg.pkt);
 547         ret = av_packet_ref(&msg.pkt,pkt);
 548         if (ret < 0)
 549             return ret;
 550     }
 551
 552     ret = av_thread_message_queue_send(fifo->queue, &msg,
 553                                        fifo->drop_pkts_on_overflow ?
 554                                        AV_THREAD_MESSAGE_NONBLOCK : 0);
 555     if (ret == AVERROR(EAGAIN)) {
 556         uint8_t overflow_set = 0;
 557
 558         /* Queue is full, set fifo->overflow_flag to 1
 559          * to let consumer thread know the queue should
 560          * be flushed. */
 561         pthread_mutex_lock(&fifo->overflow_flag_lock);
 562         if (!fifo->overflow_flag)
 563             fifo->overflow_flag = overflow_set = 1;
 564         pthread_mutex_unlock(&fifo->overflow_flag_lock);
 565
 566         if (overflow_set)
 567             av_log(avf, AV_LOG_WARNING, "FIFO queue full\n");
 568         ret = 0;
 569         goto fail;
 570     } else if (ret < 0) {
 571         goto fail;
 572     }
 573
 574     return ret;
 575 fail:
 576     if (pkt)
 577         av_packet_unref(&msg.pkt);
 578     return ret;
 579 }
 580
 581 static int fifo_write_trailer(AVFormatContext *avf)
 582 {
 583     FifoContext *fifo= avf->priv_data;
 584     int ret;
 585
 586     av_thread_message_queue_set_err_recv(fifo->queue, AVERROR_EOF);
 587
 588     ret = pthread_join(fifo->writer_thread, NULL);
 589     if (ret < 0) {
 590         av_log(avf, AV_LOG_ERROR, "pthread join error: %s\n",
 591                av_err2str(AVERROR(ret)));
 592         return AVERROR(ret);
 593     }
 594
 595     ret = fifo->write_trailer_ret;
 596     return ret;
 597 }
 598
 599 static void fifo_deinit(AVFormatContext *avf)
 600 {
 601     FifoContext *fifo = avf->priv_data;
 602
 603     av_dict_free(&fifo->format_options);
 604     avformat_free_context(fifo->avf);
 605     av_thread_message_queue_free(&fifo->queue);
 606     if (fifo->overflow_flag_lock_initialized)
 607         pthread_mutex_destroy(&fifo->overflow_flag_lock);
 608 }
 609
 610 #define OFFSET(x) offsetof(FifoContext, x)
 611 static const AVOption options[] = {
 612         {"fifo_format", "Target muxer", OFFSET(format),
 613          AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
 614
 615         {"queue_size", "Size of fifo queue", OFFSET(queue_size),
 616          AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_QUEUE_SIZE}, 1, INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
 617
 618         {"format_opts", "Options to be passed to underlying muxer", OFFSET(format_options_str),
 619          AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
 620
 621         {"drop_pkts_on_overflow", "Drop packets on fifo queue overflow not to block encoder", OFFSET(drop_pkts_on_overflow),
 622          AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
 623
 624         {"restart_with_keyframe", "Wait for keyframe when restarting output", OFFSET(restart_with_keyframe),
 625          AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
 626
 627         {"attempt_recovery", "Attempt recovery in case of failure", OFFSET(attempt_recovery),
 628         AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
 629
 630         {"max_recovery_attempts", "Maximal number of recovery attempts", OFFSET(max_recovery_attempts),
 631          AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS}, 0, INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
 632
 633         {"recovery_wait_time", "Waiting time between recovery attempts", OFFSET(recovery_wait_time),
 634          AV_OPT_TYPE_DURATION, {.i64 = FIFO_DEFAULT_RECOVERY_WAIT_TIME_USEC}, 0, INT64_MAX, AV_OPT_FLAG_ENCODING_PARAM},
 635
 636         {"recovery_wait_streamtime", "Use stream time instead of real time while waiting for recovery",
 637          OFFSET(recovery_wait_streamtime), AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
 638
 639         {"recover_any_error", "Attempt recovery regardless of type of the error", OFFSET(recover_any_error),
 640          AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
 641
 642         {NULL},
 643 };
 644
 645 static const AVClass fifo_muxer_class = {
 646     .class_name = "Fifo muxer",
 647     .item_name  = av_default_item_name,
 648     .option     = options,
 649     .version    = LIBAVUTIL_VERSION_INT,
 650 };
 651
 652 AVOutputFormat ff_fifo_muxer = {
 653     .name           = "fifo",
 654     .long_name      = NULL_IF_CONFIG_SMALL("FIFO queue pseudo-muxer"),
 655     .priv_data_size = sizeof(FifoContext),
 656     .init           = fifo_init,
 657     .write_header   = fifo_write_header,
 658     .write_packet   = fifo_write_packet,
 659     .write_trailer  = fifo_write_trailer,
 660     .deinit         = fifo_deinit,
 661     .priv_class     = &fifo_muxer_class,
 662     .flags          = AVFMT_NOFILE | AVFMT_ALLOW_FLUSH | AVFMT_TS_NEGATIVE,
 663 };