Shared\Channel

OxPHP\Shared\Channel — це обмежений канал із багатьма виробниками та багатьма споживачами, що живе в спільному реєстрі й видимий кожному PHP-воркеру процесу. Використовуйте його, коли обробник запиту і фоновий воркер або два воркери мають обмінюватися завданнями в порядку FIFO. Усередині файбера блокувальні виклики кооперативно призупиняються, тож потік воркера лишається вільним для інших запитів.

Огляд

Обмежений. Місткість задається під час створення. Коли канал заповнений, send блокується (або призупиняє файбер); trySend повідомляє результат без очікування.
MPMC. Будь-яка кількість відправників і отримувачів у різних потоках. Доставка — FIFO.
Fiber-aware. У режимі воркера з асинхронним пулом блокувальні варіанти призупиняють файбер, а не потік ОС. У традиційному режимі вони блокують потік ОС.
Типізовані результати. Кожен send/recv повертає Channel\SendResult / Channel\RecvResult — закритий канал, заповнений канал і тайм-аут є звичайними результатами для диспетчерів fan-out, тому вони подаються як варіанти результату, а не як винятки.

Довідник API

php

namespace OxPHP\Shared;

final class Channel implements Shareable, \Countable
{
    public function __construct(int $capacity);

    // ── Receive ──
    public function tryRecv(): Channel\RecvResult;                  // non-blocking
    public function recv(): Channel\RecvResult;                     // block forever / fiber-cancel
    public function recvTimeout(int $ms): Channel\RecvResult;       // bounded ($ms > 0)

    // ── Send ──
    public function trySend(mixed $value): Channel\SendResult;
    public function send(mixed $value): Channel\SendResult;
    public function sendTimeout(mixed $value, int $ms): Channel\SendResult;

    // ── Batch ──
    public function sendMany(array $values, int $ms): int;          // partial count, no throw
    public function recvMany(int $max, int $ms): array;             // partial array, no throw

    // ── Lifecycle ──
    public function close(): bool;
    public function isClosed(): bool;
    public function count(): int;
    public function id(): int;
}

namespace OxPHP\Shared\Channel;

enum RecvStatus { case Ok; case Empty; case Timeout; case Closed; }
enum SendStatus { case Ok; case Full;  case Timeout; case Closed; }

final class RecvResult {
    public function isOk(): bool;
    public function isEmpty(): bool;
    public function isTimeout(): bool;
    public function isClosed(): bool;
    public function value(): mixed;                // throws SharedException if not Ok
    public function valueOr(mixed $default): mixed;
    public function status(): RecvStatus;
}

final class SendResult {
    public function isOk(): bool;
    public function isFull(): bool;
    public function isTimeout(): bool;
    public function isClosed(): bool;
    public function status(): SendStatus;
}

Обмеження розміру

Дві межі захищають від «бомб виділення пам’яті» з недовірених вхідних даних; обидві налаштовуються змінними середовища й кидають виняток, який можна перехопити, замість аварійного завершення процесу:

Місткість — new Channel($capacity) заздалегідь виділяє масив слотів. Місткість, масив слотів якої перевищив би SHARED_MAX_CHANNEL_BYTES (типово 64 MiB, ~2 млн слотів), кидає CapacityException. Зазвичай канали мають глибину від десятків до кількох тисяч, тож ця межа спрацьовує лише на патологічних значеннях. (Недодатна $capacity — окремий TypeException.) Фактичний бюджет округлюється вгору щонайменше до одного слота, тож нульове або замале значення SHARED_MAX_CHANNEL_BYTES ніколи не відхилить мінімальний канал місткістю 1 — воно лише обмежує великі місткості.
На одне значення — кожне надіслане значення серіалізується; значення, серіалізований розмір якого перевищує SHARED_MAX_VALUE_SIZE (типово 1 MiB), кидає ValueTooLargeException. Це стосується send, trySend, sendTimeout і sendMany (який відхиляє весь пакет і нічого не надсилає, якщо хоч один елемент перевищує межу). Це та сама межа на значення, що діє в Shared\Map.

