From e594c7070aad609c4caeda760671aca00e638561 Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Tue, 5 Jun 2018 23:17:14 +0300 Subject: fixing controlled descruction --- config_test.go | 40 +++++ errors_test.go | 11 ++ ext/config_test.go | 40 ----- ext/errors_test.go | 11 -- ext/pipe_factory_test.go | 133 -------------- ext/protocol_test.go | 46 ----- ext/socket_factory_test.go | 427 --------------------------------------------- pipe_factory_test.go | 133 ++++++++++++++ protocol_test.go | 46 +++++ socket_factory_test.go | 427 +++++++++++++++++++++++++++++++++++++++++++++ state.go | 5 +- static_pool_test.go | 137 ++++++++------- worker.go | 17 +- worker_test.go | 3 +- 14 files changed, 734 insertions(+), 742 deletions(-) create mode 100644 config_test.go create mode 100644 errors_test.go delete mode 100644 ext/config_test.go delete mode 100644 ext/errors_test.go delete mode 100644 ext/pipe_factory_test.go delete mode 100644 ext/protocol_test.go delete mode 100644 ext/socket_factory_test.go create mode 100644 pipe_factory_test.go create mode 100644 protocol_test.go create mode 100644 socket_factory_test.go diff --git a/config_test.go b/config_test.go new file mode 100644 index 00000000..f4c6246d --- /dev/null +++ b/config_test.go @@ -0,0 +1,40 @@ +package roadrunner + +import ( + "github.com/stretchr/testify/assert" + "testing" + "time" +) + +func Test_NumWorkers(t *testing.T) { + cfg := Config{ + AllocateTimeout: time.Second, + DestroyTimeout: time.Second * 10, + } + err := cfg.Valid() + + assert.NotNil(t, err) + assert.Equal(t, "cfg.NumWorkers must be set", err.Error()) +} + +func Test_AllocateTimeout(t *testing.T) { + cfg := Config{ + NumWorkers: 10, + DestroyTimeout: time.Second * 10, + } + err := cfg.Valid() + + assert.NotNil(t, err) + assert.Equal(t, "cfg.AllocateTimeout must be set", err.Error()) +} + +func Test_DestroyTimeout(t *testing.T) { + cfg := Config{ + NumWorkers: 10, + AllocateTimeout: time.Second, + } + err := cfg.Valid() + + assert.NotNil(t, err) + assert.Equal(t, "cfg.DestroyTimeout must be set", err.Error()) +} diff --git a/errors_test.go b/errors_test.go new file mode 100644 index 00000000..9b0fa53e --- /dev/null +++ b/errors_test.go @@ -0,0 +1,11 @@ +package roadrunner + +import ( + "github.com/stretchr/testify/assert" + "testing" +) + +func Test_JobError_Error(t *testing.T) { + e := JobError([]byte("error")) + assert.Equal(t, "error", e.Error()) +} diff --git a/ext/config_test.go b/ext/config_test.go deleted file mode 100644 index fbdde223..00000000 --- a/ext/config_test.go +++ /dev/null @@ -1,40 +0,0 @@ -package ext - -import ( - "github.com/stretchr/testify/assert" - "testing" - "time" -) - -func Test_NumWorkers(t *testing.T) { - cfg := Config{ - AllocateTimeout: time.Second, - DestroyTimeout: time.Second * 10, - } - err := cfg.Valid() - - assert.NotNil(t, err) - assert.Equal(t, "cfg.NumWorkers must be set", err.Error()) -} - -func Test_AllocateTimeout(t *testing.T) { - cfg := Config{ - NumWorkers: 10, - DestroyTimeout: time.Second * 10, - } - err := cfg.Valid() - - assert.NotNil(t, err) - assert.Equal(t, "cfg.AllocateTimeout must be set", err.Error()) -} - -func Test_DestroyTimeout(t *testing.T) { - cfg := Config{ - NumWorkers: 10, - AllocateTimeout: time.Second, - } - err := cfg.Valid() - - assert.NotNil(t, err) - assert.Equal(t, "cfg.DestroyTimeout must be set", err.Error()) -} diff --git a/ext/errors_test.go b/ext/errors_test.go deleted file mode 100644 index 7c9d7a5b..00000000 --- a/ext/errors_test.go +++ /dev/null @@ -1,11 +0,0 @@ -package ext - -import ( - "github.com/stretchr/testify/assert" - "testing" -) - -func Test_JobError_Error(t *testing.T) { - e := JobError([]byte("error")) - assert.Equal(t, "error", e.Error()) -} diff --git a/ext/pipe_factory_test.go b/ext/pipe_factory_test.go deleted file mode 100644 index 434c31b0..00000000 --- a/ext/pipe_factory_test.go +++ /dev/null @@ -1,133 +0,0 @@ -package ext - -import ( - "github.com/stretchr/testify/assert" - "os/exec" - "testing" -) - -func Test_Pipe_Start(t *testing.T) { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") - - w, err := NewPipeFactory().SpawnWorker(cmd) - assert.NoError(t, err) - assert.NotNil(t, w) - - go func() { - assert.NoError(t, w.Wait()) - }() - - w.Stop() -} - -func Test_Pipe_StartError(t *testing.T) { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") - cmd.Start() - - w, err := NewPipeFactory().SpawnWorker(cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Pipe_PipeError(t *testing.T) { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") - cmd.StdinPipe() - - w, err := NewPipeFactory().SpawnWorker(cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Pipe_PipeError2(t *testing.T) { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") - cmd.StdoutPipe() - - w, err := NewPipeFactory().SpawnWorker(cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Pipe_Failboot(t *testing.T) { - cmd := exec.Command("php", "php-src/tests/failboot.php") - w, err := NewPipeFactory().SpawnWorker(cmd) - - assert.Nil(t, w) - assert.Error(t, err) - assert.Contains(t, err.Error(), "failboot") -} - -func Test_Pipe_Invalid(t *testing.T) { - cmd := exec.Command("php", "php-src/tests/invalid.php") - - w, err := NewPipeFactory().SpawnWorker(cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Pipe_Echo(t *testing.T) { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") - - w, _ := NewPipeFactory().SpawnWorker(cmd) - go func() { - assert.NoError(t, w.Wait()) - }() - defer w.Stop() - - res, err := w.Exec(&Payload{Body: []byte("hello")}) - - assert.NoError(t, err) - assert.NotNil(t, res) - assert.NotNil(t, res.Body) - assert.Nil(t, res.Context) - - assert.Equal(t, "hello", res.String()) -} - -func Test_Pipe_Broken(t *testing.T) { - cmd := exec.Command("php", "php-src/tests/client.php", "broken", "pipes") - - w, _ := NewPipeFactory().SpawnWorker(cmd) - go func() { - err := w.Wait() - - assert.Error(t, err) - assert.Contains(t, err.Error(), "undefined_function()") - }() - defer w.Stop() - - res, err := w.Exec(&Payload{Body: []byte("hello")}) - - assert.Error(t, err) - assert.Nil(t, res) -} - -func Benchmark_Pipe_SpawnWorker_Stop(b *testing.B) { - f := NewPipeFactory() - for n := 0; n < b.N; n++ { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") - w, _ := f.SpawnWorker(cmd) - go func() { - if w.Wait() != nil { - b.Fail() - } - }() - - w.Stop() - } -} - -func Benchmark_Pipe_Worker_ExecEcho(b *testing.B) { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") - - w, _ := NewPipeFactory().SpawnWorker(cmd) - go func() { - w.Wait() - }() - defer w.Stop() - - for n := 0; n < b.N; n++ { - if _, err := w.Exec(&Payload{Body: []byte("hello")}); err != nil { - b.Fail() - } - } -} diff --git a/ext/protocol_test.go b/ext/protocol_test.go deleted file mode 100644 index f6410ef5..00000000 --- a/ext/protocol_test.go +++ /dev/null @@ -1,46 +0,0 @@ -package ext - -import ( - "github.com/pkg/errors" - "github.com/spiral/goridge" - "github.com/stretchr/testify/assert" - "testing" -) - -type relayMock struct { - error bool - payload string -} - -func (r *relayMock) Send(data []byte, flags byte) (err error) { - if r.error { - return errors.New("send error") - } - - return nil -} - -func (r *relayMock) Receive() (data []byte, p goridge.Prefix, err error) { - return []byte(r.payload), goridge.NewPrefix().WithFlag(goridge.PayloadControl), nil -} - -func (r *relayMock) Close() error { - return nil -} - -func Test_Protocol_Errors(t *testing.T) { - err := sendPayload(&relayMock{}, make(chan int)) - assert.Error(t, err) -} - -func Test_Protocol_FetchPID(t *testing.T) { - pid, err := fetchPID(&relayMock{error: false, payload: "{\"pid\":100}"}) - assert.NoError(t, err) - assert.Equal(t, 100, pid) - - _, err = fetchPID(&relayMock{error: true, payload: "{\"pid\":100}"}) - assert.Error(t, err) - - _, err = fetchPID(&relayMock{error: false, payload: "{\"pid:100"}) - assert.Error(t, err) -} diff --git a/ext/socket_factory_test.go b/ext/socket_factory_test.go deleted file mode 100644 index 2d6108bc..00000000 --- a/ext/socket_factory_test.go +++ /dev/null @@ -1,427 +0,0 @@ -package ext - -import ( - "github.com/stretchr/testify/assert" - "net" - "os/exec" - "runtime" - "testing" - "time" -) - -func Test_Tcp_Start(t *testing.T) { - time.Sleep(time.Millisecond * 10) // to ensure free socket - - ls, err := net.Listen("tcp", "localhost:9007") - if assert.NoError(t, err) { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - assert.NoError(t, err) - assert.NotNil(t, w) - - go func() { - assert.NoError(t, w.Wait()) - }() - - w.Stop() -} - -func Test_Tcp_StartCloseFactory(t *testing.T) { - time.Sleep(time.Millisecond * 10) // to ensure free socket - - ls, err := net.Listen("tcp", "localhost:9007") - if assert.NoError(t, err) { - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") - - f := NewSocketFactory(ls, time.Minute) - defer f.Close() - - w, err := f.SpawnWorker(cmd) - assert.NoError(t, err) - assert.NotNil(t, w) - - go func() { - assert.NoError(t, w.Wait()) - }() - - w.Stop() -} - -func Test_Tcp_StartError(t *testing.T) { - time.Sleep(time.Millisecond * 10) // to ensure free socket - - ls, err := net.Listen("tcp", "localhost:9007") - if assert.NoError(t, err) { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") - cmd.Start() - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Tcp_Failboot(t *testing.T) { - time.Sleep(time.Millisecond * 10) // to ensure free socket - - ls, err := net.Listen("tcp", "localhost:9007") - if assert.NoError(t, err) { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/failboot.php") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - assert.Nil(t, w) - assert.Error(t, err) - assert.Contains(t, err.Error(), "failboot") -} - -func Test_Tcp_Timeout(t *testing.T) { - time.Sleep(time.Millisecond * 10) // to ensure free socket - - ls, err := net.Listen("tcp", "localhost:9007") - if assert.NoError(t, err) { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/slow-client.php", "echo", "tcp", "200", "0") - - w, err := NewSocketFactory(ls, time.Millisecond*100).SpawnWorker(cmd) - assert.Nil(t, w) - assert.Error(t, err) - assert.Contains(t, err.Error(), "relay timeout") -} - -func Test_Tcp_Invalid(t *testing.T) { - time.Sleep(time.Millisecond * 10) // to ensure free socket - - ls, err := net.Listen("tcp", "localhost:9007") - if assert.NoError(t, err) { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/invalid.php") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Tcp_Broken(t *testing.T) { - time.Sleep(time.Millisecond * 10) // to ensure free socket - - ls, err := net.Listen("tcp", "localhost:9007") - if assert.NoError(t, err) { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "broken", "tcp") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - go func() { - err := w.Wait() - - assert.Error(t, err) - assert.Contains(t, err.Error(), "undefined_function()") - }() - defer w.Stop() - - res, err := w.Exec(&Payload{Body: []byte("hello")}) - - assert.Error(t, err) - assert.Nil(t, res) -} - -func Test_Tcp_Echo(t *testing.T) { - time.Sleep(time.Millisecond * 10) // to ensure free socket - - ls, err := net.Listen("tcp", "localhost:9007") - if assert.NoError(t, err) { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - go func() { - assert.NoError(t, w.Wait()) - }() - defer w.Stop() - - res, err := w.Exec(&Payload{Body: []byte("hello")}) - - assert.NoError(t, err) - assert.NotNil(t, res) - assert.NotNil(t, res.Body) - assert.Nil(t, res.Context) - - assert.Equal(t, "hello", res.String()) -} - -func Test_Unix_Start(t *testing.T) { - if runtime.GOOS == "windows" { - t.Skip("not supported on " + runtime.GOOS) - } - - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "unix") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - assert.NoError(t, err) - assert.NotNil(t, w) - - go func() { - assert.NoError(t, w.Wait()) - }() - - w.Stop() -} - -func Test_Unix_Failboot(t *testing.T) { - if runtime.GOOS == "windows" { - t.Skip("not supported on " + runtime.GOOS) - } - - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/failboot.php") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - assert.Nil(t, w) - assert.Error(t, err) - assert.Contains(t, err.Error(), "failboot") -} - -func Test_Unix_Timeout(t *testing.T) { - if runtime.GOOS == "windows" { - t.Skip("not supported on " + runtime.GOOS) - } - - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/slow-client.php", "echo", "unix", "200", "0") - - w, err := NewSocketFactory(ls, time.Millisecond*100).SpawnWorker(cmd) - assert.Nil(t, w) - assert.Error(t, err) - assert.Contains(t, err.Error(), "relay timeout") -} - -func Test_Unix_Invalid(t *testing.T) { - if runtime.GOOS == "windows" { - t.Skip("not supported on " + runtime.GOOS) - } - - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/invalid.php") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Unix_Broken(t *testing.T) { - if runtime.GOOS == "windows" { - t.Skip("not supported on " + runtime.GOOS) - } - - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "broken", "unix") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - go func() { - err := w.Wait() - - assert.Error(t, err) - assert.Contains(t, err.Error(), "undefined_function()") - }() - defer w.Stop() - - res, err := w.Exec(&Payload{Body: []byte("hello")}) - - assert.Error(t, err) - assert.Nil(t, res) -} - -func Test_Unix_Echo(t *testing.T) { - if runtime.GOOS == "windows" { - t.Skip("not supported on " + runtime.GOOS) - } - - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer ls.Close() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "unix") - - w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - go func() { - assert.NoError(t, w.Wait()) - }() - defer w.Stop() - - res, err := w.Exec(&Payload{Body: []byte("hello")}) - - assert.NoError(t, err) - assert.NotNil(t, res) - assert.NotNil(t, res.Body) - assert.Nil(t, res.Context) - - assert.Equal(t, "hello", res.String()) -} - -func Benchmark_Tcp_SpawnWorker_Stop(b *testing.B) { - ls, err := net.Listen("tcp", "localhost:9007") - if err == nil { - defer ls.Close() - } else { - b.Skip("socket is busy") - } - - f := NewSocketFactory(ls, time.Minute) - for n := 0; n < b.N; n++ { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") - - w, _ := f.SpawnWorker(cmd) - go func() { - if w.Wait() != nil { - b.Fail() - } - }() - - w.Stop() - } -} - -func Benchmark_Tcp_Worker_ExecEcho(b *testing.B) { - ls, err := net.Listen("tcp", "localhost:9007") - if err == nil { - defer ls.Close() - } else { - b.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") - - w, _ := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - go func() { - w.Wait() - }() - defer w.Stop() - - for n := 0; n < b.N; n++ { - if _, err := w.Exec(&Payload{Body: []byte("hello")}); err != nil { - b.Fail() - } - } -} - -func Benchmark_Unix_SpawnWorker_Stop(b *testing.B) { - if runtime.GOOS == "windows" || runtime.GOOS == "darwin" { - b.Skip("not supported on " + runtime.GOOS) - } - - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer ls.Close() - } else { - b.Skip("socket is busy") - } - - f := NewSocketFactory(ls, time.Minute) - for n := 0; n < b.N; n++ { - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "unix") - - w, _ := f.SpawnWorker(cmd) - go func() { - if w.Wait() != nil { - b.Fail() - } - }() - - w.Stop() - } -} - -func Benchmark_Unix_Worker_ExecEcho(b *testing.B) { - if runtime.GOOS == "windows" || runtime.GOOS == "darwin" { - b.Skip("not supported on " + runtime.GOOS) - } - - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer ls.Close() - } else { - b.Skip("socket is busy") - } - - cmd := exec.Command("php", "php-src/tests/client.php", "echo", "unix") - - w, _ := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) - go func() { - w.Wait() - }() - defer w.Stop() - - for n := 0; n < b.N; n++ { - if _, err := w.Exec(&Payload{Body: []byte("hello")}); err != nil { - b.Fail() - } - } -} diff --git a/pipe_factory_test.go b/pipe_factory_test.go new file mode 100644 index 00000000..ae276ab6 --- /dev/null +++ b/pipe_factory_test.go @@ -0,0 +1,133 @@ +package roadrunner + +import ( + "github.com/stretchr/testify/assert" + "os/exec" + "testing" +) + +func Test_Pipe_Start(t *testing.T) { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") + + w, err := NewPipeFactory().SpawnWorker(cmd) + assert.NoError(t, err) + assert.NotNil(t, w) + + go func() { + assert.NoError(t, w.Wait()) + }() + + w.Stop() +} + +func Test_Pipe_StartError(t *testing.T) { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") + cmd.Start() + + w, err := NewPipeFactory().SpawnWorker(cmd) + assert.Error(t, err) + assert.Nil(t, w) +} + +func Test_Pipe_PipeError(t *testing.T) { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") + cmd.StdinPipe() + + w, err := NewPipeFactory().SpawnWorker(cmd) + assert.Error(t, err) + assert.Nil(t, w) +} + +func Test_Pipe_PipeError2(t *testing.T) { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") + cmd.StdoutPipe() + + w, err := NewPipeFactory().SpawnWorker(cmd) + assert.Error(t, err) + assert.Nil(t, w) +} + +func Test_Pipe_Failboot(t *testing.T) { + cmd := exec.Command("php", "php-src/tests/failboot.php") + w, err := NewPipeFactory().SpawnWorker(cmd) + + assert.Nil(t, w) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failboot") +} + +func Test_Pipe_Invalid(t *testing.T) { + cmd := exec.Command("php", "php-src/tests/invalid.php") + + w, err := NewPipeFactory().SpawnWorker(cmd) + assert.Error(t, err) + assert.Nil(t, w) +} + +func Test_Pipe_Echo(t *testing.T) { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") + + w, _ := NewPipeFactory().SpawnWorker(cmd) + go func() { + assert.NoError(t, w.Wait()) + }() + defer w.Stop() + + res, err := w.Exec(&Payload{Body: []byte("hello")}) + + assert.NoError(t, err) + assert.NotNil(t, res) + assert.NotNil(t, res.Body) + assert.Nil(t, res.Context) + + assert.Equal(t, "hello", res.String()) +} + +func Test_Pipe_Broken(t *testing.T) { + cmd := exec.Command("php", "php-src/tests/client.php", "broken", "pipes") + + w, _ := NewPipeFactory().SpawnWorker(cmd) + go func() { + err := w.Wait() + + assert.Error(t, err) + assert.Contains(t, err.Error(), "undefined_function()") + }() + defer w.Stop() + + res, err := w.Exec(&Payload{Body: []byte("hello")}) + + assert.Error(t, err) + assert.Nil(t, res) +} + +func Benchmark_Pipe_SpawnWorker_Stop(b *testing.B) { + f := NewPipeFactory() + for n := 0; n < b.N; n++ { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") + w, _ := f.SpawnWorker(cmd) + go func() { + if w.Wait() != nil { + b.Fail() + } + }() + + w.Stop() + } +} + +func Benchmark_Pipe_Worker_ExecEcho(b *testing.B) { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") + + w, _ := NewPipeFactory().SpawnWorker(cmd) + go func() { + w.Wait() + }() + defer w.Stop() + + for n := 0; n < b.N; n++ { + if _, err := w.Exec(&Payload{Body: []byte("hello")}); err != nil { + b.Fail() + } + } +} diff --git a/protocol_test.go b/protocol_test.go new file mode 100644 index 00000000..ed3fe461 --- /dev/null +++ b/protocol_test.go @@ -0,0 +1,46 @@ +package roadrunner + +import ( + "github.com/pkg/errors" + "github.com/spiral/goridge" + "github.com/stretchr/testify/assert" + "testing" +) + +type relayMock struct { + error bool + payload string +} + +func (r *relayMock) Send(data []byte, flags byte) (err error) { + if r.error { + return errors.New("send error") + } + + return nil +} + +func (r *relayMock) Receive() (data []byte, p goridge.Prefix, err error) { + return []byte(r.payload), goridge.NewPrefix().WithFlag(goridge.PayloadControl), nil +} + +func (r *relayMock) Close() error { + return nil +} + +func Test_Protocol_Errors(t *testing.T) { + err := sendPayload(&relayMock{}, make(chan int)) + assert.Error(t, err) +} + +func Test_Protocol_FetchPID(t *testing.T) { + pid, err := fetchPID(&relayMock{error: false, payload: "{\"pid\":100}"}) + assert.NoError(t, err) + assert.Equal(t, 100, pid) + + _, err = fetchPID(&relayMock{error: true, payload: "{\"pid\":100}"}) + assert.Error(t, err) + + _, err = fetchPID(&relayMock{error: false, payload: "{\"pid:100"}) + assert.Error(t, err) +} diff --git a/socket_factory_test.go b/socket_factory_test.go new file mode 100644 index 00000000..f6b1350c --- /dev/null +++ b/socket_factory_test.go @@ -0,0 +1,427 @@ +package roadrunner + +import ( + "github.com/stretchr/testify/assert" + "net" + "os/exec" + "runtime" + "testing" + "time" +) + +func Test_Tcp_Start(t *testing.T) { + time.Sleep(time.Millisecond * 10) // to ensure free socket + + ls, err := net.Listen("tcp", "localhost:9007") + if assert.NoError(t, err) { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + assert.NoError(t, err) + assert.NotNil(t, w) + + go func() { + assert.NoError(t, w.Wait()) + }() + + w.Stop() +} + +func Test_Tcp_StartCloseFactory(t *testing.T) { + time.Sleep(time.Millisecond * 10) // to ensure free socket + + ls, err := net.Listen("tcp", "localhost:9007") + if assert.NoError(t, err) { + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") + + f := NewSocketFactory(ls, time.Minute) + defer f.Close() + + w, err := f.SpawnWorker(cmd) + assert.NoError(t, err) + assert.NotNil(t, w) + + go func() { + assert.NoError(t, w.Wait()) + }() + + w.Stop() +} + +func Test_Tcp_StartError(t *testing.T) { + time.Sleep(time.Millisecond * 10) // to ensure free socket + + ls, err := net.Listen("tcp", "localhost:9007") + if assert.NoError(t, err) { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "pipes") + cmd.Start() + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + assert.Error(t, err) + assert.Nil(t, w) +} + +func Test_Tcp_Failboot(t *testing.T) { + time.Sleep(time.Millisecond * 10) // to ensure free socket + + ls, err := net.Listen("tcp", "localhost:9007") + if assert.NoError(t, err) { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/failboot.php") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + assert.Nil(t, w) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failboot") +} + +func Test_Tcp_Timeout(t *testing.T) { + time.Sleep(time.Millisecond * 10) // to ensure free socket + + ls, err := net.Listen("tcp", "localhost:9007") + if assert.NoError(t, err) { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/slow-client.php", "echo", "tcp", "200", "0") + + w, err := NewSocketFactory(ls, time.Millisecond*100).SpawnWorker(cmd) + assert.Nil(t, w) + assert.Error(t, err) + assert.Contains(t, err.Error(), "relay timeout") +} + +func Test_Tcp_Invalid(t *testing.T) { + time.Sleep(time.Millisecond * 10) // to ensure free socket + + ls, err := net.Listen("tcp", "localhost:9007") + if assert.NoError(t, err) { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/invalid.php") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + assert.Error(t, err) + assert.Nil(t, w) +} + +func Test_Tcp_Broken(t *testing.T) { + time.Sleep(time.Millisecond * 10) // to ensure free socket + + ls, err := net.Listen("tcp", "localhost:9007") + if assert.NoError(t, err) { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "broken", "tcp") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + go func() { + err := w.Wait() + + assert.Error(t, err) + assert.Contains(t, err.Error(), "undefined_function()") + }() + defer w.Stop() + + res, err := w.Exec(&Payload{Body: []byte("hello")}) + + assert.Error(t, err) + assert.Nil(t, res) +} + +func Test_Tcp_Echo(t *testing.T) { + time.Sleep(time.Millisecond * 10) // to ensure free socket + + ls, err := net.Listen("tcp", "localhost:9007") + if assert.NoError(t, err) { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + go func() { + assert.NoError(t, w.Wait()) + }() + defer w.Stop() + + res, err := w.Exec(&Payload{Body: []byte("hello")}) + + assert.NoError(t, err) + assert.NotNil(t, res) + assert.NotNil(t, res.Body) + assert.Nil(t, res.Context) + + assert.Equal(t, "hello", res.String()) +} + +func Test_Unix_Start(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("not supported on " + runtime.GOOS) + } + + ls, err := net.Listen("unix", "sock.unix") + if err == nil { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "unix") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + assert.NoError(t, err) + assert.NotNil(t, w) + + go func() { + assert.NoError(t, w.Wait()) + }() + + w.Stop() +} + +func Test_Unix_Failboot(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("not supported on " + runtime.GOOS) + } + + ls, err := net.Listen("unix", "sock.unix") + if err == nil { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/failboot.php") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + assert.Nil(t, w) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failboot") +} + +func Test_Unix_Timeout(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("not supported on " + runtime.GOOS) + } + + ls, err := net.Listen("unix", "sock.unix") + if err == nil { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/slow-client.php", "echo", "unix", "200", "0") + + w, err := NewSocketFactory(ls, time.Millisecond*100).SpawnWorker(cmd) + assert.Nil(t, w) + assert.Error(t, err) + assert.Contains(t, err.Error(), "relay timeout") +} + +func Test_Unix_Invalid(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("not supported on " + runtime.GOOS) + } + + ls, err := net.Listen("unix", "sock.unix") + if err == nil { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/invalid.php") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + assert.Error(t, err) + assert.Nil(t, w) +} + +func Test_Unix_Broken(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("not supported on " + runtime.GOOS) + } + + ls, err := net.Listen("unix", "sock.unix") + if err == nil { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "broken", "unix") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + go func() { + err := w.Wait() + + assert.Error(t, err) + assert.Contains(t, err.Error(), "undefined_function()") + }() + defer w.Stop() + + res, err := w.Exec(&Payload{Body: []byte("hello")}) + + assert.Error(t, err) + assert.Nil(t, res) +} + +func Test_Unix_Echo(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("not supported on " + runtime.GOOS) + } + + ls, err := net.Listen("unix", "sock.unix") + if err == nil { + defer ls.Close() + } else { + t.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "unix") + + w, err := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + go func() { + assert.NoError(t, w.Wait()) + }() + defer w.Stop() + + res, err := w.Exec(&Payload{Body: []byte("hello")}) + + assert.NoError(t, err) + assert.NotNil(t, res) + assert.NotNil(t, res.Body) + assert.Nil(t, res.Context) + + assert.Equal(t, "hello", res.String()) +} + +func Benchmark_Tcp_SpawnWorker_Stop(b *testing.B) { + ls, err := net.Listen("tcp", "localhost:9007") + if err == nil { + defer ls.Close() + } else { + b.Skip("socket is busy") + } + + f := NewSocketFactory(ls, time.Minute) + for n := 0; n < b.N; n++ { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") + + w, _ := f.SpawnWorker(cmd) + go func() { + if w.Wait() != nil { + b.Fail() + } + }() + + w.Stop() + } +} + +func Benchmark_Tcp_Worker_ExecEcho(b *testing.B) { + ls, err := net.Listen("tcp", "localhost:9007") + if err == nil { + defer ls.Close() + } else { + b.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "tcp") + + w, _ := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + go func() { + w.Wait() + }() + defer w.Stop() + + for n := 0; n < b.N; n++ { + if _, err := w.Exec(&Payload{Body: []byte("hello")}); err != nil { + b.Fail() + } + } +} + +func Benchmark_Unix_SpawnWorker_Stop(b *testing.B) { + if runtime.GOOS == "windows" || runtime.GOOS == "darwin" { + b.Skip("not supported on " + runtime.GOOS) + } + + ls, err := net.Listen("unix", "sock.unix") + if err == nil { + defer ls.Close() + } else { + b.Skip("socket is busy") + } + + f := NewSocketFactory(ls, time.Minute) + for n := 0; n < b.N; n++ { + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "unix") + + w, _ := f.SpawnWorker(cmd) + go func() { + if w.Wait() != nil { + b.Fail() + } + }() + + w.Stop() + } +} + +func Benchmark_Unix_Worker_ExecEcho(b *testing.B) { + if runtime.GOOS == "windows" || runtime.GOOS == "darwin" { + b.Skip("not supported on " + runtime.GOOS) + } + + ls, err := net.Listen("unix", "sock.unix") + if err == nil { + defer ls.Close() + } else { + b.Skip("socket is busy") + } + + cmd := exec.Command("php", "php-src/tests/client.php", "echo", "unix") + + w, _ := NewSocketFactory(ls, time.Minute).SpawnWorker(cmd) + go func() { + w.Wait() + }() + defer w.Stop() + + for n := 0; n < b.N; n++ { + if _, err := w.Exec(&Payload{Body: []byte("hello")}); err != nil { + b.Fail() + } + } +} diff --git a/state.go b/state.go index 6c27c4c1..7bd5437d 100644 --- a/state.go +++ b/state.go @@ -24,14 +24,13 @@ type State interface { const ( // StateInactive - no associated process StateInactive int64 = iota + // StateReady - ready for job. StateReady + // StateWorking - working on given payload. StateWorking - // StateDestructing process is being destructed. - StateDestructing - // StateStopping - process is being softly stopped. StateStopping diff --git a/static_pool_test.go b/static_pool_test.go index 7e2315d7..5fe7e062 100644 --- a/static_pool_test.go +++ b/static_pool_test.go @@ -187,6 +187,9 @@ func Test_StaticPool_Broken_FromOutside(t *testing.T) { p.Observe(func(e int, ctx interface{}) { if err, ok := ctx.(error); ok { assert.Contains(t, err.Error(), "exit status 1") + } + + if e == EventWorkerCreate { close(destructed) } }) @@ -200,73 +203,73 @@ func Test_StaticPool_Broken_FromOutside(t *testing.T) { } } -// -//func Test_StaticPool_AllocateTimeout(t *testing.T) { -// p, err := NewPool( -// func() *exec.Cmd { return exec.Command("php", "php-src/tests/client.php", "delay", "pipes") }, -// NewPipeFactory(), -// Config{ -// NumWorkers: 1, -// AllocateTimeout: time.Millisecond * 50, -// DestroyTimeout: time.Second, -// }, -// ) -// -// assert.NotNil(t, p) -// assert.NoError(t, err) -// -// done := make(chan interface{}) -// go func() { -// _, err := p.Exec(&Payload{Body: []byte("100")}) -// assert.NoError(t, err) -// close(done) -// }() -// -// // to ensure that worker is already busy -// time.Sleep(time.Millisecond * 10) -// -// _, err = p.Exec(&Payload{Body: []byte("10")}) -// assert.Error(t, err) -// assert.Contains(t, err.Error(), "worker timeout") -// -// <-done -// p.Destroy() -//} -// -//func Test_StaticPool_Replace_Worker(t *testing.T) { -// p, err := NewPool( -// func() *exec.Cmd { return exec.Command("php", "php-src/tests/client.php", "pid", "pipes") }, -// NewPipeFactory(), -// Config{ -// NumWorkers: 1, -// MaxExecutions: 1, -// AllocateTimeout: time.Second, -// DestroyTimeout: time.Second, -// }, -// ) -// defer p.Destroy() -// -// assert.NotNil(t, p) -// assert.NoError(t, err) -// -// var lastPID string -// lastPID = strconv.Itoa(*p.Workers()[0].Pid) -// -// res, err := p.Exec(&Payload{Body: []byte("hello")}) -// assert.Equal(t, lastPID, string(res.Body)) -// -// for i := 0; i < 10; i++ { -// res, err := p.Exec(&Payload{Body: []byte("hello")}) -// -// assert.NoError(t, err) -// assert.NotNil(t, res) -// assert.NotNil(t, res.Body) -// assert.Nil(t, res.Context) -// -// assert.NotEqual(t, lastPID, string(res.Body)) -// lastPID = string(res.Body) -// } -//} + +func Test_StaticPool_AllocateTimeout(t *testing.T) { + p, err := NewPool( + func() *exec.Cmd { return exec.Command("php", "php-src/tests/client.php", "delay", "pipes") }, + NewPipeFactory(), + Config{ + NumWorkers: 1, + AllocateTimeout: time.Millisecond * 50, + DestroyTimeout: time.Second, + }, + ) + + assert.NotNil(t, p) + assert.NoError(t, err) + + done := make(chan interface{}) + go func() { + _, err := p.Exec(&Payload{Body: []byte("100")}) + assert.NoError(t, err) + close(done) + }() + + // to ensure that worker is already busy + time.Sleep(time.Millisecond * 10) + + _, err = p.Exec(&Payload{Body: []byte("10")}) + assert.Error(t, err) + assert.Contains(t, err.Error(), "worker timeout") + + <-done + p.Destroy() +} + +func Test_StaticPool_Replace_Worker(t *testing.T) { + p, err := NewPool( + func() *exec.Cmd { return exec.Command("php", "php-src/tests/client.php", "pid", "pipes") }, + NewPipeFactory(), + Config{ + NumWorkers: 1, + MaxExecutions: 1, + AllocateTimeout: time.Second, + DestroyTimeout: time.Second, + }, + ) + defer p.Destroy() + + assert.NotNil(t, p) + assert.NoError(t, err) + + var lastPID string + lastPID = strconv.Itoa(*p.Workers()[0].Pid) + + res, err := p.Exec(&Payload{Body: []byte("hello")}) + assert.Equal(t, lastPID, string(res.Body)) + + for i := 0; i < 10; i++ { + res, err := p.Exec(&Payload{Body: []byte("hello")}) + + assert.NoError(t, err) + assert.NotNil(t, res) + assert.NotNil(t, res.Body) + assert.Nil(t, res.Context) + + assert.NotEqual(t, lastPID, string(res.Body)) + lastPID = string(res.Body) + } +} // identical to replace but controlled on worker side func Test_StaticPool_Stop_Worker(t *testing.T) { diff --git a/worker.go b/worker.go index ab0af052..5db5ed82 100644 --- a/worker.go +++ b/worker.go @@ -114,7 +114,7 @@ func (w *Worker) Wait() error { return nil } - if w.state.Value() != StateDestructing { + if w.state.Value() != StateStopping { w.state.set(StateErrored) } else { w.state.set(StateStopped) @@ -130,17 +130,6 @@ func (w *Worker) Wait() error { // Stop sends soft termination command to the worker and waits for process completion. func (w *Worker) Stop() error { - return w.doStop(StateStopping) -} - -// Destroy is identical to stop command but does mark workers with different state. Destroyed workers won't -// throw error state on completion of process destruction (exit status). -func (w *Worker) Destroy() error { - return w.doStop(StateDestructing) -} - -// actual stopping. -func (w *Worker) doStop(state int64) error { select { case <-w.waitDone: return nil @@ -148,7 +137,7 @@ func (w *Worker) doStop(state int64) error { w.mu.Lock() defer w.mu.Unlock() - w.state.set(state) + w.state.set(StateStopping) err := sendPayload(w.rl, &stopCommand{Stop: true}) <-w.waitDone @@ -163,7 +152,7 @@ func (w *Worker) Kill() error { case <-w.waitDone: return nil default: - w.state.set(StateDestructing) + w.state.set(StateStopping) err := w.cmd.Process.Signal(os.Kill) <-w.waitDone diff --git a/worker_test.go b/worker_test.go index 362db1f6..cb88d797 100644 --- a/worker_test.go +++ b/worker_test.go @@ -13,6 +13,7 @@ func Test_GetState(t *testing.T) { w, err := NewPipeFactory().SpawnWorker(cmd) go func() { assert.NoError(t, w.Wait()) + assert.Equal(t, StateStopped, w.State().Value()) }() assert.NoError(t, err) @@ -20,7 +21,7 @@ func Test_GetState(t *testing.T) { assert.Equal(t, StateReady, w.State().Value()) w.Stop() - assert.Equal(t, StateStopped, w.State().Value()) + assert.Equal(t, StateStopping, w.State().Value()) } func Test_Echo(t *testing.T) { -- cgit v1.2.3