WIP: Add concurrency for parent packages#2016
Open
alerickson wants to merge 13 commits into
Open
Conversation
InstallPackageAsync was calling the synchronous InstallPackage() which writes to _cmdletPassedIn.WriteDebug, causing cross-thread cmdlet stream writes when used from Parallel.ForEach. - Refactor InstallPackageAsync to not call InstallPackage(); inline the null-version check and enqueue all messages via the provided queues - Add a queue-aware InstallVersion overload that uses ConcurrentQueue parameters instead of _cmdletPassedIn.Write* calls, for use by the async install path - Keep the original InstallVersion(out ErrorRecord) overload intact for the synchronous path
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
alerickson
requested review from
SydneyhSmith,
adityapatwardhan,
anamnavi and
shammu1
as code owners
July 24, 2026 23:11
|
Azure Pipelines: There may be pipelines that require an authorized user to comment /azp run to run. |
Member
Author
|
/azp run |
|
Azure Pipelines: Successfully started running 1 pipeline(s). |
There was a problem hiding this comment.
Pull request overview
This PR is working toward enabling concurrent (parallel) package resolution/installation workflows by adding async server API entry points and refactoring parent-package installation to separate pipeline-thread operations (ShouldProcess/host output) from worker-thread network/download work.
Changes:
- Implemented async
Find*Async/InstallPackageAsyncoverrides across multiple server API call implementations. - Refactored
InstallHelper.InstallPackages()into a 3-phase model (pipeline-thread gating, parallel downloads, pipeline-thread finalization) with per-parent message queues. - Updated dependency resolution in
FindHelperto allow concurrent dependency-closure computation and to reduce shared mutable state.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 10 comments.
Show a summary per file
| File | Description |
|---|---|
| src/code/V3ServerAPICalls.cs | Implements async find/install paths and routes diagnostics through concurrent queues for V3 feeds. |
| src/code/V2ServerAPICalls.cs | Adjusts async pagination loop to use the async helper for subsequent pages. |
| src/code/NuGetServerAPICalls.cs | Adds async find/install methods and a queue-based HTTP helper for some async paths. |
| src/code/LocalServerApiCalls.cs | Adds basic async wrappers for local repository find/install operations. |
| src/code/InstallHelper.cs | Refactors parent install flow into phases and introduces parallel parent/dependency download work items. |
| src/code/FindHelper.cs | Enables concurrent dependency resolution and attempts to localize state for thread safety. |
| src/code/ContainerRegistryServerAPICalls.cs | Adds async wrappers for container registry find/install operations. |
Comments suppressed due to low confidence (1)
src/code/FindHelper.cs:1222
- This method drains queued messages by calling Utils.WriteOutConcurrentQueue(_cmdletPassedIn, ...) inside FindHelper. When invoked from worker threads (e.g., during parallel parent installs), this will write to cmdlet streams off the pipeline thread. Message draining should be done by the top-level pipeline-thread caller (like InstallHelper Phase 3), not inside this helper.
}
Utils.WriteOutConcurrentQueue(_cmdletPassedIn, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
}
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+66
to
+69
| // We need to explicitly add 'Id eq <packageName>' whenever $filter is used, otherwise arbitrary results are returned. | ||
| filterBuilder.AddCriterion($"Id eq '{packageName}'"); | ||
| filterBuilder.AddCriterion($"NormalizedVersion eq '{packageName}'"); | ||
|
|
| return new FindResults(stringResponse: Utils.EmptyStrArray, hashtableResponse: emptyHashResponses, responseType: v3FindResponseType); | ||
| } | ||
| _cmdletPassedIn.WriteDebug($"'{packageName}' version parsed as '{requiredVersion}'"); | ||
| //_cmdletPassedIn.WriteDebug($"'{packageName}' version parsed as '{requiredVersion}'"); |
Comment on lines
1742
to
1746
| debugMsgs.Enqueue($"Count is '{count}'"); | ||
| // skip 100 | ||
| skip += 100; | ||
| // TODO: this should be an async method | ||
| var tmpResponse = FindVersionGlobbing(packageName, versionRange, includePrerelease, type, skip, getOnlyLatest, out ErrorRecord errRecord); | ||
| if (errRecord != null) | ||
| { | ||
| Utils.EnqueueIfNotNull(errorMsgs, errRecord); | ||
| return new FindResults(stringResponse: Utils.EmptyStrArray, hashtableResponse: emptyHashResponses, responseType: v2FindResponseType); | ||
| } | ||
| var tmpResponse = await FindVersionGlobbingAsync(packageName, versionRange, includePrerelease, type, skip, getOnlyLatest, errorMsgs, warningMsgs, debugMsgs, verboseMsgs); | ||
| responses.Add(tmpResponse); |
Comment on lines
921
to
929
| var pkgToInstallName = pkgToBeInstalled.Name; | ||
| var pkgToInstallVersion = Utils.GetFullVersionString(pkgToBeInstalled.Version.ToString(), pkgToBeInstalled.Prerelease); | ||
| Stream responseStream = currentServer.InstallPackage(pkgToInstallName, pkgToInstallVersion, true, out ErrorRecord installNameErrRecord); | ||
|
|
||
| if (installNameErrRecord != null) | ||
| { | ||
| _cmdletPassedIn.WriteError(installNameErrRecord); | ||
| errorMsgs.Enqueue(installNameErrRecord); | ||
| return packagesHash; | ||
| } |
Comment on lines
+1180
to
+1184
| // Use a local instance so multiple parent packages can resolve their dependency closures concurrently | ||
| // without racing on shared state. | ||
| ConcurrentDictionary<string, PSResourceInfo> depPkgsFound = new ConcurrentDictionary<string, PSResourceInfo>(); | ||
| _cmdletPassedIn.WriteDebug($"In FindHelper::FindDependencyPackages() - {currentPkg.Name}"); | ||
| FindDependencyPackagesHelper(currentServer, currentResponseUtil, currentPkg, repository); | ||
| FindDependencyPackagesHelper(currentServer, currentResponseUtil, currentPkg, repository, depPkgsFound); |
Comment on lines
521
to
+530
| public override Task<Stream> InstallPackageAsync(string packageName, string packageVersion, bool includePrerelease, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs) | ||
| { | ||
| throw new NotImplementedException("InstallPackageAsync is not implemented for NuGetServerAPICalls."); | ||
| debugMsgs.Enqueue("In NuGetServerAPICalls::InstallPackageAsync()"); | ||
| Stream results = InstallPackage(packageName, packageVersion, includePrerelease, out ErrorRecord errRecord); | ||
| if (errRecord != null) | ||
| { | ||
| errorMsgs.Enqueue(errRecord); | ||
| } | ||
|
|
||
| return Task.FromResult(results); |
Comment on lines
906
to
910
| } | ||
| }); | ||
|
|
||
| Utils.WriteOutConcurrentQueue(_cmdletPassedIn, errorMsgs, warningMsgs, debugMsgs, verboseMsgs); | ||
| if (errorMsgs.Count > 0) | ||
| if (!errorMsgs.IsEmpty) | ||
| { |
Comment on lines
238
to
248
| public override Task<FindResults> FindNameAsync(string packageName, bool includePrerelease, ResourceType type, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs) | ||
| { | ||
| throw new NotImplementedException("FindNameAsync is not implemented for NuGetServerAPICalls."); | ||
| debugMsgs.Enqueue("In NuGetServerAPICalls::FindNameAsync()"); | ||
| FindResults findResponse = FindName(packageName, includePrerelease, type, out ErrorRecord errRecord); | ||
| if (errRecord != null) | ||
| { | ||
| errorMsgs.Enqueue(errRecord); | ||
| } | ||
|
|
||
| return Task.FromResult(findResponse); | ||
| } |
Comment on lines
91
to
101
| public override Task<FindResults> FindVersionAsync(string packageName, string version, ResourceType type, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs) | ||
| { | ||
| throw new NotImplementedException("FindVersionAsync is not implemented for ContainerRegistryServerAPICalls."); | ||
| debugMsgs.Enqueue("In ContainerRegistryServerAPICalls::FindVersionAsync()"); | ||
| FindResults findResponse = FindVersion(packageName, version, type, out ErrorRecord errRecord); | ||
| if (errRecord != null) | ||
| { | ||
| errorMsgs.Enqueue(errRecord); | ||
| } | ||
|
|
||
| return Task.FromResult(findResponse); | ||
| } |
Comment on lines
87
to
+96
| public override Task<FindResults> FindVersionGlobbingAsync(string packageName, VersionRange versionRange, bool includePrerelease, ResourceType type, bool getOnlyLatest, ConcurrentQueue<ErrorRecord> errorMsgs, ConcurrentQueue<string> warningMsgs, ConcurrentQueue<string> debugMsgs, ConcurrentQueue<string> verboseMsgs) | ||
| { | ||
| throw new NotImplementedException("FindVersionGlobbingAsync is not implemented for NuGetServerAPICalls."); | ||
| debugMsgs.Enqueue("In NuGetServerAPICalls::FindVersionGlobbingAsync()"); | ||
| FindResults findResponse = FindVersionGlobbing(packageName, versionRange, includePrerelease, type, getOnlyLatest, out ErrorRecord errRecord); | ||
| if (errRecord != null) | ||
| { | ||
| errorMsgs.Enqueue(errRecord); | ||
| } | ||
|
|
||
| return Task.FromResult(findResponse); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
PR Summary
PR Context
PR Checklist
.h,.cpp,.cs,.ps1and.psm1files have the correct copyright headerWIP:or[ WIP ]to the beginning of the title (theWIPbot will keep its status check atPendingwhile the prefix is present) and remove the prefix when the PR is ready.