Consume data sooner, remove redundant classes

Author: danog

src/Psalm/Internal/Codebase/Analyzer.php

@@ -4,6 +4,7 @@
 
 namespace Psalm\Internal\Codebase;
 
+use Amp\Future;
 use InvalidArgumentException;
 use PhpParser;
 use Psalm\CodeLocation;
@@ -31,6 +32,7 @@
 use SebastianBergmann\Diff\Output\StrictUnifiedDiffOutputBuilder;
 use UnexpectedValueException;
 
+use function Amp\Future\await;
 use function array_filter;
 use function array_intersect_key;
 use function array_merge;
@@ -313,13 +315,15 @@ private function doAnalysis(ProjectAnalyzer $project_analyzer, int $pool_size):
             $this->progress->debug('Forking analysis' . "\n");
 
             // Wait for all tasks to complete and collect the results.
-            $pool->runAll(new InitAnalyzerTask);
+            await($pool->runAll(new InitAnalyzerTask));
             $pool->run($this->files_to_analyze, AnalyzerTask::class, $task_done_closure);
             $forked_pool_data = $pool->runAll(new ShutdownAnalyzerTask);
 
             $this->progress->debug('Collecting forked analysis results' . "\n");
 
-            foreach ($forked_pool_data as $pool_data) {
+            foreach (Future::iterate($forked_pool_data) as $pool_data) {
+                $pool_data = $pool_data->await();
+
                 IssueBuffer::addIssues($pool_data['issues']);
                 IssueBuffer::addFixableIssues($pool_data['fixable_issue_counts']);
 

src/Psalm/Internal/Codebase/Scanner.php

@@ -25,6 +25,7 @@
 use Throwable;
 use UnexpectedValueException;
 
+use function Amp\Future\await;
 use function array_filter;
 use function array_merge;
 use function array_pop;
@@ -308,7 +309,7 @@ private function scanFilePaths(int $pool_size): bool
                 $this->progress,
             );
 
-            $pool->runAll(new InitScannerTask);
+            await($pool->runAll(new InitScannerTask));
             $pool->run($files_to_scan, ScannerTask::class, function (): void {
                 $this->progress->taskDone(0);
             });
@@ -317,6 +318,8 @@ private function scanFilePaths(int $pool_size): bool
             $forked_pool_data = $pool->runAll(new ShutdownScannerTask);
 
             foreach ($forked_pool_data as $pool_data) {
+                $pool_data = $pool_data->await();
+
                 IssueBuffer::addIssues($pool_data['issues']);
 
                 $this->codebase->statements_provider->addChangedMembers(
@@ -351,12 +354,9 @@ private function scanFilePaths(int $pool_size): bool
                 }
             }
         } else {
-            $i = 0;
-
             foreach ($files_to_scan as $file_path => $_) {
+                $this->scanAPath($file_path);
                 $this->progress->taskDone(0);
-                $this->scanAPath($i, $file_path);
-                ++$i;
             }
         }
 
@@ -707,7 +707,7 @@ public function isForked(): void
         $this->is_forked = true;
     }
 
-    public function scanAPath(int $_, string $file_path): void
+    public function scanAPath(string $file_path): void
     {
         $this->scanFile(
             $file_path,

src/Psalm/Internal/Fork/ForkMessage.php

@@ -17,11 +17,6 @@
 final class InitAnalyzerTask implements Task
 {
     public function run(Channel $channel, Cancellation $cancellation): mixed
-    {
-        self::runStatic();
-        return null;
-    }
-    public static function runStatic(): void
     {
         $project_analyzer = ProjectAnalyzer::getInstance();
         $codebase = $project_analyzer->getCodebase();
@@ -41,5 +36,6 @@ public static function runStatic(): void
         $file_reference_provider->setFileReferencesToMissingClassMembers([]);
         $file_reference_provider->setReferencesToMixedMemberNames([]);
         $file_reference_provider->setMethodParamUses([]);
+        return null;
     }
 }

src/Psalm/Internal/Fork/ForkProcessDoneMessage.php

@@ -20,12 +20,6 @@
 final class InitScannerTask implements Task
 {
     final public function run(Channel $channel, Cancellation $cancellation): mixed
-    {
-        self::runStatic();
-        return null;
-    }
-
-    public static function runStatic(): void
     {
         $analyzer = ProjectAnalyzer::getInstance();
         $analyzer->progress->debug('Initialising forked process for scanning' . PHP_EOL);
@@ -40,5 +34,7 @@ public static function runStatic(): void
         $statements_provider->resetDiffs();
 
         $analyzer->progress->debug('Have initialised forked process for scanning' . PHP_EOL);
+
+        return null;
     }
 }

src/Psalm/Internal/Fork/ForkProcessErrorMessage.php

@@ -110,7 +110,7 @@ public function run(
     /**
      * @template T
      * @param Task<T, void, void> $task
-     * @return array<int, T>
+     * @return array<int, Future<T>>
      */
     public function runAll(Task $task): array
     {
@@ -123,8 +123,6 @@ public function runAll(Task $task): array
         for ($x = 0; $x < $this->threads; $x++) {
             $workers []= $this->pool->getWorker();
         }
-        return await(
-            array_map(fn(Worker $w): Future => $w->submit($task)->getFuture(), $workers),
-        );
+        return array_map(fn(Worker $w): Future => $w->submit($task)->getFuture(), $workers);
     }
 }

src/Psalm/Internal/Fork/ForkTaskDoneMessage.php

@@ -20,6 +20,6 @@ public function __construct(private string $file)
     }
     public function run(Channel $channel, Cancellation $cancellation): mixed
     {
-        return ProjectAnalyzer::getInstance()->getCodebase()->scanner->scanAPath(0, $this->file);
+        return ProjectAnalyzer::getInstance()->getCodebase()->scanner->scanAPath($this->file);
     }
 }

src/Psalm/Internal/Fork/InitAnalyzerTask.php

@@ -24,14 +24,6 @@ final class ShutdownAnalyzerTask implements Task
      * @return WorkerData
      */
     public function run(Channel $channel, Cancellation $cancellation): mixed
-    {
-        return self::getPoolData();
-    }
-
-    /**
-     * @return WorkerData
-     */
-    public static function getPoolData(): array
     {
         $project_analyzer        = ProjectAnalyzer::getInstance();
         $codebase                = $project_analyzer->getCodebase();

src/Psalm/Internal/Fork/InitScannerTask.php

@@ -24,13 +24,6 @@ final class ShutdownScannerTask implements Task
      * @return PoolData
      */
     public function run(Channel $channel, Cancellation $cancellation): mixed
-    {
-        return self::getPoolData();
-    }
-    /**
-     * @return PoolData
-     */
-    public static function getPoolData(): array
     {
         $project_analyzer = ProjectAnalyzer::getInstance();
         $project_analyzer->progress->debug('Collecting data from forked scanner process' . PHP_EOL);