perf(library): prefetch network metadata with bounded concurrency

Read up to three WebDAV tracks or two SMB tracks through a scan-owned metadata service while retaining listing order, bounded output, artwork and progress. Keep pause and cancellation independent of playback and add host regressions plus a native Android latency benchmark.
This commit is contained in:
zarzet committed 2026-10-10 23:22:41 +07:00
1 parent 0d70e413ed
commit 5eafd3acc4
6 files changed
+637 -28

No files matched your search

+79 -21
View File
@@ -1,3 +1,4 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:isolate';
@@ -13,7 +14,7 @@ import 'package:spotiflac_android/utils/logger.dart';
final _log = AppLogger('LocalLibrary');
/// A single tag read at a time; only NDJSON and artwork are stored locally.
/// Bounded tag prefetch; only NDJSON and artwork are stored locally.
/// Failed listings never count as empty directories. Partial scans retain old
/// rows through the same ingestion policy used for local/SAF scans.
class NetworkLibraryScanner {
@@ -23,19 +24,19 @@ class NetworkLibraryScanner {
Future<Directory> Function()? supportDirectory,
Future<Directory> Function()? temporaryDirectory,
Future<RandomAccessFile> Function(File)? openOutput,
int? metadataConcurrency,
}) : _storage = storage ?? NetworkStorageService.instance,
_readMetadata =
readMetadata ??
((source) => NetworkMetadataService.instance.read(
source,
forceRefresh: true,
)),
_readMetadata = readMetadata,
_metadataConcurrency = metadataConcurrency == null
? null
: RangeError.checkValueInInterval(metadataConcurrency, 1, 4),
_supportDirectory = supportDirectory ?? getApplicationSupportDirectory,
_temporaryDirectory = temporaryDirectory ?? getTemporaryDirectory,
_openOutput = openOutput ?? ((file) => file.open(mode: FileMode.write));
final NetworkStorageService _storage;
final Future<Map<String, dynamic>> Function(String) _readMetadata;
final Future<Map<String, dynamic>> Function(String)? _readMetadata;
final int? _metadataConcurrency;
final Future<Directory> Function() _supportDirectory, _temporaryDirectory;
final Future<RandomAccessFile> Function(File) _openOutput;
@@ -76,6 +77,21 @@ class NetworkLibraryScanner {
final pending = <String>[root];
final visited = <String>{};
final seenFiles = <String>{};
// Leave room for playback and avoid saturating a small NAS/router.
final concurrency =
_metadataConcurrency ??
(connection.protocol == NetworkProtocol.smb ? 2 : 3);
final metadata = _readMetadata == null
? NetworkMetadataService(
storage: _storage,
maxConcurrentReads: concurrency,
cacheDirectory: _temporaryDirectory,
)
: null;
final read =
_readMetadata ??
((String source) => metadata!.read(source, forceRefresh: true));
final prefetched = <_PendingNetworkTrack>[];
try {
final writer = await _openOutput(file);
output = writer;
@@ -115,23 +131,51 @@ class NetworkLibraryScanner {
'${entries.where((entry) => entry.directory).length} folders, '
'${entries.where((entry) => entry.audio).length} audio files',
);
for (final entry in entries) {
await checkpoint();
final path = networkPath(entry.path);
// Reject provider parent links and paths outside this folder.
if (!path.startsWith(folder) || path == folder) continue;
if (entry.directory) {
if (!path.endsWith('/')) {
throw const FormatException('Invalid folder entry');
final iterator = entries.iterator;
Future<void> fill() async {
while (prefetched.length < concurrency && iterator.moveNext()) {
await checkpoint();
final entry = iterator.current;
final path = networkPath(entry.path);
// Reject provider parent links and paths outside this folder.
if (!path.startsWith(folder) || path == folder) continue;
if (entry.directory) {
if (!path.endsWith('/')) {
throw const FormatException('Invalid folder entry');
}
pending.add(path);
continue;
}
pending.add(path);
continue;
if (!entry.audio || !seenFiles.add(path)) continue;
final trackSource = NetworkStorageService.source(
connection.id,
path,
);
// Capture failures immediately: a later result can fail while the
// first read is still running. Consume results in listing order.
final result = Future.sync(() => read(trackSource))
.then<_NetworkTags>(
(tags) => (tags: tags, error: null, stack: null),
onError: (Object error, StackTrace stack) =>
(tags: null, error: error, stack: stack),
);
prefetched.add((entry: entry, source: trackSource, result: result));
}
if (!entry.audio || !seenFiles.add(path)) continue;
final trackSource = NetworkStorageService.source(connection.id, path);
}
await fill();
while (prefetched.isNotEmpty) {
await checkpoint();
final next = prefetched.removeAt(0);
final entry = next.entry;
final trackSource = next.source;
String? encodedRow;
try {
final tags = await _readMetadata(trackSource);
final result = await next.result;
if (result.error != null) {
Error.throwWithStackTrace(result.error!, result.stack!);
}
final tags = result.tags!;
if (tags['error'] != null || tags['metadataFromFilename'] == true) {
throw StateError(
tags['error']?.toString() ?? 'Network tags unavailable',
@@ -196,6 +240,7 @@ class NetworkLibraryScanner {
}
processed++;
onProgress(processed, errors, entry.name);
await fill();
}
}
await checkpoint();
@@ -214,6 +259,8 @@ class NetworkLibraryScanner {
errorCount: errors,
);
} catch (_) {
await metadata?.cancelPending();
await Future.wait(prefetched.map((track) => track.result));
try {
await output?.close();
} catch (error) {
@@ -229,6 +276,17 @@ class NetworkLibraryScanner {
}
}
typedef _NetworkTags = ({
Map<String, dynamic>? tags,
Object? error,
StackTrace? stack,
});
typedef _PendingNetworkTrack = ({
NetworkEntry entry,
String source,
Future<_NetworkTags> result,
});
// Keep the isolate closure outside scan's scope so it cannot capture an open
// output handle or the buffering closures, which are not isolate-sendable.
Future<String?> _promoteCoverInBackground(String cover, String directory) =>
+51 -7
View File
@@ -19,13 +19,18 @@ class NetworkMetadataService {
NetworkStorageService? storage,
NetworkTagReader? reader,
Future<Directory> Function()? cacheDirectory,
int maxConcurrentReads = 1,
}) : _storage = storage ?? NetworkStorageService.instance,
_useNativeReader = reader == null,
_reader =
reader ??
((url, name) =>
PlatformBridge.readFileMetadata(url, displayName: name)),
_directory = cacheDirectory ?? getTemporaryDirectory;
_directory = cacheDirectory ?? getTemporaryDirectory,
_lanes = List.generate(
RangeError.checkValueInInterval(maxConcurrentReads, 1, 4),
(_) => (tail: Future<void>.value(), pending: 0),
);
static final instance = NetworkMetadataService();
final NetworkStorageService _storage;
@@ -34,12 +39,17 @@ class NetworkMetadataService {
final Future<Directory> Function() _directory;
final _cache = <String, ({DateTime time, Map<String, dynamic> value})>{};
final _pending = <String, Future<Map<String, dynamic>>>{};
Future<void> _tail = Future.value();
final List<({Future<void> tail, int pending})> _lanes;
final _requestIds = <String>{};
bool _stopped = false;
Future<Map<String, dynamic>> read(
String source, {
bool forceRefresh = false,
}) {
if (_stopped) {
return Future.error(StateError('Network metadata reader stopped'));
}
if (forceRefresh) _cache.remove(source);
final cached = _cache[source];
if (cached != null &&
@@ -51,16 +61,44 @@ class NetworkMetadataService {
if (_pending.length >= 4) {
return Future.error(StateError('Network metadata reader is busy'));
}
final operation = _tail.then((_) => _load(source));
_tail = operation.then<void>((_) {}, onError: (Object _, StackTrace _) {});
var lane = 0;
for (var i = 1; i < _lanes.length; i++) {
if (_lanes[i].pending < _lanes[lane].pending) lane = i;
}
final requestId = 'network_metadata_${NetworkStorageService.newId()}';
_requestIds.add(requestId);
final operation = _lanes[lane].tail.then((_) => _load(source, requestId));
_lanes[lane] = (
tail: operation.then<void>((_) {}, onError: (Object _, StackTrace _) {}),
pending: _lanes[lane].pending + 1,
);
final result = operation.whenComplete(() {
_pending.remove(source);
_requestIds.remove(requestId);
_lanes[lane] = (
tail: _lanes[lane].tail,
pending: _lanes[lane].pending - 1,
);
});
_pending[source] = result;
return result;
}
Future<Map<String, dynamic>> _load(String source) async {
/// Scan-owned readers can stop their work without cancelling playback reads.
Future<void> cancelPending() async {
_stopped = true;
await Future.wait([
for (final id in _requestIds.toList())
PlatformBridge.cancelNativeDataJob(id).catchError((Object _) {}),
]);
}
void _checkStopped() {
if (_stopped) throw StateError('Network metadata reader stopped');
}
Future<Map<String, dynamic>> _load(String source, String requestId) async {
_checkStopped();
final uri = Uri.parse(source);
if (uri.scheme != 'network') {
throw const FormatException('Expected a network source');
@@ -76,21 +114,26 @@ class NetworkMetadataService {
name = '${name.substring(0, name.length - 5)}.m4a';
}
final url = await _storage.resolve(source);
_checkStopped();
Map<String, dynamic>? loaded;
if (_useNativeReader) {
try {
final directory = await _directory();
_checkStopped();
loaded = await PlatformBridge.runNativeDataJob({
'operation': 'network_metadata',
'url': url,
'display_name': name,
'cache_directory': '${(await _directory()).path}/network_metadata',
});
'cache_directory': '${directory.path}/network_metadata',
}, requestId: requestId);
} on MissingPluginException {
// Test/unsupported hosts retain the same reader on an isolate.
}
}
if (loaded == null) {
_checkStopped();
final tags = await _reader(url, name);
_checkStopped();
String directory = '';
if (tags['cover_base64'] is String &&
(tags['cover_base64'] as String).isNotEmpty) {
@@ -104,6 +147,7 @@ class NetworkMetadataService {
'metadata': tags,
});
}
_checkStopped();
final result = loaded ?? (throw StateError('Network metadata is empty'));
final metadata = Map<String, dynamic>.from(result['metadata'] as Map);
if (metadata['error']?.toString().isNotEmpty == true) {