Skip to content

Commit b425007

Browse files
nicolas-grekasnicolas-grekas
committed
[HttpClient] Don't read from the network faster than the CPU can deal with
1 parent f53699e commit b425007

File tree

6 files changed

+77
-100
lines changed

6 files changed

+77
-100
lines changed

‎CurlHttpClient.php‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -113,7 +113,7 @@ public function request(string $method, string $url, array $options = []): Respo
113113
$url = implode('', $url);
114114

115115
if (!isset($options['normalized_headers']['user-agent'])) {
116-
$options['normalized_headers']['user-agent'][] = $options['headers'][] = 'User-Agent: Symfony HttpClient/Curl';
116+
$options['headers'][] = 'User-Agent: Symfony HttpClient/Curl';
117117
}
118118

119119
$curlopts = [
@@ -194,8 +194,8 @@ public function request(string $method, string $url, array $options = []): Respo
194194
$curlopts[CURLOPT_NOSIGNAL] = true;
195195
}
196196

197-
if (!isset($options['normalized_headers']['accept-encoding']) && CURL_VERSION_LIBZ & self::$curlVersion['features']) {
198-
$curlopts[CURLOPT_ENCODING] = 'gzip'; // Expose only one encoding, some servers mess up when more are provided
197+
if (\extension_loaded('zlib') && !isset($options['normalized_headers']['accept-encoding'])) {
198+
$options['headers'][] = 'Accept-Encoding: gzip'; // Expose only one encoding, some servers mess up when more are provided
199199
}
200200

201201
foreach ($options['headers'] as $header) {

‎NativeHttpClient.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ public function request(string $method, string $url, array $options = []): Respo
7777
$options['headers'][] = 'Content-Type: application/x-www-form-urlencoded';
7878
}
7979

80-
if ($gzipEnabled = \extension_loaded('zlib') && !isset($options['normalized_headers']['accept-encoding'])) {
80+
if (\extension_loaded('zlib') && !isset($options['normalized_headers']['accept-encoding'])) {
8181
// gzip is the most widely available algo, no need to deal with deflate
8282
$options['headers'][] = 'Accept-Encoding: gzip';
8383
}
@@ -210,7 +210,7 @@ public function request(string $method, string $url, array $options = []): Respo
210210
$context = stream_context_create($context, ['notification' => $notification]);
211211
self::configureHeadersAndProxy($context, $host, $options['headers'], $proxy, $noProxy);
212212

213-
return new NativeResponse($this->multi, $context, implode('', $url), $options, $gzipEnabled, $info, $resolveRedirect, $onProgress, $this->logger);
213+
return new NativeResponse($this->multi, $context, implode('', $url), $options, $info, $resolveRedirect, $onProgress, $this->logger);
214214
}
215215

216216
/**

‎Response/CurlResponse.php‎

Lines changed: 13 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ public function __construct(CurlClientState $multi, $ch, array $options = null,
5252

5353
$this->id = $id = (int) $ch;
5454
$this->logger = $logger;
55+
$this->shouldBuffer = $options['buffer'] ?? true;
5556
$this->timeout = $options['timeout'] ?? null;
5657
$this->info['http_method'] = $method;
5758
$this->info['user_data'] = $options['user_data'] ?? null;
@@ -65,30 +66,25 @@ public function __construct(CurlClientState $multi, $ch, array $options = null,
6566
curl_setopt($ch, CURLOPT_PRIVATE, \in_array($method, ['GET', 'HEAD', 'OPTIONS', 'TRACE'], true) && 1.0 < (float) ($options['http_version'] ?? 1.1) ? 'H2' : 'H0'); // H = headers + retry counter
6667
}
6768

68-
if (null === $content = &$this->content) {
69-
$content = ($options['buffer'] ?? true) ? fopen('php://temp', 'w+') : null;
70-
} else {
71-
// Move the pushed response to the activity list
72-
if (ftell($content)) {
73-
rewind($content);
74-
$multi->handlesActivity[$id][] = stream_get_contents($content);
75-
}
76-
$content = ($options['buffer'] ?? true) ? $content : null;
77-
}
78-
7969
curl_setopt($ch, CURLOPT_HEADERFUNCTION, static function ($ch, string $data) use (&$info, &$headers, $options, $multi, $id, &$location, $resolveRedirect, $logger): int {
8070
return self::parseHeaderLine($ch, $data, $info, $headers, $options, $multi, $id, $location, $resolveRedirect, $logger);
8171
});
8272

8373
if (null === $options) {
8474
// Pushed response: buffer until requested
85-
curl_setopt($ch, CURLOPT_WRITEFUNCTION, static function ($ch, string $data) use (&$content): int {
86-
return fwrite($content, $data);
75+
curl_setopt($ch, CURLOPT_WRITEFUNCTION, static function ($ch, string $data) use ($multi, $id): int {
76+
$multi->handlesActivity[$id][] = $data;
77+
curl_pause($ch, CURLPAUSE_RECV);
78+
79+
return \strlen($data);
8780
});
8881

8982
return;
9083
}
9184

85+
$this->inflate = !isset($options['normalized_headers']['accept-encoding']);
86+
curl_pause($ch, CURLPAUSE_CONT);
87+
9288
if ($onProgress = $options['on_progress']) {
9389
$url = isset($info['url']) ? ['url' => $info['url']] : [];
9490
curl_setopt($ch, CURLOPT_NOPROGRESS, false);
@@ -108,33 +104,16 @@ public function __construct(CurlClientState $multi, $ch, array $options = null,
108104
});
109105
}
110106

111-
curl_setopt($ch, CURLOPT_WRITEFUNCTION, static function ($ch, string $data) use (&$content, $multi, $id): int {
107+
curl_setopt($ch, CURLOPT_WRITEFUNCTION, static function ($ch, string $data) use ($multi, $id): int {
112108
$multi->handlesActivity[$id][] = $data;
113109

114-
return null !== $content ? fwrite($content, $data) : \strlen($data);
110+
return \strlen($data);
115111
});
116112

117113
$this->initializer = static function (self $response) {
118-
if (null !== $response->info['error']) {
119-
throw new TransportException($response->info['error']);
120-
}
121-
122114
$waitFor = curl_getinfo($ch = $response->handle, CURLINFO_PRIVATE);
123115

124-
if ('H' === $waitFor[0] || 'D' === $waitFor[0]) {
125-
try {
126-
foreach (self::stream([$response]) as $chunk) {
127-
if ($chunk->isFirst()) {
128-
break;
129-
}
130-
}
131-
} catch (\Throwable $e) {
132-
// Persist timeouts thrown during initialization
133-
$response->info['error'] = $e->getMessage();
134-
$response->close();
135-
throw $e;
136-
}
137-
}
116+
return 'H' === $waitFor[0] || 'D' === $waitFor[0];
138117
};
139118

140119
// Schedule the request in a non-blocking way
@@ -221,6 +200,7 @@ public function __destruct()
221200
*/
222201
private function close(): void
223202
{
203+
$this->inflate = null;
224204
unset($this->multi->openHandles[$this->id], $this->multi->handlesActivity[$this->id]);
225205
curl_multi_remove_handle($this->multi->handle, $this->handle);
226206
curl_setopt_array($this->handle, [

‎Response/MockResponse.php‎

Lines changed: 3 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@ public function cancel(): void
9393
*/
9494
protected function close(): void
9595
{
96+
$this->inflate = null;
9697
$this->body = [];
9798
}
9899

@@ -104,16 +105,9 @@ public static function fromRequest(string $method, string $url, array $options,
104105
$response = new self([]);
105106
$response->requestOptions = $options;
106107
$response->id = ++self::$idSequence;
107-
$response->content = ($options['buffer'] ?? true) ? fopen('php://temp', 'w+') : null;
108+
$response->shouldBuffer = $options['buffer'] ?? true;
108109
$response->initializer = static function (self $response) {
109-
if (null !== $response->info['error']) {
110-
throw new TransportException($response->info['error']);
111-
}
112-
113-
if (\is_array($response->body[0] ?? null)) {
114-
// Consume the first chunk if it's not yielded yet
115-
self::stream([$response])->current();
116-
}
110+
return \is_array($response->body[0] ?? null);
117111
};
118112

119113
$response->info['redirect_count'] = 0;
@@ -186,11 +180,6 @@ protected static function perform(ClientState $multi, array &$responses): void
186180
} else {
187181
// Data or timeout chunk
188182
$multi->handlesActivity[$id][] = $chunk;
189-
190-
if (\is_string($chunk) && null !== $response->content) {
191-
// Buffer response body
192-
fwrite($response->content, $chunk);
193-
}
194183
}
195184
}
196185
}

‎Response/NativeResponse.php‎

Lines changed: 12 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -32,14 +32,13 @@ final class NativeResponse implements ResponseInterface
3232
private $onProgress;
3333
private $remaining;
3434
private $buffer;
35-
private $inflate;
3635
private $multi;
3736
private $debugBuffer;
3837

3938
/**
4039
* @internal
4140
*/
42-
public function __construct(NativeClientState $multi, $context, string $url, $options, bool $gzipEnabled, array &$info, callable $resolveRedirect, ?callable $onProgress, ?LoggerInterface $logger)
41+
public function __construct(NativeClientState $multi, $context, string $url, array $options, array &$info, callable $resolveRedirect, ?callable $onProgress, ?LoggerInterface $logger)
4342
{
4443
$this->multi = $multi;
4544
$this->id = (int) $context;
@@ -50,27 +49,17 @@ public function __construct(NativeClientState $multi, $context, string $url, $op
5049
$this->info = &$info;
5150
$this->resolveRedirect = $resolveRedirect;
5251
$this->onProgress = $onProgress;
53-
$this->content = $options['buffer'] ? fopen('php://temp', 'w+') : null;
52+
$this->inflate = !isset($options['normalized_headers']['accept-encoding']);
53+
$this->shouldBuffer = $options['buffer'] ?? true;
5454

55-
// Temporary resources to dechunk/inflate the response stream
55+
// Temporary resource to dechunk the response stream
5656
$this->buffer = fopen('php://temp', 'w+');
57-
$this->inflate = $gzipEnabled ? inflate_init(ZLIB_ENCODING_GZIP) : null;
5857

5958
$info['user_data'] = $options['user_data'];
6059
++$multi->responseCount;
6160

6261
$this->initializer = static function (self $response) {
63-
if (null !== $response->info['error']) {
64-
throw new TransportException($response->info['error']);
65-
}
66-
67-
if (null === $response->remaining) {
68-
foreach (self::stream([$response]) as $chunk) {
69-
if ($chunk->isFirst()) {
70-
break;
71-
}
72-
}
73-
}
62+
return null === $response->remaining;
7463
};
7564
}
7665

@@ -165,7 +154,7 @@ private function open(): void
165154
stream_set_blocking($h, false);
166155
$this->context = $this->resolveRedirect = null;
167156

168-
// Create dechunk and inflate buffers
157+
// Create dechunk buffers
169158
if (isset($this->headers['content-length'])) {
170159
$this->remaining = (int) $this->headers['content-length'][0];
171160
} elseif ('chunked' === ($this->headers['transfer-encoding'][0] ?? null)) {
@@ -175,10 +164,6 @@ private function open(): void
175164
$this->remaining = -2;
176165
}
177166

178-
if ($this->inflate && 'gzip' !== ($this->headers['content-encoding'][0] ?? null)) {
179-
$this->inflate = null;
180-
}
181-
182167
$this->multi->handlesActivity[$this->id] = [new FirstChunk()];
183168

184169
if ('HEAD' === $context['http']['method'] || \in_array($this->info['http_code'], [204, 304], true)) {
@@ -188,7 +173,7 @@ private function open(): void
188173
return;
189174
}
190175

191-
$this->multi->openHandles[$this->id] = [$h, $this->buffer, $this->inflate, $this->content, $this->onProgress, &$this->remaining, &$this->info];
176+
$this->multi->openHandles[$this->id] = [$h, $this->buffer, $this->onProgress, &$this->remaining, &$this->info];
192177
}
193178

194179
/**
@@ -228,15 +213,15 @@ private static function perform(NativeClientState $multi, array &$responses = nu
228213
$multi->handles = [];
229214
}
230215

231-
foreach ($multi->openHandles as $i => [$h, $buffer, $inflate, $content, $onProgress]) {
216+
foreach ($multi->openHandles as $i => [$h, $buffer, $onProgress]) {
232217
$hasActivity = false;
233-
$remaining = &$multi->openHandles[$i][5];
234-
$info = &$multi->openHandles[$i][6];
218+
$remaining = &$multi->openHandles[$i][3];
219+
$info = &$multi->openHandles[$i][4];
235220
$e = null;
236221

237222
// Read incoming buffer and write it to the dechunk one
238223
try {
239-
while ($remaining && '' !== $data = (string) fread($h, 0 > $remaining ? 16372 : $remaining)) {
224+
if ($remaining && '' !== $data = (string) fread($h, 0 > $remaining ? 16372 : $remaining)) {
240225
fwrite($buffer, $data);
241226
$hasActivity = true;
242227
$multi->sleep = false;
@@ -264,16 +249,8 @@ private static function perform(NativeClientState $multi, array &$responses = nu
264249
rewind($buffer);
265250
ftruncate($buffer, 0);
266251

267-
if (null !== $inflate && false === $data = @inflate_add($inflate, $data)) {
268-
$e = new TransportException('Error while processing content unencoding.');
269-
}
270-
271-
if ('' !== $data && null === $e) {
252+
if (null === $e) {
272253
$multi->handlesActivity[$i][] = $data;
273-
274-
if (null !== $content && \strlen($data) !== fwrite($content, $data)) {
275-
$e = new TransportException(sprintf('Failed writing %d bytes to the response buffer.', \strlen($data)));
276-
}
277254
}
278255
}
279256

0 commit comments

Comments
 (0)