From 02123df8cc15896f97c1580e49d8e2a991e37ddd Mon Sep 17 00:00:00 2001 From: sangtran Date: Thu, 3 Sep 2026 23:10:33 +0700 Subject: Moving all legacy quacker source code from github to cgit --- fanout/example_test.go | 45 +++++ fanout/fanout.go | 203 ++++++++++++++++++++++ fanout/fanout_bench_test.go | 65 +++++++ fanout/fanout_test.go | 412 ++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 725 insertions(+) create mode 100644 fanout/example_test.go create mode 100644 fanout/fanout.go create mode 100644 fanout/fanout_bench_test.go create mode 100644 fanout/fanout_test.go (limited to 'fanout') diff --git a/fanout/example_test.go b/fanout/example_test.go new file mode 100644 index 0000000..50e7e96 --- /dev/null +++ b/fanout/example_test.go @@ -0,0 +1,45 @@ +package fanout_test + +import ( + "context" + "fmt" + "sort" + "sync" + + "git.sangtran.io.vn/quacker.git/fanout" +) + +func ExampleNewFanOut() { + fo, _ := fanout.NewFanOut[int]( + fanout.WithWorkerCount(2), + fanout.WithBufferSize(10), + fanout.WithStrategy(fanout.RoundRobin), + ) + + input := make(chan int, 4) + for i := 1; i <= 4; i++ { + input <- i + } + close(input) + + fo.Run(context.Background(), input) + + var mu sync.Mutex + var all []int + var wg sync.WaitGroup + + for _, ch := range fo.Outputs() { + wg.Go(func() { + for v := range ch { + mu.Lock() + all = append(all, v) + mu.Unlock() + } + }) + } + + wg.Wait() + sort.Ints(all) + fmt.Println(all) + // Output: [1 2 3 4] +} diff --git a/fanout/fanout.go b/fanout/fanout.go new file mode 100644 index 0000000..1cd2043 --- /dev/null +++ b/fanout/fanout.go @@ -0,0 +1,203 @@ +// Package fanout distributes values from a single input channel to multiple output channels. +package fanout + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" +) + +// FanOutObserver defines callbacks for monitoring fan-out lifecycle events. +type FanOutObserver interface { + OnDistribute() + OnOutputClosed() +} + +// FanOutStrategy defines how values are distributed across output channels. +type FanOutStrategy int + +const ( + // RoundRobin distributes values evenly across output channels in rotation. + RoundRobin FanOutStrategy = iota + // Broadcast sends each value to all output channels. + Broadcast +) + +// ErrInvalidStrategy is returned when an unknown FanOutStrategy is provided. +var ErrInvalidStrategy = errors.New("fanout: unknown distribution strategy") + +// FanOutConfig holds configuration options for creating a FanOut instance. +type FanOutConfig struct { + WorkerCount int + BufferSize int + Strategy FanOutStrategy + Observer FanOutObserver +} + +// Option is a functional option for configuring a FanOut instance. +type Option func(*FanOutConfig) + +// FanOut distributes values from a single input channel to multiple output channels. +type FanOut[T any] struct { + cfg *FanOutConfig + outputs []chan T + running atomic.Bool + cachedOutputs []<-chan T + once sync.Once +} + +// NewFanOut creates a new FanOut instance with the provided options. +func NewFanOut[T any](opts ...Option) (*FanOut[T], error) { + cfg := &FanOutConfig{ + WorkerCount: 1, + BufferSize: 0, + Observer: nil, + Strategy: RoundRobin, + } + + for _, opt := range opts { + opt(cfg) + } + + if cfg.WorkerCount < 1 { + return nil, fmt.Errorf("fanout: worker count must be at least 1, got %d", cfg.WorkerCount) + } + + if cfg.BufferSize < 0 { + return nil, fmt.Errorf("fanout: buffer size must not be negative, got %d", cfg.BufferSize) + } + + if cfg.Strategy != RoundRobin && cfg.Strategy != Broadcast { + return nil, ErrInvalidStrategy + } + + outputs := make([]chan T, cfg.WorkerCount) + for i := range outputs { + outputs[i] = make(chan T, cfg.BufferSize) + } + + return &FanOut[T]{ + cfg: cfg, + outputs: outputs, + }, nil +} + +// Run starts distributing values from the input channel to all output channels. +func (f *FanOut[T]) Run(ctx context.Context, input <-chan T) { + if input == nil { + panic("fanout: input channel must not be nil") + } + + if !f.running.CompareAndSwap(false, true) { + panic("fanout: Run() called multiple times") + } + + go func() { + idx := 0 + defer f.running.Store(false) + defer f.closeAllOutput() + for { + select { + case value, ok := <-input: + if !ok { + return + } + switch f.cfg.Strategy { + case RoundRobin: + f.roundRobin(ctx, value, &idx) + case Broadcast: + f.broadcast(ctx, value) + } + case <-ctx.Done(): + return + } + } + }() +} + +// Outputs returns receive-only channel references for the distributed streams. +func (f *FanOut[T]) Outputs() []<-chan T { + f.once.Do(func() { + f.cachedOutputs = make([]<-chan T, len(f.outputs)) + for i, ch := range f.outputs { + f.cachedOutputs[i] = ch + } + }) + + return f.cachedOutputs +} + +func safeObserve(fn func()) { + defer func() { recover() }() + fn() +} + +func (f *FanOut[T]) roundRobin(ctx context.Context, value T, index *int) { + target := *index % f.cfg.WorkerCount + select { + case f.outputs[target] <- value: + *index = (*index + 1) % f.cfg.WorkerCount + if f.cfg.Observer != nil { + safeObserve(f.cfg.Observer.OnDistribute) + } + case <-ctx.Done(): + return + } +} + +func (f *FanOut[T]) broadcast(ctx context.Context, value T) { + var wg sync.WaitGroup + for _, ch := range f.outputs { + wg.Go(func() { + select { + case ch <- value: + if f.cfg.Observer != nil { + safeObserve(f.cfg.Observer.OnDistribute) + } + case <-ctx.Done(): + return + } + }) + } + wg.Wait() +} + +func (f *FanOut[T]) closeAllOutput() { + for _, ch := range f.outputs { + close(ch) + } + + if f.cfg.Observer != nil { + safeObserve(f.cfg.Observer.OnOutputClosed) + } +} + +// WithWorkerCount sets the number of output channels for the fan-out. +func WithWorkerCount(n int) Option { + return func(cfg *FanOutConfig) { + cfg.WorkerCount = n + } +} + +// WithBufferSize sets the buffer capacity for each output channel. +func WithBufferSize(size int) Option { + return func(cfg *FanOutConfig) { + cfg.BufferSize = size + } +} + +// WithStrategy sets the distribution strategy for the fan-out. +func WithStrategy(strategy FanOutStrategy) Option { + return func(cfg *FanOutConfig) { + cfg.Strategy = strategy + } +} + +// WithObserver attaches a FanOutObserver to receive lifecycle event callbacks. +func WithObserver(observer FanOutObserver) Option { + return func(cfg *FanOutConfig) { + cfg.Observer = observer + } +} diff --git a/fanout/fanout_bench_test.go b/fanout/fanout_bench_test.go new file mode 100644 index 0000000..0b88d87 --- /dev/null +++ b/fanout/fanout_bench_test.go @@ -0,0 +1,65 @@ +package fanout + +import ( + "context" + "testing" +) + +func BenchmarkFanOut_RoundRobin(b *testing.B) { + b.ReportAllocs() + + fo, _ := NewFanOut[int]( + WithWorkerCount(4), + WithBufferSize(64), + WithStrategy(RoundRobin), + ) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + input := make(chan int, 64) + fo.Run(ctx, input) + + // Drain outputs + for _, ch := range fo.Outputs() { + go func() { + for range ch { + } + }() + } + + for b.Loop() { + input <- 1 + } + + close(input) +} + +func BenchmarkFanOut_Broadcast(b *testing.B) { + b.ReportAllocs() + + fo, _ := NewFanOut[int]( + WithWorkerCount(4), + WithBufferSize(64), + WithStrategy(Broadcast), + ) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + input := make(chan int, 64) + fo.Run(ctx, input) + + for _, ch := range fo.Outputs() { + go func() { + for range ch { + } + }() + } + + for b.Loop() { + input <- 1 + } + + close(input) +} diff --git a/fanout/fanout_test.go b/fanout/fanout_test.go new file mode 100644 index 0000000..3f15217 --- /dev/null +++ b/fanout/fanout_test.go @@ -0,0 +1,412 @@ +package fanout + +import ( + "context" + "sync" + "testing" + "time" +) + +type mockObserver struct { + name string + t *testing.T +} + +func (m *mockObserver) OnDistribute() { + m.t.Logf("fanout: %s OnDistribute", m.name) +} + +func (m *mockObserver) OnOutputClosed() { + m.t.Logf("fanout: %s OnOutputClosed", m.name) +} + +func TestFanOut_NewFanOut(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + opts []Option + shouldErr bool + }{ + { + name: "worker size is invalid", + opts: []Option{WithWorkerCount(-10)}, + shouldErr: true, + }, + { + name: "buffer size is invalid", + opts: []Option{WithBufferSize(-10), WithWorkerCount(1)}, + shouldErr: true, + }, + { + name: "valid options", + opts: []Option{WithBufferSize(1), WithWorkerCount(1)}, + shouldErr: false, + }, + { + name: "worker count is zero", + opts: []Option{WithWorkerCount(0)}, + shouldErr: true, + }, + { + name: "valid broadcast strategy", + opts: []Option{WithWorkerCount(5), WithBufferSize(10), WithStrategy(Broadcast)}, + shouldErr: false, + }, + { + name: "invalid strategy", + opts: []Option{WithWorkerCount(1), WithStrategy(FanOutStrategy(99))}, + shouldErr: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + _, err := NewFanOut[int](test.opts...) + if (err != nil) != test.shouldErr { + t.Errorf("validation error = %v, want err %v", err, test.shouldErr) + } + }) + } +} + +func TestFanOut_ConcurrentStress(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + workerCount := 5 + fo, err := NewFanOut[int]( + WithWorkerCount(workerCount), + WithBufferSize(100), + ) + if err != nil { + t.Fatalf("new fanout error: %v", err) + } + + input := make(chan int, 1000) + fo.Run(ctx, input) + + var wg sync.WaitGroup + + for i := 0; i < 10; i++ { + wg.Go(func() { + outs := fo.Outputs() + if len(outs) != workerCount { + t.Errorf("expected %d outputs, got %d", workerCount, len(outs)) + } + }) + } + + for _, ch := range fo.Outputs() { + wg.Go(func() { + for range ch { + } + }) + } + + producers := 10 + messagesPerProducer := 100 + + var producerWg sync.WaitGroup + producerWg.Add(producers) + + for i := 0; i < producers; i++ { + go func() { + defer producerWg.Done() + for j := 0; j < messagesPerProducer; j++ { + select { + case input <- j: + case <-ctx.Done(): + return + } + } + }() + } + + go func() { + producerWg.Wait() + close(input) + }() + + wg.Wait() +} + +func TestFanOut_RunBroadcast(t *testing.T) { + t.Parallel() + var wg sync.WaitGroup + workerCount := 2 + size := 10 + obs := &mockObserver{ + name: "fo broadcast", + t: t, + } + fo, _ := NewFanOut[int]( + WithWorkerCount(workerCount), + WithStrategy(Broadcast), + WithObserver(obs), + ) + + ch := make(chan int, size) + wg.Go(func() { + for i := range size { + ch <- i + } + close(ch) + }) + + fo.Run(t.Context(), ch) + + res := make([]int, workerCount) + for i, cha := range fo.Outputs() { + wg.Go(func() { + count := 0 + for v := range cha { + t.Logf("%v", v) + count++ + res[i] = count + } + }) + } + + wg.Wait() + + expected := size + for i, count := range res { + if count != expected { + t.Errorf("worker %d received %d items, expected %d", i, count, expected) + } + } +} + +func TestFanOut_RunRoundRobin(t *testing.T) { + t.Parallel() + var wg sync.WaitGroup + size := 10 + workerCount := 2 + obs := &mockObserver{ + name: "fo round robin", + t: t, + } + fo, _ := NewFanOut[int]( + WithWorkerCount(workerCount), + WithStrategy(RoundRobin), + WithObserver(obs), + ) + + ch := make(chan int, size) + wg.Go(func() { + for i := range size { + ch <- i + } + close(ch) + }) + + fo.Run(t.Context(), ch) + res := make([]int, workerCount) + for i, cha := range fo.Outputs() { + wg.Go(func() { + count := 0 + for v := range cha { + t.Logf("%v", v) + count++ + res[i] = count + } + }) + } + + wg.Wait() + expected := size / workerCount + + for i, count := range res { + if count != expected { + t.Errorf("worker %d received %d items, expected %d", i, count, expected) + } + } +} + +func TestFanOut_ContextTimeOut(t *testing.T) { + t.Parallel() + fo, _ := NewFanOut[int](WithWorkerCount(2)) + ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + defer cancel() + + input := make(chan int, 10) + fo.Run(ctx, input) + + var wg sync.WaitGroup + done := make(chan struct{}) + for _, ch := range fo.Outputs() { + wg.Go(func() { + for v := range ch { + t.Logf("received %d", v) + } + }) + } + + go func() { + wg.Wait() + close(done) + }() + + go func() { + i := 0 + for { + select { + case input <- i: + i++ + time.Sleep(time.Millisecond) + case <-ctx.Done(): + return + } + } + }() + + select { + case <-done: + case <-time.After(200 * time.Millisecond): + t.Fatal("timeout waiting for cleanup") + } +} + +func TestFanOut_ContextCancel(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + + fo, err := NewFanOut[int](WithWorkerCount(3)) + if err != nil { + t.Fatalf("failed to create fanout: %v", err) + } + + input := make(chan int, 10) + fo.Run(ctx, input) + + var wg sync.WaitGroup + for _, ch := range fo.Outputs() { + wg.Go(func() { + for range ch { + } + }) + } + + cancel() + close(input) + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timeout waiting for shutdown") + } +} + +func TestFanOut_ContextDoneRoundRobin(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + + fo, err := NewFanOut[int]( + WithWorkerCount(3), + WithBufferSize(0), + WithStrategy(RoundRobin), + ) + if err != nil { + t.Fatalf("new fanout error: %v", err) + } + + input := make(chan int, 1) + input <- 1 + fo.Run(ctx, input) + + cancel() + + for i, ch := range fo.Outputs() { + select { + case _, ok := <-ch: + if ok { + for range ch { + } + } + case <-time.After(time.Second): + t.Fatalf("worker %d: timeout waiting for close", i) + } + } +} + +func TestFanOut_ContextDoneBroadcast(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + + fo, err := NewFanOut[int]( + WithWorkerCount(3), + WithStrategy(Broadcast), + WithBufferSize(0), + ) + if err != nil { + t.Fatalf("new fanout error: %v", err) + } + + input := make(chan int, 1) + input <- 1 + fo.Run(ctx, input) + + cancel() + + for i, ch := range fo.Outputs() { + select { + case _, ok := <-ch: + if ok { + for range ch { + } + } + case <-time.After(time.Second): + t.Fatalf("worker %d: timeout waiting for close", i) + } + } +} + +func TestFanOut_RunMultipleTime(t *testing.T) { + t.Parallel() + + ctx := t.Context() + fo, err := NewFanOut[int]( + WithWorkerCount(3), + WithBufferSize(0), + ) + if err != nil { + t.Fatalf("new fanout error: %v", err) + } + + input := make(chan int, 10) + fo.Run(ctx, input) + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic when calling Run() multiple times") + } + }() + + fo.Run(ctx, input) +} + +func TestFanOut_NilInput(t *testing.T) { + t.Parallel() + + fo, err := NewFanOut[int](WithWorkerCount(1)) + if err != nil { + t.Fatalf("failed to create fanout: %v", err) + } + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic when passing nil input channel") + } + }() + + fo.Run(t.Context(), nil) +} -- cgit v1.3.1