Політики очікування

Для кожного напрямку є три варіанти методу; суфікс кодує політику очікування:

Суфікс	Поведінка
`try*`	Неблокувальний. Повідомляє `Empty` / `Full`, якщо виклик не може виконатися.
(без суфікса)	Блокується назавжди або доки файбер запиту не буде скасовано.
`*Timeout`	Обмежене очікування. `$ms` має бути `> 0` — для нескінченного або неблокувального очікування використовуйте інші форми.

$ms — це завжди додатне ціле число мілісекунд для recvTimeout / sendTimeout / sendMany / recvMany. Нуль, від’ємні, нецілі та відсутні значення кидають OxPHP\Shared\TypeException на рівні мосту — це обмеження винесено з тіла методу, щоб трихотомія документувала себе сама.

Досяжні варіанти результату для кожного методу

SendResult::Full і RecvResult::Empty повертають лише неблокувальні виклики try* — блокувальний варіант або отримує слот/елемент, або вичерпує бюджет, і тоді результат — Timeout, а не Full / Empty. Повна матриця досяжності:

Метод	`Ok`	`Full` / `Empty`	`Timeout`	`Closed`
`trySend`	✓	`Full` ✓	—	✓
`send`	✓	—	—	✓
`sendTimeout`	✓	—	✓	✓
`tryRecv`	✓	`Empty` ✓	—	✓
`recv`	✓	—	—	✓
`recvTimeout`	✓	—	✓	✓

Перевірки isFull() / isEmpty() на результаті блокувального виклику — мертвий код. У прикладах із match нижче ці гілки залишено з коментарями unreachable, щоб асиметрію було очевидно видно.

Диспетчеризація результату

RecvResult і SendResult містять дискримінант status() і (лише RecvResult) корисне навантаження. Є дві рівноцінні ідіоми:

php

use OxPHP\Shared\Channel;
use OxPHP\Shared\Channel\RecvStatus;

$ch = new Channel(capacity: 64);

// Boolean accessors — terse when only one outcome matters.
$result = $ch->tryRecv();
if ($result->isOk()) {
    process($result->value());
} elseif ($result->isEmpty()) {
    backoff();
} else {
    break; // closed
}

// Exhaustive match — exhaustiveness checking catches new variants.
$r = $ch->recvTimeout(1500);
match ($r->status()) {
    RecvStatus::Ok      => process($r->value()),
    RecvStatus::Timeout => $logger->debug('idle'),
    RecvStatus::Closed  => break,
    RecvStatus::Empty   => /* unreachable: only tryRecv returns Empty */ ,
};

// Symmetric send-side dispatch — only Ok / Timeout / Closed are reachable.
use OxPHP\Shared\Channel\SendStatus;
$s = $ch->sendTimeout($value, 1500);
match ($s->status()) {
    SendStatus::Ok      => /* delivered */ ,
    SendStatus::Timeout => $logger->debug('backpressure'),
    SendStatus::Closed  => break,
    SendStatus::Full    => /* unreachable: only trySend returns Full */ ,
};

// Safe accessor — never throws.
$value = $ch->tryRecv()->valueOr('fallback');

RecvResult::value() кидає OxPHP\Shared\SharedException, якщо його викликати на варіанті, відмінному від Ok, — спершу використовуйте isOk() / valueOr() / status(). Так зроблено навмисно: помилка роботи з відсутнім навантаженням проявляється як гучна відмова, а не як тихий null.

Поведінка у файбері та поза ним

Ті самі виклики методів поводяться по-різному залежно від того, чи виконується PHP у цей момент усередині файбера:

