From 2f317e778eb6551fe1041942d2475cebb54ce60b Mon Sep 17 00:00:00 2001 From: Neeraj Gupta Date: Sat, 29 Aug 2026 06:01:25 +0530 Subject: [PATCH] Add read-ahead random access source --- README.md | 2 + lib/random_access_source.dart | 1 + lib/src/read_ahead_ra_source.dart | 128 +++++++++++++++ test/read_ahead_test.dart | 250 ++++++++++++++++++++++++++++++ 4 files changed, 381 insertions(+) create mode 100644 lib/src/read_ahead_ra_source.dart create mode 100644 test/read_ahead_test.dart diff --git a/README.md b/README.md index 0b8b494..2a20e86 100644 --- a/README.md +++ b/README.md @@ -54,3 +54,5 @@ Implementations: - Use `FileRASource` for `File` (`dart:io`) and `Blob` (`package:web`). - `await FileRASource.openPath(path)`: Opens a `FileRASource` from a file path. - `await FileRASource.loadFile(file)`: Loads a `FileRASource` from a `PlatformFile`. +- Wrap a source with `ReadAheadRASource` to coalesce small, nearby reads. + - `ReadAheadRASource(source)`: Uses a 4 KiB read-ahead buffer. diff --git a/lib/random_access_source.dart b/lib/random_access_source.dart index 0fd1d11..7863371 100644 --- a/lib/random_access_source.dart +++ b/lib/random_access_source.dart @@ -4,3 +4,4 @@ library; export 'src/bytes_ra_source.dart'; export 'src/file_ra_source.dart'; export 'src/random_access_source.dart'; +export 'src/read_ahead_ra_source.dart'; diff --git a/lib/src/read_ahead_ra_source.dart b/lib/src/read_ahead_ra_source.dart new file mode 100644 index 0000000..6abee50 --- /dev/null +++ b/lib/src/read_ahead_ra_source.dart @@ -0,0 +1,128 @@ +import 'dart:math' as math; +import 'dart:typed_data'; + +import 'random_access_source.dart'; + +/// Coalesces small reads from another [RandomAccessSource]. +/// +/// The wrapped source must not be accessed directly and is closed by [close]. +/// Its contents must remain unchanged while they are cached. +/// Operations on this stateful source must not overlap. +class ReadAheadRASource extends RandomAccessSource { + /// Wraps [source] with a read-ahead buffer. + ReadAheadRASource(RandomAccessSource source, {int bufferSize = 4096}) + : _source = source, + _bufferSize = bufferSize { + if (bufferSize <= 0) { + throw RangeError.value(bufferSize, 'bufferSize', 'Must be positive'); + } + } + + final RandomAccessSource _source; + final int _bufferSize; + + Uint8List _buffer = Uint8List(0); + int _bufferStart = 0; + int? _position; + bool _closed = false; + + @override + Future length() async { + _checkOpen(); + return await _source.length(); + } + + @override + Future readByte() async { + final bytes = await read(1); + return bytes.isEmpty ? -1 : bytes[0]; + } + + @override + Future read(int count) async { + _checkOpen(); + if (count < 0) { + throw RangeError.value(count, 'count', 'Must not be negative'); + } + if (count == 0) { + return Uint8List(0); + } + + final position = await _currentPosition(); + final bufferEnd = _bufferStart + _buffer.length; + if (position < _bufferStart || position + count > bufferEnd) { + await _source.seek(position); + final buffer = await _source.read(math.max(_bufferSize, count)); + _buffer = buffer; + _bufferStart = position; + } + + final start = position - _bufferStart; + final length = math.min(count, _buffer.length - start); + if (length <= 0) { + return Uint8List(0); + } + + _position = position + length; + return Uint8List.sublistView(_buffer, start, start + length); + } + + @override + Future readInto(List buffer, int offset, int count) async { + _checkOpen(); + RangeError.checkValidRange(offset, offset + count, buffer.length); + final position = await _currentPosition(); + final bytes = await read(count); + try { + buffer.setRange(offset, offset + bytes.length, bytes); + } catch (_) { + _position = position; + rethrow; + } + return bytes.length; + } + + @override + Future position() async { + _checkOpen(); + return await _currentPosition(); + } + + @override + Future seek(int position) async { + _checkOpen(); + if (position < 0) { + throw RangeError.value(position, 'position', 'Must not be negative'); + } + _position = position; + } + + @override + Future readToEnd() async { + _checkOpen(); + final position = await _currentPosition(); + await _source.seek(position); + final bytes = await _source.readToEnd(); + _position = position + bytes.length; + return bytes; + } + + @override + Future close() async { + if (_closed) { + return; + } + _closed = true; + _buffer = Uint8List(0); + await _source.close(); + } + + Future _currentPosition() async => + _position ??= await _source.position(); + + void _checkOpen() { + if (_closed) { + throw StateError('Source is closed'); + } + } +} diff --git a/test/read_ahead_test.dart b/test/read_ahead_test.dart new file mode 100644 index 0000000..56047d7 --- /dev/null +++ b/test/read_ahead_test.dart @@ -0,0 +1,250 @@ +import 'dart:math' as math; +import 'dart:typed_data'; + +import 'package:random_access_source/random_access_source.dart'; +import 'package:test/test.dart'; + +void main() { + test('Coalesces reads and seeks within the buffer', () async { + final source = _TrackingSource(_bytes(12)); + final buffered = ReadAheadRASource(source, bufferSize: 4); + + expect(await buffered.readByte(), 0); + expect(await buffered.read(2), [1, 2]); + await buffered.seek(1); + expect(await buffered.read(2), [1, 2]); + expect(source.reads, [4]); + expect(source.seeks, [0]); + + expect(await buffered.read(2), [3, 4]); + expect(source.reads, [4, 4]); + expect(source.seeks, [0, 3]); + }); + + test('Uses the requested size when it exceeds the buffer', () async { + final source = _TrackingSource(_bytes(12)); + final buffered = ReadAheadRASource(source, bufferSize: 4); + + expect(await buffered.read(6), [0, 1, 2, 3, 4, 5]); + expect(source.reads, [6]); + }); + + test('Starts at the wrapped source position', () async { + final source = _TrackingSource(_bytes(6)); + await source.seek(2); + source.seeks.clear(); + + final buffered = ReadAheadRASource(source, bufferSize: 2); + expect(await buffered.read(2), [2, 3]); + expect(await buffered.position(), 4); + expect(source.seeks, [2]); + }); + + test('A seek before the first read overrides the wrapped position', () async { + final source = _TrackingSource(_bytes(6)); + await source.seek(2); + final buffered = ReadAheadRASource(source, bufferSize: 2); + + await buffered.seek(4); + expect(await buffered.readByte(), 4); + }); + + test('Handles EOF and readToEnd', () async { + final source = _TrackingSource(_bytes(5)); + final buffered = ReadAheadRASource(source, bufferSize: 4); + + await buffered.seek(2); + expect(await buffered.readToEnd(), [2, 3, 4]); + expect(await buffered.position(), 5); + expect(await buffered.readByte(), -1); + expect(await buffered.position(), 5); + }); + + test('Handles buffer boundaries and positions past EOF', () async { + final source = _TrackingSource(_bytes(5)); + final buffered = ReadAheadRASource(source, bufferSize: 4); + + expect(await buffered.read(4), [0, 1, 2, 3]); + expect(await buffered.readByte(), 4); + expect(await buffered.readByte(), -1); + await buffered.seek(8); + expect(await buffered.readByte(), -1); + expect(await buffered.position(), 8); + }); + + test('Advances by the bytes returned from a short read', () async { + final source = _TrackingSource(_bytes(6), maxRead: 2); + final buffered = ReadAheadRASource(source, bufferSize: 4); + + expect(await buffered.read(4), [0, 1]); + expect(await buffered.position(), 2); + expect(await buffered.read(4), [2, 3]); + expect(await buffered.position(), 4); + }); + + test('Keeps position and cache when a refill fails', () async { + final source = _TrackingSource(_bytes(8)); + final buffered = ReadAheadRASource(source, bufferSize: 4); + + expect(await buffered.read(2), [0, 1]); + await buffered.seek(6); + source.failNextRead = true; + await expectLater(buffered.readByte(), throwsStateError); + expect(await buffered.position(), 6); + + await buffered.seek(1); + expect(await buffered.readByte(), 1); + await buffered.seek(6); + expect(await buffered.readByte(), 6); + expect(source.reads, [4, 4]); + }); + + test('Validates arguments without changing position', () async { + final source = _TrackingSource(_bytes(4)); + final buffered = ReadAheadRASource(source); + + await expectLater(buffered.read(-1), throwsRangeError); + await expectLater(buffered.seek(-1), throwsRangeError); + for (final arguments in [(-1, 1), (0, -1), (1, 2)]) { + await expectLater( + buffered.readInto(Uint8List(2), arguments.$1, arguments.$2), + throwsRangeError, + ); + } + expect(await buffered.read(0), isEmpty); + expect(await buffered.position(), 0); + expect(source.reads, isEmpty); + }); + + test('Reads into a range and stops at EOF', () async { + final source = _TrackingSource(_bytes(4)); + final buffered = ReadAheadRASource(source); + final destination = Uint8List.fromList([9, 9, 9, 9]); + + await buffered.seek(3); + expect(await buffered.readInto(destination, 1, 3), 1); + expect(destination, [9, 3, 9, 9]); + expect(await buffered.position(), 4); + }); + + test('Restores position when the destination rejects a write', () async { + final source = _TrackingSource(_bytes(4)); + final buffered = ReadAheadRASource(source); + final destination = List.unmodifiable([0]); + + await expectLater( + buffered.readInto(destination, 0, 1), + throwsUnsupportedError, + ); + expect(await buffered.position(), 0); + expect(await buffered.readByte(), 0); + expect(source.reads, [4096]); + }); + + test('Closes the wrapped source', () async { + final source = _TrackingSource(_bytes(2)); + final buffered = ReadAheadRASource(source); + + await buffered.close(); + await buffered.close(); + expect(source.closeCalls, 1); + + final operations = Function()>[ + () async { + await buffered.length(); + }, + () async { + await buffered.readByte(); + }, + () async { + await buffered.read(0); + }, + () async { + await buffered.readInto(Uint8List(0), 0, 0); + }, + () async { + await buffered.position(); + }, + () async { + await buffered.seek(0); + }, + () async { + await buffered.readToEnd(); + }, + ]; + for (final operation in operations) { + await expectLater(operation(), throwsStateError); + } + }); + + test('Requires a positive buffer size', () { + for (final size in [0, -1]) { + expect( + () => ReadAheadRASource(_TrackingSource(_bytes(1)), bufferSize: size), + throwsRangeError, + ); + } + }); + + test('Keeps returned bytes valid after a refill', () async { + final source = _TrackingSource(_bytes(6)); + final buffered = ReadAheadRASource(source, bufferSize: 2); + + final first = await buffered.read(2); + expect(await buffered.read(2), [2, 3]); + expect(first, [0, 1]); + }); +} + +Uint8List _bytes(int length) => + Uint8List.fromList(List.generate(length, (index) => index)); + +class _TrackingSource extends RandomAccessSource { + _TrackingSource(Uint8List bytes, {this.maxRead}) + : _source = BytesRASource(bytes); + + final BytesRASource _source; + final int? maxRead; + final List reads = []; + final List seeks = []; + bool failNextRead = false; + int closeCalls = 0; + + @override + Future close() async { + closeCalls++; + await _source.close(); + } + + @override + Future length() => _source.length(); + + @override + Future position() => _source.position(); + + @override + Future read(int count) async { + if (failNextRead) { + failNextRead = false; + throw StateError('Read failed'); + } + reads.add(count); + return _source.read(maxRead == null ? count : math.min(count, maxRead!)); + } + + @override + Future readByte() => _source.readByte(); + + @override + Future readInto(List buffer, int offset, int count) => + _source.readInto(buffer, offset, count); + + @override + Future readToEnd() => _source.readToEnd(); + + @override + Future seek(int position) async { + seeks.add(position); + await _source.seek(position); + } +}