Parallel migrations (#57)

* parallelize migration-related commands

* Fix code style (php-cs-fixer)

---------

Co-authored-by: PHP CS Fixer <phpcsfixer@example.com>

src/Commands/Migrate.php

@@ -8,16 +8,18 @@ use Illuminate\Contracts\Events\Dispatcher;
 use Illuminate\Database\Console\Migrations\MigrateCommand;
 use Illuminate\Database\Migrations\Migrator;
 use Illuminate\Database\QueryException;
+use Illuminate\Support\LazyCollection;
 use Stancl\Tenancy\Concerns\DealsWithMigrations;
 use Stancl\Tenancy\Concerns\ExtendsLaravelCommand;
 use Stancl\Tenancy\Concerns\HasTenantOptions;
+use Stancl\Tenancy\Concerns\ParallelCommand;
 use Stancl\Tenancy\Database\Exceptions\TenantDatabaseDoesNotExistException;
 use Stancl\Tenancy\Events\DatabaseMigrated;
 use Stancl\Tenancy\Events\MigratingDatabase;
 
 class Migrate extends MigrateCommand
 {
-    use HasTenantOptions, DealsWithMigrations, ExtendsLaravelCommand;
+    use HasTenantOptions, DealsWithMigrations, ExtendsLaravelCommand, ParallelCommand;
 
     protected $description = 'Run migrations for tenant(s)';
 
@@ -31,6 +33,7 @@ class Migrate extends MigrateCommand
         parent::__construct($migrator, $dispatcher);
 
         $this->addOption('skip-failing', description: 'Continue execution if migration fails for a tenant');
+        $this->addProcessesOption();
 
         $this->specifyParameters();
     }
@@ -47,26 +50,50 @@ class Migrate extends MigrateCommand
             return 1;
         }
 