Усередині файбера (режим воркера + oxphp_async(...)): блокувальні варіанти (send, recv, sendTimeout, recvTimeout) створюють синтетичний проміс, реєструють у каналі waker і призупиняють файбер. Потік воркера повертається до планувальника й обробляє інші файбери, доки канал не сповістить waker.
Поза файбером (традиційний режим або неасинхронний шлях виклику): блокувальні варіанти блокують потік воркера ОС через crossbeam_channel. Доки виклик не повернеться, жодна інша робота в цьому потоці не виконується.

Традиційний режим теж отримує семантику каналу, просто платить за неї заблокованим потоком. Для будь-якого конвеєра, що покладається на очікування, рекомендовано розгортання в режимі воркера.

php

<?php
// Traditional mode: this recvTimeout blocks the worker thread for up to 2 seconds.
$ch = new OxPHP\Shared\Channel(16);
$r = $ch->recvTimeout(2000);

// Worker mode: wrap in oxphp_async and recv suspends cooperatively.
oxphp_worker(function () use ($ch) {
    $consumer = oxphp_async(function () use ($ch) {
        for (;;) {
            $r = $ch->recvTimeout(5000);
            if ($r->isOk()) { process($r->value()); continue; }
            if ($r->isClosed()) { break; }
            // Timeout — keep waiting.
        }
    });
    oxphp_async_await($consumer);
});

Скасування файбера

Скасування файбера (переривання запиту, вичерпання дедлайну на рівні SAPI) і далі проявляється як OxPHP\Async\AsyncException. Воно не перетворюється на RecvResult::Closed. Канал не закрився — вас скасувало середовище виконання. Прокиньте виняток далі або перехопіть його, як доречно.

Семантика закриття

close() ідемпотентний; повторний виклик нічого не робить. Після закриття:

send, sendTimeout, trySend повертають SendResult::Closed.
sendMany повертає кількість елементів, фактично прийнятих до закриття (0, якщо канал уже закритий).
recv, recvTimeout, tryRecv і далі вичерпують буферизовані елементи як RecvResult::Ok($value), а коли буфер порожній, повертають RecvResult::Closed.
recvMany повертає ті елементи, які вдається вичерпати, можливо — порожній масив.
isClosed() повертає true.
Заблоковані відправники прокидаються з SendResult::Closed; заблоковані отримувачі — з RecvResult::Closed.

php

<?php
$ch = new OxPHP\Shared\Channel(4);
$ch->send('one');
$ch->send('two');
$ch->close();

// Drain leftovers.
for (;;) {
    $r = $ch->recv();
    if ($r->isClosed()) { break; }
    echo $r->value(), "\n"; // one, two
}

// Further sends report Closed without throwing.
$result = $ch->send('three');
assert($result->isClosed());

Схема коректного завершення конвеєра така: виробники зупиняються, сторона виробника викликає close(), споживачі вичерпують канал у циклі for (;;) { $r = $ch->recv(); if ($r->isClosed()) break; … } і природно виходять. Жоден споживач помилково не побачить навантаження null — цей сигнал несе варіант результату.

Вичерпання під час завершення роботи

Коли процес OxPHP завершується, реєстр OxPHP\Shared викликає close() для кожного запису, зокрема для каналів. З погляду PHP це виглядає так само, як явний close():

Заблоковані виклики recv* повертають RecvResult::Closed.
Заблоковані виклики send* повертають SendResult::Closed.

Завжди перевіряйте варіант результату

Код, який вважає, що recv()->value() завжди безпечний, кине SharedException під час завершення роботи або щоразу, коли інший власник закриє канал.

Пакетні операції

sendMany і recvMany існують для конвеєрів, що переміщують елементи групами. Надавайте їм перевагу, якщо регулярно обробляєте 10+ елементів за раз: кожен пакет — це один FFI-перехід замість N, що відчутно зменшує накладні витрати на елемент у циклах, обмежених пропускною здатністю.

php

<?php
$ch = new OxPHP\Shared\Channel(1024);

// Send an array; returns the count actually accepted.
$sent = $ch->sendMany([1, 2, 3, 4, 5], 100);  // 5 within 100ms budget

