Add concurrency for parent packages

src/code/FindHelper.cs

@@ -46,10 +46,6 @@ internal class FindHelper
         // If running 'Install-PSResource Az, TestModule, NewTestModule', it will contain one parent and its dependencies.
         private ConcurrentDictionary<string, List<string>> _packagesFound;
 
-        // Creates a new instance of depPkgsFound each time FindDependencyPackages() is called.
-        // This will eventually return the PSResourceInfo object to the main cmdlet class.
-        private ConcurrentDictionary<string, PSResourceInfo> depPkgsFound;
-
         // Contains the latest found version of a particular package.
         private ConcurrentDictionary<string, PSResourceInfo> _knownLatestPkgVersion;
 
@@ -1060,13 +1056,36 @@ private IEnumerable<PSResourceInfo> SearchByNames(ServerApiCall currentServer, R
             // After retrieving all packages find their dependencies
             if (_includeDependencies)
             {
-                foreach (PSResourceInfo currentPkg in parentPkgs)
+                // Resolving each parent package's dependency closure is independent work, so do it concurrently.
+                // yield return cannot be used inside Parallel.ForEach, so collect results into a thread-safe bag first.
+                ConcurrentBag<PSResourceInfo> dependencyPkgs = new ConcurrentBag<PSResourceInfo>();
+                int processorCount = Environment.ProcessorCount;
+                int maxDegreeOfParallelism = processorCount * 4;
+                if (parentPkgs.Count > processorCount)
+                {
+                    Parallel.ForEach(parentPkgs, new ParallelOptions { MaxDegreeOfParallelism = maxDegreeOfParallelism }, currentPkg =>
+                    {
+                        foreach (PSResourceInfo pkgDep in FindDependencyPackages(currentServer, currentResponseUtil, currentPkg, repository))
+                        {
+                            dependencyPkgs.Add(pkgDep);
+                        }
+                    });
+                }
+                else
                 {
-                    foreach (PSResourceInfo pkgDep in FindDependencyPackages(currentServer, currentResponseUtil, currentPkg, repository))
+                    foreach (PSResourceInfo currentPkg in parentPkgs)
                     {
-                        yield return pkgDep;
+                        foreach (PSResourceInfo pkgDep in FindDependencyPackages(currentServer, currentResponseUtil, currentPkg, repository))
+                        {
+                            dependencyPkgs.Add(pkgDep);
+                        }
                     }
                 }
+
+                foreach (PSResourceInfo pkgDep in dependencyPkgs)
+                {
+                    yield return pkgDep;
+                }
             }
         }
 
