feat(app): bounded-parallel prepare (default 3) replaces serial queue
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -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<Directory> 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<void> _tail = Future<void>.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<T> _serialize<T>(Future<T> Function() op) {
|
||||
final result = _tail.then((_) => op());
|
||||
_tail = result.then<void>((_) {}, onError: (Object _) {});
|
||||
/// 在途准备计数 + FIFO 等待队列（有界并发信号量）。
|
||||
int _activePrepares = 0;
|
||||
final Queue<Completer<void>> _prepareWaiters = Queue<Completer<void>>();
|
||||
|
||||
/// 有界并发执行：超额排队（FIFO）；错误由调用方经返回值处理，不设全局断链。
|
||||
Future<T> _runBounded<T>(Future<T> Function() op) {
|
||||
final ticket = _acquirePrepareSlot();
|
||||
final result = ticket.then((_) => op());
|
||||
result.whenComplete(_releasePrepareSlot).ignore();
|
||||
return result;
|
||||
}
|
||||
|
||||
Future<void> _acquirePrepareSlot() {
|
||||
if (_activePrepares < _effectivePrepareConcurrency) {
|
||||
_activePrepares++;
|
||||
return Future<void>.value();
|
||||
}
|
||||
final waiter = Completer<void>();
|
||||
_prepareWaiters.addLast(waiter);
|
||||
return waiter.future;
|
||||
}
|
||||
|
||||
void _releasePrepareSlot() {
|
||||
final waiter = _prepareWaiters.isEmpty
|
||||
? null
|
||||
: _prepareWaiters.removeFirst();
|
||||
if (waiter == null) {
|
||||
_activePrepares--;
|
||||
return;
|
||||
}
|
||||
// 名额直接移交队首等待者（计数不变）。
|
||||
waiter.complete();
|
||||
}
|
||||
|
||||
Future<UploadTask> _prepare({
|
||||
required String accountId,
|
||||
required String albumId,
|
||||
|
||||
@@ -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<void>.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);
|
||||
|
||||
Reference in New Issue
Block a user