/**
  * {@inheritDoc}
  */
 protected function forkWorkers($numProcs)
 {
     $this->prepareEnvironment();
     // Create the child processes
     for ($i = 0; $i < $numProcs; $i++) {
         $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
         // Do the fork
         $pid = pcntl_fork();
         if ($pid === -1 || $pid === false) {
             echo "Error creating child processes\n";
             exit(1);
         }
         if (!$pid) {
             $this->initChild();
             $this->childNumber = $i;
             $this->input = $sockets[0];
             $this->output = $sockets[0];
             fclose($sockets[1]);
             return 'child';
         } else {
             // This is the parent process
             $this->children[$pid] = true;
             fclose($sockets[0]);
             $childSockets[] = $sockets[1];
         }
     }
     $this->feedChildren($childSockets);
     foreach ($childSockets as $socket) {
         fclose($socket);
     }
     return 'parent';
 }

 function test()
 {
     $pipe = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     $base = new Base();
     $called = 0;
     $base->read($pipe[0], function ($event) use(&$called) {
         $called++;
         fgets($event->fd);
         if ($called === 3) {
             $event->base->halt();
         }
     });
     $pid = pcntl_fork();
     if (!$pid) {
         fwrite($pipe[1], "foo\n");
         usleep(500);
         fwrite($pipe[1], "bar\n");
         usleep(500);
         fwrite($pipe[1], "baz\n");
         usleep(500);
         exit;
     }
     $base->loop();
     pcntl_waitpid($pid, $s);
     $this->assertEquals(3, $called);
 }

 /**
  * Executes the callback in a different thread. All arguments to this method will be passed on to the callback.
  *
  * The callback will be invoked but the script will not wait for it to finish.
  * @return null
  */
 public function run()
 {
     $pid = @pcntl_fork();
     if ($pid == -1) {
         throw new ZiboException('Could not run the thread: unable to fork the callback');
     }
     if ($pid) {
         // parent process code
         $this->pid = $pid;
     } else {
         // child process code
         pcntl_signal(SIGTERM, array($this, 'signalHandler'));
         try {
             $this->callback->invokeWithArrayArguments(func_get_args());
         } catch (Exception $exception) {
             $message = $exception->getMessage();
             if (!$message) {
                 $message = get_class($exception);
             }
             Zibo::getInstance()->runEvent(Zibo::EVENT_LOG, $message, $exception->getTraceAsString(), 1);
             echo $message . "\n";
             echo $exception->getTraceAsString();
         }
         exit;
     }
 }

 public function spawn($function, $preFunction = null, $postFunction = null)
 {
     $pid = pcntl_fork();
     if ($pid === -1) {
         // for some reason the fork failed.
         return false;
     } else {
         if ($pid) {
             // fork is created and running give the pid back to the parent.
             return $pid;
         } else {
             // we are in the child process
             if (!is_null($preFunction)) {
                 $preFunction();
                 // run the pre function
             }
             $function();
             if (!is_null($postFunction)) {
                 $postFunction();
                 // run the post function
             }
             exit;
             // functions are done kill the scripts
         }
     }
 }

 public function testServer()
 {
     $pid = pcntl_fork();
     if ($pid == 0) {
         socket_close($this->input[0]);
         socket_close($this->output[0]);
         $input = new \PHPixie\Illusion\Socket($this->input[1]);
         $output = new \PHPixie\Illusion\Socket($this->output[1]);
         $server = new \PHPixie\Illusion\Server(4747, 'localhost', $input, $output);
         $server->run();
     } else {
         socket_close($this->input[1]);
         socket_close($this->output[1]);
         $input = new \PHPixie\Illusion\Socket($this->input[0]);
         $output = new \PHPixie\Illusion\Socket($this->output[0]);
         $message = array('action' => 'route', 'method' => 'GET', 'headers' => array('Content-Type: text/plain'), 'path' => '/hello', 'response' => 'world');
         $input->write($message);
         $url = 'http://localhost:4747/hello';
         $response = $output->read(true);
         $this->assertEquals(array(array('url' => $url)), $response);
         $contents = file_get_contents($url);
         $this->assertEquals('world', $contents);
         $input->write(array('action' => 'stop'));
         sleep(2);
     }
 }

 public function run($concurrent = 5)
 {
     $this->completed = 0;
     foreach ($this->workQueue as $data) {
         $pid = pcntl_fork();
         // clone
         switch ($pid) {
             case -1:
                 throw new \Exception("Out of memory!");
             case 0:
                 // child process
                 call_user_func($this->callback, $data);
                 exit(0);
             default:
                 // parent process
                 $this->processes[$pid] = TRUE;
                 // log the child process ID
         }
         // wait on a process to finish if we're at our concurrency limit
         while (count($this->processes) >= $concurrent) {
             $this->reapChild();
             usleep(500);
         }
     }
     // wait on remaining processes to finish
     while (count($this->processes) > 0) {
         $this->reapChild();
         usleep(500);
     }
 }

 /**
  * Start the thread
  *
  * @throws RuntimeException
  */
 public function execute()
 {
     $this->threadKey = 'thread_' . rand(1000, 9999) . rand(1000, 9999) . rand(1000, 9999) . rand(1000, 9999);
     if (($this->pid = pcntl_fork()) == -1) {
         throw new RuntimeException('Couldn\'t fork the process');
     }
     if ($this->pid) {
         // Parent
         //pcntl_wait($status); //Protect against Zombie children
     } else {
         // Child.
         pcntl_signal(SIGTERM, array($this, 'signalHandler'));
         $args = func_get_args();
         $callable = $this->callable;
         if (!is_string($callable)) {
             $callable = (array) $this->callable;
         }
         try {
             $return = call_user_func_array($callable, (array) $args);
             if (!is_null($return)) {
                 $this->saveResult($return);
             }
             // Executed only in PHP 7, will not match in PHP 5.x
         } catch (\Throwable $t) {
             $this->saveResult($t);
             // Executed only in PHP 5. Remove when PHP 5.x is no longer necessary.
         } catch (\Exception $ex) {
             $this->saveResult($ex);
         }
         exit(0);
     }
 }

 protected function launchJob($jobID)
 {
     if (FALSE === $this->fire('onLauncher', array(&$this))) {
         usleep(20000);
         return false;
     }
     $pid = pcntl_fork();
     if ($pid == -1) {
         $this->fire('onLaunchJobError', array(&$this));
         return false;
     } else {
         if ($pid) {
             $this->currentObjects[$pid] = $this->childObject;
             $this->currentJobs[$pid] = $jobID;
             if (isset($this->signalQueue[$pid])) {
                 $this->childSignalHandler(SIGCHLD, $pid, $this->signalQueue[$pid]);
                 unset($this->signalQueue[$pid]);
             }
         } else {
             unset($this->currentObjects);
             $exitStatus = 0;
             setproctitle($this->childsProcName);
             $this->fire('onLaunchJob', array(&$this));
             exit($exitStatus);
         }
     }
     return true;
 }

 /**
  * Start
  *
  * @param OutputInterface $output
  *
  * @return void
  */
 protected function start(OutputInterface $output)
 {
     $output->writeln('Starting daemon...');
     if (file_exists($this->pidfile)) {
         $output->writeln('<error>Daemon process is already running</error>');
         return null;
     }
     $pid = pcntl_fork();
     if ($pid == -1) {
         $output->writeln('<error>Could not fork</error>');
         return null;
     } elseif ($pid) {
         file_put_contents($this->pidfile, $pid);
         $output->writeln('Daemon started with PID ' . $pid);
     } else {
         $terminated = false;
         pcntl_signal(SIGTERM, function ($signo) use(&$terminated) {
             if ($signo == SIGTERM) {
                 $terminated = true;
             }
         });
         while (!$terminated) {
             $this->executeOperation();
             pcntl_signal_dispatch();
             sleep(1);
         }
         $output->writeln('Daemon stopped');
     }
 }

 public function startAction()
 {
     $config = Yaf_Registry::get('monitor_config');
     $dataName = $config['loghandle'][$this->logType]['data'];
     $stepTime = intval($config['loghandle'][$this->logType]['step']);
     $logPersistentService = new Service_Log_Persistent();
     $logApp = $logPersistentService->getAppId();
     foreach ($logApp as $table => $appId) {
         $pid = pcntl_fork();
         if (!$pid) {
             $logDataService = new Service_Log_Data($this->logType, $dataName);
             $dataResult = $logDataService->getWeblog($table, $stepTime);
             if (!$dataResult) {
                 return false;
             }
             $etlModel = Model_Loghandle_Etl_Factory::create($this->logType, 'key');
             $extraData = array('logapp' => $logApp, 'data' => $dataResult, 'table' => $table);
             $transData = $etlModel->transform($extraData);
             $firstData = current($dataResult);
             $time = $firstData['time']->sec;
             $etlModel->load($transData, $time);
             exit;
         }
     }
     return false;
 }

 function background()
 {
     /*
      * This prefers to Starting PHP 5.4 (dotdeb), maybe earlier for some version/distrib
      * seems MySQL connection using mysqli driver get lost when fork.
      * Need to unset it so that child process recreate it.
      *
      * @todo FIXME cleaner way to do it ?
      */
     global $_DB_DATAOBJECT;
     unset($_DB_DATAOBJECT['CONNECTIONS']);
     $pid = pcntl_fork();
     if ($pid < 0) {
         // error
         common_log(LOG_ERR, "Could not fork.");
         return false;
     } else {
         if ($pid > 0) {
             // parent
             common_log(LOG_INFO, "Successfully forked.");
             exit(0);
         } else {
             // child
             return true;
         }
     }
 }

 /**
  * Run jobs
  *
  * Run all jobs provided by the job provider.
  *
  * Jobs are run parallel in the background. The number of jobs executed in 
  * parallel can be specified as the second parameter.
  *
  * Returns once all jobs have been executed.
  * 
  * @param JobProvider $jobs 
  * @param int $parallel 
  * @return void
  */
 public function run(JobProvider $jobs, $parallel = 4)
 {
     $this->logger->startExecutor($this, $jobs);
     $forks = array();
     $jobNr = 0;
     while ($jobs->hasJobs() || count($forks)) {
         while (count($forks) < $parallel && ($job = $jobs->getNextJob())) {
             $this->logger->progressJob(++$jobNr);
             if (($forks[] = pcntl_fork()) === 0) {
                 // We are the newly forked child, just execute the job
                 call_user_func($job);
                 exit(0);
             }
         }
         do {
             // Check if the registered jobs are still alive
             if ($pid = pcntl_wait($status)) {
                 // Job has finished
                 $jobId = array_search($pid, $forks);
                 unset($forks[$jobId]);
             }
         } while (count($forks) >= $parallel);
     }
     $this->logger->finishedExecutor();
 }

