actor/dispatcher_test.go

package actor

import (
	"reflect"
	"sync"
	"sync/atomic"
	"testing"
)

// queuedDispatcher deliberately postpones execution until the test takes a
// task. Its mutex protects the test scheduler, not the actor's mailbox.
type queuedDispatcher struct {
	mu        sync.Mutex
	tasks     []func()
	submitted int
}

func (d *queuedDispatcher) Schedule(task func()) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.tasks = append(d.tasks, task)
	d.submitted++
}

func (d *queuedDispatcher) stats() (submitted, pending int) {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.submitted, len(d.tasks)
}

func (d *queuedDispatcher) take(t *testing.T) func() {
	t.Helper()
	d.mu.Lock()
	defer d.mu.Unlock()
	if len(d.tasks) == 0 {
		t.Fatal("dispatcher has no pending task")
	}
	task := d.tasks[0]
	d.tasks[0] = nil
	d.tasks = d.tasks[1:]
	return task
}

func TestQueuedDispatcherOwnsMailboxBeforeTaskStarts(t *testing.T) {
	d := &queuedDispatcher{}
	var got []int
	var active atomic.Int32
	a := New(func(message int) {
		if count := active.Add(1); count != 1 {
			t.Errorf("%d concurrent handlers", count)
		}
		defer active.Add(-1)
		got = append(got, message)
	}, WithFixedCapacity[int](4), WithDispatcher[int](d))
	for message := range 200 {
		a.Send(message)
	}
	if len(got) != 0 || !a.mailbox.running.Load() {
		t.Fatal("Send did not retain consumption ownership for the queued task")
	}
	if submitted, pending := d.stats(); submitted != 1 || pending != 1 {
		t.Fatalf("before execution: submitted=%d pending=%d, want 1, 1", submitted, pending)
	}
	d.take(t)()
	if len(got) != 200 || a.mailbox.running.Load() {
		t.Fatalf("first task did not finish draining: handled=%d", len(got))
	}
	for message, value := range got {
		if message != value {
			t.Fatalf("FIFO: got[%d]=%d", message, value)
		}
	}
	a.Send(200)
	if len(got) != 200 {
		t.Error("next wave ran before its task was executed")
	}
	if submitted, pending := d.stats(); submitted != 2 || pending != 1 {
		t.Fatalf("second wave: submitted=%d pending=%d, want 2, 1", submitted, pending)
	}
	d.take(t)()
	if len(got) != 201 || got[200] != 200 || a.mailbox.running.Load() {
		t.Error("second task did not drain the next wave")
	}
}

func TestSelfSendAndYieldDoNotRescheduleActiveTask(t *testing.T) {
	d := &queuedDispatcher{}
	var a *Actor[int]
	var got []int
	depth, yields := 0, 0
	a = New(func(message int) {
		depth++
		defer func() { depth-- }()
		if depth != 1 {
			t.Errorf("reentered handler at depth %d", depth)
		}
		got = append(got, message)
		if message == 0 {
			a.Send(1001)
		}
	}, WithDispatcher[int](d))
	a.mailbox.yield = func() {
		yields++
		if !a.mailbox.running.Load() {
			t.Error("yield released consumption ownership")
		}
		if yields == 1 {
			a.Send(1000)
		}
		if submitted, pending := d.stats(); submitted != 1 || pending != 0 {
			t.Errorf("active task rescheduled: submitted=%d pending=%d", submitted, pending)
		}
	}
	for message := range 130 {
		a.Send(message)
	}
	d.take(t)()
	want := make([]int, 130, 132)
	for i := range want {
		want[i] = i
	}
	// The first reversal yields before handler 0 runs, so publication of 1000
	// precedes the self-send of 1001. Both follow the already detached batch.
	want = append(want, 1000, 1001)
	if !reflect.DeepEqual(got, want) {
		t.Errorf("yield/self-send publication order: got %v, want %v", got, want)
	}
	if yields == 0 || a.mailbox.running.Load() {
		t.Error("task did not yield and finish")
	}
}

func TestSharedDispatcherKeepsActorsIndependent(t *testing.T) {
	const actors, messages = 3, 100
	d := &queuedDispatcher{}
	var all [actors]*Actor[int]
	var entered, release [actors]chan struct{}
	var count [actors]int
	var active [actors]atomic.Int32
	for i := range actors {
		entered[i], release[i] = make(chan struct{}), make(chan struct{})
		all[i] = New(func(message int) {
			if current := active[i].Add(1); current != 1 {
				t.Errorf("actor %d has %d concurrent handlers", i, current)
			}
			defer active[i].Add(-1)
			if message != count[i] {
				t.Errorf("actor %d: message=%d, want %d", i, message, count[i])
			}
			count[i]++
			if message == 0 {
				close(entered[i])
				<-release[i]
			}
		}, WithFixedCapacity[int](4), WithDispatcher[int](d))
		for message := range messages {
			all[i].Send(message)
		}
	}
	var workers sync.WaitGroup
	for range actors {
		task := d.take(t)
		workers.Add(1)
		go func() {
			defer workers.Done()
			task()
		}()
	}
	for i := range actors {
		await(t, entered[i])
		all[i].Send(messages)
	}
	if submitted, pending := d.stats(); submitted != actors || pending != 0 {
		t.Errorf("shared dispatcher: submitted=%d pending=%d", submitted, pending)
	}
	for i := range actors {
		close(release[i])
	}
	done := make(chan struct{})
	go func() {
		workers.Wait()
		close(done)
	}()
	await(t, done)
	for i := range actors {
		if count[i] != messages+1 || all[i].mailbox.running.Load() {
			t.Errorf("actor %d did not finish independently: handled=%d", i, count[i])
		}
	}
}

