Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
98.26%
282 / 287
93.10%
27 / 29
 CRAP
0.00%
0 / 1
AmpConnectionPool
98.26%
282 / 287
93.10%
27 / 29
 122
0.00%
0 / 1
 __construct
90.32%
28 / 31
0.00%
0 / 1
 14.18
 __destruct
100.00%
1 / 1
100.00%
1 / 1
 1
 acquire
100.00%
12 / 12
100.00%
1 / 1
 6
 release
100.00%
20 / 20
100.00%
1 / 1
 7
 close
100.00%
10 / 10
100.00%
1 / 1
 4
 tuneFromOptions
100.00%
19 / 19
100.00%
1 / 1
 8
 advertised
100.00%
3 / 3
100.00%
1 / 1
 3
 admit
100.00%
10 / 10
100.00%
1 / 1
 6
 reserve
100.00%
3 / 3
100.00%
1 / 1
 1
 makeRoom
100.00%
12 / 12
100.00%
1 / 1
 5
 wait
100.00%
35 / 35
100.00%
1 / 1
 9
 dispatch
100.00%
12 / 12
100.00%
1 / 1
 5
 giveBack
100.00%
4 / 4
100.00%
1 / 1
 2
 connect
100.00%
16 / 16
100.00%
1 / 1
 4
 vet
100.00%
15 / 15
100.00%
1 / 1
 5
 isQuiet
100.00%
8 / 8
100.00%
1 / 1
 3
 sweep
100.00%
11 / 11
100.00%
1 / 1
 6
 evictOldestIdle
100.00%
15 / 15
100.00%
1 / 1
 10
 occupancy
100.00%
15 / 15
100.00%
1 / 1
 5
 slotOf
100.00%
1 / 1
100.00%
1 / 1
 2
 watch
100.00%
6 / 6
100.00%
1 / 1
 1
 assertOpen
100.00%
2 / 2
100.00%
1 / 1
 2
 startSweeper
81.82%
9 / 11
0.00%
0 / 1
 4.10
 sweepIdle
100.00%
3 / 3
100.00%
1 / 1
 2
 stopSweeper
100.00%
3 / 3
100.00%
1 / 1
 2
 endpointCap
100.00%
5 / 5
100.00%
1 / 1
 1
 endpointOf
100.00%
1 / 1
100.00%
1 / 1
 1
 bucketOf
100.00%
2 / 2
100.00%
1 / 1
 2
 tlsToken
100.00%
1 / 1
100.00%
1 / 1
 1
