mirror of
https://github.com/p1neappleXpress/OpenFlux.git
synced 2026-10-20 06:11:02 +08:00
acac97b07d
* phpbox: survive any host's limits and keep streams alive across handovers Found on a local emulation of a free host (Apache + PHP-FPM, 60 s CPU per request, the host's disabled functions, a gateway that gives up on the request): - disabled set_time_limit/ignore_user_abort/getmypid/getenv/ini_set ended the node at once under PHP 8 (a disabled function is undefined): guarded everywhere; ping lists what the host took away - time budget per generation on both clocks (CPU on Linux, wall on Windows/macOS, RLIMIT_CPU), handover early when CPU runs short, and learning: a generation that died early leaves its age in host.json and the next ones plan below it; on a host not yet proven, handovers start at 40 s and grow as generations end well - cups: the carrier slept 18 ms per message inside the loop (a 64 KB frame ~0.4 s), so under a download new connections timed out and acks waited; now a paced outbox in the mux: control frames first, streams in turn, whole packets per message, at most ~72 KB/s (the server drops a member that sends faster) - two generations in one room: each now ignores the other's messages (and its own echoes, before decoding them), and messages carry whole packets, so a packet cut across messages is never mixed with another member's bytes; the Go cupsonline receiver keeps one reassembly buffer per member - a successor takes streams only after its link stayed up 8 s (a Mail.ru document drops new connections right after they join, and streams handed over then died) - a repeated OPEN is answered again instead of redialled; the client asks again after 4 s without an answer - the page restarts a node that ended without Stop while it is open - --mode=stream without --inbound is SOCKS5 on every OS again (macOS took utun) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * phpbox: send to cups only as fast as the server confirms; release rebuilds an existing tag - the cups server shows every member's message back to it; our own coming back is its confirmation. At most 6 unconfirmed messages: past that the server is behind, and the client's OPENs and its pings queued behind our echoes (close 3012 'no pong' under a download). The link's close code and reason go to the log. - release.yml: a moved tag replaces the release's assets and notes, as the apps do. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * phpbox: count the cups server's echoes reliably; learn a host limit only from two deaths - echoes were matched with a regex on the raw line that never matched the server's JSON, so the window waited out its 3 s timeout each time (~6 KB/s under load): a substring test on the member uuid instead; a live room confirms 10 of 10 - one sudden end (a host restart) no longer teaches a limit that shortens every run for a week: two ends at about the same age do Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * phpbox: hand over early and consistently; never reach for a longer run than the host allows The emulated hosts exposed the real cause of the node dying: the 'hold idle to prove a longer run' recorded a lifetime (107s) longer than a host's hidden wall-clock kill (90s), so later generations set their handover past that limit and were killed before they could start a successor - the chain broke with no generation left to carry the lesson. - a generation now hands over at ~2/3 of the known limit, never later than 45s, even on a host that looks generous: a free host's real limit is unknown until a generation dies by it, and that death must happen after a successor exists. Deaths only lower it. - removed the idle hold and the lifetime-growth; a learned wall/CPU limit still lowers the handover, and a wall limit is believed only after two deaths at a similar age - while a successor shares the room, the predecessor caps its reading (OVERLAP_RATE): the document server copies our download to the successor too, and at full speed the client's OPENs to the successor waited behind it Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * CHANGELOG: the node-survival and stream-proxy fixes Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * CHANGELOG: fold the node-survival fixes into 0.3.0 (rebuilt in place) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: p1neappleXpress <a@a.a> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
699 lines
32 KiB
PHP
699 lines
32 KiB
PHP
<?php
|
|
// The carrier-agnostic phpbox stream mux, shared by every exit. A Carrier
|
|
// moves opaque "packets" (each one mux frame) over some link (cups, mailru);
|
|
// the Mux turns them into OPEN/DATA/CLOSE streams and dials the destinations.
|
|
//
|
|
// Everything runs in one non-blocking select loop (no fork, no threads: the
|
|
// free hosts forbid them):
|
|
// - dials are asynchronous, so one slow destination never stalls the others;
|
|
// - writes to a destination are queued when its socket is full (a partial
|
|
// fwrite used to drop the rest of the frame);
|
|
// - the frame buffer is consumed by offset, not re-copied on every frame.
|
|
// Flow control: a client that says so in OPEN ("host:port\0fc") gets windows. The exit
|
|
// reads a destination only while that stream (and all streams together) have less than a
|
|
// window of DATA unacknowledged, and acknowledges what it wrote to destinations. Without
|
|
// it a saturated carrier's queue (a document server's) grows by megabytes and every small
|
|
// frame - a TLS handshake, a CLOSE - waits behind it: downloads starved new connections.
|
|
// Acks carry running totals (uint32), so a lost one is repaired by the next.
|
|
// Hooks ($log, $onTick) let the node report what happens without the mux
|
|
// knowing about files or pages.
|
|
|
|
require_once __DIR__ . '/util.php';
|
|
|
|
interface Carrier
|
|
{
|
|
public function setMux(Mux $m): void;
|
|
public function connect(): bool; // auth + join; false on failure
|
|
public function sockets(): array; // read sockets to select on (its WS)
|
|
public function onReadable($sock): void; // decode link -> $mux->onPacket(pkt) per packet
|
|
public function sendPacket(string $frame): void; // encode + send one mux frame
|
|
}
|
|
|
|
/**
|
|
* A carrier whose server takes one message per interval (cups.online throttles cursor updates). The mux paces it
|
|
* from its own loop instead of the carrier sleeping between messages: a sleep inside the loop stalled every stream
|
|
* (a 64 KB frame was ~22 messages, ~0.4 s asleep), so under a download new connections timed out and acks waited.
|
|
* A message carries whole packets only (2-byte length + one frame each, several per message): two senders in one
|
|
* room - generations handing over - then never cut into each other's packets on the receiving side.
|
|
*/
|
|
interface PacedCarrier extends Carrier
|
|
{
|
|
public function msgCap(): int; // bytes of packets one message holds
|
|
public function interval(): float; // seconds between messages, at least
|
|
public function rate(): int; // packet bytes per second the server forwards (0 = no limit)
|
|
public function unconfirmed(): int; // messages sent that the server has not shown back yet
|
|
public function confirmAll(): void; // forget them (they will not be shown back)
|
|
public function sendPackets(string $packets): void; // send one message now (no waiting)
|
|
}
|
|
|
|
final class Mux
|
|
{
|
|
const OPEN = 1, DATA = 2, CLOSE = 3, OPEN_OK = 4, OPEN_ERR = 5, ACK = 6;
|
|
const MASK = 0xFFFFFFFF;
|
|
const ACK_EVERY = 32768; // ack after this many bytes were written to a destination (or when its queue empties)
|
|
const STALL_AFTER = 20; // seconds blocked on a window with no ack: assume the acks were lost
|
|
const ATTEMPT_TO = 3.0; // seconds to wait for one address of a destination to accept
|
|
const MAX_TRIES = 3; // addresses tried before the client is told the open failed
|
|
// Bytes read from a destination per frame. Measured over Mail.ru: 64 KB frames move 3 parallel 4 MB downloads in
|
|
// 22 s where 16 KB frames need 41 s - the carrier's cost is per message, not per byte. Tunable with $readChunk.
|
|
const READ_CHUNK = 65536;
|
|
const MAX_WBUF = 8388608; // queued bytes per stream before it is cut
|
|
// Paced carriers: DATA waiting for the carrier, per stream and in all, before the mux stops reading destinations.
|
|
const OUT_STREAM_MAX = 49152;
|
|
const OUT_ALL_MAX = 262144;
|
|
// Messages a paced carrier may have at the server unconfirmed. cups.online sends every member's cursors to every
|
|
// member, the sender too: our own message coming back is the server saying it took it. Past a few unconfirmed,
|
|
// the server is behind - and what it has queued for us (the client's OPENs, its pings) waits behind our echoes:
|
|
// it ended the link with "no pong" under a download. Sending only as fast as it confirms keeps that queue short.
|
|
const OUT_WINDOW = 12; // ~0.5 s of the server's work at its ~72 KB/s: enough to keep it busy
|
|
const OUT_CONFIRM_TO = 3.0; // seconds without a confirmation before the window is assumed lost (a reconnect)
|
|
|
|
/** @var null|callable(string,string):void fn($level, $message): error|warn|info|debug */
|
|
public $log = null;
|
|
/** @var null|callable(Mux):mixed called about once a second; false stops the run */
|
|
public $onTick = null;
|
|
/** Destination names are sensitive (like the core's --sensitive): off hides the host. */
|
|
public bool $sensitive = false;
|
|
/** @var null|callable(string):array fn($host): addresses - lets tests steer the dial; default is DNS */
|
|
public $resolver = null;
|
|
|
|
/** false = draining: keep serving the streams we have, ignore new OPENs (a newer generation owns them). */
|
|
/** bytes read from a destination per DATA frame: bigger frames cost fewer carrier messages */
|
|
public int $readChunk = self::READ_CHUNK;
|
|
/** flow-control windows in bytes: per stream, and over all streams; 0 = never honour a client's request for them */
|
|
public int $streamWindow = 262144;
|
|
public int $totalWindow = 524288;
|
|
/** seconds a stream may carry no data either way before the exit closes it (its CLOSE can be lost on the way); 0 = never */
|
|
public int $idleTimeout = 300;
|
|
public bool $accepting = true;
|
|
/**
|
|
* Bytes per second read from destinations, over all streams (0 = no limit). A generation sets it while its successor
|
|
* is in the room: the document server sends every member's messages to every member, the successor gets ours too,
|
|
* and at full speed its queue at the server grew by seconds - the client's OPENs to it waited behind our download.
|
|
*/
|
|
public int $readRate = 0;
|
|
private float $rateSec = 0.0;
|
|
private int $rateUsed = 0;
|
|
/** @var null|callable(int):bool fn($sid): during a handover both generations see the same OPEN; true = this one owns it */
|
|
public $claim = null;
|
|
|
|
/** streams opened / closed / failed to open, bytes client->dst and dst->client */
|
|
public array $stats = ['opened' => 0, 'closed' => 0, 'failed' => 0, 'up' => 0, 'down' => 0];
|
|
|
|
private string $muxBuf = '';
|
|
private int $muxOff = 0;
|
|
private array $socks = []; // stream_id => connected destination socket
|
|
private array $pending = []; // stream_id => ['s' => socket|null, 'until' => float, 'ips' => untried addresses, 'tries', 'port', 'label', 't0']
|
|
private array $wbuf = []; // stream_id => bytes waiting for the socket to accept them
|
|
private array $act = []; // stream_id => last time data moved on it
|
|
private array $fc = []; // stream_id => ['sent','acked','wrote','ackedSent','since'] for streams with flow control
|
|
private array $ackDirty = []; // stream_id => true: bytes written since the last ack
|
|
private int $inAll = 0; // unacknowledged DATA over all streams
|
|
|
|
// Outbox for a paced carrier: control frames (OPEN_OK, OPEN_ERR, ACK) go first; each stream's DATA and CLOSE keep
|
|
// their order in that stream's queue, and streams take turns, so a new connection's handshake is not stuck behind
|
|
// another stream's download.
|
|
private bool $paced = false;
|
|
private array $ctlQ = []; // encoded control frames
|
|
private array $bulkQ = []; // stream_id => list of [type, payload]
|
|
private array $bulkLen = []; // stream_id => DATA bytes queued
|
|
private int $bulkAll = 0;
|
|
private int $rrNext = -1; // the stream served last; the next message starts after it
|
|
private float $nextSlot = 0.0;
|
|
private int $lastUnconfirmed = 0;
|
|
private float $confirmSeen = 0.0;
|
|
|
|
public function __construct(private Carrier $c)
|
|
{
|
|
$c->setMux($this);
|
|
$this->paced = $c instanceof PacedCarrier;
|
|
}
|
|
|
|
/** Frames waiting for a paced carrier (0 for an unpaced one). */
|
|
public function outboxBytes(): int
|
|
{
|
|
return $this->bulkAll + array_sum(array_map('strlen', $this->ctlQ));
|
|
}
|
|
|
|
public function activeStreams(): int
|
|
{
|
|
return count($this->socks) + count($this->pending);
|
|
}
|
|
|
|
private function say(string $lvl, string $msg): void
|
|
{
|
|
if ($this->log) {
|
|
($this->log)($lvl, $msg);
|
|
}
|
|
}
|
|
|
|
private float $nextReconnect = 0.0;
|
|
private int $reconnectFails = 0;
|
|
|
|
/** The carrier's socket is gone (the server closed it, or the link broke): join again, with backoff. */
|
|
private function reconnect(): void
|
|
{
|
|
$why = method_exists($this->c, 'why') ? $this->c->why() : '';
|
|
$this->say('warn', 'link to the carrier is down' . ($why !== '' ? " ($why)" : '') . '; reconnecting'
|
|
. ($this->reconnectFails ? " (attempt " . ($this->reconnectFails + 1) . ')' : ''));
|
|
$ok = false;
|
|
try {
|
|
$ok = $this->c->connect();
|
|
} catch (Throwable $e) {
|
|
$this->say('warn', 'reconnect failed: ' . $e->getMessage());
|
|
}
|
|
if ($ok) {
|
|
$this->reconnectFails = 0;
|
|
$this->nextReconnect = 0.0;
|
|
$this->stats['reconnects'] = ($this->stats['reconnects'] ?? 0) + 1;
|
|
$this->say('info', 'reconnected to the carrier (frames in flight were lost; open streams may stall)');
|
|
} else {
|
|
$this->reconnectFails++;
|
|
$this->nextReconnect = microtime(true) + min(10.0, 0.5 * (2 ** min($this->reconnectFails, 5)));
|
|
}
|
|
}
|
|
|
|
/** run holds the exit in the room for up to $cap seconds, pumping both ways. */
|
|
public function run(int $cap): string
|
|
{
|
|
$start = microtime(true);
|
|
$nextTick = 0.0;
|
|
$lastStats = $this->stats;
|
|
$lastStatsAt = $start;
|
|
$reason = 'cap';
|
|
while (true) {
|
|
$now = microtime(true);
|
|
if ($now - $start >= $cap) {
|
|
break;
|
|
}
|
|
if ($now >= $nextTick) {
|
|
$nextTick = $now + 1.0;
|
|
$this->expirePending($now);
|
|
$this->reapIdle($now);
|
|
if ($this->onTick && ($this->onTick)($this) === false) {
|
|
$reason = 'stopped';
|
|
break;
|
|
}
|
|
if ($now - $lastStatsAt >= 5.0) {
|
|
$dUp = $this->stats['up'] - $lastStats['up'];
|
|
$dDown = $this->stats['down'] - $lastStats['down'];
|
|
if ($dUp || $dDown || $this->activeStreams()) {
|
|
$this->say('debug', sprintf('traffic %ds: %d streams, up %s, down %s',
|
|
(int)($now - $lastStatsAt), $this->activeStreams(), self::human($dUp), self::human($dDown))
|
|
. ($this->paced ? sprintf(', waiting to send %s, unconfirmed %d', self::human($this->outboxBytes()), $this->c->unconfirmed()) : ''));
|
|
}
|
|
$lastStats = $this->stats;
|
|
$lastStatsAt = $now;
|
|
}
|
|
}
|
|
$carrierSocks = $this->c->sockets();
|
|
if (!$carrierSocks && microtime(true) >= $this->nextReconnect) {
|
|
$this->reconnect(); // the link to the room/document is gone: join again
|
|
$carrierSocks = $this->c->sockets();
|
|
}
|
|
$read = $carrierSocks;
|
|
foreach ($this->socks as $sid => $sock) {
|
|
if (!$this->blocked($sid, $now)) { $read[] = $sock; }
|
|
}
|
|
$write = [];
|
|
foreach ($this->pending as $p) {
|
|
if ($p['s']) { $write[] = $p['s']; }
|
|
}
|
|
foreach (array_keys($this->wbuf) as $sid) {
|
|
if (isset($this->socks[$sid])) {
|
|
$write[] = $this->socks[$sid];
|
|
}
|
|
}
|
|
$this->pump(microtime(true));
|
|
$wait = 200000;
|
|
if ($this->ctlQ || $this->bulkQ) { // wake up for the next message slot
|
|
$wait = max(1000, min($wait, (int)(($this->nextSlot - microtime(true)) * 1e6)));
|
|
}
|
|
if (!$read && !$write) {
|
|
usleep(min($wait, 100000));
|
|
continue;
|
|
}
|
|
$e = null;
|
|
if (@stream_select($read, $write, $e, 0, $wait) > 0) {
|
|
foreach ($write as $s) {
|
|
$this->onDstWritable($s);
|
|
}
|
|
foreach ($read as $s) {
|
|
if (in_array($s, $carrierSocks, true)) {
|
|
$this->c->onReadable($s);
|
|
} else {
|
|
$this->onDstReadable($s);
|
|
}
|
|
}
|
|
}
|
|
$this->flushAcks();
|
|
$this->pump(microtime(true));
|
|
}
|
|
// Tell the client which streams end with us, so it reconnects at once instead of waiting for a timeout.
|
|
foreach (array_keys($this->socks + $this->pending) as $sid) {
|
|
$this->sendFrame(self::CLOSE, $sid, '');
|
|
}
|
|
$this->drainOutbox(5.0);
|
|
foreach ($this->socks as $s) {
|
|
@fclose($s);
|
|
}
|
|
foreach ($this->pending as $p) {
|
|
if ($p['s']) { @fclose($p['s']); }
|
|
}
|
|
$this->socks = $this->pending = $this->wbuf = [];
|
|
return $reason;
|
|
}
|
|
|
|
/** onPacket takes reassembled link bytes and drains whole mux frames. */
|
|
public function onPacket(string $pkt): void
|
|
{
|
|
$this->muxBuf .= $pkt;
|
|
$n = strlen($this->muxBuf);
|
|
while ($n - $this->muxOff >= 9) {
|
|
$h = unpack('Ctype/Nsid/Nlen', substr($this->muxBuf, $this->muxOff, 9));
|
|
if ($n - $this->muxOff < 9 + $h['len']) {
|
|
break;
|
|
}
|
|
$payload = substr($this->muxBuf, $this->muxOff + 9, $h['len']);
|
|
$this->muxOff += 9 + $h['len'];
|
|
$this->applyMux($h['type'], $h['sid'], $payload);
|
|
}
|
|
if ($this->muxOff > 0) { // compact once per packet, not once per frame
|
|
$this->muxBuf = $this->muxOff >= $n ? '' : substr($this->muxBuf, $this->muxOff);
|
|
$this->muxOff = 0;
|
|
}
|
|
}
|
|
|
|
public function sendFrame(int $type, int $sid, string $payload): void
|
|
{
|
|
if (!$this->paced) {
|
|
$this->c->sendPacket(pack('CNN', $type, $sid, strlen($payload)) . $payload);
|
|
return;
|
|
}
|
|
if ($type === self::DATA || $type === self::CLOSE) {
|
|
$this->bulkQ[$sid][] = [$type, $payload];
|
|
$this->bulkLen[$sid] = ($this->bulkLen[$sid] ?? 0) + strlen($payload);
|
|
$this->bulkAll += strlen($payload);
|
|
} else {
|
|
$this->ctlQ[] = pack('CNN', $type, $sid, strlen($payload)) . $payload;
|
|
}
|
|
}
|
|
|
|
/** Send the next message if its slot has come (paced carriers): one message per interval, never a sleep. */
|
|
private function pump(float $now): void
|
|
{
|
|
if (!$this->paced || (!$this->ctlQ && !$this->bulkQ) || $now < $this->nextSlot) {
|
|
return;
|
|
}
|
|
$u = $this->c->unconfirmed();
|
|
if ($u < $this->lastUnconfirmed || $u === 0) { $this->confirmSeen = $now; } // the server is taking them
|
|
$this->lastUnconfirmed = $u;
|
|
if ($u >= self::OUT_WINDOW) {
|
|
if ($now - $this->confirmSeen < self::OUT_CONFIRM_TO) {
|
|
return; // wait for the server to catch up
|
|
}
|
|
$this->c->confirmAll(); // none came back for long: they are not coming (a reconnect)
|
|
$this->confirmSeen = $now;
|
|
}
|
|
$msg = $this->buildMessage($this->c->msgCap());
|
|
if ($msg === '') {
|
|
return;
|
|
}
|
|
$this->c->sendPackets($msg);
|
|
$rate = $this->c->rate();
|
|
$this->nextSlot = max($this->nextSlot, $now) + max($this->c->interval(), $rate > 0 ? strlen($msg) / $rate : 0.0);
|
|
}
|
|
|
|
/** Whole packets for one message: control frames first, then the streams in turn (DATA cut to fit). */
|
|
private function buildMessage(int $cap): string
|
|
{
|
|
$buf = '';
|
|
while ($this->ctlQ && strlen($buf) + 2 + strlen($this->ctlQ[0]) <= $cap) {
|
|
$f = array_shift($this->ctlQ);
|
|
$buf .= pack('n', strlen($f)) . $f;
|
|
}
|
|
$sids = array_keys($this->bulkQ);
|
|
if (!$sids) {
|
|
return $buf;
|
|
}
|
|
sort($sids);
|
|
$start = 0; // the first stream after the one served last
|
|
foreach ($sids as $i => $sid) {
|
|
if ($sid > $this->rrNext) { $start = $i; break; }
|
|
}
|
|
$order = array_merge(array_slice($sids, $start), array_slice($sids, 0, $start));
|
|
foreach ($order as $sid) {
|
|
while (isset($this->bulkQ[$sid])) {
|
|
$room = $cap - strlen($buf) - 2 - 9;
|
|
[$type, $payload] = $this->bulkQ[$sid][0];
|
|
if ($type === self::CLOSE) {
|
|
if ($room < 0) { break 2; }
|
|
$buf .= pack('n', 9) . pack('CNN', self::CLOSE, $sid, 0);
|
|
$this->shiftBulk($sid);
|
|
continue;
|
|
}
|
|
if ($room < 256 && strlen($payload) > $room) { break 2; } // too little left to be worth a cut
|
|
$part = strlen($payload) > $room ? substr($payload, 0, $room) : $payload;
|
|
$buf .= pack('n', 9 + strlen($part)) . pack('CNN', self::DATA, $sid, strlen($part)) . $part;
|
|
$this->bulkLen[$sid] -= strlen($part);
|
|
$this->bulkAll -= strlen($part);
|
|
if (strlen($part) < strlen($payload)) {
|
|
$this->bulkQ[$sid][0][1] = substr($payload, strlen($part));
|
|
$this->rrNext = $sid;
|
|
break 2; // the message is full
|
|
}
|
|
$this->shiftBulk($sid);
|
|
}
|
|
$this->rrNext = $sid;
|
|
}
|
|
return $buf;
|
|
}
|
|
|
|
private function shiftBulk(int $sid): void
|
|
{
|
|
array_shift($this->bulkQ[$sid]);
|
|
if (!$this->bulkQ[$sid]) {
|
|
unset($this->bulkQ[$sid], $this->bulkLen[$sid]);
|
|
}
|
|
}
|
|
|
|
/** At the end of a run: give what is queued a few seconds to go out. */
|
|
private function drainOutbox(float $limit): void
|
|
{
|
|
$until = microtime(true) + $limit;
|
|
while (($this->ctlQ || $this->bulkQ) && microtime(true) < $until) {
|
|
$this->pump(microtime(true));
|
|
usleep(max(1000, (int)(($this->nextSlot - microtime(true)) * 1e6)));
|
|
}
|
|
}
|
|
|
|
private function label(string $host, int $port): string
|
|
{
|
|
return $this->sensitive ? "$host:$port" : "*:$port";
|
|
}
|
|
|
|
private function applyMux(int $type, int $sid, string $payload): void
|
|
{
|
|
if ($type === self::OPEN) {
|
|
// The same OPEN again: the client did not hear our answer (lost while a carrier reconnected) and asks
|
|
// again. Ids are never reused by a client, so this is that stream: answer again, do not dial twice.
|
|
if (isset($this->socks[$sid])) {
|
|
$this->sendFrame(self::OPEN_OK, $sid, isset($this->fc[$sid]) ? 'fc' : '');
|
|
return;
|
|
}
|
|
if (isset($this->pending[$sid])) {
|
|
return; // still dialing: the answer is on its way
|
|
}
|
|
if (!$this->accepting) {
|
|
$this->say('debug', "stream $sid left to the newer generation");
|
|
return;
|
|
}
|
|
if ($this->claim && !($this->claim)($sid)) {
|
|
$this->say('debug', "stream $sid taken by the other generation");
|
|
return;
|
|
}
|
|
[$target, $caps] = array_pad(explode("\0", $payload, 2), 2, '');
|
|
[$host, $port] = array_pad(explode(':', $target, 2), 2, '');
|
|
$port = (int)$port;
|
|
$wantFc = $this->streamWindow > 0 && in_array('fc', explode(',', $caps), true);
|
|
$label = $this->label($host, $port);
|
|
if (!in_array($port, [80, 443], true) && !PhpboxUtil::testMode()) {
|
|
$this->say('debug', "stream $sid refused $label: port");
|
|
$this->stats['failed']++;
|
|
$this->sendFrame(self::OPEN_ERR, $sid, 'port');
|
|
return;
|
|
}
|
|
$ips = $this->resolver ? array_values(($this->resolver)($host)) : PhpboxUtil::resolveAll($host);
|
|
if (!$ips || PhpboxUtil::isPrivate($host)) {
|
|
$this->say('debug', "stream $sid refused $label: " . (!$ips ? 'no such host' : 'blocked'));
|
|
$this->stats['failed']++;
|
|
$this->sendFrame(self::OPEN_ERR, $sid, !$ips ? 'dns' : 'blocked');
|
|
return;
|
|
}
|
|
$this->pending[$sid] = ['s' => null, 'ips' => $ips, 'port' => $port, 'tries' => 0, 'label' => $label, 't0' => microtime(true), 'fc' => $wantFc];
|
|
$this->dial($sid);
|
|
} elseif ($type === self::DATA) {
|
|
if (isset($this->socks[$sid])) {
|
|
$this->write($sid, $payload);
|
|
} elseif (isset($this->pending[$sid])) {
|
|
$this->stats['up'] += strlen($payload);
|
|
$this->wbuf[$sid] = ($this->wbuf[$sid] ?? '') . $payload; // sent when the dial completes
|
|
}
|
|
} elseif ($type === self::ACK) {
|
|
if (isset($this->fc[$sid]) && strlen($payload) === 4) {
|
|
$v = unpack('N', $payload)[1];
|
|
$f = &$this->fc[$sid];
|
|
$delta = ($v - $f['acked']) & self::MASK;
|
|
if ($delta !== 0 && $delta <= (($f['sent'] - $f['acked']) & self::MASK)) { // never beyond what was sent
|
|
$f['acked'] = $v;
|
|
$f['since'] = 0.0;
|
|
$this->inAll -= $delta;
|
|
}
|
|
unset($f);
|
|
}
|
|
} elseif ($type === self::CLOSE) {
|
|
if (isset($this->socks[$sid]) || isset($this->pending[$sid])) {
|
|
$this->closeStream($sid, false);
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Start (or restart on the next address) the connect for a pending stream. */
|
|
private function dial(int $sid): void
|
|
{
|
|
$p = &$this->pending[$sid];
|
|
while ($p['ips']) {
|
|
$ip = array_shift($p['ips']);
|
|
$p['tries']++;
|
|
$addr = str_contains($ip, ':') ? "[$ip]" : $ip;
|
|
$s = @stream_socket_client("tcp://$addr:{$p['port']}", $en, $es, 0, STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT);
|
|
if ($s) {
|
|
stream_set_blocking($s, false);
|
|
$p['s'] = $s;
|
|
$p['until'] = microtime(true) + self::ATTEMPT_TO;
|
|
$this->say('debug', "stream $sid dialing {$p['label']}" . ($p['tries'] > 1 ? " (address {$p['tries']})" : ''));
|
|
return;
|
|
}
|
|
$this->say('debug', "stream $sid dial {$p['label']} failed at once: $es");
|
|
if ($p['tries'] >= self::MAX_TRIES) { break; }
|
|
}
|
|
$label = $p['label'];
|
|
unset($p);
|
|
unset($this->pending[$sid], $this->wbuf[$sid]);
|
|
$this->stats['failed']++;
|
|
$this->sendFrame(self::OPEN_ERR, $sid, 'refused');
|
|
}
|
|
|
|
/** The current address did not work: try the next one, or tell the client the open failed. */
|
|
private function redial(int $sid, string $why): void
|
|
{
|
|
$p = $this->pending[$sid];
|
|
if ($p['s']) { @fclose($p['s']); }
|
|
$this->pending[$sid]['s'] = null;
|
|
if ($p['ips'] && $p['tries'] < self::MAX_TRIES) {
|
|
$this->say('debug', "stream $sid {$p['label']} $why; trying the next address");
|
|
$this->dial($sid);
|
|
return;
|
|
}
|
|
unset($this->pending[$sid], $this->wbuf[$sid]);
|
|
$this->stats['failed']++;
|
|
$this->say('debug', "stream $sid dial {$p['label']} $why");
|
|
$this->sendFrame(self::OPEN_ERR, $sid, $why === 'timed out' ? 'timeout' : 'refused');
|
|
}
|
|
|
|
/** Try to hand bytes to the destination now; queue what does not fit. */
|
|
private function write(int $sid, string $data): void
|
|
{
|
|
$this->stats['up'] += strlen($data);
|
|
$this->act[$sid] = microtime(true);
|
|
if (isset($this->wbuf[$sid])) {
|
|
$this->wbuf[$sid] .= $data; // keep order behind what is already queued
|
|
} else {
|
|
$n = @fwrite($this->socks[$sid], $data);
|
|
if ($n === false) {
|
|
$this->closeStream($sid, true);
|
|
return;
|
|
}
|
|
$this->wrote($sid, $n);
|
|
if ($n < strlen($data)) {
|
|
$this->wbuf[$sid] = substr($data, $n);
|
|
}
|
|
}
|
|
if (isset($this->wbuf[$sid]) && strlen($this->wbuf[$sid]) > self::MAX_WBUF) {
|
|
$this->say('warn', "stream $sid cut: destination not accepting data (" . self::human(strlen($this->wbuf[$sid])) . ' queued)');
|
|
$this->closeStream($sid, true);
|
|
}
|
|
}
|
|
|
|
private function onDstWritable($s): void
|
|
{
|
|
$sid = false;
|
|
foreach ($this->pending as $id => $p) {
|
|
if ($p['s'] === $s) { $sid = $id; break; }
|
|
}
|
|
if ($sid !== false) { // a dial finished: connected or refused
|
|
$p = $this->pending[$sid];
|
|
if (@stream_socket_get_name($s, true) === false) {
|
|
$this->redial($sid, 'refused');
|
|
return;
|
|
}
|
|
unset($this->pending[$sid]);
|
|
$this->socks[$sid] = $s;
|
|
$this->act[$sid] = microtime(true);
|
|
if (!empty($p['fc'])) {
|
|
$this->fc[$sid] = ['sent' => 0, 'acked' => 0, 'wrote' => 0, 'ackedSent' => 0, 'since' => 0.0];
|
|
}
|
|
$this->stats['opened']++;
|
|
$this->say('debug', sprintf('stream %d open %s (%d ms)', $sid, $p['label'], (int)((microtime(true) - $p['t0']) * 1000)));
|
|
$this->sendFrame(self::OPEN_OK, $sid, isset($this->fc[$sid]) ? 'fc' : '');
|
|
} else {
|
|
$sid = array_search($s, $this->socks, true);
|
|
if ($sid === false) { return; }
|
|
}
|
|
if (isset($this->wbuf[$sid])) { // flush what was queued
|
|
$n = @fwrite($s, $this->wbuf[$sid]);
|
|
if ($n === false) {
|
|
$this->closeStream($sid, true);
|
|
} elseif ($n >= strlen($this->wbuf[$sid])) {
|
|
unset($this->wbuf[$sid]);
|
|
$this->wrote($sid, $n);
|
|
} elseif ($n > 0) {
|
|
$this->wbuf[$sid] = substr($this->wbuf[$sid], $n);
|
|
$this->wrote($sid, $n);
|
|
}
|
|
}
|
|
}
|
|
|
|
private function onDstReadable($s): void
|
|
{
|
|
$sid = array_search($s, $this->socks, true);
|
|
if ($sid === false) { return; }
|
|
$d = @fread($s, $this->readChunk);
|
|
if ($d === '' || $d === false) {
|
|
if ($d === false || feof($s)) { $this->closeStream($sid, true); }
|
|
return;
|
|
}
|
|
$this->stats['down'] += strlen($d);
|
|
$this->rateUsed += strlen($d);
|
|
$this->act[$sid] = microtime(true);
|
|
if (isset($this->fc[$sid])) {
|
|
$this->fc[$sid]['sent'] = ($this->fc[$sid]['sent'] + strlen($d)) & self::MASK;
|
|
$this->inAll += strlen($d);
|
|
}
|
|
$this->sendFrame(self::DATA, $sid, $d);
|
|
}
|
|
|
|
/** True while a stream has a window of unacknowledged DATA out (or all streams together do): stop reading it. */
|
|
private function blocked(int $sid, float $now): bool
|
|
{
|
|
if ($this->paced && (($this->bulkLen[$sid] ?? 0) >= self::OUT_STREAM_MAX || $this->bulkAll >= self::OUT_ALL_MAX)) {
|
|
return true; // the carrier is behind: let it catch up before reading more
|
|
}
|
|
if ($this->readRate > 0) {
|
|
if ($now - $this->rateSec >= 1.0) { $this->rateSec = $now; $this->rateUsed = 0; }
|
|
if ($this->rateUsed >= $this->readRate) { return true; }
|
|
}
|
|
if (!isset($this->fc[$sid])) {
|
|
return false;
|
|
}
|
|
$f = &$this->fc[$sid];
|
|
$inflight = ($f['sent'] - $f['acked']) & self::MASK;
|
|
if ($inflight === 0 || ($inflight < $this->streamWindow && $this->inAll < $this->totalWindow)) {
|
|
$f['since'] = 0.0;
|
|
return false; // a stream with nothing in flight may always read
|
|
}
|
|
if ($f['since'] === 0.0) {
|
|
$f['since'] = $now;
|
|
} elseif ($now - $f['since'] > self::STALL_AFTER) {
|
|
$this->say('warn', "stream $sid: no ack for " . self::STALL_AFTER . "s with " . self::human($inflight) . ' in flight; assuming they were lost');
|
|
$this->inAll -= $inflight;
|
|
$f['acked'] = $f['sent'];
|
|
$f['since'] = 0.0;
|
|
return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
/** $n more bytes reached a destination: owe the client an ack for them. */
|
|
private function wrote(int $sid, int $n): void
|
|
{
|
|
if (isset($this->fc[$sid]) && $n > 0) {
|
|
$this->fc[$sid]['wrote'] = ($this->fc[$sid]['wrote'] + $n) & self::MASK;
|
|
$this->ackDirty[$sid] = true;
|
|
}
|
|
}
|
|
|
|
/** Send owed acks: after a real chunk was written, or once a stream's write queue has emptied (keeps the client's pipe full). */
|
|
private function flushAcks(): void
|
|
{
|
|
foreach ($this->ackDirty as $sid => $_) {
|
|
if (!isset($this->fc[$sid])) {
|
|
unset($this->ackDirty[$sid]);
|
|
continue;
|
|
}
|
|
$f = &$this->fc[$sid];
|
|
$due = ($f['wrote'] - $f['ackedSent']) & self::MASK;
|
|
if ($due >= self::ACK_EVERY || ($due > 0 && !isset($this->wbuf[$sid]))) {
|
|
$f['ackedSent'] = $f['wrote'];
|
|
unset($this->ackDirty[$sid]);
|
|
$this->sendFrame(self::ACK, $sid, pack('N', $f['wrote']));
|
|
}
|
|
unset($f);
|
|
}
|
|
}
|
|
|
|
/** Close streams that carried nothing for $idleTimeout s: a lost CLOSE must not leak a socket on the host. */
|
|
private function reapIdle(float $now): void
|
|
{
|
|
if ($this->idleTimeout <= 0) {
|
|
return;
|
|
}
|
|
foreach ($this->socks as $sid => $_) {
|
|
if ($now - ($this->act[$sid] ?? $now) > $this->idleTimeout) {
|
|
$this->say('debug', "stream $sid idle for {$this->idleTimeout}s: closing it");
|
|
$this->closeStream($sid, true);
|
|
}
|
|
}
|
|
}
|
|
|
|
private function expirePending(float $now): void
|
|
{
|
|
foreach ($this->pending as $sid => $p) {
|
|
if ($now >= ($p['until'] ?? 0)) {
|
|
$this->redial($sid, 'timed out');
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Close one stream; tell the client when the destination (not the client) ended it. */
|
|
private function closeStream(int $sid, bool $notify): void
|
|
{
|
|
if (isset($this->socks[$sid])) {
|
|
@fclose($this->socks[$sid]);
|
|
$this->stats['closed']++;
|
|
} elseif (isset($this->pending[$sid]) && $this->pending[$sid]['s']) {
|
|
@fclose($this->pending[$sid]['s']);
|
|
}
|
|
if (isset($this->fc[$sid])) {
|
|
$this->inAll -= ($this->fc[$sid]['sent'] - $this->fc[$sid]['acked']) & self::MASK;
|
|
unset($this->fc[$sid], $this->ackDirty[$sid]);
|
|
}
|
|
unset($this->socks[$sid], $this->pending[$sid], $this->wbuf[$sid], $this->act[$sid]);
|
|
if (!$notify && isset($this->bulkQ[$sid])) { // the client is done with it: what we still hold is for no one
|
|
$this->bulkAll -= $this->bulkLen[$sid] ?? 0;
|
|
unset($this->bulkQ[$sid], $this->bulkLen[$sid]);
|
|
}
|
|
if ($notify) {
|
|
$this->sendFrame(self::CLOSE, $sid, '');
|
|
}
|
|
$this->say('debug', "stream $sid closed" . ($notify ? ' by destination' : ' by client'));
|
|
}
|
|
|
|
public static function human(int $b): string
|
|
{
|
|
if ($b < 1024) { return "$b B"; }
|
|
if ($b < 1048576) { return sprintf('%.1f KB', $b / 1024); }
|
|
return sprintf('%.2f MB', $b / 1048576);
|
|
}
|
|
}
|