[HttpClient] fix monitoring timeouts when other streams are active

Author: nicolas-grekas

Internal/NativeClientState.php

@@ -28,8 +28,6 @@ final class NativeClientState extends ClientState
     public $responseCount = 0;
     /** @var string[] */
     public $dnsCache = [];
-    /** @var resource[] */
-    public $handles = [];
     /** @var bool */
     public $sleep = false;
 

Response/NativeResponse.php

@@ -220,11 +220,6 @@ private static function schedule(self $response, array &$runningResponses): void
      */
     private static function perform(ClientState $multi, array &$responses = null): void
     {
-        // List of native handles for stream_select()
-        if (null !== $responses) {
-            $multi->handles = [];
-        }
-
         foreach ($multi->openHandles as $i => [$h, $buffer, $onProgress]) {
             $hasActivity = false;
             $remaining = &$multi->openHandles[$i][3];
@@ -291,8 +286,6 @@ private static function perform(ClientState $multi, array &$responses = null): v
                 $multi->handlesActivity[$i][] = $e;
                 unset($multi->openHandles[$i]);
                 $multi->sleep = false;
-            } elseif (null !== $responses) {
-                $multi->handles[] = $h;
             }
         }
 
@@ -307,7 +300,7 @@ private static function perform(ClientState $multi, array &$responses = null): v
             }
         }
 
-        if (\count($multi->handles) >= $multi->maxHostConnections) {
+        if (\count($multi->openHandles) >= $multi->maxHostConnections) {
             return;
         }
 
@@ -318,10 +311,6 @@ private static function perform(ClientState $multi, array &$responses = null): v
                 $multi->sleep = false;
                 self::perform($multi);
 
-                if (null !== $response->handle) {
-                    $multi->handles[] = $response->handle;
-                }
-
                 break;
             }
         }
@@ -335,7 +324,8 @@ private static function perform(ClientState $multi, array &$responses = null): v
     private static function select(ClientState $multi, float $timeout): int
     {
         $_ = [];
+        $handles = array_column($multi->openHandles, 0);
 
-        return (!$multi->sleep = !$multi->sleep) ? -1 : stream_select($multi->handles, $_, $_, (int) $timeout, (int) (1E6 * ($timeout - (int) $timeout)));
+        return (!$multi->sleep = !$multi->sleep) ? -1 : stream_select($handles, $_, $_, (int) $timeout, (int) (1E6 * ($timeout - (int) $timeout)));
     }
 }

Response/ResponseTrait.php

@@ -316,7 +316,7 @@ public static function stream(iterable $responses, float $timeout = null): \Gene
         }
 
         $lastActivity = microtime(true);
-        $isTimeout = false;
+        $enlapsedTimeout = 0;
 
         while (true) {
             $hasActivity = false;
@@ -338,15 +338,15 @@ public static function stream(iterable $responses, float $timeout = null): \Gene
                     } elseif (!isset($multi->openHandles[$j])) {
                         unset($responses[$j]);
                         continue;
-                    } elseif ($isTimeout) {
+                    } elseif ($enlapsedTimeout >= $timeoutMax) {
                         $multi->handlesActivity[$j] = [new ErrorChunk($response->offset, sprintf('Idle timeout reached for "%s".', $response->getInfo('url')))];
                     } else {
                         continue;
                     }
 
                     while ($multi->handlesActivity[$j] ?? false) {
                         $hasActivity = true;
-                        $isTimeout = false;
+                        $enlapsedTimeout = 0;
 
                         if (\is_string($chunk = array_shift($multi->handlesActivity[$j]))) {
                             if (null !== $response->inflate && false === $chunk = @inflate_add($response->inflate, $chunk)) {
@@ -379,7 +379,7 @@ public static function stream(iterable $responses, float $timeout = null): \Gene
                             }
                         } elseif ($chunk instanceof ErrorChunk) {
                             unset($responses[$j]);
-                            $isTimeout = true;
+                            $enlapsedTimeout = $timeoutMax;
                         } elseif ($chunk instanceof FirstChunk) {
                             if ($response->logger) {
                                 $info = $response->getInfo();
@@ -447,10 +447,11 @@ public static function stream(iterable $responses, float $timeout = null): \Gene
                 continue;
             }
 
-            switch (self::select($multi, $timeoutMin)) {
-                case -1: usleep(min(500, 1E6 * $timeoutMin)); break;
-                case 0: $isTimeout = microtime(true) - $lastActivity > $timeoutMax; break;
+            if (-1 === self::select($multi, min($timeoutMin, $timeoutMax - $enlapsedTimeout))) {
+                usleep(min(500, 1E6 * $timeoutMin));
             }
+
+            $enlapsedTimeout = microtime(true) - $lastActivity;
         }
     }
 }

Tests/CurlHttpClientTest.php

@@ -112,6 +112,15 @@ public function testHttp2PushVulcainWithUnusedResponse()
         $this->assertSame($expected, $logger->logs);
     }
 
+    public function testTimeoutIsNotAFatalError()
+    {
+        if ('\\' === \DIRECTORY_SEPARATOR) {
+            $this->markTestSkipped('Too transient on Windows');
+        }
+
+        parent::testTimeoutIsNotAFatalError();
+    }
+
     private function getVulcainClient(): CurlHttpClient
     {
         if (\PHP_VERSION_ID >= 70300 && \PHP_VERSION_ID < 70304) {

Tests/HttpClientTestCase.php

@@ -72,16 +72,17 @@ public function testToStream404()
         $this->assertSame($response, stream_get_meta_data($stream)['wrapper_data']->getResponse());
         $this->assertSame(404, $response->getStatusCode());
 
-        $this->expectException(ClientException::class);
         $response = $client->request('GET', 'http://localhost:8057/404');
-        $stream = $response->toStream();
+        $this->expectException(ClientException::class);
+        $response->toStream();
     }
 
     public function testNonBlockingStream()
     {
         $client = $this->getHttpClient(__FUNCTION__);
         $response = $client->request('GET', 'http://localhost:8057/timeout-body');
         $stream = $response->toStream();
+        usleep(10000);
 
         $this->assertTrue(stream_set_blocking($stream, false));
         $this->assertSame('<1>', fread($stream, 8192));
@@ -99,6 +100,7 @@ public function testTimeoutIsNotAFatalError()
         $response = $client->request('GET', 'http://localhost:8057/timeout-body', [
             'timeout' => 0.25,
         ]);
+        $this->assertSame(200, $response->getStatusCode());
 
         try {
             $response->getContent();

Tests/MockHttpClientTest.php

@@ -69,6 +69,10 @@ protected function getHttpClient(string $testCase): HttpClientInterface
                 $this->markTestSkipped("MockHttpClient doesn't unzip");
                 break;
 
+            case 'testTimeoutWithActiveConcurrentStream':
+                $this->markTestSkipped('Real transport required');
+                break;
+
             case 'testDestruct':
                 $this->markTestSkipped("MockHttpClient doesn't timeout on destruct");
                 break;