Page MenuHomeWickedGov Phorge
Files F6055333

ForkPool.php
No OneTemporary
Actions

Size
12 KB
Referenced Files
None
Subscribers
None

ForkPool.php
View Options

<?php
declare(strict_types=1);
namespace Phan;
use AssertionError;
use Closure;
use InvalidArgumentException;
use Phan\ForkPool\Progress;
use Phan\ForkPool\Reader;
use Phan\ForkPool\Writer;
use Phan\Library\StringUtil;
use function count;
use function intval;
use function unserialize;
use const EXIT_FAILURE;
use const EXIT_SUCCESS;
/**
* Fork off to n-processes and divide up tasks between
* each process.
*
* @internal
*/
class ForkPool
{
/** @var list<int> a list of process ids that have been forked*/
private $child_pid_list = [];
/** @var list<resource> a list of read strings for $this->child_pid_list */
private $read_streams = [];
/** @var list<Reader> a list of Readers for $this->child_pid_list */
private $readers = [];
/** @var list<Progress> a map from workers to their progress */
private $progress = [];
/** @var float the maximum memory usage at any time during the run. Excludes finished workers. */
private $max_total_mem = 0;
/** @var list<IssueInstance> the combination of issues emitted by all workers */
private $issues = [];
/** @var bool did any of the child processes fail (e.g. crash or send data that couldn't be unserialized) */
private $did_have_error = false;
/** @var float the previous update time */
private $previous_update_time = 0.0;
private function updateProgress(int $i, Progress $progress): void
{
// fwrite(STDERR, "Received progress from $i " . json_encode($progress) . "\n");
$this->progress[$i] = $progress;
$time = \microtime(true);
// If not enough time has elapsed, then don't update the progress bar.
// Making the update frequency based on time (instead of the number of files)
// prevents the terminal from rapidly flickering while processing small files.
$interval = Config::getValue('progress_bar_sample_interval');
if ($time - $this->previous_update_time < $interval) {
// Make sure to output 100%, to avoid confusion.
// https://github.com/phan/phan/issues/2694
if ($progress->progress < 1.0) {
return;
}
}
if ($this->previous_update_time) {
$this->previous_update_time += $interval;
} else {
$this->previous_update_time = $time;
}
$this->renderAggregateProgress();
}
private function renderAggregateProgress(): void
{
$total_progress = 0.0;
$total_cur_mem = 0.0;
$total_max_mem = 0.0;
$file_count = 0;
$analyzed_files = 0;
foreach ($this->progress as $progress) {
$total_progress += $progress->progress;
$total_cur_mem += $progress->cur_mem / 1024 / 1024;
if ($progress->cur_mem) {
$total_max_mem += $progress->max_mem / 1024 / 1024;
}
$file_count += $progress->file_count;
$analyzed_files += $progress->analyzed_files;
}
$this->max_total_mem = \max($this->max_total_mem, $total_max_mem);
CLI::outputProgressLine('analyze', $total_progress / count($this->progress), $total_cur_mem, $this->max_total_mem, $file_count, $analyzed_files);
}
/**
* @param array<list> $process_task_data_iterator
* An array of task data items to be divided up among the
* workers. The size of this is the number of forked processes.
*
* @param Closure():void $startup_closure
* A closure to execute upon starting a child
*
* @param Closure(int,mixed,int) $task_closure
* A method to execute on each task data.
* This closure must return an array (to be gathered).
*
* @param Closure():array $shutdown_closure
* A closure to execute upon shutting down a child
* @throws InvalidArgumentException if count($process_task_data_iterator) < 2
* @throws AssertionError if pcntl is disabled before using this
* @suppress PhanAccessMethodInternal
*/
public function __construct(
array $process_task_data_iterator,
Closure $startup_closure,
Closure $task_closure,
Closure $shutdown_closure
) {
$process_task_data_iterator = \array_values($process_task_data_iterator);
$pool_size = count($process_task_data_iterator);
if ($pool_size < 2) {
throw new InvalidArgumentException('The pool size must be >= 2 to use the fork pool.');
}
if (!\extension_loaded('pcntl')) {
throw new AssertionError('The pcntl extension must be loaded in order for Phan to be able to fork.');
}
// We'll keep track of if this is the parent process
// so that we can tell who will be doing the waiting
$is_parent = false;
// Fork as many times as requested to get the given
// pool size
for ($proc_id = 0; $proc_id < $pool_size; $proc_id++) {
// Create an IPC socket pair.
$sockets = \stream_socket_pair(\STREAM_PF_UNIX, \STREAM_SOCK_STREAM, \STREAM_IPPROTO_IP);
if (!$sockets) {
\error_log("unable to create stream socket pair");
exit(EXIT_FAILURE);
}
// Fork
$pid = \pcntl_fork();
if ($pid < 0) {
\error_log(\posix_strerror(\posix_get_last_error()));
exit(EXIT_FAILURE);
}
// Parent
if ($pid > 0) {
$is_parent = true;
$this->child_pid_list[] = $pid;
$read_stream = self::streamForParent($sockets);
$i = \count($this->progress);
$task_data_iterator = $process_task_data_iterator[$proc_id];
$this->progress[] = new Progress(0.0, 0, count($task_data_iterator));
$this->read_streams[] = $read_stream;
$this->readers[intval($read_stream)] = new Reader($read_stream, function (string $notification_type, string $payload) use ($i): void {
switch ($notification_type) {
case Writer::TYPE_PROGRESS:
$progress = unserialize($payload);
$this->updateProgress($i, $progress);
break;
case Writer::TYPE_ISSUE_LIST:
// This worker has stopped, and should no longer be using memory shortly
// @phan-suppress-next-line PhanAccessReadOnlyProperty
$this->progress[$i]->cur_mem = 0;
$issues = unserialize($payload);
if (\is_array($issues) && $issues) {
\array_push($this->issues, ...$issues);
}
break;
}
});
continue;
}
// Child
if ($pid === 0) {
$is_parent = false;
break;
}
}
// If we're the parent, return
if ($is_parent) {
return;
}
// Get the write stream for the child.
if (!isset($sockets)) {
throw new AssertionError('$sockets must be set if this is the child process');
}
$write_stream = self::streamForChild($sockets);
Writer::initialize($write_stream);
// Execute anything the children wanted to execute upon
// starting up
$startup_closure();
// Get the work for this process
$task_data_iterator = $process_task_data_iterator[$proc_id];
$task_count = \count($task_data_iterator);
foreach ($task_data_iterator as $i => $task_data) {
$task_closure($i, $task_data, $task_count);
}
// Execute each child's shutdown closure before
// exiting the process
$results = $shutdown_closure();
// Serialize this child's produced results and send them to the parent.
Writer::emitIssues($results ?: []);
\fclose($write_stream);
// Children exit after completing their work
exit(EXIT_SUCCESS);
}
/**
* Prepare the socket pair to be used in a parent process and
* return the stream the parent will use to read results.
*
* @param array{0:resource, 1:resource} $sockets the socket pair for IPC
* @return resource
*/
private static function streamForParent(array $sockets)
{
[$for_read, $for_write] = $sockets;
// The parent will not use the write channel, so it
// must be closed to prevent deadlock.
\fclose($for_write);
// stream_select will be used to read multiple streams, so these
// must be set to non-blocking mode.
if (!\stream_set_blocking($for_read, false)) {
\error_log('unable to set read stream to non-blocking');
exit(EXIT_FAILURE);
}
return $for_read;
}
/**
* Prepare the socket pair to be used in a child process and return
* the stream the child will use to write results.
*
* @param array{0:resource, 1:resource} $sockets the socket pair for IPC
* @return resource
*/
private static function streamForChild(array $sockets)
{
[$for_read, $for_write] = $sockets;
// The while will not use the read channel, so it must
// be closed to prevent deadlock.
\fclose($for_read);
return $for_write;
}
/**
* Read the results that each child process has serialized on their write streams.
* The results are returned in an array, one for each worker. The order of the results
* is not maintained.
*
* @return list<IssueInstance>
* @suppress PhanAccessMethodInternal
*/
private function readResultsFromChildren(): array
{
// Create an array of all active streams, indexed by
// resource id.
$streams = [];
foreach ($this->read_streams as $stream) {
$streams[intval($stream)] = $stream;
}
// Read the data off of all the stream.
while (count($streams) > 0) {
$needs_read = \array_values($streams);
$needs_write = null;
$needs_except = null;
// Wait for data on at least one stream.
$num = \stream_select($needs_read, $needs_write, $needs_except, null /* no timeout */);
if ($num === false) {
\error_log("unable to select on read stream");
exit(EXIT_FAILURE);
}
// For each stream that was ready, read the content.
foreach ($this->readers as $reader) {
$reader->readMessages();
}
foreach ($needs_read as $file) {
if (\feof($file)) {
\fclose($file);
$idx = intval($file);
unset($streams[$idx]);
}
}
}
$this->assertAnalysisWorkersExitedNormally();
return $this->issues;
}
/**
* Exit with a non-zero exit code if any of the workers exited without sending a valid response.
* @suppress PhanAccessMethodInternal
*/
private function assertAnalysisWorkersExitedNormally(): void
{
// Verify that the readers worked.
$saw_errors = false;
foreach ($this->readers as $reader) {
$errors = $reader->computeErrorsAfterRead();
if (StringUtil::isNonZeroLengthString($errors)) {
// @phan-suppress-next-line PhanPluginRemoveDebugCall
\fwrite(\STDERR, "Saw errors for an analysis worker:\n" . $errors);
$saw_errors = true;
}
}
if ($saw_errors) {
exit(EXIT_FAILURE);
}
}
/**
* Wait for all child processes to complete
* @return list<IssueInstance>
*/
public function wait(): array
{
// Read all the streams from child processes into an array.
$content = $this->readResultsFromChildren();
// Wait for all children to return
foreach ($this->child_pid_list as $child_pid) {
if (\pcntl_waitpid($child_pid, $status) < 0) {
\error_log(\posix_strerror(\posix_get_last_error()));
}
// Check to see if the child died a graceful death
if (\pcntl_wifsignaled($status)) {
$return_code = \pcntl_wexitstatus($status);
$term_sig = \pcntl_wtermsig($status);
$this->did_have_error = true;
\error_log("Child terminated with return code $return_code and signal $term_sig");
}
}
return $content;
}
/**
* Returns true if this had an error, e.g. due to memory limits or due to a child process crashing.
*/
public function didHaveError(): bool
{
return $this->did_have_error;
}
}

File Metadata

Mime Type
text/x-php
Expires
Oct 5 2026, 00:26 (5 w, 6 d ago)
Storage Engine
local-disk
Storage Format
Raw Data
Storage Handle
23/bb/827a2a25a219ff428ad9c30eb164
Default Alt Text
ForkPool.php (12 KB)

Event Timeline