Supervised pool

1 files changed, 145 insertions, 0 deletions

diff --git a/supervisor_test.go b/supervisor_test.go
new file mode 100644
index 00000000..68c2ddaf
--- /dev/null
+++ b/supervisor_test.go
@@ -0,0 +1,145 @@
+package roadrunner
+
+import (
+	"context"
+	"os/exec"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var cfgSupervised = Config{
+	NumWorkers:      int64(1),
+	AllocateTimeout: time.Second,
+	DestroyTimeout:  time.Second,
+	Supervisor: &SupervisorConfig{
+		WatchTick:       1,
+		TTL:             100,
+		IdleTTL:         100,
+		ExecTTL:         100,
+		MaxWorkerMemory: 100,
+	},
+}
+
+func TestSupervisedPool_Exec(t *testing.T) {
+	ctx := context.Background()
+	p, err := NewPool(
+		ctx,
+		func() *exec.Cmd { return exec.Command("php", "tests/memleak.php", "pipes") },
+		NewPipeFactory(),
+		cfgSupervised,
+	)
+
+	assert.NoError(t, err)
+	assert.NotNil(t, p)
+	stopCh := make(chan struct{})
+	defer p.Destroy(context.Background())
+
+	go func() {
+		for ; ; {
+			select {
+			case <-stopCh:
+				return
+			default:
+				workers := p.Workers()
+				if len(workers) > 0 {
+					s, err := WorkerProcessState(workers[0])
+					assert.NoError(t, err)
+					assert.NotNil(t, s)
+					// since this is soft limit, double max memory limit watch
+					if (s.MemoryUsage / MB) > cfgSupervised.Supervisor.MaxWorkerMemory*2 {
+						t.Fatal("max memory reached")
+					}
+				}
+			}
+		}
+	}()
+
+	for i := 0; i < 100; i++ {
+		time.Sleep(time.Millisecond * 50)
+		_, err = p.Exec(Payload{
+			Context: []byte(""),
+			Body:    []byte("foo"),
+		})
+		assert.NoError(t, err)
+	}
+
+	stopCh <- struct{}{}
+}
+
+func TestSupervisedPool_ExecTTL_TimedOut(t *testing.T) {
+	var cfgExecTTL = Config{
+		NumWorkers:      int64(1),
+		AllocateTimeout: time.Second,
+		DestroyTimeout:  time.Second,
+		Supervisor: &SupervisorConfig{
+			WatchTick:       1,
+			TTL:             100,
+			IdleTTL:         100,
+			ExecTTL:         1,
+			MaxWorkerMemory: 100,
+		},
+	}
+	ctx := context.Background()
+	p, err := NewPool(
+		ctx,
+		func() *exec.Cmd { return exec.Command("php", "tests/sleep.php", "pipes") },
+		NewPipeFactory(),
+		cfgExecTTL,
+	)
+
+	assert.NoError(t, err)
+	assert.NotNil(t, p)
+	defer p.Destroy(context.Background())
+
+	pid := p.Workers()[0].Pid()
+
+	resp, err := p.ExecWithContext(context.Background(), Payload{
+		Context: []byte(""),
+		Body:    []byte("foo"),
+	})
+
+	assert.Error(t, err)
+	assert.Empty(t, resp)
+
+	time.Sleep(time.Second * 3)
+	// should be new worker with new pid
+	assert.NotEqual(t, pid, p.Workers()[0].Pid())
+}
+
+func
+TestSupervisedPool_ExecTTL_OK(t *testing.T) {
+	var cfgExecTTL = Config{
+		NumWorkers:      int64(1),
+		AllocateTimeout: time.Second,
+		DestroyTimeout:  time.Second,
+		Supervisor: &SupervisorConfig{
+			WatchTick:       1,
+			TTL:             100,
+			IdleTTL:         100,
+			ExecTTL:         4,
+			MaxWorkerMemory: 100,
+		},
+	}
+	ctx := context.Background()
+	p, err := NewPool(
+		ctx,
+		func() *exec.Cmd { return exec.Command("php", "tests/sleep.php", "pipes") },
+		NewPipeFactory(),
+		cfgExecTTL,
+	)
+
+	assert.NoError(t, err)
+	assert.NotNil(t, p)
+	defer p.Destroy(context.Background())
+
+	time.Sleep(time.Millisecond * 100)
+	resp, err := p.Exec(Payload{
+		Context: []byte(""),
+		Body:    []byte("foo"),
+	})
+
+	assert.NoError(t, err)
+	assert.Empty(t, resp)
+}