feature[RR v2024]: max_jobs jitter to prevent thundering herd problem

ipc/pipe/pipe_factory.go

@@ -31,12 +31,12 @@ type sr struct {
 	err error
 }
 
-// SpawnWorkerWithTimeout creates new Process and connects it to goridge relay,
+// SpawnWorkerWithContext Creates new Process and connects it to goridge relay,
 // method Wait() must be handled on level above.
-func (f *Factory) SpawnWorkerWithTimeout(ctx context.Context, cmd *exec.Cmd) (*worker.Process, error) {
+func (f *Factory) SpawnWorkerWithContext(ctx context.Context, cmd *exec.Cmd, options ...worker.Options) (*worker.Process, error) {
 	spCh := make(chan sr)
 	go func() {
-		w, err := worker.InitBaseWorker(cmd, worker.WithLog(f.log))
+		w, err := worker.InitBaseWorker(cmd, options...)
 		if err != nil {
 			select {
 			case spCh <- sr{
@@ -140,44 +140,6 @@ func (f *Factory) SpawnWorkerWithTimeout(ctx context.Context, cmd *exec.Cmd) (*w
 	}
 }
 
-func (f *Factory) SpawnWorker(cmd *exec.Cmd) (*worker.Process, error) {
-	w, err := worker.InitBaseWorker(cmd, worker.WithLog(f.log))
-	if err != nil {
-		return nil, err
-	}
-
-	in, err := cmd.StdoutPipe()
-	if err != nil {
-		return nil, err
-	}
-
-	out, err := cmd.StdinPipe()
-	if err != nil {
-		return nil, err
-	}
-
-	// Init new PIPE relay
-	relay := pipe.NewPipeRelay(in, out)
-	w.AttachRelay(relay)
-
-	// Start the worker
-	err = w.Start()
-	if err != nil {
-		return nil, err
-	}
-
-	// errors bundle
-	_, err = internal.Pid(relay)
-	if err != nil {
-		_ = w.Kill()
-		return nil, err
-	}
-
-	// everything ok, set ready state
-	w.State().Transition(fsm.StateReady)
-	return w, nil
-}
-
 // Close the factory.
 func (f *Factory) Close() error {
 	return nil

ipc/pipe/pipe_factory_bench_test.go

@@ -11,37 +11,12 @@ import (
 	"go.uber.org/zap"
 )
 
-func Benchmark_WorkerPipeNoTTL(b *testing.B) {
-	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
-
-	log, _ = zap.NewDevelopment()
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
-	require.NoError(b, err)
-
-	go func() {
-		_ = w.Wait()
-	}()
-
-	b.ResetTimer()
-	b.ReportAllocs()
-
-	for i := 0; i < b.N; i++ {
-		res, err := w.Exec(context.Background(), &payload.Payload{Body: []byte("hello")})
-		assert.NoError(b, err)
-		assert.NotNil(b, res)
-	}
-
-	b.Cleanup(func() {
-		assert.NoError(b, w.Stop())
-	})
-}
-
 func Benchmark_WorkerPipeTTL(b *testing.B) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 	ctx := context.Background()
 
 	log, _ = zap.NewDevelopment()
-	w, err := NewPipeFactory(log).SpawnWorkerWithTimeout(ctx, cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(ctx, cmd)
 	require.NoError(b, err)
 
 	go func() {

ipc/pipe/pipe_factory_spawn_test.go

@@ -20,7 +20,7 @@ var log = zap.NewNop()
 func Test_GetState2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	go func() {
 		assert.NoError(t, w.Wait())
 		assert.Equal(t, fsm.StateStopped, w.State().CurrentState())
@@ -36,7 +36,7 @@ func Test_GetState2(t *testing.T) {
 func Test_Kill2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
@@ -59,7 +59,7 @@ func Test_Kill2(t *testing.T) {
 func Test_Pipe_Start2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	assert.NoError(t, err)
 	assert.NotNil(t, w)
 
@@ -77,7 +77,7 @@ func Test_Pipe_StartError2(t *testing.T) {
 		t.Errorf("error running the command: error %v", err)
 	}
 
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	assert.Error(t, err)
 	assert.Nil(t, w)
 }
@@ -89,7 +89,7 @@ func Test_Pipe_PipeError3(t *testing.T) {
 		t.Errorf("error creating the STDIN pipe: error %v", err)
 	}
 
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	assert.Error(t, err)
 	assert.Nil(t, w)
 }
@@ -101,28 +101,28 @@ func Test_Pipe_PipeError4(t *testing.T) {
 		t.Errorf("error creating the STDIN pipe: error %v", err)
 	}
 
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	assert.Error(t, err)
 	assert.Nil(t, w)
 }
 
 func Test_Pipe_Failboot2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/failboot.php")
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	assert.Nil(t, w)
 	assert.Error(t, err)
 }
 
 func Test_Pipe_Invalid2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/invalid.php")
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	assert.Error(t, err)
 	assert.Nil(t, w)
 }
 
 func Test_Pipe_Echo2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	assert.NoError(t, err)
 
 	res, err := w.Exec(context.Background(), &payload.Payload{Body: []byte("hello")})
@@ -145,7 +145,7 @@ func Test_Pipe_Echo2(t *testing.T) {
 
 func Test_Pipe_Broken2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "broken", "pipes")
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	assert.NoError(t, err)
 	require.NotNil(t, w)
 
@@ -162,7 +162,7 @@ func Benchmark_Pipe_SpawnWorker_Stop2(b *testing.B) {
 	f := NewPipeFactory(log)
 	for n := 0; n < b.N; n++ {
 		cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
-		w, _ := f.SpawnWorker(cmd)
+		w, _ := f.SpawnWorkerWithContext(context.Background(), cmd)
 		go func() {
 			if w.Wait() != nil {
 				b.Fail()
@@ -179,7 +179,7 @@ func Benchmark_Pipe_SpawnWorker_Stop2(b *testing.B) {
 func Benchmark_Pipe_Worker_ExecEcho2(b *testing.B) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 
-	w, _ := NewPipeFactory(log).SpawnWorker(cmd)
+	w, _ := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 
 	b.ReportAllocs()
 	b.ResetTimer()
@@ -205,7 +205,7 @@ func Benchmark_Pipe_Worker_ExecEcho2(b *testing.B) {
 
 func Benchmark_Pipe_Worker_ExecEcho4(b *testing.B) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	if err != nil {
 		b.Fatal(err)
 	}
@@ -226,7 +226,7 @@ func Benchmark_Pipe_Worker_ExecEcho4(b *testing.B) {
 
 func Benchmark_Pipe_Worker_ExecEchoWithoutContext2(b *testing.B) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	if err != nil {
 		b.Fatal(err)
 	}
@@ -248,7 +248,7 @@ func Benchmark_Pipe_Worker_ExecEchoWithoutContext2(b *testing.B) {
 func Test_Echo2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -276,7 +276,7 @@ func Test_Echo2(t *testing.T) {
 func Test_BadPayload2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 
-	w, _ := NewPipeFactory(log).SpawnWorker(cmd)
+	w, _ := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 
 	go func() {
 		assert.NoError(t, w.Wait())
@@ -296,7 +296,7 @@ func Test_BadPayload2(t *testing.T) {
 func Test_String2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 
-	w, _ := NewPipeFactory(log).SpawnWorker(cmd)
+	w, _ := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	go func() {
 		assert.NoError(t, w.Wait())
 	}()
@@ -315,7 +315,7 @@ func Test_String2(t *testing.T) {
 func Test_Echo_Slow2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/slow-client.php", "echo", "pipes", "10", "10")
 
-	w, _ := NewPipeFactory(log).SpawnWorker(cmd)
+	w, _ := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	go func() {
 		assert.NoError(t, w.Wait())
 	}()
@@ -339,7 +339,7 @@ func Test_Echo_Slow2(t *testing.T) {
 func Test_Broken2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "broken", "pipes")
 
-	w, err := NewPipeFactory(log).SpawnWorker(cmd)
+	w, err := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -355,7 +355,7 @@ func Test_Broken2(t *testing.T) {
 func Test_Error2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "error", "pipes")
 
-	w, _ := NewPipeFactory(log).SpawnWorker(cmd)
+	w, _ := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	go func() {
 		assert.NoError(t, w.Wait())
 	}()
@@ -380,7 +380,7 @@ func Test_Error2(t *testing.T) {
 func Test_NumExecs2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 
-	w, _ := NewPipeFactory(log).SpawnWorker(cmd)
+	w, _ := NewPipeFactory(log).SpawnWorkerWithContext(context.Background(), cmd)
 	go func() {
 		assert.NoError(t, w.Wait())
 	}()