src/Workers/ForkingWorker.php
public function perform(): void
    {
        declare(ticks=1);

        $this->who = 'master:' . $this->name;
        $this->logContext = ['type' => $this->who, 'job.identifier' => null];
        $this->logger->info('{type}: worker started', $this->logContext);

        $this->reserver->beforeWork();

        $didWork = false;

        while (true) {
            // Don't wait on any processes if we're already in shutdown mode.
            if ($this->isShuttingDown() == true) {
                break;
            }

            while ($this->paused) {
                usleep(250000);
            }

            if ($didWork) {
                $this->title(
                    sprintf(
                        'Waiting for %s with interval %d sec',
                        implode(',', $this->reserver->getQueues()),
                        $this->interval
                    )
                );
                $didWork = false;
            }

            $job = $this->reserve();
            if ($job === null) {
                if ($this->interval == 0) {
                    break;
                }
                usleep($this->interval * 1000000);
                continue;
            }

            $this->setCurrentJob($job);
            $this->logContext['job.identifier'] = $job->jid;

            // fork processes
            $this->childStart();
            $this->watchdogStart($this->client->createSubscriber(['ql:log']));

            $this->title(sprintf('Forked %d at %s', $this->childPID, strftime('%F %T')));

            // Parent process, sit and wait
            while ($this->childProcesses > 0) {
                $status = null;
                $pid   = pcntl_wait($status, WUNTRACED);
                if ($pid > 0) {
                    if ($pid === $this->childPID) {
                        $exited = $this->childProcessStatus($status);
                    } elseif ($pid === $this->watchdogPID) {
                        $exited = $this->watchdogProcessStatus($status);
                    } else {
                        // unexpected?
                        $this->logger->info(sprintf("master received status for unknown PID %d; exiting\n", $pid));
                        exit(1);
                    }

                    if ($exited) {
                        --$this->childProcesses;
                        switch ($pid) {
                            case $this->childPID:
                                $this->childPID = null;
                                if ($this->watchdogPID) {
                                    // shutdown watchdog immediately if child has exited
                                    posix_kill($this->watchdogPID, SIGKILL);
                                }
                                break;

                            case $this->watchdogPID:
                                $this->watchdogPID = null;
                                break;
                        }
                    }
                }
            }

            foreach ($this->sockets as $socket) {
                socket_close($socket);
            }

            $this->sockets  = [];
            $this->setCurrentJob(null);
            $this->logContext['job.identifier'] = null;
            $didWork = true;

            /**
             * We need to reconnect due to bug in Redis library that always sends QUIT on destruction of \Redis
             * rather than just leaving socket around. This call will sometimes generate a broken pipe notice
             */
            $old = error_reporting();
            error_reporting($old & ~E_NOTICE);
            try {
                $this->client->reconnect();
            } finally {
                error_reporting($old);
            }
        }
    }
public function perform(): void
    {
        declare(ticks=1);

        $this->who = 'master:' . $this->name;
        $this->logContext = ['type' => $this->who, 'job.identifier' => null];
        $this->logger->info('{type}: worker started', $this->logContext);

        $this->reserver->beforeWork();

        $didWork = false;

        while (true) {
            // Don't wait on any processes if we're already in shutdown mode.
            if ($this->isShuttingDown() == true) {
                break;
            }

            while ($this->paused) {
                usleep(250000);
            }

            if ($didWork) {
                $this->title(
                    sprintf(
                        'Waiting for %s with interval %d sec',
                        implode(',', $this->reserver->getQueues()),
                        $this->interval
                    )
                );
                $didWork = false;
            }

            $job = $this->reserve();
            if ($job === null) {
                if ($this->interval == 0) {
                    break;
                }
                usleep($this->interval * 1000000);
                continue;
            }

            $this->setCurrentJob($job);
            $this->logContext['job.identifier'] = $job->jid;

            // fork processes
            $this->childStart();
            $this->watchdogStart($this->client->createSubscriber(['ql:log']));

            $this->title(sprintf('Forked %d at %s', $this->childPID, strftime('%F %T')));

            // Parent process, sit and wait
            while ($this->childProcesses > 0) {
                $status = null;
                $pid   = pcntl_wait($status, WUNTRACED);
                if ($pid > 0) {
                    if ($pid === $this->childPID) {
                        $exited = $this->childProcessStatus($status);
                    } elseif ($pid === $this->watchdogPID) {
                        $exited = $this->watchdogProcessStatus($status);
                    } else {
                        // unexpected?
                        $this->logger->info(sprintf("master received status for unknown PID %d; exiting\n", $pid));
                        exit(1);
                    }

                    if ($exited) {
                        --$this->childProcesses;
                        switch ($pid) {
                            case $this->childPID:
                                $this->childPID = null;
                                if ($this->watchdogPID) {
                                    // shutdown watchdog immediately if child has exited
                                    posix_kill($this->watchdogPID, SIGKILL);
                                }
                                break;

                            case $this->watchdogPID:
                                $this->watchdogPID = null;
                                break;
                        }
                    }
                }
            }

            foreach ($this->sockets as $socket) {
                socket_close($socket);
            }

            $this->sockets  = [];
            $this->setCurrentJob(null);
            $this->logContext['job.identifier'] = null;
            $didWork = true;

            /**
             * We need to reconnect due to bug in Redis library that always sends QUIT on destruction of \Redis
             * rather than just leaving socket around. This call will sometimes generate a broken pipe notice
             */
            $old = error_reporting();
            error_reporting($old & ~E_NOTICE);
            try {
                $this->client->reconnect();
            } finally {
                error_reporting($old);
            }
        }
    }
