protected function connectToAddress(string $address, float $timeout, bool $encrypt)
 {
     $url = \sprintf('%s://%s', $this->protocol, $address);
     $errno = null;
     $errstr = null;
     $context = \stream_context_create(\array_replace_recursive($this->options, ['socket' => ['connect' => $address]]));
     $socket = @\stream_socket_client($url, $errno, $errstr, $timeout ?: null, \STREAM_CLIENT_CONNECT | \STREAM_CLIENT_ASYNC_CONNECT, $context);
     if (!\is_resource($socket)) {
         throw new SocketException(\sprintf('Could not connect to "%s" (%s): [%s] %s', $url, $this->peer, $errno, \trim($errstr)));
     }
     try {
         \stream_set_blocking($socket, false);
         \stream_set_read_buffer($socket, 0);
         \stream_set_write_buffer($socket, 0);
         if ($this->tcpNoDelay) {
             Socket::setTcpNoDelay($socket, true);
         }
         if ($encrypt) {
             yield from $this->encryptSocketClient($socket, $timeout);
         } else {
             (yield new AwaitWrite($socket, $timeout));
         }
         return $socket;
     } catch (\Throwable $e) {
         Socket::shutdown($socket);
         throw $e;
     }
 }

 public function __construct($stream)
 {
     parent::__construct($stream);
     stream_set_blocking($stream, 0);
     stream_set_write_buffer($stream, 0);
     $this->read = self::reader();
     $this->write = self::writer();
 }

 /**
  * Create a new pipe-based log handler.
  * 
  * @param string $pipe Output pipe, defaults to STDERR.
  */
 public function __construct($pipe = 'php://stderr', int $threshold = Logger::ALL, array $origins = [])
 {
     $pipe = \fopen($pipe, 'wb');
     \stream_set_blocking($pipe, false);
     \stream_set_write_buffer($pipe, 0);
     $this->writer = new SocketWriter($pipe);
     $this->threshold = $threshold;
     $this->origins = $origins;
 }

 public function _connect()
 {
     $this->_socket = @pfsockopen($this->_host, $this->_port, $errno, $errstr, $this->_timeout);
     if ($this->_socket === false) {
         return new PHPTCP_Error($errno, $errstr);
     }
     stream_set_write_buffer($this->_socket, 0);
     socket_set_timeout($this->_socket, $this->_timeout);
 }

 /**
  * Connect to Host
  *
  * @param  string         $host
  * @param  array          $ssl
  * @return AbstractClient
  */
 protected function connect($host, array $ssl)
 {
     $this->socket = stream_socket_client($host, $errno, $errstr, ini_get('default_socket_timeout'), STREAM_CLIENT_CONNECT, stream_context_create(array('ssl' => $ssl)));
     if (!$this->socket) {
         throw new Exception\RuntimeException(sprintf('Unable to connect: %s: %d (%s)', $host, $errno, $errstr));
     }
     stream_set_blocking($this->socket, 0);
     stream_set_write_buffer($this->socket, 0);
     return $this;
 }

 /**
  * @param resource $resource Stream resource.
  * @param bool $autoClose True to close the resource on destruct, false to leave it open.
  */
 public function __construct($resource, bool $autoClose = true)
 {
     parent::__construct($resource, $autoClose);
     stream_set_write_buffer($resource, 0);
     stream_set_chunk_size($resource, self::CHUNK_SIZE);
     $this->writeQueue = new \SplQueue();
     $this->onCancelled = function (Throwable $exception) {
         $this->free($exception);
     };
 }

 /**
  * add a new request to the pool.
  */
 public function add(Resource $r)
 {
     $stream = $r->stream;
     $id = (int) $stream;
     if (isset($this->streams[$id])) {
         return $this->streams[$id];
     }
     stream_set_blocking($stream, 0);
     stream_set_write_buffer($stream, 0);
     return $this->streams[$id] = $r;
 }

 protected function _connect()
 {
     $sURL = $this->asurls[$this->_environment];
     $streamContext = stream_context_create(array('ssl' => array('verify_peer' => isset($this->_rootCertificationAuthorityFile), 'cafile' => $this->_rootCertificationAuthorityFile, 'local_cert' => $this->_providerCertificateFile)));
     $this->_hSocket = @stream_socket_client($sURL, $nError, $sError, $this->_connectTimeout, STREAM_CLIENT_CONNECT, $streamContext);
     if (!$this->_hSocket) {
         throw new Exception("Unable to connect to '{$sURL}': {$sError} ({$nError})");
     }
     stream_set_blocking($this->_hSocket, 0);
     stream_set_write_buffer($this->_hSocket, 0);
     return true;
 }

 /**
  * @param resource $resource
  * @param LoopInterface $loop
  * @param bool $autoClose
  * @throws InvalidArgumentException
  */
 public function __construct($resource, LoopInterface $loop, $autoClose = true)
 {
     parent::__construct($resource, $autoClose);
     if (function_exists('stream_set_write_buffer')) {
         stream_set_write_buffer($this->resource, 0);
     }
     $this->loop = $loop;
     $this->listening = false;
     $this->paused = true;
     $this->buffer = new Buffer();
     $this->resume();
 }

 public function record($level, $logStr)
 {
     $timeStr = '[' . date('Y-m-d H:i:s O') . ']';
     $fp = fopen($this->_filename, 'ab');
     if (false !== $fp) {
         stream_set_write_buffer($fp, $this->_write_buff);
         fwrite($fp, $timeStr . ' ' . $logStr . PHP_EOL);
         fclose($fp);
     } else {
         return false;
     }
     return true;
 }

 /**
  * Connect to Kafka via socket
  * 
  * @return void
  */
 public function connect()
 {
     if (!is_resource($this->conn)) {
         $this->conn = stream_socket_client('tcp://' . $this->host . ':' . $this->port, $errno, $errstr);
         if (!$this->conn) {
             throw new RuntimeException($errstr, $errno);
         }
         stream_set_timeout($this->conn, $this->socketTimeout);
         stream_set_read_buffer($this->conn, $this->socketBufferSize);
         stream_set_write_buffer($this->conn, $this->socketBufferSize);
         //echo "\nConnected to ".$this->host.":".$this->port."\n";
     }
 }

 public function stream_set_option($option, $arg1, $arg2)
 {
     switch ($option) {
         case STREAM_OPTION_BLOCKING:
             stream_set_blocking($this->fileSource, $arg1);
             break;
         case STREAM_OPTION_READ_TIMEOUT:
             stream_set_timeout($this->fileSource, $arg1, $arg2);
             break;
         case STREAM_OPTION_WRITE_BUFFER:
             stream_set_write_buffer($this->fileSource, $arg1, $arg2);
     }
 }

 public function run()
 {
     if (!($ipcSocket = @stream_socket_client($this->ipcUri, $errno, $errstr, 5))) {
         throw new \RuntimeException(sprintf("Failed connecting to IPC server: (%d) %s", $errno, $errstr));
     }
     stream_set_write_buffer($ipcSocket, 0);
     stream_socket_shutdown($ipcSocket, STREAM_SHUT_RD);
     $openMsg = $this->getThreadId() . "\n";
     if (@fwrite($ipcSocket, $openMsg) !== strlen($openMsg)) {
         throw new \RuntimeException("Failed writing open message to IPC server");
     }
     $this->ipcSocket = $ipcSocket;
 }

 private function init()
 {
     $connection = @stream_socket_client('tcp://' . $this->host . ':' . $this->port);
     stream_set_write_buffer($connection, 0);
     stream_set_read_buffer($connection, 0);
     if (!$connection) {
         throw new \Exception('No SyncDaemon available.');
     }
     $data = trim(fgets($connection));
     if ($data != self::ACCEPT) {
         throw new \Exception('Connection faild. Wrong answer: ' . $data);
     }
     $this->connection = $connection;
 }

 /**
  * {@inheritdoc}
  */
 public function connect()
 {
     if ($this->sock) {
         return;
     }
     $this->sock = stream_socket_client($this->uri, $errno, $errstr, $this->timeout);
     if (!$this->sock) {
         throw new \RuntimeException("Can't connect to the server at '{$this->uri}' ({$errno}: {$errstr})");
     }
     stream_set_timeout($this->sock, $this->timeout);
     stream_set_blocking($this->sock, $this->blocking);
     stream_set_read_buffer($this->sock, 0);
     // 64k
     stream_set_write_buffer($this->sock, 0);
 }

 public function __construct($host, $port, $connection_timeout, $read_write_timeout = null, $context = null, $blocking = false)
 {
     $errstr = $errno = null;
     $this->sock = null;
     if ($context) {
         $remote = sprintf('tls://%s:%s', $host, $port);
         $this->sock = @stream_socket_client($remote, $errno, $errstr, $connection_timeout, STREAM_CLIENT_CONNECT, $context);
     } else {
         $remote = sprintf('tcp://%s:%s', $host, $port);
         $this->sock = @stream_socket_client($remote, $errno, $errstr, $connection_timeout, STREAM_CLIENT_CONNECT);
     }
     if (!$this->sock) {
         throw new RuntimeException("Error Connecting to server({$errno}): {$errstr} ");
     }
     if (null !== $read_write_timeout) {
         if (!stream_set_timeout($this->sock, $read_write_timeout)) {
             throw new \Exception("Timeout (stream_set_timeout) could not be set");
         }
     }
     /**
      * Manually set blocking & write buffer settings and make sure they are successfuly set
      * Use non-blocking as we dont want to stuck waiting for socket data while fread() w/o timeout
      */
     if (!stream_set_blocking($this->sock, $blocking)) {
         throw new \Exception("Blocking could not be set");
     }
     $rbuff = stream_set_read_buffer($this->sock, 0);
     if (!(0 === $rbuff)) {
         throw new \Exception("Read buffer could not be set");
     }
     /**
      * ! this is not reliably returns success (0)
      * ! but default buffer is pretty high (few Kb), so will not affect sending single small pushes
      *
      * Output using fwrite() is normally buffered at 8K.
      * This means that if there are two processes wanting to write to the same output stream (a file),
      * each is paused after 8K of data to allow the other to write.
      *
      * Ensures that all writes with fwrite() are completed
      * before other processes are allowed to write to that output stream
      */
     stream_set_write_buffer($this->sock, 0);
     /**
      * Set small chunk size (default=4096/8192)
      * Setting this to small values (100bytes) still does NOT help detecting feof()
      */
     stream_set_chunk_size($this->sock, 1024);
 }

 /**
  * Open a log file.
  *
  * @param string $date  The date for the log file.
  */
 private function openLog($date)
 {
     assert('is_string($date)');
     if ($this->file !== NULL && $this->file !== FALSE) {
         fclose($this->file);
         $this->file = NULL;
     }
     $fileName = $this->logDir . '/' . $date . '.log';
     $this->file = @fopen($fileName, 'a');
     if ($this->file === FALSE) {
         throw new SimpleSAML_Error_Exception('Error opening log file: ' . var_export($fileName, TRUE));
     }
     // Disable output buffering
     stream_set_write_buffer($this->file, 0);
     $this->fileDate = $date;
 }

 public static function connect(string $url, $workerId) : IpcClient
 {
     $errno = null;
     $errstr = null;
     $socket = @\stream_socket_client($url, $errno, $errstr, 5);
     if ($socket === false) {
         throw new \RuntimeException(\sprintf('Failed to connecto to IPC server "%s": [%s] %s', $url, $errno, $errstr));
     }
     \stream_set_read_buffer($socket, 0);
     \stream_set_write_buffer($socket, 0);
     $socket = new SocketStream($socket);
     $transmitter = new SocketTransmitter($socket);
     // Perform a blocking handshake.
     $transmitter->sendSync($workerId, SocketTransmitter::TYPE_HANDSHAKE);
     return new IpcClient($workerId, $socket, $transmitter);
 }

 function puts($file, $string, $mode = 'w')
 {
     $pointer = @fopen($file, $mode);
     if (!$pointer) {
         $pointer = fopen($file, 'w');
     }
     if ($pointer) {
         stream_set_write_buffer($pointer, 0);
         if (flock($pointer, LOCK_EX)) {
             rewind($pointer);
             fputs($pointer, $string);
             flock($pointer, LOCK_UN);
         }
         fclose($pointer);
     }
 }

 public function accept()
 {
     $client_socket = \stream_socket_accept($this->server_sock);
     $client_socket_id = (int) $client_socket;
     \stream_set_blocking($client_socket, 0);
     $this->client_sock[$client_socket_id] = $client_socket;
     $this->client_num++;
     if ($this->client_num > $this->max_connect) {
         $this->_closeSocket($client_socket);
         return false;
     } else {
         //设置写缓冲区
         \stream_set_write_buffer($client_socket, $this->write_buffer_size);
         return $client_socket_id;
     }
 }