43 lines
802 B
Go
43 lines
802 B
Go
package cyutils
|
|
|
|
import (
|
|
"sync"
|
|
)
|
|
|
|
func WithWorkers[T any](workers int, arr []T, fn func(worker int, item T)) {
|
|
workersChan := make(chan struct{}, workers)
|
|
ingestChan := make(chan T, len(arr))
|
|
|
|
for _, item := range arr {
|
|
ingestChan <- item
|
|
}
|
|
close(ingestChan)
|
|
|
|
wg := sync.WaitGroup{}
|
|
for i := 0; i < workers; i++ {
|
|
workersChan <- struct{}{}
|
|
wg.Add(1)
|
|
go func(worker int) {
|
|
defer wg.Done()
|
|
defer func() { <-workersChan }()
|
|
for {
|
|
item, ok := <-ingestChan
|
|
// We're done processing totally
|
|
if !ok {
|
|
return
|
|
}
|
|
fn(worker, item)
|
|
}
|
|
}(i)
|
|
}
|
|
wg.Wait()
|
|
}
|
|
func Batched[T any](arr []T, batchSize int, fn func(batch []T)) {
|
|
for i := 0; i < len(arr); i += batchSize {
|
|
start := i
|
|
end := min(i+batchSize, len(arr))
|
|
batch := arr[start:end]
|
|
fn(batch)
|
|
}
|
|
}
|