perf(downloads): index queued positions for scheduling

This commit is contained in:
zarzet committed 2026-10-08 17:51:56 +07:00
1 parent 5a4d0716db
commit e4b6156249
6 files changed
+373 -5

No files matched your search

@@ -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) {
@@ -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<int> _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 = <Completer<void>>[];
final started = <String>[];
final scheduler = DownloadScheduler<DownloadItem>(
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<DownloadItem>.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<void>();
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<String, int> _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<DownloadItem>.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<DownloadItem>.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 = <Map<String, dynamic>>[];
setUpAll(() => Future<void>.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 = <String, List<int>>{'scan': [], 'indexed': []};
final indexCosts = <Map<String, int>>[];
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,
});
});
}
}
}
+2 -4
View File
@@ -2025,10 +2025,8 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
_startMultiProgressPolling();
try {
await DownloadScheduler<DownloadItem>(
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,
+94
View File
@@ -128,6 +128,79 @@ class _QueueItemIds extends ListBase<String> {
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<int> words;
final int firstWord;
factory _QueuedPositions.fromWords(List<int> words) =>
_QueuedPositions._(ChunkedList<int>.from(words), _findFirst(words, 0));
static int _findFirst(List<int> words, int start) {
for (var index = start; index < words.length; index++) {
if (words[index] != 0) return index;
}
return -1;
}
Iterable<int> 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<DownloadItem> previousItems,
List<DownloadItem> nextItems,
Set<int> changedIndices,
) {
Map<int, int>? 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<int>.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<int>.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<DownloadItem> 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<DownloadItem> _items;
final int? _firstDownloadingIndex;
final int? _firstFinalizingIndex;
final _QueuedPositions _queuedPositions;
final Map<String, List<int>>? _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<DownloadItem> items,
required int? firstDownloadingIndex,
required int? firstFinalizingIndex,
required _QueuedPositions queuedPositions,
Map<String, List<int>>? 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<int>.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,
);
}
+4
View File
@@ -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,
@@ -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<DownloadItem>.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<DownloadItem>.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(