PR #25404 opened by Boosung Kim (boosungkim) URL: https://code.ffmpeg.org/FFmpeg/FFmpeg/pulls/25404 Patch URL: https://code.ffmpeg.org/FFmpeg/FFmpeg/pulls/25404.patch
The async protocol's background thread writes into the shared `AVFifo` without holding the mutex. `async_seek()`'s fast path also reads and moves the ring without it. Readers can therefore see a torn FIFO state, typically just after a seek resets the ring. The result is wrong data, `--assert-level=2` aborts in `ring_read()`, and with default asserts an abort in `av_fifo_drain2()` (`Assertion cur_size >= size failed at libavutil/fifo.c:270`). The fix does the blocking inner read into a stack buffer outside the lock and commits it to the FIFO under the lock. The read is dropped if a seek arrived in the meantime. `async_seek()`'s ring inspection and drain now run under the lock. `av_fifo_peek()` and forward fast-seek errors are now returned instead of ignored. ## Testing Before/after (macOS arm64, master `d19f86ff4c`): | Test | Before | After | |------|--------|-------| | Seek/verify stress, seeks ≥16 MiB apart, 10×100k seeks | wrong data in 9/10 runs | 0/10 | | Seek/verify stress, seeks within 6 MB, 20×200k seeks | wrong data in 20/20 runs | 0/20 | | Same, `--assert-level=2` | 20/20 runs abort in `ring_read()` | 0/20 | | TSAN, `ffprobe -show_packets async:cache:http://…`, 40 runs | races in 40/40 | 0/40 | | `ffprobe -show_packets` output vs. `file:` (MP4, ticket6132 sample) | identical | identical | | `ffmpeg` MKV remux + mpeg4/aac transcode vs. `file:` over async URLs | byte-identical to `file:` | byte-identical to `file:` | ## Validations - `async_seek_stress.c` is a single file that links libavformat. `async_seek_stress -g pattern-256m.bin 268435456` writes a 256 MiB file: the byte at offset `pos` is `(pos * 0x9E3779B97F4A7C15ULL) >> 56`. The test opens that path as an AVIO URL and repeats `iterations` times: draw a seek position, `avio_seek`, read 4096 bytes with `avio_read`, check every byte against the same formula. The optional seed selects the xorshift sequence. - Far-seek rows in the table: no fifth argument, so each new offset satisfies `llabs(pos - prev) >= 16<<20` (ring reset every seek). Command shape: `async_seek_stress async:file:pattern-256m.bin 100000 <seed>`. Stock usually printed `wrong data from byte 0` on the first bad seek; patched ended with `bad_reads=0`. - Near-seek rows: fifth argument `6000000`, so each offset is `prev ± 6 MiB` (uniform), which exercises forward/backward fast seek inside the ring. Same 4096-byte read after each seek. With `--assert-level=2`, stock aborted in `ring_read()` before finishing the run; patched completed with `bad_reads=0`. - HTTP / TSAN rows: any HTTP server with Range support works; I used `npx http-server media -p 8793 -c-1` on localhost, no throttling. Input file `test-chunkdur.mp4` (5 min lavfi H.264/AAC, then `ffmpeg -c copy -chunk_duration 4000000`). Probe URL: `async:cache:http://127.0.0.1:8793/test-chunkdur.mp4`. Looped `ffprobe -show_packets` with a `--toolchain=clang-tsan` build, 20 runs each for stock and patched. - No-output-change rows: `ffprobe -show_packets -of csv` on that MP4 through `file:`, `async:file:`, and the HTTP URL above — `cmp` identical. `ffmpeg` stream copy to MKV with `-fflags +bitexact` and mpeg4/aac transcode the same three ways; sha256 matched the `file:` reference on stock and patched. For ticket6132, 200× `ffprobe -show_packets async:file:<path>` vs `file:` — same CSV, no aborts. Behaviour note: a forward fast seek that reaches EOF or fails before reading anything now returns that error. Previously it returned the unchanged position. Seeks that make partial progress behave as before. From 84d11c22f1f92b2473d7f90d18297c17daf6f881 Mon Sep 17 00:00:00 2001 From: boosungkim <[email protected]> Date: Fri, 23 Oct 2026 16:52:05 -0700 Subject: [PATCH] avformat/async: Fix data races on the ring buffer The background thread called av_fifo_write_from_cb() without holding the mutex, so the FIFO was updated concurrently with reads and drains on the main thread. The fast seek path in async_seek() also accessed the ring without the mutex. Read from the inner protocol into a stack buffer outside the lock, then av_fifo_write() under the lock. Discard the read if a seek was requested while blocked. Hold the mutex while async_seek() inspects and drains the ring. Propagate errors from av_fifo_peek() and from forward fast seek. Reproduced with ThreadSanitizer and a seek/read stress test. Wrong reads and assert failures in ring_read()/libavutil/fifo.c on stock, none on patched. ffprobe -show_packets output unchanged on tested inputs. Signed-off-by: Boosung Kim <[email protected]> --- libavformat/async.c | 70 ++++++++++++++++++++++----------------------- 1 file changed, 35 insertions(+), 35 deletions(-) diff --git a/libavformat/async.c b/libavformat/async.c index ab63cce5db2..55f6a4008fa 100644 --- a/libavformat/async.c +++ b/libavformat/async.c @@ -66,7 +66,6 @@ typedef struct AsyncContext { int seek_completed; int64_t seek_ret; - int inner_io_error; int io_error; int io_eof_reached; @@ -120,8 +119,11 @@ static int ring_read(RingBuffer *ring, void *dest, int buf_size) int ret = 0; av_assert2(buf_size <= ring_size(ring)); - if (dest) + if (dest) { ret = av_fifo_peek(ring->fifo, dest, buf_size, ring->read_pos); + if (ret < 0) + return ret; + } ring->read_pos += buf_size; if (ring->read_pos > ring->read_back_capacity) { @@ -132,31 +134,6 @@ static int ring_read(RingBuffer *ring, void *dest, int buf_size) return ret; } -static int wrapped_url_read(void *src, void *dst, size_t *size) -{ - URLContext *h = src; - AsyncContext *c = h->priv_data; - int ret; - - ret = ffurl_read(c->inner, dst, *size); - *size = ret > 0 ? ret : 0; - c->inner_io_error = ret < 0 ? ret : 0; - - return c->inner_io_error; -} - -static int ring_write(RingBuffer *ring, URLContext *h, size_t size) -{ - int ret; - - av_assert2(size <= ring_space(ring)); - ret = av_fifo_write_from_cb(ring->fifo, wrapped_url_read, h, &size); - if (ret < 0) - return ret; - - return size; -} - static int ring_size_of_read_back(RingBuffer *ring) { return ring->read_pos; @@ -189,6 +166,7 @@ static void *async_buffer_task(void *arg) URLContext *h = arg; AsyncContext *c = h->priv_data; RingBuffer *ring = &c->ring; + uint8_t buf[4096]; int ret = 0; int64_t seek_ret; @@ -234,13 +212,26 @@ static void *async_buffer_task(void *arg) pthread_mutex_unlock(&c->mutex); to_copy = FFMIN(4096, fifo_space); - ret = ring_write(ring, h, to_copy); + ret = ffurl_read(c->inner, buf, to_copy); pthread_mutex_lock(&c->mutex); - if (ret <= 0) { + if (ret > 0) { + /* + * A seek may have been requested while the inner read blocked. + * In that case the bytes belong to the old position and must not + * be committed before the seek resets the ring. + */ + if (!c->seek_request) { + int write_ret = av_fifo_write(ring->fifo, buf, ret); + if (write_ret < 0) { + c->io_eof_reached = 1; + c->io_error = write_ret; + } + } + } else { c->io_eof_reached = 1; - if (c->inner_io_error < 0) - c->io_error = c->inner_io_error; + if (ret < 0) + c->io_error = ret; } pthread_cond_signal(&c->cond_wakeup_main); @@ -360,7 +351,9 @@ static int async_read_internal(URLContext *h, void *dest, int size) fifo_size = ring_size(ring); to_copy = FFMIN(to_read, fifo_size); if (to_copy > 0) { - ring_read(ring, dest, to_copy); + ret = ring_read(ring, dest, to_copy); + if (ret < 0) + break; if (dest) dest = (uint8_t *)dest + to_copy; c->logical_pos += to_copy; @@ -417,10 +410,13 @@ static int64_t async_seek(URLContext *h, int64_t pos, int whence) if (new_logical_pos < 0) return AVERROR(EINVAL); + pthread_mutex_lock(&c->mutex); + fifo_size = ring_size(ring); fifo_size_of_read_back = ring_size_of_read_back(ring); if (new_logical_pos == c->logical_pos) { /* current position */ + pthread_mutex_unlock(&c->mutex); return c->logical_pos; } else if ((new_logical_pos >= (c->logical_pos - fifo_size_of_read_back)) && (new_logical_pos < (c->logical_pos + fifo_size + SHORT_SEEK_THRESHOLD))) { @@ -432,24 +428,28 @@ static int64_t async_seek(URLContext *h, int64_t pos, int whence) if (pos_delta > 0) { // fast seek forwards - async_read_internal(h, NULL, pos_delta); + pthread_mutex_unlock(&c->mutex); + ret = async_read_internal(h, NULL, pos_delta); + if (ret < 0) + return ret; } else { // fast seek backwards ring_drain(ring, pos_delta); c->logical_pos = new_logical_pos; + pthread_mutex_unlock(&c->mutex); } return c->logical_pos; } else if (c->logical_size <= 0) { /* can not seek */ + pthread_mutex_unlock(&c->mutex); return AVERROR(EINVAL); } else if (new_logical_pos > c->logical_size) { /* beyond end */ + pthread_mutex_unlock(&c->mutex); return AVERROR(EINVAL); } - pthread_mutex_lock(&c->mutex); - c->seek_request = 1; c->seek_pos = new_logical_pos; c->seek_whence = SEEK_SET; -- 2.54.0 _______________________________________________ ffmpeg-devel mailing list -- [email protected] To unsubscribe send an email to [email protected]
