/**
  * {@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';
 }

 /**
  * Test the process method to make sure that the FastCGI connection and the
  * environment has been initialized.
  *
  * @return void
  */
 public function testProcess()
 {
     // create a mock response
     $mockFastCgiRequest = $this->getMock('Crunch\\FastCGI\\Request', array(), array(), '', false);
     $mockFastCgiResponse = $this->getMock('Crunch\\FastCGI\\Response');
     // create a pair of sockets
     $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     // create the mock connection, pass the client socket to the constructor
     $mockFastCgiConnection = $this->getMock('Crunch\\FastCGI\\Connection', array('newRequest', 'request'), array(array_shift($sockets), '127.0.0.1', '9000'));
     $mockFastCgiConnection->expects($this->once())->method('newRequest')->will($this->returnValue($mockFastCgiRequest));
     $mockFastCgiConnection->expects($this->once())->method('request')->will($this->returnValue($mockFastCgiResponse));
     // create the mock client
     $mockFastCgiClient = $this->getMock('Crunch\\FastCGI\\Client', array('connect'), array(), '', false);
     $mockFastCgiClient->expects($this->once())->method('connect')->will($this->returnValue($mockFastCgiConnection));
     // create a mock version of the module
     $mockFastCgiModule = $this->getMock('AppserverIo\\WebServer\\Modules\\FastCgiModule', array('getFastCgiClient'));
     $mockFastCgiModule->expects($this->once())->method('getFastCgiClient')->will($this->returnValue($mockFastCgiClient));
     // prepare the array with the server vars
     $serverVars = array(array(ServerVars::SERVER_HANDLER, FastCgiModule::MODULE_NAME), array(ServerVars::REQUEST_METHOD, Protocol::METHOD_POST), array(ServerVars::SCRIPT_FILENAME, '/opt/appserver/webapps/test.php'), array(ServerVars::QUERY_STRING, 'test=test'), array(ServerVars::SCRIPT_NAME, '/index.php'), array(ServerVars::REQUEST_URI, '/test.php/test?test=test'), array(ServerVars::DOCUMENT_ROOT, '/opt/appserver/webapps'), array(ServerVars::SERVER_PROTOCOL, 'HTTP/1.1'), array(ServerVars::HTTPS, 'off'), array(ServerVars::SERVER_SOFTWARE, 'appserver'), array(ServerVars::REMOTE_ADDR, '127.0.0.1'), array(ServerVars::REMOTE_PORT, 63752), array(ServerVars::SERVER_ADDR, '127.0.0.1'), array(ServerVars::SERVER_PORT, 9080), array(ServerVars::SERVER_NAME, 'localhost'));
     // create a mock HTTP request instance
     $mockHttpRequest = $this->getMock('AppserverIo\\Http\\HttpRequest');
     $mockHttpRequest->expects($this->once())->method('getHeaders')->will($this->returnValue(array()));
     $mockHttpRequest->expects($this->once())->method('getBodyStream')->will($this->returnValue(fopen('php://memory', 'rw')));
     // create a mock HTTP request context instance
     $mockRequestContext = $this->getMockForAbstractClass('AppserverIo\\Server\\Interfaces\\RequestContextInterface');
     $mockRequestContext->expects($this->any())->method('hasServerVar')->will($this->returnValue(true));
     $mockRequestContext->expects($this->any())->method('hasHeader')->will($this->returnValue(false));
     $mockRequestContext->expects($this->any())->method('getServerVar')->will($this->returnValueMap($serverVars));
     $mockRequestContext->expects($this->once())->method('getEnvVars')->will($this->returnValue(array()));
     // create a mock HTTP response instance
     $mockHttpResponse = $this->getMock('AppserverIo\\Http\\HttpResponse');
     // process the FastCGI request
     $mockFastCgiModule->process($mockHttpRequest, $mockHttpResponse, $mockRequestContext, ModuleHooks::REQUEST_POST);
 }

 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);
 }

 /**
  * Opens up a new socket pair
  *
  * @param bool $blocking
  */
 public function __construct($blocking = false)
 {
     $this->socket = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     foreach ([self::READ, self::WRITE] as $socket) {
         stream_set_blocking($this->socket[$socket], $blocking);
     }
 }

 public function testSave()
 {
     /* This is expected to fail */
     $drv = new JSONDriver("/path/to/file/that/doesnt/exist");
     $drv->setTable('foo', 'bar', new Table());
     $this->assertFalse(is_file("/path/to/file/that/doesnt/exist"));
     /* Start tests that try to write to a real (temp) file. */
     $testFile = tempnam(sys_get_temp_dir(), "JSDTest");
     $drv = new JSONDriver($testFile);
     /* Set data to something that will cause json_encode to fail. */
     $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     $drv->setData($sockets);
     $drv->setTable('foo', 'bar', new Table());
     $this->assertEmpty(file_get_contents($testFile));
     $drv->setData(null);
     /* Now write something */
     $tbl = new Table("tbl", [new Column("col", new Integer(4))]);
     $drv->setTable('foo', 'bar', $tbl);
     $this->assertEquals(['foo'], $drv->getDatabaseNames(), 'Adding table should add database');
     $this->assertEquals(['bar'], $drv->getSchemaNames('foo'), 'Adding table should add schema');
     $this->assertEquals(['tbl'], $drv->getTableNames('foo', 'bar'), 'Adding table should index table');
     $this->assertFalse($drv->getSchemaNames('baz'), 'Non-existant db must cause failure');
     $this->assertFalse($drv->getTableNames('baz', 'boz'), 'Non-existant schema must cause failure');
     $data = json_decode(file_get_contents($testFile));
     $this->assertTrue(isset($data->foo->bar->tbl));
 }

 public function create()
 {
     $this->sockets = @stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     if (false === $this->sockets) {
         throw new ConnectionException('Could not create socket pair');
     }
 }

 /**
  * Add a new tunnel.
  * Calls from parent process only.
  *
  * @throws \RuntimeException If could not create a new pair socket
  */
 public function add()
 {
     list($this->parentSocket, $this->childSocket) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     if ($this->parentSocket === null || $this->childSocket === null) {
         throw new \RuntimeException(sprintf('Could not create a new pair socket: %s', socket_strerror(socket_last_error())));
     }
 }

 /**
  * @dataProvider loopProvider
  */
 public function testBufferReadsLargeChunks($condition, $loopFactory)
 {
     if (true !== $condition()) {
         return $this->markTestSkipped('Loop implementation not available');
     }
     $loop = $loopFactory();
     list($sockA, $sockB) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
     $streamA = new Stream($sockA, $loop);
     $streamB = new Stream($sockB, $loop);
     $bufferSize = 4096;
     $streamA->bufferSize = $bufferSize;
     $streamB->bufferSize = $bufferSize;
     $testString = str_repeat("*", $streamA->bufferSize + 1);
     $buffer = "";
     $streamB->on('data', function ($data, $streamB) use(&$buffer, &$testString) {
         $buffer .= $data;
     });
     $streamA->write($testString);
     $loop->tick();
     $loop->tick();
     $loop->tick();
     $streamA->close();
     $streamB->close();
     $this->assertEquals($testString, $buffer);
 }

 /**
  * Test signal interrupt when a stream is attached to the loop
  * @dataProvider signalProvider
  */
 public function testSignalInterruptWithStream($sigName, $signal)
 {
     if (!extension_loaded('pcntl')) {
         $this->markTestSkipped('"pcntl" extension is required to run this test.');
     }
     // dispatch signal handler every 10ms
     $this->loop->addPeriodicTimer(0.01, function () {
         pcntl_signal_dispatch();
     });
     // add stream to the loop
     list($writeStream, $readStream) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     $this->loop->addReadStream($readStream, function ($stream, $loop) {
         /** @var $loop LoopInterface */
         $read = fgets($stream);
         if ($read === "end loop\n") {
             $loop->stop();
         }
     });
     $this->loop->addTimer(0.05, function () use($writeStream) {
         fwrite($writeStream, "end loop\n");
     });
     $this->setUpSignalHandler($signal);
     // spawn external process to send signal to current process id
     $this->forkSendSignal($signal);
     $this->loop->run();
     $this->assertTrue($this->_signalHandled);
 }

 protected function spawnWorkers()
 {
     $master = null;
     $workers = array();
     for ($i = 0; $i < $this->config['master']['workers']; $i++) {
         //создаём парные сокеты, через них будут связываться мастер и воркер
         $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
         $pid = pcntl_fork();
         //создаём форк
         if ($pid == -1) {
             die("error: pcntl_fork\r\n");
         } elseif ($pid) {
             //мастер
             fclose($pair[0]);
             $workers[intval($pair[1])] = $pair[1];
             //один из пары будет в мастере
         } else {
             //воркер
             fclose($pair[1]);
             $master = $pair[0];
             //второй в воркере
             break;
         }
     }
     return array($pid, $master, $workers);
 }

 protected function spawnWorkers()
 {
     $master = null;
     $workers = array();
     $i = 0;
     while ($i < $this->config['workers']) {
         $i++;
         //socket pair
         $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
         $pid = pcntl_fork();
         //fork
         if ($pid == -1) {
             die("error: pcntl_fork\r\n");
         } elseif ($pid) {
             //master
             fclose($pair[0]);
             $workers[$pid] = $pair[1];
             //pair in master
         } else {
             //worker
             fclose($pair[1]);
             $master = $pair[0];
             //pair in worker
             break;
         }
     }
     return array($pid, $master, $workers);
 }

 /**
  * 构造函数 创建一个管道，避免select空fd
  * @return void
  */
 public function __construct()
 {
     $this->channel = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     if ($this->channel) {
         stream_set_blocking($this->channel[0], 0);
         $this->readFds[0] = $this->channel[0];
     }
 }

 /**
  * @param callable $func filter proc
  * @param callable $ctor constructor
  * @param callable $dtor destructor
  */
 function __construct(callable $func, callable $ctor = null, callable $dtor = null)
 {
     /*
      * We don't have pipe(2) support, so we'll use socketpair(2) instead.
      */
     list($this->input, $this->output) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     stream_filter_append($this->input, "atick\\IO\\StreamFilter", STREAM_FILTER_WRITE, compact("func", "ctor", "dtor"));
     stream_set_blocking($this->output, false);
 }

 /**
  * @covers React\Stream\Buffer::write
  * @covers React\Stream\Buffer::handleWrite
  */
 public function testWriteDetectsWhenOtherSideIsClosed()
 {
     list($a, $b) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     $loop = $this->createWriteableLoopMock();
     $buffer = new Buffer($a, $loop);
     $buffer->softLimit = 4;
     $buffer->on('error', $this->expectCallableOnce());
     fclose($b);
     $buffer->write("foo");
 }

 /**
  * 构造函数
  * @return void
  */
 public function __construct()
 {
     // 创建一个管道，放入监听读的描述符集合中，避免空轮询
     $this->channel = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     if ($this->channel) {
         stream_set_blocking($this->channel[0], 0);
         $this->_readFds[0] = $this->channel[0];
     }
     // 初始化优先队列(最大堆)
     $this->_scheduler = new \SplPriorityQueue();
     $this->_scheduler->setExtractFlags(\SplPriorityQueue::EXTR_BOTH);
 }

 /**
  * Construct.
  * @return void
  */
 public function __construct()
 {
     // Create a pipeline and put into the collection of the read to read the descriptor to avoid empty polling.
     $this->channel = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     if ($this->channel) {
         stream_set_blocking($this->channel[0], 0);
         $this->_readFds[0] = $this->channel[0];
     }
     // Init SplPriorityQueue.
     $this->_scheduler = new \SplPriorityQueue();
     $this->_scheduler->setExtractFlags(\SplPriorityQueue::EXTR_BOTH);
 }

 function process(Queue $queue)
 {
     if (file_exists($this->pidFile)) {
         $serverPid = trim(file_get_contents($this->pidFile));
         # When the process is not running anymore and a PID file exists
         # than the PID file was not correctly unlinked on the last shutdown.
         if (!posix_kill($serverPid, 0)) {
             unlink($this->pidFile);
         } else {
             throw new UnexpectedValueException("Server is already running with PID {$serverPid}.");
         }
     }
     if (false === @file_put_contents($this->pidFile, posix_getpid())) {
         throw new UnexpectedValueException(sprintf('Could not create %s. Maybe you have no permission to write it.', $this->pidFile));
     }
     $this->selfPipe = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     $this->queue = $queue;
     $queue->process($this);
     # Spawn up the initial worker pool.
     $this->spawnWorkers();
     register_shutdown_function(array($this, "stopListening"));
     pcntl_signal(SIGTTIN, array($this, "incrementWorkerCount"));
     pcntl_signal(SIGTTOU, array($this, "decrementWorkerCount"));
     pcntl_signal(SIGQUIT, array($this, "shutdown"));
     pcntl_signal(SIGINT, function () {
         exit;
     });
     # Monitor the child processes.
     for (;;) {
         pcntl_signal_dispatch();
         if ($this->shutdown) {
             foreach ($this->workers as $pid => $info) {
                 posix_kill($pid, SIGQUIT);
                 pcntl_waitpid($pid, $procStatus);
             }
             $this->stopListening();
             return;
         }
         $read = array($this->selfPipe[1]);
         $write = null;
         $exception = null;
         $readySize = @stream_select($read, $write, $exception, 10);
         # Handle the heartbeat sent by a worker.
         if ($readySize > 0 and $read) {
             $childPid = trim(fgets($read[0]));
             $this->workers[$childPid]["heartbeat"] = time();
         }
         $this->removeStaleWorkers();
         # Maintain a stable worker count. Compares the actual worker count
         # to the configured worker count and spawns workers when necessary.
         $this->spawnWorkers();
     }
 }

 /**
  * Creates a pair of socket channels that are connected to each other. This
  * is mostly useful for writing unit tests of, e.g., protocol channels.
  *
  *   list($x, $y) = PhutilSocketChannel::newChannelPair();
  *
  * @task construct
  */
 public static function newChannelPair()
 {
     $sockets = null;
     $domain = phutil_is_windows() ? STREAM_PF_INET : STREAM_PF_UNIX;
     $pair = stream_socket_pair($domain, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
     if (!$pair) {
         throw new Exception("stream_socket_pair() failed!");
     }
     $x = new PhutilSocketChannel($pair[0]);
     $y = new PhutilSocketChannel($pair[1]);
     return array($x, $y);
 }

 public static function createChildren($children, callable $callback = null)
 {
     if (!function_exists('pcntl_fork')) {
         throw new \Exception('Cannot spawn children; pcntl_fork method is required.');
     }
     if (is_array($children)) {
         $childCount = count($children);
     } else {
         $childCount = (int) $children;
     }
     $parent = null;
     for ($x = 0; $x < $childCount; $x++) {
         // Create a socket pair so parent and child can communicate
         list($parentSocket, $childSocket) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
         $pid = pcntl_fork();
         $key = isset($children[$x]) ? $children[$x] : $x;
         switch ($pid) {
             case -1:
                 throw new \Exception('Unable to fork');
             case 0:
                 // Child doesn't need parentSocket
                 fclose($parentSocket);
                 // Enable non-blocking mode
                 stream_set_blocking($childSocket, 0);
                 // Create the child process object and send in the startup parameters
                 $child = new ChildProcess($key, $childSocket);
                 // Two modes of operation here - callback or freeflow
                 if ($callback) {
                     // Call the user defined function and send in the process object
                     call_user_func($callback, $child);
                     // Kill the child process
                     $child->shutdown();
                 } else {
                     // Send the child back, calling script must shut down manually
                     return $child;
                 }
             default:
                 // Parent doesn't need childSocket
                 fclose($childSocket);
                 // Enable non-blocking mode
                 stream_set_blocking($parentSocket, 0);
                 // Initialized here to prevent children from having an instance
                 if (!$parent instanceof ParentProcess) {
                     $parent = new ParentProcess();
                 }
                 $parent->addChild(new Child($key, $pid, $parentSocket));
                 break;
         }
     }
     return $parent;
 }

 /**
  * @throws InstantiationException
  */
 public function __construct()
 {
     if (($sockets = @stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP)) === false) {
         throw new InstantiationException('stream_socket_pair() could not establish connection.');
     }
     $pid = pcntl_fork();
     if ($pid == -1) {
         throw new InstantiationException('pcntl_fork() could not create subprocess.');
     } else {
         if ($pid) {
             $this->createParent($pid, $sockets);
         } else {
             $this->createChild($pid, $sockets);
         }
     }
 }