diff --git a/apps/mobile/lib/features/uploads/preparer.dart b/apps/mobile/lib/features/uploads/preparer.dart index 65b1598..dda6076 100644 --- a/apps/mobile/lib/features/uploads/preparer.dart +++ b/apps/mobile/lib/features/uploads/preparer.dart @@ -7,8 +7,8 @@ /// [UploadPreparer.retry]),lastError 携带客户端伪码([PrepErrorCodes])。 /// /// 设计取舍(对齐计划 Task 6 评审口径): -/// - **串行执行**:文件复制是最容易造成 IO 风暴的环节(连拍场景一次 -/// 落盘多张),队列化逐张处理;传输并发由 UploadEngine 另行控制; +/// - **有界并发执行(默认 3)**:文件复制仍是 IO 最重环节,上限钳位 +/// 防连拍(一次落盘多张)IO 风暴;传输并发由 UploadEngine 另行控制; /// - **磁盘空间不显式预检**:引入 disk_space 插件成本(新依赖 + 平台 /// 误差)高于收益——复制失败本身即失败信号(ENOSPC 走 copy 失败分支 /// → preparationFailed,用户可见可重试)。此为有意裁剪,登记为计划 @@ -19,6 +19,7 @@ library; import 'dart:async'; +import 'dart:collection'; import 'dart:io'; import 'dart:math'; @@ -45,15 +46,21 @@ abstract final class PrepErrorCodes { static const String copyFailed = 'PREP_COPY_FAILED'; } +/// 准备阶段有界并发上限(dart-define PREPARE_CONCURRENCY 覆盖;1=串行)。 +const int kMaxConcurrentPrepares = + int.fromEnvironment('PREPARE_CONCURRENCY', defaultValue: 3); + /// 上传准备器:入队唯一入口(队列 UI/连拍页调用;引擎消费产出)。 /// -/// 所有操作串行(连拍入队逐张复制,避免 IO 风暴,见库级注释)。 +/// 所有操作有界并发执行(默认 3):文件复制仍是 IO 最重环节,上限钳位 +/// 防连拍风暴;传输并发由 UploadEngine 另行控制(见库级注释)。 class UploadPreparer { UploadPreparer( this._store, { Future Function()? baseDirProvider, Random? random, this.maxBytes = kMaxUploadBytes, + this.maxConcurrentPrepares = kMaxConcurrentPrepares, this.onEnqueued, }) : _baseDirProvider = baseDirProvider ?? getApplicationSupportDirectory, _random = random ?? Random.secure(); @@ -71,8 +78,11 @@ class UploadPreparer { /// 入队成功回调(唤醒传输引擎立即出队)。 final void Function()? onEnqueued; - /// 串行队列尾部。 - Future _tail = Future.value(); + /// 准备并发上限(运行时 <1 钳位为 1)。 + final int maxConcurrentPrepares; + + int get _effectivePrepareConcurrency => + maxConcurrentPrepares < 1 ? 1 : maxConcurrentPrepares; /// 准备一张图片并入队。 /// @@ -88,7 +98,7 @@ class UploadPreparer { required String sourcePath, String? filename, }) { - return _serialize( + return _runBounded( () => _prepare( accountId: accountId, albumId: albumId, @@ -108,19 +118,46 @@ class UploadPreparer { required String taskId, required String sourcePath, }) { - return _serialize( + return _runBounded( () => _retry(accountId: accountId, taskId: taskId, sourcePath: sourcePath), ); } - /// 串行化:所有准备操作首尾相接(错误吞掉以不断链,错误经返回值上抛)。 - Future _serialize(Future Function() op) { - final result = _tail.then((_) => op()); - _tail = result.then((_) {}, onError: (Object _) {}); + /// 在途准备计数 + FIFO 等待队列(有界并发信号量)。 + int _activePrepares = 0; + final Queue> _prepareWaiters = Queue>(); + + /// 有界并发执行:超额排队(FIFO);错误由调用方经返回值处理,不设全局断链。 + Future _runBounded(Future Function() op) { + final ticket = _acquirePrepareSlot(); + final result = ticket.then((_) => op()); + result.whenComplete(_releasePrepareSlot).ignore(); return result; } + Future _acquirePrepareSlot() { + if (_activePrepares < _effectivePrepareConcurrency) { + _activePrepares++; + return Future.value(); + } + final waiter = Completer(); + _prepareWaiters.addLast(waiter); + return waiter.future; + } + + void _releasePrepareSlot() { + final waiter = _prepareWaiters.isEmpty + ? null + : _prepareWaiters.removeFirst(); + if (waiter == null) { + _activePrepares--; + return; + } + // 名额直接移交队首等待者(计数不变)。 + waiter.complete(); + } + Future _prepare({ required String accountId, required String albumId, diff --git a/apps/mobile/test/features/uploads/preparer_test.dart b/apps/mobile/test/features/uploads/preparer_test.dart index 925eb61..8d6fcc0 100644 --- a/apps/mobile/test/features/uploads/preparer_test.dart +++ b/apps/mobile/test/features/uploads/preparer_test.dart @@ -47,11 +47,15 @@ void main() { return f; } - UploadPreparer preparer({int maxBytes = UploadPreparer.kMaxUploadBytes}) => + UploadPreparer preparer({ + int maxBytes = UploadPreparer.kMaxUploadBytes, + int maxConcurrentPrepares = kMaxConcurrentPrepares, + }) => UploadPreparer( store, baseDirProvider: () async => managedBase, maxBytes: maxBytes, + maxConcurrentPrepares: maxConcurrentPrepares, ); group('prepare(快乐路径)', () { @@ -216,9 +220,66 @@ void main() { }); }); - group('串行执行(连拍防 IO 风暴)', () { - test('并发 prepare 按提交顺序串行完成,各自独立成行', () async { - final p = preparer(); + group('有界并发执行(连拍防 IO 风暴的上限钳位)', () { + /// 构造带并发探针的准备器:baseDirProvider 在 _prepare 内被 await, + /// 其重叠度是「在途 prepare 数」的可观测下界(且受信号量上限约束)。 + (UploadPreparer, int Function()) probePreparer(int limit) { + var active = 0; + var peak = 0; + final p = UploadPreparer( + store, + baseDirProvider: () async { + active++; + if (active > peak) peak = active; + await Future.delayed(const Duration(milliseconds: 30)); + active--; + return managedBase; + }, + maxConcurrentPrepares: limit, + ); + return (p, () => peak); + } + + test('并发 prepare 峰值 = maxConcurrentPrepares(≤上限且真并行),9 任务全部 queued', () async { + final (p, peakOf) = probePreparer(3); + final srcs = [ + for (var i = 0; i < 9; i++) + sourceFile('par-$i.jpg', _jpgBytes + [i]), + ]; + final tasks = await Future.wait([ + for (final src in srcs) + p.prepare(accountId: 'alice', albumId: 'alb-1', sourcePath: src.path), + ]); + // 全部入队、ID/幂等键互不相同、副本都在。 + expect(tasks.map((t) => t.state), everyElement(UploadTaskState.queued)); + expect(tasks.map((t) => t.id).toSet().length, 9); + expect(tasks.map((t) => t.clientTaskId).toSet().length, 9); + for (final t in tasks) { + expect(File(t.sourceFilePath).existsSync(), isTrue); + } + // 峰值恰为 3:≤3 证明上限钳位(防连拍 IO 风暴初衷保留); + // 达到 3 证明未退化为串行。 + expect(peakOf(), 3); + }); + + test('maxConcurrentPrepares=1 等价旧串行行为:峰值 1、任务仍全部成功', () async { + final (p, peakOf) = probePreparer(1); + final srcs = [ + for (var i = 0; i < 4; i++) + sourceFile('ser-$i.jpg', _jpgBytes + [i]), + ]; + final tasks = await Future.wait([ + for (final src in srcs) + p.prepare(accountId: 'alice', albumId: 'alb-1', sourcePath: src.path), + ]); + expect(tasks.map((t) => t.state), everyElement(UploadTaskState.queued)); + expect(peakOf(), 1, reason: '并发 1 时任意时刻在途 prepare 至多 1(串行)'); + }); + }); + + group('串行回归(maxConcurrentPrepares=1 等价旧全局串行队列)', () { + test('并发=1:prepare 按提交顺序串行完成,各自独立成行', () async { + final p = preparer(maxConcurrentPrepares: 1); final srcs = [ for (var i = 0; i < 3; i++) sourceFile('burst-$i.jpg', _jpgBytes + [i]), @@ -234,7 +295,7 @@ void main() { for (final t in tasks) { expect(File(t.sourceFilePath).existsSync(), isTrue); } - // createdAt 严格递增(串行执行的间接证据:无交叉写库)。 + // createdAt 非降(并发=1 串行的间接证据:无交叉写库)。 final stamps = tasks.map((t) => t.createdAt).toList(); for (var i = 1; i < stamps.length; i++) { expect(!stamps[i].isBefore(stamps[i - 1]), isTrue);