Close connections on destruct

[danog]

Fixes #395, the root cause is likely some leaked reference somewhere and this is more of a hotfix, but it does the job.

[trowski]

Commented on the source issue that I'm not able to reproduce this issue this is fixing. I can envision scenarios where the connection pool is destroyed before a request has completed, so I'm not eager to abruptly close the connections.

[kelunik]

I have similar thoughts, so I don’t think we should accept this change.

[danog]

The fully reproduces the leaks:

<?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();

[danog]

Ping

[danog]

A better reproducer:

<?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', '16M');

final class Test {

	public static function t(): void {

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

			if ($x % 1000 === 0) {
				echo "counter: $x\n";
				self::resolveIps($ips);
				while (gc_collect_cycles());
				echo "memory: " . intval(memory_get_usage() / 1024). " KB \n";
				$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();

[trowski]

Fix by #400.