diff --git a/src/Commands/Migrate.php b/src/Commands/Migrate.php index 8fbd9c69..9f09910e 100644 --- a/src/Commands/Migrate.php +++ b/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; } } diff --git a/src/Commands/MigrateFresh.php b/src/Commands/MigrateFresh.php index 93b93e7c..00a40540 100644 --- a/src/Commands/MigrateFresh.php +++ b/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('--step', null, InputOption::VALUE_NONE, 'Force the migrations to be run so they can be rolled back individually.'); + $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->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', function () { - $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, - ]); - }); + $this->components->task('Dropping tables', fn () => $success = $success && $this->wipeDB()); + $this->components->task('Migrating', fn () => $success = $success && $this->migrateTenant($tenant)); }); - 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; } } diff --git a/src/Commands/Rollback.php b/src/Commands/Rollback.php index 651766f7..d32db47c 100644 --- a/src/Commands/Rollback.php +++ b/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) { - $this->components->info("Tenant: {$tenant->getTenantKey()}"); + if ($this->getProcesses() > 1) { + return $this->runConcurrently($this->getTenantChunks()->map(function ($chunk) { + return $this->getTenants(array_values($chunk->all())); + })); + } - event(new RollingBackDatabase($tenant)); - - // Rollback - parent::handle(); - - event(new DatabaseRolledBack($tenant)); - }); - - return 0; + return $this->rollbackTenants($this->getTenants()) ? 0 : 1; } 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 diff --git a/src/Concerns/HasTenantOptions.php b/src/Concerns/HasTenantOptions.php index 9b3db143..09309766 100644 --- a/src/Concerns/HasTenantOptions.php +++ b/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() diff --git a/src/Concerns/ParallelCommand.php b/src/Concerns/ParallelCommand.php new file mode 100644 index 00000000..7f49fcb6 --- /dev/null +++ b/src/Concerns/ParallelCommand.php @@ -0,0 +1,105 @@ +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; + } +}