Skip to content

perf(files_external): pipeline SFTP stream writes #62377

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
susnux merged 1 commit into from
Jul 29, 2026
Merged

perf(files_external): pipeline SFTP stream writes #62377

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions apps/files_external/lib/Lib/Storage/SFTPReflection.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,15 +32,15 @@ trait SFTPReflection {
/**
* Invoke a method that is private in phpseclib v3.
*/
private function invokeSftp(SFTP $sftp, string $method, array $arguments = []): mixed {
protected function invokeSftp(SFTP $sftp, string $method, array $arguments = []): mixed {
self::$sftpReflectionMethods[$method] ??= new \ReflectionMethod(SFTP::class, $method);
return self::$sftpReflectionMethods[$method]->invokeArgs($sftp, $arguments);
}

/**
* Read a property that is private/protected in phpseclib v3.
*/
private function getSftpProperty(SFTP $sftp, string $property): mixed {
protected function getSftpProperty(SFTP $sftp, string $property): mixed {
self::$sftpReflectionProperties[$property] ??= new \ReflectionProperty(SFTP::class, $property);
return self::$sftpReflectionProperties[$property]->getValue($sftp);
}
Expand Down
91 changes: 77 additions & 14 deletions apps/files_external/lib/Lib/Storage/SFTPWriteStream.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,27 @@ class SFTPWriteStream implements File {

private string $path;

/** Payload budget for a single SSH_FXP_WRITE packet */
private int $packetSize = 0;

/** Number of SSH_FXP_WRITE packets sent but not yet acknowledged */
private int $outstanding = 0;

/** Mirrors phpseclib's NET_SFTP_UPLOAD_QUEUE_SIZE */
private const int QUEUE_SIZE = 1024;

/** 2^32, used to split a 64-bit file offset into its high and low 32-bit words */
private const int UINT32_MODULUS = 2 ** 32;

/** Default SFTP payload size (32 KiB) used when the negotiated maximum is unavailable */
private const int DEFAULT_MAX_PACKET_SIZE = 1 << 15;

/** Bytes reserved in an SSH_FXP_WRITE packet for the handle-length, offset and data-length fields */
private const int WRITE_PACKET_OVERHEAD = 25;

/** Length of the 32-bit "string" length prefix preceding the handle in an SSH_FXP_HANDLE response */
private const int STRING_LENGTH_PREFIX = 4;

public static function register($protocol = 'sftpwrite') {
if (in_array($protocol, stream_get_wrappers(), true)) {
return false;
Expand Down Expand Up @@ -77,11 +98,10 @@ public function stream_open($path, $mode, $options, &$opened_path) {
}

$remote_file = $this->sftp->realpath($path);

$this->path = $remote_file;
if ($remote_file === false) {
return false;
}
$this->path = $remote_file;

$packet = pack('Na*N2', strlen($remote_file), $remote_file, NET_SFTP_OPEN_WRITE | NET_SFTP_OPEN_CREATE | NET_SFTP_OPEN_TRUNCATE, 0);
try {
Expand All @@ -93,7 +113,7 @@ public function stream_open($path, $mode, $options, &$opened_path) {
$response = $this->invokeSftp($this->sftp, 'get_sftp_packet');
switch ($this->getSftpProperty($this->sftp, 'packet_type')) {
case NET_SFTP_HANDLE:
$this->handle = substr($response, 4);
$this->handle = substr($response, self::STRING_LENGTH_PREFIX);
break;
case NET_SFTP_STATUS: // presumably SSH_FX_NO_SUCH_FILE or SSH_FX_PERMISSION_DENIED
$this->invokeSftp($this->sftp, 'logError', [$response]);
Expand All @@ -103,6 +123,11 @@ public function stream_open($path, $mode, $options, &$opened_path) {
return false;
}

// Size each SSH_FXP_WRITE to the negotiated maximum packet, leaving room
// for the handle and the packet header (mirrors phpseclib's SFTP::put()).
$maxPacket = $this->getSftpProperty($this->sftp, 'max_sftp_packet') ?: self::DEFAULT_MAX_PACKET_SIZE;
$this->packetSize = max(1, $maxPacket - strlen($this->handle) - self::WRITE_PACKET_OVERHEAD);

return true;
}

Expand All @@ -128,15 +153,55 @@ public function stream_write($data) {

$this->buffer .= $data;

if (strlen($this->buffer) > 64 * 1024) {
if (!$this->stream_flush()) {
// Send full packets as soon as enough data is buffered, without waiting
// for each acknowledgement, so the upload stays pipelined.
while (strlen($this->buffer) >= $this->packetSize) {
if (!$this->sendChunk(substr($this->buffer, 0, $this->packetSize))) {
return false;
}
$this->buffer = substr($this->buffer, $this->packetSize);
}

return $written;
}

/**
* Send a single SSH_FXP_WRITE packet without waiting for its response.
*
* Acknowledgements are only drained once the queue is full or the stream is
* flushed/closed. Blocking on every packet would turn each write into a full
* round trip, which is what made the previous implementation slow.
*/
private function sendChunk(string $chunk): bool {
$size = strlen($chunk);
$packet = pack('Na*N3a*', strlen($this->handle), $this->handle, $this->internalPosition / self::UINT32_MODULUS, $this->internalPosition, $size, $chunk);
try {
$this->invokeSftp($this->sftp, 'send_sftp_packet', [NET_SFTP_WRITE, $packet]);
} catch (\Throwable) {
return false;
}
$this->internalPosition += $size;
$this->outstanding++;

if ($this->outstanding >= self::QUEUE_SIZE) {
return $this->drainResponses();
}

return true;
}

/**
* Read the acknowledgements for all packets sent since the last drain.
*/
private function drainResponses(): bool {
if ($this->outstanding === 0) {
return true;
}
$result = $this->invokeSftp($this->sftp, 'read_put_responses', [$this->outstanding]);
$this->outstanding = 0;
return $result;
}

#[\Override]
public function stream_set_option($option, $arg1, $arg2) {
return false;
Expand All @@ -159,17 +224,15 @@ public function stream_lock($operation) {

#[\Override]
public function stream_flush() {
$size = strlen($this->buffer);
$packet = pack('Na*N3a*', strlen($this->handle), $this->handle, $this->internalPosition / 4294967296, $this->internalPosition, $size, $this->buffer);
try {
$this->invokeSftp($this->sftp, 'send_sftp_packet', [NET_SFTP_WRITE, $packet]);
} catch (\Throwable) {
return false;
// Flush the trailing partial packet, then wait for all outstanding writes.
if ($this->buffer !== '') {
if (!$this->sendChunk($this->buffer)) {
return false;
}
$this->buffer = '';
}
$this->internalPosition += $size;
$this->buffer = '';

return $this->invokeSftp($this->sftp, 'read_put_responses', [1]);
return $this->drainResponses();
}

#[\Override]
Expand Down
Loading
Loading