-        foreach ($this->getTenants() as $tenant) {
+        if ($this->getProcesses() > 1) {
+            return $this->runConcurrently($this->getTenantChunks()->map(function ($chunk) {
+                return $this->getTenants(array_values($chunk->all()));
+            }));
+        }
+
+        return $this->migrateTenants($this->getTenants()) ? 0 : 1;
+    }
+
+    protected function childHandle(...$args): bool
+    {
+        $chunk = $args[0];
+
+        return $this->migrateTenants($chunk);
+    }
+
+    protected function migrateTenants(LazyCollection $tenants): bool
+    {
+        $success = true;
+
+        foreach ($tenants as $tenant) {
             try {
                 $this->components->info("Migrating tenant {$tenant->getTenantKey()}");
 
-                $tenant->run(function ($tenant) {
+                $tenant->run(function ($tenant) use (&$success) {
                     event(new MigratingDatabase($tenant));
+
                     // Migrate
-                    parent::handle();
+                    if (parent::handle() !== 0) {
+                        $success = false;
+                    }
 
                     event(new DatabaseMigrated($tenant));
                 });
             } catch (TenantDatabaseDoesNotExistException|QueryException $e) {
+                $this->components->error("Migration failed for tenant {$tenant->getTenantKey()}: {$e->getMessage()}");
+                $success = false;
+
                 if (! $this->option('skip-failing')) {
                     throw $e;
                 }
-
-                $this->components->warn("Migration failed for tenant {$tenant->getTenantKey()}: {$e->getMessage()}");
             }
         }
 
-        return 0;
+        return $success;
     }
 }

src/Commands/MigrateFresh.php

@@ -5,13 +5,19 @@ declare(strict_types=1);
 namespace Stancl\Tenancy\Commands;
 
 use Illuminate\Database\Console\Migrations\BaseCommand;
+use Illuminate\Database\QueryException;
+use Illuminate\Support\LazyCollection;
 use Stancl\Tenancy\Concerns\DealsWithMigrations;
 use Stancl\Tenancy\Concerns\HasTenantOptions;
+use Stancl\Tenancy\Concerns\ParallelCommand;
+use Stancl\Tenancy\Database\Contracts\TenantWithDatabase;
+use Stancl\Tenancy\Database\Exceptions\TenantDatabaseDoesNotExistException;
 use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface as OI;
 
 class MigrateFresh extends BaseCommand
 {
-    use HasTenantOptions, DealsWithMigrations;
+    use HasTenantOptions, DealsWithMigrations, ParallelCommand;
 
     protected $description = 'Drop all tables and re-run all migrations for tenant(s)';
 
@@ -19,34 +25,90 @@ class MigrateFresh extends BaseCommand
     {
         parent::__construct();
 
-        $this->addOption('--drop-views', null, InputOption::VALUE_NONE, 'Drop views along with tenant tables.', null);
+        $this->addOption('drop-views', null, InputOption::VALUE_NONE, 'Drop views along with tenant tables.', null);
-        $this->addOption('--step', null, InputOption::VALUE_NONE, 'Force the migrations to be run so they can be rolled back individually.');
+        $this->addOption('step', null, InputOption::VALUE_NONE, 'Force the migrations to be run so they can be rolled back individually.');
+        $this->addProcessesOption();
 
         $this->setName('tenants:migrate-fresh');
     }
 
     public function handle(): int
     {
-        tenancy()->runForMultiple($this->getTenants(), function ($tenant) {
+        $success = true;
+
+        if ($this->getProcesses() > 1) {
+            return $this->runConcurrently($this->getTenantChunks()->map(function ($chunk) {
+                return $this->getTenants(array_values($chunk->all()));
+            }));
+        }
+
+        tenancy()->runForMultiple($this->getTenants(), function ($tenant) use (&$success) {
             $this->components->info("Tenant: {$tenant->getTenantKey()}");
-
+            $this->components->task('Dropping tables', fn () => $success = $success && $this->wipeDB());
-            $this->components->task('Dropping tables', function () {
+            $this->components->task('Migrating', fn () => $success = $success && $this->migrateTenant($tenant));
-                $this->callSilently('db:wipe', array_filter([
-                    '--database' => 'tenant',
-                    '--drop-views' => $this->option('drop-views'),
-                    '--force' => true,
-                ]));
-            });
-
-            $this->components->task('Migrating', function () use ($tenant) {
-                $this->callSilent('tenants:migrate', [
-                    '--tenants' => [$tenant->getTenantKey()],
-                    '--step' => $this->option('step'),
-                    '--force' => true,
-                ]);
-            });
         });
 
-        return 0;
+        return $success ? 0 : 1;
+    }
+
+    protected function wipeDB(): bool
+    {
+        return $this->callSilently('db:wipe', array_filter([
+            '--database' => 'tenant',
+            '--drop-views' => $this->option('drop-views'),
+            '--force' => true,
+        ])) === 0;
+    }
+
+    protected function migrateTenant(TenantWithDatabase $tenant): bool
+    {
+        return $this->callSilently('tenants:migrate', [
+            '--tenants' => [$tenant->getTenantKey()],
+            '--step' => $this->option('step'),
+            '--force' => true,
+        ]) === 0;
+    }
+
+    protected function childHandle(...$args): bool
+    {
+        $chunk = $args[0];
+
+        return $this->migrateFreshTenants($chunk);
+    }
+
+    protected function migrateFreshTenants(LazyCollection $tenants): bool
+    {
+        $success = true;
+
+        foreach ($tenants as $tenant) {
+            try {
+                $tenant->run(function ($tenant) use (&$success) {
+                    $this->components->info("Wiping database of tenant {$tenant->getTenantKey()}", OI::VERBOSITY_VERY_VERBOSE);
+                    if ($this->wipeDB()) {
+                        $this->components->info("Wiped database of tenant {$tenant->getTenantKey()}", OI::VERBOSITY_VERBOSE);
+                    } else {
+                        $success = false;
+                        $this->components->error("Wiping database of tenant {$tenant->getTenantKey()} failed!");
+                    }
+
+                    $this->components->info("Migrating database of tenant {$tenant->getTenantKey()}", OI::VERBOSITY_VERY_VERBOSE);
+                    if ($this->migrateTenant($tenant)) {
+                        $this->components->info("Migrated database of tenant {$tenant->getTenantKey()}", OI::VERBOSITY_VERBOSE);
+                    } else {
+                        $success = false;
+                        $this->components->error("Migrating database of tenant {$tenant->getTenantKey()} failed!");
+                    }
+                });
+            } catch (TenantDatabaseDoesNotExistException|QueryException $e) {
+                $this->components->error("Migration failed for tenant {$tenant->getTenantKey()}: {$e->getMessage()}");
+                $success = false;
+
+                if (! $this->option('skip-failing')) {
+                    throw $e;
+                }
+            }
+        }
+
+        return $success;
     }
 }

src/Commands/Rollback.php

@@ -6,15 +6,19 @@ namespace Stancl\Tenancy\Commands;
 
 use Illuminate\Database\Console\Migrations\RollbackCommand;
 use Illuminate\Database\Migrations\Migrator;
+use Illuminate\Database\QueryException;
+use Illuminate\Support\LazyCollection;
 use Stancl\Tenancy\Concerns\DealsWithMigrations;
 use Stancl\Tenancy\Concerns\ExtendsLaravelCommand;
 use Stancl\Tenancy\Concerns\HasTenantOptions;
+use Stancl\Tenancy\Concerns\ParallelCommand;
+use Stancl\Tenancy\Database\Exceptions\TenantDatabaseDoesNotExistException;
 use Stancl\Tenancy\Events\DatabaseRolledBack;
 use Stancl\Tenancy\Events\RollingBackDatabase;
 
 class Rollback extends RollbackCommand
 {
-    use HasTenantOptions, DealsWithMigrations, ExtendsLaravelCommand;
+    use HasTenantOptions, DealsWithMigrations, ExtendsLaravelCommand, ParallelCommand;
 
     protected $description = 'Rollback migrations for tenant(s).';
 
@@ -22,6 +26,9 @@ class Rollback extends RollbackCommand
     {
         parent::__construct($migrator);
 
+        $this->addProcessesOption();
+        $this->addOption('skip-failing', description: 'Continue execution if migration fails for a tenant');
+
         $this->specifyTenantSignature();
     }
 
@@ -33,18 +40,13 @@ class Rollback extends RollbackCommand
             return 1;
         }
 
-        tenancy()->runForMultiple($this->getTenants(), function ($tenant) {
+        if ($this->getProcesses() > 1) {
-            $this->components->info("Tenant: {$tenant->getTenantKey()}");
+            return $this->runConcurrently($this->getTenantChunks()->map(function ($chunk) {
+                return $this->getTenants(array_values($chunk->all()));
+            }));
+        }
 
-            event(new RollingBackDatabase($tenant));
+        return $this->rollbackTenants($this->getTenants()) ? 0 : 1;
-
-            // Rollback
-            parent::handle();
-
-            event(new DatabaseRolledBack($tenant));
-        });
-
-        return 0;
     }
 
     protected static function getTenantCommandName(): string
@@ -52,6 +54,44 @@ class Rollback extends RollbackCommand
         return 'tenants:rollback';
     }
 
+    protected function childHandle(...$args): bool
+    {
+        $chunk = $args[0];
+
+        return $this->rollbackTenants($chunk);
+    }
+
+    protected function rollbackTenants(LazyCollection $tenants): bool
+    {
+        $success = true;
+
+        foreach ($tenants as $tenant) {
+            try {
+                $this->components->info("Tenant {$tenant->getTenantKey()}");
+
+                $tenant->run(function ($tenant) use (&$success) {
+                    event(new RollingBackDatabase($tenant));
+
+                    // Rollback
+                    if (parent::handle() !== 0) {
+                        $success = false;
+                    }
+
+                    event(new DatabaseRolledBack($tenant));
+                });
+            } catch (TenantDatabaseDoesNotExistException|QueryException $e) {
+                $this->components->error("Rollback failed for tenant {$tenant->getTenantKey()}: {$e->getMessage()}");
+                $success = false;
+
+                if (! $this->option('skip-failing')) {
+                    throw $e;
+                }
+            }
+        }
+
+        return $success;
+    }
+
     protected function setParameterDefaults(): void
     {
         // Parameters that this command doesn't support, but can be in tenancy.migration_parameters

src/Concerns/HasTenantOptions.php

@@ -4,6 +4,7 @@ declare(strict_types=1);
 
 namespace Stancl\Tenancy\Concerns;
 
+use Illuminate\Database\Eloquent\Builder;
 use Illuminate\Support\LazyCollection;
 use Stancl\Tenancy\Database\Concerns\PendingScope;
 use Symfony\Component\Console\Input\InputOption;
@@ -21,16 +22,23 @@ trait HasTenantOptions
         ], parent::getOptions());
     }
 
