Skip to content

Implement unimplemented concurrency methods for all server classes #2012

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.

Sign up for GitHub

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

Open
alerickson wants to merge 11 commits into
base: master
Choose a base branch
from
Open

Implement unimplemented concurrency methods for all server classes #2012

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
ba2d230
Implement unimplemented concurrency methods for all server classes
alerickson Jun 30, 2026
fbb940f
build fixes
alerickson Jun 30, 2026
a4730af
Fix thread-safety in ContainerRegistryServerAPICalls.InstallPackageAsync
Copilot Jul 1, 2026
f800970
Fix copilot commit for container registry
alerickson Jul 1, 2026
6ab7367
Avoid cmdlet stream writes in NuGet FindVersionAsync path
Copilot Jul 1, 2026
dbd1689
Fix V3 async helper logging to use debug queues
Copilot Jul 1, 2026
efb000d
Potential fix for pull request finding
alerickson Jul 1, 2026
0e9c684
Fix specific-version async dependency error handling path
Copilot Jul 1, 2026
e3c1af5
Scope async dependency queue handling to current operation
Copilot Jul 1, 2026
9b92e77
Fix version-range async path: reset errRecord and flush concurrent qu…
Copilot Jul 1, 2026
900db22
build fixes
alerickson Jul 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 49 additions & 4 deletions src/code/ContainerRegistryServerAPICalls.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,14 +82,40 @@ public ContainerRegistryServerAPICalls(PSRepositoryInfo repository, PSCmdlet cmd

#region Overridden Methods

/// <summary>
/// Async find method which allows for searching for single name with specific version.
/// Name: no wildcard support
/// Version: no wildcard support
/// This is the concurrent (parallel) counterpart of FindVersion().
/// </summary>
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);
}

/// <summary>
/// Async find method which allows for searching for single name with version range.
/// Name: no wildcard support
/// Version: supports wildcards
/// This is the concurrent (parallel) counterpart of FindVersionGlobbing().
/// </summary>
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 ContainerRegistryServerAPICalls.");
debugMsgs.Enqueue("In ContainerRegistryServerAPICalls::FindVersionGlobbingAsync()");
FindResults findResponse = FindVersionGlobbing(packageName, versionRange, includePrerelease, type, getOnlyLatest, out ErrorRecord errRecord);
if (errRecord != null)
{
errorMsgs.Enqueue(errRecord);
}

return Task.FromResult(findResponse);
}

/// <summary>
Expand Down Expand Up @@ -158,9 +184,21 @@ public override FindResults FindName(string packageName, bool includePrerelease,
}


/// <summary>
/// Async find method which allows for searching for single name and returns latest version.
/// Name: no wildcard support
/// This is the concurrent (parallel) counterpart of FindName().
/// </summary>
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 ContainerRegistryServerAPICalls.");
debugMsgs.Enqueue("In ContainerRegistryServerAPICalls::FindNameAsync()");
FindResults findResponse = FindName(packageName, includePrerelease, type, out ErrorRecord errRecord);
if (errRecord != null)
{
errorMsgs.Enqueue(errRecord);
}

return Task.FromResult(findResponse);
}

/// <summary>
Expand Down Expand Up @@ -329,7 +367,14 @@ public override Stream InstallPackage(string packageName, string packageVersion,
/// </summary>
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("FindNameAsync is not implemented for ContainerRegistryServerAPICalls.");
debugMsgs.Enqueue("In ContainerRegistryServerAPICalls::InstallPackageAsync()");
Stream results = InstallPackage(packageName, packageVersion, includePrerelease, out ErrorRecord errRecord);
if (errRecord != null)
Comment thread
alerickson marked this conversation as resolved.
Outdated
{
errorMsgs.Enqueue(errRecord);
}

return Task.FromResult(results);
}