function main($argv)
{
    global $cfgArr;
    if (!isset($argv[1])) {
        exit("请提供参数");
    }
    $act = $argv[1];
    $nbr_workers = intval($cfgArr['worker']['worker_num']);
    // $nbr_workers = isset($argv[2]) ? intval($argv[2]) : $nbr_workers;
    if ($act == 'stop') {
        $exec_str = "ps -ef |grep workers.php |awk '{print \$2}'|xargs kill -9";
        exec($exec_str);
    }
    if ($act == 'start' || $act == 'restart') {
        //v($nbr_workers);
        //$exec_str = "ps -ef |grep workers.php |awk '{print $2}'|xargs kill -9";
        //exec( $exec_str );
        for ($worker_nbr = 0; $worker_nbr < $nbr_workers; $worker_nbr++) {
            $pid = pcntl_fork();
            if ($pid == 0) {
                //子进程得到的$pid为0, 所以这里是子进程执行的逻辑。
                worker_thread2($worker_nbr);
            }
        }
    }
    sleep(1);
}

 /**
  * Launch a job from the job queue
  * @param Schedule_Maintenance_Job $job
  */
 protected function launchJob($job)
 {
     $pid = pcntl_fork();
     if ($pid == -1) {
         //Problem launching the job
         Logger::error('Could not launch new job with id [ ' . $job->getId() . ' ], exiting');
         return false;
     } else {
         if ($pid) {
             $this->currentJobs[$pid] = $job->getId();
             if (isset($this->signalQueue[$pid])) {
                 $this->childSignalHandler(SIGCHLD, $pid, $this->signalQueue[$pid]);
                 unset($this->signalQueue[$pid]);
             }
         } else {
             //Forked child
             try {
                 Logger::debug("Executing job [ " . $job->getId() . " ] as forked child");
                 Pimcore_Resource::reset();
                 $job->execute();
             } catch (Exception $e) {
                 Logger::error($e);
                 Logger::error("Failed to execute job with id [ " . $job->getId() . " ] and method [ " . $job->getMethod() . " ]");
             }
             $job->unlock();
             Logger::debug("Done with job [ " . $job->getId() . " ]");
             exit(0);
         }
     }
     return true;
 }

 /**
  * Запуск демона. 
  * @return Undefined Запуск демона не может ничего возвращать. Демон - это конечная программа, поэтому должен все необходимые действия делать сам.
  * @throws Exception Исключение в случае неудачи запуска демона.
  */
 public function start()
 {
     $name = __CLASS__;
     $this->logger->console("Starting '{$name}' daemon.");
     //Запуск в режиме дебага, если необходимо.
     if ($this->debug) {
         return $this->runDebug();
     }
     // Создаем дочерний процесс
     // весь код после pcntl_fork() будет выполняться
     // двумя процессами: родительским и дочерним
     $pid = pcntl_fork();
     if ($pid == -1) {
         // Не удалось создать дочерний процесс
         $msg = "Unable to create child process. Exit.";
         $logger->console($msg);
         throw new Exception($msg);
     }
     if ($pid) {
         //родительский процесс уходит, так как мы работаем в фоновом режиме
         $this->logger->console("Child process pid: {$pid}");
         return;
     }
     // А этот код выполнится дочерним процессом
     $childpid = getmypid();
     $this->run($childpid);
     //Так как демон конечная программа, выполнение php я остановлю насильно
     exit;
 }

 public final function execute()
 {
     $hasThreads = function_exists('pcntl_signal');
     if (!$hasThreads || Cli::getInstance()->isSimulation()) {
         flush();
         try {
             return $this->executeNoThread();
         } catch (Interrupt $e) {
             throw $e;
         } catch (Exception $e) {
             echo $e;
         }
         return;
     }
     pcntl_signal(SIGCHLD, SIG_IGN);
     $pid = pcntl_fork();
     if ($pid < 1) {
         $this->_run();
         posix_kill(posix_getpid(), 9);
         pcntl_waitpid(posix_getpid(), $temp = 0, WNOHANG);
         pcntl_wifexited($temp);
         exit;
         //Make sure we exit...
     } else {
         $this->pid = $pid;
     }
 }