func TestDispatcherFuncAsynchronousConcurrentDrain(t *testing.T) {
	const producers, messages, selfMessages = 8, 400, 16
	type message struct{ producer, sequence int }
	var tasks sync.WaitGroup
	var submitted, active atomic.Int32
	d := DispatcherFunc(func(task func()) {
		submitted.Add(1)
		tasks.Add(1)
		go func() {
			defer tasks.Done()
			task()
		}()
	})
	entered, release := make(chan struct{}), make(chan struct{})
	last := make([]int, producers+1)
	seen := make([][]bool, producers+1)
	for producer := range producers + 1 {
		last[producer] = -1
		seen[producer] = make([]bool, messages+1)
	}
	handled, panics := 0, 0
	var a *Actor[message]
	a = New(func(m message) {
		if current := active.Add(1); current != 1 {
			t.Errorf("async actor has %d concurrent handlers", current)
		}
		defer active.Add(-1)
		if m.producer == -1 {
			a.Send(message{producers, 0})
			close(entered)
			<-release
			return
		}
		if m.sequence != last[m.producer]+1 || seen[m.producer][m.sequence] {
			t.Errorf("async producer %d: sequence=%d after %d", m.producer, m.sequence, last[m.producer])
		}
		last[m.producer] = m.sequence
		seen[m.producer][m.sequence] = true
		handled++
		if m.producer == producers {
			if m.sequence+1 < selfMessages {
				a.Send(message{producers, m.sequence + 1})
			}
		} else if m.sequence == 7 {
			panic(m.producer)
		}
	}, WithFixedCapacity[message](4), WithBatchSize[message](8), WithDispatcher[message](d),
		WithPanicHandler(func(m message, value any, stack []byte) {
			panics++
			if value != m.producer || len(stack) == 0 {
				t.Error("async panic callback lost message, panic value, or stack")
			}
			if m.producer == 0 {
				panic("callback failed")
			}
		}))
	a.Send(message{producer: -1})
	await(t, entered)
	var senders sync.WaitGroup
	for producer := range producers {
		senders.Add(1)
		go func() {
			defer senders.Done()
			for sequence := range messages {
				a.Send(message{producer, sequence})
			}
		}()
	}
	sendersDone := make(chan struct{})
	go func() {
		senders.Wait()
		close(sendersDone)
	}()
	await(t, sendersDone)
	if got := submitted.Load(); got != 1 {
		t.Errorf("queued/running async task submitted %d times", got)
	}
	close(release)
	awaitTasks := func() {
		done := make(chan struct{})
		go func() {
			tasks.Wait()
			close(done)
		}()
		await(t, done)
	}
	// Wait for the task itself to return, not just its final handler, before
	// inspecting ownership and the consumer's ordinary state.
	awaitTasks()
	if handled != producers*messages+selfMessages || panics != producers {
		t.Errorf("async drain: handled=%d panics=%d", handled, panics)
	}
	for producer := range producers {
		for sequence := range messages {
			if !seen[producer][sequence] {
				t.Fatalf("async missing producer=%d sequence=%d", producer, sequence)
			}
		}
	}
	if last[producers] != selfMessages-1 {
		t.Error("async self-send chain did not finish")
	}
	a.Send(message{0, messages})
	awaitTasks()
	if submitted.Load() != 2 || handled != producers*messages+selfMessages+1 {
		t.Error("async next wave did not schedule and finish")
	}
	if a.mailbox.running.Load() || !a.mailbox.queue.empty() {
		t.Error("async task returned with consumption ownership or queued messages")
	}
	for i := range a.mailbox.queue.fixed {
		n := &a.mailbox.queue.fixed[i]
		if n.value != (message{}) || n.next != nil {
			t.Errorf("async fixed node %d retains a message or link", i)
		}
	}
}

type nilSafeDispatcher struct{}

func (*nilSafeDispatcher) Schedule(task func()) { task() }

func TestCustomTypedNilDispatcherUsesItsContract(t *testing.T) {
	var dispatcher *nilSafeDispatcher
	handled := false
	a := New(func(int) { handled = true }, WithDispatcher[int](dispatcher))
	a.Send(0)
	if !handled {
		t.Error("valid typed-nil dispatcher was not called")
	}
}

func TestInlineFixedStorageSendDoesNotAllocate(t *testing.T) {
	a := New(func(int) {}, WithFixedCapacity[int](4))
	if allocations := testing.AllocsPerRun(1000, func() { a.Send(1) }); allocations != 0 {
		t.Errorf("light-load inline Send allocated %v times per call", allocations)
	}
}