Fork one worker process.

@param string $start_file
430637
public static function forkOneWorkerForWindows($start_file)
    {
        $start_file = realpath($start_file);
        $std_file = sys_get_temp_dir() . '/'.str_replace(array('/', "\\", ':'), '_', $start_file).'.out.txt';

        $descriptorspec = array(
            0 => array('pipe', 'a'), // stdin
            1 => array('file', $std_file, 'w'), // stdout
            2 => array('file', $std_file, 'w') // stderr
        );


        $pipes       = array();
        $process     = proc_open("php \"$start_file\" -q", $descriptorspec, $pipes);
        $std_handler = fopen($std_file, 'a+');
        stream_set_blocking($std_handler, 0);

        if (empty(static::$globalEvent)) {
            static::$globalEvent = new Select();
            Timer::init(static::$globalEvent);
        }
        $timer_id = Timer::add(0.1, function()use($std_handler)
        {
            Worker::safeEcho(fread($std_handler, 65535));
        });

        // 保存子进程句柄
        static::$_processForWindows[$start_file] = array($process, $start_file, $timer_id);
    }
109186
async def start_worker(
        self,
        cmd: List[str],
        input_source: str,
        output: Optional[str] = None,
        extra_cmd: Optional[str] = None,
        pattern: Optional[str] = None,
        reading: str = FFMPEG_STDERR,
    ) -> None:
        """"""
        if self.is_running:
            _LOGGER.warning("Can't start worker. It is allready running!")
            return

        if reading == FFMPEG_STDERR:
            stdout = False
            stderr = True
        else:
            stdout = True
            stderr = False

        # start ffmpeg and reading to queue
        await self.open(
            cmd=cmd,
            input_source=input_source,
            output=output,
            extra_cmd=extra_cmd,
            stdout_pipe=stdout,
            stderr_pipe=stderr,
        )

        self._input = await self.get_reader(reading)

        # start background processing
        self._read_task = self._loop.create_task(self._process_lines(pattern))
        self._loop.create_task(self._worker_process())
255898
protected static function init()
    {
        set_error_handler(function($code, $msg, $file, $line){
            Worker::safeEcho("$msg in file $file on line $line\n");
        });

        // Start file.
        $backtrace        = debug_backtrace();
        static::$_startFile = $backtrace[count($backtrace) - 1]['file'];


        $unique_prefix = str_replace('/', '_', static::$_startFile);

        // Pid file.
        if (empty(static::$pidFile)) {
            static::$pidFile = __DIR__ . "/../$unique_prefix.pid";
        }

        // Log file.
        if (empty(static::$logFile)) {
            static::$logFile = __DIR__ . '/../workerman.log';
        }
        $log_file = (string)static::$logFile;
        if (!is_file($log_file)) {
            touch($log_file);
            chmod($log_file, 0622);
        }

        // State.
        static::$_status = static::STATUS_STARTING;

        // For statistics.
        static::$_globalStatistics['start_timestamp'] = time();
        static::$_statisticsFile                      = sys_get_temp_dir() . "/$unique_prefix.status";

        // Process title.
        static::setProcessTitle('WorkerMan: master process  start_file=' . static::$_startFile);

        // Init data for worker id.
        static::initId();

        // Timer init.
        Timer::init();
    }
1211898
public function runProcess($output_file = '/dev/null', $append = false)
    {
        switch (PHP_OS) {
            case 'Windows':
                shell_exec(sprintf('%s &', $this->function_code, $output_file));
                break;
            case 'Linux':
                $this->pid = (int)shell_exec(sprintf('php -r "%s" %s %s 2>&1 & echo $!', str_replace('"', '\"', $this->function_code), ($append) ? '>>' : '>', $output_file));
                break;
            default:
                throw new RuntimeException(sprintf(
                    'Could not execute command "%s" because operating system "%s" is not supported', $this->function_code, PHP_OS
                ));
        }
    }
1032550
protected static function forkWorkersForWindows()
    {
        $files = static::getStartFilesForWindows();
        global $argv;
        if(in_array('-q', $argv) || count($files) === 1)
        {
            if(count(static::$_workers) > 1)
            {
                static::safeEcho("@@@ Error: multi workers init in one php file are not support @@@\r\n");
                static::safeEcho("@@@ Please visit http://wiki.workerman.net/Multi_woker_for_win @@@\r\n");
            }
            elseif(count(static::$_workers) <= 0)
            {
                exit("@@@no worker inited@@@\r\n\r\n");
            }

            reset(static::$_workers);
            /** @var Worker $worker */
            $worker = current(static::$_workers);

            // Display UI.
            static::safeEcho(str_pad($worker->name, 21) . str_pad($worker->getSocketName(), 36) . str_pad($worker->count, 10) . "[ok]\n");
            $worker->listen();
            $worker->run();
            exit("@@@child exit@@@\r\n");
        }
        else
        {
            static::$globalEvent = new \Workerman\Events\Select();
            Timer::init(static::$globalEvent);
            foreach($files as $start_file)
            {
                static::forkOneWorkerForWindows($start_file);
            }
        }
    }
800148
protected static function forkOneWorker($worker)
    {
        // Get available worker id.
        $id = self::getId($worker->workerId, 0);
        if ($id === false) {
            return;
        }
        $process = new swoole_process(function(swoole_process $process) use($worker,$id){
            if ($worker->reusePort) {
                $worker->listen();
            }
            if (self::$_status === self::STATUS_STARTING) {
                self::resetStd();
            }
            self::$_pidMap  = array();
            self::$_workers = array($worker->workerId => $worker);
            Timer::delAll();
            $process->name('WorkerMan: worker process  ' . $worker->name . ' ' . $worker->getSocketName());
            $worker->setUserAndGroup();
            $worker->id = $id;
            $worker->run();
        });
        $pid = $process->start();
        self::$_pidMap[$worker->workerId][$pid] = $pid;
        self::$_idMap[$worker->workerId][$id]   = $pid;


    }
