aiden
/
pixelfed
mirror of https://github.com/pixelfed/pixelfed
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #6966 from pixelfed/debug/emoji-cloud-migration

Browse Source 
perf: parallelise emoji cloud migration with worker processes
pull/6970/head
Shlee 3 weeks ago committed by GitHub
parent b6e8571009 00564ac221
commit 67ae799cb4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 118 additions and 8 deletions
Show Stats Download Patch File Download Diff File

126
app/Console/Commands/Admin/EmojiMoveStorageLocalToCloud.php
Unescape Escape View File

@ -9,6 +9,7 @@ use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Storage;
use Illuminate\Support\Str;
use Symfony\Component\Process\Process;
class EmojiMoveStorageLocalToCloud extends Command
{
@ -21,6 +22,12 @@ class EmojiMoveStorageLocalToCloud extends Command
*/
protected $signature = 'admin:EmojiMoveStorageLocalToCloud
{--limit=0 : Max files to process this run (0 = no limit, process all)}
{--offset=0 : Skip this many files before processing (for manual chunking)}
{--workers=1 : Spawn N parallel worker processes to upload concurrently}
{--stride=1 : Internal: total worker count for strided sharding}
{--shard=0 : Internal: this worker index (0-based) for strided sharding}
{--skip-verify : Do not re-check the cloud copy size after upload (faster)}
{--skip-cloud-check : Do not HEAD the cloud object first; always upload (faster, idempotent)}
{--dry-run : Report what would happen without copying or writing}
{--keep-local : Do not delete local files after verifying the cloud copy}
{--debug : Print detailed diagnostics}
@ -87,6 +94,16 @@ class EmojiMoveStorageLocalToCloud extends Command
$this->line(' file count: '.count($localDisk->files('public/emoji')));
}
$workers = max(1, (int) $this->option('workers'));
// When asked to parallelise, spawn N child processes that each handle a
// strided slice of the file list (worker w processes indexes where
// index % workers === w). This gives real concurrency for the
// I/O-bound S3 uploads without any extra dependencies.
if ($workers > 1) {
return $this->runParallel($workers);
}
if (! $this->option('dry-run') && ! $this->option('force')) {
if (! $this->confirm('Begin migrating local custom emoji to cloud?', true)) {
$this->comment('Aborted.');
@ -96,6 +113,9 @@ class EmojiMoveStorageLocalToCloud extends Command
}
$limit = (int) $this->option('limit');
$offset = max(0, (int) $this->option('offset'));
$stride = max(1, (int) $this->option('stride'));
$shard = max(0, (int) $this->option('shard'));
$moved = 0;
$skipped = 0;
$failed = 0;
@ -107,14 +127,24 @@ class EmojiMoveStorageLocalToCloud extends Command
// so a DB filter on uri would wrongly exclude them.
$files = $localDisk->exists('public/emoji') ? $localDisk->files('public/emoji') : [];
if ($offset > 0) {
$files = array_slice($files, $offset);
}
if ($limit > 0) {
$files = array_slice($files, 0, $limit);
}
$bar = $this->output->createProgressBar(count($files));
$bar->start();
$showBar = $stride === 1 && ! $debug;
$bar = $showBar ? $this->output->createProgressBar(count($files)) : null;
$bar?->start();
foreach ($files as $i => $localPath) {
// Strided sharding for parallel workers: only handle our slice.
if ($stride > 1 && ($i % $stride) !== $shard) {
continue;
}
foreach ($files as $localPath) {
$filename = basename($localPath);
// Preserve dotfiles such as a directory .gitignore, and the
@ -122,7 +152,7 @@ class EmojiMoveStorageLocalToCloud extends Command
// /storage/emoji/missing.png onerror fallback (must stay local).
if (str_starts_with($filename, '.') || $filename === 'missing.png') {
$skipped++;
$bar->advance();
$bar?->advance();
continue;
}
@ -135,10 +165,10 @@ class EmojiMoveStorageLocalToCloud extends Command
'skipped' => $skipped++,
default => $failed++,
};
$bar->advance();
$bar?->advance();
}
$bar->finish();
$bar?->finish();
$this->newLine(2);
if ($moved > 0 && ! $this->option('dry-run')) {
@ -153,12 +183,91 @@ class EmojiMoveStorageLocalToCloud extends Command
return self::SUCCESS;
}
/**
* Spawn N child worker processes, each handling a strided slice of the
* files, and wait for them all to finish.
*/
protected function runParallel(int $workers): int
{
if (! $this->option('force') && ! $this->confirm("Begin migrating local custom emoji to cloud using {$workers} parallel workers?", true)) {
$this->comment('Aborted.');
return self::SUCCESS;
}
$php = PHP_BINARY;
$artisan = base_path('artisan');
// Forwarded flags each worker should inherit.
$forward = ['--force'];
foreach (['dry-run', 'keep-local', 'skip-verify', 'skip-cloud-check', 'debug'] as $flag) {
if ($this->option($flag)) {
$forward[] = '--'.$flag;
}
}
if ((int) $this->option('limit') > 0) {
$forward[] = '--limit='.(int) $this->option('limit');
}
if ((int) $this->option('offset') > 0) {
$forward[] = '--offset='.(int) $this->option('offset');
}
$this->info("Launching {$workers} workers...");
$procs = [];
for ($w = 0; $w < $workers; $w++) {
$cmd = array_merge(
[$php, $artisan, 'admin:EmojiMoveStorageLocalToCloud'],
$forward,
['--stride='.$workers, '--shard='.$w]
);
$process = new Process($cmd);
$process->setTimeout(null);
$process->start();
$procs[$w] = $process;
}
// Stream each worker's output prefixed with its shard id.
while (array_filter($procs, fn ($p) => $p->isRunning())) {
foreach ($procs as $w => $process) {
if ($out = $process->getIncrementalOutput()) {
foreach (explode("\n", rtrim($out, "\n")) as $line) {
if ($line !== '') {
$this->line("[worker {$w}] ".$line);
}
}
}
}
usleep(200000);
}
$failed = 0;
foreach ($procs as $w => $process) {
if (! $process->isSuccessful()) {
$failed++;
$this->error("Worker {$w} exited with code ".$process->getExitCode());
}
}
if (! $this->option('dry-run')) {
Cache::forget('pf:custom_emoji');
}
$this->newLine();
$this->info('All workers finished'.($failed ? " ({$failed} failed)" : '.'));
return $failed ? self::FAILURE : self::SUCCESS;
}
/**
* @return string one of moved|skipped|failed
*/
protected function migrateFile(string $localPath, string $mediaPath, $localDisk, $cloudDisk, bool $debug = false): string
{
if ($cloudDisk->exists($mediaPath)) {
// The upfront cloud HEAD is an extra S3 round-trip per file. Skip it
// with --skip-cloud-check for a faster, always-upload (idempotent) run.
if (! $this->option('skip-cloud-check') && $cloudDisk->exists($mediaPath)) {
if ($debug) {
$this->newLine();
$this->line(' [skip] already on cloud: '.$mediaPath);
@ -185,7 +294,8 @@ class EmojiMoveStorageLocalToCloud extends Command
$size = (int) $localDisk->size($localPath);
$cloudDisk->put($mediaPath, $localDisk->get($localPath), 'public');
if (! $this->verify($localPath, $mediaPath, $localDisk, $cloudDisk)) {
// Verify is another S3 round-trip; skippable with --skip-verify.
if (! $this->option('skip-verify') && ! $this->verify($localPath, $mediaPath, $localDisk, $cloudDisk)) {
$this->warn(PHP_EOL.'Verify failed for '.$mediaPath.'; left local copy intact.');
return 'failed';

Write Preview
Loading…
Cancel
Save