1<?php
2
3/**
4 * SPDX-License-Identifier: EUPL-1.2
5 *
6 * This file is part of icap-flow.
7 *
8 * Licensed under the EUPL, Version 1.2 only (the "Licence");
9 * you may not use this work except in compliance with the Licence.
10 * You may obtain a copy of the Licence at:
11 *
12 *     https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
13 *
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the Licence is distributed on an "AS IS" basis,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 */
18
19declare(strict_types=1);
20
21namespace Ndrstmr\Icap\Transport;
22
23use Amp\Cancellation;
24use Amp\CancelledException;
25use Amp\DeferredCancellation;
26use Amp\Socket\ClientTlsContext;
27use Amp\Socket\Socket as SocketInterface;
28use Closure;
29use Ndrstmr\Icap\Config;
30use Ndrstmr\Icap\DTO\IcapResponse;
31use Ndrstmr\Icap\Exception\IcapConnectionException;
32use Ndrstmr\Icap\Exception\IcapMalformedResponseException;
33use Ndrstmr\Icap\Exception\IcapTimeoutException;
34use Ndrstmr\Icap\HeaderFields;
35use Revolt\EventLoop;
36use WeakMap;
37use WeakReference;
38
39/**
40 * In-process keep-alive pool of amphp sockets, with a limit on connections.
41 *
42 * **Limits (M04).** `maxConnectionsPerHost` is the most connections the pool has to one endpoint (host and port,
43 * whatever the TLS context), and `maxConnections` the most it has at all (default: the same number). Connections
44 * being set up, in use and idle all count, and room is reserved before the connector is asked. A call that finds
45 * no room fails with {@see IcapConnectionException} unless a queue is configured: `maxWaiters` calls may then wait,
46 * first come first served, each for at most `waitTimeout` seconds ({@see IcapTimeoutException}); the caller's
47 * cancellation ends a wait as `Amp\CancelledException`. A wait ends at its deadline on the monotonic clock, also
48 * when something held up the event loop and the timer has not run: a call whose wait is over is served nothing.
49 * Only an idle socket gives way for a new connection, never one in use. Without a pool (or with
50 * {@see NullConnectionPool}) the application limits its own concurrency.
51 *
52 * **Ownership (M06).** The pool takes back only sockets it lent out, and for the bucket they were lent for: a socket
53 * it never lent out, and one given back for another endpoint or TLS context, is closed, never pooled. A socket that
54 * is given back again while it is idle stays in the pool once and is handed out once; the second release changes
55 * nothing. What the pool cannot tell is a second release of a socket that has been handed out again meanwhile from
56 * the release of its new holder (the interface carries no lease): a session guards itself against giving back
57 * twice. A socket the holder closed itself, or that was collected, gives its room back, and so does one whose
58 * peer hung up and which its holder read to the end of the stream: it is closed whole. After {@see close()} idle
59 * sockets are closed, calls that wait fail, a connection that is still being set up is closed when it arrives, and
60 * sockets in use stay with their holders and are closed when they are given back.
61 *
62 * **Idle sockets (M05).** Idle sockets are a LIFO stack per bucket (host, port and TLS context; most recently used
63 * first). Each records when it became idle on a monotonic clock. Every acquire ages out all of them, in every
64 * bucket, and drops empty buckets; a timer (unreferenced, running only while something is idle, gone with
65 * {@see close()} and with the pool) does so when nothing is acquired. The pool does not watch idle
66 * sockets: when it hands one out it reads it once, and a socket whose peer has closed, that has received bytes
67 * nobody asked for, or that fails, is closed instead of handed out. That is no guarantee that the peer still
68 * listens; a request on a socket that died a moment later fails like any other.
69 *
70 * Security domains are separate pools; a rotated TLS context is a new context (or a new pool, with the old one
71 * closed). Certificate files that change under a context in use change nothing for the sockets already open.
72 */
73final class AmpConnectionPool implements ConnectionPoolInterface
74{
75    /** The idle timer runs at half the idle time, but not more often than this and not less often than that. */
76    private const float MIN_SWEEP_INTERVAL = 0.05;
77    private const float MAX_SWEEP_INTERVAL = 10.0;
78
79    /** The most endpoints whose Max-Connections the pool keeps; a pool serves a handful of servers, not an open set. */
80    private const int MAX_ENDPOINT_LIMITS = 256;
81
82    /** @var array<string, list<array{socket: SocketInterface, idleSince: float}>> idle sockets per bucket, oldest first */
83    private array $idle = [];
84
85    /** @var WeakMap<SocketInterface, PoolSlot> every open socket the pool lent out or holds idle */
86    private WeakMap $slots;
87
88    /** @var array<int, PoolSlot> connections being set up: room is reserved, no socket yet */
89    private array $connecting = [];
90
91    /** @var array<int, PoolWaiter> calls that wait for room, oldest first */
92    private array $waiters = [];
93
94    private int $lastWaiter = 0;
95
96    private bool $closed = false;
97
98    private ?string $sweeper = null;
99
100    /** @var WeakMap<ClientTlsContext, int> */
101    private WeakMap $tlsTokens;
102
103    private int $lastTlsToken = 0;
104
105    private int $maxConnections;
106
107    private float $waitTimeout;
108
109    private float $sweepInterval;
110
111    /** @var array<string, int> Max-Connections per endpoint, as advertised */
112    private array $serverMaxByEndpoint = [];
113
114    /** @var Closure(Config, ?Cancellation): SocketInterface */
115    private Closure $connector;
116
117    /** @var Closure(): float */
118    private Closure $clock;
119
120    /**
121     * @param int                                                    $maxConnectionsPerHost cap on the connections (being set up, in use and idle) to one endpoint, host and port, whatever the TLS context
122     * @param (Closure(Config, ?Cancellation): SocketInterface)|null $connector             optional override — production code uses the amphp connector; tests can inject pre-built socket pairs
123     * @param int|null                                               $serverMaxConnections  Max-Connections (RFC 3507 §4.10.2) taken as the limit of every endpoint of the pool; it can only lower `maxConnectionsPerHost`. Per endpoint: {@see tuneFromOptions()}
124     * @param float                                                  $maxIdleSeconds        idle sockets older than this are closed (default 30 s)
125     * @param (Closure(): float)|null                                $clock                 clock for idle-age and wait-deadline checks; defaults to a monotonic one (`hrtime`); injectable for deterministic tests
126     * @param int|null                                               $maxConnections        cap on the connections of the whole pool, all endpoints together; default `maxConnectionsPerHost`
127     * @param int                                                    $maxWaiters            how many calls may wait for a connection when there is none; 0 (default) is no queue: a call that finds the pool full fails at once
128     * @param float|null                                             $waitTimeout           the longest a call waits, in seconds; required with a queue
129     */
130    public function __construct(
131        private int $maxConnectionsPerHost = 8,
132        ?Closure $connector = null,
133        private ?int $serverMaxConnections = null,
134        private float $maxIdleSeconds = 30.0,
135        ?Closure $clock = null,
136        ?int $maxConnections = null,
137        private int $maxWaiters = 0,
138        ?float $waitTimeout = null,
139    ) {
140        if ($maxConnectionsPerHost < 1) {
141            throw new \InvalidArgumentException(
142                'maxConnectionsPerHost must be >= 1, got: ' . $maxConnectionsPerHost,
143            );
144        }
145
146        if ($serverMaxConnections !== null && $serverMaxConnections < 1) {
147            throw new \InvalidArgumentException(
148                'serverMaxConnections must be >= 1, got: ' . $serverMaxConnections,
149            );
150        }
151
152        if (!is_finite($maxIdleSeconds) || $maxIdleSeconds <= 0.0) {
153            throw new \InvalidArgumentException(
154                'maxIdleSeconds must be a finite number > 0, got: ' . var_export($maxIdleSeconds, true),
155            );
156        }
157
158        if ($maxConnections !== null && $maxConnections < 1) {
159            throw new \InvalidArgumentException(
160                'maxConnections must be >= 1, got: ' . $maxConnections,
161            );
162        }
163
164        if ($maxWaiters < 0) {
165            throw new \InvalidArgumentException(
166                'maxWaiters must be >= 0, got: ' . $maxWaiters,
167            );
168        }
169
170        if (($waitTimeout !== null && (!is_finite($waitTimeout) || $waitTimeout <= 0.0)) || ($maxWaiters > 0 && $waitTimeout === null)) {
171            throw new \InvalidArgumentException(
172                'waitTimeout must be a finite number > 0, and is required with maxWaiters > 0, got: ' . var_export($waitTimeout, true),
173            );
174        }
175
176        $this->maxConnections = $maxConnections ?? $maxConnectionsPerHost;
177        $this->waitTimeout = $waitTimeout ?? 0.0;
178        $this->sweepInterval = max(self::MIN_SWEEP_INTERVAL, min($maxIdleSeconds / 2, self::MAX_SWEEP_INTERVAL));
179        $this->connector = $connector ?? SocketOpener::open(...);
180        $this->clock = $clock ?? static fn (): float => hrtime(true) / 1e9;
181        $this->slots = new WeakMap();
182        $this->tlsTokens = new WeakMap();
183    }
184
185    public function __destruct()
186    {
187        $this->stopSweeper();
188    }
189
190    /**
191     * @throws IcapConnectionException When the pool is closed, no connection can be offered (the limit is reached
192     *         and there is no queue, or the queue is full), or the default connector fails; a connector of your own
193     *         passes what it throws on unchanged
194     * @throws IcapTimeoutException    When a call waited for the whole `waitTimeout` without being served
195     * @throws CancelledException      When the caller's cancellation ended the call
196     */
197    #[\Override]
198    public function acquire(Config $config, ?Cancellation $cancellation = null): SocketInterface
199    {
200        $this->assertOpen();
201        // Cancelled before it began: neither the idle sockets nor the connector are asked.
202        $cancellation?->throwIfRequested();
203
204        $endpoint = $this->endpointOf($config);
205        $bucket = $this->bucketOf($config, $endpoint);
206
207        while (true) {
208            // Calls that wait come first: a call that arrives now does not pass them.
209            $grant = $this->waiters === [] ? $this->admit($endpoint, $bucket) : null;
210            $grant ??= $this->wait($endpoint, $bucket, $cancellation);
211
212            if ($grant instanceof SocketInterface) {
213                $grant = $this->vet($grant, $cancellation);
214                if ($grant === null) {
215                    // The room it held fell under a cap meanwhile: ask again, as any call would.
216                    continue;
217                }
218            }
219
220            return $grant instanceof PoolSlot ? $this->connect($config, $cancellation, $grant) : $grant;
221        }
222    }
223
224    #[\Override]
225    public function release(Config $config, SocketInterface $socket): void
226    {
227        $slot = $this->slotOf($socket);
228
229        if ($slot === null) {
230            // Not a socket the pool is lending out: one it never issued, or one that is closed and forgotten.
231            $socket->close();
232            $this->dispatch();
233
234            return;
235        }
236
237        if ($slot->idle) {
238            // Given back before: it is in the pool once, and stays there once.
239            return;
240        }
241
242        if ($this->closed || $socket->isClosed() || $slot->bucket !== $this->bucketOf($config, $this->endpointOf($config))) {
243            // The pool is closed, or the socket is no use, or it is given back for another endpoint or TLS context
244            // than it was lent for: another bucket would be served a connection it must not have.
245            $socket->close();
246            $this->dispatch();
247
248            return;
249        }
250
251        [, $perEndpoint] = $this->occupancy();
252        if (($perEndpoint[$slot->endpoint] ?? 0) > $this->endpointCap($slot->endpoint)) {
253            // A cap fell below what is in use: what comes back beyond it is closed.
254            $socket->close();
255            $this->dispatch();
256
257            return;
258        }
259
260        $slot->idle = true;
261        $this->idle[$slot->bucket][] = ['socket' => $socket, 'idleSince' => ($this->clock)()];
262        $this->startSweeper();
263        $this->dispatch();
264    }
265
266    #[\Override]
267    public function close(): void
268    {
269        $this->closed = true;
270        $this->stopSweeper();
271
272        foreach ($this->idle as $entries) {
273            foreach ($entries as $entry) {
274                $entry['socket']->close();
275            }
276        }
277        $this->idle = [];
278
279        // Calls that wait find the pool closed. Sockets in use stay with their holders.
280        foreach ($this->waiters as $waiter) {
281            $waiter->wake();
282        }
283        $this->serverMaxByEndpoint = [];
284        $this->occupancy();
285    }
286
287    /**
288     * Take the `Max-Connections` header of an OPTIONS response as a limit (RFC 3507 §4.10.2). It can only lower
289     * what the pool allows itself: the effective cap is the smaller of `maxConnectionsPerHost` and the number the
290     * server advertised. Idle sockets above the new cap are closed at once; sockets in use run to their end, and
291     * what comes back beyond the cap is closed.
292     *
293     * The pool does not know which service an OPTIONS response belongs to, only the endpoint, and only if told:
294     * with `$config` the number applies to that endpoint (host and port). Without it the pool cannot tell whose
295     * limit it is and applies it conservatively to every endpoint it serves. The pool keeps the number of at most
296     * 256 endpoints; for a further one it refuses, rather than forget a limit it was told, which would be to
297     * bypass it. A cap that rose serves the calls that wait.
298     *
299     * Typical usage after an OPTIONS round-trip:
300     *
301     *     $response = $client->options('/avscan')->await();
302     *     $pool->tuneFromOptions($response, $config);
303     *
304     * @see IcapResponse
305     *
306     * @throws IcapMalformedResponseException When the response carries conflicting `Max-Connections` fields, or a
307     *         value that is not a positive decimal number of at most nine digits
308     * @throws IcapConnectionException         When the pool is closed, or already keeps the limits of as many
309     *         endpoints as it takes and `$config` names another one
310     */
311    public function tuneFromOptions(IcapResponse $response, ?Config $config = null): void
312    {
313        $value = HeaderFields::single($response->headers, 'Max-Connections');
314        if ($value === null) {
315            return;
316        }
317        $advertised = self::advertised($value);
318        $this->assertOpen();
319
320        if ($config === null) {
321            $this->serverMaxConnections = $advertised;
322        } else {
323            $endpoint = $this->endpointOf($config);
324            if (!isset($this->serverMaxByEndpoint[$endpoint]) && count($this->serverMaxByEndpoint) >= self::MAX_ENDPOINT_LIMITS) {
325                throw new IcapConnectionException(sprintf(
326                    'The pool already keeps the Max-Connections of %d endpoints and takes no more: use another pool for other servers.',
327                    self::MAX_ENDPOINT_LIMITS,
328                ));
329            }
330            $this->serverMaxByEndpoint[$endpoint] = $advertised;
331        }
332
333        // Idle sockets above a cap that fell are closed now; sockets in use keep running and are closed on return.
334        [, $perEndpoint] = $this->occupancy();
335        foreach ($perEndpoint as $endpoint => $count) {
336            while ($count > $this->endpointCap($endpoint) && $this->evictOldestIdle($endpoint)) {
337                $count--;
338            }
339        }
340
341        // A cap that rose is room for the calls that wait.
342        $this->dispatch();
343    }
344
345    /**
346     * A positive decimal number of at most nine digits, checked before it is cast: (int) reads '2.5' as 2,
347     * '2abc' as 2 and '1e3' as 1000, and a longer number is no connection limit anyone means.
348     *
349     * @throws IcapMalformedResponseException
350     */
351    private static function advertised(string $value): int
352    {
353        if (preg_match('/^[0-9]{1,9}\z/', $value) !== 1 || (int) $value < 1) {
354            // What the server sent stays out of the message: it would end up in logs and in handlers.
355            throw new IcapMalformedResponseException('Invalid Max-Connections header in the ICAP response: not a positive decimal number of at most nine digits.');
356        }
357
358        return (int) $value;
359    }
360
361    /**
362     * One atomic step of admission, without a suspension: an idle socket of the bucket, the most recently used
363     * first, or reserved room for a new connection, or nothing.
364     */
365    private function admit(string $endpoint, string $bucket): SocketInterface|PoolSlot|null
366    {
367        $this->sweep();
368
369        $entry = isset($this->idle[$bucket]) ? array_pop($this->idle[$bucket]) : null;
370        if ($entry !== null) {
371            if ($this->idle[$bucket] === []) {
372                unset($this->idle[$bucket]);
373            }
374            $slot = $this->slotOf($entry['socket']);
375            if ($slot !== null) {
376                $slot->idle = false;
377            }
378
379            return $entry['socket'];
380        }
381
382        return $this->makeRoom($endpoint) ? $this->reserve($endpoint, $bucket) : null;
383    }
384
385    /**
386     * Room for one connection, counted from now until the connection is a socket or has failed.
387     */
388    private function reserve(string $endpoint, string $bucket): PoolSlot
389    {
390        $slot = new PoolSlot($endpoint, $bucket);
391        $this->connecting[spl_object_id($slot)] = $slot;
392
393        return $slot;
394    }
395
396    /**
397     * Whether a new connection to the endpoint fits, closing idle sockets (the oldest first) if that is what
398     * stands in the way. A socket in use is never closed to make room: its scan is still running.
399     */
400    private function makeRoom(string $endpoint): bool
401    {
402        [$total, $perEndpoint] = $this->occupancy();
403        $here = $perEndpoint[$endpoint] ?? 0;
404
405        while ($here >= $this->endpointCap($endpoint)) {
406            if (!$this->evictOldestIdle($endpoint)) {
407                return false;
408            }
409            $here--;
410            $total--;
411        }
412
413        while ($total >= $this->maxConnections) {
414            if (!$this->evictOldestIdle(null)) {
415                return false;
416            }
417            $total--;
418        }
419
420        return true;
421    }
422
423    /**
424     * Waits, first come first served, for room or an idle socket. Returns what the pool granted.
425     *
426     * @throws IcapConnectionException No queue, or the queue is full, or the pool was closed
427     * @throws IcapTimeoutException    The wait ran out
428     * @throws CancelledException      The caller gave up
429     */
430    private function wait(string $endpoint, string $bucket, ?Cancellation $cancellation): SocketInterface|PoolSlot
431    {
432        if ($this->maxWaiters === 0) {
433            throw new IcapConnectionException('No connection is available: the limit of the pool is reached, and the pool has no queue.');
434        }
435        if (count($this->waiters) >= $this->maxWaiters) {
436            throw new IcapConnectionException(sprintf(
437                'No connection is available: the limit of the pool is reached, and its queue of %d calls is full.',
438                $this->maxWaiters,
439            ));
440        }
441
442        $waiter = new PoolWaiter($endpoint, $bucket, ($this->clock)() + $this->waitTimeout);
443        $id = ++$this->lastWaiter;
444        $this->waiters[$id] = $waiter;
445        // Timer and subscription are removed on every way out, not left to destructors: an exception that is
446        // still referenced keeps the arguments of its stack frames alive.
447        $timer = EventLoop::delay($this->waitTimeout, static function () use ($waiter): void {
448            $waiter->wake();
449        });
450        $watch = $cancellation?->subscribe(static function () use ($waiter): void {
451            $waiter->wake();
452        });
453
454        try {
455            // This call may be servable although the calls before it are not.
456            $this->dispatch();
457            $waiter->await();
458        } finally {
459            EventLoop::cancel($timer);
460            if ($cancellation !== null && $watch !== null) {
461                $cancellation->unsubscribe($watch);
462            }
463            unset($this->waiters[$id]);
464        }
465
466        $grant = $waiter->grant;
467
468        try {
469            // The caller's cancellation wins over whatever else ended the wait.
470            $cancellation?->throwIfRequested();
471            $this->assertOpen();
472            if ($grant === null || ($this->clock)() >= $waiter->deadline) {
473                // Also what was granted after the deadline, or before it but taken over only after it, because
474                // the event loop was held up: the wait is over, and the call gets nothing.
475                throw new IcapTimeoutException(sprintf(
476                    'No connection became available within the pool wait timeout of %s seconds.',
477                    $this->waitTimeout,
478                ));
479            }
480        } catch (\Throwable $e) {
481            // Granted in the same moment the wait ended: the room goes back, once.
482            if ($grant !== null) {
483                $this->giveBack($grant);
484            }
485
486            throw $e;
487        }
488
489        return $grant;
490    }
491
492    /**
493     * Hands room to the calls that wait, oldest first, as far as it goes. Called wherever room may have come free.
494     */
495    private function dispatch(): void
496    {
497        if ($this->waiters === []) {
498            return;
499        }
500
501        $now = ($this->clock)();
502        foreach ($this->waiters as $id => $waiter) {
503            if ($now >= $waiter->deadline) {
504                // Its wait is over, whether or not its timer has run yet: it finds that out when it wakes.
505                continue;
506            }
507
508            $grant = $this->admit($waiter->endpoint, $waiter->bucket);
509            if ($grant === null) {
510                continue;
511            }
512            $waiter->grant = $grant;
513            unset($this->waiters[$id]);
514            $waiter->wake();
515        }
516    }
517
518    /**
519     * Takes back what a call was granted and will not use: the room, or the idle socket (which is closed).
520     */
521    private function giveBack(SocketInterface|PoolSlot $grant): void
522    {
523        if ($grant instanceof PoolSlot) {
524            unset($this->connecting[spl_object_id($grant)]);
525        } else {
526            $grant->close();
527        }
528
529        $this->dispatch();
530    }
531
532    /**
533     * Opens the connection the reserved room is for. The room goes back on every way out that is not a socket.
534     */
535    private function connect(Config $config, ?Cancellation $cancellation, PoolSlot $slot): SocketInterface
536    {
537        $socket = null;
538
539        try {
540            $socket = ($this->connector)($config, $cancellation);
541
542            // A connector need not honour the token it was given, and the pool may have been closed while it
543            // worked: what it hands back after that is not handed out.
544            $cancellation?->throwIfRequested();
545            $this->assertOpen();
546
547            if ($this->slotOf($socket) !== null && !$socket->isClosed()) {
548                // A socket the pool has lent out already is its holder's; leave it alone.
549                $socket = null;
550                throw new IcapConnectionException('The connector returned a socket that is in use.');
551            }
552
553            unset($this->connecting[spl_object_id($slot)]);
554            $this->slots[$socket] = $slot;
555            $this->watch($socket);
556
557            return $socket;
558        } catch (\Throwable $e) {
559            $socket?->close();
560            unset($this->connecting[spl_object_id($slot)]);
561            $this->dispatch();
562
563            throw $e;
564        }
565    }
566
567    /**
568     * Checks an idle socket before it is handed out: it is read once, and any byte or the end of the stream
569     * means it may not be. A socket that is not fit is closed, and its room is used for a new connection, so
570     * the call that was served does not lose its turn (unless a cap fell below the room it held).
571     *
572     * @return SocketInterface|PoolSlot|null the socket; or the room for a new connection; or null
573     *
574     * @throws IcapConnectionException|CancelledException When the pool was closed or the caller gave up meanwhile
575     */
576    private function vet(SocketInterface $socket, ?Cancellation $cancellation): SocketInterface|PoolSlot|null
577    {
578        try {
579            $quiet = $this->isQuiet($socket);
580            $cancellation?->throwIfRequested();
581            $this->assertOpen();
582        } catch (\Throwable $e) {
583            $socket->close();
584            $this->dispatch();
585
586            throw $e;
587        }
588
589        if ($quiet) {
590            return $socket;
591        }
592
593        $slot = $this->slotOf($socket);
594        $socket->close();
595
596        // The connection that is gone left room for exactly one in its place, unless a cap fell below it meanwhile.
597        if ($slot === null || !$this->makeRoom($slot->endpoint)) {
598            $this->dispatch();
599
600            return null;
601        }
602
603        return $this->reserve($slot->endpoint, $slot->bucket);
604    }
605
606    /**
607     * Nothing may be waiting on an idle connection: a byte is data nobody asked for, the end of the stream is
608     * the peer going away, an error is an error. The read carries a cancellation that is already requested, so
609     * it takes what is there and ends in the same turn of the event loop; nothing is left watching the socket,
610     * and what the next user is sent is the next user's. A closed socket answers with the end of the stream. The
611     * socket is closed when it is not quiet, so no byte that is taken here is missed.
612     */
613    private function isQuiet(SocketInterface $socket): bool
614    {
615        $probe = new DeferredCancellation();
616        $probe->cancel();
617
618        try {
619            $socket->read($probe->getCancellation());
620
621            return false;
622        } catch (CancelledException) {
623            return true;
624        } catch (\Throwable) {
625            return false;
626        }
627    }
628
629    /**
630     * Closes the idle sockets that are closed, or older than `maxIdleSeconds`, in every bucket, and drops the
631     * buckets that are left empty.
632     */
633    private function sweep(): void
634    {
635        $now = ($this->clock)();
636
637        foreach ($this->idle as $bucket => $entries) {
638            $kept = [];
639            foreach ($entries as $entry) {
640                if ($entry['socket']->isClosed() || ($now - $entry['idleSince']) > $this->maxIdleSeconds) {
641                    $entry['socket']->close();
642                    continue;
643                }
644                $kept[] = $entry;
645            }
646
647            if ($kept === []) {
648                unset($this->idle[$bucket]);
649            } else {
650                $this->idle[$bucket] = $kept;
651            }
652        }
653    }
654
655    /**
656     * Closes the oldest idle socket, of the endpoint or of any endpoint, to make room.
657     */
658    private function evictOldestIdle(?string $endpoint): bool
659    {
660        $oldest = null;
661
662        foreach ($this->idle as $bucket => $entries) {
663            $entry = $entries[0] ?? null;
664            if ($entry === null || ($endpoint !== null && $this->slotOf($entry['socket'])?->endpoint !== $endpoint)) {
665                continue;
666            }
667            if ($oldest === null || $entry['idleSince'] < $oldest[1]) {
668                $oldest = [$bucket, $entry['idleSince']];
669            }
670        }
671
672        if ($oldest === null) {
673            return false;
674        }
675
676        $entry = array_shift($this->idle[$oldest[0]]);
677        if ($this->idle[$oldest[0]] === []) {
678            unset($this->idle[$oldest[0]]);
679        }
680        if ($entry !== null) {
681            $entry['socket']->close();
682        }
683
684        return true;
685    }
686
687    /**
688     * The connections the pool answers for: those being set up, in use and idle, in all and per endpoint. A socket
689     * that is closed is no connection, however it was closed; one that was collected is not even in the registry.
690     *
691     * @return array{int, array<string, int>}
692     */
693    private function occupancy(): array
694    {
695        $total = 0;
696        $perEndpoint = [];
697
698        foreach ($this->connecting as $slot) {
699            $total++;
700            $perEndpoint[$slot->endpoint] = ($perEndpoint[$slot->endpoint] ?? 0) + 1;
701        }
702
703        $closed = [];
704        foreach ($this->slots as $socket => $slot) {
705            if ($socket->isClosed()) {
706                $closed[] = $socket;
707                continue;
708            }
709            $total++;
710            $perEndpoint[$slot->endpoint] = ($perEndpoint[$slot->endpoint] ?? 0) + 1;
711        }
712        foreach ($closed as $socket) {
713            unset($this->slots[$socket]);
714        }
715
716        return [$total, $perEndpoint];
717    }
718
719    private function slotOf(SocketInterface $socket): ?PoolSlot
720    {
721        return $this->slots->offsetExists($socket) ? $this->slots[$socket] : null;
722    }
723
724    /**
725     * A socket that its holder closes itself, or that is collected, gives its room back at once (the registry
726     * does not count it); calls that wait are told when it happens.
727     *
728     * The callback is the socket's, and amphp calls it when the reading end closes. That happens when the peer
729     * hung up and the holder read the end of the stream, and the writing end may be open still: the socket is not
730     * closed, its room is not free, and when the holder closes it later nothing tells the pool again. A connection
731     * nobody can read an answer from is no use for ICAP, so it is closed whole here, and its room is free now.
732     * Neither the pool nor the socket is held by the callback: a collected socket closes itself.
733     */
734    private function watch(SocketInterface $socket): void
735    {
736        $pool = WeakReference::create($this);
737        $watched = WeakReference::create($socket);
738        $socket->onClose(static function () use ($pool, $watched): void {
739            $watched->get()?->close();
740            $pool->get()?->dispatch();
741        });
742    }
743
744    private function assertOpen(): void
745    {
746        if ($this->closed) {
747            throw new IcapConnectionException('Pool is closed.');
748        }
749    }
750
751    private function startSweeper(): void
752    {
753        if ($this->sweeper !== null || $this->closed) {
754            return;
755        }
756
757        // The timer holds the pool weakly and does not keep the event loop running: a pool nobody holds any
758        // more, or a loop that has nothing else to do, is not kept alive by it.
759        $pool = WeakReference::create($this);
760        $this->sweeper = EventLoop::repeat($this->sweepInterval, static function (string $id) use ($pool): void {
761            $self = $pool->get();
762            if ($self === null) {
763                EventLoop::cancel($id);
764
765                return;
766            }
767            $self->sweepIdle();
768        });
769        EventLoop::unreference($this->sweeper);
770    }
771
772    private function sweepIdle(): void
773    {
774        $this->sweep();
775        if ($this->idle === []) {
776            $this->stopSweeper();
777        }
778    }
779
780    private function stopSweeper(): void
781    {
782        if ($this->sweeper !== null) {
783            EventLoop::cancel($this->sweeper);
784            $this->sweeper = null;
785        }
786    }
787
788    /**
789     * The cap on the connections to one endpoint: the pool's own, lowered by what servers advertised.
790     */
791    private function endpointCap(string $endpoint): int
792    {
793        return min(
794            $this->maxConnectionsPerHost,
795            $this->serverMaxConnections ?? PHP_INT_MAX,
796            $this->serverMaxByEndpoint[$endpoint] ?? PHP_INT_MAX,
797        );
798    }
799
800    /**
801     * What a connection counts against: host and port, whatever the TLS context. Host names are not case sensitive.
802     */
803    private function endpointOf(Config $config): string
804    {
805        return strtolower($config->host) . ':' . $config->port;
806    }
807
808    /**
809     * Where an idle socket may be given out: the endpoint and, for TLS, the context it was opened for.
810     */
811    private function bucketOf(Config $config, string $endpoint): string
812    {
813        $tls = $config->getTlsContext();
814
815        return $tls === null ? $endpoint : $endpoint . ':tls:' . $this->tlsToken($tls);
816    }
817
818    /**
819     * The pool's own identity for a TLS context: one number per context
820     * object, handed out once and never again. An object hash would not do:
821     * it is recycled as soon as the context is collected, and connectTls()
822     * connects with a clone of a context without peer name, so nothing keeps
823     * the original alive while its idle socket is. The next context to take
824     * over the hash would then be served the old socket (B07).
825     *
826     * The counter only grows. Past PHP_INT_MAX the increment throws a
827     * TypeError instead of wrapping, so an identity is never issued twice.
828     */
829    private function tlsToken(ClientTlsContext $tls): int
830    {
831        return $this->tlsTokens[$tls] ??= ++$this->lastTlsToken;
832    }
833}