Read ahead: a plan that cannot be made leaves the restore to the usual reads; only a whole buffered SFTP packet counts as an answer waiting
This commit is contained in:
@@ -150,7 +150,7 @@ final class SftpRangeFetcher {
|
|||||||
$buffered = $vars['requestBuffer'] ?? null;
|
$buffered = $vars['requestBuffer'] ?? null;
|
||||||
$channels = $vars['channel_buffers'] ?? null;
|
$channels = $vars['channel_buffers'] ?? null;
|
||||||
if ((is_array($buffered) && isset($job->queue[0]) && array_key_exists($job->queue[0][0], $buffered))
|
if ((is_array($buffered) && isset($job->queue[0]) && array_key_exists($job->queue[0][0], $buffered))
|
||||||
|| (is_string($vars['packet_buffer'] ?? null) && $vars['packet_buffer'] !== '')
|
|| (is_string($vars['packet_buffer'] ?? null) && self::wholePacket($vars['packet_buffer']))
|
||||||
|| (is_array($channels) && array_filter($channels) !== [])) {
|
|| (is_array($channels) && array_filter($channels) !== [])) {
|
||||||
$ready[] = $c;
|
$ready[] = $c;
|
||||||
continue;
|
continue;
|
||||||
@@ -173,6 +173,12 @@ final class SftpRangeFetcher {
|
|||||||
return $n === false || $n === 0 ? array_keys($this->jobs) : array_keys($read);
|
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 {
|
private function start(int $c, int $i): void {
|
||||||
[$path, $offset, $length] = $this->ranges[$i];
|
[$path, $offset, $length] = $this->ranges[$i];
|
||||||
$fh = $this->spoolFile();
|
$fh = $this->spoolFile();
|
||||||
|
|||||||
@@ -243,6 +243,7 @@ final class Repository {
|
|||||||
@rmdir($dir);
|
@rmdir($dir);
|
||||||
}
|
}
|
||||||
$plan = new ReadAheadPlan(rtrim($target, '/') . '/.ngb-restore-spool-' . bin2hex(random_bytes(4)));
|
$plan = new ReadAheadPlan(rtrim($target, '/') . '/.ngb-restore-spool-' . bin2hex(random_bytes(4)));
|
||||||
|
try {
|
||||||
foreach ($this->streamEntries($snapshotId) as $entry) {
|
foreach ($this->streamEntries($snapshotId) as $entry) {
|
||||||
if ($prefix !== '' && !str_starts_with($entry['p'], $prefix)) {
|
if ($prefix !== '' && !str_starts_with($entry['p'], $prefix)) {
|
||||||
continue;
|
continue;
|
||||||
@@ -256,6 +257,9 @@ final class Repository {
|
|||||||
$plan->add('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length);
|
$plan->add('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
} catch (\Throwable) {
|
||||||
|
return null; // reading ahead is only faster: the restore itself reads the snapshot again and reports what is wrong
|
||||||
|
}
|
||||||
if (count($plan->ranges()) < 2 || (!is_dir($plan->spool) && !@mkdir($plan->spool, 0700, true))) {
|
if (count($plan->ranges()) < 2 || (!is_dir($plan->spool) && !@mkdir($plan->spool, 0700, true))) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -96,7 +96,7 @@ class SftpRangeFetcherTest extends TestCase {
|
|||||||
$pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
|
$pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
|
||||||
$sockets[] = $pair;
|
$sockets[] = $pair;
|
||||||
$c->fsock = $pair[0];
|
$c->fsock = $pair[0];
|
||||||
$c->packet_buffer = 'more';
|
$c->packet_buffer = pack('N', 4) . 'more'; // a whole packet
|
||||||
}
|
}
|
||||||
$ranges = array_slice($this->ranges(), 0, 3);
|
$ranges = array_slice($this->ranges(), 0, 3);
|
||||||
$start = microtime(true);
|
$start = microtime(true);
|
||||||
|
|||||||
@@ -66,6 +66,15 @@ class ReadAheadTest extends TestCase {
|
|||||||
$this->assertLessThanOrEqual(count($backend->inner->list('packs')) + 1, count($backend->asked));
|
$this->assertLessThanOrEqual(count($backend->inner->list('packs')) + 1, count($backend->asked));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public function testAPlanThatCannotBeMadeLeavesTheRestoreToTheUsualReads(): void {
|
||||||
|
$backend = new ReadAheadBackend(new LocalBackend($this->root), failSnapshotReads: 1);
|
||||||
|
|
||||||
|
$this->assertRestored($this->restore($backend));
|
||||||
|
|
||||||
|
$this->assertSame([], $backend->asked, 'no reading ahead without a plan');
|
||||||
|
$this->assertSame(2, $backend->snapshotReads, 'the restore itself read the snapshot again');
|
||||||
|
}
|
||||||
|
|
||||||
public function testTheEmptySpoolOfAStoppedRestoreIsRemoved(): void {
|
public function testTheEmptySpoolOfAStoppedRestoreIsRemoved(): void {
|
||||||
$target = $this->tempDir();
|
$target = $this->tempDir();
|
||||||
mkdir("$target/.ngb-restore-spool-0badc0de", 0700); // a restore that was killed
|
mkdir("$target/.ngb-restore-spool-0badc0de", 0700); // a restore that was killed
|
||||||
@@ -195,6 +204,8 @@ final class ReadAheadBackend implements IBackend, IReadsAhead {
|
|||||||
private bool $unavailable = false,
|
private bool $unavailable = false,
|
||||||
/** reading a pack the usual way fails */
|
/** reading a pack the usual way fails */
|
||||||
private bool $packsGone = false,
|
private bool $packsGone = false,
|
||||||
|
/** this many reads of a snapshot fail, then they work */
|
||||||
|
private int $failSnapshotReads = 0,
|
||||||
) {
|
) {
|
||||||
$this->counter = new PackOpenCountingBackend($inner);
|
$this->counter = new PackOpenCountingBackend($inner);
|
||||||
}
|
}
|
||||||
@@ -238,7 +249,15 @@ final class ReadAheadBackend implements IBackend, IReadsAhead {
|
|||||||
$this->counter->put($path, $stream);
|
$this->counter->put($path, $stream);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public int $snapshotReads = 0;
|
||||||
|
|
||||||
public function get(string $path) {
|
public function get(string $path) {
|
||||||
|
if (str_starts_with($path, 'snapshots/')) {
|
||||||
|
$this->snapshotReads++;
|
||||||
|
if ($this->snapshotReads <= $this->failSnapshotReads) {
|
||||||
|
throw new BackendException('Connection lost reading ' . $path);
|
||||||
|
}
|
||||||
|
}
|
||||||
if ($this->packsGone && str_starts_with($path, 'packs/')) {
|
if ($this->packsGone && str_starts_with($path, 'packs/')) {
|
||||||
throw new BackendException('Not found: ' . $path);
|
throw new BackendException('Not found: ' . $path);
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user