diff --git a/integration_test/album_replaygain_completion_test.dart b/integration_test/album_replaygain_completion_test.dart index e384207a..1de0f2f5 100644 --- a/integration_test/album_replaygain_completion_test.dart +++ b/integration_test/album_replaygain_completion_test.dart @@ -324,7 +324,12 @@ registerExtension({ next.items .where((item) => item.status == DownloadStatus.finalizing) .firstOrNull, - ); + ) && + orderedEquals( + next.items.where( + (item) => item.status == DownloadStatus.queued, + ), + ).matches(next.queuedItems, {}); for (final item in next.items) { final transitions = states.putIfAbsent(item.track.id, () => []); if (transitions.lastOrNull != item.status.name) { diff --git a/integration_test/queue_candidate_selection_test.dart b/integration_test/queue_candidate_selection_test.dart new file mode 100644 index 00000000..717afe1f --- /dev/null +++ b/integration_test/queue_candidate_selection_test.dart @@ -0,0 +1,171 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:integration_test/integration_test.dart'; +import 'package:spotiflac_android/models/download_item.dart'; +import 'package:spotiflac_android/providers/download_queue_state.dart'; +import 'package:spotiflac_android/services/download_scheduler.dart'; +import 'package:spotiflac_android/utils/chunked_list.dart'; + +import '../test/support/queue_removal_benchmark.dart'; + +Future _measure( + DownloadQueueState initial, { + required bool indexed, +}) async { + final expected = initial.items + .where((item) => item.status == DownloadStatus.queued) + .map((item) => item.id) + .toList(); + final watch = Stopwatch(); + for (var wave = 0; wave < 100; wave++) { + var state = initial; + var paused = false; + final work = >[]; + final started = []; + final scheduler = DownloadScheduler( + queuedItems: () => indexed + ? state.queuedItems + : state.items.where((item) => item.status == DownloadStatus.queued), + idOf: (item) => item.id, + isPaused: () => paused, + concurrency: () => 3, + start: (item) { + started.add(item.id); + final previous = state.items; + final index = state.lookup.indexByItemId[item.id]!; + final next = ChunkedList.from( + previous, + ).updated({index: item.copyWith(status: DownloadStatus.downloading)}); + state = state.copyWith( + items: next, + lookup: state.lookup.updatedForIndices( + previousItems: previous, + nextItems: next, + changedIndices: [index], + ), + ); + final pending = Completer(); + work.add(pending); + return pending.future; + }, + ); + watch.start(); + final run = scheduler.run(); + watch.stop(); + expect(started, expected); + expect(work, hasLength(3)); + expect(state.activeDownloadsCount, 3); + expect(state.queuedItems, isEmpty); + paused = true; + for (final pending in work) { + pending.complete(); + } + await run; + } + return watch.elapsedMicroseconds; +} + +Map _indexCosts(DownloadQueueState initial) { + final build = Stopwatch()..start(); + var state = initial; + for (var i = 0; i < 20; i++) { + state = state.copyWith(items: state.items); + } + build.stop(); + final activeIndices = [ + for (var i = 0; i < state.items.length; i++) + if (state.items[i].status == DownloadStatus.queued) i, + ]; + final active = ChunkedList.from(state.items).updated({ + for (final index in activeIndices) + index: state.items[index].copyWith(status: DownloadStatus.downloading), + }); + state = state.copyWith( + items: active, + lookup: state.lookup.updatedForIndices( + previousItems: state.items, + nextItems: active, + changedIndices: activeIndices, + ), + ); + final progress = Stopwatch()..start(); + for (var tick = 0; tick < 5000; tick++) { + final index = activeIndices[tick % activeIndices.length]; + final next = ChunkedList.from( + state.items, + ).updated({index: state.items[index].copyWith(progress: tick / 5000)}); + state = state.copyWith( + items: next, + lookup: state.lookup.updatedForIndices( + previousItems: state.items, + nextItems: next, + changedIndices: [index], + ), + ); + } + progress.stop(); + expect(state.items.length, initial.items.length); + return { + 'build_20_us': build.elapsedMicroseconds, + 'progress_5000_us': progress.elapsedMicroseconds, + }; +} + +void main() { + final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized(); + final cases = >[]; + setUpAll(() => Future.delayed(const Duration(seconds: 2))); + tearDownAll(() { + binding.reportData ??= {}; + binding.reportData!['queue_candidate_selection'] = { + 'platform': Platform.operatingSystem, + 'completed_cases': cases.length, + 'note': + 'Five alternating scan/indexed pairs after warmup: 100 actual scheduler admissions of three items including immutable status/index updates. Both variants maintain current indexes. Setup, async work/drain and rendering excluded. Separate full-index-build and progress-only costs compared to retained pre-index baseline. iOS debug correctness only.', + 'cases': cases, + }; + }); + for (final count in [1000, 10000]) { + for (final pattern in ['front', 'tail', 'spread']) { + testWidgets('queued candidates $count/$pattern', (tester) async { + final fixture = await queueRemovalFixture(count); + final queued = switch (pattern) { + 'front' => {0, 1, 2}, + 'tail' => {count - 3, count - 2, count - 1}, + _ => {0, count ~/ 2, count - 1}, + }; + final initial = fixture.copyWith( + items: [ + for (var i = 0; i < count; i++) + fixture.items[i].copyWith( + status: queued.contains(i) + ? DownloadStatus.queued + : DownloadStatus.completed, + ), + ], + ); + await _measure(initial, indexed: false); + await _measure(initial, indexed: true); + _indexCosts(initial); + final samples = >{'scan': [], 'indexed': []}; + final indexCosts = >[]; + for (var i = 0; i < 5; i++) { + for (final indexed in i.isEven ? [false, true] : [true, false]) { + samples[indexed ? 'indexed' : 'scan']!.add( + await _measure(initial, indexed: indexed), + ); + } + indexCosts.add(_indexCosts(initial)); + } + cases.add({ + 'queue_size': count, + 'pattern': pattern, + 'admission_100_us': samples, + 'index_costs': indexCosts, + }); + }); + } + } +} diff --git a/lib/providers/download_queue_provider.dart b/lib/providers/download_queue_provider.dart index 61d28a54..6d3b652a 100644 --- a/lib/providers/download_queue_provider.dart +++ b/lib/providers/download_queue_provider.dart @@ -2025,10 +2025,8 @@ class DownloadQueueNotifier extends Notifier { _startMultiProgressPolling(); try { await DownloadScheduler( - queuedItems: () => state.items.where( - (item) => - item.status == DownloadStatus.queued && - !_pausePendingItemIds.contains(item.id), + queuedItems: () => state.queuedItems.where( + (item) => !_pausePendingItemIds.contains(item.id), ), idOf: (item) => item.id, isPaused: () => state.isPaused, diff --git a/lib/providers/download_queue_state.dart b/lib/providers/download_queue_state.dart index 6e2315ee..30a1775d 100644 --- a/lib/providers/download_queue_state.dart +++ b/lib/providers/download_queue_state.dart @@ -128,6 +128,79 @@ class _QueueItemIds extends ListBase { throw UnsupportedError('Immutable queue IDs'); } +/// One bit per queued position. Sparse status changes copy only the affected +/// word chunks; progress-only updates share the index. This also skips gaps +/// between candidates without rescanning DownloadItem objects during launch. +class _QueuedPositions { + const _QueuedPositions.empty() : words = const [], firstWord = -1; + _QueuedPositions._(this.words, this.firstWord); + final List words; + final int firstWord; + + factory _QueuedPositions.fromWords(List words) => + _QueuedPositions._(ChunkedList.from(words), _findFirst(words, 0)); + + static int _findFirst(List words, int start) { + for (var index = start; index < words.length; index++) { + if (words[index] != 0) return index; + } + return -1; + } + + Iterable get indices sync* { + if (firstWord < 0) return; + for (var index = firstWord; index < words.length; index++) { + var remaining = words[index]; + while (remaining != 0) { + final lowestBit = remaining & -remaining; + yield (index << 5) + lowestBit.bitLength - 1; + remaining ^= lowestBit; + } + } + } + + _QueuedPositions updated( + List previousItems, + List nextItems, + Set changedIndices, + ) { + Map? changes; + for (final index in changedIndices) { + final wasQueued = previousItems[index].status == DownloadStatus.queued; + final isQueued = nextItems[index].status == DownloadStatus.queued; + if (wasQueued == isQueued) continue; + changes ??= {}; + final word = index >> 5; + final bit = 1 << (index & 31); + final value = changes[word] ?? words[word]; + changes[word] = isQueued ? value | bit : value & ~bit; + } + if (changes == null) return this; + final next = ChunkedList.from(words).updated(changes); + var first = firstWord; + for (final entry in changes.entries) { + if (entry.value != 0 && (first < 0 || entry.key < first)) { + first = entry.key; + } + } + if (first >= 0 && next[first] == 0) first = _findFirst(next, first + 1); + return _QueuedPositions._(next, first); + } + + _QueuedPositions removing(int index, int nextLength) { + final shifted = List.of(words); + final first = index >> 5; + for (var word = first; word < words.length; word++) { + final lowMask = word == first ? (1 << (index & 31)) - 1 : 0; + final carry = word + 1 < words.length ? (words[word + 1] & 1) << 31 : 0; + shifted[word] = + (words[word] & lowMask) | ((words[word] >> 1) & ~lowMask) | carry; + } + if (shifted.length > ((nextLength + 31) >> 5)) shifted.removeLast(); + return _QueuedPositions.fromWords(shifted); + } +} + /// Immutable queue state shared by the notifier and read-only UI consumers. /// /// Keeping this model outside the notifier implementation makes queue state @@ -206,6 +279,12 @@ class DownloadQueueState { int get activeDownloadsCount => items.isEmpty ? 0 : lookup.activeDownloadsCount; + /// Ordered candidates from this snapshot. Raw/custom states without a + /// matching immutable lookup retain the ordinary status-filter behavior. + Iterable get queuedItems => identical(items, lookup._items) + ? lookup._queuedPositions.indices.map((index) => items[index]) + : items.where((item) => item.status == DownloadStatus.queued); + DownloadQueueState withoutItem(String id) { final index = lookup.indexByItemId[id]; if (lookup.indexByItemId.length != items.length || @@ -251,6 +330,7 @@ class DownloadQueueLookup { final List _items; final int? _firstDownloadingIndex; final int? _firstFinalizingIndex; + final _QueuedPositions _queuedPositions; final Map>? _duplicateTrackPositions; DownloadItem? get firstDownloading => @@ -272,6 +352,7 @@ class DownloadQueueLookup { _items = const [], _firstDownloadingIndex = null, _firstFinalizingIndex = null, + _queuedPositions = const _QueuedPositions.empty(), _duplicateTrackPositions = null; DownloadQueueLookup._({ @@ -288,10 +369,12 @@ class DownloadQueueLookup { required List items, required int? firstDownloadingIndex, required int? firstFinalizingIndex, + required _QueuedPositions queuedPositions, Map>? duplicateTrackPositions, }) : _items = items, _firstDownloadingIndex = firstDownloadingIndex, _firstFinalizingIndex = firstFinalizingIndex, + _queuedPositions = queuedPositions, _duplicateTrackPositions = duplicateTrackPositions; static int? _firstAfterRemoval( @@ -429,6 +512,7 @@ class DownloadQueueLookup { finalizingCount, nextItems, ), + queuedPositions: _queuedPositions.removing(index, nextItems.length), duplicateTrackPositions: duplicates, ); } @@ -445,6 +529,7 @@ class DownloadQueueLookup { var finalizingCount = 0; int? firstDownloadingIndex; int? firstFinalizingIndex; + final queuedWords = List.filled((items.length + 31) >> 5, 0); for (var index = 0; index < items.length; index++) { final item = items[index]; byTrackIndex.putIfAbsent(item.track.id, () => index); @@ -454,6 +539,9 @@ class DownloadQueueLookup { notCompletedItemIds.add(item.id); } if (_countsAsQueued(item.status)) queuedCount++; + if (item.status == DownloadStatus.queued) { + queuedWords[index >> 5] |= 1 << (index & 31); + } if (item.status == DownloadStatus.completed) completedCount++; if (item.status == DownloadStatus.failed) failedCount++; if (item.status == DownloadStatus.downloading) { @@ -481,6 +569,7 @@ class DownloadQueueLookup { items: snapshot, firstDownloadingIndex: firstDownloadingIndex, firstFinalizingIndex: firstFinalizingIndex, + queuedPositions: _QueuedPositions.fromWords(queuedWords), ); } @@ -604,6 +693,11 @@ class DownloadQueueLookup { snapshot, normalizedChanged, ), + queuedPositions: _queuedPositions.updated( + previousItems, + snapshot, + normalizedChanged, + ), duplicateTrackPositions: _duplicateTrackPositions, ); } diff --git a/test/download_queue_removal_test.dart b/test/download_queue_removal_test.dart index d98ddf0d..2af76a20 100644 --- a/test/download_queue_removal_test.dart +++ b/test/download_queue_removal_test.dart @@ -39,6 +39,10 @@ void _check(DownloadQueueState state) { expect(actual.failedCount, expected.failedCount); expect(actual.activeDownloadsCount, expected.activeDownloadsCount); expect(actual.finalizingCount, expected.finalizingCount); + expect( + state.queuedItems, + state.items.where((item) => item.status == DownloadStatus.queued), + ); for (final entry in { DownloadStatus.downloading: actual.firstDownloading, DownloadStatus.finalizing: actual.firstFinalizing, diff --git a/test/download_queue_sparse_update_test.dart b/test/download_queue_sparse_update_test.dart index 8621ffb2..78c0ae2f 100644 --- a/test/download_queue_sparse_update_test.dart +++ b/test/download_queue_sparse_update_test.dart @@ -40,6 +40,94 @@ DownloadItem _item(int index) => DownloadItem( void main() { TestWidgetsFlutterBinding.ensureInitialized(); + test( + 'queued positions span word/chunk boundaries and preserve iterables', + () { + final queued = { + 0, + 1, + 30, + 31, + 32, + 33, + 63, + 64, + 65, + 1023, + 1024, + 2047, + 2048, + 2099, + }; + var state = const DownloadQueueState().copyWith( + items: [ + for (var i = 0; i < 2100; i++) + _item(i).copyWith( + status: queued.contains(i) + ? DownloadStatus.queued + : DownloadStatus.completed, + ), + ], + ); + final old = state; + final captured = state.queuedItems; + final original = state.items + .where((item) => item.status == DownloadStatus.queued) + .toList(); + for (final removed in [31, 32, 0, 2048, 2047, 33, 1024, 2099, 500, 64]) { + state = state.withoutItem('item-$removed'); + expect( + state.queuedItems, + state.items.where((item) => item.status == DownloadStatus.queued), + ); + expect(captured, original); + expect(old.queuedItems, original); + } + final previous = state.items; + final next = ChunkedList.from(previous).updated({ + for (var i = 0; i < previous.length; i++) + i: previous[i].copyWith(status: DownloadStatus.completed), + }); + state = state.copyWith( + items: next, + lookup: state.lookup.updatedForIndices( + previousItems: previous, + nextItems: next, + changedIndices: List.generate(previous.length, (i) => i), + ), + ); + expect(state.queuedItems, isEmpty); + final restored = ChunkedList.from(next).updated({ + for (final i in [31, 32, 63, 64, 2047, 2048]) + i: next[i].copyWith(status: DownloadStatus.queued), + }); + state = state.copyWith( + items: restored, + lookup: state.lookup.updatedForIndices( + previousItems: next, + nextItems: restored, + changedIndices: [2048, 31, 64, 32, 2047, 63], + ), + ); + expect(state.queuedItems, [ + for (final i in [31, 32, 63, 64, 2047, 2048]) restored[i], + ]); + expect(captured, original); + }, + ); + + test('raw and mismatched queue states keep ordinary status filtering', () { + final items = [_item(0), _item(1).copyWith(status: DownloadStatus.queued)]; + final raw = DownloadQueueState(items: items); + expect(raw.queuedItems, [items[1]]); + final unrelated = DownloadQueueLookup.fromItems([_item(3), _item(4)]); + final mismatched = DownloadQueueState(items: items, lookup: unrelated); + expect(mismatched.queuedItems, [items[1]]); + items[0] = items[0].copyWith(status: DownloadStatus.queued); + expect(raw.queuedItems, items); + expect(mismatched.queuedItems, items); + }); + test( 'sparse updates preserve snapshots and do not revisit the source list', () { @@ -100,6 +188,10 @@ void main() { expect(state.lookup.finalizingCount, rebuilt.finalizingCount); expect(state.lookup.firstDownloading, same(rebuilt.firstDownloading)); expect(state.lookup.firstFinalizing, same(rebuilt.firstFinalizing)); + expect( + state.queuedItems, + state.items.where((item) => item.status == DownloadStatus.queued), + ); expect(state.lookup.notCompletedItemIds, rebuilt.notCompletedItemIds); expect( old.lookup.byItemId[old.items[index].id], @@ -119,6 +211,10 @@ void main() { 'notification candidates follow batch transitions and keep old values', () { void check(DownloadQueueState state) { + expect( + state.queuedItems, + state.items.where((item) => item.status == DownloadStatus.queued), + ); expect( state.lookup.firstDownloading, same(