diff --git a/pkg/pyproc/cancellation_test.go b/pkg/pyproc/cancellation_test.go index 3b2d841..a4edef7 100644 --- a/pkg/pyproc/cancellation_test.go +++ b/pkg/pyproc/cancellation_test.go @@ -9,7 +9,7 @@ import ( ) // createTestPool creates a new pool for testing -func createTestPool(t *testing.T, id string) *Pool { +func createTestPoolWithConfig(t *testing.T, id string, workers int, maxInFlight int, perWorker int) *Pool { requireUnixSocket(t) // Use /tmp directly with short names to avoid 104 char Unix socket path limit on macOS // Note: pool.go adds "-0" for worker ID, so keep the base path very short @@ -19,9 +19,10 @@ func createTestPool(t *testing.T, id string) *Pool { poolOpts := PoolOptions{ Config: PoolConfig{ - Workers: 1, - MaxInFlight: 3, // Allow 3 concurrent requests for the concurrent test - HealthInterval: 100 * time.Millisecond, + Workers: workers, + MaxInFlight: maxInFlight, + MaxInFlightPerWorker: perWorker, + HealthInterval: 100 * time.Millisecond, }, WorkerConfig: WorkerConfig{ ID: id, @@ -59,6 +60,10 @@ func createTestPool(t *testing.T, id string) *Pool { return pool } +func createTestPool(t *testing.T, id string) *Pool { + return createTestPoolWithConfig(t, id, 1, 3, 1) +} + // TestContextCancellation tests that context cancellation propagates to Python workers func TestContextCancellation(t *testing.T) { // Skip if running in CI without Python @@ -121,7 +126,7 @@ func TestContextCancellation(t *testing.T) { }) t.Run("MultipleConcurrentCancellations", func(t *testing.T) { - pool := createTestPool(t, "mc") + pool := createTestPoolWithConfig(t, "mc", 3, 3, 1) t.Cleanup(func() { if err := pool.Shutdown(context.Background()); err != nil { t.Errorf("Failed to shutdown pool: %v", err) diff --git a/pkg/pyproc/pool_concurrency_test.go b/pkg/pyproc/pool_concurrency_test.go new file mode 100644 index 0000000..9212533 --- /dev/null +++ b/pkg/pyproc/pool_concurrency_test.go @@ -0,0 +1,149 @@ +package pyproc + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/YuminosukeSato/pyproc/internal/protocol" +) + +func TestPoolCall_SerializesPerWorker(t *testing.T) { + paths := []string{ + tempSocketPath(t, "serial-w0"), + tempSocketPath(t, "serial-w1"), + } + var collisions atomic.Int32 + for _, path := range paths { + path := path + var inflight atomic.Int32 + stop := startUnixServer(t, path, func(req protocol.Request) *protocol.Response { + if inflight.Add(1) > 1 { + collisions.Add(1) + } + time.Sleep(120 * time.Millisecond) + inflight.Add(-1) + resp, _ := protocol.NewResponse(req.ID, map[string]bool{"ok": true}) + return resp + }) + t.Cleanup(stop) + } + + workers := []workerHandle{ + newStubWorker(paths[0], true), + newStubWorker(paths[1], true), + } + p := newPoolWithWorkers(PoolConfig{ + Workers: 2, + MaxInFlight: 4, + MaxInFlightPerWorker: 1, + HealthInterval: 10 * time.Millisecond, + }, workers) + for _, pw := range p.workers { + pw.healthy.Store(true) + } + + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := p.Call(ctx, "echo", map[string]string{"msg": "x"}, &map[string]any{}); err != nil { + t.Errorf("call failed: %v", err) + } + }() + } + wg.Wait() + + if collisions.Load() != 0 { + t.Fatalf("expected no concurrent requests per worker, got %d", collisions.Load()) + } +} + +func TestPoolCall_OversubscribeBlocksWithContext(t *testing.T) { + path := tempSocketPath(t, "oversub") + firstStarted := make(chan struct{}) + var reqCount atomic.Int32 + stop := startUnixServer(t, path, func(req protocol.Request) *protocol.Response { + if reqCount.Add(1) == 1 { + close(firstStarted) + time.Sleep(200 * time.Millisecond) + } + resp, _ := protocol.NewResponse(req.ID, map[string]bool{"ok": true}) + return resp + }) + t.Cleanup(stop) + + workers := []workerHandle{newStubWorker(path, true)} + p := newPoolWithWorkers(PoolConfig{ + Workers: 1, + MaxInFlight: 2, + MaxInFlightPerWorker: 1, + }, workers) + p.workers[0].healthy.Store(true) + + firstErrCh := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + firstErrCh <- p.Call(ctx, "echo", map[string]string{"msg": "first"}, &map[string]any{}) + }() + + <-firstStarted + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + err := p.Call(ctx, "echo", map[string]string{"msg": "second"}, &map[string]any{}) + if err == nil { + t.Fatal("expected deadline exceeded for oversubscribed call") + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected deadline exceeded, got %v", err) + } + + if firstErr := <-firstErrCh; firstErr != nil { + t.Fatalf("first call failed: %v", firstErr) + } +} + +func TestPoolCall_ShutdownConcurrent(t *testing.T) { + path := tempSocketPath(t, "shutdown") + started := make(chan struct{}) + stop := startUnixServer(t, path, func(req protocol.Request) *protocol.Response { + close(started) + time.Sleep(150 * time.Millisecond) + resp, _ := protocol.NewResponse(req.ID, map[string]bool{"ok": true}) + return resp + }) + t.Cleanup(stop) + + workers := []workerHandle{newStubWorker(path, true)} + p := newPoolWithWorkers(PoolConfig{ + Workers: 1, + MaxInFlight: 1, + MaxInFlightPerWorker: 1, + }, workers) + p.workers[0].healthy.Store(true) + + callErrCh := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + callErrCh <- p.Call(ctx, "echo", map[string]string{"msg": "x"}, &map[string]any{}) + }() + + <-started + + if err := p.Shutdown(context.Background()); err != nil { + t.Fatalf("shutdown failed: %v", err) + } + + if callErr := <-callErrCh; callErr != nil { + t.Fatalf("call failed during shutdown: %v", callErr) + } +} diff --git a/pkg/pyproc/pool_error_test.go b/pkg/pyproc/pool_error_test.go index e5c65d7..ddecfd1 100644 --- a/pkg/pyproc/pool_error_test.go +++ b/pkg/pyproc/pool_error_test.go @@ -165,6 +165,28 @@ func TestPool_DefaultMaxInFlight(t *testing.T) { } } +func TestPool_DefaultMaxInFlightPerWorker(t *testing.T) { + opts := PoolOptions{ + Config: PoolConfig{ + Workers: 1, + MaxInFlight: 1, + MaxInFlightPerWorker: 0, + }, + WorkerConfig: WorkerConfig{ + SocketPath: "/tmp/test.sock", + }, + } + + pool, err := NewPool(opts, nil) + if err != nil { + t.Fatalf("NewPool failed: %v", err) + } + + if pool.opts.Config.MaxInFlightPerWorker != 1 { + t.Errorf("expected default MaxInFlightPerWorker 1, got %d", pool.opts.Config.MaxInFlightPerWorker) + } +} + func TestPool_DefaultHealthInterval(t *testing.T) { opts := PoolOptions{ Config: PoolConfig{