FFmpeg coverage

Directory: ../../../ffmpeg/
File: src/libavformat/fifo.c
Date: 2026-09-22 08:26:45
Exec Total Coverage
Lines: 241 342 70.5%
Functions: 17 18 94.4%
Branches: 92 184 50.0%
Line Branch Exec Source
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 <stdatomic.h>
23
24 #include "libavutil/avassert.h"
25 #include "libavutil/opt.h"
26 #include "libavutil/time.h"
27 #include "libavutil/thread.h"
28 #include "libavutil/threadmessage.h"
29 #include "avformat.h"
30 #include "internal.h"
31 #include "mux.h"
32
33 #define FIFO_DEFAULT_QUEUE_SIZE 60
34 #define FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS 0
35 #define FIFO_DEFAULT_RECOVERY_WAIT_TIME_USEC 5000000 // 5 seconds
36
37 typedef struct FifoContext {
38 const AVClass *class;
39 AVFormatContext *avf;
40
41 char *format;
42 AVDictionary *format_options;
43
44 int queue_size;
45 AVThreadMessageQueue *queue;
46
47 pthread_t writer_thread;
48
49 /* Return value of last write_trailer_call */
50 int write_trailer_ret;
51
52 /* Time to wait before next recovery attempt
53 * This can refer to the time in processed stream,
54 * or real time. */
55 int64_t recovery_wait_time;
56
57 /* Maximal number of unsuccessful successive recovery attempts */
58 int max_recovery_attempts;
59
60 /* Whether to attempt recovery from failure */
61 int attempt_recovery;
62
63 /* If >0 stream time will be used when waiting
64 * for the recovery attempt instead of real time */
65 int recovery_wait_streamtime;
66
67 /* If >0 recovery will be attempted regardless of error code
68 * (except AVERROR_EXIT, so exit request is never ignored) */
69 int recover_any_error;
70
71 /* Whether to drop packets in case the queue is full. */
72 int drop_pkts_on_overflow;
73
74 /* Whether to wait for keyframe when recovering
75 * from failure or queue overflow */
76 int restart_with_keyframe;
77
78 pthread_mutex_t overflow_flag_lock;
79 int overflow_flag_lock_initialized;
80 /* Value > 0 signals queue overflow */
81 volatile uint8_t overflow_flag;
82
83 atomic_int_least64_t queue_duration;
84 int64_t last_sent_dts;
85 int64_t timeshift;
86 } FifoContext;
87
88 typedef struct FifoThreadContext {
89 AVFormatContext *avf;
90
91 /* Timestamp of last failure.
92 * This is either pts in case stream time is used,
93 * or microseconds as returned by av_gettime_relative() */
94 int64_t last_recovery_ts;
95
96 /* Number of current recovery process
97 * Value > 0 means we are in recovery process */
98 int recovery_nr;
99
100 /* If > 0 all frames will be dropped until keyframe is received */
101 uint8_t drop_until_keyframe;
102
103 /* Value > 0 means that the previous write_header call was successful
104 * so finalization by calling write_trailer and ff_io_close must be done
105 * before exiting / reinitialization of underlying muxer */
106 uint8_t header_written;
107
108 int64_t last_received_dts;
109
110 /* If > 0 at least one of the streams is a video stream */
111 uint8_t has_video_stream;
112 } FifoThreadContext;
113
114 typedef enum FifoMessageType {
115 FIFO_NOOP,
116 FIFO_WRITE_HEADER,
117 FIFO_WRITE_PACKET,
118 FIFO_FLUSH_OUTPUT
119 } FifoMessageType;
120
121 typedef struct FifoMessage {
122 FifoMessageType type;
123 AVPacket pkt;
124 } FifoMessage;
125
126 51 static int fifo_thread_write_header(FifoThreadContext *ctx)
127 {
128 51 AVFormatContext *avf = ctx->avf;
129 51 FifoContext *fifo = avf->priv_data;
130 51 AVFormatContext *avf2 = fifo->avf;
131 51 AVDictionary *format_options = NULL;
132 int ret, i;
133
134 51 ret = av_dict_copy(&format_options, fifo->format_options, 0);
135
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 51 times.
 51 if (ret < 0)
136 ✗ goto end;
137
138 51 ret = ff_format_output_open(avf2, avf->url, &format_options);
139
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 51 times.
 51 if (ret < 0) {
140 ✗ av_log(avf, AV_LOG_ERROR, "Error opening %s: %s\n", avf->url,
141 ✗ av_err2str(ret));
142 ✗ goto end;
143 }
144
145
2/2
✓ Branch 0 taken 52 times.
✓ Branch 1 taken 51 times.
 103 for (i = 0;i < avf2->nb_streams; i++)
146 52 ffstream(avf2->streams[i])->cur_dts = 0;
147
148 51 ret = avformat_write_header(avf2, &format_options);
149
1/2
✓ Branch 0 taken 51 times.
✗ Branch 1 not taken.
 51 if (!ret) {
150 51 ctx->header_written = 1;
151 } else {
152 ✗ ff_format_io_close(avf2, &avf2->pb);
153 }
154
155 // Check for options unrecognized by underlying muxer
156
1/2
✓ Branch 0 taken 51 times.
✗ Branch 1 not taken.
 51 if (format_options) {
157 ✗ const AVDictionaryEntry *entry = NULL;
158 ✗ while ((entry = av_dict_iterate(format_options, entry)))
159 ✗ av_log(avf2, AV_LOG_ERROR, "Unknown option '%s'\n", entry->key);
160 ✗ ret = AVERROR(EINVAL);
161 }
162
163 51 end:
164 51 av_dict_free(&format_options);
165 51 return ret;
166 }
167
168 3 static int fifo_thread_flush_output(FifoThreadContext *ctx)
169 {
170 3 AVFormatContext *avf = ctx->avf;
171 3 FifoContext *fifo = avf->priv_data;
172 3 AVFormatContext *avf2 = fifo->avf;
173
174 3 return av_write_frame(avf2, NULL);
175 }
176
177 ✗ static int64_t next_duration(AVFormatContext *avf, AVPacket *pkt, int64_t *last_dts)
178 {
179 ✗ AVStream *st = avf->streams[pkt->stream_index];
180 ✗ int64_t dts = av_rescale_q(pkt->dts, st->time_base, AV_TIME_BASE_Q);
181 ✗ int64_t duration = (*last_dts == AV_NOPTS_VALUE ? 0 : dts - *last_dts);
182 ✗ *last_dts = dts;
183 ✗ return duration;
184 }
185
186 145 static int fifo_thread_write_packet(FifoThreadContext *ctx, AVPacket *pkt)
187 {
188 145 AVFormatContext *avf = ctx->avf;
189 145 FifoContext *fifo = avf->priv_data;
190 145 AVFormatContext *avf2 = fifo->avf;
191 AVRational src_tb, dst_tb;
192 int ret, s_idx;
193 int64_t orig_pts, orig_dts, orig_duration;
194 145 enum AVMediaType stream_codec_type = avf->streams[pkt->stream_index]->codecpar->codec_type;
195
196
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 145 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 145 if (fifo->timeshift && pkt->dts != AV_NOPTS_VALUE)
197 ✗ atomic_fetch_sub_explicit(&fifo->queue_duration, next_duration(avf, pkt, &ctx->last_received_dts), memory_order_relaxed);
198
199
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 145 times.
 145 if (ctx->drop_until_keyframe) {
200 ✗ if (pkt->flags & AV_PKT_FLAG_KEY) {
201 ✗ if (!ctx->has_video_stream) {
202 ✗ ctx->drop_until_keyframe = 0;
203 ✗ av_log(avf, AV_LOG_VERBOSE, "Keyframe received, recovering...\n");
204 } else {
205 ✗ if (stream_codec_type == AVMEDIA_TYPE_VIDEO) {
206 ✗ ctx->drop_until_keyframe = 0;
207 ✗ av_log(avf, AV_LOG_VERBOSE, "Video keyframe received, recovering...\n");
208 } else {
209 ✗ av_log(avf, AV_LOG_VERBOSE, "Dropping non-video keyframe\n");
210 ✗ av_packet_unref(pkt);
211 ✗ return 0;
212 }
213 }
214 } else {
215 ✗ av_log(avf, AV_LOG_VERBOSE, "Dropping non-keyframe packet\n");
216 ✗ av_packet_unref(pkt);
217 ✗ return 0;
218 }
219 }
220
221 145 orig_pts = pkt->pts;
222 145 orig_dts = pkt->dts;
223 145 orig_duration = pkt->duration;
224 145 s_idx = pkt->stream_index;
225 145 src_tb = avf->streams[s_idx]->time_base;
226 145 dst_tb = avf2->streams[s_idx]->time_base;
227 145 av_packet_rescale_ts(pkt, src_tb, dst_tb);
228
229 145 ret = av_write_frame(avf2, pkt);
230
2/2
✓ Branch 0 taken 100 times.
✓ Branch 1 taken 45 times.
 145 if (ret >= 0) {
231 100 av_packet_unref(pkt);
232 } else {
233 // avoid scaling twice
234 45 pkt->pts = orig_pts;
235 45 pkt->dts = orig_dts;
236 45 pkt->duration = orig_duration;
237 }
238 145 return ret;
239 }
240
241 51 static int fifo_thread_write_trailer(FifoThreadContext *ctx)
242 {
243 51 AVFormatContext *avf = ctx->avf;
244 51 FifoContext *fifo = avf->priv_data;
245 51 AVFormatContext *avf2 = fifo->avf;
246 int ret;
247
248
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 51 times.
 51 if (!ctx->header_written)
249 ✗ return 0;
250
251 51 ret = av_write_trailer(avf2);
252 51 ff_format_io_close(avf2, &avf2->pb);
253
254 51 return ret;
255 }
256
257 154 static int fifo_thread_dispatch_message(FifoThreadContext *ctx, FifoMessage *msg)
258 {
259 154 int ret = AVERROR(EINVAL);
260
261
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 154 times.
 154 if (msg->type == FIFO_NOOP)
262 ✗ return 0;
263
264
2/2
✓ Branch 0 taken 51 times.
✓ Branch 1 taken 103 times.
 154 if (!ctx->header_written) {
265 51 ret = fifo_thread_write_header(ctx);
266
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 51 times.
 51 if (ret < 0)
267 ✗ return ret;
268 }
269
270
3/4
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 145 times.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
 154 switch(msg->type) {
271 6 case FIFO_WRITE_HEADER:
272
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 av_assert0(ret >= 0);
273 6 return ret;
274 145 case FIFO_WRITE_PACKET:
275 145 return fifo_thread_write_packet(ctx, &msg->pkt);
276 3 case FIFO_FLUSH_OUTPUT:
277 3 return fifo_thread_flush_output(ctx);
278 }
279
280 ✗ av_assert0(0);
281 return AVERROR(EINVAL);
282 }
283
284 75 static int is_recoverable(const FifoContext *fifo, int err_no) {
285
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 75 times.
 75 if (!fifo->attempt_recovery)
286 ✗ return 0;
287
288
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 75 times.
 75 if (fifo->recover_any_error)
289 ✗ return err_no != AVERROR_EXIT;
290
291
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 75 times.
 75 switch (err_no) {
292 ✗ case AVERROR(EINVAL):
293 case AVERROR(ENOSYS):
294 case AVERROR_EOF:
295 case AVERROR_EXIT:
296 case AVERROR_PATCHWELCOME:
297 ✗ return 0;
298 75 default:
299 75 return 1;
300 }
301 }
302
303 3 static void free_message(void *msg)
304 {
305 3 FifoMessage *fifo_msg = msg;
306
307
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 if (fifo_msg->type == FIFO_WRITE_PACKET)
308 3 av_packet_unref(&fifo_msg->pkt);
309 3 }
310
311 30 static int fifo_thread_process_recovery_failure(FifoThreadContext *ctx, AVPacket *pkt,
312 int err_no)
313 {
314 30 AVFormatContext *avf = ctx->avf;
315 30 FifoContext *fifo = avf->priv_data;
316 int ret;
317
318 30 av_log(avf, AV_LOG_INFO, "Recovery failed: %s\n",
319 30 av_err2str(err_no));
320
321
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 30 times.
 30 if (fifo->recovery_wait_streamtime) {
322 ✗ if (pkt->pts == AV_NOPTS_VALUE)
323 ✗ av_log(avf, AV_LOG_WARNING, "Packet does not contain presentation"
324 " timestamp, recovery will be attempted immediately");
325 ✗ ctx->last_recovery_ts = pkt->pts;
326 } else {
327 30 ctx->last_recovery_ts = av_gettime_relative();
328 }
329
330
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 30 times.
 30 if (fifo->max_recovery_attempts &&
331 ✗ ctx->recovery_nr >= fifo->max_recovery_attempts) {
332 ✗ av_log(avf, AV_LOG_ERROR,
333 "Maximal number of %d recovery attempts reached.\n",
334 fifo->max_recovery_attempts);
335 ✗ ret = err_no;
336 } else {
337 30 ret = AVERROR(EAGAIN);
338 }
339
340 30 return ret;
341 }
342
343 45 static int fifo_thread_attempt_recovery(FifoThreadContext *ctx, FifoMessage *msg, int err_no)
344 {
345 45 AVFormatContext *avf = ctx->avf;
346 45 FifoContext *fifo = avf->priv_data;
347 45 AVPacket *pkt = &msg->pkt;
348 int64_t time_since_recovery;
349 int ret;
350
351
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 45 times.
 45 if (!is_recoverable(fifo, err_no)) {
352 ✗ ret = err_no;
353 ✗ goto fail;
354 }
355
356
1/2
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
 45 if (ctx->header_written) {
357 45 fifo->write_trailer_ret = fifo_thread_write_trailer(ctx);
358 45 ctx->header_written = 0;
359 }
360
361
2/2
✓ Branch 0 taken 15 times.
✓ Branch 1 taken 30 times.
 45 if (!ctx->recovery_nr) {
362 15 ctx->last_recovery_ts = fifo->recovery_wait_streamtime ?
363
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 15 times.
 15 AV_NOPTS_VALUE : 0;
364 } else {
365
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 30 times.
 30 if (fifo->recovery_wait_streamtime) {
366 ✗ if (ctx->last_recovery_ts == AV_NOPTS_VALUE) {
367 ✗ AVRational tb = avf->streams[pkt->stream_index]->time_base;
368 ✗ time_since_recovery = av_rescale_q(pkt->pts - ctx->last_recovery_ts,
369 ✗ tb, AV_TIME_BASE_Q);
370 } else {
371 /* Enforce recovery immediately */
372 ✗ time_since_recovery = fifo->recovery_wait_time;
373 }
374 } else {
375 30 time_since_recovery = av_gettime_relative() - ctx->last_recovery_ts;
376 }
377
378
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 30 times.
 30 if (time_since_recovery < fifo->recovery_wait_time)
379 ✗ return AVERROR(EAGAIN);
380 }
381
382 45 ctx->recovery_nr++;
383
384
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 45 times.
 45 if (fifo->max_recovery_attempts) {
385 ✗ av_log(avf, AV_LOG_VERBOSE, "Recovery attempt #%d/%d\n",
386 ctx->recovery_nr, fifo->max_recovery_attempts);
387 } else {
388 45 av_log(avf, AV_LOG_VERBOSE, "Recovery attempt #%d\n",
389 ctx->recovery_nr);
390 }
391
392
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 45 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 45 if (fifo->restart_with_keyframe && fifo->drop_pkts_on_overflow)
393 ✗ ctx->drop_until_keyframe = 1;
394
395 45 ret = fifo_thread_dispatch_message(ctx, msg);
396
2/2
✓ Branch 0 taken 30 times.
✓ Branch 1 taken 15 times.
 45 if (ret < 0) {
397
1/2
✓ Branch 1 taken 30 times.
✗ Branch 2 not taken.
 30 if (is_recoverable(fifo, ret)) {
398 30 return fifo_thread_process_recovery_failure(ctx, pkt, ret);
399 } else {
400 ✗ goto fail;
401 }
402 } else {
403 15 av_log(avf, AV_LOG_INFO, "Recovery successful\n");
404 15 ctx->recovery_nr = 0;
405 }
406
407 15 return 0;
408
409 ✗ fail:
410 ✗ free_message(msg);
411 ✗ return ret;
412 }
413
414 15 static int fifo_thread_recover(FifoThreadContext *ctx, FifoMessage *msg, int err_no)
415 {
416 15 AVFormatContext *avf = ctx->avf;
417 15 FifoContext *fifo = avf->priv_data;
418 int ret;
419
420 do {
421
3/4
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 30 times.
✓ Branch 3 taken 15 times.
 45 if (!fifo->recovery_wait_streamtime && ctx->recovery_nr > 0) {
422 30 int64_t time_since_recovery = av_gettime_relative() - ctx->last_recovery_ts;
423 30 int64_t time_to_wait = FFMAX(0, fifo->recovery_wait_time - time_since_recovery);
424
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 30 times.
 30 if (time_to_wait)
425 ✗ av_usleep(FFMIN(10000, time_to_wait));
426 }
427
428 45 ret = fifo_thread_attempt_recovery(ctx, msg, err_no);
429
3/4
✓ Branch 0 taken 30 times.
✓ Branch 1 taken 15 times.
✓ Branch 2 taken 30 times.
✗ Branch 3 not taken.
 45 } while (ret == AVERROR(EAGAIN) && !fifo->drop_pkts_on_overflow);
430
431
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 15 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 15 if (ret == AVERROR(EAGAIN) && fifo->drop_pkts_on_overflow) {
432 ✗ if (msg->type == FIFO_WRITE_PACKET)
433 ✗ av_packet_unref(&msg->pkt);
434 ✗ ret = 0;
435 }
436
437 15 return ret;
438 }
439
440 6 static void *fifo_consumer_thread(void *data)
441 {
442 6 AVFormatContext *avf = data;
443 6 FifoContext *fifo = avf->priv_data;
444 6 AVThreadMessageQueue *queue = fifo->queue;
445 6 FifoMessage msg = {fifo->timeshift ? FIFO_NOOP : FIFO_WRITE_HEADER, {0}};
446 int ret, i;
447
448 FifoThreadContext fifo_thread_ctx;
449 6 memset(&fifo_thread_ctx, 0, sizeof(FifoThreadContext));
450 6 fifo_thread_ctx.avf = avf;
451 6 fifo_thread_ctx.last_received_dts = AV_NOPTS_VALUE;
452
453 6 ff_thread_setname("fifo-consumer");
454
455
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 5 times.
 11 for (i = 0; i < avf->nb_streams; i++) {
456
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 5 times.
 6 if (avf->streams[i]->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
457 1 fifo_thread_ctx.has_video_stream = 1;
458 1 break;
459 }
460 }
461
462 103 while (1) {
463 109 uint8_t just_flushed = 0;
464
465
1/2
✓ Branch 0 taken 109 times.
✗ Branch 1 not taken.
 109 if (!fifo_thread_ctx.recovery_nr)
466 109 ret = fifo_thread_dispatch_message(&fifo_thread_ctx, &msg);
467
468
3/4
✓ Branch 0 taken 94 times.
✓ Branch 1 taken 15 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 94 times.
 109 if (ret < 0 || fifo_thread_ctx.recovery_nr > 0) {
469 15 int rec_ret = fifo_thread_recover(&fifo_thread_ctx, &msg, ret);
470
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 15 times.
 15 if (rec_ret < 0) {
471 ✗ av_thread_message_queue_set_err_send(queue, rec_ret);
472 ✗ break;
473 }
474 }
475
476 /* If the queue is full at the moment when fifo_write_packet
477 * attempts to insert new message (packet) to the queue,
478 * it sets the fifo->overflow_flag to 1 and drops packet.
479 * Here in consumer thread, the flag is checked and if it is
480 * set, the queue is flushed and flag cleared. */
481 109 pthread_mutex_lock(&fifo->overflow_flag_lock);
482
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 108 times.
 109 if (fifo->overflow_flag) {
483 1 av_thread_message_flush(queue);
484
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (fifo->restart_with_keyframe)
485 ✗ fifo_thread_ctx.drop_until_keyframe = 1;
486 1 fifo->overflow_flag = 0;
487 1 just_flushed = 1;
488 }
489 109 pthread_mutex_unlock(&fifo->overflow_flag_lock);
490
491
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 108 times.
 109 if (just_flushed)
492 1 av_log(avf, AV_LOG_INFO, "FIFO queue flushed\n");
493
494
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 109 times.
 109 if (fifo->timeshift)
495 ✗ while (atomic_load_explicit(&fifo->queue_duration, memory_order_relaxed) < fifo->timeshift)
496 ✗ av_usleep(10000);
497
498 109 ret = av_thread_message_queue_recv(queue, &msg, 0);
499
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 103 times.
 109 if (ret < 0) {
500 6 av_thread_message_queue_set_err_send(queue, ret);
501 6 break;
502 }
503 }
504
505 6 fifo->write_trailer_ret = fifo_thread_write_trailer(&fifo_thread_ctx);
506
507 6 return NULL;
508 }
509
510 6 static int fifo_mux_init(AVFormatContext *avf, const AVOutputFormat *oformat,
511 const char *filename)
512 {
513 6 FifoContext *fifo = avf->priv_data;
514 AVFormatContext *avf2;
515 6 int ret = 0, i;
516
517 6 ret = avformat_alloc_output_context2(&avf2, oformat, NULL, filename);
518
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (ret < 0)
519 ✗ return ret;
520
521 6 fifo->avf = avf2;
522
523 6 avf2->interrupt_callback = avf->interrupt_callback;
524 6 avf2->max_delay = avf->max_delay;
525 6 ret = av_dict_copy(&avf2->metadata, avf->metadata, 0);
526
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (ret < 0)
527 ✗ return ret;
528 6 avf2->opaque = avf->opaque;
529 6 avf2->io_close2 = avf->io_close2;
530 6 avf2->io_open = avf->io_open;
531 6 avf2->flags = avf->flags;
532
533
2/2
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 6 times.
 13 for (i = 0; i < avf->nb_streams; ++i) {
534 7 AVStream *st = ff_stream_clone(avf2, avf->streams[i]);
535
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7 times.
 7 if (!st)
536 ✗ return AVERROR(ENOMEM);
537 }
538
539 6 return 0;
540 }
541
542 6 static int fifo_init(AVFormatContext *avf)
543 {
544 6 FifoContext *fifo = avf->priv_data;
545 const AVOutputFormat *oformat;
546 6 int ret = 0;
547
548
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 6 if (fifo->recovery_wait_streamtime && !fifo->drop_pkts_on_overflow) {
549 ✗ av_log(avf, AV_LOG_ERROR, "recovery_wait_streamtime can be turned on"
550 " only when drop_pkts_on_overflow is also turned on\n");
551 ✗ return AVERROR(EINVAL);
552 }
553 6 atomic_init(&fifo->queue_duration, 0);
554 6 fifo->last_sent_dts = AV_NOPTS_VALUE;
555
556 #ifdef FIFO_TEST
557 /* This exists for the fifo_muxer test tool. */
558
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 4 if (fifo->format && !strcmp(fifo->format, "fifo_test")) {
559 extern const FFOutputFormat ff_fifo_test_muxer;
560 4 oformat = &ff_fifo_test_muxer.p;
561 } else
562 #endif
563 2 oformat = av_guess_format(fifo->format, avf->url, NULL);
564
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (!oformat) {
565 ✗ ret = AVERROR_MUXER_NOT_FOUND;
566 ✗ return ret;
567 }
568
569 6 ret = fifo_mux_init(avf, oformat, avf->url);
570
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (ret < 0)
571 ✗ return ret;
572
573 6 ret = av_thread_message_queue_alloc(&fifo->queue, (unsigned) fifo->queue_size,
574 sizeof(FifoMessage));
575
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (ret < 0)
576 ✗ return ret;
577
578 6 av_thread_message_queue_set_free_func(fifo->queue, free_message);
579
580 6 ret = pthread_mutex_init(&fifo->overflow_flag_lock, NULL);
581
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (ret < 0)
582 ✗ return AVERROR(ret);
583 6 fifo->overflow_flag_lock_initialized = 1;
584
585 6 return 0;
586 }
587
588 6 static int fifo_write_header(AVFormatContext *avf)
589 {
590 6 FifoContext * fifo = avf->priv_data;
591 int ret;
592
593 6 ret = pthread_create(&fifo->writer_thread, NULL, fifo_consumer_thread, avf);
594
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (ret) {
595 ✗ av_log(avf, AV_LOG_ERROR, "Failed to start thread: %s\n",
596 ✗ av_err2str(AVERROR(ret)));
597 ✗ ret = AVERROR(ret);
598 }
599
600 6 return ret;
601 }
602
603 109 static int fifo_write_packet(AVFormatContext *avf, AVPacket *pkt)
604 {
605 109 FifoContext *fifo = avf->priv_data;
606
2/2
✓ Branch 0 taken 106 times.
✓ Branch 1 taken 3 times.
 109 FifoMessage msg = {.type = pkt ? FIFO_WRITE_PACKET : FIFO_FLUSH_OUTPUT};
607 int ret;
608
609
2/2
✓ Branch 0 taken 106 times.
✓ Branch 1 taken 3 times.
 109 if (pkt) {
610 106 ret = av_packet_ref(&msg.pkt,pkt);
611
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 106 times.
 106 if (ret < 0)
612 ✗ return ret;
613 }
614
615 109 ret = av_thread_message_queue_send(fifo->queue, &msg,
616 109 fifo->drop_pkts_on_overflow ?
617 AV_THREAD_MESSAGE_NONBLOCK : 0);
618
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 106 times.
 109 if (ret == AVERROR(EAGAIN)) {
619 3 uint8_t overflow_set = 0;
620
621 /* Queue is full, set fifo->overflow_flag to 1
622 * to let consumer thread know the queue should
623 * be flushed. */
624 3 pthread_mutex_lock(&fifo->overflow_flag_lock);
625
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 2 times.
 3 if (!fifo->overflow_flag)
626 1 fifo->overflow_flag = overflow_set = 1;
627 3 pthread_mutex_unlock(&fifo->overflow_flag_lock);
628
629
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 2 times.
 3 if (overflow_set)
630 1 av_log(avf, AV_LOG_WARNING, "FIFO queue full\n");
631 3 ret = 0;
632 3 goto fail;
633
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 106 times.
 106 } else if (ret < 0) {
634 ✗ goto fail;
635 }
636
637
1/6
✗ Branch 0 not taken.
✓ Branch 1 taken 106 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 106 if (fifo->timeshift && pkt && pkt->dts != AV_NOPTS_VALUE)
638 ✗ atomic_fetch_add_explicit(&fifo->queue_duration, next_duration(avf, pkt, &fifo->last_sent_dts), memory_order_relaxed);
639
640 106 return ret;
641 3 fail:
642
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 if (pkt)
643 3 av_packet_unref(&msg.pkt);
644 3 return ret;
645 }
646
647 6 static int fifo_write_trailer(AVFormatContext *avf)
648 {
649 6 FifoContext *fifo= avf->priv_data;
650 int ret;
651
652 6 av_thread_message_queue_set_err_recv(fifo->queue, AVERROR_EOF);
653
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (fifo->timeshift) {
654 ✗ int64_t now = av_gettime_relative();
655 ✗ int64_t elapsed = 0;
656 ✗ FifoMessage msg = {FIFO_NOOP};
657 do {
658 ✗ int64_t delay = av_gettime_relative() - now;
659 ✗ if (delay < 0) { // Discontinuity?
660 ✗ delay = 10000;
661 ✗ now = av_gettime_relative();
662 } else {
663 ✗ now += delay;
664 }
665 ✗ atomic_fetch_add_explicit(&fifo->queue_duration, delay, memory_order_relaxed);
666 ✗ elapsed += delay;
667 ✗ if (elapsed > fifo->timeshift)
668 ✗ break;
669 ✗ av_usleep(10000);
670 ✗ ret = av_thread_message_queue_send(fifo->queue, &msg, AV_THREAD_MESSAGE_NONBLOCK);
671 ✗ } while (ret >= 0 || ret == AVERROR(EAGAIN));
672 ✗ atomic_store(&fifo->queue_duration, INT64_MAX);
673 }
674
675 6 ret = pthread_join(fifo->writer_thread, NULL);
676
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (ret < 0) {
677 ✗ av_log(avf, AV_LOG_ERROR, "pthread join error: %s\n",
678 ✗ av_err2str(AVERROR(ret)));
679 ✗ return AVERROR(ret);
680 }
681
682 6 ret = fifo->write_trailer_ret;
683 6 return ret;
684 }
685
686 6 static void fifo_deinit(AVFormatContext *avf)
687 {
688 6 FifoContext *fifo = avf->priv_data;
689
690 6 avformat_free_context(fifo->avf);
691 6 av_thread_message_queue_free(&fifo->queue);
692
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 if (fifo->overflow_flag_lock_initialized)
693 6 pthread_mutex_destroy(&fifo->overflow_flag_lock);
694 6 }
695
696 #define OFFSET(x) offsetof(FifoContext, x)
697 static const AVOption options[] = {
698 {"attempt_recovery", "Attempt recovery in case of failure", OFFSET(attempt_recovery),
699 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
700
701 {"drop_pkts_on_overflow", "Drop packets on fifo queue overflow not to block encoder", OFFSET(drop_pkts_on_overflow),
702 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
703
704 {"fifo_format", "Target muxer", OFFSET(format),
705 AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
706
707 {"format_opts", "Options to be passed to underlying muxer", OFFSET(format_options),
708 AV_OPT_TYPE_DICT, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
709
710 {"max_recovery_attempts", "Maximal number of recovery attempts", OFFSET(max_recovery_attempts),
711 AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS}, 0, INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
712
713 {"queue_size", "Size of fifo queue", OFFSET(queue_size),
714 AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_QUEUE_SIZE}, 1, INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
715
716 {"recovery_wait_streamtime", "Use stream time instead of real time while waiting for recovery",
717 OFFSET(recovery_wait_streamtime), AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
718
719 {"recovery_wait_time", "Waiting time between recovery attempts", OFFSET(recovery_wait_time),
720 AV_OPT_TYPE_DURATION, {.i64 = FIFO_DEFAULT_RECOVERY_WAIT_TIME_USEC}, 0, INT64_MAX, AV_OPT_FLAG_ENCODING_PARAM},
721
722 {"recover_any_error", "Attempt recovery regardless of type of the error", OFFSET(recover_any_error),
723 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
724
725 {"restart_with_keyframe", "Wait for keyframe when restarting output", OFFSET(restart_with_keyframe),
726 AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
727
728 {"timeshift", "Delay fifo output", OFFSET(timeshift),
729 AV_OPT_TYPE_DURATION, {.i64 = 0}, 0, INT64_MAX, AV_OPT_FLAG_ENCODING_PARAM},
730
731 {NULL},
732 };
733
734 static const AVClass fifo_muxer_class = {
735 .class_name = "Fifo muxer",
736 .item_name = av_default_item_name,
737 .option = options,
738 .version = LIBAVUTIL_VERSION_INT,
739 };
740
741 const FFOutputFormat ff_fifo_muxer = {
742 .p.name = "fifo",
743 .p.long_name = NULL_IF_CONFIG_SMALL("FIFO queue pseudo-muxer"),
744 .p.priv_class = &fifo_muxer_class,
745 .p.flags = AVFMT_NOFILE | AVFMT_TS_NEGATIVE,
746 .priv_data_size = sizeof(FifoContext),
747 .init = fifo_init,
748 .write_header = fifo_write_header,
749 .write_packet = fifo_write_packet,
750 .write_trailer = fifo_write_trailer,
751 .deinit = fifo_deinit,
752 .flags_internal = FF_OFMT_FLAG_ALLOW_FLUSH,
753 };
754