ОсновиPDO Pool

PDO Pool ​

Імпорт майже готовий: воркери перевіряють адреси через GeoDirectory. Лишається тільки зберегти перевірені адреси в базу даних. Звучить банально:

php
$pdo = new PDO('mysql:host=localhost;dbname=profiles', 'app', 'secret');

for ($i = 0; $i < 10; $i++) {
    $workers->spawn(function () use ($queue, $pdo) {
        foreach ($queue as $address) {
                checkAddress($address);
                saveAddress($pdo, $address);
        }
    });
}

Один об'єкт PDO, який ділять десять корутин. У розділі про канали ми казали, що гонитви даних не трапляються в межах одного потоку, тож усе має бути гаразд, правда ж?

Ні. Це справедливо для пам'яті. Але з'єднання з базою даних це не пам'ять, це протокол поверх сокета: запит, відповідь, запит, відповідь, строго по черзі. Кожен запит до бази це точка очікування, у якій корутина засинає, а керування переходить до іншої. І та інша записує свій власний запит у той самий сокет:

php
// Воркер 1
$pdo->beginTransaction();
$pdo->exec("INSERT INTO addresses ...");
// точка очікування: воркер 1 засинає, воркер 2 прокидається

// Воркер 2
$pdo->beginTransaction(); // на тому самому з'єднанні!
$pdo->exec("UPDATE ...");
$pdo->commit(); // фіксує і свою транзакцію, і чужу

Відповіді переплутуються, транзакції фіксують чужу роботу. Гонитви повернулися, тільки тепер вони живуть у з'єднанні, а не в пам'яті.

Гаразд, то дати кожній корутині власне з'єднання?

php
$workers->spawn(function () use ($queue) {
    $pdo = new PDO(/* ... */); // власне з'єднання
    // ...
});

Годиться для десяти воркерів. Але уявіть не завдання імпорту, а сервер, де корутина створюється на кожен запит. Тисяча корутин означає тисячу TCP-з'єднань. MySQL за замовчуванням дозволяє 151, PostgreSQL 100. І відкривати з'єднання заради кількох мілісекунд роботи просто дорого: рукостискання з базою може тривати довше за сам запит.

Знайомо звучить? У розділі про канали була та сама розвилка: сто тисяч з'єднань до GeoDirectory чи черга на десять.

Пул: черга навпаки ​

Розв'язання називається пулом з'єднань: відкрити N з'єднань заздалегідь і роздавати їх корутинам на час їхньої роботи. Як виявляється, ми вже вміємо його будувати. Пул це канал, який тримає з'єднання:

php
$pool = new Channel(5);

for ($i = 0; $i < 5; $i++) {
    $pool->send(new PDO(/* ... */));
}

// Усередині корутини:
$pdo = $pool->recv();   // узяти з'єднання
saveAddress($pdo, $address);
$pool->send($pdo);      // повернути його

Вільні з'єднання лежать у буфері. Якщо всі вони зайняті, recv присипляє корутину, доки хтось не поверне з'єднання. Та сама синхронізація з попереднього розділу, тільки черга тримає ресурси, а не завдання.

Схема працює, але має слабке місце: треба не забути повернути з'єднання, хоч би що сталося, зокрема виняток чи скасування. А ще є транзакції, обірвані з'єднання, перепідключення. Для PDO про все це вже подбано, просто в ядрі.

PDO Pool ​

Пул вбудований у сам PDO і вмикається через атрибути конструктора:

php
$pdo = new PDO('mysql:host=localhost;dbname=profiles', 'app', 'secret', [
    PDO::ATTR_POOL_ENABLED => true,
    PDO::ATTR_POOL_MIN     => 2,
    PDO::ATTR_POOL_MAX     => 10,
]);

Ззовні нічого не змінилося: той найперший приклад, де єдиний $pdo іде до десяти воркерів, тепер коректний. Об'єкт $pdo більше не з'єднання, це фасад для пулу. Коли корутина виконує свій перший запит, пул дає їй виділене з'єднання, і всі запити цієї корутини йдуть через нього. Щойно корутина завершується, з'єднання повертається в пул, готове для наступної корутини.

Ніяких ручних recv і send: отримання й повернення відбуваються самі, у потрібні моменти, хоч би як усе обернулося. Код виглядає як звичайний синхронний PHP зі звичайним PDO, і саме в цьому й був увесь задум.

Транзакції ​

Транзакція це стан, який належить з'єднанню, тож пул ставиться до неї особливо: поки транзакція відкрита, з'єднання закріплене за своєю корутиною і не повернеться в пул:

php
$workers->spawn(function () use ($pdo, $queue) {
    // ...
    $pdo->beginTransaction();
    $pdo->exec("INSERT INTO addresses (user_id, region) VALUES (...)");
    $pdo->exec("UPDATE users SET address_checked = 1 WHERE id = ...");
    $pdo->commit();
    // тільки тепер з'єднання може повернутися в пул
});

А що, як корутина завершується, не викликавши commit? Пригадайте розділ про Scope: воркера могли скасувати просто посеред транзакції, і це нормальний сценарій, а не катастрофа. Перед поверненням з'єднання пул автоматично виконує ROLLBACK. Незавершена транзакція не витече в наступну корутину й не зависне в базі даних.

Коли з'єднання обривається ​

Імпорт може тривати годину. За годину база даних може перезавантажитися, мережа може моргнути, а DBA може вбити сесію. У класичному PHP скрипт просто впав би, але застосунок, що працює довго, має вміти продовжувати роботу.

Пул перевіряє з'єднання, коли їх повертають: обірване знищується замість того, щоб бути переданим наступній корутині. А якщо запит зазнає невдачі через обірване з'єднання, достатньо просто повторити його на тому самому $pdo, пул дасть корутині свіже з'єднання:

php
try {
    saveAddress($pdo, $address);
} catch (PDOException $e) {
    saveAddress($pdo, $address); // пул уже підставив нове з'єднання
}

Не потрібно писати логіку перепідключення, не потрібно перестворювати об'єкт PDO. Просто повторіть запит, про решту подбає пул.

Зрештою, код працює так, ніби в кожної корутини було власне з'єднання з базою даних. Насправді з'єднань лише десять, і пул постійно передає їх з рук у руки, пильнує за транзакціями та відкидає мертві. Але жодна з цих кухонних справ не видна ззовні, і в цьому головна користь: ми просто пишемо звичайний код із PDO.

До речі, наші воркери досі працюють напівсліпо: вони перевіряють адреси й зберігають їх, але ніхто не рахує, скільки адрес виявилося поганих, чи які саме. Як отримати результати назад від корутин і зручно їх зібрати? Ось цим ми й займемося в наступному розділі.