Connections leak when destroying client

[danog]

The following code causes a memory/FD leak:

<?php

require 'vendor/autoload.php';

use Amp\Http\Client\Request;
use Amp\Http\Client\HttpClientBuilder;
use function Amp\async;
use function Amp\Future\await;


ini_set('memory_limit', '64M');

final class Test {

	public static function t(): void {

		$ips = [];
		$counter = 0;
		for ($x = 0; $x < 1000000; $x++) {
			$ip = long2ip($x);
			$ips[$ip] = $ip;

			if ($x % 1000 === 0) {
				echo "counter: $counter\n";
				self::resolveIps($ips);
				$ips = [];
			}
		}
	}

	private static function resolveIps(array $ips, int $threads = 10): void {
		$client = (new HttpClientBuilder())
			->retry(0)
			->build()
		;

		$futures = [];
		foreach ($ips as $ip => &$item) {
			if (str_starts_with($ip, '172.')) {
				$item = [
					'country' => '',
					'city' => '',
				];
				continue;
			}
			$futures[] = async(function () use (&$client, $ip, &$item) {
				$response = $client->request(new Request('http://127.0.0.1:16003/?ip=' . $ip));
				$responseString = $response->getBody()->buffer();
			});

			if (count($futures) % $threads === 0) {
				await($futures);
				$futures = [];
			}

		}unset($item);

		if (count($futures) > 0) {
			await($futures);
			unset($futures);
		}
	}
}


Test::t();

[trowski]

Running the script above with some modifications to hit the hello-world.php example in http-server, I'm not able to reproduce this issue. Is there something I'm missing which needs to be done?

[danog]

It takes quite a few iterations, so it's hard to reproduce especially on the slower http-server, reproducing faster with a go server.

Tweaking the parameters reproduces on http-server as well:

<?php

require 'vendor/autoload.php';

use Amp\Http\Client\Request;
use Amp\Http\Client\HttpClientBuilder;
use Amp\Http\HttpStatus;
use Amp\Http\Server\DefaultErrorHandler;
use Amp\Http\Server\Request as ServerRequest;
use Amp\Http\Server\RequestHandler;
use Amp\Http\Server\Response;
use Amp\Http\Server\SocketHttpServer;
use Amp\Log\ConsoleFormatter;
use Amp\Log\StreamHandler;
use Monolog\Logger;
use Monolog\Processor\PsrLogMessageProcessor;
use Psr\Log\NullLogger;

use function Amp\async;
use function Amp\ByteStream\getStdout;
use function Amp\Future\await;

$requestHandler = new class() implements RequestHandler {
    public function handleRequest(ServerRequest $request) : Response
    {
        return new Response(
            status: HttpStatus::OK,
            headers: ['Content-Type' => 'text/plain'],
            body: 'Hello, world!',
        );
    }
};
$errorHandler = new DefaultErrorHandler();

$server = SocketHttpServer::createForDirectAccess(new NullLogger);
$server->expose('127.0.0.1:1337');
$server->start($requestHandler, $errorHandler);

ini_set('memory_limit', '8M');

final class Test {

	public static function t(): void {

		$ips = [];
		$counter = 0;
		for ($x = 0; $x < 1000000; $x++) {
			$ip = long2ip($x);
			$ips[$ip] = $ip;

			if ($x % 1000 === 0) {
				echo "counter: $counter\n";
				self::resolveIps($ips);
				$ips = [];
			}
		}
	}

	private static function resolveIps(array $ips, int $threads = 10): void {
		$client = (new HttpClientBuilder())
			->retry(0)
			->build()
		;

		$futures = [];
		foreach ($ips as $ip => &$item) {
			if (str_starts_with($ip, '172.')) {
				$item = [
					'country' => '',
					'city' => '',
				];
				continue;
			}
			$futures[] = async(function () use (&$client, $ip, &$item) {
                $r = new Request('http://127.0.0.1:1337/?ip=' . $ip);
                $r->setHeader('Connection', 'keep-alive');
				$response = $client->request($r);
				$responseString = $response->getBody()->buffer();
			});

			if (count($futures) % $threads === 0) {
				await($futures);
				$futures = [];
			}

		}unset($item);

		if (count($futures) > 0) {
			await($futures);
			unset($futures);
		}
	}
}


Test::t();