ForkController.php

Go to the documentation of this file.

<?php
namespace MediaWiki\Maintenance;
 
use MediaWiki\MediaWikiServices;
use RuntimeException;
use Wikimedia\ObjectCache\RedisConnectionPool;
 
class ForkController {
    protected $children = [];
 
    protected $childNumber = 0;
 
    protected $termReceived = false;
 
    protected $flags = 0;
 
    protected $procsToStart = 0;
 
    protected static $RESTARTABLE_SIGNALS = [];
 
    protected $exitStatuses = [];
 
    private const RESTART_ON_ERROR = 1;
 
    public function __construct( $numProcs, $flags = 0 ) {
        if ( !wfIsCLI() ) {
            throw new RuntimeException( "MediaWiki\Maintenance\ForkController cannot be used from the web." );
        } elseif ( !extension_loaded( 'pcntl' ) ) {
            throw new RuntimeException(
                'MediaWiki\Maintenance\ForkController requires pcntl extension to be installed.'
            );
        } elseif ( !extension_loaded( 'posix' ) ) {
            throw new RuntimeException(
                'MediaWiki\Maintenance\ForkController requires posix extension to be installed.'
            );
        }
        $this->procsToStart = $numProcs;
        $this->flags = $flags;
 
        // Define this only after confirming PCNTL support
        self::$RESTARTABLE_SIGNALS = [
            SIGFPE, SIGILL, SIGSEGV, SIGBUS,
            SIGABRT, SIGSYS, SIGPIPE, SIGXCPU, SIGXFSZ,
        ];
    }
 
    public function start() {
        // Trap SIGTERM
        pcntl_signal( SIGTERM, [ $this, 'handleTermSignal' ], false );
 
        do {
            // Start child processes
            if ( $this->procsToStart ) {
                if ( $this->forkWorkers( $this->procsToStart ) == 'child' ) {
                    return 'child';
                }
                $this->procsToStart = 0;
            }
 
            // Check child status
            $status = false;
            $deadPid = pcntl_wait( $status );
 
            if ( $deadPid > 0 ) {
                // Respond to child process termination
                unset( $this->children[$deadPid] );
                if ( $this->flags & self::RESTART_ON_ERROR ) {
                    if ( pcntl_wifsignaled( $status ) ) {
                        // Restart if the signal was abnormal termination
                        // Don't restart if it was deliberately killed
                        $signal = pcntl_wtermsig( $status );
                        if ( in_array( $signal, self::$RESTARTABLE_SIGNALS ) ) {
                            echo "Worker exited with signal $signal, restarting\n";
                            $this->procsToStart++;
                        }
                    } elseif ( pcntl_wifexited( $status ) ) {
                        // Restart on non-zero exit status
                        $exitStatus = pcntl_wexitstatus( $status );
                        if ( $exitStatus != 0 ) {
                            echo "Worker exited with status $exitStatus, restarting\n";
                            $this->procsToStart++;
                        } else {
                            echo "Worker exited normally\n";
                        }
                    }
                }
 
                if ( pcntl_wifexited( $status ) ) {
                    $exitStatus = pcntl_wexitstatus( $status );
                    echo "Worker exited with status $exitStatus\n";
                    $this->exitStatuses[] = $exitStatus;
                }
 
                // Throttle restarts
                if ( $this->procsToStart ) {
                    usleep( 500_000 );
                }
            }
 
            // Run signal handlers
            if ( function_exists( 'pcntl_signal_dispatch' ) ) {
                pcntl_signal_dispatch();
            } else {
                declare( ticks=1 ) {
                    // @phan-suppress-next-line PhanPluginDuplicateExpressionAssignment
                    $status = $status;
                }
            }
            // Respond to TERM signal
            if ( $this->termReceived ) {
                foreach ( $this->children as $childPid => $unused ) {
                    posix_kill( $childPid, SIGTERM );
                }
                $this->termReceived = false;
            }
        } while ( count( $this->children ) );
        pcntl_signal( SIGTERM, SIG_DFL );
        return 'done';
    }
 
    public function allSuccessful(): bool {
        return array_reduce(
            $this->exitStatuses,
            static fn ( $acc, $status ) => $acc && ( $status === 0 ),
            true
        );
    }
 
    public function getChildNumber() {
        return $this->childNumber;
    }
 
    protected function prepareEnvironment() {
        // Don't share DB, storage, or memcached connections
        MediaWikiServices::resetChildProcessServices();
        MediaWikiServices::getInstance()->getObjectCacheFactory()->clear();
        RedisConnectionPool::destroySingletons();
    }
 
    protected function forkWorkers( $numProcs ) {
        $this->prepareEnvironment();
 
        // Create the child processes
        for ( $i = 0; $i < $numProcs; $i++ ) {
            // Do the fork
            $pid = pcntl_fork();
            if ( $pid === -1 ) {
                echo "Error creating child processes\n";
                exit( 1 );
            }
 
            if ( !$pid ) {
                $this->initChild();
                $this->childNumber = $i;
                return 'child';
            } else {
                // This is the parent process
                $this->children[$pid] = true;
            }
        }
 
        return 'parent';
    }
 
    protected function initChild() {
        $this->children = null;
        pcntl_signal( SIGTERM, SIG_DFL );
    }
 
    protected function handleTermSignal( int $signal ) {
        $this->termReceived = true;
    }
}
 
class_alias( ForkController::class, 'ForkController' );