avformat: Add fifo pseudo-muxer
[ffmpeg.git] / 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 };