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