function sync($t_username, $s_email, $s_pwd, $apikey)
{
    $pid = pcntl_fork();
    if (!$pid) {
        doSync($t_username, $s_email, $s_pwd, $apikey);
    }
}

 /**
  * Daemonize the current process so it can run in the background.
  *
  * If $pidfile is supplied, the process ID is written there.
  * Provide absolute path names for the parameters to avoid file not found errors or logs inside the source code folder.
  *
  * If an error occurred, a RuntimeException is thrown.
  *
  * @param string $pidfile File to write the process ID of the daemon to
  * @param string $stderr File to redirect STDERR to
  * @param string $stdout File to redirect STDOUT to
  * @param string $stdin File to read STDIN from
  * @throws \RuntimeException
  * @return true
  */
 public static function daemonize($pidfile = null, $stderr = '/dev/null', $stdout = '/dev/null', $stdin = '/dev/null')
 {
     // Allow only cli scripts to daemonize, otherwise you may confuse your webserver
     if (\php_sapi_name() !== 'cli') {
         throw new \RuntimeException('Can only daemonize a CLI process!');
     }
     self::checkPID($pidfile);
     self::reopenFDs($stdin, $stdout, $stderr);
     if (($pid1 = @\pcntl_fork()) < 0) {
         throw new \RuntimeException('Failed to fork, reason: "' . \pcntl_strerror(\pcntl_get_last_error()) . '"');
     } elseif ($pid1 > 0) {
         exit;
     }
     if (@posix_setsid() === -1) {
         throw new \RuntimeException('Failed to become session leader, reason: "' . \posix_strerror(\posix_get_last_error()) . '"');
     }
     if (($pid2 = @\pcntl_fork()) < 0) {
         throw new \RuntimeException('Failed to fork, reason: "' . \pcntl_strerror(\pcntl_get_last_error()) . '"');
     } elseif ($pid2 > 0) {
         exit;
     }
     chdir('/');
     umask(022);
     self::writePID($pidfile);
     return true;
 }

 /**
  * Run some code in a thread.
  *
  * @param callable $func The function to execute
  * @param array|mixed $args The arguments (or a single argument) to pass to the function
  *
  * @return int The pid of the thread created to execute this code
  */
 public function call(callable $func, $args = null)
 {
     $pid = pcntl_fork();
     if ($pid == -1) {
         throw new \Exception("Failed to fork");
     }
     # If this is the child process, then run the requested function
     if (!$pid) {
         try {
             if ($args === null) {
                 $func();
             } else {
                 call_user_func_array($func, $args);
             }
         } catch (\Exception $e) {
             $memory = shmop_open($this->memoryKey, "c", 0644, static::SHARED_MEMORY_LIMIT);
             $errors = shmop_read($memory, 0, static::SHARED_MEMORY_LIMIT);
             $errors = trim($errors);
             if ($errors) {
                 $errors .= "\n";
             }
             $errors .= "Exception: " . $e->getMessage() . " (" . $e->getFile() . ":" . $e->getLine() . ")";
             shmop_write($memory, $errors, 0);
             shmop_close($memory);
             exit(1);
         }
         # Then we must exit or else we will end up the child process running the parent processes code
         die;
     }
     $this->threads[$pid] = $pid;
     return $pid;
 }

 public function run()
 {
     $count = 0;
     $socket = stream_socket_server($this->socketString, &$errno, &$errstr);
     if (!$socket) {
         throw new RuntimeException($errstr);
     }
     while (true) {
         while (count($this->children) < $this->maxConnections) {
             $count++;
             $pid = pcntl_fork();
             if ($pid == -1) {
                 throw new RuntimeException('Couldn\'t fork');
             } elseif ($pid == 0) {
                 $this->createChild($socket, $count);
                 exit;
             }
             $this->children[] = $pid;
         }
         while (pcntl_wait($status, WNOHANG or WUNTRACED) > 0) {
             usleep(500000);
         }
         while (list($key, $val) = each($this->children)) {
             if (!posix_kill($val, 0)) {
                 unset($this->children[$key]);
             }
         }
         $this->children = array_values($this->children);
         usleep(500000);
     }
 }