DAG: execute recompute levels concurrently

2026-08-02 17:52:26 -04:00
parent 0c8efc8b69
commit cb6877be36
5 changed files with 108 additions and 42 deletions
--- a/docs/freecad-full-parity-plan.zh-CN.md
+++ b/docs/freecad-full-parity-plan.zh-CN.md
@@ -324,7 +324,7 @@ Geometry Facade 还接入了 Bitbybit IO 的 STEP 和 ASCII STL 导出，作为

 | 任务 | 实施措施 | 验证结果 | 剩余边界 |
 |---|---|---|---|
-| P1-03 / DAG-06 重算执行器 | `RecomputeCoordinator` 为每次运行分配 generation；后发运行取消前发；节点完成后校验 AbortSignal、文档 ID/版本和当前 generation；结果状态为 `completed/failed/cancelled/stale` | 取消、过期版本、依赖失败传播和 Facade 提交测试通过 | 每个 PartDesign 特征尚未绑定真实 OCCT 输入/输出 Shape；尚未实现 Worker 并行层 |
+| P1-03 / DAG-06 重算执行器 | `RecomputeCoordinator` 为每次运行分配 generation；后发运行取消前发；节点完成后校验 AbortSignal、文档 ID/版本和当前 generation；结果状态为 `completed/failed/cancelled/stale`；按 DAG `levels` 使用 `Promise.all` 并行执行同层节点，并按稳定计划顺序合并结果和进度 | 取消、过期版本、依赖失败传播、独立节点并行和 Facade 提交测试通过 | 每个 PartDesign 特征尚未绑定真实 OCCT 输入/输出 Shape；同层并行目前是调度层并发，不等同于每节点独立 Worker；尚未实现最近有效 Shape 保留 |
 | DAG-08 失败传播 | 上游 `error` 将下游置为 `upstream-failed`，保存错误根因和 dirty 闭包；成功后只提交当前文档版本 | 上游失败和下游跳过测试通过 | 抑制对象、最近有效 Shape、完整 FreeCAD 状态机和 Report 树仍待实现 |
 | SK-04 求解器边界 | Sketch 节点通过可替换 executor 求解；冲突/无效约束阻止下游，求解结果作为对象更新返回 | 基础 Sketch solve 与异步对象更新测试通过 | 当前是基础 TypeScript solver，不是 `planegcs` WASM；高级约束、拖拽、外部几何和 B-spline 未完成 |
 | FC-03 ZIP 安全读取 | 先读中央目录再解压；限制 archive/entry/total 字节、压缩比、条目数；拒绝路径穿越、重复路径、加密项、ZIP64 和 XML entity | 缺 Document.xml、路径穿越、压缩炸弹测试通过 | 尚未恢复 BRep/Shape、视图、表达式和参数化对象 |
--- a/docs/web-cad-implementation-plan.zh-CN.md
+++ b/docs/web-cad-implementation-plan.zh-CN.md
@@ -1476,7 +1476,7 @@ SQLite 是运行时的主存储，不要求项目包直接暴露数据库内部
 |---|---|---|---|---|
 | P4-01 | 实现 Document/Object/Container/Feature/Shape | 领域包和序列化投影 | G1 | UUID、类型、标签和生命周期稳定 |
 | P4-02 | 实现 Property/Link/Expression/Unit | 元数据、校验和单位换算 | P4-01 | Data/View 编辑器由 metadata 驱动 |
