diff --git a/plugins/common/parallel.go b/plugins/common/parallel.go new file mode 100644 index 000000000..779bc4f90 --- /dev/null +++ b/plugins/common/parallel.go @@ -0,0 +1,135 @@ +package common + +import ( + "fmt" + "runtime" + "sync" +) + +// ParallelCommand is a type that declares functions +// the ps plugins can execute in parallel +type ParallelCommand func(string) error + +// ParallelCommandRun contains all the arguments +// necessary for a parallel command run +type ParallelCommandRun struct { + Name string +} + +// ParallelCommandResult is the result of a parallel +// command run +type ParallelCommandResult struct { + Name string + Error error +} + +// RunCommandAgainstAllApps runs a given ParallelCommand against all apps +func RunCommandAgainstAllApps(command ParallelCommand, commandName string, parallelCount int) error { + runInSerial := false + + if parallelCount < -1 { + return fmt.Errorf("Invalid value %d for --parallel flag", parallelCount) + } + + if parallelCount == -1 { + cpuCount := runtime.NumCPU() + LogWarn(fmt.Sprintf("Setting --parallel=%d value to CPU count of %d", parallelCount, cpuCount)) + parallelCount = cpuCount + } + + if parallelCount == 0 || parallelCount == 1 { + LogWarn(fmt.Sprintf("Running %s in serial mode", commandName)) + runInSerial = true + } + + if runInSerial { + return runCommandAgainstAllAppsSerially(command, commandName) + } + + return runCommandAgainstAllAppsInParallel(command, commandName, parallelCount) +} + +func runCommandAgainstAllAppsInParallel(command ParallelCommand, commandName string, parallelCount int) error { + apps, err := DokkuApps() + if err != nil { + LogWarn(err.Error()) + return nil + } + + jobs := make(chan string, parallelCount) + results := make(chan ParallelCommandResult, len(apps)) + + go allocateJobs(apps, jobs) + done := make(chan error) + go aggregateResults(results, done) + createParallelWorkerPool(jobs, results, command, parallelCount) + err = <-done + + return err +} + +func runCommandAgainstAllAppsSerially(command ParallelCommand, commandName string) error { + apps, err := DokkuApps() + if err != nil { + LogWarn(err.Error()) + return nil + } + + errorCount := 0 + for _, appName := range apps { + LogInfo1(fmt.Sprintf("Running %s against app %s", commandName, appName)) + if err = command(appName); err != nil { + errorCount++ + } + } + + if errorCount > 0 { + return fmt.Errorf("%s command returned %d errors", commandName, errorCount) + } + + return nil +} + +func allocateJobs(input []string, jobs chan string) { + for _, job := range input { + jobs <- job + } + close(jobs) +} + +func aggregateResults(results chan ParallelCommandResult, done chan error) { + var parallelError error + errorCount := 0 + for result := range results { + if result.Error != nil { + LogWarn(fmt.Sprintf("Error running command against %s", result.Name)) + errorCount++ + } + } + if errorCount > 0 { + parallelError = fmt.Errorf("Encountered %d errors during parallel run", errorCount) + } + done <- parallelError +} + +func createParallelWorker(jobs chan string, results chan ParallelCommandResult, command ParallelCommand, wg *sync.WaitGroup, workerID int) { + for job := range jobs { + LogInfo1(fmt.Sprintf("Running command against %s", job)) + output := ParallelCommandResult{ + Name: job, + Error: command(job), + } + results <- output + } + wg.Done() +} + +func createParallelWorkerPool(jobs chan string, results chan ParallelCommandResult, command ParallelCommand, numberOfWorkers int) { + var wg sync.WaitGroup + for i := 0; i < numberOfWorkers; i++ { + wg.Add(1) + go createParallelWorker(jobs, results, command, &wg, i) + } + wg.Wait() + close(results) +} diff --git a/plugins/ps/ps.go b/plugins/ps/ps.go index 1a9818943..81c52e8c2 100644 --- a/plugins/ps/ps.go +++ b/plugins/ps/ps.go @@ -2,11 +2,8 @@ package ps import ( "fmt" - "runtime" "strconv" "strings" - "sync" - "time" "github.com/dokku/dokku/plugins/common" ) @@ -15,23 +12,6 @@ import ( // and defaults to -1 (false) const RunInSerial = 0 -// ParallelCommand is a type that declares functions -// the ps plugins can execute in parallel -type ParallelCommand func(string) error - -// ParallelCommandRun contains all the arguments -// necessary for a parallel command run -type ParallelCommandRun struct { - AppName string -} - -// ParallelCommandResult is the result of a parallel -// command run -type ParallelCommandResult struct { - AppName string - Error error -} - var ( // DefaultProperties is a map of all valid ps properties with corresponding default property values DefaultProperties = map[string]string{ @@ -39,122 +19,6 @@ var ( } ) -// RunCommandAgainstAllApps runs a given ParallelCommand against all apps -func RunCommandAgainstAllApps(command ParallelCommand, commandName string, parallelCount int) error { - runInSerial := false - - if parallelCount == -1 { - cpuCount := runtime.NumCPU() - common.LogWarn(fmt.Sprintf("Setting --parallel=%d value to CPU count of %d", parallelCount, cpuCount)) - parallelCount = cpuCount - } - - if parallelCount == 0 || parallelCount == 1 { - common.LogWarn(fmt.Sprintf("Running %s in serial mode", commandName)) - runInSerial = true - } - - if runInSerial { - return RunCommandAgainstAllAppsSerially(command, commandName) - } - - return RunCommandAgainstAllAppsInParallel(command, commandName, parallelCount) -} - -// RunCommandAgainstAllAppsInParallel runs a given -// ParallelCommand against all apps in parallel -func RunCommandAgainstAllAppsInParallel(command ParallelCommand, commandName string, parallelCount int) error { - apps, err := common.DokkuApps() - if err != nil { - common.LogWarn(err.Error()) - return nil - } - - jobs := make(chan string, parallelCount) - results := make(chan ParallelCommandResult, len(apps)) - - allocate := func() { - for _, appName := range apps { - jobs <- appName - } - close(jobs) - } - - result := func(done chan error) { - var parallelError error - errorCount := 0 - for result := range results { - if result.Error != nil { - common.LogWarn(fmt.Sprintf("Error running %s against app %s", commandName, result.AppName)) - errorCount++ - } - } - if errorCount > 0 { - parallelError = fmt.Errorf("Encountered %d errors during parallel run", errorCount) - } - done <- parallelError - } - - worker := func(wg *sync.WaitGroup, workerID int) { - for appName := range jobs { - common.LogInfo1(fmt.Sprintf("Running %s against app %s", commandName, appName)) - output := ParallelCommandResult{ - AppName: appName, - Error: command(appName), - } - results <- output - } - wg.Done() - } - - createParallelWorkerPool := func(numberOfWorkers int) { - var wg sync.WaitGroup - for i := 0; i < numberOfWorkers; i++ { - wg.Add(1) - go worker(&wg, i) - } - wg.Wait() - close(results) - } - - startTime := time.Now() - go allocate() - done := make(chan error) - go result(done) - createParallelWorkerPool(parallelCount) - err = <-done - endTime := time.Now() - diff := endTime.Sub(startTime) - - common.LogInfo2Quiet(fmt.Sprintf("Total time taken: %.2f seconds", diff.Seconds())) - - return err -} - -// RunCommandAgainstAllAppsSerially runs a given -// ParallelCommand against all apps in serial -func RunCommandAgainstAllAppsSerially(command ParallelCommand, commandName string) error { - apps, err := common.DokkuApps() - if err != nil { - common.LogWarn(err.Error()) - return nil - } - - errorCount := 0 - for _, appName := range apps { - common.LogInfo1(fmt.Sprintf("Running %s against app %s", commandName, appName)) - if err = command(appName); err != nil { - errorCount++ - } - } - - if errorCount > 0 { - return fmt.Errorf("%s command returned %d errors", commandName, errorCount) - } - - return nil -} - // Rebuild rebuilds app from base image func Rebuild(appName string) error { return common.PlugnTrigger("receive-app", []string{appName}...) diff --git a/plugins/ps/subcommands.go b/plugins/ps/subcommands.go index d4103fb90..bda148ff8 100644 --- a/plugins/ps/subcommands.go +++ b/plugins/ps/subcommands.go @@ -27,7 +27,7 @@ func CommandInspect(appName string) error { // CommandRebuild rebuilds an app from source func CommandRebuild(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Rebuild, "rebuild", parallelCount) + return common.RunCommandAgainstAllApps(Rebuild, "rebuild", parallelCount) } if appName == "" { @@ -67,7 +67,7 @@ func CommandReport(appName string, infoFlag string) error { // CommandRestart restarts an app func CommandRestart(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Restart, "restart", parallelCount) + return common.RunCommandAgainstAllApps(Restart, "restart", parallelCount) } if appName == "" { @@ -150,7 +150,7 @@ func CommandSet(appName string, property string, value string) error { // CommandStart starts an app func CommandStart(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Start, "start", parallelCount) + return common.RunCommandAgainstAllApps(Start, "start", parallelCount) } if appName == "" { @@ -167,7 +167,7 @@ func CommandStart(appName string, allApps bool, parallelCount int) error { // CommandStop stops an app func CommandStop(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Stop, "stop", parallelCount) + return common.RunCommandAgainstAllApps(Stop, "stop", parallelCount) } if appName == "" {