diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 531343f..eac5c40 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -17,10 +17,14 @@ jobs: go: [stable] os: ["ubuntu-latest"] steps: + - name: Check out code + uses: actions/checkout@v7 + - name: Set up Go ${{ matrix.go }} uses: actions/setup-go@v7 # action page: with: go-version: ${{ matrix.go }} + cache-dependency-path: go.sum - name: Set up PHP ${{ matrix.php }} uses: shivammathur/setup-php@v2 # action page: @@ -28,19 +32,9 @@ jobs: php-version: ${{ matrix.php }} extensions: sockets - - name: Check out code - uses: actions/checkout@v7 - - name: Install Composer dependencies run: cd tests && composer update --prefer-dist --no-progress --ansi - - name: Init Go modules Cache # Docs: - uses: actions/cache@v6 - with: - path: ~/go/pkg/mod - key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }} - restore-keys: ${{ runner.os }}-go- - - name: Install Go dependencies run: go mod download diff --git a/Makefile b/Makefile index f9720c2..f4cf78a 100644 --- a/Makefile +++ b/Makefile @@ -5,22 +5,8 @@ SHELL = /bin/sh test_coverage: rm -rf coverage-ci mkdir ./coverage-ci - go test -v -race -cover -tags=debug -timeout 30m -coverpkg=./... -coverprofile=./coverage-ci/pipe.out -covermode=atomic ./ipc/pipe - go test -v -race -cover -tags=debug -timeout 30m -coverpkg=./... -coverprofile=./coverage-ci/socket.out -covermode=atomic ./ipc/socket - go test -v -race -cover -tags=debug -timeout 30m -coverpkg=./... -coverprofile=./coverage-ci/pool_static.out -covermode=atomic ./pool/static_pool - go test -v -race -cover -tags=debug -timeout 30m -coverpkg=./... -coverprofile=./coverage-ci/worker.out -covermode=atomic ./worker - go test -v -race -cover -tags=debug -timeout 30m -coverpkg=./... -coverprofile=./coverage-ci/worker_stack.out -covermode=atomic ./worker_watcher - go test -v -race -cover -tags=debug -timeout 30m -coverpkg=./... -coverprofile=./coverage-ci/channel.out -covermode=atomic ./worker_watcher/container/channel - go test -v -race -cover -tags=debug -timeout 30m -coverpkg=./... -coverprofile=./coverage-ci/ratelimiter.out -covermode=atomic ./pool/ratelimiter - echo 'mode: atomic' > ./coverage-ci/summary.txt - tail -q -n +2 ./coverage-ci/*.out >> ./coverage-ci/summary.txt + go test -p=1 -v -race -cover -tags=debug -timeout 30m -coverpkg=./... -coverprofile=./coverage-ci/summary.txt -covermode=atomic ./... test: ## Run application tests - go test -v -race ./ipc/pipe - go test -v -race ./ipc/socket - go test -v -race ./pool/static_pool - go test -v -race ./worker - go test -v -race ./worker_watcher - go test -v -race ./worker_watcher/container/channel - go test -v -race ./pool/ratelimiter + go test -p=1 -v -race ./... go test -v -race -fuzz=FuzzStaticPoolEcho -fuzztime=30s -tags=debug ./pool/static_pool diff --git a/ipc/pipe/pipe.go b/ipc/pipe/pipe.go index 16d5b15..5e27fdb 100755 --- a/ipc/pipe/pipe.go +++ b/ipc/pipe/pipe.go @@ -15,15 +15,11 @@ import ( // Factory connects to stack using standard // streams (STDIN, STDOUT pipes). -type Factory struct { - log *slog.Logger -} +type Factory struct{} // NewPipeFactory returns a new factory instance. -func NewPipeFactory(log *slog.Logger) *Factory { - return &Factory{ - log: log, - } +func NewPipeFactory(_ *slog.Logger) *Factory { + return &Factory{} } type sr struct { diff --git a/ipc/pipe/pipe_spawn_test.go b/ipc/pipe/pipe_spawn_test.go deleted file mode 100644 index b905376..0000000 --- a/ipc/pipe/pipe_spawn_test.go +++ /dev/null @@ -1,412 +0,0 @@ -package pipe - -import ( - "os/exec" - "sync" - "testing" - "time" - - "log/slog" - - "github.com/roadrunner-server/errors" - "github.com/roadrunner-server/pool/v2/fsm" - "github.com/roadrunner-server/pool/v2/payload" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -var log = slog.New(slog.DiscardHandler) - -func Test_GetState2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - go func() { - assert.NoError(t, w.Wait()) - assert.Equal(t, fsm.StateStopped, w.State().CurrentState()) - }() - - assert.NoError(t, err) - assert.NotNil(t, w) - - assert.Equal(t, fsm.StateReady, w.State().CurrentState()) - assert.NoError(t, w.Stop()) -} - -func Test_Kill2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - wg := &sync.WaitGroup{} - wg.Go(func() { - assert.Error(t, w.Wait()) - assert.Equal(t, fsm.StateErrored, w.State().CurrentState()) - }) - - assert.NoError(t, err) - assert.NotNil(t, w) - - assert.Equal(t, fsm.StateReady, w.State().CurrentState()) - err = w.Kill() - if err != nil { - t.Errorf("error killing the Process: error %v", err) - } - wg.Wait() -} - -func Test_Pipe_Start2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - assert.NoError(t, err) - assert.NotNil(t, w) - - go func() { - assert.NoError(t, w.Wait()) - }() - - assert.NoError(t, w.Stop()) -} - -func Test_Pipe_StartError2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - err := cmd.Start() - if err != nil { - t.Errorf("error running the command: error %v", err) - } - - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Pipe_PipeError3(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - _, err := cmd.StdinPipe() - if err != nil { - t.Errorf("error creating the STDIN pipe: error %v", err) - } - - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Pipe_PipeError4(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - _, err := cmd.StdinPipe() - if err != nil { - t.Errorf("error creating the STDIN pipe: error %v", err) - } - - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), 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).SpawnWorkerWithContext(t.Context(), 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).SpawnWorkerWithContext(t.Context(), 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).SpawnWorkerWithContext(t.Context(), cmd) - assert.NoError(t, err) - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - - assert.NoError(t, err) - assert.NotNil(t, res) - assert.NotNil(t, res.Body) - assert.Empty(t, res.Context) - - go func() { - if w.Wait() != nil { - t.Fail() - } - }() - - assert.Equal(t, "hello", res.String()) - err = w.Stop() - assert.NoError(t, err) -} - -func Test_Pipe_Broken2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "broken", "pipes") - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - assert.NoError(t, err) - require.NotNil(t, w) - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - assert.Error(t, err) - assert.Nil(t, res) - - time.Sleep(time.Second) - err = w.Stop() - assert.Error(t, err) -} - -func Benchmark_Pipe_SpawnWorker_Stop2(b *testing.B) { - f := NewPipeFactory(log) - for b.Loop() { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - w, _ := f.SpawnWorkerWithContext(b.Context(), cmd) - go func() { - if w.Wait() != nil { - b.Fail() - } - }() - - err := w.Stop() - if err != nil { - b.Errorf("error stopping the worker: error %v", err) - } - } -} - -func Benchmark_Pipe_Worker_ExecEcho2(b *testing.B) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - - w, _ := NewPipeFactory(log).SpawnWorkerWithContext(b.Context(), cmd) - - b.ReportAllocs() - - go func() { - err := w.Wait() - if err != nil { - b.Errorf("error waiting the worker: error %v", err) - } - }() - defer func() { - err := w.Stop() - if err != nil { - b.Errorf("error stopping the worker: error %v", err) - } - }() - - for b.Loop() { - if _, err := w.Exec(b.Context(), &payload.Payload{Body: []byte("hello")}); err != nil { - b.Fail() - } - } -} - -func Benchmark_Pipe_Worker_ExecEcho4(b *testing.B) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - w, err := NewPipeFactory(log).SpawnWorkerWithContext(b.Context(), cmd) - if err != nil { - b.Fatal(err) - } - - defer func() { - err = w.Stop() - if err != nil { - b.Errorf("error stopping the Process: error %v", err) - } - }() - - for b.Loop() { - if _, err := w.Exec(b.Context(), &payload.Payload{Body: []byte("hello")}); err != nil { - b.Fail() - } - } -} - -func Benchmark_Pipe_Worker_ExecEchoWithoutContext2(b *testing.B) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - w, err := NewPipeFactory(log).SpawnWorkerWithContext(b.Context(), cmd) - if err != nil { - b.Fatal(err) - } - - defer func() { - err = w.Stop() - if err != nil { - b.Errorf("error stopping the Process: error %v", err) - } - }() - - for b.Loop() { - if _, err := w.Exec(b.Context(), &payload.Payload{Body: []byte("hello")}); err != nil { - b.Fail() - } - } -} - -func Test_Echo2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - if err != nil { - t.Fatal(err) - } - - go func() { - assert.NoError(t, w.Wait()) - }() - defer func() { - err = w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } - }() - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - - assert.Nil(t, err) - assert.NotNil(t, res) - assert.NotNil(t, res.Body) - assert.Empty(t, res.Context) - - assert.Equal(t, "hello", res.String()) -} - -func Test_BadPayload2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - - w, _ := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - - go func() { - assert.NoError(t, w.Wait()) - }() - defer func() { - err := w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } - }() - - res, err := w.Exec(t.Context(), &payload.Payload{}) - assert.NoError(t, err) - assert.NotNil(t, res) -} - -func Test_String2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - - w, _ := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - go func() { - assert.NoError(t, w.Wait()) - }() - defer func() { - err := w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } - }() - - assert.Contains(t, w.String(), "php ../../tests/client.php echo pipes") - assert.Contains(t, w.String(), "ready") - assert.Contains(t, w.String(), "num_execs: 0") -} - -func Test_Echo_Slow2(t *testing.T) { - cmd := exec.Command("php", "../../tests/slow-client.php", "echo", "pipes", "10", "10") - - w, _ := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - go func() { - assert.NoError(t, w.Wait()) - }() - defer func() { - err := w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } - }() - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - - assert.Nil(t, err) - assert.NotNil(t, res) - assert.NotNil(t, res.Body) - assert.Empty(t, res.Context) - - assert.Equal(t, "hello", res.String()) -} - -func Test_Broken2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "broken", "pipes") - - w, err := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - if err != nil { - t.Fatal(err) - } - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - assert.NotNil(t, err) - assert.Nil(t, res) - - time.Sleep(time.Second * 3) - assert.Error(t, w.Stop()) -} - -func Test_Error2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "error", "pipes") - - w, _ := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - go func() { - assert.NoError(t, w.Wait()) - }() - - defer func() { - err := w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } - }() - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - assert.NotNil(t, err) - assert.Nil(t, res) - - if errors.Is(errors.SoftJob, err) == false { - t.Fatal("error should be of type errors.ErrSoftJob") - } - assert.Contains(t, err.Error(), "hello") -} - -func Test_NumExecs2(t *testing.T) { - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - - w, _ := NewPipeFactory(log).SpawnWorkerWithContext(t.Context(), cmd) - go func() { - assert.NoError(t, w.Wait()) - }() - defer func() { - err := w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } - }() - - _, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - if err != nil { - t.Errorf("fail to execute payload: error %v", err) - } - assert.Equal(t, uint64(1), w.State().NumExecs()) - w.State().Transition(fsm.StateReady) - - _, err = w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - if err != nil { - t.Errorf("fail to execute payload: error %v", err) - } - assert.Equal(t, uint64(2), w.State().NumExecs()) - w.State().Transition(fsm.StateReady) - - _, err = w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - if err != nil { - t.Errorf("fail to execute payload: error %v", err) - } - assert.Equal(t, uint64(3), w.State().NumExecs()) - w.State().Transition(fsm.StateReady) -} diff --git a/ipc/pipe/pipe_test.go b/ipc/pipe/pipe_test.go index b067fc0..781fdd0 100644 --- a/ipc/pipe/pipe_test.go +++ b/ipc/pipe/pipe_test.go @@ -2,6 +2,7 @@ package pipe import ( "context" + "log/slog" "os" "os/exec" "runtime" @@ -18,6 +19,8 @@ import ( "github.com/stretchr/testify/require" ) +var log = slog.New(slog.DiscardHandler) + func Test_GetState(t *testing.T) { t.Parallel() ctx := t.Context() @@ -107,21 +110,6 @@ func Test_Pipe_PipeError(t *testing.T) { assert.Nil(t, w) } -func Test_Pipe_PipeError2(t *testing.T) { - t.Parallel() - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - // error cause - _, err := cmd.StdinPipe() - if err != nil { - t.Errorf("error creating the STDIN pipe: error %v", err) - } - - ctx := t.Context() - w, err := NewPipeFactory(log).SpawnWorkerWithContext(ctx, cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - func Test_Pipe_Failboot(t *testing.T) { cmd := exec.Command("php", "../../tests/failboot.php") ctx := t.Context() @@ -483,7 +471,7 @@ func Test_Pipe_SpawnTimeout_ReapsWorker(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*10) defer cancel() - w, err := NewPipeFactory(log).SpawnWorkerWithContext(ctx, exec.Command("php", "../../tests/client.php", "echo", "pipes")) + w, err := NewPipeFactory(log).SpawnWorkerWithContext(ctx, exec.Command("php", "../../tests/slow-client.php", "echo", "pipes", "200", "0")) require.Error(t, err) require.Nil(t, w) diff --git a/ipc/socket/socket.go b/ipc/socket/socket.go index d5b8d7a..b0d3ce9 100755 --- a/ipc/socket/socket.go +++ b/ipc/socket/socket.go @@ -11,7 +11,6 @@ import ( "log/slog" "github.com/roadrunner-server/errors" - "github.com/roadrunner-server/goridge/v4/pkg/relay" "github.com/roadrunner-server/goridge/v4/pkg/socket" "github.com/roadrunner-server/pool/v2/fsm" "github.com/roadrunner-server/pool/v2/internal" @@ -65,7 +64,7 @@ func (f *Factory) listen() error { } // the relay is reused for all later worker traffic, which sets its own bounds _ = conn.SetReadDeadline(time.Time{}) - f.attachRelayToPid(pid, rl) + f.relays.Store(pid, rl) } } @@ -160,8 +159,3 @@ func (f *Factory) findRelayWithContext(ctx context.Context, w *worker.Process) ( } } } - -// attachRelayToPid stores the relay associated with the specific pid -func (f *Factory) attachRelayToPid(pid int64, relay relay.Relay) { - f.relays.Store(pid, relay) -} diff --git a/ipc/socket/socket_spawn_test.go b/ipc/socket/socket_spawn_test.go deleted file mode 100644 index c182ffc..0000000 --- a/ipc/socket/socket_spawn_test.go +++ /dev/null @@ -1,518 +0,0 @@ -package socket - -import ( - "context" - "net" - "os/exec" - "sync" - "testing" - "time" - - "log/slog" - - "github.com/roadrunner-server/pool/v2/payload" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -var log = slog.New(slog.DiscardHandler) - -func Test_Tcp_Start2(t *testing.T) { - ls, err := net.Listen("tcp", "127.0.0.1:9007") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "echo", "tcp") - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*5) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - assert.NoError(t, err) - assert.NotNil(t, w) - - go func() { - assert.NoError(t, w.Wait()) - }() - - err = w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } -} - -func Test_Tcp_StartCloseFactory2(t *testing.T) { - ls, err := net.Listen("tcp", "127.0.0.1:9007") - if assert.NoError(t, err) { - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "echo", "tcp") - f := NewSocketServer(ls, log) - defer func() { - err = ls.Close() - if err != nil { - t.Errorf("error closing the listener: error %v", err) - } - }() - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*5) - defer cancel() - - w, err := f.SpawnWorkerWithContext(ctx, cmd) - assert.NoError(t, err) - assert.NotNil(t, w) - - go func() { - require.NoError(t, w.Wait()) - }() - - err = w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } -} - -func Test_Tcp_StartError2(t *testing.T) { - ls, err := net.Listen("tcp", "127.0.0.1:9007") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes") - err = cmd.Start() - if err != nil { - t.Errorf("error executing the command: error %v", err) - } - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*5) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Tcp_Failboot2(t *testing.T) { - ls, err := net.Listen("tcp", "127.0.0.1:9007") - if assert.NoError(t, err) { - defer func() { - err3 := ls.Close() - if err3 != nil { - t.Errorf("error closing the listener: error %v", err3) - } - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/failboot.php") - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*5) - defer cancel() - - w, err2 := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - assert.Nil(t, w) - assert.Error(t, err2) -} - -func Test_Tcp_Invalid2(t *testing.T) { - ls, err := net.Listen("tcp", "127.0.0.1:9007") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/invalid.php") - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*15) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Tcp_Broken2(t *testing.T) { - ls, err := net.Listen("tcp", "127.0.0.1:9007") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "broken", "tcp") - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*15) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - if err != nil { - t.Fatal(err) - } - wg := sync.WaitGroup{} - wg.Go(func() { - errW := w.Wait() - assert.Error(t, errW) - }) - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - assert.Error(t, err) - assert.Nil(t, res) - wg.Wait() - - time.Sleep(time.Second) - err2 := w.Stop() - // write tcp 127.0.0.1:9007->127.0.0.1:34204: use of closed network connection - // but process exited - assert.NoError(t, err2) -} - -func Test_Tcp_Echo2(t *testing.T) { - ls, err := net.Listen("tcp", "127.0.0.1:9007") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "echo", "tcp") - - ctx, cancel := context.WithTimeout(t.Context(), time.Minute) - defer cancel() - - w, _ := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - go func() { - assert.NoError(t, w.Wait()) - }() - defer func() { - err = w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } - }() - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - - assert.NoError(t, err) - assert.NotNil(t, res) - assert.NotNil(t, res.Body) - assert.Empty(t, res.Context) - - assert.Equal(t, "hello", res.String()) -} - -func Test_Unix_Start2(t *testing.T) { - ls, err := net.Listen("unix", "sock.unix") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "echo", "unix") - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(t.Context(), cmd) - assert.NoError(t, err) - assert.NotNil(t, w) - - go func() { - assert.NoError(t, w.Wait()) - }() - - err = w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } -} - -func Test_Unix_Failboot2(t *testing.T) { - ls, err := net.Listen("unix", "sock.unix") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/failboot.php") - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*5) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - assert.Nil(t, w) - assert.Error(t, err) -} - -func Test_Unix_Timeout2(t *testing.T) { - ls, err := net.Listen("unix", "sock.unix") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/slow-client.php", "echo", "unix", "200", "0") - - ctx, cancel := context.WithTimeout(t.Context(), time.Millisecond*100) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - - assert.Nil(t, w) - assert.Error(t, err) - assert.Contains(t, err.Error(), "Timeout") -} - -func Test_Unix_Invalid2(t *testing.T) { - ls, err := net.Listen("unix", "sock.unix") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/invalid.php") - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*15) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - assert.Error(t, err) - assert.Nil(t, w) -} - -func Test_Unix_Broken2(t *testing.T) { - ls, err := net.Listen("unix", "sock.unix") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "broken", "unix") - - ctx, cancel := context.WithTimeout(t.Context(), time.Second*15) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - if err != nil { - t.Fatal(err) - } - wg := &sync.WaitGroup{} - wg.Go(func() { - errW := w.Wait() - assert.Error(t, errW) - }) - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - - assert.Error(t, err) - assert.Nil(t, res) - wg.Wait() - - time.Sleep(time.Second) - err = w.Stop() - assert.NoError(t, err) -} - -func Test_Unix_Echo2(t *testing.T) { - ls, err := net.Listen("unix", "sock.unix") - if assert.NoError(t, err) { - defer func() { - errC := ls.Close() - require.NoError(t, errC) - }() - } else { - t.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "echo", "unix") - - ctx, cancel := context.WithTimeout(t.Context(), time.Minute) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - if err != nil { - t.Fatal(err) - } - go func() { - assert.NoError(t, w.Wait()) - }() - defer func() { - err = w.Stop() - if err != nil { - t.Errorf("error stopping the Process: error %v", err) - } - }() - - res, err := w.Exec(t.Context(), &payload.Payload{Body: []byte("hello")}) - - assert.NoError(t, err) - assert.NotNil(t, res) - assert.NotNil(t, res.Body) - assert.Empty(t, res.Context) - - assert.Equal(t, "hello", res.String()) -} - -func Benchmark_Tcp_SpawnWorker_Stop2(b *testing.B) { - ls, err := net.Listen("unix", "sock.unix") - if assert.NoError(b, err) { - defer func() { - errC := ls.Close() - require.NoError(b, errC) - }() - } else { - b.Skip("socket is busy") - } - - f := NewSocketServer(ls, log) - for b.Loop() { - cmd := exec.Command("php", "../../tests/client.php", "echo", "tcp") - - w, err := f.SpawnWorkerWithContext(b.Context(), cmd) - if err != nil { - b.Fatal(err) - } - go func() { - assert.NoError(b, w.Wait()) - }() - - err = w.Stop() - if err != nil { - b.Errorf("error stopping the Process: error %v", err) - } - } -} - -func Benchmark_Tcp_Worker_ExecEcho2(b *testing.B) { - ls, err := net.Listen("unix", "sock.unix") - if assert.NoError(b, err) { - defer func() { - errC := ls.Close() - require.NoError(b, errC) - }() - } else { - b.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "echo", "tcp") - - ctx, cancel := context.WithTimeout(b.Context(), time.Second*15) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - if err != nil { - b.Fatal(err) - } - defer func() { - err = w.Stop() - if err != nil { - b.Errorf("error stopping the Process: error %v", err) - } - }() - - for b.Loop() { - if _, err := w.Exec(b.Context(), &payload.Payload{Body: []byte("hello")}); err != nil { - b.Fail() - } - } -} - -func Benchmark_Unix_SpawnWorker_Stop2(b *testing.B) { - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer func() { - errC := ls.Close() - require.NoError(b, errC) - }() - } else { - b.Skip("socket is busy") - } - - f := NewSocketServer(ls, log) - for b.Loop() { - cmd := exec.Command("php", "../../tests/client.php", "echo", "unix") - - w, err := f.SpawnWorkerWithContext(b.Context(), cmd) - if err != nil { - b.Fatal(err) - } - err = w.Stop() - if err != nil { - b.Errorf("error stopping the Process: error %v", err) - } - } -} - -func Benchmark_Unix_Worker_ExecEcho2(b *testing.B) { - ls, err := net.Listen("unix", "sock.unix") - if err == nil { - defer func() { - errC := ls.Close() - require.NoError(b, errC) - }() - } else { - b.Skip("socket is busy") - } - - cmd := exec.Command("php", "../../tests/client.php", "echo", "unix") - - ctx, cancel := context.WithTimeout(b.Context(), time.Second*15) - defer cancel() - - w, err := NewSocketServer(ls, log).SpawnWorkerWithContext(ctx, cmd) - if err != nil { - b.Fatal(err) - } - defer func() { - err = w.Stop() - if err != nil { - b.Errorf("error stopping the Process: error %v", err) - } - }() - - for b.Loop() { - if _, err := w.Exec(b.Context(), &payload.Payload{Body: []byte("hello")}); err != nil { - b.Fail() - } - } -} diff --git a/ipc/socket/socket_test.go b/ipc/socket/socket_test.go index 241fe3a..9f4ca5d 100755 --- a/ipc/socket/socket_test.go +++ b/ipc/socket/socket_test.go @@ -2,6 +2,7 @@ package socket import ( "context" + "log/slog" "net" "os/exec" "sync" @@ -14,6 +15,8 @@ import ( "github.com/stretchr/testify/require" ) +var log = slog.New(slog.DiscardHandler) + func Test_Tcp_Start(t *testing.T) { ctx := t.Context() time.Sleep(time.Millisecond * 10) // to ensure free socket diff --git a/pool/static_pool/dyn_allocator_test.go b/pool/static_pool/dyn_allocator_test.go index 50255bf..ba1a9be 100644 --- a/pool/static_pool/dyn_allocator_test.go +++ b/pool/static_pool/dyn_allocator_test.go @@ -119,7 +119,7 @@ func Test_DynamicPool_OverMax(t *testing.T) { p, err := NewPool( t.Context(), func(cmd []string) *exec.Cmd { - return exec.Command("php", "../../tests/worker-slow-dyn.php") + return exec.Command("php", "../../tests/sleep-ttl.php") }, pipe.NewPipeFactory(slog.Default()), dynAllCfg, @@ -220,7 +220,7 @@ func Test_DynamicPool(t *testing.T) { p, errp := NewPool( t.Context(), func(cmd []string) *exec.Cmd { - return exec.Command("php", "../../tests/worker-slow-dyn.php") + return exec.Command("php", "../../tests/sleep-ttl.php") }, pipe.NewPipeFactory(slog.Default()), dynAllCfg, @@ -285,7 +285,7 @@ func Test_DynamicPool_500W(t *testing.T) { p, err := NewPool( t.Context(), func(cmd []string) *exec.Cmd { - return exec.Command("php", "../../tests/worker-slow-dyn.php") + return exec.Command("php", "../../tests/sleep-ttl.php") }, pipe.NewPipeFactory(slog.Default()), dynAllCfg, @@ -655,7 +655,7 @@ func Test_DynAllocator_SpawnRate_CappedByMaxWorkers(t *testing.T) { p, err := NewPool( t.Context(), func(cmd []string) *exec.Cmd { - return exec.Command("php", "../../tests/worker-slow-dyn.php") + return exec.Command("php", "../../tests/sleep-ttl.php") }, pipe.NewPipeFactory(slog.Default()), cfg, @@ -764,7 +764,7 @@ func Test_DynAllocator_RateLimit_ThunderingHerd(t *testing.T) { p, err := NewPool( t.Context(), func(cmd []string) *exec.Cmd { - return exec.Command("php", "../../tests/worker-slow-dyn.php") + return exec.Command("php", "../../tests/sleep-ttl.php") }, pipe.NewPipeFactory(slog.Default()), cfg, diff --git a/pool/static_pool/pool.go b/pool/static_pool/pool.go index b8606f8..b4f7987 100644 --- a/pool/static_pool/pool.go +++ b/pool/static_pool/pool.go @@ -30,10 +30,6 @@ type Pool struct { cfg *pool.Config // logger log *slog.Logger - // worker command creator - cmd pool.Command - // creates and connects to workers - factory pool.Factory // manages worker states and TTLs ww *workerWatcher.WorkerWatcher // dynamic allocator @@ -63,11 +59,9 @@ func NewPool(ctx context.Context, cmd pool.Command, factory pool.Factory, cfg *p cfg.InitDefaults() p := &Pool{ - cfg: cfg, - cmd: cmd, - factory: factory, - log: log, - stopCh: make(chan struct{}), + cfg: cfg, + log: log, + stopCh: make(chan struct{}), } // options may adjust the config (e.g. WithNumWorkers), so validation and derived @@ -118,10 +112,7 @@ func NewPool(ctx context.Context, cmd pool.Command, factory pool.Factory, cfg *p } // add workers to the watcher - err = p.ww.Watch(workers) - if err != nil { - return nil, err - } + p.ww.Watch(workers) if p.cfg.Supervisor != nil { if p.cfg.Supervisor.ExecTTL != 0 { @@ -353,10 +344,7 @@ func (sp *Pool) Reset(ctx context.Context) error { return err } // add the NEW workers to the watcher - err = sp.ww.Watch(workers) - if err != nil { - return err - } + sp.ww.Watch(workers) return nil } diff --git a/pool/static_pool/pool_test.go b/pool/static_pool/pool_test.go index 8580d23..891f287 100644 --- a/pool/static_pool/pool_test.go +++ b/pool/static_pool/pool_test.go @@ -59,7 +59,7 @@ func Test_MaxWorkers(t *testing.T) { p, err := NewPool( t.Context(), func(cmd []string) *exec.Cmd { - return exec.Command("php", "../../tests/worker-slow-dyn.php") + return exec.Command("php", "../../tests/sleep-ttl.php") }, pipe.NewPipeFactory(slog.Default()), dynAllCfg, @@ -402,34 +402,6 @@ func Test_StaticPool_Echo(t *testing.T) { }) } -func Test_StaticPool_Echo_NilContext(t *testing.T) { - p, err := NewPool( - t.Context(), - func(cmd []string) *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") }, - pipe.NewPipeFactory(slog.Default()), - testCfg, - slog.Default(), - ) - assert.NoError(t, err) - - assert.NotNil(t, p) - - re, err := p.Exec(t.Context(), &payload.Payload{Body: []byte("hello"), Context: nil}, make(chan struct{})) - assert.NoError(t, err) - - res := <-re - - assert.NotNil(t, res) - assert.NotNil(t, res.Body()) - assert.Empty(t, res.Context()) - - assert.Equal(t, "hello", res.Payload().String()) - t.Cleanup(func() { - p.Destroy(t.Context()) - p = nil - }) -} - func Test_StaticPool_Echo_Context(t *testing.T) { p, err := NewPool( t.Context(), diff --git a/pool/static_pool/supervisor_test.go b/pool/static_pool/supervisor_test.go index 5a3cac4..3f573c2 100644 --- a/pool/static_pool/supervisor_test.go +++ b/pool/static_pool/supervisor_test.go @@ -59,36 +59,6 @@ func Test_SupervisedPool_Exec(t *testing.T) { t.Cleanup(func() { p.Destroy(t.Context()) }) } -func Test_SupervisedPool_AddRemoveWorkers(t *testing.T) { - p, err := NewPool( - t.Context(), - func(cmd []string) *exec.Cmd { return exec.Command("php", "../../tests/memleak.php", "pipes") }, - pipe.NewPipeFactory(slog.Default()), - cfgSupervised, - slog.Default(), - ) - - require.NoError(t, err) - require.NotNil(t, p) - - time.Sleep(time.Second) - - pidBefore := p.Workers()[0].Pid() - - for range 10 { - time.Sleep(time.Second) - _, err = p.Exec(t.Context(), &payload.Payload{ - Context: []byte(""), - Body: []byte("foo"), - }, make(chan struct{})) - require.NoError(t, err) - } - - time.Sleep(time.Second) - require.NotEqual(t, pidBefore, p.Workers()[0].Pid()) - t.Cleanup(func() { p.Destroy(t.Context()) }) -} - func Test_SupervisedPool_ImmediateDestroy(t *testing.T) { p, err := NewPool( t.Context(), diff --git a/tests/composer.json b/tests/composer.json index 844ade6..1e806de 100644 --- a/tests/composer.json +++ b/tests/composer.json @@ -2,17 +2,8 @@ "minimum-stability": "dev", "prefer-stable": true, "require": { - "nyholm/psr7": "^1.4", - "spiral/roadrunner-http": "^3.5", "spiral/roadrunner-worker": "^3.5", - "spiral/tokenizer": ">=2.7", - "spiral/goridge": "^4.0", - "spiral/roadrunner-metrics": "^3.0" - }, - "autoload": { - "psr-4": { - "Temporal\\Tests\\": "src" - } + "spiral/goridge": "^4.0" }, "name": "test/test", "description": "test" diff --git a/tests/gzip-large-file.txt b/tests/gzip-large-file.txt deleted file mode 100644 index 4c3eef8..0000000 --- a/tests/gzip-large-file.txt +++ /dev/null @@ -1,19 +0,0 @@ -Lorem ipsum dolor sit amet, consectetur adipiscing elit. Phasellus sodales consectetur fringilla. Phasellus rhoncus lectus nec diam vehicula euismod. Proin eget mollis libero, at ullamcorper ex. Proin sollicitudin, ligula vitae efficitur iaculis, est arcu iaculis ex, in finibus dolor lorem sit amet orci. Pellentesque placerat fermentum est, sed aliquam nulla aliquam non. Donec non tellus vitae elit iaculis ullamcorper. Sed ultricies arcu vel nunc maximus, nec iaculis eros varius. Duis iaculis a mi sed rutrum. Aliquam aliquet ornare arcu id pulvinar. Maecenas luctus orci at efficitur mollis. Aliquam eu placerat magna, id blandit libero. - -Pellentesque habitant morbi tristique senectus et netus et malesuada fames ac turpis egestas. Pellentesque mattis nibh in sagittis dictum. Ut eu neque mollis, pharetra nunc in, sodales neque. Aenean viverra commodo arcu, sit amet egestas magna interdum non. Sed pulvinar ornare urna quis sagittis. Duis tempor at augue sed laoreet. Curabitur aliquam at arcu a semper. Sed eu sem ante. Praesent pellentesque magna lacus, eget ultricies elit pharetra et. Sed ut nisl ultricies orci ultricies porttitor nec id dui. Etiam eu ipsum sagittis, maximus lectus non, semper justo. Fusce orci tellus, congue et consequat nec, facilisis accumsan enim. Proin facilisis quis libero a viverra. Curabitur neque magna, euismod in lacinia nec, scelerisque nec ex. - -Donec nec ullamcorper turpis. Praesent ultrices nunc non dignissim auctor. Morbi at congue lacus. Ut ut diam nec magna finibus ullamcorper. Donec auctor est ut tempor euismod. Proin tincidunt ipsum ac leo aliquet finibus. Curabitur laoreet turpis id enim rutrum, quis tincidunt sem maximus. Vivamus quis diam at lorem viverra pretium. Quisque eget diam scelerisque, hendrerit purus at, blandit lacus. Sed quis consectetur lectus, vitae molestie neque. Quisque in nisi augue. - -Proin ultricies maximus tellus, sed malesuada metus posuere at. Vestibulum a purus ut libero tincidunt dictum et vel sapien. In auctor a metus in ullamcorper. Nullam pharetra pharetra ipsum, non gravida massa tempus non. Proin et varius nunc. Interdum et malesuada fames ac ante ipsum primis in faucibus. Quisque ullamcorper porttitor gravida. Class aptent taciti sociosqu ad litora torquent per conubia nostra, per inceptos himenaeos. In hac habitasse platea dictumst. Donec sit amet rhoncus turpis. Cras vel dui non felis consectetur varius vitae id turpis. Sed vitae hendrerit diam. Aenean neque turpis, laoreet eget libero a, pellentesque malesuada ipsum. Praesent bibendum ultrices nisi eu ornare. Nulla dignissim eu ipsum non iaculis. - -Ut eget lacus porta, posuere neque a, feugiat mi. Sed vehicula sed dui non imperdiet. Nulla quis fringilla nunc. Nunc ipsum dui, hendrerit eu diam sit amet, vehicula convallis libero. Pellentesque habitant morbi tristique senectus et netus et malesuada fames ac turpis egestas. Pellentesque vel diam id felis ornare lobortis a in metus. Mauris feugiat tellus sit amet felis pellentesque ornare sit amet ac augue. Phasellus ut porta purus. Morbi vestibulum, nunc sed pellentesque rhoncus, nibh turpis varius justo, vel mollis ipsum risus at libero. Praesent eget convallis arcu. Donec placerat justo et odio dignissim interdum. Vestibulum rhoncus faucibus tempor. Quisque ipsum tortor, ullamcorper ut sagittis quis, elementum eget metus. Vestibulum ante ipsum primis in faucibus orci luctus et ultrices posuere cubilia Curae; - -Suspendisse in mi non eros fermentum euismod a a tortor. Morbi tincidunt sagittis arcu, ut cursus urna luctus vel. Curabitur malesuada, dui at rutrum pretium, turpis sem vulputate urna, eget suscipit odio ipsum vitae enim. Nunc pellentesque nibh sed facilisis aliquam. Sed ac est sit amet dolor aliquam volutpat in ac risus. Nulla facilisi. Etiam gravida tincidunt metus vitae consectetur. Phasellus rutrum metus ut porttitor varius. Nullam sollicitudin orci ac vehicula accumsan. Maecenas vitae malesuada est. Nam viverra ante ante, id mollis justo sagittis sed. Ut facilisis, lectus id dignissim volutpat, nisl elit fringilla elit, vel facilisis neque lectus et neque. Etiam erat justo, interdum quis neque at, mattis tristique lorem. Mauris malesuada nibh ante, eget eleifend sapien condimentum efficitur. Suspendisse metus turpis, viverra vitae tortor eget, porttitor scelerisque leo. Aenean ac fermentum tortor, ut dapibus mi. - -Fusce semper velit ac tempor lobortis. Suspendisse venenatis eros est, quis sollicitudin dolor consequat vitae. Vivamus tristique erat quis eros laoreet, et dictum nibh sagittis. Integer iaculis pretium orci non venenatis. Aenean ultricies purus ac dui tempor, ut fringilla turpis feugiat. Suspendisse sodales vestibulum aliquet. Donec elit lectus, porta sit amet mollis eu, porttitor eu sem. Vivamus sit amet nisi vel leo commodo suscipit faucibus porttitor orci. Quisque ut sollicitudin quam, luctus pharetra risus. Ut tempus diam odio, elementum porttitor lacus sollicitudin non. Curabitur eget eleifend nisi. Nullam porttitor sagittis gravida. Donec sodales a lorem vel fermentum. Aliquam ac rhoncus quam. - -Fusce pulvinar imperdiet diam et sollicitudin. Quisque a urna id ex porttitor venenatis sit amet id ligula. In fringilla euismod orci vitae pulvinar. In facilisis non nibh ac rutrum. Quisque aliquam sem sit amet pharetra pellentesque. Donec imperdiet libero sed sapien consequat vehicula. Integer vitae ornare sapien. Vivamus pretium felis at urna gravida, quis luctus sem feugiat. Fusce nisl metus, dictum vitae maximus in, interdum id nunc. Phasellus sodales nibh at mi porta egestas quis nec nisl. - -Fusce vel suscipit erat, quis vulputate quam. Orci varius natoque penatibus et magnis dis parturient montes, nascetur ridiculus mus. Etiam a odio nulla. Suspendisse ac tincidunt erat. Duis tortor sem, blandit sed arcu vehicula, faucibus vehicula eros. Sed gravida faucibus quam sed condimentum. Donec sed imperdiet sem. Quisque purus mauris, ultrices sit amet nisl ut, luctus convallis enim. Quisque sit amet nisi ut lacus ullamcorper placerat et id dolor. Mauris tincidunt quam id posuere sodales. Integer tristique efficitur varius. Nulla facilisi. Morbi sed odio eget neque ornare congue. Duis dui massa, maximus ac ullamcorper nec, tempus ut magna. Morbi dictum nisl sagittis orci hendrerit, ac consequat odio pulvinar. Ut mollis sodales egestas. - -Maecenas ante libero, aliquam in enim quis, aliquet malesuada mauris. Quisque porta massa id sem lacinia, id blandit diam mollis. Donec rhoncus vestibulum ante. Aliquam id ligula a erat bibendum condimentum congue non eros. Praesent ac lectus convallis, laoreet quam vitae, viverra arcu. Morbi finibus ligula risus, et ornare turpis iaculis eget. Donec placerat metus vel ex mattis tristique. Donec varius dapibus leo eu pulvinar. Fusce et diam mauris. Duis consequat pharetra urna id malesuada. \ No newline at end of file diff --git a/tests/http/client.php b/tests/http/client.php deleted file mode 100644 index fa9865b..0000000 --- a/tests/http/client.php +++ /dev/null @@ -1,51 +0,0 @@ -waitRequest()) { - try { - $psr7->respond(handleRequest($req, new \Nyholm\Psr7\Response())); - } catch (\Throwable $e) { - $psr7->getWorker()->error((string)$e); - } -} diff --git a/tests/http/cookie.php b/tests/http/cookie.php deleted file mode 100644 index 97673ef..0000000 --- a/tests/http/cookie.php +++ /dev/null @@ -1,334 +0,0 @@ -getBody()->write(strtoupper($req->getCookieParams()['input'])); - - return $resp->withAddedHeader( - "Set-Cookie", - (new Cookie('output', 'cookie-output'))->createHeader() - ); -} - -final class Cookie -{ - /** - * The name of the cookie. - * - * @var string - */ - private $name = ''; - /** - * The value of the cookie. This value is stored on the clients computer; do not store sensitive - * information. - * - * @var string|null - */ - private $value = null; - /** - * Cookie lifetime. This value specified in seconds and declares period of time in which cookie - * will expire relatively to current time() value. - * - * @var int|null - */ - private $lifetime = null; - /** - * The path on the server in which the cookie will be available on. - * - * If set to '/', the cookie will be available within the entire domain. If set to '/foo/', - * the cookie will only be available within the /foo/ directory and all sub-directories such as - * /foo/bar/ of domain. The default value is the current directory that the cookie is being set - * in. - * - * @var string|null - */ - private $path = null; - /** - * The domain that the cookie is available. To make the cookie available on all subdomains of - * example.com then you'd set it to '.example.com'. The . is not required but makes it - * compatible with more browsers. Setting it to www.example.com will make the cookie only - * available in the www subdomain. Refer to tail matching in the spec for details. - * - * @var string|null - */ - private $domain = null; - /** - * Indicates that the cookie should only be transmitted over a secure HTTPS connection from the - * client. When set to true, the cookie will only be set if a secure connection exists. - * On the server-side, it's on the programmer to send this kind of cookie only on secure - * connection - * (e.g. with respect to $_SERVER["HTTPS"]). - * - * @var bool|null - */ - private $secure = null; - /** - * When true the cookie will be made accessible only through the HTTP protocol. This means that - * the cookie won't be accessible by scripting languages, such as JavaScript. This setting can - * effectively help to reduce identity theft through XSS attacks (although it is not supported - * by all browsers). - * - * @var bool - */ - private $httpOnly = true; - - /** - * New Cookie instance, cookies used to schedule cookie set while dispatching Response. - * - * @link http://php.net/manual/en/function.setcookie.php - * - * @param string $name The name of the cookie. - * @param string $value The value of the cookie. This value is stored on the clients - * computer; do not store sensitive information. - * @param int $lifetime Cookie lifetime. This value specified in seconds and declares period - * of time in which cookie will expire relatively to current time() - * value. - * @param string $path The path on the server in which the cookie will be available on. - * If set to '/', the cookie will be available within the entire - * domain. - * If set to '/foo/', the cookie will only be available within the - * /foo/ - * directory and all sub-directories such as /foo/bar/ of domain. The - * default value is the current directory that the cookie is being set - * in. - * @param string $domain The domain that the cookie is available. To make the cookie - * available - * on all subdomains of example.com then you'd set it to - * '.example.com'. - * The . is not required but makes it compatible with more browsers. - * Setting it to www.example.com will make the cookie only available in - * the www subdomain. Refer to tail matching in the spec for details. - * @param bool $secure Indicates that the cookie should only be transmitted over a secure - * HTTPS connection from the client. When set to true, the cookie will - * only be set if a secure connection exists. On the server-side, it's - * on the programmer to send this kind of cookie only on secure - * connection (e.g. with respect to $_SERVER["HTTPS"]). - * @param bool $httpOnly When true the cookie will be made accessible only through the HTTP - * protocol. This means that the cookie won't be accessible by - * scripting - * languages, such as JavaScript. This setting can effectively help to - * reduce identity theft through XSS attacks (although it is not - * supported by all browsers). - */ - public function __construct( - string $name, - string $value = null, - int $lifetime = null, - string $path = null, - string $domain = null, - bool $secure = false, - bool $httpOnly = true - ) { - $this->name = $name; - $this->value = $value; - $this->lifetime = $lifetime; - $this->path = $path; - $this->domain = $domain; - $this->secure = $secure; - $this->httpOnly = $httpOnly; - } - - /** - * The name of the cookie. - * - * @return string - */ - public function getName(): string - { - return $this->name; - } - - /** - * The value of the cookie. This value is stored on the clients computer; do not store sensitive - * information. - * - * @return string|null - */ - public function getValue() - { - return $this->value; - } - - /** - * The time the cookie expires. This is a Unix timestamp so is in number of seconds since the - * epoch. In other words, you'll most likely set this with the time function plus the number of - * seconds before you want it to expire. Or you might use mktime. - * - * Will return null if lifetime is not specified. - * - * @return int|null - */ - public function getExpires() - { - if ($this->lifetime === null) { - return null; - } - - return time() + $this->lifetime; - } - - /** - * The path on the server in which the cookie will be available on. - * - * If set to '/', the cookie will be available within the entire domain. If set to '/foo/', - * the cookie will only be available within the /foo/ directory and all sub-directories such as - * /foo/bar/ of domain. The default value is the current directory that the cookie is being set - * in. - * - * @return string|null - */ - public function getPath() - { - return $this->path; - } - - /** - * The domain that the cookie is available. To make the cookie available on all subdomains of - * example.com then you'd set it to '.example.com'. The . is not required but makes it - * compatible with more browsers. Setting it to www.example.com will make the cookie only - * available in the www subdomain. Refer to tail matching in the spec for details. - * - * @return string|null - */ - public function getDomain() - { - return $this->domain; - } - - /** - * Indicates that the cookie should only be transmitted over a secure HTTPS connection from the - * client. When set to true, the cookie will only be set if a secure connection exists. - * On the server-side, it's on the programmer to send this kind of cookie only on secure - * connection - * (e.g. with respect to $_SERVER["HTTPS"]). - * - * @return bool - */ - public function isSecure(): bool - { - return $this->secure; - } - - /** - * When true the cookie will be made accessible only through the HTTP protocol. This means that - * the cookie won't be accessible by scripting languages, such as JavaScript. This setting can - * effectively help to reduce identity theft through XSS attacks (although it is not supported - * by all browsers). - * - * @return bool - */ - public function isHttpOnly(): bool - { - return $this->httpOnly; - } - - /** - * Get new cookie with altered value. Original cookie object should not be changed. - * - * @param string $value - * - * @return Cookie - */ - public function withValue(string $value): self - { - $cookie = clone $this; - $cookie->value = $value; - - return $cookie; - } - - /** - * Convert cookie instance to string. - * - * @link http://www.w3.org/Protocols/rfc2109/rfc2109 - * @return string - */ - public function createHeader(): string - { - $header = [ - rawurlencode($this->name) . '=' . rawurlencode($this->value) - ]; - if ($this->lifetime !== null) { - $header[] = 'Expires=' . gmdate(\DateTime::COOKIE, $this->getExpires()); - $header[] = 'Max-Age=' . $this->lifetime; - } - if (!empty($this->path)) { - $header[] = 'Path=' . $this->path; - } - if (!empty($this->domain)) { - $header[] = 'Domain=' . $this->domain; - } - if ($this->secure) { - $header[] = 'Secure'; - } - if ($this->httpOnly) { - $header[] = 'HttpOnly'; - } - - return join('; ', $header); - } - - /** - * New Cookie instance, cookies used to schedule cookie set while dispatching Response. - * Static constructor. - * - * @link http://php.net/manual/en/function.setcookie.php - * - * @param string $name The name of the cookie. - * @param string $value The value of the cookie. This value is stored on the clients - * computer; do not store sensitive information. - * @param int $lifetime Cookie lifetime. This value specified in seconds and declares period - * of time in which cookie will expire relatively to current time() - * value. - * @param string $path The path on the server in which the cookie will be available on. - * If set to '/', the cookie will be available within the entire - * domain. - * If set to '/foo/', the cookie will only be available within the - * /foo/ - * directory and all sub-directories such as /foo/bar/ of domain. The - * default value is the current directory that the cookie is being set - * in. - * @param string $domain The domain that the cookie is available. To make the cookie - * available - * on all subdomains of example.com then you'd set it to - * '.example.com'. - * The . is not required but makes it compatible with more browsers. - * Setting it to www.example.com will make the cookie only available in - * the www subdomain. Refer to tail matching in the spec for details. - * @param bool $secure Indicates that the cookie should only be transmitted over a secure - * HTTPS connection from the client. When set to true, the cookie will - * only be set if a secure connection exists. On the server-side, it's - * on the programmer to send this kind of cookie only on secure - * connection (e.g. with respect to $_SERVER["HTTPS"]). - * @param bool $httpOnly When true the cookie will be made accessible only through the HTTP - * protocol. This means that the cookie won't be accessible by - * scripting - * languages, such as JavaScript. This setting can effectively help to - * reduce identity theft through XSS attacks (although it is not - * supported by all browsers). - * - * @return Cookie - */ - public static function create( - string $name, - string $value = null, - int $lifetime = null, - string $path = null, - string $domain = null, - bool $secure = false, - bool $httpOnly = true - ): self { - return new self($name, $value, $lifetime, $path, $domain, $secure, $httpOnly); - } - - /** - * @return string - */ - public function __toString(): string - { - return $this->createHeader(); - } -} diff --git a/tests/http/data.php b/tests/http/data.php deleted file mode 100644 index 6570936..0000000 --- a/tests/http/data.php +++ /dev/null @@ -1,17 +0,0 @@ -getParsedBody(); - - ksort($data); - ksort($data['arr']); - ksort($data['arr']['x']['y']); - - $resp->getBody()->write(json_encode($data)); - - return $resp; -} diff --git a/tests/http/echo.php b/tests/http/echo.php deleted file mode 100644 index 08e29a2..0000000 --- a/tests/http/echo.php +++ /dev/null @@ -1,10 +0,0 @@ -getBody()->write(strtoupper($req->getQueryParams()['hello'])); - return $resp->withStatus(201); -} diff --git a/tests/http/echoDelay.php b/tests/http/echoDelay.php deleted file mode 100644 index 78e8547..0000000 --- a/tests/http/echoDelay.php +++ /dev/null @@ -1,11 +0,0 @@ -getBody()->write(strtoupper($req->getQueryParams()['hello'])); - return $resp->withStatus(201); -} diff --git a/tests/http/echoerr.php b/tests/http/echoerr.php deleted file mode 100644 index 7e1d05e..0000000 --- a/tests/http/echoerr.php +++ /dev/null @@ -1,12 +0,0 @@ -getQueryParams()['hello'])); - - $resp->getBody()->write(strtoupper($req->getQueryParams()['hello'])); - return $resp->withStatus(201); -} diff --git a/tests/http/env.php b/tests/http/env.php deleted file mode 100644 index 3755bde..0000000 --- a/tests/http/env.php +++ /dev/null @@ -1,10 +0,0 @@ -getBody()->write($_SERVER['ENV_KEY']); - return $resp; -} diff --git a/tests/http/error.php b/tests/http/error.php deleted file mode 100644 index 527e406..0000000 --- a/tests/http/error.php +++ /dev/null @@ -1,9 +0,0 @@ -getBody()->write(strtoupper($req->getHeaderLine('input'))); - - return $resp->withAddedHeader("Header", $req->getQueryParams()['hello']); -} diff --git a/tests/http/headers.php b/tests/http/headers.php deleted file mode 100644 index b6f3967..0000000 --- a/tests/http/headers.php +++ /dev/null @@ -1,11 +0,0 @@ -getBody()->write(json_encode($req->getHeaders())); - - return $resp; -} diff --git a/tests/http/ip.php b/tests/http/ip.php deleted file mode 100644 index 49eb928..0000000 --- a/tests/http/ip.php +++ /dev/null @@ -1,11 +0,0 @@ -getBody()->write($req->getServerParams()['REMOTE_ADDR']); - - return $resp; -} diff --git a/tests/http/memleak.php b/tests/http/memleak.php deleted file mode 100644 index 197a7fb..0000000 --- a/tests/http/memleak.php +++ /dev/null @@ -1,11 +0,0 @@ -getHeaderLine("Content-Type") != 'application/json') { - $resp->getBody()->write("invalid content-type"); - return $resp; - } - - // we expect json body - $p = json_decode($req->getBody(), true); - $resp->getBody()->write(json_encode(array_flip($p))); - - return $resp; -} diff --git a/tests/http/pid.php b/tests/http/pid.php deleted file mode 100644 index f22d8e2..0000000 --- a/tests/http/pid.php +++ /dev/null @@ -1,11 +0,0 @@ -getBody()->write((string)getmypid()); - - return $resp; -} diff --git a/tests/http/push.php b/tests/http/push.php deleted file mode 100644 index d88fc07..0000000 --- a/tests/http/push.php +++ /dev/null @@ -1,10 +0,0 @@ -getBody()->write(strtoupper($req->getQueryParams()['hello'])); - return $resp->withAddedHeader("Http2-Push", __FILE__)->withStatus(201); -} diff --git a/tests/http/request-uri.php b/tests/http/request-uri.php deleted file mode 100644 index d4c8755..0000000 --- a/tests/http/request-uri.php +++ /dev/null @@ -1,10 +0,0 @@ -getBody()->write($_SERVER['REQUEST_URI']); - return $resp; -} diff --git a/tests/http/server.php b/tests/http/server.php deleted file mode 100644 index 393d962..0000000 --- a/tests/http/server.php +++ /dev/null @@ -1,11 +0,0 @@ -getBody()->write(strtoupper($req->getHeaderLine('input'))); - - return $resp->withAddedHeader("Header", $req->getQueryParams()['hello']); -} diff --git a/tests/http/slow-client.php b/tests/http/slow-client.php deleted file mode 100644 index be554a7..0000000 --- a/tests/http/slow-client.php +++ /dev/null @@ -1,52 +0,0 @@ -waitRequest()) { - try { - $psr7->respond(handleRequest($req, new \Nyholm\Psr7\Response())); - } catch (\Throwable $e) { - $psr7->getWorker()->error((string) $e); - } -} diff --git a/tests/http/stuck.php b/tests/http/stuck.php deleted file mode 100644 index 2dea057..0000000 --- a/tests/http/stuck.php +++ /dev/null @@ -1,11 +0,0 @@ -getBody()->write(strtoupper($req->getQueryParams()['hello'])); - return $resp->withStatus(201); -} diff --git a/tests/http/upload.php b/tests/http/upload.php deleted file mode 100644 index 5752624..0000000 --- a/tests/http/upload.php +++ /dev/null @@ -1,35 +0,0 @@ -getUploadedFiles(); - array_walk_recursive($files, function (&$v) { - /** - * @var \Psr\Http\Message\UploadedFileInterface $v - */ - - if ($v->getError()) { - $v = [ - 'name' => $v->getClientFilename(), - 'size' => $v->getSize(), - 'mime' => $v->getClientMediaType(), - 'error' => $v->getError(), - ]; - } else { - $v = [ - 'name' => $v->getClientFilename(), - 'size' => $v->getSize(), - 'mime' => $v->getClientMediaType(), - 'error' => $v->getError(), - 'sha512' => hash('sha512', $v->getStream()->__toString()), - ]; - } - }); - - $resp->getBody()->write(json_encode($files, JSON_UNESCAPED_SLASHES)); - - return $resp; -} diff --git a/tests/http/user-agent.php b/tests/http/user-agent.php deleted file mode 100644 index 03d7a2c..0000000 --- a/tests/http/user-agent.php +++ /dev/null @@ -1,10 +0,0 @@ -getBody()->write($_SERVER['HTTP_USER_AGENT']); - return $resp; -} diff --git a/tests/issue659.php b/tests/issue659.php deleted file mode 100644 index 2a0e4f1..0000000 --- a/tests/issue659.php +++ /dev/null @@ -1,21 +0,0 @@ -waitRequest()) { - $psr7->getWorker()->error("test_error"); -} diff --git a/tests/metrics-issue-571.php b/tests/metrics-issue-571.php deleted file mode 100644 index 947ae1f..0000000 --- a/tests/metrics-issue-571.php +++ /dev/null @@ -1,37 +0,0 @@ -getRPCAddress()) -); - -$metrics->declare( - 'test', - RoadRunner\Metrics\Collector::counter()->withHelp('Test counter') -); - -while ($req = $worker->waitRequest()) { - try { - $rsp = new \Nyholm\Psr7\Response(); - $rsp->getBody()->write("hello world"); - - $metrics->add('test', 1); - - $worker->respond($rsp); - } catch (\Throwable $e) { - $worker->getWorker()->error((string)$e); - } -} diff --git a/tests/pipes_test_script.sh b/tests/pipes_test_script.sh deleted file mode 100755 index c759b0a..0000000 --- a/tests/pipes_test_script.sh +++ /dev/null @@ -1,2 +0,0 @@ -#!/usr/bin/env bash -php ../../tests/client.php echo pipes diff --git a/tests/psr-worker-bench.php b/tests/psr-worker-bench.php deleted file mode 100644 index 1513211..0000000 --- a/tests/psr-worker-bench.php +++ /dev/null @@ -1,30 +0,0 @@ -waitRequest()) { - try { - $resp = new \Nyholm\Psr7\Response(); - $resp->getBody()->write("hello world"); - - $psr7->respond($resp); - } catch (\Throwable $e) { - $psr7->getWorker()->error((string)$e); - } -} diff --git a/tests/psr-worker-post.php b/tests/psr-worker-post.php deleted file mode 100644 index 2f54af5..0000000 --- a/tests/psr-worker-post.php +++ /dev/null @@ -1,30 +0,0 @@ -waitRequest()) { - try { - $resp = new \Nyholm\Psr7\Response(); - $resp->getBody()->write((string) $req->getBody()); - - $psr7->respond($resp); - } catch (\Throwable $e) { - $psr7->getWorker()->error((string)$e); - } -} diff --git a/tests/psr-worker-slow.php b/tests/psr-worker-slow.php deleted file mode 100644 index 153dff6..0000000 --- a/tests/psr-worker-slow.php +++ /dev/null @@ -1,29 +0,0 @@ -waitRequest()) { - try { - $resp = new \Nyholm\Psr7\Response(); - sleep(mt_rand(1,20)); - $resp->getBody()->write("hello world"); - - $psr7->respond($resp); - } catch (\Throwable $e) { - $psr7->getWorker()->error((string)$e); - } -} diff --git a/tests/psr-worker.php b/tests/psr-worker.php deleted file mode 100644 index db53eee..0000000 --- a/tests/psr-worker.php +++ /dev/null @@ -1,28 +0,0 @@ -waitRequest()) { - try { - $resp = new \Nyholm\Psr7\Response(); - $resp->getBody()->write(str_repeat("hello world", 1000)); - - $psr7->respond($resp); - } catch (\Throwable $e) { - $psr7->getWorker()->error((string)$e); - } -} diff --git a/tests/raw-error.php b/tests/raw-error.php deleted file mode 100644 index 3caf46c..0000000 --- a/tests/raw-error.php +++ /dev/null @@ -1,21 +0,0 @@ -waitRequest()) {} diff --git a/tests/sample.txt b/tests/sample.txt deleted file mode 100644 index d64a3d9..0000000 --- a/tests/sample.txt +++ /dev/null @@ -1 +0,0 @@ -sample diff --git a/tests/slow-pid.php b/tests/slow-pid.php deleted file mode 100644 index 8278561..0000000 --- a/tests/slow-pid.php +++ /dev/null @@ -1,18 +0,0 @@ -waitPayload()) { - try { - sleep(1); - $rr->respond(new RoadRunner\Payload((string)getmypid())); - } catch (\Throwable $e) { - $rr->error((string)$e); - } - } diff --git a/tests/socket_test_script.sh b/tests/socket_test_script.sh deleted file mode 100755 index 3948c4f..0000000 --- a/tests/socket_test_script.sh +++ /dev/null @@ -1,2 +0,0 @@ -#!/usr/bin/env bash -php ../../tests/client.php echo tcp diff --git a/tests/src/Activity/SimpleActivity.php b/tests/src/Activity/SimpleActivity.php deleted file mode 100644 index 576b126..0000000 --- a/tests/src/Activity/SimpleActivity.php +++ /dev/null @@ -1,63 +0,0 @@ -", $user->name, $user->email)); - } - - #[ActivityMethod] - public function slow( - string $input - ): string { - sleep(2); - - return strtolower($input); - } - - #[ActivityMethod] - public function sha512( - Bytes $input - ): string { - return hash("sha512", ($input->getData())); - } - - public function updateRunID(WorkflowExecution $e): WorkflowExecution - { - $e->setRunId('updated'); - return $e; - } - - #[ActivityMethod] - public function fail() - { - throw new \Error("failed activity"); - } -} \ No newline at end of file diff --git a/tests/src/Client/StartNewWorkflow.php b/tests/src/Client/StartNewWorkflow.php deleted file mode 100644 index 67bc1d0..0000000 --- a/tests/src/Client/StartNewWorkflow.php +++ /dev/null @@ -1,23 +0,0 @@ -stub = $client->newWorkflowStub(SimpleDTOWorkflow::class); - } - - public function __invoke() - { - } -} diff --git a/tests/src/Workflow/SagaWorkflow.php b/tests/src/Workflow/SagaWorkflow.php deleted file mode 100644 index e47c020..0000000 --- a/tests/src/Workflow/SagaWorkflow.php +++ /dev/null @@ -1,54 +0,0 @@ -withStartToCloseTimeout(60) - ->withRetryOptions(RetryOptions::new()->withMaximumAttempts(1)) - ); - - $saga = new Workflow\Saga(); - $saga->setParallelCompensation(true); - - try { - yield $simple->echo('test'); - $saga->addCompensation( - function () use ($simple) { - yield $simple->echo('compensate echo'); - } - ); - - yield $simple->lower('TEST'); - $saga->addCompensation( - function () use ($simple) { - yield $simple->lower('COMPENSATE LOWER'); - } - ); - - yield $simple->fail(); - } catch (\Throwable $e) { - yield $saga->compensate(); - throw $e; - } - } -} diff --git a/tests/stream_worker.php b/tests/stream_worker.php deleted file mode 100644 index 7540ba6..0000000 --- a/tests/stream_worker.php +++ /dev/null @@ -1,34 +0,0 @@ -chunk_size = 10 * 10 * 1024; -$fp = "fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"; - -while ($worker->waitPayload()) { - try { - $resp = (new Response())->withBody(Stream::create($fp)); - $psr7->respond($resp); - } catch (\Throwable $e) { - $psr7->getWorker()->error((string)$e); - } -} - -?> \ No newline at end of file diff --git a/tests/worker-slow-dyn.php b/tests/worker-slow-dyn.php deleted file mode 100644 index 3f4aa93..0000000 --- a/tests/worker-slow-dyn.php +++ /dev/null @@ -1,31 +0,0 @@ -waitRequest()) { - try { - $resp = new \Nyholm\Psr7\Response(); - sleep(10); - $resp->getBody()->write("hello world"); - - $psr7->respond($resp); - } catch (\Throwable $e) { - $psr7->getWorker()->error((string)$e); - } -} diff --git a/worker/options.go b/worker/options.go index 6b74825..1f95ede 100644 --- a/worker/options.go +++ b/worker/options.go @@ -1,9 +1,8 @@ package worker import ( - "crypto/rand" "log/slog" - "math/big" + "math/rand/v2" ) const ( @@ -20,31 +19,7 @@ func WithLog(z *slog.Logger) Options { func WithMaxExecs(maxExecs uint64) Options { return func(p *Process) { - p.maxExecs = calculateMaxExecsJitter(maxExecs, maxExecsPercentJitter, p.log) + percent := rand.Uint64N(maxExecsPercentJitter) //nolint:gosec // This value only sets the worker restart limit. + p.maxExecs = maxExecs + uint64(float64(maxExecs)*float64(percent)/100) } } - -func calculateMaxExecsJitter(maxExecs, jitter uint64, log *slog.Logger) uint64 { - if maxExecs == 0 { - return 0 - } - - random, err := rand.Int(rand.Reader, big.NewInt(int64(jitter))) //nolint:gosec - - if err != nil { - if log != nil { - log.Debug("jitter calculation error", "error", err, "jitter", jitter) - } - return maxExecs - } - - percent := random.Uint64() - - if percent == 0 { - return maxExecs - } - - result := (float64(maxExecs) * float64(percent)) / 100.0 - - return maxExecs + uint64(result) -} diff --git a/worker/worker.go b/worker/worker.go index e662e09..e46ecce 100644 --- a/worker/worker.go +++ b/worker/worker.go @@ -111,7 +111,7 @@ func InitBaseWorker(cmd *exec.Cmd, options ...Options) (*Process, error) { // https://linux.die.net/man/7/pipe // see pipe capacity buf := make([]byte, 65536) - errCopy := copyBuffer(w, rc, buf) + _, errCopy := io.CopyBuffer(w, rc, buf) if errCopy != nil { w.log.Debug("stderr", "error", errCopy) } @@ -458,12 +458,6 @@ func (w *Process) MaxExecs() uint64 { return w.maxExecs } -// copyBuffer copies src to dst through buf until EOF. -func copyBuffer(dst io.Writer, src io.Reader, buf []byte) error { - _, err := io.CopyBuffer(dst, src, buf) - return err -} - // sendFrame sends frame to the worker func (w *Process) sendFrame(p *payload.Payload) error { const op = errors.Op("sync_worker_send_frame") diff --git a/worker_watcher/worker_watcher.go b/worker_watcher/worker_watcher.go index 6ebf591..d169673 100644 --- a/worker_watcher/worker_watcher.go +++ b/worker_watcher/worker_watcher.go @@ -57,7 +57,7 @@ func NewSyncWorkerWatcher(allocator Allocator, log *slog.Logger, numWorkers uint return ww } -func (ww *WorkerWatcher) Watch(workers []*worker.Process) error { +func (ww *WorkerWatcher) Watch(workers []*worker.Process) { ww.mu.Lock() defer ww.mu.Unlock() @@ -68,8 +68,6 @@ func (ww *WorkerWatcher) Watch(workers []*worker.Process) error { ww.workers.Store(workers[i].Pid(), workers[i]) ww.addToWatch(workers[i]) } - - return nil } // NumWorkers returns the live number of workers tracked by the watcher. @@ -274,33 +272,22 @@ func (ww *WorkerWatcher) Reset(ctx context.Context) uint64 { select { case <-tt: ww.mu.RLock() - - // that might be one of the workers is working. To proceed, all workers should be inside a channel - if ww.numWorkers.Load() != uint64(ww.container.Len()) { //nolint:gosec - ww.mu.RUnlock() + idle := ww.numWorkers.Load() == uint64(ww.container.Len()) //nolint:gosec + ww.mu.RUnlock() + if !idle { continue } - ww.mu.RUnlock() - // All workers at this moment are in the container - // Pop operation is blocked; push can't be done, since it's not possible to pop - ww.mu.Lock() - ww.stopWatchedWorkers() - ww.container.ResetDone() - - // todo: rustatian, do we need this mutex? - ww.mu.Unlock() - - return ww.numWorkers.Load() case <-ctx.Done(): - // kill workers - ww.mu.Lock() - ww.stopWatchedWorkers() - ww.container.ResetDone() - ww.mu.Unlock() - - return ww.numWorkers.Load() } + break } + + ww.mu.Lock() + ww.stopWatchedWorkers() + ww.container.ResetDone() + ww.mu.Unlock() + + return ww.numWorkers.Load() } // Destroy all underlying containers (but let them complete the task) @@ -321,33 +308,22 @@ func (ww *WorkerWatcher) Destroy(ctx context.Context) { select { case <-tt: ww.mu.RLock() - // that might be one of the workers is working - if ww.numWorkers.Load() != uint64(ww.container.Len()) { //nolint:gosec - ww.mu.RUnlock() + idle := ww.numWorkers.Load() == uint64(ww.container.Len()) //nolint:gosec + ww.mu.RUnlock() + if !idle { continue } - - ww.mu.RUnlock() - // All workers at this moment are in the container - // Pop operation is blocked, push can't be done, since it's not possible to pop - - ww.mu.Lock() - ww.stopWatchedWorkers() - ww.numWorkers.Store(0) - ww.destroyed.Store(true) - ww.mu.Unlock() - return case <-ctx.Done(): - // kill workers ww.log.Debug("destroy: context canceled", "error", ctx.Err()) - ww.mu.Lock() - ww.stopWatchedWorkers() - ww.numWorkers.Store(0) - ww.destroyed.Store(true) - ww.mu.Unlock() - return } + break } + + ww.mu.Lock() + ww.stopWatchedWorkers() + ww.numWorkers.Store(0) + ww.destroyed.Store(true) + ww.mu.Unlock() } // List - this is O(n) operation, and it will return copy of the actual workers @@ -386,26 +362,17 @@ func (ww *WorkerWatcher) wait(w *worker.Process) { err = ww.Allocate() if err != nil { - // the watcher is shutting down and the dead worker gets no replacement; drop it - // from the count so Destroy's drain can converge - if errors.Is(errors.WatcherStopped, err) { - for { - n := ww.numWorkers.Load() - if n == 0 || ww.numWorkers.CompareAndSwap(n, n-1) { - break - } - } - return - } - - // dead worker was not replaced; saturate at zero so a concurrent removal cannot - // wrap the counter + // The worker has no replacement. Keep the count at or above zero. for { n := ww.numWorkers.Load() if n == 0 || ww.numWorkers.CompareAndSwap(n, n-1) { break } } + if errors.Is(errors.WatcherStopped, err) { + return + } + ww.log.Error("failed to allocate the worker", "internal_event_name", events.EventWorkerError.String(), "error", err) if ww.numWorkers.Load() == 0 { panic("no workers available, can't run the application")