FFmpeg coverage


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