fd24e31912
- Modify unit test commands in the Makefile to include additional skip patterns for memory leak tests, improving test coverage and accuracy. - Expand benchmark and memory leak detection to include the event module, ensuring comprehensive testing across all components. - Add new design and TODO documentation files for the event module to facilitate future development.
382 lines
11 KiB
Go
382 lines
11 KiB
Go
package event_test
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"runtime"
|
||
"sync"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/yaoapp/yao/event"
|
||
"github.com/yaoapp/yao/event/types"
|
||
)
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Shared bench handler: lightweight, simulates minimal real work.
|
||
// ---------------------------------------------------------------------------
|
||
|
||
type benchHandler struct {
|
||
processed atomic.Int64
|
||
}
|
||
|
||
func (h *benchHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) {
|
||
h.processed.Add(1)
|
||
if ev.IsCall {
|
||
resp <- types.Result{Data: "ok"}
|
||
}
|
||
}
|
||
|
||
func (h *benchHandler) Shutdown(ctx context.Context) error { return nil }
|
||
|
||
// benchListener counts received events.
|
||
type benchListener struct {
|
||
received atomic.Int64
|
||
}
|
||
|
||
func (l *benchListener) OnEvent(ev *types.Event) { l.received.Add(1) }
|
||
func (l *benchListener) Shutdown(ctx context.Context) error { return nil }
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Benchmark: Push throughput (no queue, pure worker dispatch)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
func BenchmarkPush_NoQueue(b *testing.B) {
|
||
event.Reset()
|
||
h := &benchHandler{}
|
||
event.Register("bench", h, event.MaxWorkers(512))
|
||
_ = event.Start()
|
||
b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() })
|
||
|
||
ctx := context.Background()
|
||
b.ResetTimer()
|
||
b.RunParallel(func(pb *testing.PB) {
|
||
for pb.Next() {
|
||
_, _ = event.Push(ctx, "bench.work", nil)
|
||
}
|
||
})
|
||
b.StopTimer()
|
||
|
||
// Drain workers
|
||
time.Sleep(100 * time.Millisecond)
|
||
b.ReportMetric(float64(h.processed.Load()), "events_handled")
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Benchmark: Call throughput (no queue, synchronous round-trip)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
func BenchmarkCall_NoQueue(b *testing.B) {
|
||
event.Reset()
|
||
h := &benchHandler{}
|
||
event.Register("bench", h, event.MaxWorkers(512))
|
||
_ = event.Start()
|
||
b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() })
|
||
|
||
ctx := context.Background()
|
||
b.ResetTimer()
|
||
b.RunParallel(func(pb *testing.PB) {
|
||
for pb.Next() {
|
||
_, _, _ = event.Call(ctx, "bench.get", nil)
|
||
}
|
||
})
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Benchmark: Push throughput with Queue (serial per queue)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
func BenchmarkPush_WithQueue(b *testing.B) {
|
||
event.Reset()
|
||
h := &benchHandler{}
|
||
event.Register("bench", h, event.MaxWorkers(512), event.QueueSize(8192))
|
||
_ = event.Start()
|
||
b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() })
|
||
|
||
qID, _ := event.QueueCreate("bench")
|
||
b.Cleanup(func() { event.QueueRelease(qID) })
|
||
|
||
ctx := context.Background()
|
||
b.ResetTimer()
|
||
for i := 0; i < b.N; i++ {
|
||
_, _ = event.Push(ctx, "bench.work", nil, event.Queue(qID))
|
||
}
|
||
b.StopTimer()
|
||
|
||
event.QueueRelease(qID)
|
||
time.Sleep(200 * time.Millisecond)
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Scenario: 1000 concurrent users, each with trace + job queues.
|
||
//
|
||
// Simulates:
|
||
// - 1000 users × 2 queues (trace + job) = 2000 queues
|
||
// - Each user pushes 20 trace events + 5 job events + 1 Call per queue
|
||
// - 200 SSE subscribers watching "trace.*" and "job.*"
|
||
// - 2 Listeners (trace.* + job.*)
|
||
//
|
||
// Reports: total duration, events/sec, memory delta.
|
||
// ---------------------------------------------------------------------------
|
||
|
||
func TestScenario_1000Users(t *testing.T) {
|
||
event.Reset()
|
||
defer event.Reset()
|
||
|
||
traceH := &benchHandler{}
|
||
jobH := &benchHandler{}
|
||
event.Register("trace", traceH, event.MaxWorkers(512), event.ReservedWorkers(20), event.QueueSize(8192))
|
||
event.Register("job", jobH, event.MaxWorkers(256), event.ReservedWorkers(10), event.QueueSize(4096))
|
||
|
||
traceL := &benchListener{}
|
||
jobL := &benchListener{}
|
||
event.Listen("trace.*", traceL)
|
||
event.Listen("job.*", jobL)
|
||
|
||
_ = event.Start()
|
||
defer func() { _ = event.Stop(context.Background()) }()
|
||
|
||
const (
|
||
numUsers = 1000
|
||
tracePushPerUser = 20
|
||
jobPushPerUser = 5
|
||
callsPerQueue = 1
|
||
numSubscribers = 200
|
||
subscriberBufSize = 256
|
||
)
|
||
|
||
// --- Subscribers ---
|
||
subChans := make([]chan *types.Event, numSubscribers)
|
||
subIDs := make([]string, numSubscribers)
|
||
for i := 0; i < numSubscribers; i++ {
|
||
ch := make(chan *types.Event, subscriberBufSize)
|
||
subChans[i] = ch
|
||
pattern := "trace.*"
|
||
if i%2 == 1 {
|
||
pattern = "job.*"
|
||
}
|
||
subIDs[i] = event.Subscribe(pattern, ch)
|
||
}
|
||
defer func() {
|
||
for _, id := range subIDs {
|
||
event.Unsubscribe(id)
|
||
}
|
||
}()
|
||
|
||
// Drain subscribers in background
|
||
var subReceived atomic.Int64
|
||
subDone := make(chan struct{})
|
||
go func() {
|
||
defer close(subDone)
|
||
for _, ch := range subChans {
|
||
go func(c chan *types.Event) {
|
||
for range c {
|
||
subReceived.Add(1)
|
||
}
|
||
}(ch)
|
||
}
|
||
}()
|
||
|
||
// --- Memory before ---
|
||
runtime.GC()
|
||
var memBefore runtime.MemStats
|
||
runtime.ReadMemStats(&memBefore)
|
||
|
||
// --- Run ---
|
||
start := time.Now()
|
||
var wg sync.WaitGroup
|
||
|
||
for u := 0; u < numUsers; u++ {
|
||
wg.Add(1)
|
||
go func(userID int) {
|
||
defer wg.Done()
|
||
ctx := event.WithSID(context.Background(), fmt.Sprintf("sess-%d", userID))
|
||
ctx = event.WithAuth(ctx, &types.AuthorizedInfo{UserID: fmt.Sprintf("u-%d", userID)})
|
||
|
||
// Create trace queue
|
||
traceQID, err := event.QueueCreate("trace")
|
||
if err != nil {
|
||
t.Errorf("user %d: trace QueueCreate: %v", userID, err)
|
||
return
|
||
}
|
||
|
||
// Create job queue
|
||
jobQID, err := event.QueueCreate("job")
|
||
if err != nil {
|
||
t.Errorf("user %d: job QueueCreate: %v", userID, err)
|
||
event.QueueRelease(traceQID)
|
||
return
|
||
}
|
||
|
||
// Push trace events
|
||
for i := 0; i < tracePushPerUser; i++ {
|
||
_, _ = event.Push(ctx, "trace.add", i, event.Queue(traceQID))
|
||
}
|
||
|
||
// Push job events
|
||
for i := 0; i < jobPushPerUser; i++ {
|
||
_, _ = event.Push(ctx, "job.progress", i, event.Queue(jobQID))
|
||
}
|
||
|
||
// Call on each queue
|
||
for i := 0; i < callsPerQueue; i++ {
|
||
callCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||
_, _, _ = event.Call(callCtx, "trace.get", nil, event.Queue(traceQID))
|
||
cancel()
|
||
|
||
callCtx2, cancel2 := context.WithTimeout(ctx, 5*time.Second)
|
||
_, _, _ = event.Call(callCtx2, "job.status", nil, event.Queue(jobQID))
|
||
cancel2()
|
||
}
|
||
|
||
// Release queues
|
||
event.QueueRelease(traceQID)
|
||
event.QueueRelease(jobQID)
|
||
}(u)
|
||
}
|
||
|
||
wg.Wait()
|
||
elapsed := time.Since(start)
|
||
|
||
// Wait for queues to drain
|
||
time.Sleep(500 * time.Millisecond)
|
||
|
||
// --- Memory after ---
|
||
runtime.GC()
|
||
var memAfter runtime.MemStats
|
||
runtime.ReadMemStats(&memAfter)
|
||
|
||
// --- Results ---
|
||
totalPush := int64(numUsers) * int64(tracePushPerUser+jobPushPerUser)
|
||
totalCall := int64(numUsers) * int64(callsPerQueue) * 2
|
||
totalEvents := totalPush + totalCall
|
||
traceProcessed := traceH.processed.Load()
|
||
jobProcessed := jobH.processed.Load()
|
||
listenerTrace := traceL.received.Load()
|
||
listenerJob := jobL.received.Load()
|
||
memDeltaMB := float64(memAfter.TotalAlloc-memBefore.TotalAlloc) / 1024 / 1024
|
||
|
||
t.Logf("=== 1000-User Scenario Results ===")
|
||
t.Logf("Users: %d", numUsers)
|
||
t.Logf("Queues created: %d (trace: %d, job: %d)", numUsers*2, numUsers, numUsers)
|
||
t.Logf("Subscribers: %d", numSubscribers)
|
||
t.Logf("Total events: %d (push: %d, call: %d)", totalEvents, totalPush, totalCall)
|
||
t.Logf("Trace processed: %d", traceProcessed)
|
||
t.Logf("Job processed: %d", jobProcessed)
|
||
t.Logf("Listener trace: %d", listenerTrace)
|
||
t.Logf("Listener job: %d", listenerJob)
|
||
t.Logf("Sub received: %d", subReceived.Load())
|
||
t.Logf("Elapsed: %v", elapsed)
|
||
t.Logf("Throughput: %.0f events/sec", float64(totalEvents)/elapsed.Seconds())
|
||
t.Logf("Memory delta: %.2f MB (TotalAlloc)", memDeltaMB)
|
||
|
||
// --- Assertions ---
|
||
expectedProcessed := totalPush + totalCall
|
||
actualProcessed := traceProcessed + jobProcessed
|
||
if actualProcessed < expectedProcessed {
|
||
t.Errorf("processed %d < expected %d (some events lost)", actualProcessed, expectedProcessed)
|
||
}
|
||
|
||
if elapsed > 30*time.Second {
|
||
t.Errorf("scenario took %v, expected < 30s", elapsed)
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Benchmark: Queue create/release churn (lifecycle overhead)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
func BenchmarkQueueCreateRelease(b *testing.B) {
|
||
event.Reset()
|
||
h := &benchHandler{}
|
||
event.Register("bench", h, event.QueueSize(64))
|
||
_ = event.Start()
|
||
b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() })
|
||
|
||
b.ResetTimer()
|
||
b.RunParallel(func(pb *testing.PB) {
|
||
for pb.Next() {
|
||
qID, err := event.QueueCreate("bench")
|
||
if err != nil {
|
||
b.Fatalf("QueueCreate: %v", err)
|
||
}
|
||
event.QueueRelease(qID)
|
||
}
|
||
})
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Benchmark: Subscriber notify throughput (fanout to 200 subscribers)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
func BenchmarkSubscriberFanout(b *testing.B) {
|
||
event.Reset()
|
||
h := &benchHandler{}
|
||
event.Register("bench", h, event.MaxWorkers(512))
|
||
_ = event.Start()
|
||
b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() })
|
||
|
||
const numSubs = 200
|
||
for i := 0; i < numSubs; i++ {
|
||
ch := make(chan *types.Event, 1024)
|
||
event.Subscribe("bench.*", ch)
|
||
go func(c chan *types.Event) {
|
||
for range c {
|
||
}
|
||
}(ch)
|
||
}
|
||
|
||
ctx := context.Background()
|
||
b.ResetTimer()
|
||
b.RunParallel(func(pb *testing.PB) {
|
||
for pb.Next() {
|
||
_, _ = event.Push(ctx, "bench.work", nil)
|
||
}
|
||
})
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Benchmark: Mixed Push/Call with 2000 queues (1000 users × 2)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
func BenchmarkMixed_2000Queues(b *testing.B) {
|
||
event.Reset()
|
||
h := &benchHandler{}
|
||
event.Register("mix", h, event.MaxWorkers(512), event.ReservedWorkers(20), event.QueueSize(4096))
|
||
_ = event.Start()
|
||
b.Cleanup(func() { _ = event.Stop(context.Background()); event.Reset() })
|
||
|
||
const numQueues = 2000
|
||
queueIDs := make([]string, numQueues)
|
||
for i := 0; i < numQueues; i++ {
|
||
qID, err := event.QueueCreate("mix")
|
||
if err != nil {
|
||
b.Fatalf("QueueCreate %d: %v", i, err)
|
||
}
|
||
queueIDs[i] = qID
|
||
}
|
||
b.Cleanup(func() {
|
||
for _, qID := range queueIDs {
|
||
event.QueueRelease(qID)
|
||
}
|
||
time.Sleep(200 * time.Millisecond)
|
||
})
|
||
|
||
ctx := context.Background()
|
||
b.ResetTimer()
|
||
b.RunParallel(func(pb *testing.PB) {
|
||
i := 0
|
||
for pb.Next() {
|
||
qID := queueIDs[i%numQueues]
|
||
if i%10 == 0 {
|
||
callCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
||
_, _, _ = event.Call(callCtx, "mix.get", nil, event.Queue(qID))
|
||
cancel()
|
||
} else {
|
||
_, _ = event.Push(ctx, "mix.work", nil, event.Queue(qID))
|
||
}
|
||
i++
|
||
}
|
||
})
|
||
}
|