5 ms·
func process(stream chan int) { p := pool.New().WithMaxGoroutines(10) for elem := range stream { elem := elem p.Go(func(
by ilyt 4y ago
func process(stream chan int) {
p := pool.New().WithMaxGoroutines(10)
for elem := range stream {
elem := elem
p.Go(func() {
handle(elem)
})
}
p.Wait()
I did something similar just with input (optionally output) channel. Close input, goroutines stop, when all of them stop the output is closed [1]. No need to incur function call just to add elements (although I'd imagine go would just inline it so it might not matter either way)
This
func mapStream(
in chan int,
out chan int,
f func(int) int,
) {
s := stream.New().WithMaxGoroutines(10)
for elem := range in {
elem := elem
s.Go(func() stream.Callback {
res := f(elem)
return func() { out <- res }
})
}
s.Wait()
}
also seems awfully verbose vs just function (that is now easy and safe thanks to generics) with this signature
WorkerPool[T1, T2 any](input chan T1, output chan T2, worker func(T1) T2, concurrency int)
I do like idea of waitgroup on steroids, I might steal it for my generic library.
* [1] https://github.com/XANi/goneric/blob/master/worker.go#L92 https://github.com/XANi/goneric/blob/master/worker.go#L92
- camdencheek 4y agoFair criticism. The nice thing about the current API is it works with any input that's iterable (channels, slices, readers, etc.) and any output (callback, channel, append to slice, etc.). In most code I write, I avoid channels because I find them easy to misuse. I used channels in the examples because it's the easiest way to represent "some potentially unbounded stream of input."
- ilyt 4y agoI just wrote a bunch of separate short functions for different types (channels, slices, maps). There is more code duplication but code itself is simpler, and less boiler-platey and more embeddable. For example out := MapSlice( func(i int) string { return fmt.Sprintf("-=0x%02x=-", i) }, MapSlice( func(i int) int { return i + 1 }, MapSlice( func(i int) int { return i * i }, GenSlice(10, func(idx int) int { return idx }), ), ), ) or piping workers out := WorkerPoolBackgroundClose( WorkerPoolBackgroundClose( WorkerPoolBackgroundClose( GenChanNClose(3, func(idx int) int { return idx + 1 }), func(i int) string { return strconv.Itoa(i) }, 4), func(s string) string { return ">" + s }, 5, ), func(s string) string { return " |" + s }, 6)