-| P4-03 | 实现依赖 DAG、脏标记和拓扑调度 | recompute scheduler、稳定 `levels` | P4-01 | 只重算受影响闭包；同层对象无依赖，可安全交给并行 Worker |
+| P4-03 | 实现依赖 DAG、脏标记和拓扑调度 | recompute scheduler、稳定 `levels`、同层并发执行 | P4-01 | 只重算受影响闭包；同层对象无依赖；调度器以 `Promise.all` 并发执行并按计划顺序提交结果，未来再下沉到并行 Worker |
 | P4-04 | 实现事务、撤销/重做和保存点 | Command Bus、undo journal | P4-03 | 100 次撤销/重做可回到原 hash |
 | P4-05 | 实现 Body/Tip/特征顺序/可见性 | PartDesign 领域规则 | P4-01/P3-04 | 非法多实体和 Tip 操作被拒绝 |
 | P4-06 | 实现诊断树和失败恢复 | Diagnostic model、repair actions | P4-03 | 失败节点可定位、回退或修复 |
--- a/src/facade/mockFacade.ts
+++ b/src/facade/mockFacade.ts
@@ -133,6 +133,7 @@ const createDocument = (label = 'Pump Housing'): DocumentSnapshot => {
 const selectionRequired = new Set(['pad', 'pocket', 'revolution', 'fillet', 'chamfer', 'hole', 'linear-pattern', 'polar-pattern', 'measure-distance', 'measure-angle', 'measure-area', 'solve-sketch'])
 const systemCommands = new Set(['new-document', 'save', 'select-object'])
 const implementedCommandIds = new Set(['new-document', 'save', 'select-object', 'create-body', 'create-sketch', 'new-sketch', 'pad', 'pocket', 'fillet', 'chamfer', 'solve-sketch'])
+const partDesignCommands = new Set(['create-body', 'create-sketch', 'pad', 'pocket', 'fillet', 'chamfer'])
 const featureCommands: Record<string, { label: string; detail: string }> = {
  'create-body': { label: 'Body', detail: 'Part Design body' },
  'create-sketch': { label: 'Sketch', detail: 'Fully constrained' },
@@ -146,7 +147,8 @@ const commandState = (commandId: string, activeWorkbench: WorkbenchId, selectedO
  const known = systemCommands.has(commandId) || Object.values(workbenchDefinitions).some((definition) => definition.groups.some((group) => group.commands.some((command) => command.id === commandId)))
  if (!known) return { id: commandId, status: 'disabled', reason: 'Command is not registered in the active manifest.' }
  if (!implementedCommandIds.has(commandId)) return { id: commandId, status: 'disabled', reason: 'Command is visible in the FreeCAD-compatible manifest but its BitBybit business executor is not implemented yet.' }
-  if (commandId === 'pad' && activeWorkbench !== 'Part Design') return { id: commandId, status: 'disabled', reason: 'Switch to Part Design to use Pad.' }
+  if (partDesignCommands.has(commandId) && activeWorkbench !== 'Part Design') return { id: commandId, status: 'disabled', reason: `Switch to Part Design to use ${commandId}.` }
+  if ((commandId === 'new-sketch' || commandId === 'solve-sketch') && activeWorkbench !== 'Sketcher') return { id: commandId, status: 'disabled', reason: `Switch to Sketcher to use ${commandId}.` }
  if (selectionRequired.has(commandId) && !selectedObjectId) return { id: commandId, status: 'disabled', reason: 'Select a compatible object or sub-shape first.' }
  return { id: commandId, status: 'enabled' }
 }
--- a/src/facade/recomputeEngine.ts
+++ b/src/facade/recomputeEngine.ts
@@ -128,52 +128,61 @@ export class RecomputeCoordinator {
      }
    }

-    for (const objectId of plan.order) {
+    let processed = 0
+    for (const level of plan.levels) {
      if (controller.signal.aborted || this.active?.generation !== generation) return terminalResult('cancelled')
      if (this.currentDocumentVersion(document.id) !== document.version) return terminalResult('stale')

-      const failedDependency = graph.dependenciesOf(objectId).find((dependencyId) => objectStates[dependencyId] === 'error' || objectStates[dependencyId] === 'upstream-failed')
-      if (failedDependency) {
-        objectStates[objectId] = 'upstream-failed'
-        skipped.push(objectId)
-        errors.push({ objectId, code: 'UPSTREAM_FAILED', message: `Dependency ${failedDependency} did not recompute successfully.` })
-        options.onProgress?.({ generation, documentVersion: document.version, objectId, completed: completed.length, total: plan.order.length, state: 'upstream-failed' })
-        continue
-      }
+      // Nodes in a level have no dependencies on each other. Execute them together,
+      // then merge outcomes in plan order so persistence and UI events stay deterministic.
+      const outcomes = await Promise.all(level.map(async (objectId) => {
+        const failedDependency = graph.dependenciesOf(objectId).find((dependencyId) => objectStates[dependencyId] === 'error' || objectStates[dependencyId] === 'upstream-failed')
+        if (failedDependency) return { objectId, state: 'upstream-failed' as const, error: { objectId, code: 'UPSTREAM_FAILED', message: `Dependency ${failedDependency} did not recompute successfully.` } }

-      const object = objectById.get(objectId)
-      if (!object) {
-        objectStates[objectId] = 'error'
-        failed.push(objectId)
-        errors.push({ objectId, code: 'OBJECT_NOT_FOUND', message: `Document object does not exist: ${objectId}` })
-        continue
-      }
+        const object = objectById.get(objectId)
+        if (!object) return { objectId, state: 'error' as const, error: { objectId, code: 'OBJECT_NOT_FOUND', message: `Document object does not exist: ${objectId}` } }

-      try {
-        const result = await this.executeNode(object, document, {
-          documentId: document.id,
-          documentVersion: document.version,
-          generation,
-          signal: controller.signal,
-        })
-        if (controller.signal.aborted || this.active?.generation !== generation) return terminalResult('cancelled')
-        if (this.currentDocumentVersion(document.id) !== document.version) return terminalResult('stale')
-        if (result.status === 'failed') {
-          objectStates[objectId] = 'error'
-          failed.push(objectId)
-          errors.push(...(result.errors?.length ? result.errors : [{ objectId, code: 'RECOMPUTE_FAILED', message: `${objectId} failed to recompute.` }]))
-        } else {
-          objectStates[objectId] = 'up-to-date'
-          completed.push(objectId)
-          if (result.updatedObject) objectUpdates.push(result.updatedObject)
+        try {
+          const result = await this.executeNode(object, document, {
+            documentId: document.id,
+            documentVersion: document.version,
+            generation,
+            signal: controller.signal,
+          })
+          if (result.status === 'failed') {
+            return {
+              objectId,
+              state: 'error' as const,
+              errors: result.errors?.length ? result.errors : [{ objectId, code: 'RECOMPUTE_FAILED', message: `${objectId} failed to recompute.` }],
+            }
+          }
+          return { objectId, state: 'up-to-date' as const, updatedObject: result.updatedObject }
+        } catch (error) {
+          if (controller.signal.aborted || isAbortError(error)) return { objectId, state: 'cancelled' as const }
+          return { objectId, state: 'error' as const, error: { objectId, code: 'RECOMPUTE_EXCEPTION', message: error instanceof Error ? error.message : String(error) } }
        }
-      } catch (error) {
-        if (controller.signal.aborted || isAbortError(error)) return terminalResult('cancelled')
-        objectStates[objectId] = 'error'
-        failed.push(objectId)
-        errors.push({ objectId, code: 'RECOMPUTE_EXCEPTION', message: error instanceof Error ? error.message : String(error) })
+      }))
+
+      if (controller.signal.aborted || this.active?.generation !== generation) return terminalResult('cancelled')
+      if (this.currentDocumentVersion(document.id) !== document.version) return terminalResult('stale')
+
+      for (const outcome of outcomes) {
+        if (outcome.state === 'cancelled') return terminalResult('cancelled')
+        objectStates[outcome.objectId] = outcome.state
+        processed += 1
+        if (outcome.state === 'up-to-date') {
+          completed.push(outcome.objectId)
+          if (outcome.updatedObject) objectUpdates.push(outcome.updatedObject)
+        } else if (outcome.state === 'upstream-failed') {
+          skipped.push(outcome.objectId)
+          if (outcome.error) errors.push(outcome.error)
+        } else {
+          failed.push(outcome.objectId)
+          if ('errors' in outcome && outcome.errors) errors.push(...outcome.errors)
+          else if (outcome.error) errors.push(outcome.error)
+        }
+        options.onProgress?.({ generation, documentVersion: document.version, objectId: outcome.objectId, completed: processed, total: plan.order.length, state: outcome.state })
      }
-      options.onProgress?.({ generation, documentVersion: document.version, objectId, completed: completed.length, total: plan.order.length, state: objectStates[objectId] })
    }

    if (this.active?.generation === generation) this.active = null
--- a/tests/facade.test.ts
+++ b/tests/facade.test.ts
@@ -277,6 +277,50 @@ test('generation-aware recompute cancels an older run before accepting its resul
  version = 2
 })

+test('recompute executes independent dependency levels concurrently and merges in stable order', async () => {
+  const document: DocumentSnapshot = {
+    ...recomputeDocumentFixture(),
+    tree: [
+      { id: 'root', label: 'Root', type: 'feature', state: 'dirty' },
+      { id: 'left', label: 'Left', type: 'feature', state: 'dirty' },
+      { id: 'right', label: 'Right', type: 'feature', state: 'dirty' },
+    ],
+    objects: [
+      { id: 'root', typeId: 'Part::Feature', properties: [] },
+      { id: 'left', typeId: 'Part::Feature', properties: [] },
+      { id: 'right', typeId: 'Part::Feature', properties: [] },
+    ],
+    dependencies: [
+      { sourceId: 'left', targetId: 'root', relation: 'link' },
+      { sourceId: 'right', targetId: 'root', relation: 'link' },
+    ],
+    recompute: {
+      generation: 0,
+      status: 'idle',
+      objectStates: { root: 'touched', left: 'touched', right: 'touched' },
+      dirtyObjects: ['root'],
+      order: [],
+      errors: [],
+    },
+  }
+  let active = 0
+  let maximumActive = 0
+  const coordinator = new RecomputeCoordinator(async (object) => {
+    active += 1
+    maximumActive = Math.max(maximumActive, active)
+    await new Promise((resolve) => setTimeout(resolve, 5))
+    active -= 1
+    return { status: 'success', updatedObject: object }
+  }, () => 1)
+
+  const result = await coordinator.run(document)
+  assert.equal(result.status, 'completed')
+  assert.equal(maximumActive, 2)
+  assert.deepEqual(result.order, ['root', 'left', 'right'])
+  assert.deepEqual(result.completed, ['root', 'left', 'right'])
+  assert.deepEqual(result.levels, [['root'], ['left', 'right']])
+})
+
 test('recompute rejects results from an obsolete document version', async () => {
  let version = 1
  const coordinator = new RecomputeCoordinator(async () => {
@@ -395,6 +439,17 @@ test('manifest commands without a Bitbybit executor are explicitly unsupported',
  assert.equal(facade.getState().document.version, before)
 })

+test('implemented commands enforce their FreeCAD workbench preconditions', () => {
+  const facade = createMockFacade()
+  facade.gui.workbench.setActive('Sketcher')
+  assert.equal(facade.gui.command.getState('create-body').status, 'disabled')
+  assert.match(facade.gui.command.getState('create-body').reason || '', /Part Design/)
+  assert.equal(facade.gui.command.getState('new-sketch').status, 'enabled')
+  facade.gui.workbench.setActive('Part Design')
+  assert.equal(facade.gui.command.getState('new-sketch').status, 'disabled')
+  assert.equal(facade.gui.command.getState('pocket').status, 'enabled')
+})
+
 test('command events carry a document-scoped request context', () => {
  const facade = createMockFacade()
  const events: string[] = []