Skip to content

Commit d05b20e

Browse files
nicolas-grekasnicolas-grekas
committed
[HttpClient] fix monitoring responses issued before reset()
1 parent ec58f58 commit d05b20e

File tree

4 files changed

+79
-63
lines changed

4 files changed

+79
-63
lines changed

‎CurlHttpClient.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -306,9 +306,9 @@ public function stream($responses, float $timeout = null): ResponseStreamInterfa
306306
throw new \TypeError(sprintf('"%s()" expects parameter 1 to be an iterable of CurlResponse objects, "%s" given.', __METHOD__, \is_object($responses) ? \get_class($responses) : \gettype($responses)));
307307
}
308308

309-
if (\is_resource($this->multi->handle) || $this->multi->handle instanceof \CurlMultiHandle) {
309+
if (\is_resource($mh = $this->multi->handles[0] ?? null) || $mh instanceof \CurlMultiHandle) {
310310
$active = 0;
311-
while (\CURLM_CALL_MULTI_PERFORM === curl_multi_exec($this->multi->handle, $active)) {
311+
while (\CURLM_CALL_MULTI_PERFORM === curl_multi_exec($mh, $active)) {
312312
}
313313
}
314314

‎Internal/CurlClientState.php‎

Lines changed: 26 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,8 @@
2323
*/
2424
final class CurlClientState extends ClientState
2525
{
26-
/** @var \CurlMultiHandle|resource */
27-
public $handle;
26+
/** @var array<\CurlMultiHandle|resource> */
27+
public $handles = [];
2828
/** @var PushedResponse[] */
2929
public $pushedResponses = [];
3030
/** @var DnsCache */
@@ -41,20 +41,20 @@ public function __construct(int $maxHostConnections, int $maxPendingPushes)
4141
{
4242
self::$curlVersion = self::$curlVersion ?? curl_version();
4343

44-
$this->handle = curl_multi_init();
44+
array_unshift($this->handles, $mh = curl_multi_init());
4545
$this->dnsCache = new DnsCache();
4646
$this->maxHostConnections = $maxHostConnections;
4747
$this->maxPendingPushes = $maxPendingPushes;
4848

4949
// Don't enable HTTP/1.1 pipelining: it forces responses to be sent in order
5050
if (\defined('CURLPIPE_MULTIPLEX')) {
51-
curl_multi_setopt($this->handle, \CURLMOPT_PIPELINING, \CURLPIPE_MULTIPLEX);
51+
curl_multi_setopt($mh, \CURLMOPT_PIPELINING, \CURLPIPE_MULTIPLEX);
5252
}
5353
if (\defined('CURLMOPT_MAX_HOST_CONNECTIONS')) {
54-
$maxHostConnections = curl_multi_setopt($this->handle, \CURLMOPT_MAX_HOST_CONNECTIONS, 0 < $maxHostConnections ? $maxHostConnections : \PHP_INT_MAX) ? 0 : $maxHostConnections;
54+
$maxHostConnections = curl_multi_setopt($mh, \CURLMOPT_MAX_HOST_CONNECTIONS, 0 < $maxHostConnections ? $maxHostConnections : \PHP_INT_MAX) ? 0 : $maxHostConnections;
5555
}
5656
if (\defined('CURLMOPT_MAXCONNECTS') && 0 < $maxHostConnections) {
57-
curl_multi_setopt($this->handle, \CURLMOPT_MAXCONNECTS, $maxHostConnections);
57+
curl_multi_setopt($mh, \CURLMOPT_MAXCONNECTS, $maxHostConnections);
5858
}
5959

6060
// Skip configuring HTTP/2 push when it's unsupported or buggy, see https://bugs.php.net/77535
@@ -67,45 +67,40 @@ public function __construct(int $maxHostConnections, int $maxPendingPushes)
6767
return;
6868
}
6969

70-
curl_multi_setopt($this->handle, \CURLMOPT_PUSHFUNCTION, function ($parent, $pushed, array $requestHeaders) use ($maxPendingPushes) {
71-
return $this->handlePush($parent, $pushed, $requestHeaders, $maxPendingPushes);
70+
// Clone to prevent a circular reference
71+
$multi = clone $this;
72+
$multi->handles = [$mh];
73+
$multi->pushedResponses = &$this->pushedResponses;
74+
$multi->logger = &$this->logger;
75+
$multi->handlesActivity = &$this->handlesActivity;
76+
$multi->openHandles = &$this->openHandles;
77+
$multi->lastTimeout = &$this->lastTimeout;
78+
79+
curl_multi_setopt($mh, \CURLMOPT_PUSHFUNCTION, static function ($parent, $pushed, array $requestHeaders) use ($multi, $maxPendingPushes) {
80+
return $multi->handlePush($parent, $pushed, $requestHeaders, $maxPendingPushes);
7281
});
7382
}
7483

7584
public function reset()
7685
{
77-
if ($this->logger) {
78-
foreach ($this->pushedResponses as $url => $response) {
79-
$this->logger->debug(sprintf('Unused pushed response: "%s"', $url));
86+
foreach ($this->pushedResponses as $url => $response) {
87+
$this->logger && $this->logger->debug(sprintf('Unused pushed response: "%s"', $url));
88+
89+
foreach ($this->handles as $mh) {
90+
curl_multi_remove_handle($mh, $response->handle);
8091
}
92+
curl_close($response->handle);
8193
}
8294

8395
$this->pushedResponses = [];
8496
$this->dnsCache->evictions = $this->dnsCache->evictions ?: $this->dnsCache->removals;
8597
$this->dnsCache->removals = $this->dnsCache->hostnames = [];
8698

87-
if (\is_resource($this->handle) || $this->handle instanceof \CurlMultiHandle) {
88-
if (\defined('CURLMOPT_PUSHFUNCTION')) {
89-
curl_multi_setopt($this->handle, \CURLMOPT_PUSHFUNCTION, null);
90-
}
91-
92-
curl_multi_close($this->handle);
93-
$this->__construct($this->maxHostConnections, $this->maxPendingPushes);
99+
if (\defined('CURLMOPT_PUSHFUNCTION')) {
100+
curl_multi_setopt($this->handles[0], \CURLMOPT_PUSHFUNCTION, null);
94101
}
95-
}
96-
97-
public function __wakeup()
98-
{
99-
throw new \BadMethodCallException('Cannot unserialize '.__CLASS__);
100-
}
101102

102-
public function __destruct()
103-
{
104-
foreach ($this->openHandles as [$ch]) {
105-
if (\is_resource($ch) || $ch instanceof \CurlHandle) {
106-
curl_setopt($ch, \CURLOPT_VERBOSE, false);
107-
}
108-
}
103+
$this->__construct($this->maxHostConnections, $this->maxPendingPushes);
109104
}
110105

111106
private function handlePush($parent, $pushed, array $requestHeaders, int $maxPendingPushes): int

‎Response/CurlResponse.php‎

Lines changed: 38 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,7 @@ public function __construct(CurlClientState $multi, $ch, array $options = null,
150150
// Schedule the request in a non-blocking way
151151
$multi->lastTimeout = null;
152152
$multi->openHandles[$id] = [$ch, $options];
153-
curl_multi_add_handle($multi->handle, $ch);
153+
curl_multi_add_handle($multi->handles[0], $ch);
154154

155155
$this->canary = new Canary(static function () use ($ch, $multi, $id) {
156156
unset($multi->openHandles[$id], $multi->handlesActivity[$id]);
@@ -160,7 +160,9 @@ public function __construct(CurlClientState $multi, $ch, array $options = null,
160160
return;
161161
}
162162

163-
curl_multi_remove_handle($multi->handle, $ch);
163+
foreach ($multi->handles as $mh) {
164+
curl_multi_remove_handle($mh, $ch);
165+
}
164166
curl_setopt_array($ch, [
165167
\CURLOPT_NOPROGRESS => true,
166168
\CURLOPT_PROGRESSFUNCTION => null,
@@ -242,7 +244,7 @@ public function __destruct()
242244
*/
243245
private static function schedule(self $response, array &$runningResponses): void
244246
{
245-
if (isset($runningResponses[$i = (int) $response->multi->handle])) {
247+
if (isset($runningResponses[$i = (int) $response->multi->handles[0]])) {
246248
$runningResponses[$i][1][$response->id] = $response;
247249
} else {
248250
$runningResponses[$i] = [$response->multi, [$response->id => $response]];
@@ -274,39 +276,47 @@ private static function perform(ClientState $multi, array &$responses = null): v
274276

275277
try {
276278
self::$performing = true;
277-
$active = 0;
278-
while (\CURLM_CALL_MULTI_PERFORM === ($err = curl_multi_exec($multi->handle, $active))) {
279-
}
280-
281-
if (\CURLM_OK !== $err) {
282-
throw new TransportException(curl_multi_strerror($err));
283-
}
284279

285-
while ($info = curl_multi_info_read($multi->handle)) {
286-
if (\CURLMSG_DONE !== $info['msg']) {
287-
continue;
280+
foreach ($multi->handles as $i => $mh) {
281+
$active = 0;
282+
while (\CURLM_CALL_MULTI_PERFORM === ($err = curl_multi_exec($mh, $active))) {
288283
}
289-
$result = $info['result'];
290-
$id = (int) $ch = $info['handle'];
291-
$waitFor = @curl_getinfo($ch, \CURLINFO_PRIVATE) ?: '_0';
292284

293-
if (\in_array($result, [\CURLE_SEND_ERROR, \CURLE_RECV_ERROR, /*CURLE_HTTP2*/ 16, /*CURLE_HTTP2_STREAM*/ 92], true) && $waitFor[1] && 'C' !== $waitFor[0]) {
294-
curl_multi_remove_handle($multi->handle, $ch);
295-
$waitFor[1] = (string) ((int) $waitFor[1] - 1); // decrement the retry counter
296-
curl_setopt($ch, \CURLOPT_PRIVATE, $waitFor);
297-
curl_setopt($ch, \CURLOPT_FORBID_REUSE, true);
285+
if (\CURLM_OK !== $err) {
286+
throw new TransportException(curl_multi_strerror($err));
287+
}
298288

299-
if (0 === curl_multi_add_handle($multi->handle, $ch)) {
289+
while ($info = curl_multi_info_read($mh)) {
290+
if (\CURLMSG_DONE !== $info['msg']) {
300291
continue;
301292
}
302-
}
293+
$result = $info['result'];
294+
$id = (int) $ch = $info['handle'];
295+
$waitFor = @curl_getinfo($ch, \CURLINFO_PRIVATE) ?: '_0';
296+
297+
if (\in_array($result, [\CURLE_SEND_ERROR, \CURLE_RECV_ERROR, /*CURLE_HTTP2*/ 16, /*CURLE_HTTP2_STREAM*/ 92], true) && $waitFor[1] && 'C' !== $waitFor[0]) {
298+
curl_multi_remove_handle($mh, $ch);
299+
$waitFor[1] = (string) ((int) $waitFor[1] - 1); // decrement the retry counter
300+
curl_setopt($ch, \CURLOPT_PRIVATE, $waitFor);
301+
curl_setopt($ch, \CURLOPT_FORBID_REUSE, true);
302+
303+
if (0 === curl_multi_add_handle($mh, $ch)) {
304+
continue;
305+
}
306+
}
307+
308+
if (\CURLE_RECV_ERROR === $result && 'H' === $waitFor[0] && 400 <= ($responses[(int) $ch]->info['http_code'] ?? 0)) {
309+
$multi->handlesActivity[$id][] = new FirstChunk();
310+
}
303311

304-
if (\CURLE_RECV_ERROR === $result && 'H' === $waitFor[0] && 400 <= ($responses[(int) $ch]->info['http_code'] ?? 0)) {
305-
$multi->handlesActivity[$id][] = new FirstChunk();
312+
$multi->handlesActivity[$id][] = null;
313+
$multi->handlesActivity[$id][] = \in_array($result, [\CURLE_OK, \CURLE_TOO_MANY_REDIRECTS], true) || '_0' === $waitFor || curl_getinfo($ch, \CURLINFO_SIZE_DOWNLOAD) === curl_getinfo($ch, \CURLINFO_CONTENT_LENGTH_DOWNLOAD) ? null : new TransportException(sprintf('%s for "%s".', curl_strerror($result), curl_getinfo($ch, \CURLINFO_EFFECTIVE_URL)));
306314
}
307315

308-
$multi->handlesActivity[$id][] = null;
309-
$multi->handlesActivity[$id][] = \in_array($result, [\CURLE_OK, \CURLE_TOO_MANY_REDIRECTS], true) || '_0' === $waitFor || curl_getinfo($ch, \CURLINFO_SIZE_DOWNLOAD) === curl_getinfo($ch, \CURLINFO_CONTENT_LENGTH_DOWNLOAD) ? null : new TransportException(sprintf('%s for "%s".', curl_strerror($result), curl_getinfo($ch, \CURLINFO_EFFECTIVE_URL)));
316+
if (!$active && 0 < $i) {
317+
curl_multi_close($mh);
318+
unset($multi->handles[$i]);
319+
}
310320
}
311321
} finally {
312322
self::$performing = false;
@@ -325,7 +335,7 @@ private static function select(ClientState $multi, float $timeout): int
325335
$timeout = min($timeout, 0.01);
326336
}
327337

328-
return curl_multi_select($multi->handle, $timeout);
338+
return curl_multi_select($multi->handles[array_key_last($multi->handles)], $timeout);
329339
}
330340

331341
/**

‎Tests/CurlHttpClientTest.php‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -143,9 +143,20 @@ public function testHandleIsReinitOnReset()
143143
$r = new \ReflectionProperty($httpClient, 'multi');
144144
$r->setAccessible(true);
145145
$clientState = $r->getValue($httpClient);
146-
$initialHandleId = (int) $clientState->handle;
146+
$initialHandleId = (int) $clientState->handles[0];
147147
$httpClient->reset();
148-
self::assertNotSame($initialHandleId, (int) $clientState->handle);
148+
self::assertNotSame($initialHandleId, (int) $clientState->handles[0]);
149+
}
150+
151+
public function testProcessAfterReset()
152+
{
153+
$client = $this->getHttpClient(__FUNCTION__);
154+
155+
$response = $client->request('GET', 'http://127.0.0.1:8057/json');
156+
157+
$client->reset();
158+
159+
$this->assertSame(['application/json'], $response->getHeaders()['content-type']);
149160
}
150161

151162
private function getVulcainClient(): CurlHttpClient

0 commit comments

Comments
 (0)