public function perform ( ) : void { declare ( ticks = 1 ) ; $ this -> who = 'master:' . $ this -> name ; $ this -> logContext = [ 'type' => $ this -> who , 'job.identifier' => null ] ; $ this -> logger -> info ( '{type}: worker started' , $ this -> logContext ) ; $ this -> reserver -> beforeWork ( ) ; $ didWork = false ; while ( true ) { // Don't wait on any processes if we're already in shutdown mode. if ( $ this -> isShuttingDown ( ) == true ) { break ; } while ( $ this -> paused ) { usleep ( 250000 ) ; } if ( $ didWork ) { $ this -> title ( sprintf ( 'Waiting for %s with interval %d sec' , implode ( ',' , $ this -> reserver -> getQueues ( ) ) , $ this -> interval ) ) ; $ didWork = false ; } $ job = $ this -> reserve ( ) ; if ( $ job === null ) { if ( $ this -> interval == 0 ) { break ; } usleep ( $ this -> interval * 1000000 ) ; continue ; } $ this -> setCurrentJob ( $ job ) ; $ this -> logContext [ 'job.identifier' ] = $ job -> jid ; // fork processes $ this -> childStart ( ) ; $ this -> watchdogStart ( $ this -> client -> createSubscriber ( [ 'ql:log' ] ) ) ; $ this -> title ( sprintf ( 'Forked %d at %s' , $ this -> childPID , strftime ( '%F %T' ) ) ) ; // Parent process, sit and wait while ( $ this -> childProcesses > 0 ) { $ status = null ; $ pid = pcntl_wait ( $ status , WUNTRACED ) ; if ( $ pid > 0 ) { if ( $ pid === $ this -> childPID ) { $ exited = $ this -> childProcessStatus ( $ status ) ; } elseif ( $ pid === $ this -> watchdogPID ) { $ exited = $ this -> watchdogProcessStatus ( $ status ) ; } else { // unexpected? $ this -> logger -> info ( sprintf ( "master received status for unknown PID %d; exiting\n" , $ pid ) ) ; exit ( 1 ) ; } if ( $ exited ) { -- $ this -> childProcesses ; switch ( $ pid ) { case $ this -> childPID : $ this -> childPID = null ; if ( $ this -> watchdogPID ) { // shutdown watchdog immediately if child has exited posix_kill ( $ this -> watchdogPID , SIGKILL ) ; } break ; case $ this -> watchdogPID : $ this -> watchdogPID = null ; break ; } } } } foreach ( $ this -> sockets as $ socket ) { socket_close ( $ socket ) ; } $ this -> sockets = [ ] ; $ this -> setCurrentJob ( null ) ; $ this -> logContext [ 'job.identifier' ] = null ; $ didWork = true ; /**
             * We need to reconnect due to bug in Redis library that always sends QUIT on destruction of \Redis
             * rather than just leaving socket around. This call will sometimes generate a broken pipe notice
             */ $ old = error_reporting ( ) ; error_reporting ( $ old & ~ E_NOTICE ) ; try { $ this -> client -> reconnect ( ) ; } finally { error_reporting ( $ old ) ; } } }
{@inheritdoc}

@return void
{ @inheritdoc }