FFmpeg coverage


Directory: ../../../ffmpeg/
File: src/libavformat/async.c
Date: 2026-08-15 14:54:27
Exec Total Coverage
Lines: 0 243 0.0%
Functions: 0 17 0.0%
Branches: 0 84 0.0%

Line Branch Exec Source
1 /*
2 * Input async protocol.
3 * Copyright (c) 2015 Zhang Rui <bbcallen@gmail.com>
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
9 * License 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 GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with FFmpeg; if not, write to the Free Software
19 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
20 *
21 * Based on libavformat/cache.c by Michael Niedermayer
22 */
23
24 /**
25 * @TODO
26 * support timeout
27 * support work with concatdec, hls
28 */
29
30 #include <inttypes.h>
31 #include <string.h>
32
33 #include "libavutil/avassert.h"
34 #include "libavutil/avstring.h"
35 #include "libavutil/error.h"
36 #include "libavutil/fifo.h"
37 #include "libavutil/log.h"
38 #include "libavutil/opt.h"
39 #include "libavutil/thread.h"
40 #include "url.h"
41
42 #if HAVE_UNISTD_H
43 #include <unistd.h>
44 #endif
45
46 #define BUFFER_CAPACITY (4 * 1024 * 1024)
47 #define READ_BACK_CAPACITY (4 * 1024 * 1024)
48 #define SHORT_SEEK_THRESHOLD (256 * 1024)
49
50 typedef struct RingBuffer
51 {
52 AVFifo *fifo;
53 int read_back_capacity;
54
55 int read_pos;
56 } RingBuffer;
57
58 typedef struct AsyncContext {
59 AVClass *class;
60 URLContext *inner;
61
62 int seek_request;
63 int64_t seek_pos;
64 int seek_whence;
65 int seek_completed;
66 int64_t seek_ret;
67
68 int inner_io_error;
69 int io_error;
70 int io_eof_reached;
71
72 int64_t logical_pos;
73 int64_t logical_size;
74 RingBuffer ring;
75
76 pthread_cond_t cond_wakeup_main;
77 pthread_cond_t cond_wakeup_background;
78 pthread_mutex_t mutex;
79 pthread_t async_buffer_thread;
80
81 int abort_request;
82 AVIOInterruptCB interrupt_callback;
83 } AsyncContext;
84
85 static int ring_init(RingBuffer *ring, unsigned int capacity, int read_back_capacity)
86 {
87 memset(ring, 0, sizeof(RingBuffer));
88 ring->fifo = av_fifo_alloc2(capacity + read_back_capacity, 1, 0);
89 if (!ring->fifo)
90 return AVERROR(ENOMEM);
91
92 ring->read_back_capacity = read_back_capacity;
93 return 0;
94 }
95
96 static void ring_destroy(RingBuffer *ring)
97 {
98 av_fifo_freep2(&ring->fifo);
99 }
100
101 static void ring_reset(RingBuffer *ring)
102 {
103 av_fifo_reset2(ring->fifo);
104 ring->read_pos = 0;
105 }
106
107 static int ring_size(RingBuffer *ring)
108 {
109 return av_fifo_can_read(ring->fifo) - ring->read_pos;
110 }
111
112 static int ring_space(RingBuffer *ring)
113 {
114 return av_fifo_can_write(ring->fifo);
115 }
116
117 static int ring_read(RingBuffer *ring, void *dest, int buf_size)
118 {
119 int ret = 0;
120
121 av_assert2(buf_size <= ring_size(ring));
122 if (dest)
123 ret = av_fifo_peek(ring->fifo, dest, buf_size, ring->read_pos);
124 ring->read_pos += buf_size;
125
126 if (ring->read_pos > ring->read_back_capacity) {
127 av_fifo_drain2(ring->fifo, ring->read_pos - ring->read_back_capacity);
128 ring->read_pos = ring->read_back_capacity;
129 }
130
131 return ret;
132 }
133
134 static int wrapped_url_read(void *src, void *dst, size_t *size)
135 {
136 URLContext *h = src;
137 AsyncContext *c = h->priv_data;
138 int ret;
139
140 ret = ffurl_read(c->inner, dst, *size);
141 *size = ret > 0 ? ret : 0;
142 c->inner_io_error = ret < 0 ? ret : 0;
143
144 return c->inner_io_error;
145 }
146
147 static int ring_write(RingBuffer *ring, URLContext *h, size_t size)
148 {
149 int ret;
150
151 av_assert2(size <= ring_space(ring));
152 ret = av_fifo_write_from_cb(ring->fifo, wrapped_url_read, h, &size);
153 if (ret < 0)
154 return ret;
155
156 return size;
157 }
158
159 static int ring_size_of_read_back(RingBuffer *ring)
160 {
161 return ring->read_pos;
162 }
163
164 static int ring_drain(RingBuffer *ring, int offset)
165 {
166 av_assert2(offset >= -ring_size_of_read_back(ring));
167 av_assert2(offset <= ring_size(ring));
168 ring->read_pos += offset;
169 return 0;
170 }
171
172 static int async_check_interrupt(void *arg)
173 {
174 URLContext *h = arg;
175 AsyncContext *c = h->priv_data;
176
177 if (c->abort_request)
178 return 1;
179
180 if (ff_check_interrupt(&c->interrupt_callback))
181 c->abort_request = 1;
182
183 return c->abort_request;
184 }
185
186 static void *async_buffer_task(void *arg)
187 {
188 URLContext *h = arg;
189 AsyncContext *c = h->priv_data;
190 RingBuffer *ring = &c->ring;
191 int ret = 0;
192 int64_t seek_ret;
193
194 ff_thread_setname("async");
195
196 while (1) {
197 int fifo_space, to_copy;
198
199 pthread_mutex_lock(&c->mutex);
200 if (async_check_interrupt(h)) {
201 c->io_eof_reached = 1;
202 c->io_error = AVERROR_EXIT;
203 pthread_cond_signal(&c->cond_wakeup_main);
204 pthread_mutex_unlock(&c->mutex);
205 break;
206 }
207
208 if (c->seek_request) {
209 seek_ret = ffurl_seek(c->inner, c->seek_pos, c->seek_whence);
210 if (seek_ret >= 0) {
211 c->io_eof_reached = 0;
212 c->io_error = 0;
213 ring_reset(ring);
214 }
215
216 c->seek_completed = 1;
217 c->seek_ret = seek_ret;
218 c->seek_request = 0;
219
220
221 pthread_cond_signal(&c->cond_wakeup_main);
222 pthread_mutex_unlock(&c->mutex);
223 continue;
224 }
225
226 fifo_space = ring_space(ring);
227 if (c->io_eof_reached || fifo_space <= 0) {
228 pthread_cond_signal(&c->cond_wakeup_main);
229 pthread_cond_wait(&c->cond_wakeup_background, &c->mutex);
230 pthread_mutex_unlock(&c->mutex);
231 continue;
232 }
233 pthread_mutex_unlock(&c->mutex);
234
235 to_copy = FFMIN(4096, fifo_space);
236 ret = ring_write(ring, h, to_copy);
237
238 pthread_mutex_lock(&c->mutex);
239 if (ret <= 0) {
240 c->io_eof_reached = 1;
241 if (c->inner_io_error < 0)
242 c->io_error = c->inner_io_error;
243 }
244
245 pthread_cond_signal(&c->cond_wakeup_main);
246 pthread_mutex_unlock(&c->mutex);
247 }
248
249 return NULL;
250 }
251
252 static int async_open(URLContext *h, const char *arg, int flags, AVDictionary **options)
253 {
254 AsyncContext *c = h->priv_data;
255 int ret;
256 AVIOInterruptCB interrupt_callback = {.callback = async_check_interrupt, .opaque = h};
257
258 av_strstart(arg, "async:", &arg);
259
260 ret = ring_init(&c->ring, BUFFER_CAPACITY, READ_BACK_CAPACITY);
261 if (ret < 0)
262 goto fifo_fail;
263
264 /* wrap interrupt callback */
265 c->interrupt_callback = h->interrupt_callback;
266 ret = ffurl_open_whitelist(&c->inner, arg, flags, &interrupt_callback, options, h->protocol_whitelist, h->protocol_blacklist, h);
267 if (ret != 0) {
268 av_log(h, AV_LOG_ERROR, "ffurl_open failed : %s, %s\n", av_err2str(ret), arg);
269 goto url_fail;
270 }
271
272 c->logical_size = ffurl_size(c->inner);
273 h->is_streamed = c->inner->is_streamed;
274
275 ret = pthread_mutex_init(&c->mutex, NULL);
276 if (ret != 0) {
277 ret = AVERROR(ret);
278 av_log(h, AV_LOG_ERROR, "pthread_mutex_init failed : %s\n", av_err2str(ret));
279 goto mutex_fail;
280 }
281
282 ret = pthread_cond_init(&c->cond_wakeup_main, NULL);
283 if (ret != 0) {
284 ret = AVERROR(ret);
285 av_log(h, AV_LOG_ERROR, "pthread_cond_init failed : %s\n", av_err2str(ret));
286 goto cond_wakeup_main_fail;
287 }
288
289 ret = pthread_cond_init(&c->cond_wakeup_background, NULL);
290 if (ret != 0) {
291 ret = AVERROR(ret);
292 av_log(h, AV_LOG_ERROR, "pthread_cond_init failed : %s\n", av_err2str(ret));
293 goto cond_wakeup_background_fail;
294 }
295
296 ret = pthread_create(&c->async_buffer_thread, NULL, async_buffer_task, h);
297 if (ret) {
298 ret = AVERROR(ret);
299 av_log(h, AV_LOG_ERROR, "pthread_create failed : %s\n", av_err2str(ret));
300 goto thread_fail;
301 }
302
303 return 0;
304
305 thread_fail:
306 pthread_cond_destroy(&c->cond_wakeup_background);
307 cond_wakeup_background_fail:
308 pthread_cond_destroy(&c->cond_wakeup_main);
309 cond_wakeup_main_fail:
310 pthread_mutex_destroy(&c->mutex);
311 mutex_fail:
312 ffurl_closep(&c->inner);
313 url_fail:
314 ring_destroy(&c->ring);
315 fifo_fail:
316 return ret;
317 }
318
319 static int async_close(URLContext *h)
320 {
321 AsyncContext *c = h->priv_data;
322 int ret;
323
324 pthread_mutex_lock(&c->mutex);
325 c->abort_request = 1;
326 pthread_cond_signal(&c->cond_wakeup_background);
327 pthread_mutex_unlock(&c->mutex);
328
329 ret = pthread_join(c->async_buffer_thread, NULL);
330 if (ret != 0)
331 av_log(h, AV_LOG_ERROR, "pthread_join(): %s\n", av_err2str(ret));
332
333 pthread_cond_destroy(&c->cond_wakeup_background);
334 pthread_cond_destroy(&c->cond_wakeup_main);
335 pthread_mutex_destroy(&c->mutex);
336 ffurl_closep(&c->inner);
337 ring_destroy(&c->ring);
338
339 return 0;
340 }
341
342 static int async_read_internal(URLContext *h, void *dest, int size)
343 {
344 AsyncContext *c = h->priv_data;
345 RingBuffer *ring = &c->ring;
346 int read_complete = !dest;
347 int to_read = size;
348 int ret = 0;
349
350 pthread_mutex_lock(&c->mutex);
351
352 while (to_read > 0) {
353 int fifo_size, to_copy;
354 if (async_check_interrupt(h)) {
355 ret = AVERROR_EXIT;
356 break;
357 }
358 fifo_size = ring_size(ring);
359 to_copy = FFMIN(to_read, fifo_size);
360 if (to_copy > 0) {
361 ring_read(ring, dest, to_copy);
362 if (dest)
363 dest = (uint8_t *)dest + to_copy;
364 c->logical_pos += to_copy;
365 to_read -= to_copy;
366 ret = size - to_read;
367
368 if (to_read <= 0 || !read_complete)
369 break;
370 } else if (c->io_eof_reached) {
371 if (ret <= 0) {
372 if (c->io_error)
373 ret = c->io_error;
374 else
375 ret = AVERROR_EOF;
376 }
377 break;
378 }
379 pthread_cond_signal(&c->cond_wakeup_background);
380 pthread_cond_wait(&c->cond_wakeup_main, &c->mutex);
381 }
382
383 pthread_cond_signal(&c->cond_wakeup_background);
384 pthread_mutex_unlock(&c->mutex);
385
386 return ret;
387 }
388
389 static int async_read(URLContext *h, unsigned char *buf, int size)
390 {
391 return async_read_internal(h, buf, size);
392 }
393
394 static int64_t async_seek(URLContext *h, int64_t pos, int whence)
395 {
396 AsyncContext *c = h->priv_data;
397 RingBuffer *ring = &c->ring;
398 int64_t ret;
399 int64_t new_logical_pos;
400 int fifo_size;
401 int fifo_size_of_read_back;
402
403 if (whence == AVSEEK_SIZE) {
404 av_log(h, AV_LOG_TRACE, "async_seek: AVSEEK_SIZE: %"PRId64"\n", (int64_t)c->logical_size);
405 return c->logical_size;
406 } else if (whence == SEEK_CUR) {
407 av_log(h, AV_LOG_TRACE, "async_seek: %"PRId64"\n", pos);
408 new_logical_pos = pos + c->logical_pos;
409 } else if (whence == SEEK_SET){
410 av_log(h, AV_LOG_TRACE, "async_seek: %"PRId64"\n", pos);
411 new_logical_pos = pos;
412 } else {
413 return AVERROR(EINVAL);
414 }
415 if (new_logical_pos < 0)
416 return AVERROR(EINVAL);
417
418 fifo_size = ring_size(ring);
419 fifo_size_of_read_back = ring_size_of_read_back(ring);
420 if (new_logical_pos == c->logical_pos) {
421 /* current position */
422 return c->logical_pos;
423 } else if ((new_logical_pos >= (c->logical_pos - fifo_size_of_read_back)) &&
424 (new_logical_pos < (c->logical_pos + fifo_size + SHORT_SEEK_THRESHOLD))) {
425 int pos_delta = (int)(new_logical_pos - c->logical_pos);
426 /* fast seek */
427 av_log(h, AV_LOG_TRACE, "async_seek: fask_seek %"PRId64" from %d dist:%d/%d\n",
428 new_logical_pos, (int)c->logical_pos,
429 (int)(new_logical_pos - c->logical_pos), fifo_size);
430
431 if (pos_delta > 0) {
432 // fast seek forwards
433 async_read_internal(h, NULL, pos_delta);
434 } else {
435 // fast seek backwards
436 ring_drain(ring, pos_delta);
437 c->logical_pos = new_logical_pos;
438 }
439
440 return c->logical_pos;
441 } else if (c->logical_size <= 0) {
442 /* can not seek */
443 return AVERROR(EINVAL);
444 } else if (new_logical_pos > c->logical_size) {
445 /* beyond end */
446 return AVERROR(EINVAL);
447 }
448
449 pthread_mutex_lock(&c->mutex);
450
451 c->seek_request = 1;
452 c->seek_pos = new_logical_pos;
453 c->seek_whence = SEEK_SET;
454 c->seek_completed = 0;
455 c->seek_ret = 0;
456
457 while (1) {
458 if (async_check_interrupt(h)) {
459 ret = AVERROR_EXIT;
460 break;
461 }
462 if (c->seek_completed) {
463 if (c->seek_ret >= 0)
464 c->logical_pos = c->seek_ret;
465 ret = c->seek_ret;
466 break;
467 }
468 pthread_cond_signal(&c->cond_wakeup_background);
469 pthread_cond_wait(&c->cond_wakeup_main, &c->mutex);
470 }
471
472 pthread_mutex_unlock(&c->mutex);
473
474 return ret;
475 }
476
477 #define OFFSET(x) offsetof(AsyncContext, x)
478 #define D AV_OPT_FLAG_DECODING_PARAM
479
480 static const AVOption options[] = {
481 {NULL},
482 };
483
484 #undef D
485 #undef OFFSET
486
487 static const AVClass async_context_class = {
488 .class_name = "Async",
489 .item_name = av_default_item_name,
490 .option = options,
491 .version = LIBAVUTIL_VERSION_INT,
492 };
493
494 const URLProtocol ff_async_protocol = {
495 .name = "async",
496 .url_open2 = async_open,
497 .url_read = async_read,
498 .url_seek = async_seek,
499 .url_close = async_close,
500 .priv_data_size = sizeof(AsyncContext),
501 .priv_data_class = &async_context_class,
502 };
503
504 #if 0
505
506 #define TEST_SEEK_POS (1536)
507 #define TEST_STREAM_SIZE (2048)
508
509 typedef struct TestContext {
510 AVClass *class;
511 int64_t logical_pos;
512 int64_t logical_size;
513
514 /* options */
515 int opt_read_error;
516 } TestContext;
517
518 static int async_test_open(URLContext *h, const char *arg, int flags, AVDictionary **options)
519 {
520 TestContext *c = h->priv_data;
521 c->logical_pos = 0;
522 c->logical_size = TEST_STREAM_SIZE;
523 return 0;
524 }
525
526 static int async_test_close(URLContext *h)
527 {
528 return 0;
529 }
530
531 static int async_test_read(URLContext *h, unsigned char *buf, int size)
532 {
533 TestContext *c = h->priv_data;
534 int i;
535 int read_len = 0;
536
537 if (c->opt_read_error)
538 return c->opt_read_error;
539
540 if (c->logical_pos >= c->logical_size)
541 return AVERROR_EOF;
542
543 for (i = 0; i < size; ++i) {
544 buf[i] = c->logical_pos & 0xFF;
545
546 c->logical_pos++;
547 read_len++;
548
549 if (c->logical_pos >= c->logical_size)
550 break;
551 }
552
553 return read_len;
554 }
555
556 static int64_t async_test_seek(URLContext *h, int64_t pos, int whence)
557 {
558 TestContext *c = h->priv_data;
559 int64_t new_logical_pos;
560
561 if (whence == AVSEEK_SIZE) {
562 return c->logical_size;
563 } else if (whence == SEEK_CUR) {
564 new_logical_pos = pos + c->logical_pos;
565 } else if (whence == SEEK_SET){
566 new_logical_pos = pos;
567 } else {
568 return AVERROR(EINVAL);
569 }
570 if (new_logical_pos < 0)
571 return AVERROR(EINVAL);
572
573 c->logical_pos = new_logical_pos;
574 return new_logical_pos;
575 }
576
577 #define OFFSET(x) offsetof(TestContext, x)
578 #define D AV_OPT_FLAG_DECODING_PARAM
579
580 static const AVOption async_test_options[] = {
581 { "async-test-read-error", "cause read fail",
582 OFFSET(opt_read_error), AV_OPT_TYPE_INT, { .i64 = 0 }, INT_MIN, INT_MAX, .flags = D },
583 {NULL},
584 };
585
586 #undef D
587 #undef OFFSET
588
589 static const AVClass async_test_context_class = {
590 .class_name = "Async-Test",
591 .item_name = av_default_item_name,
592 .option = async_test_options,
593 .version = LIBAVUTIL_VERSION_INT,
594 };
595
596 const URLProtocol ff_async_test_protocol = {
597 .name = "async-test",
598 .url_open2 = async_test_open,
599 .url_read = async_test_read,
600 .url_seek = async_test_seek,
601 .url_close = async_test_close,
602 .priv_data_size = sizeof(TestContext),
603 .priv_data_class = &async_test_context_class,
604 };
605
606 int main(void)
607 {
608 URLContext *h = NULL;
609 int i;
610 int ret;
611 int64_t size;
612 int64_t pos;
613 int64_t read_len;
614 unsigned char buf[4096];
615 AVDictionary *opts = NULL;
616
617 ffurl_register_protocol(&ff_async_protocol);
618 ffurl_register_protocol(&ff_async_test_protocol);
619
620 /*
621 * test normal read
622 */
623 ret = ffurl_open_whitelist(&h, "async:async-test:", AVIO_FLAG_READ,
624 NULL, NULL, NULL, NULL, NULL);
625 printf("open: %d\n", ret);
626
627 size = ffurl_size(h);
628 printf("size: %"PRId64"\n", size);
629
630 pos = ffurl_seek(h, 0, SEEK_CUR);
631 read_len = 0;
632 while (1) {
633 ret = ffurl_read(h, buf, sizeof(buf));
634 if (ret == AVERROR_EOF) {
635 printf("read-error: AVERROR_EOF at %"PRId64"\n", ffurl_seek(h, 0, SEEK_CUR));
636 break;
637 }
638 else if (ret == 0)
639 break;
640 else if (ret < 0) {
641 printf("read-error: %d at %"PRId64"\n", ret, ffurl_seek(h, 0, SEEK_CUR));
642 goto fail;
643 } else {
644 for (i = 0; i < ret; ++i) {
645 if (buf[i] != (pos & 0xFF)) {
646 printf("read-mismatch: actual %d, expecting %d, at %"PRId64"\n",
647 (int)buf[i], (int)(pos & 0xFF), pos);
648 break;
649 }
650 pos++;
651 }
652 }
653
654 read_len += ret;
655 }
656 printf("read: %"PRId64"\n", read_len);
657
658 /*
659 * test normal seek
660 */
661 ret = ffurl_read(h, buf, 1);
662 printf("read: %d\n", ret);
663
664 pos = ffurl_seek(h, TEST_SEEK_POS, SEEK_SET);
665 printf("seek: %"PRId64"\n", pos);
666
667 read_len = 0;
668 while (1) {
669 ret = ffurl_read(h, buf, sizeof(buf));
670 if (ret == AVERROR_EOF)
671 break;
672 else if (ret == 0)
673 break;
674 else if (ret < 0) {
675 printf("read-error: %d at %"PRId64"\n", ret, ffurl_seek(h, 0, SEEK_CUR));
676 goto fail;
677 } else {
678 for (i = 0; i < ret; ++i) {
679 if (buf[i] != (pos & 0xFF)) {
680 printf("read-mismatch: actual %d, expecting %d, at %"PRId64"\n",
681 (int)buf[i], (int)(pos & 0xFF), pos);
682 break;
683 }
684 pos++;
685 }
686 }
687
688 read_len += ret;
689 }
690 printf("read: %"PRId64"\n", read_len);
691
692 ret = ffurl_read(h, buf, 1);
693 printf("read: %d\n", ret);
694
695 /*
696 * test read error
697 */
698 ffurl_close(h);
699 av_dict_set_int(&opts, "async-test-read-error", -10000, 0);
700 ret = ffurl_open_whitelist(&h, "async:async-test:", AVIO_FLAG_READ,
701 NULL, &opts, NULL, NULL, NULL);
702 printf("open: %d\n", ret);
703
704 ret = ffurl_read(h, buf, 1);
705 printf("read: %d\n", ret);
706
707 fail:
708 av_dict_free(&opts);
709 ffurl_close(h);
710 return 0;
711 }
712
713 #endif
714