599696
def start_workers(self):
        """"""

        if self._started:
            raise InternalError("The method start() was called twice on a BaseRunnable object.")

        self._started = True

        for worker in self._workers:
            worker.start()
265813
private function ensureProcessIsStarted(): void
    {
        if (is_resource($this->process) === false) {
            $this->startProcess();

            $this->handleStartupErrors();
        }
    }
1062308
public function start() {

        // indicate the library that it should execute queued requests
        $this->_queue = false;

        // if we have to pause between batches of requests, process them sequentially, in batches
        if ($this->pause_interval > 0) $this->_process_paused();

        // if we don't have to pause between batches of requests, process them all at once
        else $this->_process();

    }
166183
def run_command_as_worker(self, command, after_post_fork_hook=False):
        
        self._set('worker-exec2' if after_post_fork_hook else 'worker-exec', command, multi=True)

        return self._section
839185
public function fork()
    {
        $pair = $this->pair();

        $pid = pcntl_fork();
        if ($pid < 0) {  // Failed fork
            throw new \RuntimeException('Failed to fork child process.');
        } elseif ($pid === 0) {  // is worker process
            fclose($pair[0]);
            foreach ($this->sockets as $socket) {
                fclose($socket);
            }
            $this->sockets = array();
            $this->loop($pair[1]);
            exit;
        }

        fclose($pair[1]);
        $this->sockets[$pid] = $pair[0];

        return $pid;
    }
529273
public function process(): void
    {
        $this->queue = msg_get_queue(ftok(__DIR__.DIRECTORY_SEPARATOR.'Queue.php', 't'));
        $this->catchSignal();
        $this->spawnInitialWorkers();
        $this->main();
    }
756575
public static function fork(callable $onStart = null, callable $onError = null, $id = 0)
    {
        if (!self::hasPcntl()) {
            return false;
        }

        $info = [];
        $pid = \pcntl_fork();

        // at parent, get forked child info
        if ($pid > 0) {
            $info = [
                'id'        => $id,
                'pid'       => $pid,
                'startTime' => \time(),
            ];
        } elseif ($pid === 0) { // at child
            $pid = self::getPid();

            if ($onStart) {
                $onStart($pid, $id);
            }
        } else {
            if ($onError) {
                $onError($pid);
            }

            throw new \RuntimeException('Fork child process failed! exiting.');
        }

        return $info;
    }
761619
function process(file, context) {
  fis.emit('process:start', file);
  pipe(file, 'parser');
  pipe(file, 'preprocessor');
  pipe(file, 'standard');
  postStandard(file, context);
  pipe(file, 'postprocessor');
  pipe(file, 'optimizer');
  fis.emit('process:end', file);
}
498383
protected function startupWorker($pool) : void
    {
        $workerId = $this->getId();
        $baseCmd = 'php %s --config %s --pool %s --name %s';
        $launcherPath = __DIR__ . '/cli/worker.php';
        $startupCmd = sprintf(
            $baseCmd,
            $launcherPath,
            $this->getConfigPath(),
            $pool,
            $workerId
        );

        exec(escapeshellcmd($startupCmd) . ' >> ' . $this->logPath.'/'.$pool.'.log 2>&1 &');
        $this->reloadPids();
    }
390670
public function start()
    {
        $this->logger->info("crontab start");
        $crontab = $this->createCrontab();
        $loop = Factory::create();

        // add periodic timer
        $loop->addPeriodicTimer(60, function () use ($crontab, $loop) {
            $loop->addTimer(60 - time() % 60, function () use ($crontab) {
                $pid = pcntl_fork();
                if ($pid > 0) {
                    return;
                } elseif ($pid == 0) {
                    $crontab->start(time());
                    exit();
                } else {
                    $this->logger->error("could not fork");
                }
            });
        });

        // recover the sub processes
        $loop->addPeriodicTimer(60, function () {
            while (($pid = pcntl_waitpid(0, $status, WNOHANG)) > 0) {
                $message = "process exit. pid:" . $pid . ". exit code:" . $status;
                $this->logger->info($message);
            }
        });

        $loop->run();
    }
867255
public static function openProcess($start_file)
    {
        // 保存子进程的输出
        $start_file = realpath($start_file);
        $std_file = sys_get_temp_dir() . '/'.str_replace(array('/', "\\", ':'), '_', $start_file).'.out.txt';
        // 将stdou stderr 重定向到文件
        $descriptorspec = array(
                0 => array('pipe', 'a'), // stdin
                1 => array('file', $std_file, 'w'), // stdout
                2 => array('file', $std_file, 'w') // stderr
        );
        
        // 保存stdin句柄，用来探测子进程是否关闭
        $pipes = array();
       
        // 打开子进程
        $process= proc_open("php \"$start_file\" -q", $descriptorspec, $pipes);
        
        // 打开stdout stderr 文件句柄
        $std_handler = fopen($std_file, 'a+');
        // 非阻塞
        stream_set_blocking($std_handler, 0);
        // 定时读取子进程的stdout stderr
        $timer_id = Timer::add(0.1, function()use($std_handler)
        {
            echo fread($std_handler, 65535);
        });
        
        // 保存子进程句柄
        static::$_process[$start_file] = array($process, $start_file, $timer_id);
    }