Improve CPU utilization (~74% -> ~51%) when writing to socket is faster

This improves CPU profile by skipping `read()` calls in Responder
when there is no new data in buffer.

Was tested with stdin `head -c 1G </dev/urandom | pv -q -L 10000000`
which reads 1G file with 10M/s rate limit (1 connection), then exits

Before:

```
     141366.582835      task-clock (msec)     #    1.318 CPUs utilized
         2,830,199      context-switches      #    0.020 M/sec
            50,323      cpu-migrations        #    0.356 K/sec
             5,363      page-faults           #    0.038 K/sec
   476,500,948,484      cycles                #    3.371 GHz
   340,863,872,685      instructions          #    0.72  insn per cycle
    72,803,147,450      branches              #  514.995 M/sec
     2,394,570,464      branch-misses         #    3.29% of all branches

     107.284806111 seconds time elapsed
```

After:
```
      83205.359732      task-clock (msec)     #    0.776 CPUs utilized
           279,904      context-switches      #    0.003 M/sec
            15,297      cpu-migrations        #    0.184 K/sec
             5,318      page-faults           #    0.064 K/sec
    79,455,374,201      cycles                #    0.955 GHz
    44,104,516,695      instructions          #    0.56  insn per cycle
     9,306,984,820      branches              #  111.856 M/sec
       452,310,830      branch-misses         #    4.86% of all branches

     107.261390647 seconds time elapsed
```

Co-authored-by: Aaron Piotrowski <aaron@trowski.com>

src/Bufferer/BuffererInterface.php

@@ -9,6 +9,7 @@ interface BuffererInterface
     public function __invoke();
     public function getFilePath(): string;
     public function isBuffering(): bool;
+    public function waitForWrite(): Promise;
     public function getMimeType(): Promise;
     public function getCurrentProgress(): int;
 }

src/Bufferer/PipeBufferer.php

@@ -2,9 +2,9 @@
 
 namespace Ostrolucky\Stdinho\Bufferer;
 
-use Amp\ByteStream\InputStream;
 use Amp\ByteStream\ResourceInputStream;
 use Amp\ByteStream\ResourceOutputStream;
+use Amp\Deferred;
 use Amp\Promise;
 use Ostrolucky\Stdinho\ProgressBar;
 use Psr\Log\LoggerInterface;
@@ -21,6 +21,10 @@ class PipeBufferer implements BuffererInterface
     private $progressBar;
 
     private $buffering = true;
+    /**
+     * @var Deferred|null
+     */
+    private $deferred;
 
     /**
      * @param resource $inputStream
@@ -33,7 +37,7 @@ public function __construct(
     ) {
         $this->logger = $logger;
         $this->inputStream = new ResourceInputStream($inputStream);
-        $this->outputStream = new ResourceOutputStream($fileOutput = $outputPath ? fopen($outputPath, 'w') : tmpfile());
+        $this->outputStream = new ResourceOutputStream($fileOutput = $outputPath ? fopen($outputPath, 'wb') : tmpfile());
         $this->mimeType = new \Amp\Deferred;
         $this->filePath = $outputPath ?: stream_get_meta_data($fileOutput)['uri'];
         $this->progressBar = new ProgressBar($output, 0, 'buffer');
@@ -46,6 +50,7 @@ public function __invoke(): \Generator
         $bytesDownloaded = 0;
         while (null !== $chunk = yield $this->inputStream->read()) {
             yield $this->outputStream->write($chunk);
+            $this->resolveDeferrer();
 
             if ($bytesDownloaded === 0) {
                 $mimeType = (new \finfo(FILEINFO_MIME))->buffer($chunk);
@@ -59,6 +64,7 @@ public function __invoke(): \Generator
         $this->buffering = false;
         $this->progressBar->finish();
         $this->logger->debug("Stdin transfer done, $bytesDownloaded bytes downloaded");
+        $this->resolveDeferrer();
     }
 
     public function getFilePath(): string
@@ -76,8 +82,24 @@ public function isBuffering(): bool
         return $this->buffering;
     }
 
+    public function waitForWrite(): Promise
+    {
+        return ($this->deferred = $this->deferred ?: new Deferred())->promise();
+    }
+
     public function getCurrentProgress(): int
     {
         return $this->progressBar->step;
     }
+
+    private function resolveDeferrer(): void
+    {
+        if (!$this->deferred) {
+            return;
+        }
+
+        $deferrer = $this->deferred;
+        $this->deferred = null;
+        $deferrer->resolve();
+    }
 }

src/Bufferer/ResolvedBufferer.php

@@ -31,6 +31,11 @@ public function isBuffering(): bool
         return false;
     }
 
+    public function waitForWrite(): Promise
+    {
+        return new Success();
+    }
+
     public function getMimeType(): Promise
     {
         return new Success($this->mimeType);

src/Responder.php

@@ -22,13 +22,13 @@ public function __construct(LoggerInterface $logger, BuffererInterface $bufferer
         $this->consoleOutput = $consoleOutput;
     }
 
-    public function __invoke(Socket $socket)
+    public function __invoke(Socket $socket): \Generator
     {
         $remoteAddress = $socket->getRemoteAddress();
         $this->logger->debug("Accepted connection from $remoteAddress:\n" . trim(yield $socket->read()));
 
         /** @var Handle $handle */
-        $handle = yield \Amp\File\open($this->bufferer->getFilePath(), 'r');
+        $handle = yield \Amp\File\open($this->bufferer->getFilePath(), 'rb');
 
         $header = "HTTP/1.1 200 OK\nContent-Type: " . yield $this->bufferer->getMimeType();
         if (!$this->bufferer->isBuffering()) {
@@ -48,6 +48,7 @@ public function __invoke(Socket $socket)
             while (($chunk = yield $handle->read()) || $this->bufferer->isBuffering()) {
                 // we reached end of the buffer, but it's still buffering
                 if ($chunk === null) {
+                    yield $this->bufferer->waitForWrite();
                     continue;
                 }