// Drain up to 10 items with a 100ms deadline.
$batch = $ch->recvMany(10, 100);

На що варто звернути увагу:

Обидва пакетні методи повертають часткові результати в разі тайм-ауту або закриття посеред пакета — ніколи не виняток. Щоб виявити частковий прогрес, перевіряйте ціле число / довжину масиву.
sendMany на вже закритому каналі повертає 0.
recvMany на закритому й порожньому каналі повертає [].
$ms має бути > 0. Перевантаження «вичерпати все, що є в буфері» немає — якщо такий сценарій важливий, викликайте tryRecv() у щільному циклі; окремий пакетний варіант ми додамо, коли його попросить реальне місце виклику.

Винятки

Виняток	Коли виникає
`CapacityException`	`new Channel($capacity)`, коли масив слотів перевищив би `SHARED_MAX_CHANNEL_BYTES` (типово 64 MiB, ~2 млн слотів).
`ValueTooLargeException`	`send` / `trySend` / `sendTimeout` / `sendMany`, коли серіалізований розмір значення перевищує `SHARED_MAX_VALUE_SIZE` (типово 1 MiB).
`TypeException`	Недодатна `$capacity`; `$ms`, що дорівнює нулю, від’ємне, неціле або відсутнє; або надсилання значення, яке не є скаляром, `null` чи вкладеним масивом shareable-значень.
`SharedException`	`RecvResult::value()` на варіанті, відмінному від Ok, або `clone $channel`.
`AsyncException`	Блокувальний `recv` / `send`, перерваний скасуванням файбера; проявляється як `OxPHP\Async\AsyncException`, а не як варіант Result.

Спостережуваність

Внутрішній сервер показує канали в загальних ендпойнтах спільного реєстру. Він запускається, лише коли задано INTERNAL_ADDR (наприклад, INTERNAL_ADDR=127.0.0.1:9090); без нього недоступні ні ці ендпойнти, ні /metrics нижче:

GET /__ox_shared/summary містить кошик Channel із полями count, bytes і ops.
GET /__ox_shared/entries перелічує записи реєстру з їхніми ID (приймає параметр запиту limit).
GET /__ox_shared/entry?id=<id> повертає стан окремого каналу: capacity, count, closed, senders_blocked, receivers_blocked.

Лічильник ops (і загальнореєстровий oxphp_shared_operations_total{type="Channel"}) рахує кожну спробу recv незалежно від результату — успішне отримання, порожній/закритий канал і recvTimeout, що завершився тайм-аутом, збільшують його однаково. Метрика відстежує звернення до каналу, а не лише успішно передані значення.

Експорт для Prometheus на /metrics:

text

oxphp_shared_channel_count{channel_id="<id>"}               gauge
oxphp_shared_channel_senders_blocked{channel_id="<id>"}     gauge
oxphp_shared_channel_receivers_blocked{channel_id="<id>"}   gauge
oxphp_shared_channel_items_sent_total{channel_id="<id>"}    counter
oxphp_shared_channel_items_dropped_total{channel_id="<id>"} counter

items_dropped_total збільшується на «хвіст» часткового sendMany, який не вмістився.

Типові сценарії

HTTP-виробник, асинхронний споживач

php

<?php
$work = new OxPHP\Shared\Channel(256);

$consumer = oxphp_async(function () use ($work) {
    for (;;) {
        $r = $work->recvTimeout(30000);
        if ($r->isOk()) { process_job($r->value()); continue; }
        if ($r->isClosed()) { break; }
        // Timeout — health-check, then keep waiting.
    }
});

oxphp_worker(function () use ($work) {
    $work->send(['url' => $_POST['url'], 'tries' => 3]);
    echo "queued";
});

Fan-out на кількох споживачів

Запустіть N асинхронних споживачів на одному каналі; реєстр гарантує, що кожен елемент отримає рівно один із них.

php

<?php
$ch = new OxPHP\Shared\Channel(1024);

