READ size per connection, lowered when a server answers less than asked */ private array $chunks = []; /** @var array the range each busy connection reads */ private array $jobs = []; /** @var array ranges read (or failed) and not yet given to the caller */ private array $done = []; private int $requestId = 0x4e520000; /** * @param array $connections phpseclib 2 connections with their SFTP channel open * @param list $ranges paths below $root */ private function __construct( private array $connections, private string $root, private array $ranges, private string $spool, ) { foreach (array_keys($connections) as $c) { $this->chunks[$c] = self::CHUNK; } } /** * See IReadsAhead::readAhead(). The connections are ended when the generator ends. * * @param list $connections * @param list $ranges * @return \Generator */ public static function fetch(array $connections, string $root, array $ranges, string $spool): \Generator { $fetcher = new self($connections, rtrim($root, '/'), $ranges, $spool); try { yield from $fetcher->run(); } finally { $fetcher->end(); } } /** @return \Generator */ private function run(): \Generator { $count = count($this->ranges); $next = 0; for ($want = 0; $want < $count; $want++) { while (!array_key_exists($want, $this->done)) { foreach (array_keys($this->connections) as $c) { if (!isset($this->jobs[$c]) && $next < $count && $next < $want + self::AHEAD) { $this->start($c, $next++); } } if ($this->jobs === [] && !array_key_exists($want, $this->done)) { // no connection left: the rest the usual way $this->done[$want] = null; $next = max($next, $want + 1); break; } foreach ($this->ready() as $c) { $this->step($c); } } $fh = $this->done[$want]; unset($this->done[$want]); try { yield $want => $fh; } finally { if (is_resource($fh)) { fclose($fh); } } } } /** * The busy connections that have an answer waiting, so that a connection that waits for a * lost packet to be sent again does not hold up the others; all of them when the connections * do not tell (then they are taken in turn). * * @return list */ private function ready(): array { $sockets = []; $ready = []; foreach ($this->jobs as $c => $job) { $sftp = $this->connections[$c]; $vars = get_object_vars($sftp); // only what is public: phpseclib 2's are // Answers phpseclib has read already: the socket does not show them. Its request // buffer holds whole answers read while it waited for another; its packet buffer and // channel buffers hold what came in with the last packet it read (small answers, such // as a CLOSE and the next OPEN, often arrive together). $buffered = $vars['requestBuffer'] ?? null; $channels = $vars['channel_buffers'] ?? null; if ((is_array($buffered) && isset($job->queue[0]) && array_key_exists($job->queue[0][0], $buffered)) || (is_string($vars['packet_buffer'] ?? null) && self::wholePacket($vars['packet_buffer'])) || (is_array($channels) && array_filter($channels) !== [])) { $ready[] = $c; continue; } $socket = $vars['fsock'] ?? null; if (!is_resource($socket)) { return array_keys($this->jobs); } $sockets[$c] = $socket; } if ($ready !== [] || $sockets === []) { return $ready; } $read = $sockets; $write = null; $except = null; // Data in PHP's own buffer of a socket counts as readable. Nothing within the time: // each in turn, where phpseclib's own timeout applies. $n = @stream_select($read, $write, $except, self::SELECT_TIMEOUT); return $n === false || $n === 0 ? array_keys($this->jobs) : array_keys($read); } /** Whether $buffer starts with a whole SFTP packet (length, then that many bytes). */ private static function wholePacket(string $buffer): bool { $length = strlen($buffer) >= 4 ? unpack('N', $buffer) : false; return is_array($length) && strlen($buffer) >= 4 + $length[1]; } private function start(int $c, int $i): void { [$path, $offset, $length] = $this->ranges[$i]; $fh = $this->spoolFile(); if ($fh === null) { $this->done[$i] = null; return; } if ($length <= 0) { $this->done[$i] = $fh; return; } $this->jobs[$c] = new SftpRangeJob($i, $fh, $offset, $offset + $length); try { $remote = $this->connections[$c]->_realpath($this->root . '/' . ltrim($path, '/')); if (!is_string($remote)) { $this->finish($c); return; } $this->send($c, self::FXP_OPEN, pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0), self::OPEN); } catch (\Throwable) { $this->drop($c); } } /** Take the oldest answer on connection $c, then keep its requests in flight. */ private function step(int $c): void { $job = $this->jobs[$c]; [$id, $kind, $offset, $length] = array_shift($job->queue) ?? throw new \LogicException('no request in flight'); $sftp = $this->connections[$c]; try { $response = $sftp->_get_sftp_packet($id); if (!is_string($response)) { throw new BackendException('SFTP connection lost'); } $type = $sftp->packet_type; if ($kind === self::OPEN) { if ($type === self::FXP_HANDLE) { $job->handle = substr($response, 4); } else { $job->failed = true; // missing, or no permission } } elseif ($kind === self::READ) { if (!$job->failed) { $this->received($c, $type, $response, $offset, $length); } } else { $status = $type === self::FXP_STATUS && strlen($response) >= 4 ? unpack('N', $response) : false; $closed = is_array($status) && $status[1] === self::FX_OK; $this->finish($c); if (!$closed) { // handles might pile up on the server: not used again $this->drop($c); } return; } $this->refill($c); } catch (\Throwable) { $this->drop($c); } } private function received(int $c, int $type, string $response, int $offset, int $length): void { $job = $this->jobs[$c]; $declared = strlen($response) >= 4 ? unpack('N', $response) : false; $n = is_array($declared) ? $declared[1] : 0; if ($type !== self::FXP_DATA || $n <= 0 || $n !== strlen($response) - 4 || $n > $length) { // end of file before the end of the range, a server error or a malformed answer $job->failed = true; return; } if (fseek($job->fh, $offset - $job->start) !== 0 || fwrite($job->fh, substr($response, 4)) !== $n) { $job->failed = true; return; } $job->got += $n; if ($n < $length) { // a server that caps a READ: ask for the rest, and no more than it gives from now on $job->redo[] = [$offset + $n, $length - $n]; $this->chunks[$c] = max(self::MIN_CHUNK, min($this->chunks[$c], $n)); } } /** Send READs until DEPTH are in flight; when nothing is left to ask for, close the file. */ private function refill(int $c): void { $job = $this->jobs[$c]; if ($job->handle !== '' && !$job->failed) { while (count($job->queue) < self::DEPTH) { if ($job->redo !== []) { [$offset, $length] = array_shift($job->redo); } elseif ($job->next < $job->end) { $offset = $job->next; $length = min($this->chunks[$c], $job->end - $offset); $job->next += $length; } else { break; } $packet = pack('Na*N3', strlen($job->handle), $job->handle, intdiv($offset, 4294967296), $offset % 4294967296, $length); $this->send($c, self::FXP_READ, $packet, self::READ, $offset, $length); } } if ($job->queue !== []) { return; } if ($job->handle === '') { $this->finish($c); // not opened: nothing to close return; } $this->send($c, self::FXP_CLOSE, pack('Na*', strlen($job->handle), $job->handle), self::CLOSE); } private function send(int $c, int $type, string $packet, int $kind, int $offset = 0, int $length = 0): void { $this->requestId = $this->requestId >= 0x7fffffff ? 0x4e520000 : $this->requestId + 1; if (!$this->connections[$c]->_send_sftp_packet($type, $packet, $this->requestId)) { throw new BackendException('SFTP connection failed while sending a request'); } $this->jobs[$c]->queue[] = [$this->requestId, $kind, $offset, $length]; } /** The range of connection $c is done: its stream for the caller, or null when it failed. */ private function finish(int $c): void { $job = $this->jobs[$c]; unset($this->jobs[$c]); if ($job->failed || $job->got !== $job->end - $job->start || !rewind($job->fh)) { self::closeFile($job->fh); $this->done[$job->i] = null; return; } $this->done[$job->i] = $job->fh; } /** A connection in an unknown state: its range fails, and it is not used again. */ private function drop(int $c): void { if (isset($this->jobs[$c])) { $job = $this->jobs[$c]; unset($this->jobs[$c]); self::closeFile($job->fh); $this->done[$job->i] = null; } self::disconnect($this->connections[$c]); unset($this->connections[$c]); } /** @return resource|null a new file in the spool, already unlinked: nothing is left behind if the process dies */ private function spoolFile() { $path = $this->spool . '/' . bin2hex(random_bytes(8)); $fh = @fopen($path, 'x+b'); if ($fh === false) { return null; } @unlink($path); return $fh; } private function end(): void { foreach ($this->jobs as $job) { self::closeFile($job->fh); } foreach ($this->done as $fh) { if (is_resource($fh)) { fclose($fh); } } $this->jobs = []; $this->done = []; foreach ($this->connections as $sftp) { self::disconnect($sftp); } $this->connections = []; } private static function closeFile(mixed $fh): void { if (is_resource($fh)) { fclose($fh); } } private static function disconnect(object $sftp): void { try { if (is_callable([$sftp, 'disconnect'])) { $sftp->disconnect(); } } catch (\Throwable) { } } }