12845706260

Committed 18 Jan 2025 12:40PM UTC coverage: 94.954% (-0.01%) from 94.964%

Build # 12845706260

Build Type

push

github

Committed by

web-flow

Commit Message

deps: bump phpcompatibility/phpcompatibility-symfony from 1.2.1 to 1.2.2 in /dev-tools (#8378)

Run Details

27889 of 29371 relevant lines covered (94.95%)

43.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.73

/src/Runner/Runner.php

<?php

declare(strict_types=1);

/*
 * This file is part of PHP CS Fixer.
 *
 * (c) Fabien Potencier <fabien@symfony.com>
 *     Dariusz Rumiński <dariusz.ruminski@gmail.com>
 *
 * This source file is subject to the MIT license that is bundled
 * with this source code in the file LICENSE.
 */

namespace PhpCsFixer\Runner;

use Clue\React\NDJson\Decoder;
use Clue\React\NDJson\Encoder;
use PhpCsFixer\AbstractFixer;
use PhpCsFixer\Cache\CacheManagerInterface;
use PhpCsFixer\Cache\Directory;
use PhpCsFixer\Cache\DirectoryInterface;
use PhpCsFixer\Console\Command\WorkerCommand;
use PhpCsFixer\Differ\DifferInterface;
use PhpCsFixer\Error\Error;
use PhpCsFixer\Error\ErrorsManager;
use PhpCsFixer\Error\SourceExceptionFactory;
use PhpCsFixer\FileReader;
use PhpCsFixer\Fixer\FixerInterface;
use PhpCsFixer\Linter\LinterInterface;
use PhpCsFixer\Linter\LintingException;
use PhpCsFixer\Linter\LintingResultInterface;
use PhpCsFixer\Preg;
use PhpCsFixer\Runner\Event\AnalysisStarted;
use PhpCsFixer\Runner\Event\FileProcessed;
use PhpCsFixer\Runner\Parallel\ParallelAction;
use PhpCsFixer\Runner\Parallel\ParallelConfig;
use PhpCsFixer\Runner\Parallel\ParallelConfigFactory;
use PhpCsFixer\Runner\Parallel\ParallelisationException;
use PhpCsFixer\Runner\Parallel\ProcessFactory;
use PhpCsFixer\Runner\Parallel\ProcessIdentifier;
use PhpCsFixer\Runner\Parallel\ProcessPool;
use PhpCsFixer\Runner\Parallel\WorkerException;
use PhpCsFixer\Tokenizer\Tokens;
use React\EventLoop\StreamSelectLoop;
use React\Socket\ConnectionInterface;
use React\Socket\TcpServer;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
use Symfony\Component\Filesystem\Exception\IOException;
use Symfony\Contracts\EventDispatcher\Event;

/**
 * @author Dariusz Rumiński <dariusz.ruminski@gmail.com>
 * @author Greg Korba <greg@codito.dev>
 *
 * @phpstan-type _RunResult array<string, array{appliedFixers: list<string>, diff: string}>
 */
final class Runner
{
    /**
     * Buffer size used in the NDJSON decoder for communication between main process and workers.
     *
     * @see https://github.com/PHP-CS-Fixer/PHP-CS-Fixer/pull/8068
     */
    private const PARALLEL_BUFFER_SIZE = 16 * (1_024 * 1_024 /* 1MB */);

    private DifferInterface $differ;

    private DirectoryInterface $directory;

    private ?EventDispatcherInterface $eventDispatcher;

    private ErrorsManager $errorsManager;

    private CacheManagerInterface $cacheManager;

    private bool $isDryRun;

    private LinterInterface $linter;

    /**
     * @var null|\Traversable<array-key, \SplFileInfo>
     */
    private ?\Traversable $fileIterator = null;

    private int $fileCount;

    /**
     * @var list<FixerInterface>
     */
    private array $fixers;

    private bool $stopOnViolation;

    private ParallelConfig $parallelConfig;

    private ?InputInterface $input;

    private ?string $configFile;

    /**
     * @param null|\Traversable<array-key, \SplFileInfo> $fileIterator
     * @param list<FixerInterface>                       $fixers
     */
    public function __construct(
        ?\Traversable $fileIterator,
        array $fixers,
        DifferInterface $differ,
        ?EventDispatcherInterface $eventDispatcher,
        ErrorsManager $errorsManager,
        LinterInterface $linter,
        bool $isDryRun,
        CacheManagerInterface $cacheManager,
        ?DirectoryInterface $directory = null,
        bool $stopOnViolation = false,
        // @TODO Make these arguments required in 4.0
        ?ParallelConfig $parallelConfig = null,
        ?InputInterface $input = null,
        ?string $configFile = null
    ) {
        // Required only for main process (calculating workers count)
        $this->fileCount = null !== $fileIterator ? \count(iterator_to_array($fileIterator)) : 0;

        $this->fileIterator = $fileIterator;
        $this->fixers = $fixers;
        $this->differ = $differ;
        $this->eventDispatcher = $eventDispatcher;
        $this->errorsManager = $errorsManager;
        $this->linter = $linter;
        $this->isDryRun = $isDryRun;
        $this->cacheManager = $cacheManager;
        $this->directory = $directory ?? new Directory('');
        $this->stopOnViolation = $stopOnViolation;
        $this->parallelConfig = $parallelConfig ?? ParallelConfigFactory::sequential();
        $this->input = $input;
        $this->configFile = $configFile;
    }

    /**
     * @TODO consider to drop this method and make iterator parameter obligatory in constructor,
     * more in https://github.com/PHP-CS-Fixer/PHP-CS-Fixer/pull/7777/files#r1590447581
     *
     * @param \Traversable<array-key, \SplFileInfo> $fileIterator
     */
    public function setFileIterator(iterable $fileIterator): void
    {
        $this->fileIterator = $fileIterator;

        // Required only for main process (calculating workers count)
        $this->fileCount = \count(iterator_to_array($fileIterator));
    }

    /**
     * @return _RunResult
     */
    public function fix(): array
    {
        if (0 === $this->fileCount) {
            return [];
        }

        // @TODO 4.0: Remove condition and its body, as no longer needed when param will be required in the constructor.
        // This is a fallback only in case someone calls `new Runner()` in a custom repo and does not provide v4-ready params in v3-codebase.
        if (null === $this->input) {
            return $this->fixSequential();
        }

        if (
            1 === $this->parallelConfig->getMaxProcesses()
            || $this->fileCount <= $this->parallelConfig->getFilesPerProcess()
        ) {
            return $this->fixSequential();
        }

        return $this->fixParallel();
    }

    /**
     * Heavily inspired by {@see https://github.com/phpstan/phpstan-src/blob/9ce425bca5337039fb52c0acf96a20a2b8ace490/src/Parallel/ParallelAnalyser.php}.
     *
     * @return _RunResult
     */
    private function fixParallel(): array
    {
        $this->dispatchEvent(AnalysisStarted::NAME, new AnalysisStarted(AnalysisStarted::MODE_PARALLEL, $this->isDryRun));

        $changed = [];
        $streamSelectLoop = new StreamSelectLoop();
        $server = new TcpServer('127.0.0.1:0', $streamSelectLoop);
        $serverPort = parse_url($server->getAddress() ?? '', PHP_URL_PORT);

        if (!is_numeric($serverPort)) {
            throw new ParallelisationException(\sprintf(
                'Unable to parse server port from "%s"',
                $server->getAddress() ?? ''
            ));
        }

        $processPool = new ProcessPool($server);
        $maxFilesPerProcess = $this->parallelConfig->getFilesPerProcess();
        $fileIterator = $this->getFilteringFileIterator();
        $fileIterator->rewind();

        $getFileChunk = static function () use ($fileIterator, $maxFilesPerProcess): array {
            $files = [];

            while (\count($files) < $maxFilesPerProcess) {
                $current = $fileIterator->current();

                if (null === $current) {
                    break;
                }

                $files[] = $current->getRealPath();

                $fileIterator->next();
            }

            return $files;
        };

        // [REACT] Handle worker's handshake (init connection)
        $server->on('connection', static function (ConnectionInterface $connection) use ($processPool, $getFileChunk): void {
            $jsonInvalidUtf8Ignore = \defined('JSON_INVALID_UTF8_IGNORE') ? JSON_INVALID_UTF8_IGNORE : 0;
            $decoder = new Decoder(
                $connection,
                true,
                512,
                $jsonInvalidUtf8Ignore,
                self::PARALLEL_BUFFER_SIZE
            );
            $encoder = new Encoder($connection, $jsonInvalidUtf8Ignore);

            // [REACT] Bind connection when worker's process requests "hello" action (enables 2-way communication)
            $decoder->on('data', static function (array $data) use ($processPool, $getFileChunk, $decoder, $encoder): void {
                if (ParallelAction::WORKER_HELLO !== $data['action']) {
                    return;
                }

                $identifier = ProcessIdentifier::fromRaw($data['identifier']);
                $process = $processPool->getProcess($identifier);
                $process->bindConnection($decoder, $encoder);
                $fileChunk = $getFileChunk();

                if (0 === \count($fileChunk)) {
                    $process->request(['action' => ParallelAction::RUNNER_THANK_YOU]);
                    $processPool->endProcessIfKnown($identifier);

                    return;
                }

                $process->request(['action' => ParallelAction::RUNNER_REQUEST_ANALYSIS, 'files' => $fileChunk]);
            });
        });

        $processesToSpawn = min(
            $this->parallelConfig->getMaxProcesses(),
            max(
                1,
                (int) ceil($this->fileCount / $this->parallelConfig->getFilesPerProcess()),
            )
        );
        $processFactory = new ProcessFactory($this->input);

        for ($i = 0; $i < $processesToSpawn; ++$i) {
            $identifier = ProcessIdentifier::create();
            $process = $processFactory->create(
                $streamSelectLoop,
                new RunnerConfig(
                    $this->isDryRun,
                    $this->stopOnViolation,
                    $this->parallelConfig,
                    $this->configFile
                ),
                $identifier,
                $serverPort,
            );
            $processPool->addProcess($identifier, $process);
            $process->start(
                // [REACT] Handle workers' responses (multiple actions possible)
                function (array $workerResponse) use ($processPool, $process, $identifier, $getFileChunk, &$changed): void {
                    // File analysis result (we want close-to-realtime progress with frequent cache savings)
                    if (ParallelAction::WORKER_RESULT === $workerResponse['action']) {
                        $fileAbsolutePath = $workerResponse['file'];
                        $fileRelativePath = $this->directory->getRelativePathTo($fileAbsolutePath);

                        // Dispatch an event for each file processed and dispatch its status (required for progress output)
                        $this->dispatchEvent(FileProcessed::NAME, new FileProcessed($workerResponse['status']));

                        if (isset($workerResponse['fileHash'])) {
                            $this->cacheManager->setFileHash($fileRelativePath, $workerResponse['fileHash']);
                        }

                        foreach ($workerResponse['errors'] ?? [] as $error) {
                            $this->errorsManager->report(new Error(
                                $error['type'],
                                $error['filePath'],
                                null !== $error['source']
                                    ? SourceExceptionFactory::fromArray($error['source'])
                                    : null,
                                $error['appliedFixers'],
                                $error['diff']
                            ));
                        }

                        // Pass-back information about applied changes (only if there are any)
                        if (isset($workerResponse['fixInfo'])) {
                            $changed[$fileRelativePath] = $workerResponse['fixInfo'];

                            if ($this->stopOnViolation) {
                                $processPool->endAll();

                                return;
                            }
                        }

                        return;
                    }

                    if (ParallelAction::WORKER_GET_FILE_CHUNK === $workerResponse['action']) {
                        // Request another chunk of files, if still available
                        $fileChunk = $getFileChunk();

                        if (0 === \count($fileChunk)) {
                            $process->request(['action' => ParallelAction::RUNNER_THANK_YOU]);
                            $processPool->endProcessIfKnown($identifier);

                            return;
                        }

                        $process->request(['action' => ParallelAction::RUNNER_REQUEST_ANALYSIS, 'files' => $fileChunk]);

                        return;
                    }

                    if (ParallelAction::WORKER_ERROR_REPORT === $workerResponse['action']) {
                        throw WorkerException::fromRaw($workerResponse); // @phpstan-ignore-line
                    }

                    throw new ParallelisationException('Unsupported action: '.($workerResponse['action'] ?? 'n/a'));
                },

                // [REACT] Handle errors encountered during worker's execution
                static function (\Throwable $error) use ($processPool): void {
                    $processPool->endAll();

                    throw new ParallelisationException($error->getMessage(), $error->getCode(), $error);
                },

                // [REACT] Handle worker's shutdown
                static function ($exitCode, string $output) use ($processPool, $identifier): void {
                    $processPool->endProcessIfKnown($identifier);

                    if (0 === $exitCode || null === $exitCode) {
                        return;
                    }

                    $errorsReported = Preg::matchAll(
                        \sprintf('/^(?:%s)([^\n]+)+/m', WorkerCommand::ERROR_PREFIX),
                        $output,
                        $matches
                    );

                    if ($errorsReported > 0) {
                        throw WorkerException::fromRaw(json_decode($matches[1][0], true));
                    }
                }
            );
        }

        $streamSelectLoop->run();

        return $changed;
    }

    /**
     * @return _RunResult
     */
    private function fixSequential(): array
    {
        $this->dispatchEvent(AnalysisStarted::NAME, new AnalysisStarted(AnalysisStarted::MODE_SEQUENTIAL, $this->isDryRun));

        $changed = [];
        $collection = $this->getLintingFileIterator();

        foreach ($collection as $file) {
            $fixInfo = $this->fixFile($file, $collection->currentLintingResult());

            // we do not need Tokens to still caching just fixed file - so clear the cache
            Tokens::clearCache();

            if (null !== $fixInfo) {
                $name = $this->directory->getRelativePathTo($file->__toString());
                $changed[$name] = $fixInfo;

                if ($this->stopOnViolation) {
                    break;
                }
            }
        }

        return $changed;
    }

    /**
     * @return null|array{appliedFixers: list<string>, diff: string}
     */
    private function fixFile(\SplFileInfo $file, LintingResultInterface $lintingResult): ?array
    {
        $name = $file->getPathname();

        try {
            $lintingResult->check();
        } catch (LintingException $e) {
            $this->dispatchEvent(
                FileProcessed::NAME,
                new FileProcessed(FileProcessed::STATUS_INVALID)
            );

            $this->errorsManager->report(new Error(Error::TYPE_INVALID, $name, $e));

            return null;
        }

        $old = FileReader::createSingleton()->read($file->getRealPath());

        $tokens = Tokens::fromCode($old);
        $oldHash = $tokens->getCodeHash();

        $new = $old;
        $newHash = $oldHash;

        $appliedFixers = [];

        try {
            foreach ($this->fixers as $fixer) {
                // for custom fixers we don't know is it safe to run `->fix()` without checking `->supports()` and `->isCandidate()`,
                // thus we need to check it and conditionally skip fixing
                if (
                    !$fixer instanceof AbstractFixer
                    && (!$fixer->supports($file) || !$fixer->isCandidate($tokens))
                ) {
                    continue;
                }

                $fixer->fix($file, $tokens);

                if ($tokens->isChanged()) {
                    $tokens->clearEmptyTokens();
                    $tokens->clearChanged();
                    $appliedFixers[] = $fixer->getName();
                }
            }
        } catch (\ParseError $e) {
            $this->dispatchEvent(FileProcessed::NAME, new FileProcessed(FileProcessed::STATUS_LINT));

            $this->errorsManager->report(new Error(Error::TYPE_LINT, $name, $e));

            return null;
        } catch (\Throwable $e) {
            $this->processException($name, $e);

            return null;
        }

        $fixInfo = null;

        if ([] !== $appliedFixers) {
            $new = $tokens->generateCode();
            $newHash = $tokens->getCodeHash();
        }

        // We need to check if content was changed and then applied changes.
        // But we can't simply check $appliedFixers, because one fixer may revert
        // work of other and both of them will mark collection as changed.
        // Therefore we need to check if code hashes changed.
        if ($oldHash !== $newHash) {
            $fixInfo = [
                'appliedFixers' => $appliedFixers,
                'diff' => $this->differ->diff($old, $new, $file),
            ];

            try {
                $this->linter->lintSource($new)->check();
            } catch (LintingException $e) {
                $this->dispatchEvent(FileProcessed::NAME, new FileProcessed(FileProcessed::STATUS_LINT));

                $this->errorsManager->report(new Error(Error::TYPE_LINT, $name, $e, $fixInfo['appliedFixers'], $fixInfo['diff']));

                return null;
            }

            if (!$this->isDryRun) {
                $fileName = $file->getRealPath();

                if (!file_exists($fileName)) {
                    throw new IOException(
                        \sprintf('Failed to write file "%s" (no longer) exists.', $file->getPathname()),
                        0,
                        null,
                        $file->getPathname()
                    );
                }

                if (is_dir($fileName)) {
                    throw new IOException(
                        \sprintf('Cannot write file "%s" as the location exists as directory.', $fileName),
                        0,
                        null,
                        $fileName
                    );
                }

                if (!is_writable($fileName)) {
                    throw new IOException(
                        \sprintf('Cannot write to file "%s" as it is not writable.', $fileName),
                        0,
                        null,
                        $fileName
                    );
                }

                if (false === @file_put_contents($fileName, $new)) {
                    $error = error_get_last();

                    throw new IOException(
                        \sprintf('Failed to write file "%s", "%s".', $fileName, null !== $error ? $error['message'] : 'no reason available'),
                        0,
                        null,
                        $fileName
                    );
                }
            }
        }

        $this->cacheManager->setFileHash($name, $newHash);

        $this->dispatchEvent(
            FileProcessed::NAME,
            new FileProcessed(null !== $fixInfo ? FileProcessed::STATUS_FIXED : FileProcessed::STATUS_NO_CHANGES, $name, $newHash)
        );

        return $fixInfo;
    }

    /**
     * Process an exception that occurred.
     */
    private function processException(string $name, \Throwable $e): void
    {
        $this->dispatchEvent(FileProcessed::NAME, new FileProcessed(FileProcessed::STATUS_EXCEPTION));

        $this->errorsManager->report(new Error(Error::TYPE_EXCEPTION, $name, $e));
    }

    private function dispatchEvent(string $name, Event $event): void
    {
        if (null === $this->eventDispatcher) {
            return;
        }

        $this->eventDispatcher->dispatch($event, $name);
    }

    private function getLintingFileIterator(): LintingResultAwareFileIteratorInterface
    {
        $fileFilterIterator = $this->getFilteringFileIterator();

        return $this->linter->isAsync()
            ? new FileCachingLintingFileIterator($fileFilterIterator, $this->linter)
            : new LintingFileIterator($fileFilterIterator, $this->linter);
    }

    private function getFilteringFileIterator(): FileFilterIterator
    {
        if (null === $this->fileIterator) {
            throw new \RuntimeException('File iterator is not configured. Pass paths during Runner initialisation or set them after with `setFileIterator()`.');
        }

        return new FileFilterIterator(
            $this->fileIterator instanceof \IteratorAggregate
                ? $this->fileIterator->getIterator()
                : $this->fileIterator,
            $this->eventDispatcher,
            $this->cacheManager
        );
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	/*
6	* This file is part of PHP CS Fixer.
7	*
8	* (c) Fabien Potencier <fabien@symfony.com>
9	* Dariusz Rumiński <dariusz.ruminski@gmail.com>
10	*
11	* This source file is subject to the MIT license that is bundled
12	* with this source code in the file LICENSE.
13	*/
14
15	namespace PhpCsFixer\Runner;
16
17	use Clue\React\NDJson\Decoder;
18	use Clue\React\NDJson\Encoder;
19	use PhpCsFixer\AbstractFixer;
20	use PhpCsFixer\Cache\CacheManagerInterface;
21	use PhpCsFixer\Cache\Directory;
22	use PhpCsFixer\Cache\DirectoryInterface;
23	use PhpCsFixer\Console\Command\WorkerCommand;
24	use PhpCsFixer\Differ\DifferInterface;
25	use PhpCsFixer\Error\Error;
26	use PhpCsFixer\Error\ErrorsManager;
27	use PhpCsFixer\Error\SourceExceptionFactory;
28	use PhpCsFixer\FileReader;
29	use PhpCsFixer\Fixer\FixerInterface;
30	use PhpCsFixer\Linter\LinterInterface;
31	use PhpCsFixer\Linter\LintingException;
32	use PhpCsFixer\Linter\LintingResultInterface;
33	use PhpCsFixer\Preg;
34	use PhpCsFixer\Runner\Event\AnalysisStarted;
35	use PhpCsFixer\Runner\Event\FileProcessed;
36	use PhpCsFixer\Runner\Parallel\ParallelAction;
37	use PhpCsFixer\Runner\Parallel\ParallelConfig;
38	use PhpCsFixer\Runner\Parallel\ParallelConfigFactory;
39	use PhpCsFixer\Runner\Parallel\ParallelisationException;
40	use PhpCsFixer\Runner\Parallel\ProcessFactory;
41	use PhpCsFixer\Runner\Parallel\ProcessIdentifier;
42	use PhpCsFixer\Runner\Parallel\ProcessPool;
43	use PhpCsFixer\Runner\Parallel\WorkerException;
44	use PhpCsFixer\Tokenizer\Tokens;
45	use React\EventLoop\StreamSelectLoop;
46	use React\Socket\ConnectionInterface;
47	use React\Socket\TcpServer;
48	use Symfony\Component\Console\Input\InputInterface;
49	use Symfony\Component\EventDispatcher\EventDispatcherInterface;
50	use Symfony\Component\Filesystem\Exception\IOException;
51	use Symfony\Contracts\EventDispatcher\Event;
52
53	/**
54	* @author Dariusz Rumiński <dariusz.ruminski@gmail.com>
55	* @author Greg Korba <greg@codito.dev>
56	*
57	* @phpstan-type _RunResult array<string, array{appliedFixers: list<string>, diff: string}>
58	*/
59	final class Runner
60	{
61	/**
62	* Buffer size used in the NDJSON decoder for communication between main process and workers.
63	*
64	* @see https://github.com/PHP-CS-Fixer/PHP-CS-Fixer/pull/8068
65	*/
66	private const PARALLEL_BUFFER_SIZE = 16 * (1_024 * 1_024 /* 1MB */);
67
68	private DifferInterface $differ;
69
70	private DirectoryInterface $directory;
71
72	private ?EventDispatcherInterface $eventDispatcher;
73
74	private ErrorsManager $errorsManager;
75
76	private CacheManagerInterface $cacheManager;
77
78	private bool $isDryRun;
79
80	private LinterInterface $linter;
81
82	/**
83	* @var null\|\Traversable<array-key, \SplFileInfo>
84	*/
85	private ?\Traversable $fileIterator = null;
86
87	private int $fileCount;
88
89	/**
90	* @var list<FixerInterface>
91	*/
92	private array $fixers;
93
94	private bool $stopOnViolation;
95
96	private ParallelConfig $parallelConfig;
97
98	private ?InputInterface $input;
99
100	private ?string $configFile;
101
102	/**
103	* @param null\|\Traversable<array-key, \SplFileInfo> $fileIterator
104	* @param list<FixerInterface> $fixers
105	*/
106	public function __construct(
107	?\Traversable $fileIterator,
108	array $fixers,
109	DifferInterface $differ,
110	?EventDispatcherInterface $eventDispatcher,
111	ErrorsManager $errorsManager,
112	LinterInterface $linter,
113	bool $isDryRun,
114	CacheManagerInterface $cacheManager,
115	?DirectoryInterface $directory = null,
116	bool $stopOnViolation = false,
117	// @TODO Make these arguments required in 4.0
118	?ParallelConfig $parallelConfig = null,
119	?InputInterface $input = null,
120	?string $configFile = null
121	) {
122	// Required only for main process (calculating workers count)
123	$this->fileCount = null !== $fileIterator ? \count(iterator_to_array($fileIterator)) : 0;	7✔
124
125	$this->fileIterator = $fileIterator;	7✔
126	$this->fixers = $fixers;	7✔
127	$this->differ = $differ;	7✔
128	$this->eventDispatcher = $eventDispatcher;	7✔
129	$this->errorsManager = $errorsManager;	7✔
130	$this->linter = $linter;	7✔
131	$this->isDryRun = $isDryRun;	7✔
132	$this->cacheManager = $cacheManager;	7✔
133	$this->directory = $directory ?? new Directory('');	7✔
134	$this->stopOnViolation = $stopOnViolation;	7✔
135	$this->parallelConfig = $parallelConfig ?? ParallelConfigFactory::sequential();	7✔
136	$this->input = $input;	7✔
137	$this->configFile = $configFile;	7✔
138	}
139
140	/**
141	* @TODO consider to drop this method and make iterator parameter obligatory in constructor,
142	* more in https://github.com/PHP-CS-Fixer/PHP-CS-Fixer/pull/7777/files#r1590447581
143	*
144	* @param \Traversable<array-key, \SplFileInfo> $fileIterator
145	*/
146	public function setFileIterator(iterable $fileIterator): void
147	{
148	$this->fileIterator = $fileIterator;	×
149
150	// Required only for main process (calculating workers count)
151	$this->fileCount = \count(iterator_to_array($fileIterator));	×
152	}
153
154	/**
155	* @return _RunResult
156	*/
157	public function fix(): array
158	{
159	if (0 === $this->fileCount) {	7✔
160	return [];	×
161	}
162
163	// @TODO 4.0: Remove condition and its body, as no longer needed when param will be required in the constructor.
164	// This is a fallback only in case someone calls `new Runner()` in a custom repo and does not provide v4-ready params in v3-codebase.
165	if (null === $this->input) {	7✔
166	return $this->fixSequential();	3✔
167	}
168
169	if (
170	1 === $this->parallelConfig->getMaxProcesses()	4✔
171	\|\| $this->fileCount <= $this->parallelConfig->getFilesPerProcess()	4✔
172	) {
173	return $this->fixSequential();	3✔
174	}
175
176	return $this->fixParallel();	1✔
177	}
178
179	/**
180	* Heavily inspired by {@see https://github.com/phpstan/phpstan-src/blob/9ce425bca5337039fb52c0acf96a20a2b8ace490/src/Parallel/ParallelAnalyser.php}.
181	*
182	* @return _RunResult
183	*/
184	private function fixParallel(): array
185	{
186	$this->dispatchEvent(AnalysisStarted::NAME, new AnalysisStarted(AnalysisStarted::MODE_PARALLEL, $this->isDryRun));	2✔
187
188	$changed = [];	2✔
189	$streamSelectLoop = new StreamSelectLoop();	2✔
190	$server = new TcpServer('127.0.0.1:0', $streamSelectLoop);	2✔
191	$serverPort = parse_url($server->getAddress() ?? '', PHP_URL_PORT);	2✔
192
193	if (!is_numeric($serverPort)) {	2✔
194	throw new ParallelisationException(\sprintf(	×
195	'Unable to parse server port from "%s"',	×
196	$server->getAddress() ?? ''	×
197	));	×
198	}
199
200	$processPool = new ProcessPool($server);	2✔
201	$maxFilesPerProcess = $this->parallelConfig->getFilesPerProcess();	2✔
202	$fileIterator = $this->getFilteringFileIterator();	2✔
203	$fileIterator->rewind();	2✔
204
205	$getFileChunk = static function () use ($fileIterator, $maxFilesPerProcess): array {	2✔
206	$files = [];	2✔
207
208	while (\count($files) < $maxFilesPerProcess) {	2✔
209	$current = $fileIterator->current();	2✔
210
211	if (null === $current) {	2✔
212	break;	2✔
213	}
214
215	$files[] = $current->getRealPath();	2✔
216
217	$fileIterator->next();	2✔
218	}
219
220	return $files;	2✔
221	};	2✔
222
223	// [REACT] Handle worker's handshake (init connection)
224	$server->on('connection', static function (ConnectionInterface $connection) use ($processPool, $getFileChunk): void {	2✔
225	$jsonInvalidUtf8Ignore = \defined('JSON_INVALID_UTF8_IGNORE') ? JSON_INVALID_UTF8_IGNORE : 0;	2✔
226	$decoder = new Decoder(	2✔
227	$connection,	2✔
228	true,	2✔
229	512,	2✔
230	$jsonInvalidUtf8Ignore,	2✔
231	self::PARALLEL_BUFFER_SIZE	2✔
232	);	2✔
233	$encoder = new Encoder($connection, $jsonInvalidUtf8Ignore);	2✔
234
235	// [REACT] Bind connection when worker's process requests "hello" action (enables 2-way communication)
236	$decoder->on('data', static function (array $data) use ($processPool, $getFileChunk, $decoder, $encoder): void {	2✔
237	if (ParallelAction::WORKER_HELLO !== $data['action']) {	2✔
238	return;	2✔
239	}
240
241	$identifier = ProcessIdentifier::fromRaw($data['identifier']);	2✔
242	$process = $processPool->getProcess($identifier);	2✔
243	$process->bindConnection($decoder, $encoder);	2✔
244	$fileChunk = $getFileChunk();	2✔
245
246	if (0 === \count($fileChunk)) {	2✔
247	$process->request(['action' => ParallelAction::RUNNER_THANK_YOU]);	×
248	$processPool->endProcessIfKnown($identifier);	×
249
250	return;	×
251	}
252
253	$process->request(['action' => ParallelAction::RUNNER_REQUEST_ANALYSIS, 'files' => $fileChunk]);	2✔
254	});	2✔
255	});	2✔
256
257	$processesToSpawn = min(	2✔
258	$this->parallelConfig->getMaxProcesses(),	2✔
259	max(	2✔
260	1,	2✔
261	(int) ceil($this->fileCount / $this->parallelConfig->getFilesPerProcess()),	2✔
262	)	2✔
263	);	2✔
264	$processFactory = new ProcessFactory($this->input);	2✔
265
266	for ($i = 0; $i < $processesToSpawn; ++$i) {	2✔
267	$identifier = ProcessIdentifier::create();	2✔
268	$process = $processFactory->create(	2✔
269	$streamSelectLoop,	2✔
270	new RunnerConfig(	2✔
271	$this->isDryRun,	2✔
272	$this->stopOnViolation,	2✔
273	$this->parallelConfig,	2✔
274	$this->configFile	2✔
275	),	2✔
276	$identifier,	2✔
277	$serverPort,	2✔
278	);	2✔
279	$processPool->addProcess($identifier, $process);	2✔
280	$process->start(	2✔
281	// [REACT] Handle workers' responses (multiple actions possible)
282	function (array $workerResponse) use ($processPool, $process, $identifier, $getFileChunk, &$changed): void {	2✔
283	// File analysis result (we want close-to-realtime progress with frequent cache savings)
284	if (ParallelAction::WORKER_RESULT === $workerResponse['action']) {	2✔
285	$fileAbsolutePath = $workerResponse['file'];	2✔
286	$fileRelativePath = $this->directory->getRelativePathTo($fileAbsolutePath);	2✔
287
288	// Dispatch an event for each file processed and dispatch its status (required for progress output)
289	$this->dispatchEvent(FileProcessed::NAME, new FileProcessed($workerResponse['status']));	2✔
290
291	if (isset($workerResponse['fileHash'])) {	2✔
292	$this->cacheManager->setFileHash($fileRelativePath, $workerResponse['fileHash']);	2✔
293	}
294
295	foreach ($workerResponse['errors'] ?? [] as $error) {	2✔
296	$this->errorsManager->report(new Error(	×
297	$error['type'],	×
298	$error['filePath'],	×
299	null !== $error['source']	×
300	? SourceExceptionFactory::fromArray($error['source'])	×
301	: null,	×
302	$error['appliedFixers'],	×
303	$error['diff']	×
304	));	×
305	}
306
307	// Pass-back information about applied changes (only if there are any)
308	if (isset($workerResponse['fixInfo'])) {	2✔
309	$changed[$fileRelativePath] = $workerResponse['fixInfo'];	2✔
310
311	if ($this->stopOnViolation) {	2✔
312	$processPool->endAll();	×
313
314	return;	×
315	}
316	}
317
318	return;	2✔
319	}
320
321	if (ParallelAction::WORKER_GET_FILE_CHUNK === $workerResponse['action']) {	2✔
322	// Request another chunk of files, if still available
323	$fileChunk = $getFileChunk();	2✔
324
325	if (0 === \count($fileChunk)) {	2✔
326	$process->request(['action' => ParallelAction::RUNNER_THANK_YOU]);	2✔
327	$processPool->endProcessIfKnown($identifier);	2✔
328
329	return;	2✔
330	}
331
332	$process->request(['action' => ParallelAction::RUNNER_REQUEST_ANALYSIS, 'files' => $fileChunk]);	1✔
333
334	return;	1✔
335	}
336
337	if (ParallelAction::WORKER_ERROR_REPORT === $workerResponse['action']) {	×
338	throw WorkerException::fromRaw($workerResponse); // @phpstan-ignore-line	×
339	}
340
341	throw new ParallelisationException('Unsupported action: '.($workerResponse['action'] ?? 'n/a'));	×
342	},	2✔
343
344	// [REACT] Handle errors encountered during worker's execution
345	static function (\Throwable $error) use ($processPool): void {	2✔
346	$processPool->endAll();	×
347
348	throw new ParallelisationException($error->getMessage(), $error->getCode(), $error);	×
349	},	2✔
350
351	// [REACT] Handle worker's shutdown
352	static function ($exitCode, string $output) use ($processPool, $identifier): void {	2✔
353	$processPool->endProcessIfKnown($identifier);	2✔
354
355	if (0 === $exitCode \|\| null === $exitCode) {	2✔
356	return;	2✔
357	}
358
359	$errorsReported = Preg::matchAll(	×
360	\sprintf('/^(?:%s)([^\n]+)+/m', WorkerCommand::ERROR_PREFIX),	×
361	$output,	×
362	$matches	×
363	);	×
364
365	if ($errorsReported > 0) {	×
366	throw WorkerException::fromRaw(json_decode($matches[1][0], true));	×
367	}
368	}	2✔
369	);	2✔
370	}
371
372	$streamSelectLoop->run();	2✔
373
374	return $changed;	2✔
375	}
376
377	/**
378	* @return _RunResult
379	*/
380	private function fixSequential(): array
381	{
382	$this->dispatchEvent(AnalysisStarted::NAME, new AnalysisStarted(AnalysisStarted::MODE_SEQUENTIAL, $this->isDryRun));	7✔
383
384	$changed = [];	7✔
385	$collection = $this->getLintingFileIterator();	7✔
386
387	foreach ($collection as $file) {	7✔
388	$fixInfo = $this->fixFile($file, $collection->currentLintingResult());	7✔
389
390	// we do not need Tokens to still caching just fixed file - so clear the cache
391	Tokens::clearCache();	7✔
392
393	if (null !== $fixInfo) {	7✔
394	$name = $this->directory->getRelativePathTo($file->__toString());	5✔
395	$changed[$name] = $fixInfo;	5✔
396
397	if ($this->stopOnViolation) {	5✔
398	break;	2✔
399	}
400	}
401	}
402
403	return $changed;	7✔
404	}
405
406	/**
407	* @return null\|array{appliedFixers: list<string>, diff: string}
408	*/
409	private function fixFile(\SplFileInfo $file, LintingResultInterface $lintingResult): ?array
410	{
411	$name = $file->getPathname();	6✔
412
413	try {
414	$lintingResult->check();	6✔
415	} catch (LintingException $e) {	2✔
416	$this->dispatchEvent(	2✔
417	FileProcessed::NAME,	2✔
418	new FileProcessed(FileProcessed::STATUS_INVALID)	2✔
419	);	2✔
420
421	$this->errorsManager->report(new Error(Error::TYPE_INVALID, $name, $e));	2✔
422
423	return null;	2✔
424	}
425
426	$old = FileReader::createSingleton()->read($file->getRealPath());	4✔
427
428	$tokens = Tokens::fromCode($old);	4✔
429	$oldHash = $tokens->getCodeHash();	4✔
430
431	$new = $old;	4✔
432	$newHash = $oldHash;	4✔
433
434	$appliedFixers = [];	4✔
435
436	try {
437	foreach ($this->fixers as $fixer) {	4✔
438	// for custom fixers we don't know is it safe to run `->fix()` without checking `->supports()` and `->isCandidate()`,
439	// thus we need to check it and conditionally skip fixing
440	if (
441	!$fixer instanceof AbstractFixer	4✔
442	&& (!$fixer->supports($file) \|\| !$fixer->isCandidate($tokens))	4✔
443	) {
444	continue;	×
445	}
446
447	$fixer->fix($file, $tokens);	4✔
448
449	if ($tokens->isChanged()) {	4✔
450	$tokens->clearEmptyTokens();	4✔
451	$tokens->clearChanged();	4✔
452	$appliedFixers[] = $fixer->getName();	4✔
453	}
454	}
455	} catch (\ParseError $e) {	×
456	$this->dispatchEvent(FileProcessed::NAME, new FileProcessed(FileProcessed::STATUS_LINT));	×
457
458	$this->errorsManager->report(new Error(Error::TYPE_LINT, $name, $e));	×
459
460	return null;	×
461	} catch (\Throwable $e) {	×
462	$this->processException($name, $e);	×
463
464	return null;	×
465	}
466
467	$fixInfo = null;	4✔
468
469	if ([] !== $appliedFixers) {	4✔
470	$new = $tokens->generateCode();	4✔
471	$newHash = $tokens->getCodeHash();	4✔
472	}
473
474	// We need to check if content was changed and then applied changes.
475	// But we can't simply check $appliedFixers, because one fixer may revert
476	// work of other and both of them will mark collection as changed.
477	// Therefore we need to check if code hashes changed.
478	if ($oldHash !== $newHash) {	4✔
479	$fixInfo = [	4✔
480	'appliedFixers' => $appliedFixers,	4✔
481	'diff' => $this->differ->diff($old, $new, $file),	4✔
482	];	4✔
483
484	try {
485	$this->linter->lintSource($new)->check();	4✔
486	} catch (LintingException $e) {	×
487	$this->dispatchEvent(FileProcessed::NAME, new FileProcessed(FileProcessed::STATUS_LINT));	×
488
489	$this->errorsManager->report(new Error(Error::TYPE_LINT, $name, $e, $fixInfo['appliedFixers'], $fixInfo['diff']));	×
490
491	return null;	×
492	}
493
494	if (!$this->isDryRun) {	4✔
495	$fileName = $file->getRealPath();	×
496
497	if (!file_exists($fileName)) {	×
498	throw new IOException(	×
499	\sprintf('Failed to write file "%s" (no longer) exists.', $file->getPathname()),	×
500	0,	×
501	null,	×
502	$file->getPathname()	×
503	);	×
504	}
505
506	if (is_dir($fileName)) {	×
507	throw new IOException(	×
508	\sprintf('Cannot write file "%s" as the location exists as directory.', $fileName),	×
509	0,	×
510	null,	×
511	$fileName	×
512	);	×
513	}
514
515	if (!is_writable($fileName)) {	×
516	throw new IOException(	×
517	\sprintf('Cannot write to file "%s" as it is not writable.', $fileName),	×
518	0,	×
519	null,	×
520	$fileName	×
521	);	×
522	}
523
524	if (false === @file_put_contents($fileName, $new)) {	×
525	$error = error_get_last();	×
526
527	throw new IOException(	×
528	\sprintf('Failed to write file "%s", "%s".', $fileName, null !== $error ? $error['message'] : 'no reason available'),	×
529	0,	×
530	null,	×
531	$fileName	×
532	);	×
533	}
534	}
535	}
536
537	$this->cacheManager->setFileHash($name, $newHash);	4✔
538
539	$this->dispatchEvent(	4✔
540	FileProcessed::NAME,	4✔
541	new FileProcessed(null !== $fixInfo ? FileProcessed::STATUS_FIXED : FileProcessed::STATUS_NO_CHANGES, $name, $newHash)	4✔
542	);	4✔
543
544	return $fixInfo;	4✔
545	}
546
547	/**
548	* Process an exception that occurred.
549	*/
550	private function processException(string $name, \Throwable $e): void
551	{
552	$this->dispatchEvent(FileProcessed::NAME, new FileProcessed(FileProcessed::STATUS_EXCEPTION));	×
553
554	$this->errorsManager->report(new Error(Error::TYPE_EXCEPTION, $name, $e));	×
555	}
556
557	private function dispatchEvent(string $name, Event $event): void
558	{
559	if (null === $this->eventDispatcher) {	7✔
560	return;	4✔
561	}
562
563	$this->eventDispatcher->dispatch($event, $name);	3✔
564	}
565
566	private function getLintingFileIterator(): LintingResultAwareFileIteratorInterface
567	{
568	$fileFilterIterator = $this->getFilteringFileIterator();	6✔
569
570	return $this->linter->isAsync()	6✔
571	? new FileCachingLintingFileIterator($fileFilterIterator, $this->linter)	×
572	: new LintingFileIterator($fileFilterIterator, $this->linter);	6✔
573	}
574
575	private function getFilteringFileIterator(): FileFilterIterator
576	{
577	if (null === $this->fileIterator) {	7✔
578	throw new \RuntimeException('File iterator is not configured. Pass paths during Runner initialisation or set them after with `setFileIterator()`.');	×
579	}
580
581	return new FileFilterIterator(	7✔
582	$this->fileIterator instanceof \IteratorAggregate	7✔
583	? $this->fileIterator->getIterator()	7✔
584	: $this->fileIterator,	7✔
585	$this->eventDispatcher,	7✔
586	$this->cacheManager	7✔
587	);	7✔
588	}
589	}

keradus / PHP-CS-Fixer / 12845706260

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous