Memory leak with InfiniteChannel

[tyranron]

If I understand correctly, the goroutine ch.infiniteBuffer() persists in memory until all the values from inner buffer are drained.

So, in situation where there are many writes, but only few reads, and then both writer/reader are done with a channel (channel goes out of their scopes), the channel won't be garbage collected as the goroutine ch.infiniteBuffer() still runs (and we have no runtime support for our lib as for native Go channels). Which in certain circumstances (and naive usage) may lead to memory leaks.

@eapache please, correct me if I'm wrong. If not, maybe this fact needs additional mention in docs?
There is already related one, but it points to another caveat.

[eapache]

You are not wrong, this is another caveat. In general this is true of all of the buffered channel implementations here (RingChannel, InfiniteChannel, etc):

If the channel passes out of scope without being explicitly closed, it will leak a goroutine plus whatever values were stored in the channel at the time it went out of scope.

How does that sound?

[tyranron]

@eapache sounds OK, but sometimes (as in my example above) just closing channel is not enough, we also need to read all the values from internal buffer.

[eapache]

Right, good point.

[michilu]

How do you think about providing another way such as supports the cancel event of context.Context for return from goroutines?

[tyranron]

@michilu I can barely imagine how that can be applied in this situation. Would you be so kind to provide a sort of sketch/pseudocode to illustrate your idea?

[michilu]

The inner goroutine returns if a case of closing all input channels.

This is other use cases, I try implements the Multiple-Input/Multiple-Output now.
I want do not closing edge channels of input/output each time because keep the channels owned by the worker func.
But It does not work with the Multiplex/Tee funcs.

channels:
edge inputs => Multiplex -> WeakTee => edge outputs

Maybe the WeakTee works well, so it has nothing to do to edge channels of output if closed the input channel.
I must be close all input channels if stop the Multiplex/WeakMultiplex goroutine, otherwise leak goroutines of the Multiplex func.

I think to it useful to supports context.Context for this countermeasure.

signature:
Multiplex(ctx context.Context, output SimpleInChannel, inputs ...SimpleOutChannel)

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
Multiplex(ctx, output, inputs)

[tyranron]

@michilu yup, now I see... polluting channels with context.Context will give an ability to cancel inner runners of such channels. This definitely should work.