/// <summary>
Expand Down
88 changes: 28 additions & 60 deletions src/code/FindHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -912,17 +912,12 @@ private IEnumerable<PSResourceInfo> SearchByNames(ServerApiCall currentServer, R

ConcurrentDictionary<string, Task<FindResults>> cachedNetworkCalls = new ConcurrentDictionary<string, Task<FindResults>>();
Task<FindResults> response = null;
if (currentServer.Repository.ApiVersion == PSRepositoryInfo.APIVersion.V2) {
string key = $"{pkgName}|{_nugetVersion.ToNormalizedString()}|{_type}";
response = cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindVersionAsync(pkgName, _nugetVersion.ToNormalizedString(), _type, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

responses = response.GetAwaiter().GetResult();
string key = $"{pkgName}|{_nugetVersion.ToNormalizedString()}|{_type}";
response = cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindVersionAsync(pkgName, _nugetVersion.ToNormalizedString(), _type, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

Comment thread
alerickson marked this conversation as resolved.
responses = response.GetAwaiter().GetResult();

Utils.WriteOutConcurrentQueue(_cmdletPassedIn, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
}
else {
responses = currentServer.FindVersion(pkgName, _nugetVersion.ToNormalizedString(), _type, out errRecord);
}
Utils.WriteOutConcurrentQueue(_cmdletPassedIn, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
}
else
{
Expand Down Expand Up @@ -996,15 +991,10 @@ private IEnumerable<PSResourceInfo> SearchByNames(ServerApiCall currentServer, R
{
ConcurrentDictionary<string, Task<FindResults>> cachedNetworkCalls = new ConcurrentDictionary<string, Task<FindResults>>();
Task<FindResults> response = null;
if (currentServer.Repository.ApiVersion == PSRepositoryInfo.APIVersion.V2) {
string key = $"{pkgName}|{_versionRange.ToString()}|{_type}";
response = cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindVersionGlobbingAsync(pkgName, _versionRange, _prerelease, _type, getOnlyLatest: false, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

responses = response.GetAwaiter().GetResult();
}
else {
responses = currentServer.FindVersionGlobbing(pkgName, _versionRange, _prerelease, _type, getOnlyLatest: false, out errRecord);
}
string key = $"{pkgName}|{_versionRange.ToString()}|{_type}";
response = cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindVersionGlobbingAsync(pkgName, _versionRange, _prerelease, _type, getOnlyLatest: false, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

Comment thread
alerickson marked this conversation as resolved.
responses = response.GetAwaiter().GetResult();
}
else
{
Expand Down Expand Up @@ -1189,7 +1179,7 @@ internal void FindDependencyPackagesHelper(ServerApiCall currentServer, Response
//const int PARALLEL_THRESHOLD = 5; // TODO: Trottle limit from user, defaults to 5;
int processorCount = Environment.ProcessorCount;
int maxDegreeOfParallelism = processorCount * 4;
if (currentServer.Repository.ApiVersion == PSRepositoryInfo.APIVersion.V2 && currentPkg.Dependencies.Length > processorCount)
if (currentPkg.Dependencies.Length > processorCount)
{
Comment thread
alerickson marked this conversation as resolved.
Parallel.ForEach(currentPkg.Dependencies, new ParallelOptions { MaxDegreeOfParallelism = maxDegreeOfParallelism }, dep =>
{
Expand Down Expand Up @@ -1293,20 +1283,13 @@ private PSResourceInfo FindDependencyWithSpecificVersion(
Task<FindResults> response = null;
debugMsgs.Enqueue("In FindHelper::FindDependencyWithSpecificVersion()");

if (currentServer.Repository.ApiVersion == PSRepositoryInfo.APIVersion.V2)
{
// See if the network call we're making is already cached, if not, call FindNameAsync() and cache results
string key = $"{dep.Name}|{dep.VersionRange.MaxVersion.ToString()}|{_type}";
debugMsgs.Enqueue("Checking if network call is cached.");
response = _cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindVersionAsync(dep.Name, dep.VersionRange.MaxVersion.ToString(), _type, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

responses = response.GetAwaiter().GetResult();
}
else
{
responses = currentServer.FindVersion(dep.Name, dep.VersionRange.MaxVersion.ToString(), _type, out errRecord);
}


// See if the network call we're making is already cached, if not, call FindNameAsync() and cache results
string key = $"{dep.Name}|{dep.VersionRange.MaxVersion.ToString()}|{_type}";
debugMsgs.Enqueue("Checking if network call is cached.");
response = _cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindVersionAsync(dep.Name, dep.VersionRange.MaxVersion.ToString(), _type, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

responses = response.GetAwaiter().GetResult();
Comment thread
alerickson marked this conversation as resolved.
Outdated

// Error handling and Convert to PSResource object
if (errRecord != null)
Expand Down Expand Up @@ -1369,19 +1352,12 @@ private PSResourceInfo FindDependencyWithLowerBound(
Task<FindResults> response = null;
debugMsgs.Enqueue("In FindHelper::FindDependencyWithLowerBound()");

if (currentServer.Repository.ApiVersion == PSRepositoryInfo.APIVersion.V2)
{
// See if the network call we're making is already cached, if not, call FindNameAsync() and cache results
string key = $"{dep.Name}|*|{_type}";
debugMsgs.Enqueue("Checking if network call is cached.");
response = _cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindNameAsync(dep.Name, includePrerelease: true, _type, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

responses = response.GetAwaiter().GetResult();
}
else
{
responses = currentServer.FindName(dep.Name, includePrerelease: true, _type, out errRecord);
}
// See if the network call we're making is already cached, if not, call FindNameAsync() and cache results
string key = $"{dep.Name}|*|{_type}";
debugMsgs.Enqueue("Checking if network call is cached.");
response = _cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindNameAsync(dep.Name, includePrerelease: true, _type, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

responses = response.GetAwaiter().GetResult();

// Error handling and Convert to PSResource object
if (errRecord != null)
Expand Down Expand Up @@ -1445,21 +1421,13 @@ private PSResourceInfo FindDependencyWithUpperBound(

ConcurrentDictionary<string, Task<FindResults>> cachedNetworkCalls = new ConcurrentDictionary<string, Task<FindResults>>();
debugMsgs.Enqueue("In FindHelper::FindDependencyWithUpperBound()");
// See if the network call we're making is already caced, if not, call FindNameAsync() and cache results
Comment thread
Copilot marked this conversation as resolved.
Outdated
string key = $"{dep.Name}|{dep.VersionRange.MaxVersion.ToString()}|{_type}";
debugMsgs.Enqueue("Checking if network call is cached.");
response = cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindVersionGlobbingAsync(dep.Name, dep.VersionRange, includePrerelease: true, ResourceType.None, getOnlyLatest: true, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

if (currentServer.Repository.ApiVersion == PSRepositoryInfo.APIVersion.V2)
{
// See if the network call we're making is already caced, if not, call FindNameAsync() and cache results
string key = $"{dep.Name}|{dep.VersionRange.MaxVersion.ToString()}|{_type}";
debugMsgs.Enqueue("Checking if network call is cached.");
response = cachedNetworkCalls.GetOrAdd(key, _ => currentServer.FindVersionGlobbingAsync(dep.Name, dep.VersionRange, includePrerelease: true, ResourceType.None, getOnlyLatest: true, errorMsgs, warningMsgs, debugMsgs, verboseMsgs));

responses = response.GetAwaiter().GetResult();
responses = response.GetAwaiter().GetResult();

}
else
{
responses = currentServer.FindVersionGlobbing(dep.Name, dep.VersionRange, includePrerelease: true, ResourceType.None, getOnlyLatest: true, out errRecord);
}

// Error handling and Convert to PSResource object
if (errRecord != null)
Expand Down
4 changes: 2 additions & 2 deletions src/code/InstallHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -801,7 +801,7 @@ private ConcurrentDictionary<string, Hashtable> BeginPackageInstall(
}
else
{
// Concurrent updates, currently only implemented for v2 server repositories
// Concurrent updates
// Find all dependencies
if (!skipDependencyCheck)
{
Expand Down Expand Up @@ -853,7 +853,7 @@ private ConcurrentDictionary<string, Hashtable> InstallParentAndDependencyPackag
// TODO: figure out a good threshold and parallel count
int processorCount = Environment.ProcessorCount;
_cmdletPassedIn.WriteDebug($"parentAndDeps.Count is {parentAndDeps.Count}, processor count is: {processorCount}");
if (currentServer.Repository.ApiVersion == PSRepositoryInfo.APIVersion.V2 && parentAndDeps.Count > processorCount)
if (parentAndDeps.Count > processorCount)
Comment thread
alerickson marked this conversation as resolved.
{
_cmdletPassedIn.WriteDebug($"parentAndDeps.Count is greater than processor count");
// Set the maximum degree of parallelism to 32? (Invoke-Command has default of 32, that's where we got this number from)
Expand Down
65 changes: 61 additions & 4 deletions src/code/LocalServerApiCalls.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,14 +41,40 @@ public LocalServerAPICalls (PSRepositoryInfo repository, PSCmdlet cmdletPassedIn

#region Overridden Methods

/// <summary>
/// Async find method which allows for searching for single name with specific version.
/// Name: no wildcard support
/// Version: no wildcard support
/// This is the concurrent (parallel) counterpart of FindVersion().
/// </summary>
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();
debugMsgs.Enqueue("In LocalServerApiCalls::FindVersionAsync()");
FindResults findResponse = FindVersionHelper(packageName, version, tags: Utils.EmptyStrArray, type, out ErrorRecord errRecord);
if (errRecord != null)
{
errorMsgs.Enqueue(errRecord);
}

return Task.FromResult(findResponse);
}

/// <summary>
/// Async find method which allows for searching for single name with version range.
/// Name: no wildcard support
/// Version: supports wildcards
/// This is the concurrent (parallel) counterpart of FindVersionGlobbing().
/// </summary>
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();
debugMsgs.Enqueue("In LocalServerApiCalls::FindVersionGlobbingAsync()");
FindResults findResponse = FindVersionGlobbing(packageName, versionRange, includePrerelease, type, getOnlyLatest, out ErrorRecord errRecord);
if (errRecord != null)
{
errorMsgs.Enqueue(errRecord);
}

return Task.FromResult(findResponse);
}
/// <summary>
/// Find method which allows for searching for all packages from a repository and returns latest version for each.
Expand Down Expand Up @@ -124,9 +150,21 @@ public override FindResults FindName(string packageName, bool includePrerelease,
return FindNameHelper(packageName, Utils.EmptyStrArray, includePrerelease, type, out errRecord);
}

/// <summary>
/// Async find method which allows for searching for single name and returns latest version.
/// Name: no wildcard support
/// This is the concurrent (parallel) counterpart of FindName().
/// </summary>
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();
debugMsgs.Enqueue("In LocalServerApiCalls::FindNameAsync()");
FindResults findResponse = FindNameHelper(packageName, tags: Utils.EmptyStrArray, includePrerelease, type, out ErrorRecord errRecord);
if (errRecord != null)
{
errorMsgs.Enqueue(errRecord);
}

return Task.FromResult(findResponse);
}

/// <summary>
Expand Down Expand Up @@ -278,7 +316,26 @@ public override Stream InstallPackage(string packageName, string packageVersion,
/// </summary>
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 LocalServerAPICalls.");
debugMsgs.Enqueue("In LocalServerApiCalls::InstallPackageAsync()");
Stream results = new MemoryStream();
if (string.IsNullOrEmpty(packageVersion))
{
errorMsgs.Enqueue(new ErrorRecord(
exception: new ArgumentNullException($"Package version could not be found for {packageName}"),
"PackageVersionNullOrEmptyError",
ErrorCategory.InvalidArgument,
_cmdletPassedIn));

return Task.FromResult(results);
}

results = InstallVersion(packageName, packageVersion, out ErrorRecord errRecord);
if (errRecord != null)
{
errorMsgs.Enqueue(errRecord);
}

return Task.FromResult(results);
}

#endregion
Expand Down
Loading
Loading