ОсновыPool

Pool ​

В главе про PDO Pool мы набросали пул из канала за шесть строчек и тут же признались, что для настоящей жизни он наивен: вернуть ресурс надо не забыть при любом исходе, битые соединения надо распознавать, а умершие заменять. Для PDO всё это сделало ядро. Но checkAddress ходит в GeoDirectory по HTTP, и его соединения тоже жалко открывать на каждый запрос. А завтра появится Redis для кэша и SMTP для писем.

Не собирать же для каждого ресурса свой пул из канала. Правильно, для этого есть готовый примитив, тот самый, что работает внутри PDO Pool: Async\Pool.

Фабрика и деструктор ​

Пул не знает, что за ресурс он хранит. Вы объясняете ему две вещи: как ресурс создать и как уничтожить:

php
use Async\Pool;

$geo = new Pool(
    factory:    fn() => new GeoConnection('geodirectory.example.com'),
    destructor: fn($conn) => $conn->close(),
    min: 2,
    max: 10,
);

min: 2 — два соединения открываются заранее, чтобы первые корутины не ждали рукопожатия. max: 10 — больше десяти соединений пул не создаст никогда, сколько бы корутин ни просило. Знакомый предел конкурентности, только теперь он привязан не к числу воркеров, а к самому ресурсу.

Дальше цикл, который мы уже строили руками через канал:

php
$conn = $geo->acquire();

try {
    $verdict = $conn->request("/check?address=$address");
} finally {
    $geo->release($conn);
}

acquire выдаёт свободный ресурс. Если свободных нет, а лимит не достигнут, фабрика создаст новый. Если достигнут, корутина заснёт до чужого release: та же механика, что у recv на пустом канале. Обратите внимание на finally: ресурс возвращается при любом исходе, включая исключение и отмену. Это тот самый пункт, на котором спотыкался наш самодельный пул.

Ждать можно и с пределом, привычным образом:

php
$conn = $geo->acquire(timeout: 3000); // TimeoutException, если за 3 секунды не дождались

Ресурсы умирают ​

Соединение, пролежавшее в пуле полчаса, могло тихо умереть: сервер закрыл его по таймауту, сеть моргнула. В главе про PDO мы уже видели, чем это грозит: следующая корутина получает загадочную ошибку на ровном месте. Пул борется с этим на трёх рубежах:

php
$geo = new Pool(
    factory:             fn() => new GeoConnection('geodirectory.example.com'),
    destructor:          fn($conn) => $conn->close(),
    healthcheck:         fn($conn) => $conn->ping(),
    beforeAcquire:       fn($conn) => $conn->isAlive(),
    beforeRelease:       fn($conn) => !$conn->isBroken(),
    min: 2,
    max: 10,
    healthcheckInterval: 30000,
);
  • healthcheck — раз в тридцать секунд пул сам обходит свободные ресурсы и проверяет их пульс. Мёртвые уничтожаются и заменяются новыми.
  • beforeAcquire — последний осмотр перед выдачей. Не прошёл — ресурс уничтожается, корутине достаётся следующий.
  • beforeRelease — осмотр при возврате. Корутина могла сломать соединение посреди запроса; такое в пул не возвращается.

Корутины об этой кухне не знают: они просят ресурс и получают рабочий. Забота о здоровье выражена декларативно, в конструкторе, а не размазана if-ами по всему коду.

Предохранитель ​

Теперь представьте, что GeoDirectory упал целиком. Все десять соединений мертвы, каждая корутина честно ждёт таймаут, создаёт новое соединение, снова ждёт. Приложение тратит все силы на бомбардировку лежащего сервиса, и ему же мешает подняться.

В электрике для этого придуман предохранитель: цепь размыкается, пока авария не устранена. Паттерн так и называется, circuit breaker, и он встроен в пул. У пула есть три состояния: ACTIVE, всё работает; INACTIVE, сервис признан недоступным и acquire отказывает сразу, без таймаутов; RECOVERING, осторожная проверка, ожил ли сервис.

Переключать состояния можно вручную, а можно доверить стратегии:

php
use Async\CircuitBreakerStrategy;

final class FiveStrikes implements CircuitBreakerStrategy
{
    private int $failures = 0;

    public function reportSuccess(mixed $pool): void
    {
        $this->failures = 0;
        $pool->activate();
    }

    public function reportFailure(mixed $pool, Throwable $error): void
    {
        if (++$this->failures >= 5) {
            $pool->deactivate();
        }
    }

    public function shouldRecover(): bool
    {
        return true; // пробовать восстановление при первой возможности
    }
}

$geo->setCircuitBreakerStrategy(new FiveStrikes());

Пул сам сообщает стратегии о каждом успехе и каждой поломке ресурса, а через shouldRecover спрашивает, не пора ли осторожно проверить, ожил ли сервис. Пять поломок подряд, и цепь размыкается: корутины получают быстрый отказ вместо медленной пытки таймаутами, а GeoDirectory перестаёт получать бессмысленные удары и спокойно поднимается.

Если оглянуться, Pool делает ровно то, что мы в девятой главе собирали руками из канала, и вдобавок то, что руками собирать совсем не хочется: гарантированный возврат, контроль здоровья, предохранитель. PDO Pool — тот же примитив, только спрятанный за фасадом PDO. А для всего остального, HTTP-клиентов, Redis, сокетов и просто тяжёлых объектов, есть Async\Pool.

К этому моменту наш арсенал выглядит внушительно: корутины, каналы, scope, группы задач, пулы. Но всё это крутится в одном потоке операционной системы и делит одно ядро процессора. Пока задачи ждут ввода-вывода, это никого не смущает. А что делать, если работа упирается не в ожидание, а в вычисления? Тут картина меняется целиком, и об этом следующая глава.