Fix backpressure for HTTP/2 request bodies

amphp · Nov 29, 2019 · 1c59226 · 1c59226
1 parent e486025
commit 1c59226
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 6 deletions.
diff --git a/src/Connection/Internal/Http2Stream.php b/src/Connection/Internal/Http2Stream.php
@@ -71,6 +71,9 @@ final class Http2Stream
     /** @var Stream */
     public $stream;
 
+    /**@var Deferred|null */
+    public $windowSizeIncrease;
+
     public function __construct(
         int $id,
         Request $request,

diff --git a/src/Connection/Internal/InternalHttp2Connection.php b/src/Connection/Internal/InternalHttp2Connection.php
@@ -880,15 +880,16 @@ public function request(Request $request, CancellationToken $cancellationToken,
         return call(function () use ($streamId, $request, $cancellationToken, $stream, $http2stream): \Generator {
             $this->socket->reference();
 
-            $cancellationId = $cancellationToken->subscribe(function (CancelledException $exception) use ($streamId
-            ): void {
+            $onCancel = function (CancelledException $exception) use ($streamId): void {
                 if (!isset($this->streams[$streamId])) {
                     return;
                 }
 
                 $this->writeFrame(\pack("N", self::CANCEL), self::RST_STREAM, self::NO_FLAG, $streamId);
                 $this->releaseStream($streamId, $exception);
-            });
+            };
+
+            $cancellationId = $cancellationToken->subscribe($onCancel);
 
             try {
                 $headers = yield from $this->generateHeaders($request);
@@ -1121,6 +1122,12 @@ private function writeBufferedData(Http2Stream $stream): Promise
         $length = \strlen($stream->buffer);
 
         if ($length <= $windowSize) {
+            if ($stream->windowSizeIncrease) {
+                $deferred = $stream->windowSizeIncrease;
+                $stream->windowSizeIncrease = null;
+                $deferred->resolve();
+            }
+
             $this->clientWindow -= $length;
             $stream->clientWindow -= $length;
 
@@ -1141,12 +1148,17 @@ private function writeBufferedData(Http2Stream $stream): Promise
 
             $stream->buffer = "";
 
-            // TODO Read next request body chunk
-
             return $promise;
         }
 
         if ($windowSize > 0) {
+            // Read next body chunk if less than 8192 bytes will remain in the buffer
+            if ($length - 8192 < $windowSize && $stream->windowSizeIncrease) {
+                $deferred = $stream->windowSizeIncrease;
+                $stream->windowSizeIncrease = null;
+                $deferred->resolve();
+            }
+
             $data = $stream->buffer;
             $end = $windowSize - $this->frameSizeLimit;
 
@@ -1169,7 +1181,11 @@ private function writeBufferedData(Http2Stream $stream): Promise
             return $promise;
         }
 
-        return new Success;
+        if ($stream->windowSizeIncrease === null) {
+            $stream->windowSizeIncrease = new Deferred;
+        }
+
+        return $stream->windowSizeIncrease->promise();
     }
 
     private function releaseStream(int $streamId, ?\Throwable $exception = null): void