2026/9/19 5:12:52
FEATURED · 精选文章 
Spacedrive 并行任务执行设计从 JobContext 接入 TaskDispatcher 的多线程任务派发架构【免费下载链接】spacedriveSpacedrive is an open source cross-platform file explorer, powered by a virtual distributed filesystem written in Rust.项目地址: https://gitcode.com/gh_mirrors/sp/spacedrive本文基于 .tasks/core/JOB-003-parallel-task-execution.mdJOB-003展开该任务文档定义了 Spacedrive 核心的并行化改造方案让持久化 Job 系统通过JobContext暴露task_dispatcher()从而在 task-system 的多线程 worker 池上并行派发子任务。读完本文你将掌握 Spacedrive Job 系统与 task-system 的分层关系、TaskDispatcher的派发与调度原理、以及 FileCopyJob 从顺序执行迁移到并行执行的完整落地路径与验收标准。背景Durable Job System 与 task-system 的分层Spacedrive 的持久化后台执行引擎由一组任务文档定义其中 JOB-000-job-system.md 是总纲Epic负责长时任务的弹性异步执行支持暂停、恢复与中断后恢复JOB-001-job-manager.md 定义了每个 Library 独立的JobManager它构建在通用TaskSystem之上进行并发管理。这两个层次的职责划分是理解 JOB-003 的前提Job 层core/src/infra/job/面向业务语义例如文件复制、缩略图生成、媒体索引。Job 会被序列化到数据库含检查点 checkpoint可暂停、可恢复、可跨进程重启续跑。Task 层crates/task-system/通用并发执行引擎只关心如何把任务并发跑完不关心业务语义。从 crates/task-system/src/lib.rs 的模块文档可以看到 task-system 的核心能力清单按 CPU 核心数创建 worker 的轮询调度round robin、worker 之间的工作窃取work stealing、优雅暂停与取消、强制终止、优先级抢占挂起无优先级任务来执行高优先级任务以及在系统关闭时把所有 pending/running 任务交还给 dispatcher 以便落盘后重新派发。在 core/src/infra/job/manager.rs 中JobManager持有dispatcher: ArcTaskSystemJobError并统一使用JobError作为错误类型派发路径可见 manager.rs 中的self.dispatcher.get_dispatcher().dispatch_boxed(executor)。这说明Job 与 Task 共享同一个错误类型JobErrorJob 本身最终也是以Task的身份进入 task-system 执行的。问题单个 Job 只能顺序处理工作负载JOB-003 指出的核心痛点当前 Job 系统把每个 Job 当作单个 Task 执行工作完全串行。对于复制 100 个文件这类典型 I/O 密集操作当前顺序100 文件 × 500ms 50 秒 目标并行100 文件 / 10 workers 5 秒10 倍提升顺序执行意味着复制文件时 CPU 核心空闲、存储 I/O 未充分利用。JOB-003 将这类场景量化为性能预期详见下文性能预期小节并明确适用范围是I/O 密集操作文件复制、缩略图生成、媒体索引、批量删除、哈希计算等。解决方案架构Job 作为编排者Task 作为工人JOB-003 给出了清晰的架构链路JobManager creates JobExecutor with TaskDispatcher ↓ JobContext exposes ctx.task_dispatcher() ↓ Job spawns parallel tasks via dispatcher ↓ Tasks execute on multi-threaded worker pool文档末尾的 Notes 提炼出设计要义Jobs are orchestrators, tasks are workers.Job 是编排者Task 是工人。Job 不需要特殊的任务类型只需要具备派发标准 Task 的能力——这取代了此前过度设计的JOB_TASK_COMPOSITION_API.md方案。对照当前源码这个编排者定位已经体现在 core/src/infra/job/context.rs 的JobContext结构中它已内置networking: OptionArcNetworkingService与volume_manager: OptionArcVolumeManager并通过ctx.networking_service()、ctx.volume_manager()暴露见 context.rs。JOB-003 的ctx.task_dispatcher()就是遵循这一既有模式新增的第三个访问器。为什么通过 Context 而非 Job 存储传递 dispatcherJOB-003 列举了四点理由这决定了整个 Phase 1 的改动范围Job 会被序列化到数据库——dispatcher 不可序列化放进 Job 字段会破坏持久化JobExecutor 本就拥有 task-system 访问权——从 executor 注入 Context 是天然路径与既有模式一致——ctx.library()、ctx.networking_service()已是标准用法无需破坏#[derive(Job)]宏——Job 结构体字段不变宏生成的序列化/反序列化逻辑不受影响。Phase 1核心集成JOB-003a——让 Job 拿到 dispatcherPhase 1 的目标是打通Job → Context → dispatcher的访问链路共 6 项改动在JobExecutorState中添加task_dispatcher字段更新JobExecutor::new()接受 dispatcher 参数在JobContext中添加task_dispatcher字段添加ctx.task_dispatcher()访问器方法更新JobManager::dispatch()传递 dispatcher按需更新#[derive(Job)]宏。涉及文件core/src/infra/job/executor.rscore/src/infra/job/context.rscore/src/infra/job/manager.rs从源码看这 6 步的落点非常明确。JobExecutorState目前已有networking、volume_manager等可选字段executor.rs新增task_dispatcher完全同构run_inner中构造JobContext时executor.rs把 dispatcher 一并注入即可JobManager的派发路径manager.rs、manager.rs在创建 executor 时带上 dispatcher 克隆。Phase 1 验收标准Job 可以调用ctx.task_dispatcher()并获得有效 dispatcher集成测试展示 Job 派生并行任务对现有 Job 无破坏性变更。TaskDispatcher 的底层原理worker 池、轮询派发与工作窃取要理解 Phase 2 中dispatcher.dispatch_many()的行为需要先看 task-system 的实现细节。Worker 池规模crates/task-system/src/system.rs 中System::new()以std::thread::available_parallelism()的一半创建 worker 数量至少 1源码注释注明这是临时策略未来计划做成运行时可配置。派发路径System同时提供dispatch与dispatch_manysystem.rs二者都委托给内部的BaseDispatcher导出别名为BaseTaskDispatcher。在dispatch_boxedsystem.rs中可以看到若系统已 shutdown直接返回DispatcherShutdownError把任务原样退回通过AtomicWorkerId的fetch_update做轮询round robin选择(last_worker_id 1) % workers.len()把任务分发给对应 worker分发后将该 worker 标记为忙碌。而dispatch_many_boxedsystem.rs对每个任务并发执行worker.add_task()把同一批任务铺开到所有 worker 上——这正是 JOB-003 中dispatcher.dispatch_many(tasks)直接依赖的能力。此外 worker 之间还存在**工作窃取work stealing**机制用于负载均衡见 crates/task-system/src/worker/ 目录Dispatchertrait 的完整定义见 system.rs。任务控制能力TaskHandletask.rs暴露pause()、cancel()、resume()、force_abortion()与remote_controller()任务运行期间通过Interrupter在安全点检查中断请求try_check_interrupt()与check_interruption!宏task.rs是非阻塞检查而await一个Interrupter会阻塞直到收到暂停/取消指令。TaskStatus枚举task.rs覆盖Done、Canceled、ForcedAbortion、Shutdown、Error五种结局——Phase 2 的并行任务聚合逻辑就是围绕这组状态机展开的。值得强调的是task-system 的这套模式在仓库中已有完整参考实现JOB-003 的 References 指向 crates/task-system/tests/common/jobs.rs。其中SampleJob持有一个BaseTaskDispatcherSampleError用dispatch_many派出首批任务再通过FutureGroup动态追加工人任务jobs.rs完整展示了编排者 工人的经典形态可作为 JOB-003 实现的模式蓝本。Phase 2FileCopy 概念验证JOB-003b——把 FileCopyJob 并行化Phase 2 是第一个真实业务落地目标是把FileCopyJob从顺序复制迁移为并行复制共 5 项改动创建实现TaskJobError的CopyFileTask更新FileCopyJob::run()使用dispatcher.dispatch_many()实现并行任务的进度聚合保持可恢复性跟踪已完成的文件索引优雅处理部分失败。涉及文件core/src/ops/files/copy/job.rs现有core/src/ops/files/copy/task.rs新建现有 FileCopyJob 的可恢复性基础对照 core/src/ops/files/copy/job.rs 可以看到FileCopyJob已经是为可恢复而设计的#[derive(Debug, Serialize, Deserialize, Job)] pub struct FileCopyJob { pub sources: SdPathBatch, pub destination: SdPath, #[serde(default)] pub options: CopyOptions, // Internal state for resumption #[serde(default)] pub completed_indices: Vecusize, // 已完成文件索引用于恢复 #[serde(skip, default Instant::now)] started_at: Instant, #[serde(default)] pub job_metadata: super::metadata::CopyJobMetadata, }completed_indices正是 JOB-003 代码示例中过滤已完成任务的关键字段同时Jobtrait 声明const RESUMABLE: bool true。复制策略本身由CopyStrategyRouter负责job.rs 调用select_strategy_with_metadata策略实现位于 core/src/ops/files/copy/routing.rsCopyFileTask直接复用这套策略即可保持复制语义完全一致。并行化后的 FileCopyJob.run() 示例JOB-003 原文#[async_trait] impl JobHandler for FileCopyJob { async fn run(mut self, ctx: JobContext_) - JobResultSelf::Output { // Get dispatcher from context let dispatcher ctx.task_dispatcher(); // Create parallel copy tasks let tasks: Vec_ self.sources.paths.iter() .enumerate() .filter(|(idx, _)| !self.completed_indices.contains(idx)) .map(|(idx, source)| CopyFileTask { id: TaskId::new_v4(), index: idx, source: source.clone(), destination: self.destination.clone(), options: self.options.clone(), }) .collect(); // Dispatch all tasks - task system handles distribution let handles dispatcher.dispatch_many(tasks).await?; // Wait for completion and track progress for (completed, handle) in handles.into_iter().enumerate() { ctx.check_interrupt().await?; match handle.await { Ok(TaskStatus::Done(_)) { self.completed_indices.push(completed); ctx.progress(/* ... */); } Ok(TaskStatus::Error(e)) { // Handle individual task failure } _ {} } if (completed 1) % 10 0 { ctx.checkpoint().await?; } } Ok(FileCopyOutput { /* ... */ }) } }该示例的关键设计点拆解过滤已完成项!self.completed_indices.contains(idx)保证中断后重跑只处理未完成文件这是可恢复性的第一道防线批量派发dispatcher.dispatch_many(tasks)把全部任务一次性交给 task-system由 worker 池轮询分发并工作窃取Job 无需关心分配细节聚合等待逐handle.await等待每个任务完成按完成情况更新completed_indices并上报进度协作式中断每轮循环ctx.check_interrupt().await?响应暂停/取消定期检查点每完成 10 个任务调用ctx.checkpoint().await?把进度落库保证中断后可恢复。CopyFileTask 示例JOB-003 原文struct CopyFileTask { id: TaskId, index: usize, source: SdPath, destination: SdPath, options: CopyOptions, } #[async_trait] impl TaskJobError for CopyFileTask { fn id(self) - TaskId { self.id } async fn run(mut self, interrupter: Interrupter) - ResultExecStatus, JobError { // Check interruption interrupter.try_check_interrupt()?; // Execute copy strategy let strategy CopyStrategyRouter::select_strategy(/* ... */).await; let bytes_copied strategy.execute_simple(/* ... */).await?; Ok(ExecStatus::Done(/* output */)) } }CopyFileTask只实现 task-system 的TaskJobErrortrait接口定义见 crates/task-system/src/task.rsid()提供唯一标识run()在Interrupter的配合下于安全点检查中断复用CopyStrategyRouter完成实际复制并返回ExecStatus::Done。注意Tasktrait 还提供了两个可选钩子with_priority()默认false可让关键任务抢占 worker与with_timeout()默认无限等待可设置超时取消——在 Phase 2 中可按需启用。Phase 2 验收标准FileCopyJob 派生并行复制任务100 文件场景性能提升 4-8 倍Job 在中断后仍可恢复部分失败不会终止整个 Job进度上报正确。Phase 3文档与模式沉淀并行化模式落地后需要在开发者文档中沉淀经验JOB-003 列出的交付物包括在 Job 系统文档中新增并行执行指南、更新 Job 实现模板、在开发者文档中提供代码示例、以及一个演示该模式的集成测试。结合上文提到的 crates/task-system/tests/common/jobs.rs 的SampleJob该测试可作为并行编排模式的权威参照——实际上 JOB-003 的设计正是基于 Spacedrive v1 task-system 设计的成熟模式。Phase 4 与 Phase 5未来扩展方向Phase 3 之后的两个阶段是渐进式扩展JOB-003 明确标注为未来工作Phase 4扩展到其他 I/O 密集操作缩略图生成高度可并行媒体元数据提取文件删除批量操作哈希计算CPU 密集并行。Phase 5集中式资源管理全局资源池I/O、CPU、网络、数据库基于信号量的LimitedTaskDispatcher包装器优先级感知的资源分配根据系统负载动态调整限制。JOB-003 特别注明资源限制推迟到并行执行概念验证成功之后的阶段再做——先证明并行化的收益再考虑防止系统过载的节流机制避免过度工程。性能预期JOB-003 原文数据文件复制100 个 1MB 文件SSD模式计算耗时顺序100 文件 × 20ms2000ms并行10 并发10 批 × 20ms200ms10 倍提升真实场景混合大小共 10GB模式耗时顺序~102s并行~12s8.5 倍提升这些数据是 JOB-003 文档给出的预期值属于设计阶段的估算任务状态为 To Do用于论证并行化的投入产出而非已实测的基准结果验收阶段应以真实 benchmark 为准。收益总结与设计要点回顾JOB-003 给出的五项收益真正的并行任务借助工作窃取机制分布到所有 CPU 核心无架构变更完全复用既有 task-system 基础设施向后兼容既有顺序 Job 不受影响继续正常工作实现简单无需包装器、适配器或特殊 trait模式成熟源自 Spacedrive v1 的 task-system 设计。核心设计要点可归纳为Job 是可序列化、可恢复的编排者Task 是无状态工人JobContext是 Job 访问系统服务Library、网络、卷管理器、以及未来的任务派发器的统一入口。通过上下文注入而非字段持有的方式传递 dispatcher既绕开了序列化约束又保持了#[derive(Job)]宏的兼容性使整个并行化改造可以被拆成 5 个可独立验收的阶段逐步推进。关联文档与参考路径任务文档.tasks/core/JOB-003-parallel-task-execution.md本文主体来源父任务JOB-000 Durable Job System.tasks/core/JOB-000-job-system.md、JOB-001 Job Manager.tasks/core/JOB-001-job-manager.md相关任务FILE-001File Copy Job、FILE-003Cloud File Operationstask-system 实现crates/task-system/src/lib.rs、crates/task-system/src/system.rs、crates/task-system/src/task.rs并行编排参考实现crates/task-system/tests/common/jobs.rsJob 系统源码core/src/infra/job/context.rs、core/src/infra/job/executor.rs、core/src/infra/job/manager.rs复制策略core/src/ops/files/copy/routing.rs、core/src/ops/files/copy/job.rs【免费下载链接】spacedriveSpacedrive is an open source cross-platform file explorer, powered by a virtual distributed filesystem written in Rust.项目地址: https://gitcode.com/gh_mirrors/sp/spacedrive创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Spacedrive 并行任务执行设计：从 JobContext 接入 TaskDispatcher 的多线程任务派发架构
Spacedrive 并行任务执行设计从 JobContext 接入 TaskDispatcher 的多线程任务派发架构【免费下载链接】spacedriveSpacedrive is an open source cross-platform file explorer, powered by a virtual distributed filesystem written in Rust.项目地址: https://gitcode.com/gh_mirrors/sp/spacedrive本文基于 .tasks/core/JOB-003-parallel-task-execution.mdJOB-003展开该任务文档定义了 Spacedrive 核心的并行化改造方案让持久化 Job 系统通过JobContext暴露task_dispatcher()从而在 task-system 的多线程 worker 池上并行派发子任务。读完本文你将掌握 Spacedrive Job 系统与 task-system 的分层关系、TaskDispatcher的派发与调度原理、以及 FileCopyJob 从顺序执行迁移到并行执行的完整落地路径与验收标准。背景Durable Job System 与 task-system 的分层Spacedrive 的持久化后台执行引擎由一组任务文档定义其中 JOB-000-job-system.md 是总纲Epic负责长时任务的弹性异步执行支持暂停、恢复与中断后恢复JOB-001-job-manager.md 定义了每个 Library 独立的JobManager它构建在通用TaskSystem之上进行并发管理。这两个层次的职责划分是理解 JOB-003 的前提Job 层core/src/infra/job/面向业务语义例如文件复制、缩略图生成、媒体索引。Job 会被序列化到数据库含检查点 checkpoint可暂停、可恢复、可跨进程重启续跑。Task 层crates/task-system/通用并发执行引擎只关心如何把任务并发跑完不关心业务语义。从 crates/task-system/src/lib.rs 的模块文档可以看到 task-system 的核心能力清单按 CPU 核心数创建 worker 的轮询调度round robin、worker 之间的工作窃取work stealing、优雅暂停与取消、强制终止、优先级抢占挂起无优先级任务来执行高优先级任务以及在系统关闭时把所有 pending/running 任务交还给 dispatcher 以便落盘后重新派发。在 core/src/infra/job/manager.rs 中JobManager持有dispatcher: ArcTaskSystemJobError并统一使用JobError作为错误类型派发路径可见 manager.rs 中的self.dispatcher.get_dispatcher().dispatch_boxed(executor)。这说明Job 与 Task 共享同一个错误类型JobErrorJob 本身最终也是以Task的身份进入 task-system 执行的。问题单个 Job 只能顺序处理工作负载JOB-003 指出的核心痛点当前 Job 系统把每个 Job 当作单个 Task 执行工作完全串行。对于复制 100 个文件这类典型 I/O 密集操作当前顺序100 文件 × 500ms 50 秒 目标并行100 文件 / 10 workers 5 秒10 倍提升顺序执行意味着复制文件时 CPU 核心空闲、存储 I/O 未充分利用。JOB-003 将这类场景量化为性能预期详见下文性能预期小节并明确适用范围是I/O 密集操作文件复制、缩略图生成、媒体索引、批量删除、哈希计算等。解决方案架构Job 作为编排者Task 作为工人JOB-003 给出了清晰的架构链路JobManager creates JobExecutor with TaskDispatcher ↓ JobContext exposes ctx.task_dispatcher() ↓ Job spawns parallel tasks via dispatcher ↓ Tasks execute on multi-threaded worker pool文档末尾的 Notes 提炼出设计要义Jobs are orchestrators, tasks are workers.Job 是编排者Task 是工人。Job 不需要特殊的任务类型只需要具备派发标准 Task 的能力——这取代了此前过度设计的JOB_TASK_COMPOSITION_API.md方案。对照当前源码这个编排者定位已经体现在 core/src/infra/job/context.rs 的JobContext结构中它已内置networking: OptionArcNetworkingService与volume_manager: OptionArcVolumeManager并通过ctx.networking_service()、ctx.volume_manager()暴露见 context.rs。JOB-003 的ctx.task_dispatcher()就是遵循这一既有模式新增的第三个访问器。为什么通过 Context 而非 Job 存储传递 dispatcherJOB-003 列举了四点理由这决定了整个 Phase 1 的改动范围Job 会被序列化到数据库——dispatcher 不可序列化放进 Job 字段会破坏持久化JobExecutor 本就拥有 task-system 访问权——从 executor 注入 Context 是天然路径与既有模式一致——ctx.library()、ctx.networking_service()已是标准用法无需破坏#[derive(Job)]宏——Job 结构体字段不变宏生成的序列化/反序列化逻辑不受影响。Phase 1核心集成JOB-003a——让 Job 拿到 dispatcherPhase 1 的目标是打通Job → Context → dispatcher的访问链路共 6 项改动在JobExecutorState中添加task_dispatcher字段更新JobExecutor::new()接受 dispatcher 参数在JobContext中添加task_dispatcher字段添加ctx.task_dispatcher()访问器方法更新JobManager::dispatch()传递 dispatcher按需更新#[derive(Job)]宏。涉及文件core/src/infra/job/executor.rscore/src/infra/job/context.rscore/src/infra/job/manager.rs从源码看这 6 步的落点非常明确。JobExecutorState目前已有networking、volume_manager等可选字段executor.rs新增task_dispatcher完全同构run_inner中构造JobContext时executor.rs把 dispatcher 一并注入即可JobManager的派发路径manager.rs、manager.rs在创建 executor 时带上 dispatcher 克隆。Phase 1 验收标准Job 可以调用ctx.task_dispatcher()并获得有效 dispatcher集成测试展示 Job 派生并行任务对现有 Job 无破坏性变更。TaskDispatcher 的底层原理worker 池、轮询派发与工作窃取要理解 Phase 2 中dispatcher.dispatch_many()的行为需要先看 task-system 的实现细节。Worker 池规模crates/task-system/src/system.rs 中System::new()以std::thread::available_parallelism()的一半创建 worker 数量至少 1源码注释注明这是临时策略未来计划做成运行时可配置。派发路径System同时提供dispatch与dispatch_manysystem.rs二者都委托给内部的BaseDispatcher导出别名为BaseTaskDispatcher。在dispatch_boxedsystem.rs中可以看到若系统已 shutdown直接返回DispatcherShutdownError把任务原样退回通过AtomicWorkerId的fetch_update做轮询round robin选择(last_worker_id 1) % workers.len()把任务分发给对应 worker分发后将该 worker 标记为忙碌。而dispatch_many_boxedsystem.rs对每个任务并发执行worker.add_task()把同一批任务铺开到所有 worker 上——这正是 JOB-003 中dispatcher.dispatch_many(tasks)直接依赖的能力。此外 worker 之间还存在**工作窃取work stealing**机制用于负载均衡见 crates/task-system/src/worker/ 目录Dispatchertrait 的完整定义见 system.rs。任务控制能力TaskHandletask.rs暴露pause()、cancel()、resume()、force_abortion()与remote_controller()任务运行期间通过Interrupter在安全点检查中断请求try_check_interrupt()与check_interruption!宏task.rs是非阻塞检查而await一个Interrupter会阻塞直到收到暂停/取消指令。TaskStatus枚举task.rs覆盖Done、Canceled、ForcedAbortion、Shutdown、Error五种结局——Phase 2 的并行任务聚合逻辑就是围绕这组状态机展开的。值得强调的是task-system 的这套模式在仓库中已有完整参考实现JOB-003 的 References 指向 crates/task-system/tests/common/jobs.rs。其中SampleJob持有一个BaseTaskDispatcherSampleError用dispatch_many派出首批任务再通过FutureGroup动态追加工人任务jobs.rs完整展示了编排者 工人的经典形态可作为 JOB-003 实现的模式蓝本。Phase 2FileCopy 概念验证JOB-003b——把 FileCopyJob 并行化Phase 2 是第一个真实业务落地目标是把FileCopyJob从顺序复制迁移为并行复制共 5 项改动创建实现TaskJobError的CopyFileTask更新FileCopyJob::run()使用dispatcher.dispatch_many()实现并行任务的进度聚合保持可恢复性跟踪已完成的文件索引优雅处理部分失败。涉及文件core/src/ops/files/copy/job.rs现有core/src/ops/files/copy/task.rs新建现有 FileCopyJob 的可恢复性基础对照 core/src/ops/files/copy/job.rs 可以看到FileCopyJob已经是为可恢复而设计的#[derive(Debug, Serialize, Deserialize, Job)] pub struct FileCopyJob { pub sources: SdPathBatch, pub destination: SdPath, #[serde(default)] pub options: CopyOptions, // Internal state for resumption #[serde(default)] pub completed_indices: Vecusize, // 已完成文件索引用于恢复 #[serde(skip, default Instant::now)] started_at: Instant, #[serde(default)] pub job_metadata: super::metadata::CopyJobMetadata, }completed_indices正是 JOB-003 代码示例中过滤已完成任务的关键字段同时Jobtrait 声明const RESUMABLE: bool true。复制策略本身由CopyStrategyRouter负责job.rs 调用select_strategy_with_metadata策略实现位于 core/src/ops/files/copy/routing.rsCopyFileTask直接复用这套策略即可保持复制语义完全一致。并行化后的 FileCopyJob.run() 示例JOB-003 原文#[async_trait] impl JobHandler for FileCopyJob { async fn run(mut self, ctx: JobContext_) - JobResultSelf::Output { // Get dispatcher from context let dispatcher ctx.task_dispatcher(); // Create parallel copy tasks let tasks: Vec_ self.sources.paths.iter() .enumerate() .filter(|(idx, _)| !self.completed_indices.contains(idx)) .map(|(idx, source)| CopyFileTask { id: TaskId::new_v4(), index: idx, source: source.clone(), destination: self.destination.clone(), options: self.options.clone(), }) .collect(); // Dispatch all tasks - task system handles distribution let handles dispatcher.dispatch_many(tasks).await?; // Wait for completion and track progress for (completed, handle) in handles.into_iter().enumerate() { ctx.check_interrupt().await?; match handle.await { Ok(TaskStatus::Done(_)) { self.completed_indices.push(completed); ctx.progress(/* ... */); } Ok(TaskStatus::Error(e)) { // Handle individual task failure } _ {} } if (completed 1) % 10 0 { ctx.checkpoint().await?; } } Ok(FileCopyOutput { /* ... */ }) } }该示例的关键设计点拆解过滤已完成项!self.completed_indices.contains(idx)保证中断后重跑只处理未完成文件这是可恢复性的第一道防线批量派发dispatcher.dispatch_many(tasks)把全部任务一次性交给 task-system由 worker 池轮询分发并工作窃取Job 无需关心分配细节聚合等待逐handle.await等待每个任务完成按完成情况更新completed_indices并上报进度协作式中断每轮循环ctx.check_interrupt().await?响应暂停/取消定期检查点每完成 10 个任务调用ctx.checkpoint().await?把进度落库保证中断后可恢复。CopyFileTask 示例JOB-003 原文struct CopyFileTask { id: TaskId, index: usize, source: SdPath, destination: SdPath, options: CopyOptions, } #[async_trait] impl TaskJobError for CopyFileTask { fn id(self) - TaskId { self.id } async fn run(mut self, interrupter: Interrupter) - ResultExecStatus, JobError { // Check interruption interrupter.try_check_interrupt()?; // Execute copy strategy let strategy CopyStrategyRouter::select_strategy(/* ... */).await; let bytes_copied strategy.execute_simple(/* ... */).await?; Ok(ExecStatus::Done(/* output */)) } }CopyFileTask只实现 task-system 的TaskJobErrortrait接口定义见 crates/task-system/src/task.rsid()提供唯一标识run()在Interrupter的配合下于安全点检查中断复用CopyStrategyRouter完成实际复制并返回ExecStatus::Done。注意Tasktrait 还提供了两个可选钩子with_priority()默认false可让关键任务抢占 worker与with_timeout()默认无限等待可设置超时取消——在 Phase 2 中可按需启用。Phase 2 验收标准FileCopyJob 派生并行复制任务100 文件场景性能提升 4-8 倍Job 在中断后仍可恢复部分失败不会终止整个 Job进度上报正确。Phase 3文档与模式沉淀并行化模式落地后需要在开发者文档中沉淀经验JOB-003 列出的交付物包括在 Job 系统文档中新增并行执行指南、更新 Job 实现模板、在开发者文档中提供代码示例、以及一个演示该模式的集成测试。结合上文提到的 crates/task-system/tests/common/jobs.rs 的SampleJob该测试可作为并行编排模式的权威参照——实际上 JOB-003 的设计正是基于 Spacedrive v1 task-system 设计的成熟模式。Phase 4 与 Phase 5未来扩展方向Phase 3 之后的两个阶段是渐进式扩展JOB-003 明确标注为未来工作Phase 4扩展到其他 I/O 密集操作缩略图生成高度可并行媒体元数据提取文件删除批量操作哈希计算CPU 密集并行。Phase 5集中式资源管理全局资源池I/O、CPU、网络、数据库基于信号量的LimitedTaskDispatcher包装器优先级感知的资源分配根据系统负载动态调整限制。JOB-003 特别注明资源限制推迟到并行执行概念验证成功之后的阶段再做——先证明并行化的收益再考虑防止系统过载的节流机制避免过度工程。性能预期JOB-003 原文数据文件复制100 个 1MB 文件SSD模式计算耗时顺序100 文件 × 20ms2000ms并行10 并发10 批 × 20ms200ms10 倍提升真实场景混合大小共 10GB模式耗时顺序~102s并行~12s8.5 倍提升这些数据是 JOB-003 文档给出的预期值属于设计阶段的估算任务状态为 To Do用于论证并行化的投入产出而非已实测的基准结果验收阶段应以真实 benchmark 为准。收益总结与设计要点回顾JOB-003 给出的五项收益真正的并行任务借助工作窃取机制分布到所有 CPU 核心无架构变更完全复用既有 task-system 基础设施向后兼容既有顺序 Job 不受影响继续正常工作实现简单无需包装器、适配器或特殊 trait模式成熟源自 Spacedrive v1 的 task-system 设计。核心设计要点可归纳为Job 是可序列化、可恢复的编排者Task 是无状态工人JobContext是 Job 访问系统服务Library、网络、卷管理器、以及未来的任务派发器的统一入口。通过上下文注入而非字段持有的方式传递 dispatcher既绕开了序列化约束又保持了#[derive(Job)]宏的兼容性使整个并行化改造可以被拆成 5 个可独立验收的阶段逐步推进。关联文档与参考路径任务文档.tasks/core/JOB-003-parallel-task-execution.md本文主体来源父任务JOB-000 Durable Job System.tasks/core/JOB-000-job-system.md、JOB-001 Job Manager.tasks/core/JOB-001-job-manager.md相关任务FILE-001File Copy Job、FILE-003Cloud File Operationstask-system 实现crates/task-system/src/lib.rs、crates/task-system/src/system.rs、crates/task-system/src/task.rs并行编排参考实现crates/task-system/tests/common/jobs.rsJob 系统源码core/src/infra/job/context.rs、core/src/infra/job/executor.rs、core/src/infra/job/manager.rs复制策略core/src/ops/files/copy/routing.rs、core/src/ops/files/copy/job.rs【免费下载链接】spacedriveSpacedrive is an open source cross-platform file explorer, powered by a virtual distributed filesystem written in Rust.项目地址: https://gitcode.com/gh_mirrors/sp/spacedrive创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
