Skip to content

Commit ee3451b

Browse files
nicolas-grekasnicolas-grekas
committed
[HttpClient] Fix activity tracking leading to negative timeout errors
1 parent 25dd52e commit ee3451b

File tree

5 files changed

+33
-31
lines changed

5 files changed

+33
-31
lines changed

‎src/Symfony/Component/HttpClient/Response/AmpResponse.php‎

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -179,19 +179,17 @@ private static function schedule(self $response, array &$runningResponses): void
179179
/**
180180
* @param AmpClientState $multi
181181
*/
182-
private static function perform(ClientState $multi, ?array &$responses = null): void
182+
private static function perform(ClientState $multi, ?array $responses = null): void
183183
{
184-
if ($responses) {
185-
foreach ($responses as $response) {
186-
try {
187-
if ($response->info['start_time']) {
188-
$response->info['total_time'] = microtime(true) - $response->info['start_time'];
189-
($response->onProgress)();
190-
}
191-
} catch (\Throwable $e) {
192-
$multi->handlesActivity[$response->id][] = null;
193-
$multi->handlesActivity[$response->id][] = $e;
184+
foreach ($responses ?? [] as $response) {
185+
try {
186+
if ($response->info['start_time']) {
187+
$response->info['total_time'] = microtime(true) - $response->info['start_time'];
188+
($response->onProgress)();
194189
}
190+
} catch (\Throwable $e) {
191+
$multi->handlesActivity[$response->id][] = null;
192+
$multi->handlesActivity[$response->id][] = $e;
195193
}
196194
}
197195
}

‎src/Symfony/Component/HttpClient/Response/CurlResponse.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -265,11 +265,11 @@ private static function schedule(self $response, array &$runningResponses): void
265265
/**
266266
* @param CurlClientState $multi
267267
*/
268-
private static function perform(ClientState $multi, ?array &$responses = null): void
268+
private static function perform(ClientState $multi, ?array $responses = null): void
269269
{
270270
if ($multi->performing) {
271271
if ($responses) {
272-
$response = current($responses);
272+
$response = $responses[array_key_first($responses)];
273273
$multi->handlesActivity[(int) $response->handle][] = null;
274274
$multi->handlesActivity[(int) $response->handle][] = new TransportException(sprintf('Userland callback cannot use the client nor the response while processing "%s".', curl_getinfo($response->handle, \CURLINFO_EFFECTIVE_URL)));
275275
}

‎src/Symfony/Component/HttpClient/Response/MockResponse.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,7 @@ protected static function schedule(self $response, array &$runningResponses): vo
167167
$runningResponses[0][1][$response->id] = $response;
168168
}
169169

170-
protected static function perform(ClientState $multi, array &$responses): void
170+
protected static function perform(ClientState $multi, array $responses): void
171171
{
172172
foreach ($responses as $response) {
173173
$id = $response->id;

‎src/Symfony/Component/HttpClient/Response/NativeResponse.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -228,7 +228,7 @@ private static function schedule(self $response, array &$runningResponses): void
228228
/**
229229
* @param NativeClientState $multi
230230
*/
231-
private static function perform(ClientState $multi, ?array &$responses = null): void
231+
private static function perform(ClientState $multi, ?array $responses = null): void
232232
{
233233
foreach ($multi->openHandles as $i => [$pauseExpiry, $h, $buffer, $onProgress]) {
234234
if ($pauseExpiry) {

‎src/Symfony/Component/HttpClient/Response/TransportResponseTrait.php‎

Lines changed: 20 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ abstract protected static function schedule(self $response, array &$runningRespo
9292
/**
9393
* Performs all pending non-blocking operations.
9494
*/
95-
abstract protected static function perform(ClientState $multi, array &$responses): void;
95+
abstract protected static function perform(ClientState $multi, array $responses): void;
9696

9797
/**
9898
* Waits for network activity.
@@ -150,10 +150,15 @@ public static function stream(iterable $responses, ?float $timeout = null): \Gen
150150
$lastActivity = hrtime(true) / 1E9;
151151
$elapsedTimeout = 0;
152152

153-
if ($fromLastTimeout = 0.0 === $timeout && '-0' === (string) $timeout) {
154-
$timeout = null;
155-
} elseif ($fromLastTimeout = 0 > $timeout) {
156-
$timeout = -$timeout;
153+
if ((0.0 === $timeout && '-0' === (string) $timeout) || 0 > $timeout) {
154+
$timeout = $timeout ? -$timeout : null;
155+
156+
/** @var ClientState $multi */
157+
foreach ($runningResponses as [$multi]) {
158+
if (null !== $multi->lastTimeout) {
159+
$elapsedTimeout = max($elapsedTimeout, $lastActivity - $multi->lastTimeout);
160+
}
161+
}
157162
}
158163

159164
while (true) {
@@ -162,35 +167,33 @@ public static function stream(iterable $responses, ?float $timeout = null): \Gen
162167
$timeoutMin = $timeout ?? \INF;
163168

164169
/** @var ClientState $multi */
165-
foreach ($runningResponses as $i => [$multi]) {
166-
$responses = &$runningResponses[$i][1];
170+
foreach ($runningResponses as $i => [$multi, &$responses]) {
167171
self::perform($multi, $responses);
168172

169173
foreach ($responses as $j => $response) {
170174
$timeoutMax = $timeout ?? max($timeoutMax, $response->timeout);
171175
$timeoutMin = min($timeoutMin, $response->timeout, 1);
172176
$chunk = false;
173177

174-
if ($fromLastTimeout && null !== $multi->lastTimeout) {
175-
$elapsedTimeout = hrtime(true) / 1E9 - $multi->lastTimeout;
176-
}
177-
178178
if (isset($multi->handlesActivity[$j])) {
179179
$multi->lastTimeout = null;
180+
$elapsedTimeout = 0;
180181
} elseif (!isset($multi->openHandles[$j])) {
182+
$hasActivity = true;
181183
unset($responses[$j]);
182184
continue;
183185
} elseif ($elapsedTimeout >= $timeoutMax) {
184186
$multi->handlesActivity[$j] = [new ErrorChunk($response->offset, sprintf('Idle timeout reached for "%s".', $response->getInfo('url')))];
185187
$multi->lastTimeout ??= $lastActivity;
188+
$elapsedTimeout = $timeoutMax;
186189
} else {
187190
continue;
188191
}
189192

190-
while ($multi->handlesActivity[$j] ?? false) {
191-
$hasActivity = true;
192-
$elapsedTimeout = 0;
193+
$lastActivity = null;
194+
$hasActivity = true;
193195

196+
while ($multi->handlesActivity[$j] ?? false) {
194197
if (\is_string($chunk = array_shift($multi->handlesActivity[$j]))) {
195198
if (null !== $response->inflate && false === $chunk = @inflate_add($response->inflate, $chunk)) {
196199
$multi->handlesActivity[$j] = [null, new TransportException(sprintf('Error while processing content unencoding for "%s".', $response->getInfo('url')))];
@@ -227,7 +230,6 @@ public static function stream(iterable $responses, ?float $timeout = null): \Gen
227230
}
228231
} elseif ($chunk instanceof ErrorChunk) {
229232
unset($responses[$j]);
230-
$elapsedTimeout = $timeoutMax;
231233
} elseif ($chunk instanceof FirstChunk) {
232234
if ($response->logger) {
233235
$info = $response->getInfo();
@@ -274,10 +276,12 @@ public static function stream(iterable $responses, ?float $timeout = null): \Gen
274276
if ($chunk instanceof ErrorChunk && !$chunk->didThrow()) {
275277
// Ensure transport exceptions are always thrown
276278
$chunk->getContent();
279+
throw new \LogicException('A transport exception should have been thrown.');
277280
}
278281
}
279282

280283
if (!$responses) {
284+
$hasActivity = true;
281285
unset($runningResponses[$i]);
282286
}
283287

@@ -291,7 +295,7 @@ public static function stream(iterable $responses, ?float $timeout = null): \Gen
291295
}
292296

293297
if ($hasActivity) {
294-
$lastActivity = hrtime(true) / 1E9;
298+
$lastActivity ??= hrtime(true) / 1E9;
295299
continue;
296300
}
297301

0 commit comments

Comments
 (0)