Skip to content

refactor: use queue-based concurrency coordinator - #623

Merged
zhongkechen merged 3 commits into
mainfrom
codex/queue-concurrency-coordinator
Aug 19, 2026
Merged

refactor: use queue-based concurrency coordinator#623
zhongkechen merged 3 commits into
mainfrom
codex/queue-concurrency-coordinator

Conversation

@zhongkechen

Copy link
Copy Markdown
Contributor

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Issue Link, if available

N/A

Description

Replace the concurrency operation's rebuilt CompletableFuture.anyOf coordination with a persistent event queue modeled after the Python SDK coordinator.

  • Give one coordinator exclusive ownership of running children and completion counters.
  • Publish child completion, state-change, and terminal failure events through a blocking queue.
  • Preserve FIFO branch admission and bounded concurrency.
  • Keep the synchronized active-thread handoff required to avoid lost wakeups during suspension.
  • Add regression coverage for queue admission and exceptional completion while the coordinator is waiting.

Demo/Screenshots

N/A - internal coordination refactor with no user-facing UI changes.

Checklist

  • I have filled out every section of the PR template
  • I have thoroughly tested this change

Testing

Unit Tests

Yes. Added coordinator queue admission and exceptional wakeup tests. Ran the full SDK unit suite: 1,139 tests passed.

Integration Tests

Existing map and parallel integration coverage was run: 169 tests passed, including replay, suspension, failures, nesting modes, early completion, and bounded concurrency.

Examples

Not applicable. This change does not alter the public API or add a customer-facing workflow.

@zhongkechen
zhongkechen requested a review from a team August 14, 2026 05:59
@zhongkechen
zhongkechen temporarily deployed to ai-pr-review-runtime August 14, 2026 05:59 — with GitHub Actions Inactive
}

try {
return coordinatorEvents.take();

This comment was marked as outdated.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for flagging this. I added singleWorkerForkJoinPoolDoesNotStarveCoordinator in ParallelIntegrationTest using a custom ForkJoinPool(1), two parallel branches, and a five-second timeout. It completes successfully, and the full ParallelIntegrationTest suite passes (65 tests).

On Corretto/OpenJDK 17, LinkedBlockingQueue.take() waits through an AQS ConditionObject; its condition node implements ForkJoinPool.ManagedBlocker, and ConditionObject.await() invokes ForkJoinPool.managedBlock. The pool therefore compensates for the blocked coordinator even though this call site does not invoke managedBlock explicitly. Wrapping take() in another managed blocker would be redundant for the supported runtime.

The regression coverage is in commit 12a99c2.

@github-actions

This comment has been minimized.

@zhongkechen
zhongkechen temporarily deployed to ai-pr-review-runtime August 14, 2026 06:48 — with GitHub Actions Inactive
@github-actions

This comment has been minimized.

@zhongkechen
zhongkechen temporarily deployed to ai-pr-review-runtime August 19, 2026 19:15 — with GitHub Actions Inactive
@zhongkechen
zhongkechen merged commit fb3507c into main Aug 19, 2026
8 of 12 checks passed
@zhongkechen
zhongkechen deleted the codex/queue-concurrency-coordinator branch August 19, 2026 19:39
}

try {
return coordinatorEvents.take();

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Codex AI review

[P1] Avoid unmanaged blocking on the configured ForkJoinPool. On Java 17, LinkedBlockingQueue.take() is not a ForkJoinPool.ManagedBlocker. With the newly tested ForkJoinPool(1), the handler's CompletableFuture.join() creates one compensating worker for this coordinator, which then parks here while the branch tasks remain queued, so the test times out and execution deadlocks. Use ForkJoinPool.managedBlock for this wait or restore a CompletableFuture-based event signal that allows pool compensation.

@github-actions

Copy link
Copy Markdown
Contributor

Codex AI review

Found one high-severity coordinator liveness regression. Tests were not executed, as requested.

Reviewed commit 35c961ec327a9fb3afe7d8fe9eab2797b553b8f0. Workflow run

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants