From 18ba596e862360e9b1f5dec7783ab560c0ea3180 Mon Sep 17 00:00:00 2001 From: Dmitry Verkhoturov Date: Wed, 19 Aug 2026 07:54:41 +0100 Subject: [PATCH] modernise with go fix, drop deprecated rand.Seed go fix ./... replaces interface{} with any, rewrites counted loops as range over int and removes the ch := ch capture obsolete since go 1.22. README and _example are swept by hand as the go tool skips them. rand.Seed is deprecated and no longer needed, the global source is seeded randomly since go 1.20, its time import goes with it. outChs preallocated as the rewritten loop makes the capacity known, flagged by prealloc. --- README.md | 36 ++++++++++---------- _example/main.go | 14 ++++---- flow.go | 31 +++++++++-------- flow_test.go | 76 ++++++++++++++++++++--------------------- options.go | 2 +- pool/cursor.go | 6 ++-- pool/options.go | 2 +- pool/pool.go | 42 +++++++++++------------ pool/pool_test.go | 86 +++++++++++++++++++++++------------------------ pool/store.go | 8 ++--- 10 files changed, 150 insertions(+), 153 deletions(-) diff --git a/README.md b/README.md index 2a3356d..b90162a 100644 --- a/README.md +++ b/README.md @@ -34,10 +34,10 @@ running handlers gracefully and won't keep any goroutine running/leaking. ```go // ReaderHandler creates flow.Handler, reading strings from any io.Reader func ReaderHandler(reader io.Reader) Handler { - return func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { + return func(ctx context.Context, ch chan any) (chan any, func() error) { metrics := flow.GetMetrics(ctx) // metrics collects how many records read with "read" key. - readerCh := make(chan interface{}, 1000) + readerCh := make(chan any, 1000) readerFn := func() error { defer close(readerCh) @@ -70,8 +70,8 @@ func ExampleFlow_flow() { f.Add( // add handlers. Note: handlers can be added directly in New // first handler, generate 100 initial values. - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}, 100) // example of non-async handler + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any, 100) // example of non-async handler for i := 1; i <= 100; i++ { out <- i } @@ -80,8 +80,8 @@ func ExampleFlow_flow() { }, // second handler - picks odd numbers only and multiply - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}) // async handler makes its out channel + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any) // async handler makes its out channel runFn = func() error { defer close(out) // handler should close out channel for e := range in { @@ -102,8 +102,8 @@ func ExampleFlow_flow() { }, // final handler - sum all numbers - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}, 1) + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any, 1) runFn = func() error { defer close(out) sum := 0 @@ -145,8 +145,8 @@ func ExampleFlow_parallel() { f.Add( // generate 100 initial values in single handler - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}, 100) // example of non-async handler + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any, 100) // example of non-async handler for i := 1; i <= 100; i++ { out <- i } @@ -155,8 +155,8 @@ func ExampleFlow_parallel() { }, // multiple all numbers in 10 parallel handlers - f.Parallel(10, func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}) // async handler makes its out channel + f.Parallel(10, func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any) // async handler makes its out channel runFn = func() error { defer close(out) // handler should close out channel for e := range in { @@ -173,7 +173,7 @@ func ExampleFlow_parallel() { }), // print all numbers - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { runFn = func() error { defer close(out) sum := 0 @@ -221,7 +221,7 @@ terminated early, by an error with the default fail-fast or by a canceled contex ### worker function Worker function passed by user and runs in multiple workers (goroutines) concurrently. -This is the function: `type WorkerFn func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error` +This is the function: `type WorkerFn func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error` It takes `inp` parameter, does the job and optionally send result(s) with `SenderFn` to the common results channel. Error will terminate all workers unless `ContinueOnError` set. @@ -235,8 +235,8 @@ is not synchronised internally, it doesn't need to be as only the owning worker ```go type WorkerStore interface { - Set(key string, val interface{}) - Get(key string) (interface{}, bool) + Set(key string, val any) + Get(key string) (any, bool) GetInt(key string) int GetFloat(key string) float64 GetString(key string) string @@ -251,7 +251,7 @@ _alternatively state can be kept outside of workers as a slice of values and acc ### usage ```go - p := pool.New(8, func(ctx context.Context, v interface{}, sendFn pool.SenderFn, ws pool.WorkerStore) error { + p := pool.New(8, func(ctx context.Context, v any, sendFn pool.SenderFn, ws pool.WorkerStore) error { // worker function gets input v, processes it and sends result(s) to the common results channel input, ok := v.(string) // in this case it gets string as input @@ -286,7 +286,7 @@ _alternatively state can be kept outside of workers as a slice of values and acc p.Close() // indicates completion of all inputs }() - var v interface{} + var v any for cursor.Next(ctx, &v) { log.Print(v) // print value } diff --git a/_example/main.go b/_example/main.go index cbc41f8..9c6e0da 100644 --- a/_example/main.go +++ b/_example/main.go @@ -21,7 +21,7 @@ func main() { // seed channel with the list of input files. Usually seeding implemented as a separate handler but for our toy example // filling a buffered channel will do it. - seedCh := make(chan interface{}, 4) + seedCh := make(chan any, 4) seedCh <- "data/input-1.txt" seedCh <- "data/input-2.txt" seedCh <- "data/input-3.txt" @@ -41,10 +41,10 @@ func main() { } // lineSplitHandler gets file names and sends lines of text -func lineSplitHandler(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { +func lineSplitHandler(ctx context.Context, ch chan any) (chan any, func() error) { log.Print("make line split handler") metrics := flow.GetMetrics(ctx) - lineCh := make(chan interface{}, 100) + lineCh := make(chan any, 100) lineFn := func() error { log.Printf("start line split handler %d", flow.CID(ctx)) defer close(lineCh) @@ -89,10 +89,10 @@ type wordsInfo struct { // wordsHandler reads lines of text from the input and sends wordsInfo func wordsHandler(minSize int) flow.Handler { log.Printf("make words handler with minsize=%d", minSize) - return func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { + return func(ctx context.Context, ch chan any) (chan any, func() error) { log.Printf("start words handler %d with minsize=%d", flow.CID(ctx), minSize) metrics := flow.GetMetrics(ctx) - wordsCh := make(chan interface{}, 1000) + wordsCh := make(chan any, 1000) wordsFn := func() error { defer close(wordsCh) count := 0 @@ -120,8 +120,8 @@ func wordsHandler(minSize int) flow.Handler { } // sumHandler reduces all inputs with wordsInfo to the final figures and prints it as result -func sumHandler(_ context.Context, ch chan interface{}) (chan interface{}, func() error) { - nopCh := make(chan interface{}) +func sumHandler(_ context.Context, ch chan any) (chan any, func() error) { + nopCh := make(chan any) sumFn := func() error { log.Printf("start sum handler") defer close(nopCh) diff --git a/flow.go b/flow.go index 86e63ff..e21cf1b 100644 --- a/flow.go +++ b/flow.go @@ -31,8 +31,8 @@ type Flow struct { group *errgroup.Group // all handlers runs in this errgroup ctx context.Context // context used for cancellation - lastCh chan interface{} // last channel in flow - funcs []func() error // all runnable functions + lastCh chan any // last channel in flow + funcs []func() error // all runnable functions fanoutBuffer int // buffer size for fanout activateOnce sync.Once // prevents multiple activations of flow @@ -43,7 +43,7 @@ type Flow struct { // fn will be executed in a separate goroutine. fn is thread-safe and may have mutable state. It will live // all flow lifetime and usually implements read->process->write cycle. If fn returns != nil it indicates // critical failure and will stop, with canceled context, all handlers in the flow. -type Handler func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) +type Handler func(ctx context.Context, in chan any) (out chan any, runFn func() error) // New creates flow object with context and common errgroup. This errgroup used to schedule and cancel all handlers. // options defines non-default parameters. @@ -91,9 +91,9 @@ func (f *Flow) Parallel(concurrent int, handler Handler) Handler { return handler } - return func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { - var outChs []chan interface{} - for n := 0; n < concurrent; n++ { + return func(ctx context.Context, ch chan any) (chan any, func() error) { + outChs := make([]chan any, 0, concurrent) + for n := range concurrent { ctxWithID := context.WithValue(ctx, CidContextKey, n) // put n as id to context for parallel handlers out, fn := handler(ctxWithID, ch) // all parallel handlers read from the same lastCh f.funcs = append(f.funcs, fn) // register runnable with flow executor @@ -115,15 +115,15 @@ func (f *Flow) FanOut(handler Handler, handlers ...Handler) Handler { return handler } - return func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { + return func(ctx context.Context, ch chan any) (chan any, func() error) { handlers = append([]Handler{handler}, handlers...) // add head handler to head - inChs := make([]chan interface{}, len(handlers)) // input channels for forked input from ch - outChs := make([]chan interface{}, len(handlers)) // output channels for merging + inChs := make([]chan any, len(handlers)) // input channels for forked input from ch + outChs := make([]chan any, len(handlers)) // output channels for merging for i := 0; i < len(handlers); i++ { - inChs[i] = make(chan interface{}, f.fanoutBuffer) // buffered to allow async readers + inChs[i] = make(chan any, f.fanoutBuffer) // buffered to allow async readers ctxWithID := context.WithValue(ctx, CidContextKey, i) // keep i as ID for handler in context out, fn := handlers[i](ctxWithID, inChs[i]) // handle forked input f.funcs = append(f.funcs, fn) // register runnable with flow executor @@ -171,7 +171,7 @@ func (f *Flow) Wait() error { // Channel returns last (final) channel in flow. Usually consumers don't need this channel, but can be used // to return some final result(s) -func (f *Flow) Channel() chan interface{} { +func (f *Flow) Channel() chan any { return f.lastCh } @@ -181,15 +181,14 @@ func (f *Flow) Metrics() *Metrics { } // merge gets multiple channels and fan-in to a single output channel -func (f *Flow) merge(ctx context.Context, chs []chan interface{}) (mergeCh chan interface{}, mergeFn func() error) { +func (f *Flow) merge(ctx context.Context, chs []chan any) (mergeCh chan any, mergeFn func() error) { - mergeCh = make(chan interface{}) + mergeCh = make(chan any) mergeFn = func() error { defer close(mergeCh) gr, ctxGroup := errgroup.WithContext(ctx) for _, ch := range chs { - ch := ch gr.Go(func() error { for e := range ch { if err := Send(ctxGroup, mergeCh, e); err != nil { @@ -216,7 +215,7 @@ func CID(ctx context.Context) int { // Send entry to channel or returns error if context canceled. // Shortcut for send-or-fail-on-cancel most handlers implement. -func Send(ctx context.Context, ch chan interface{}, e interface{}) error { +func Send(ctx context.Context, ch chan any, e any) error { select { case ch <- e: return nil @@ -227,7 +226,7 @@ func Send(ctx context.Context, ch chan interface{}, e interface{}) error { // Recv gets entry from the channel or returns error if context canceled. // Shortcut for read-or-fail-on-cancel most handlers implement. -func Recv(ctx context.Context, ch chan interface{}) (interface{}, error) { +func Recv(ctx context.Context, ch chan any) (any, error) { select { case val := <-ch: return val, nil diff --git a/flow_test.go b/flow_test.go index 43852e4..39681be 100644 --- a/flow_test.go +++ b/flow_test.go @@ -61,8 +61,8 @@ func TestFlowWithFanOut(t *testing.T) { f := New() f.Add( - func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { - inp := make(chan interface{}, 100) + func(ctx context.Context, ch chan any) (chan any, func() error) { + inp := make(chan any, 100) for i := 1; i <= 100; i++ { inp <- 1 } @@ -90,8 +90,8 @@ func TestFlowWithFanOutAndParallel(t *testing.T) { f := New(FanOutSize(10)) f.Add( - func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { - inp := make(chan interface{}, 100) + func(ctx context.Context, ch chan any) (chan any, func() error) { + inp := make(chan any, 100) for i := 1; i <= 100; i++ { inp <- 1 } @@ -119,7 +119,7 @@ func TestFlowWithFanOutAndParallel(t *testing.T) { func TestFlowWithInputAndSecondFlow(t *testing.T) { - inp := make(chan interface{}, 100) + inp := make(chan any, 100) for i := 1; i <= 100; i++ { inp <- 1 } @@ -154,8 +154,8 @@ func TestFlowWithTimeout(t *testing.T) { var processed int64 proceed := make(chan struct{}, workers) // releases as many inputs as there are workers and no more - slowHandler := func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { - resCh := make(chan interface{}) + slowHandler := func(ctx context.Context, ch chan any) (chan any, func() error) { + resCh := make(chan any) resFn := func() error { defer close(resCh) for inp := range ch { @@ -178,7 +178,7 @@ func TestFlowWithTimeout(t *testing.T) { return resCh, resFn } - for i := 0; i < workers; i++ { + for range workers { proceed <- struct{}{} } @@ -200,8 +200,8 @@ func TestFlowParallelCanceledOnMerge(t *testing.T) { defer cancel() sent := make(chan struct{}) - emitter := func(ctx context.Context, _ chan interface{}) (chan interface{}, func() error) { - outCh := make(chan interface{}) + emitter := func(ctx context.Context, _ chan any) (chan any, func() error) { + outCh := make(chan any) return outCh, func() error { defer close(outCh) if CID(ctx) != 0 { // the second worker closes its channel without emitting anything @@ -216,8 +216,8 @@ func TestFlowParallelCanceledOnMerge(t *testing.T) { } // consumes nothing from the input channel, so the merged records have nowhere to go - stalled := func(ctx context.Context, _ chan interface{}) (chan interface{}, func() error) { - outCh := make(chan interface{}) + stalled := func(ctx context.Context, _ chan any) (chan any, func() error) { + outCh := make(chan any) return outCh, func() error { defer close(outCh) <-ctx.Done() @@ -234,8 +234,8 @@ func TestFlowParallelCanceledOnMerge(t *testing.T) { assert.EqualError(t, res.Wait(), "context canceled") } -func seedHandler(_ context.Context, _ chan interface{}) (chan interface{}, func() error) { - inp := make(chan interface{}, 100) +func seedHandler(_ context.Context, _ chan any) (chan any, func() error) { + inp := make(chan any, 100) for i := 1; i <= 100; i++ { inp <- i } @@ -245,8 +245,8 @@ func seedHandler(_ context.Context, _ chan interface{}) (chan interface{}, func( func multiplierHandler(mult, cancelOn int) Handler { - fn := func(ctx context.Context, ch chan interface{}) (chan interface{}, func() error) { - resCh := make(chan interface{}) + fn := func(ctx context.Context, ch chan any) (chan any, func() error) { + resCh := make(chan any) metrics := ctx.Value(MetricsContextKey).(*Metrics) resFn := func() error { @@ -277,8 +277,8 @@ func multiplierHandler(mult, cancelOn int) Handler { return fn } -func collectorHandler(ctx context.Context, ch chan interface{}) (chOut chan interface{}, fnRun func() error) { - resCh := make(chan interface{}, 1) +func collectorHandler(ctx context.Context, ch chan any) (chOut chan any, fnRun func() error) { + resCh := make(chan any, 1) calls := 0 resFn := func() error { @@ -315,9 +315,9 @@ func ExampleFlow_flow() { f.Add( // add handlers. Note: handlers can be added directly in New // generate 100 initial values. - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { // example of non-async handler, Add executes it right away, prior to Go call - out = make(chan interface{}, 100) + out = make(chan any, 100) for i := 1; i <= 100; i++ { out <- i } @@ -326,9 +326,9 @@ func ExampleFlow_flow() { }, // pick odd numbers only and multiply - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}) // each handler makes its out channel - runFn = func() error { // async handler returns runnable func + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any) // each handler makes its out channel + runFn = func() error { // async handler returns runnable func defer close(out) // handler should close out channel for e := range in { val := e.(int) @@ -348,8 +348,8 @@ func ExampleFlow_flow() { }, // sum all numbers - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}, 1) + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any, 1) runFn = func() error { defer close(out) sum := 0 @@ -389,8 +389,8 @@ func ExampleFlow_parallel() { f.Add( // generate 100 initial values in single handler - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}, 100) // example of non-async handler + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any, 100) // example of non-async handler for i := 1; i <= 100; i++ { out <- i } @@ -399,8 +399,8 @@ func ExampleFlow_parallel() { }, // multiple all numbers in 10 parallel handlers - f.Parallel(10, func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}) // async handler makes its out channel + f.Parallel(10, func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any) // async handler makes its out channel runFn = func() error { defer close(out) // handler should close out channel for e := range in { @@ -418,7 +418,7 @@ func ExampleFlow_parallel() { }), // print all numbers - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { runFn = func() error { defer close(out) sum := 0 @@ -452,8 +452,8 @@ func ExampleFlow_fanOut() { f.Add( // add handlers. Note: handlers can be added directly in New // generate 100 ones. - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}, 100) // example of non-async handler + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any, 100) // example of non-async handler for i := 1; i <= 100; i++ { out <- 1 } @@ -466,8 +466,8 @@ func ExampleFlow_fanOut() { f.FanOut( // first handler picks odd numbers only and multiply by 2 - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}) // async handler makes its out channel + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any) // async handler makes its out channel runFn = func() error { defer close(out) // handler should close out channel for e := range in { @@ -488,8 +488,8 @@ func ExampleFlow_fanOut() { }, // second handler picks even numbers only and multiply by 3 - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}) // async handler makes its out channel + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any) // async handler makes its out channel runFn = func() error { defer close(out) // handler should close out channel for e := range in { @@ -511,8 +511,8 @@ func ExampleFlow_fanOut() { ), // sum all numbers - func(ctx context.Context, in chan interface{}) (out chan interface{}, runFn func() error) { - out = make(chan interface{}, 1) + func(ctx context.Context, in chan any) (out chan any, runFn func() error) { + out = make(chan any, 1) runFn = func() error { defer close(out) sum := 0 diff --git a/options.go b/options.go index ebd1127..fb04912 100644 --- a/options.go +++ b/options.go @@ -25,7 +25,7 @@ func FanOutSize(size int) Option { // Input functional option defines input channels for first handler in chain. // Can be used to connect multiple flows together or seed flow from the outside, with some external data. -func Input(ch chan interface{}) Option { +func Input(ch chan any) Option { return func(f *Flow) { f.lastCh = ch } diff --git a/pool/cursor.go b/pool/cursor.go index b7ad778..7a72acc 100644 --- a/pool/cursor.go +++ b/pool/cursor.go @@ -14,7 +14,7 @@ type Cursor struct { // Next returns next result from the cursor, ok = false on completion. // Any error saved internally and can be returned by Err call -func (c *Cursor) Next(ctx context.Context, v interface{}) bool { +func (c *Cursor) Next(ctx context.Context, v any) bool { for { select { case resp, ok := <-c.ch: @@ -42,8 +42,8 @@ func (c *Cursor) Next(ctx context.Context, v interface{}) bool { } // All gets all data from the cursor -func (c *Cursor) All(ctx context.Context) (res []interface{}, err error) { - var v interface{} +func (c *Cursor) All(ctx context.Context) (res []any, err error) { + var v any for c.Next(ctx, &v) { res = append(res, v) } diff --git a/pool/options.go b/pool/options.go index 59ca737..b79c8c7 100644 --- a/pool/options.go +++ b/pool/options.go @@ -6,7 +6,7 @@ type Option func(p *Workers) // ChunkFn functional option defines chunk func distributing records to particular workers. // The function should return key string identifying the record. // Record with a given key string guaranteed to be processed by the same worker. -func ChunkFn(chunkFn func(val interface{}) string) Option { +func ChunkFn(chunkFn func(val any) string) Option { return func(p *Workers) { p.chunkFn = chunkFn } diff --git a/pool/pool.go b/pool/pool.go index 1d4d698..dc19907 100644 --- a/pool/pool.go +++ b/pool/pool.go @@ -32,7 +32,6 @@ import ( "hash/crc32" "math/rand" "sync" - "time" "github.com/go-pkgz/flow" "golang.org/x/sync/errgroup" @@ -43,7 +42,7 @@ type Workers struct { poolSize int // number of workers (goroutines) batchSize int // size of batch send to workers - chunkFn func(interface{}) string + chunkFn func(any) string resChanSize int // size of responses channel workerChanSize int // size of worker channels workerFn WorkerFn // worker function @@ -52,9 +51,9 @@ type Workers struct { store []WorkerStore // workers store, per worker ID - buf [][]interface{} + buf [][]any bufLock []sync.Mutex // guards buf, per worker ID - workersCh []chan []interface{} + workersCh []chan []any abortCh chan struct{} // closed on pool termination, releases blocked submits ctx context.Context eg *errgroup.Group @@ -64,14 +63,14 @@ type Workers struct { // response wraps data and error type response struct { - value interface{} // the actual data - err error // optional error + value any // the actual data + err error // optional error } // WorkerStore defines interface for per-worker storage type WorkerStore interface { - Set(key string, val interface{}) - Get(key string) (interface{}, bool) + Set(key string, val any) + Get(key string) (any, bool) GetInt(key string) int GetFloat(key string) float64 GetString(key string) string @@ -85,10 +84,10 @@ type contextKey string const widContextKey contextKey = "worker-id" // WorkerFn processes input record inpRec and optionally sends results to sender func -type WorkerFn func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error +type WorkerFn func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error // SenderFn func called by worker code to publish results -type SenderFn func(val interface{}) error +type SenderFn func(val any) error // CompleteFn processes input record inpRec and optionally sends response to respCh type CompleteFn func(ctx context.Context, store WorkerStore) error @@ -102,8 +101,8 @@ func New(poolSize int, workerFn WorkerFn, options ...Option) *Workers { res := Workers{ poolSize: poolSize, - workersCh: make([]chan []interface{}, poolSize), - buf: make([][]interface{}, poolSize), + workersCh: make([]chan []any, poolSize), + buf: make([][]any, poolSize), bufLock: make([]sync.Mutex, poolSize), abortCh: make(chan struct{}), store: make([]WorkerStore, poolSize), @@ -122,21 +121,20 @@ func New(poolSize int, workerFn WorkerFn, options ...Option) *Workers { // initialize workers channels and batch buffers for id := 0; id < poolSize; id++ { - res.workersCh[id] = make(chan []interface{}, res.workerChanSize) + res.workersCh[id] = make(chan []any, res.workerChanSize) if res.batchSize > 1 { - res.buf[id] = make([]interface{}, 0, poolSize) + res.buf[id] = make([]any, 0, poolSize) } res.store[id] = NewLocalStore() } - rand.Seed(time.Now().UnixNano()) return &res } // Submit record to pool, can be blocked until the record accepted by a worker or the pool terminated. // Submits after termination, i.e. after a worker failed or the context canceled, don't block and the // records may be dropped as the workers are shutting down. -func (p *Workers) Submit(v interface{}) { +func (p *Workers) Submit(v any) { // randomize distribution by default id := rand.Intn(p.poolSize) //nolint gosec @@ -147,7 +145,7 @@ func (p *Workers) Submit(v interface{}) { if p.batchSize <= 1 { // skip all buffering if batch size is 1 or less - p.send(id, append([]interface{}{}, v)) + p.send(id, append([]any{}, v)) return } @@ -157,7 +155,7 @@ func (p *Workers) Submit(v interface{}) { p.buf[id] = append(p.buf[id], v) // add to batch buffer if len(p.buf[id]) >= p.batchSize { // commit copy to workers - cp := make([]interface{}, len(p.buf[id])) + cp := make([]any, len(p.buf[id])) copy(cp, p.buf[id]) p.send(id, cp) p.buf[id] = p.buf[id][:0] // reset size, keep capacity @@ -166,7 +164,7 @@ func (p *Workers) Submit(v interface{}) { // send records to the worker with a given id, drops them if the pool terminated and nothing reads workers channels. // Both cases can be ready on termination, i.e. a shutting down worker may still get the records. -func (p *Workers) send(id int, vals []interface{}) { +func (p *Workers) send(id int, vals []any) { select { case p.workersCh[id] <- vals: case <-p.abortCh: @@ -184,7 +182,7 @@ func (p *Workers) Go(ctx context.Context) (Cursor, error) { p.ctx = context.WithValue(ctx, flow.MetricsContextKey, flow.NewMetrics()) var egCtx context.Context p.eg, egCtx = errgroup.WithContext(ctx) - worker := func(id int, inCh chan []interface{}) func() error { + worker := func(id int, inCh chan []any) func() error { return func() error { wCtx := context.WithValue(p.ctx, widContextKey, id) for { @@ -301,8 +299,8 @@ func (p *Workers) Wait(ctx context.Context) error { } // sendResponseFn makes sender func used by worker with the given context and response channel -func (p *Workers) sendResponseFn(ctx context.Context, respCh chan response) func(val interface{}) error { - return func(val interface{}) error { +func (p *Workers) sendResponseFn(ctx context.Context, respCh chan response) func(val any) error { + return func(val any) error { select { case respCh <- response{value: val}: return nil diff --git a/pool/pool_test.go b/pool/pool_test.go index 4dfa699..72f59a8 100644 --- a/pool/pool_test.go +++ b/pool/pool_test.go @@ -26,7 +26,7 @@ func TestPool(t *testing.T) { Fld string } - worker := func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + worker := func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { rec := v.(inpRec) if rec.Num%10 == 0 { err := sender(fmt.Sprintf("%s-%03d", rec.Fld, rec.Num)) @@ -38,7 +38,7 @@ func TestPool(t *testing.T) { var opts []Option if chunks { - opts = append(opts, ChunkFn(func(v interface{}) string { + opts = append(opts, ChunkFn(func(v any) string { return v.(inpRec).Fld })) } @@ -50,7 +50,7 @@ func TestPool(t *testing.T) { require.NoError(t, err) go func() { - for i := 0; i < 1000; i++ { + for i := range 1000 { p.Submit(inpRec{Num: i, Fld: fmt.Sprintf("val-%03d", i)}) } p.Close() @@ -58,7 +58,7 @@ func TestPool(t *testing.T) { n := 0 var res []string - var v interface{} + var v any for cursor.Next(ctx, &v) { log.Printf("%+v", v) res = append(res, v.(string)) @@ -130,14 +130,14 @@ func TestPoolWithStruct(t *testing.T) { k5 bool } - p := New(4, func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error { + p := New(4, func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error { i := inpRec.(int) r := resp{k1: "rec" + strconv.Itoa(i), k2: i, k3: "something", k4: []string{"foo", "bar"}, k5: true} return sender(r) }) go func() { - for i := 0; i < 1000; i++ { + for i := range 1000 { p.Submit(i) time.Sleep(time.Millisecond * time.Duration(rand.Intn(3))) //nolint gosec } @@ -161,7 +161,7 @@ func TestPoolWithStruct(t *testing.T) { func TestPoolWithStore(t *testing.T) { - worker := func(ctx context.Context, v interface{}, send SenderFn, store WorkerStore) error { + worker := func(ctx context.Context, v any, send SenderFn, store WorkerStore) error { store.Set("counter", store.GetInt("counter")+1) Metrics(ctx).Add("c", 1) require.NoError(t, send("something")) @@ -184,7 +184,7 @@ func TestPoolWithStore(t *testing.T) { require.NoError(t, err) go func() { - for i := 0; i < 1000; i++ { + for range 1000 { p.Submit("line") time.Sleep(time.Millisecond * time.Duration(rand.Intn(3))) //nolint gosec } @@ -202,7 +202,7 @@ func TestPoolWithStore(t *testing.T) { func TestPoolWaitDrainsResults(t *testing.T) { - worker := func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + worker := func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { return sender(v) } @@ -211,7 +211,7 @@ func TestPoolWaitDrainsResults(t *testing.T) { require.NoError(t, err) go func() { - for i := 0; i < 100; i++ { + for i := range 100 { p.Submit(i) } p.Close() @@ -224,7 +224,7 @@ func TestPoolWaitDrainsResults(t *testing.T) { func TestPoolWaitReturnsWorkerError(t *testing.T) { - worker := func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + worker := func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { return errors.New("some error") } @@ -241,13 +241,13 @@ func TestPoolWaitReturnsWorkerError(t *testing.T) { } func TestPoolWaitNotActivated(t *testing.T) { - p := New(1, func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { return nil }) + p := New(1, func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { return nil }) assert.EqualError(t, p.Wait(context.Background()), "workers poll not activated") } func TestPoolCanceled(t *testing.T) { - worker := func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + worker := func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { time.Sleep(100 * time.Millisecond) return sender(v) } @@ -261,7 +261,7 @@ func TestPoolCanceled(t *testing.T) { require.NoError(t, err) go func() { - for i := 0; i < 1000; i++ { + for range 1000 { p.Submit("line") time.Sleep(time.Millisecond * 100) } @@ -269,7 +269,7 @@ func TestPoolCanceled(t *testing.T) { }() n := 0 - var v interface{} + var v any for cursor.Next(ctx, &v) { n++ } @@ -279,7 +279,7 @@ func TestPoolCanceled(t *testing.T) { } func TestPoolError(t *testing.T) { - worker := func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + worker := func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { Metrics(ctx).Inc("calls") if rand.Intn(10) > 5 { //nolint gosec return errors.New("some error") @@ -294,7 +294,7 @@ func TestPoolError(t *testing.T) { require.NoError(t, err) go func() { - for i := 0; i < 1000; i++ { + for range 1000 { p.Submit("line") } p.Close() @@ -313,7 +313,7 @@ func TestPoolError(t *testing.T) { func TestPoolErrorContinue(t *testing.T) { - worker := func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + worker := func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { Metrics(ctx).Inc("calls") if rand.Intn(10) > 5 { //nolint gosec Metrics(ctx).Inc("errs") @@ -331,7 +331,7 @@ func TestPoolErrorContinue(t *testing.T) { require.NoError(t, err) go func() { - for i := 0; i < 1000; i++ { + for range 1000 { p.Submit("line") time.Sleep(time.Duration(rand.Intn(1000)) * time.Nanosecond) //nolint } @@ -351,21 +351,21 @@ func TestPoolErrorContinue(t *testing.T) { func TestPoolSubmitConcurrent(t *testing.T) { - worker := func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + worker := func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { return sender(v) } // chunkFn sends everything to the same worker, i.e. all producers share a single batch buffer - p := New(4, worker, Batch(10), ChunkFn(func(v interface{}) string { return "single" })) + p := New(4, worker, Batch(10), ChunkFn(func(v any) string { return "single" })) cursor, err := p.Go(context.Background()) require.NoError(t, err) var wg sync.WaitGroup - for i := 0; i < 8; i++ { + for i := range 8 { wg.Add(1) go func(producer int) { defer wg.Done() - for j := 0; j < 100; j++ { + for j := range 100 { p.Submit(fmt.Sprintf("%d-%03d", producer, j)) } }(i) @@ -394,7 +394,7 @@ func TestPoolSubmitAfterTermination(t *testing.T) { done := make(chan struct{}) go func() { defer close(done) - for i := 0; i < 100; i++ { + for i := range 100 { p.Submit(i) } }() @@ -411,7 +411,7 @@ func TestPoolSubmitAfterTermination(t *testing.T) { } t.Run("worker failed", func(t *testing.T) { - p := New(1, func(ctx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + p := New(1, func(ctx context.Context, v any, sender SenderFn, store WorkerStore) error { return errors.New("some error") }) cursor, err := p.Go(context.Background()) @@ -426,7 +426,7 @@ func TestPoolSubmitAfterTermination(t *testing.T) { defer cancel() started, once := make(chan struct{}), sync.Once{} - p := New(1, func(wCtx context.Context, v interface{}, sender SenderFn, store WorkerStore) error { + p := New(1, func(wCtx context.Context, v any, sender SenderFn, store WorkerStore) error { once.Do(func() { close(started) }) <-wCtx.Done() // hold the worker to make submits pile up return nil @@ -446,25 +446,25 @@ func TestWorkers_SubmitWithChunks(t *testing.T) { tbl := []struct { inp string poolSize int - buf [][]interface{} + buf [][]any }{ - {"test", 7, [][]interface{}{{"test"}, {}, {}, {}, {}, {}, {}}}, - {"test2", 7, [][]interface{}{{}, {}, {}, {"test2"}, {}, {}, {}}}, - {"test3", 7, [][]interface{}{{}, {}, {"test3"}, {}, {}, {}, {}}}, - {"test123", 7, [][]interface{}{{}, {}, {"test123"}, {}, {}, {}, {}}}, - {"test123", 1, [][]interface{}{{"test123"}}}, - {"test12345", 1, [][]interface{}{{"test12345"}}}, - {"zzzz", 2, [][]interface{}{{"zzzz"}, {}}}, - {"xxxx", 2, [][]interface{}{{}, {"xxxx"}}}, + {"test", 7, [][]any{{"test"}, {}, {}, {}, {}, {}, {}}}, + {"test2", 7, [][]any{{}, {}, {}, {"test2"}, {}, {}, {}}}, + {"test3", 7, [][]any{{}, {}, {"test3"}, {}, {}, {}, {}}}, + {"test123", 7, [][]any{{}, {}, {"test123"}, {}, {}, {}, {}}}, + {"test123", 1, [][]any{{"test123"}}}, + {"test12345", 1, [][]any{{"test12345"}}}, + {"zzzz", 2, [][]any{{"zzzz"}, {}}}, + {"xxxx", 2, [][]any{{}, {"xxxx"}}}, } - wk := func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error { + wk := func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error { return nil } for i, tt := range tbl { - p := New(tt.poolSize, wk, Batch(5), ChunkFn(func(val interface{}) string { + p := New(tt.poolSize, wk, Batch(5), ChunkFn(func(val any) string { return val.(string) + "$" })) @@ -478,18 +478,18 @@ func TestWorkers_SubmitWithChunks(t *testing.T) { func TestWorkers_SubmitNoChunkFn(t *testing.T) { - wk := func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error { + wk := func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error { return nil } p := New(8, wk, Batch(1000)) - for i := 0; i < 1000; i++ { + for i := range 1000 { p.Submit("something " + strconv.Itoa(i)) } tot := 0 - for j := 0; j < 8; j++ { + for j := range 8 { tot += len(p.buf[j]) assert.True(t, len(p.buf[j]) > 0 && len(p.buf[j]) < 1000) } @@ -500,7 +500,7 @@ func TestWorkers_SubmitNoChunkFn(t *testing.T) { // illustrates basic use of workers pool func ExampleWorkers_basic() { - workerFn := func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error { + workerFn := func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error { v, ok := inpRec.(string) if !ok { return errors.New("incorrect input type") @@ -534,7 +534,7 @@ func ExampleWorkers_basic() { // illustrates use of workers pool with all options func ExampleWorkers_withOptions() { - workerFn := func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error { + workerFn := func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error { v, ok := inpRec.(string) if !ok { return errors.New("incorrect input type") @@ -553,7 +553,7 @@ func ExampleWorkers_withOptions() { // create workers pool with chunks and batch mode. ChunkFn used to detect worker and guaranteed to send same chunk // to the same worker. This is important for stateful workers. Batch sets the size of internal buffer collecting records // internally before sending them to worker. - p := New(8, workerFn, Batch(10), ResChanSize(5), WorkerChanSize(2), ChunkFn(func(val interface{}) string { + p := New(8, workerFn, Batch(10), ResChanSize(5), WorkerChanSize(2), ChunkFn(func(val any) string { v := val.(string) return v[:4] // chunks by 4chars prefix })) diff --git a/pool/store.go b/pool/store.go index 7ea5e5d..3291f7d 100644 --- a/pool/store.go +++ b/pool/store.go @@ -5,21 +5,21 @@ import ( ) type localStore struct { - data map[string]interface{} + data map[string]any } // NewLocalStore makes map-based worker store func NewLocalStore() WorkerStore { - return &localStore{data: map[string]interface{}{}} + return &localStore{data: map[string]any{}} } // Set value for a given key -func (l *localStore) Set(key string, val interface{}) { +func (l *localStore) Set(key string, val any) { l.data[key] = val } // Get value for a given key -func (l *localStore) Get(key string) (interface{}, bool) { +func (l *localStore) Get(key string) (any, bool) { val, ok := l.data[key] return val, ok }