From 16b7228a816a875249011998d2b4fa93679a5eb3 Mon Sep 17 00:00:00 2001 From: Jason Wilder Date: Wed, 7 May 2014 11:25:02 -0600 Subject: [PATCH] Add concurrency support For #18 --- start.go | 84 +++++++++++++++++++++++++++++---------- start_test.go | 108 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 172 insertions(+), 20 deletions(-) create mode 100644 start_test.go diff --git a/start.go b/start.go index 931182a..6e062b1 100644 --- a/start.go +++ b/start.go @@ -1,11 +1,13 @@ package main import ( + "errors" "fmt" "os" "os/signal" "path/filepath" "strconv" + "strings" "sync" "time" ) @@ -13,6 +15,7 @@ import ( const shutdownGraceTime = 3 * time.Second var flagPort int +var flagConcurrency string var processes = map[string]*Process{} var shutdown_mutex = new(sync.Mutex) @@ -37,6 +40,35 @@ func init() { cmdStart.Flag.StringVar(&flagProcfile, "f", "Procfile", "procfile") cmdStart.Flag.StringVar(&flagEnv, "e", "", "env") cmdStart.Flag.IntVar(&flagPort, "p", 5000, "port") + cmdStart.Flag.StringVar(&flagConcurrency, "c", "", "concurrency") +} + +func parseConcurrency(value string) (map[string]int, error) { + concurrency := map[string]int{} + if strings.TrimSpace(value) == "" { + return concurrency, nil + } + + parts := strings.Split(value, ",") + for _, part := range parts { + if !strings.Contains(part, "=") { + return concurrency, errors.New("Parsing concurency") + } + + nameValue := strings.Split(part, "=") + n, v := strings.TrimSpace(nameValue[0]), strings.TrimSpace(nameValue[1]) + if n == "" || v == "" { + return concurrency, errors.New("Parsing concurency") + } + + numProcs, err := strconv.ParseInt(v, 10, 16) + if err != nil { + return concurrency, err + } + + concurrency[n] = int(numProcs) + } + return concurrency, nil } func runStart(cmd *Command, args []string) { @@ -52,6 +84,9 @@ func runStart(cmd *Command, args []string) { env, err := ReadEnv(flagEnv) handleError(err) + concurrency, err := parseConcurrency(flagConcurrency) + handleError(err) + of := NewOutletFactory() of.Padding = pf.LongestProcessName() @@ -77,26 +112,35 @@ func runStart(cmd *Command, args []string) { } for idx, proc := range pf.Entries { - if (singleton == "") || (singleton == proc.Name) { - shutdown_mutex.Lock() - wg.Add(1) - port := flagPort + (idx * 100) - ps := NewProcess(proc.Command, env) - processes[proc.Name] = ps - ps.Env["PORT"] = strconv.Itoa(port) - ps.Root = filepath.Dir(flagProcfile) - ps.Stdin = nil - ps.Stdout = of.CreateOutlet(proc.Name, idx, false) - ps.Stderr = of.CreateOutlet(proc.Name, idx, true) - ps.Start() - of.SystemOutput(fmt.Sprintf("starting %s on port %d", proc.Name, port)) - go func(proc ProcfileEntry, ps *Process) { - ps.Wait() - wg.Done() - delete(processes, proc.Name) - ShutdownProcesses(of) - }(proc, ps) - shutdown_mutex.Unlock() + numProcs := 1 + if value, ok := concurrency[proc.Name]; ok { + numProcs = value + } + for i := 0; i < numProcs; i++ { + if (singleton == "") || (singleton == proc.Name) { + shutdown_mutex.Lock() + wg.Add(1) + port := flagPort + (idx * 100) + ps := NewProcess(proc.Command, env) + procName := strings.Join([]string{ + proc.Name, + strconv.FormatInt(int64(i+1), 10)}, ".") + processes[procName] = ps + ps.Env["PORT"] = strconv.Itoa(port) + ps.Root = filepath.Dir(flagProcfile) + ps.Stdin = nil + ps.Stdout = of.CreateOutlet(procName, idx, false) + ps.Stderr = of.CreateOutlet(procName, idx, true) + ps.Start() + of.SystemOutput(fmt.Sprintf("starting %s on port %d", procName, port)) + go func(proc ProcfileEntry, ps *Process) { + ps.Wait() + wg.Done() + delete(processes, procName) + ShutdownProcesses(of) + }(proc, ps) + shutdown_mutex.Unlock() + } } } diff --git a/start_test.go b/start_test.go new file mode 100644 index 0000000..0808dd2 --- /dev/null +++ b/start_test.go @@ -0,0 +1,108 @@ +package main + +import "testing" + +func TestParseConcurrencyFlagEmpty(t *testing.T) { + c, err := parseConcurrency("") + if err != nil { + t.Fatal(err) + } + if len(c) > 0 { + t.Fatal("expected no concurrency settings with ''") + } +} + +func TestParseConcurrencyFlagSimle(t *testing.T) { + c, err := parseConcurrency("foo=2") + if err != nil { + t.Fatal(err) + } + + if len(c) != 1 { + t.Fatal("expected 1 concurrency settings with 'foo=2'") + } + + if c["foo"] != 2 { + t.Fatal("expected concurrency settings of 2 with 'foo=2'") + } +} + +func TestParseConcurrencyFlagMultiple(t *testing.T) { + c, err := parseConcurrency("foo=2,bar=3") + if err != nil { + t.Fatal(err) + } + + if len(c) != 2 { + t.Fatal("expected 1 concurrency settings with 'foo=2'") + } + + if c["foo"] != 2 { + t.Fatal("expected concurrency settings of 2 with 'foo=2'") + } + + if c["bar"] != 3 { + t.Fatal("expected concurrency settings of 3 with 'bar=3'") + } +} + +func TestParseConcurrencyFlagNonInt(t *testing.T) { + _, err := parseConcurrency("foo=x") + if err == nil { + t.Fatal("foo=x should fail") + } +} + +func TestParseConcurrencyFlagWhitespace(t *testing.T) { + c, err := parseConcurrency("foo = 2, bar = 3") + if err != nil { + t.Fatalf("foo = 2, bar = 4 should not fail:%s", err) + } + + if len(c) != 2 { + t.Fatal("expected 1 concurrency settings with 'foo=2'") + } + + if c["foo"] != 2 { + t.Fatal("expected concurrency settings of 2 with 'foo=2'") + } + + if c["bar"] != 3 { + t.Fatal("expected concurrency settings of 3 with 'bar=3'") + } +} + +func TestParseConcurrencyFlagMultipleEquals(t *testing.T) { + _, err := parseConcurrency("foo===2") + if err == nil { + t.Fatalf("foo===2 should fail: %s", err) + } +} + +func TestParseConcurrencyFlagNoValue(t *testing.T) { + _, err := parseConcurrency("foo=") + if err == nil { + t.Fatalf("foo= should fail: %s", err) + } + + _, err = parseConcurrency("=") + if err == nil { + t.Fatalf("= should fail: %s", err) + } + + _, err = parseConcurrency("=1") + if err == nil { + t.Fatalf("= should fail: %s", err) + } + + _, err = parseConcurrency(",") + if err == nil { + t.Fatalf(", should fail: %s", err) + } + + _, err = parseConcurrency(",,,") + if err == nil { + t.Fatalf(",,, should fail: %s", err) + } + +}