for ($i = 0; $i < 4; $i++) {
    oxphp_async(function () use ($ch, $i) {
        for (;;) {
            $r = $ch->recvTimeout(60000);
            if ($r->isOk()) { handle($i, $r->value()); continue; }
            if ($r->isClosed()) { break; }
        }
    });
}

Обмежений конвеєр зі зворотним тиском

trySend разом із лічильником відкинутих елементів дає змогу виробникові скидати навантаження замість блокування під час перевантаження:

php

<?php
if ($ch->trySend($event)->isFull()) {
    increment_dropped_metric();
}

Підводні камені

recvTimeout(0) — це TypeException, а не «неблокувальний виклик». Для неблокувального отримання використовуйте tryRecv(). Окремі назви методів усунули неоднозначність перевантаження тайм-ауту, яка була в старому ?float $timeout.
Значення мають бути shareable. Дозволено скаляри, null і вкладені масиви shareable-значень. Передавання об’єкта, який не є екземпляром Shared\*, під час надсилання кидає TypeException.
Клонування заборонено. clone $channel кидає SharedException; передавайте канал через use замикання — oxphp_async(function () use ($ch) { ... }), — щоб обидві сторони бачили той самий запис реєстру.
value() на варіанті, відмінному від Ok, кидає виняток. Це не вада, а можливість: помилка «забув перевірити isOk» стає гучною відмовою. Якщо значення за замовчуванням справді підходить, використовуйте valueOr($default).
Скасування файбера поширюється як виняток, а не як варіант Result. recv(), перерваний скасуванням запиту, кидає OxPHP\Async\AsyncException. Канали, що закриваються самі, і далі повертають RecvResult::Closed.
Скасовані очікувачі з навантаженням у дорозі. Якщо багато файберів чекають на send / recv і їх скасовують саме тоді, коли навантаження мало перейти, воно може лишатися в пам’яті до наступного пробудження. Тримайте кількість очікувачів обмеженою (наприклад, обмежте паралелізм за допомогою Shared\Counter або семафора на основі місткості каналу).

Міграція з попереднього API

Було	Стало
`$ch->tryRecv()` → `null`, якщо порожньо, кидає `ClosedException`	`$ch->tryRecv()` → `RecvResult::Empty` / `Closed`
`$ch->recv($secs)` → `null` у разі тайм-ауту/закриття	`$ch->recv()` (назавжди) / `$ch->recvTimeout($ms)` → `RecvResult`
`$ch->trySend($v): bool`	`$ch->trySend($v): SendResult`
`$ch->send($v, $secs): bool` (кидає TimeoutException/ClosedException)	`$ch->send($v)` / `$ch->sendTimeout($v, $ms)` → `SendResult`
`$ch->sendMany($vs, $secs)` (кидає TimeoutException у разі часткового надсилання)	`$ch->sendMany($vs, $ms): int` (часткова кількість, без винятку)
`?float $timeout` (секунди, NaN/INF допускалися)	`int $ms` (мілісекунди, має бути `> 0`)

Промпти для вашого агента

Ці промпти розраховані на агента з підключеним MCP-сервером документації OxPHP — потрібні сторінки він прочитає сам.

prompt

Використай OxPHP Shared Channel у моєму коді замість нинішньої черги завдань на одному хості — таблиці в базі даних, списку Redis або циклу опитування. Прочитай сторінку `shared-state/shared-channel` через MCP-сервер документації OxPHP.
Знайди в цьому репозиторії виробника і споживача, які так передають роботу, і покажи мені заміну на Channel разом з `oxphp_async` для обох сторін.

Пов’язані матеріали

Режим воркера — передумова для блокувальних варіантів, що призупиняють файбер.
Асинхронні проміси — замикання oxphp_async() є звичайним способом передати Channel фоновому файберу.
Мультиплексування файберів — пояснює, як призупинення дає змогу потоку воркера лишатися продуктивним, поки операції з каналом очікують.