-
Notifications
You must be signed in to change notification settings - Fork 6k
perf(acp): parallelize extension loading in ACP server #8098
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -756,6 +756,76 @@ impl Agent { | |
| Ok(()) | ||
| } | ||
|
|
||
| /// Load multiple extensions in parallel, persisting state once at the end. | ||
| /// | ||
| /// Unlike `add_extension`, this avoids per-extension persistence and acquires | ||
| /// the container lock once upfront to prevent serialisation of the parallel futures. | ||
| pub async fn add_extensions_bulk( | ||
| self: &Arc<Self>, | ||
| extensions: Vec<ExtensionConfig>, | ||
| session_id: &str, | ||
| ) -> Vec<ExtensionLoadResult> { | ||
| // Resolve session working_dir and container once, before spawning futures, | ||
| // so each future doesn't re-acquire the container lock. | ||
| let working_dir = match self | ||
| .config | ||
| .session_manager | ||
| .get_session(session_id, false) | ||
| .await | ||
| { | ||
| Ok(session) => Some(session.working_dir), | ||
| Err(e) => { | ||
| warn!("Failed to get session for bulk load: {}", e); | ||
| None | ||
| } | ||
| }; | ||
| let container = self.container.lock().await.clone(); | ||
|
|
||
| let extension_futures = extensions | ||
| .into_iter() | ||
| .map(|config| { | ||
| let ext_manager = Arc::clone(&self.extension_manager); | ||
| let working_dir = working_dir.clone(); | ||
| let container = container.clone(); | ||
| let sid = session_id.to_string(); | ||
|
|
||
| async move { | ||
| let name = config.name().to_string(); | ||
| match ext_manager | ||
| .add_extension(config, working_dir, container.as_ref(), Some(&sid)) | ||
| .await | ||
| { | ||
| Ok(_) => ExtensionLoadResult { | ||
| name, | ||
| success: true, | ||
| error: None, | ||
| }, | ||
| Err(e) => { | ||
| let error_msg = e.to_string(); | ||
| warn!("Failed to load extension {}: {}", name, error_msg); | ||
| ExtensionLoadResult { | ||
| name, | ||
| success: false, | ||
| error: Some(error_msg), | ||
| } | ||
| } | ||
| } | ||
| } | ||
| }) | ||
| .collect::<Vec<_>>(); | ||
|
|
||
| let results = futures::future::join_all(extension_futures).await; | ||
|
|
||
| // Persist once after all extensions are loaded | ||
| if results.iter().any(|r| r.success) { | ||
| if let Err(e) = self.persist_extension_state(session_id).await { | ||
| warn!("Failed to persist extension state after bulk load: {}", e); | ||
| } | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Useful? React with 👍 / 👎. |
||
| } | ||
|
|
||
| results | ||
| } | ||
|
|
||
| async fn add_extension_inner( | ||
| &self, | ||
| extension: ExtensionConfig, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Loading all configs via
join_allhere makes startup order-dependent when the same extension key appears more than once (e.g., a builtin also enabled in config).create_agent_for_sessionappendsself.builtinsto config-derived extensions, andExtensionManager::add_extensiondoes a pre-check and insert in separate phases, so duplicate keys can initialize concurrently and whichever future finishes last overwrites the other. The old sequential loop had deterministic behavior; this change introduces nondeterministic final extension config/tool state across runs.Useful? React with 👍 / 👎.