Attempt direct read on ResourceInputStream::read()

This change should resolve the low performance on Windows' STDIN. STDIN
on Windows is a file handle and Windows has specialized support built
into it's stream_select implementation, but a stream_select with a file
handle takes 100ms, which results in slow performance if we read in
smaller chunks and always have to wait 100ms between these.

Fixes amphp/socket#52.

lib/ResourceInputStream.php

@@ -25,17 +25,26 @@ final class ResourceInputStream implements InputStream {
     /** @var bool */
     private $readable = true;
 
+    /** @var int */
+    private $chunkSize;
+
+    /** @var bool */
+    private $useSingleRead;
+
     /**
      * @param resource $stream Stream resource.
      * @param int $chunkSize Chunk size per read operation.
+     *
+     * @throws \Error If an invalid stream or parameter has been passed.
      */
     public function __construct($stream, int $chunkSize = self::DEFAULT_CHUNK_SIZE) {
         if (!\is_resource($stream) || \get_resource_type($stream) !== 'stream') {
             throw new \Error("Expected a valid stream");
         }
 
         $meta = \stream_get_meta_data($stream);
-        $useFread = $meta["stream_type"] === "udp_socket" || $meta["stream_type"] === "STDIO";
+        $useSingleRead = $meta["stream_type"] === "udp_socket" || $meta["stream_type"] === "STDIO";
+        $this->useSingleRead = $useSingleRead;
 
         if (\strpos($meta["mode"], "r") === false && \strpos($meta["mode"], "+") === false) {
             throw new \Error("Expected a readable stream");
@@ -45,12 +54,15 @@ public function __construct($stream, int $chunkSize = self::DEFAULT_CHUNK_SIZE)
         \stream_set_read_buffer($stream, 0);
 
         $this->resource = $stream;
+        $this->chunkSize = $chunkSize;
 
         $deferred = &$this->deferred;
         $readable = &$this->readable;
 
-        $this->watcher = Loop::onReadable($this->resource, static function ($watcher, $stream) use (&$deferred, &$readable, $chunkSize, $useFread) {
-            if ($useFread) {
+        $this->watcher = Loop::onReadable($this->resource, static function ($watcher, $stream) use (
+            &$deferred, &$readable, $chunkSize, $useSingleRead
+        ) {
+            if ($useSingleRead) {
                 $data = @\fread($stream, $chunkSize);
             } else {
                 $data = @\stream_get_contents($stream, $chunkSize);
@@ -89,10 +101,32 @@ public function read(): Promise {
             return new Success; // Resolve with null on closed stream.
         }
 
-        $this->deferred = new Deferred;
-        Loop::enable($this->watcher);
+        // Attempt a direct read, because Windows suffers from slow I/O on STDIN otherwise.
+        if ($this->useSingleRead) {
+            $data = @\fread($this->resource, $this->chunkSize);
+        } else {
+            $data = @\stream_get_contents($this->resource, $this->chunkSize);
+        }
+
+        \assert($data !== false, "Trying to read from a previously fclose()'d resource. Do NOT manually fclose() resources the loop still has a reference to.");
+
+        if ($data === '') {
+            // Error suppression, because pthreads does crazy things with resources,
+            // which might be closed during two operations.
+            // See https://github.com/amphp/byte-stream/issues/32
+            if (@\feof($this->resource)) {
+                $this->readable = false;
+                Loop::cancel($this->watcher);
+                $data = null; // Stream closed, resolve read with null.
+            } else {
+                $this->deferred = new Deferred;
+                Loop::enable($this->watcher);
+
+                return $this->deferred->promise();
+            }
+        }
 
-        return $this->deferred->promise();
+        return new Success($data);
     }
 
     /**
@@ -125,7 +159,7 @@ private function free() {
         if ($this->deferred !== null) {
             $deferred = $this->deferred;
             $this->deferred = null;
-            $deferred->resolve(null);
+            $deferred->resolve();
         }
 
         Loop::cancel($this->watcher);