@@ -1158,20 +1177,50 @@ private string FormatPkgVersionString(PSResourceInfo pkg)
 
         internal IEnumerable<PSResourceInfo> FindDependencyPackages(ServerApiCall currentServer, ResponseUtil currentResponseUtil, PSResourceInfo currentPkg, PSRepositoryInfo repository)
         {
-            depPkgsFound = new ConcurrentDictionary<string, PSResourceInfo>();
-            _cmdletPassedIn.WriteDebug($"In FindHelper::FindDependencyPackages() - {currentPkg.Name}");            
-            FindDependencyPackagesHelper(currentServer, currentResponseUtil, currentPkg, repository); 
+            // Pipeline-thread callers: collect diagnostics locally and drain them to the cmdlet on this thread.
+            ConcurrentQueue<ErrorRecord> errorMsgs = new ConcurrentQueue<ErrorRecord>();
+            ConcurrentQueue<string> warningMsgs = new ConcurrentQueue<string>();
+            ConcurrentQueue<string> debugMsgs = new ConcurrentQueue<string>();
+            ConcurrentQueue<string> verboseMsgs = new ConcurrentQueue<string>();
+
+            var depPkgs = FindDependencyPackages(currentServer, currentResponseUtil, currentPkg, repository, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
+
+            Utils.WriteOutConcurrentQueue(_cmdletPassedIn, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
+            return depPkgs;
+        }
+
+        // Overload for worker-thread callers: diagnostics are routed to the caller-provided queues and drained by the caller on the pipeline thread.
+        internal IEnumerable<PSResourceInfo> FindDependencyPackages(
+            ServerApiCall currentServer,
+            ResponseUtil currentResponseUtil,
+            PSResourceInfo currentPkg,
+            PSRepositoryInfo repository,
+            ConcurrentQueue<ErrorRecord> errorMsgs,
+            ConcurrentQueue<string> warningMsgs,
+            ConcurrentQueue<string> debugMsgs,
+            ConcurrentQueue<string> verboseMsgs)
+        {
+            // 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>();
+            debugMsgs.Enqueue($"In FindHelper::FindDependencyPackages() - {currentPkg.Name}");
+            FindDependencyPackagesHelper(currentServer, currentResponseUtil, currentPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
 
             return depPkgsFound.Values.ToList();
         }
 
         // Method 2 
-        internal void FindDependencyPackagesHelper(ServerApiCall currentServer, ResponseUtil currentResponseUtil, PSResourceInfo currentPkg, PSRepositoryInfo repository)
+        internal void FindDependencyPackagesHelper(
+            ServerApiCall currentServer,
+            ResponseUtil currentResponseUtil,
+            PSResourceInfo currentPkg,
+            PSRepositoryInfo repository,
+            ConcurrentDictionary<string, PSResourceInfo> depPkgsFound,
+            ConcurrentQueue<ErrorRecord> errorMsgs,
+            ConcurrentQueue<string> warningMsgs,
+            ConcurrentQueue<string> debugMsgs,
+            ConcurrentQueue<string> verboseMsgs)
         {
-            ConcurrentQueue<ErrorRecord> errorMsgs = new ConcurrentQueue<ErrorRecord>();
-            ConcurrentQueue<string> verboseMsgs = new ConcurrentQueue<string>();
-            ConcurrentQueue<string> debugMsgs = new ConcurrentQueue<string>();
-            ConcurrentQueue<string> warningMsgs = new ConcurrentQueue<string>();
             debugMsgs.Enqueue("In FindHelper::FindDependencyPackagesHelper()");
 
             if (currentPkg.Dependencies.Length > 0)
@@ -1185,19 +1234,17 @@ internal void FindDependencyPackagesHelper(ServerApiCall currentServer, Response
                     Parallel.ForEach(currentPkg.Dependencies, new ParallelOptions { MaxDegreeOfParallelism = maxDegreeOfParallelism }, dep =>
                     {
                         debugMsgs.Enqueue($"Finding dependency '{dep.Name}' version range '{dep.VersionRange}'");
-                        FindDependencyPackageVersion(dep, currentServer, currentResponseUtil, currentPkg, repository, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
+                        FindDependencyPackageVersion(dep, currentServer, currentResponseUtil, currentPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
                     });
                 }
                 else
                 {
                     foreach (var dep in currentPkg.Dependencies)
                     {
                         debugMsgs.Enqueue($"Finding dependency '{dep.Name}' version range '{dep.VersionRange}'");
-                        FindDependencyPackageVersion(dep, currentServer, currentResponseUtil, currentPkg, repository, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
+                        FindDependencyPackageVersion(dep, currentServer, currentResponseUtil, currentPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
                     }
                 }
-
-                Utils.WriteOutConcurrentQueue(_cmdletPassedIn, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
             }
         }
 
@@ -1208,6 +1255,7 @@ private void FindDependencyPackageVersion(
             ResponseUtil currentResponseUtil, 
             PSResourceInfo currentPkg, 
             PSRepositoryInfo repository, 
+            ConcurrentDictionary<string, PSResourceInfo> depPkgsFound,
             ConcurrentQueue<ErrorRecord> errorMsgs, 
             ConcurrentQueue<string> warningMsgs,
             ConcurrentQueue<string> debugMsgs,
@@ -1228,7 +1276,7 @@ private void FindDependencyPackageVersion(
                     else
                     {
                         // Find this version from the server
-                        depPkg = FindDependencyWithLowerBound(dep, currentServer, currentResponseUtil, currentPkg, repository, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
+                        depPkg = FindDependencyWithLowerBound(dep, currentServer, currentResponseUtil, currentPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
                     }
             }
             else if (dep.VersionRange.HasLowerBound && dep.VersionRange.MinVersion.Equals(dep.VersionRange.MaxVersion))
@@ -1245,7 +1293,7 @@ private void FindDependencyPackageVersion(
                 }
                 else
                 {
-                    depPkg = FindDependencyWithSpecificVersion(dep, currentServer, currentResponseUtil, currentPkg, repository, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
+                    depPkg = FindDependencyWithSpecificVersion(dep, currentServer, currentResponseUtil, currentPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
                 }
             }
             else
@@ -1261,7 +1309,7 @@ private void FindDependencyPackageVersion(
                 }
                 else
                 {
-                    depPkg = FindDependencyWithUpperBound(dep, currentServer, currentResponseUtil, currentPkg, repository, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
+                    depPkg = FindDependencyWithUpperBound(dep, currentServer, currentResponseUtil, currentPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
                 }
             }
         }
@@ -1273,6 +1321,7 @@ private PSResourceInfo FindDependencyWithSpecificVersion(
             ResponseUtil currentResponseUtil, 
             PSResourceInfo currentPkg, 
             PSRepositoryInfo repository, 
+            ConcurrentDictionary<string, PSResourceInfo> depPkgsFound,
             ConcurrentQueue<ErrorRecord> errorMsgs, 
             ConcurrentQueue<string> warningMsgs,
             ConcurrentQueue<string> debugMsgs,
@@ -1351,7 +1400,7 @@ private PSResourceInfo FindDependencyWithSpecificVersion(
                         // This will eventually return the PSResourceInfo object to the main cmdlet class.
                         debugMsgs.Enqueue($"Adding'{key}' to list of dependency packages found");
                         depPkgsFound.TryAdd(key, depPkg);
-                        FindDependencyPackagesHelper(currentServer, currentResponseUtil, depPkg, repository);
+                        FindDependencyPackagesHelper(currentServer, currentResponseUtil, depPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
                     }
                 }
             }
@@ -1366,6 +1415,7 @@ private PSResourceInfo FindDependencyWithLowerBound(
             ResponseUtil currentResponseUtil, 
             PSResourceInfo currentPkg, 
             PSRepositoryInfo repository, 
+            ConcurrentDictionary<string, PSResourceInfo> depPkgsFound,
             ConcurrentQueue<ErrorRecord> errorMsgs, 
             ConcurrentQueue<string> warningMsgs,
             ConcurrentQueue<string> debugMsgs,
@@ -1419,7 +1469,7 @@ private PSResourceInfo FindDependencyWithLowerBound(
                         // This will eventually return the PSResourceInfo object to the main cmdlet class.
                         debugMsgs.Enqueue($"Adding'{key}' to list of dependency packages found");
                         depPkgsFound.TryAdd(key, depPkg);
-                        FindDependencyPackagesHelper(currentServer, currentResponseUtil, depPkg, repository);
+                        FindDependencyPackagesHelper(currentServer, currentResponseUtil, depPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
                     }
                 }
             }
@@ -1434,6 +1484,7 @@ private PSResourceInfo FindDependencyWithUpperBound(
             ResponseUtil currentResponseUtil, 
             PSResourceInfo currentPkg, 
             PSRepositoryInfo repository, 
+            ConcurrentDictionary<string, PSResourceInfo> depPkgsFound,
             ConcurrentQueue<ErrorRecord> errorMsgs, 
             ConcurrentQueue<string> warningMsgs,
             ConcurrentQueue<string> debugMsgs,
@@ -1490,7 +1541,7 @@ private PSResourceInfo FindDependencyWithUpperBound(
                         // This will eventually return the PSResourceInfo object to the main cmdlet class.
                         debugMsgs.Enqueue($"Adding'{key}' to list of dependency packages found");
                         depPkgsFound.TryAdd(key, depPkg);
-                        FindDependencyPackagesHelper(currentServer, currentResponseUtil, depPkg, repository);
+                        FindDependencyPackagesHelper(currentServer, currentResponseUtil, depPkg, repository, depPkgsFound, errorMsgs, warningMsgs, debugMsgs, verboseMsgs);
                     }
                 }
             }