How to incorporate buffered channels

[maguro]

I need to incorporate buffered channels with the various Rill constructs. Unfortunately, the channels they make, e.g. Batch(...), return unbuffered channels.

It would be nice to have sibling methods like

func BatchToChannel[A any](out chan Try[[]A], in <-chan Try[A], n int, timeout time.Duration)

Any suggestions?

[destel]

Hello and thank you for your interest in Rill.

The current solution to the problem is to use rill.Buffer function. Like this:

batches := rill.Batch(items, 10, 1*time.Second)
batches = rill.Buffer(batches, 100) // adds buffer of size 100 to the pipeline

Semantically, it's equivalent to the Batch function returning a buffered channel, though it has a small performance overhead due to context switching. Such overhead would be negligible for I/O-bound workloads.

This design is a trade-off I've made to keep the API simpler, safer, and smaller. One aspect of this trade-off is that Rill always creates, manages, and closes output channels on its own, making it harder for users to make mistakes. Another part of this trade-off is that all intermediate channels are unbuffered.

One idea I considered is to add optional parameters to most functions. It could look like

batches := rill.Batch(items, 10, 1*time.Second, rill.WIthBuffer(100))

This design would also theoretically allow us to remove the "ordered" versions of the functions and enable/disable ordering with an optional parameter. However, I'm still not sure about this. I'm trying to be very careful with API design, and such a change would be backwards incompatible.

[maguro]

Placing a buffered channel at the end of the unbuffered channel is not the same as rill.Batch(...) returning a buffered channel, an infinite number of items will pile up like an aneurysm in the unbuffered channel feeding into the rill.Buffer().

I'm trying to keep my memory usage fixed and use buffered channels as a push-back mechanism to prevent memory issues associated with slow downstream filters, mappers, consumers.

With that said, I'll add that I would be hard pressed to come up with an example where I'd use an unbounded channel.

[destel]

Unbuffered channels are not unbound. They're opposite: they have zero sized buffers, meaning any write operation won't succeed until there's a corresponding read operation in another goroutine. Here's the quote from the Go specification:

The capacity, in number of elements, sets the size of the buffer in the channel. If the capacity is zero or absent, the channel is unbuffered and communication succeeds only when both a sender and receiver are ready. Otherwise, the channel is buffered and communication succeeds without blocking if the buffer is not full (sends) or not empty (receives). A nil channel is never ready for communication.

So, in terms of memory usage, you're safe. The goal of the rill.Buffer function is to add a buffer between pipeline stages. This allows upstream stages to proceed to some extent, even if a downstream stages become slow and can't keep up with the load.

[maguro]

Ugh, there must be a bug in my code. :doh:

[maguro]

Yeah, definitely a brain fart on my end. Your first solution will help quite nicely. Thanks!

[destel]

You're welcome