-    protected function getTenants(): LazyCollection
+    protected function getTenants(?array $tenantKeys = null): LazyCollection
+    {
+        return $this->getTenantsQuery($tenantKeys)->cursor();
+    }
+
+    protected function getTenantsQuery(?array $tenantKeys = null): Builder
     {
         return tenancy()->query()
+            ->when($tenantKeys, function ($query) use ($tenantKeys) {
+                $query->whereIn(tenancy()->model()->getTenantKeyName(), $tenantKeys);
+            })
             ->when($this->option('tenants'), function ($query) {
                 $query->whereIn(tenancy()->model()->getTenantKeyName(), $this->option('tenants'));
             })
             ->when(tenancy()->model()::hasGlobalScope(PendingScope::class), function ($query) {
                 $query->withPending(config('tenancy.pending.include_in_queries') ?: $this->option('with-pending'));
-            })
+            });
-            ->cursor();
     }
 
     public function __construct()

src/Concerns/ParallelCommand.php

@@ -0,0 +1,105 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Stancl\Tenancy\Concerns;
+
+use ArrayAccess;
+use Illuminate\Support\Collection;
+use Illuminate\Support\Facades\DB;
+use Symfony\Component\Console\Input\InputOption;
+
+trait ParallelCommand
+{
+    public const int MAX_PROCESSES = 24;
+
+    abstract protected function childHandle(...$args): bool;
+
+    public function addProcessesOption(): void
+    {
+        $this->addOption('processes', 'p', InputOption::VALUE_OPTIONAL, 'How many processes to spawn. Maximum value: ' . static::MAX_PROCESSES . ', recommended value: core count', 1);
+    }
+
+    protected function forkProcess(...$args): int
+    {
+        $pid = pcntl_fork();
+
+        if ($pid === -1) {
+            return -1;
+        } else if ($pid) {
+            // Parent
+            return $pid;
+        } else {
+            // Child
+            DB::reconnect();
+
+            exit($this->childHandle(...$args) ? 0 : 1);
+        }
+    }
+
+    protected function getProcesses(): int
+    {
+        $processes = (int) $this->input->getOption('processes');
+
+        if (($processes < 0) || ($processes > static::MAX_PROCESSES)) {
+            $this->components->error("Maximum value for processes is " . static::MAX_PROCESSES);
+            exit(1);
+        }
+
+        if ($processes > 1 && ! function_exists('pcntl_fork')) {
+            $this->components->error("The pcntl extension is required for parallel migrations to work.");
+        }
+
+        return $processes;
+    }
+
+    protected function getTenantChunks(): Collection
+    {
+        $idCol = tenancy()->model()->getTenantKeyName();
+        $tenants = tenancy()->model()->orderBy($idCol, 'asc')->pluck($idCol);
+        return $tenants->chunk(ceil($tenants->count() / $this->getProcesses()));
+    }
+
+    protected function runConcurrently(array|ArrayAccess $args = null): int
+    {
+        $processes = $this->getProcesses();
+        $success = true;
+        $pids = [];
+
+        for ($i = 0; $i < $processes; $i++) {
+            $pid = $this->forkProcess($args !== null ? $args[$i] : null);
+
+            if ($pid === -1) {
+                $this->components->error("Unable to fork process (iteration $i)!");
+                if ($i === 0) {
+                    exit(1);
+                }
+            }
+
+            $pids[] = $pid;
+        }
+
+        // Fork equivalent of joining an array of join handles
+        foreach ($pids as $i => $pid) {
+            pcntl_waitpid($pid, $status);
+
+            $normalExit = pcntl_wifexited($status);
+
+            if ($normalExit) {
+                $exitCode = pcntl_wexitstatus($status);
+
+                if ($exitCode === 0) {
+                    $this->components->info("Child process [$i] (PID $pid) finished successfully.");
+                } else {
+                    $success = false;
+                    $this->components->error("Child process [$i] (PID $pid) completed with failures.");
+                }
+            } else {
+                $success = false;
+                $this->components->error("Child process [$i] (PID $pid) exited abnormally.");
+            }
+        }
+
+        return $success ? 0 : 1;
+    }
+}