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