Stream.group_by/2

[hamiltop]

Suggestion from elixir-lang-core mailing list.

Stream.group_by/2 would end up looking something like:

grouped_stream = Streamz.group_by(stream, fn (x) -> rem(x, 3) end) # Stream that represents %{1 => stream_one, 2 => stream_two, 3 => stream_three}
grouped_stream |> Enum.each fn ({key, stream}) ->
  spawn_link fn ->
    stream |> Enum.each fn (value) ->
      IO.puts "#{value} is #{key} mod 3"
    end
  end
end

One interesting use case (though there is probably a better way to do this) would be:

  computationally_intensive_stream
  |> Streamz.group_by(fn (x) -> rem(x, 3) end)
  |> Enum.map(fn {_, stream} -> stream end)
  |> Streamz.merge

Perhaps not intended, but that code could be used to distribute the stream and execute it in parallel. There is probably a better way to do it though. Perhaps Streamz.pmap/2 is a good idea.

Implementation would be something along the lines of one process per key in the group_by map. A router would either push new elements to the existing process or spawn a new process if a new key is created. One issue would be back pressure. A good test of whether back pressure is working properly is:

Stream.cycle([1,2,3])
  |> Streamz.group_by( fn (x) -> rem(x, 3) end)
  |> Enum.map &IO.inspect/1

And do nothing else with it. If back pressure is not working, then BEAM will start consuming GBs of memory, largely due to massive inboxes for the resulting processes.

The normal solution of requiring an ack before moving on to the next data point in the stream will not work if at least one of the resulting streams is not being consumed.

Potential solutions:

1. Do not consume from the source stream (Stream.cycle/1 in this case) until at least one downstream source is being consumed. For the streams not being consumed, either:
   a. queue the messages up.
   b. drop the messages.
2. Do not consume from the source stream until all downstream sources are being consumed.
3. Don't apply backpressure.

Problems:

1a. Only partial backpressure. If the source happens to only be 1% messages that make it to the downstream being consumed, we will build up a lot of queued messages.
1b. We lose messages. For some streams that is probably ok. For others, not so much.
2. We lose the ability to do things like filter on the result of group_by, which is a common pattern. We could solve this by requiring filter to happen before group_by, but that limits flexibility.
3. It is up to the user to not do stupid things.

I think 1a would be our best bet, though it would be pretty complicated. We could also allow for a flag to toggle the behavior between 1a and 1b.

The implementation I see is for the substream to send a :get message to the router process. The router then consumes data until it gets a data point that gets routed to the requesting process. This would result in an odd hybrid model of push/pull. A substream may have data in its inbox, but if it doesn't it would need to request data.

Thoughts and comments are welcome!

[josevalim]

I thought group_by would return a special kind of stream that would require per-key subscription without enumeration. You can even make the returned stream implement the access protocol, so you can do something like this:

stream = Stream.group_by(..., ...)
Enum.each stream["key"], fn value ->
  IO.puts "Got #{value} in key"
end

How does that affect the design decisions you asked above?

Code organization wise, instead of calling them generally CompoundStream, you could possible call them Streamz.Merge and Streamz.GroupBy, because they are really just representing the structure for the underlying computation. For merge you may not even need a struct, you can just return an anonymous function (that's how many functions in Stream does it).

One of the things I would also like to explore is that group_by and merge are potentially long running processes. So I would like to be able to start my "merger" or my "grouper" in a supervision tree and bind the incoming stream just later (although that is something we could explore later). But maybe this is overkill. :)

[hamiltop]

I like the Access protocol approach. The long running process idea is interesting. Currently I'm leveraging the fact that the process exiting clears out unwanted messages in the inbox. A long running process would require a bit more care on that front, but that's ok.

Sadly, those ideas are mostly just convenience. It doesn't really address the underlying issue - stateful streams.

An example of the problems we might face here:

data = [1,2,3,4,5,6,7,8,9,10]
grouped_data = Enum.group_by(data, &rem(&1,3))
grouped_stream = Stream.group_by(data, &rem(&1,3) )
grouped_stream[0] |> Enum.take(2) # returns [3,6]
grouped_data[0]   |> Enum.take(2) # returns [3,6]
grouped_stream[0] |> Enum.take(1) # Question 1 : should this return [3] or [9]
grouped_data[0]   |> Enum.take(1) # returns [3]
grouped_stream[1] |> Enum.take(2) # Question 2 : should this return [1,4] or [7, 10]
grouped_data[1]   |> Enum.take(2) # returns [1,4]

I think the answer to question 1 should be [9] and the answer to question 2 should be [1,4].

I think the streams produced by group_by will be stateful regardless of whether or not the original stream was stateful.

[hamiltop]

After mulling over this, here's the implementation I'm leaning towards.

GroupBy.start_link(stream) spawns a GroupBy process (probably GenServer based).

Enumeration is done by sending a message to the GroupBy process requesting data for key. The response can either be a value or :done.

When the GroupBy process receives a request for data it first looks to see if there is any buffered data. If there is, then the first element is removed and returned. If no data is available, data is pulled from the source stream until it matches the key. The matching data is sent back to the enumeration process. The "burned through" data is placed into the data buffers to be returned if that key is enumerated at a later point.

The use case this is meant to solve is where both A and B are keys being enumerated. When B is consuming the source stream, it may come across values that should be in A. Rather than lose them, it stores them away so A can fetch them.

This will solve the overall functionality, but at a fairly high cost. The buffered data may never be consumed. There are many options in dealing with this. We could set a cap. We could not save any buffered data. We could selectively save buffered data (only those keys in active enumeration, or only the keys specified as options to start_link). The above solution could be adapted to any of those scenarios fairly easily.

@josevalim What are your thoughts on this? Are we ok with GroupBy being such a different beast? Most of Stream mirrors Enum, but here we depart quite a bit in usage.

[josevalim]

I was talking with James and he suggested that there is no need to use It
directly in a supervision tree. We can use Stream.group_by directly as long
as it is a linked process (and that process is linked to something in the
tree).

The observable docs in Rx mention two kinds of observables: hot and cold.
Hot observables are basically emitting data all the time like a gen event
and so on. Cold ones are basically data structures (lists, maps, etc).

I don't think we should do buffering of hot observables. On the other hand,
there is the discussion if cold observables should be repeatable, as you
posted in an earlier issue.

It would be nice to see how RxJava treats those different
observables/streams but one option is to have a protocol that will tell you
if a stream is hot or cold so you can act accordingly.

For example, when group_by starts, you invoke this new protocol that will
let you know if it is hot or cold. If it is hot, you will start receiving
messages and you will redirect them to the published keys (or ignore if
there are no keys).

If cold, you treat as a normal data structure and guarantee repeatability
as you would get with Enum. This implies though the data structure would be
traversed multiple times (one per key to be more precise).

Another advantage of having a protocol is that you could specialize hot
observables on merge too. Today, if you pass a gen event on merge, you have
the caller process, the collector process, the consumer process per stream
and gen event process.

If you know what is hot and cold, you can get hot streams to send messages
directly to the collector, skipping the consumer one.

Another approach would be to have different functions with different names.
group_by starts one stream per key, the other functions starts the stream
once and publish it per key. This may be the simplest approach.

This is way too much fun. :)

On Monday, July 21, 2014, hamiltop notifications@github.com wrote:

After mulling over this, here's the implementation I'm leaning towards.

GroupBy.start_link(stream) spawns a GroupBy process (probably GenServer
based).

Enumeration is done by sending a message to the GroupBy process requesting
data for key. The response can either be a value or :done.

When the GroupBy process receives a request for data it first looks to see
if there is any buffered data. If there is, then the first element is
removed and returned. If no data is available, data is pulled from the
source stream until it matches the key. The matching data is sent back to
the enumeration process. The "burned through" data is placed into the data
buffers to be returned if that key is enumerated at a later point.

The use case this is meant to solve is where both A and B are keys being
enumerated. When B is consuming the source stream, it may come across
values that should be in A. Rather than lose them, it stores them away so A
can fetch them.

This will solve the overall functionality, but at a fairly high cost. The
buffered data may never be consumed. There are many options in dealing with
this. We could set a cap. We could not save any buffered data. We could
selective save buffered data (only those keys in active enumeration, or
only the keys specified as options to start_link). The above solution could
be adapted to any of those scenarios fairly easily.

@josevalim https://github.com/josevalim What are your thoughts on this?
Are we ok with GroupBy being such a different beast? Most of Stream mirrors
Enum, but here we depart quite a bit in usage.

—
Reply to this email directly or view it on GitHub
#3 (comment).

José Valimwww.plataformatec.com.br
http://www.plataformatec.com.br/Founder and Lead Developer

[hamiltop]

So it sounds like for cold streams that group by is nothing more than

steam |> Enum.filter &( group_fun.(&1) == key )

And hot streams are as I described, but with selective buffering. We would
still want to buffer subscribed keys.

I guess if you don't want to lose any data in a hot stream, make sure you
have comprehension subscription. I'm thinking about the case of a network
stream. Is the small window between calling group_by and setting
subscriptions a problem? We have a similar issue right now with GenEvent.

Another possibility is to provide default subscription. If no other
subscriptions match, then it goes to the default subscriber. That also
provides the interesting ability of "skimming" the stream.

I'm going to go ahead and build out the hot stream case with a configurable
default subscription. I think the idea of a Stream protocol with merge and
hot/cold needs to bake a bit more. My implementation should be able to fit
into that design anyway.
On Jul 20, 2014 11:59 PM, "José Valim" notifications@github.com wrote:

I was talking with James and he suggested that there is no need to use It
directly in a supervision tree. We can use Stream.group_by directly as
long
as it is a linked process (and that process is linked to something in the
tree).

The observable docs in Rx mention two kinds of observables: hot and cold.
Hot observables are basically emitting data all the time like a gen event
and so on. Cold ones are basically data structures (lists, maps, etc).

I don't think we should do buffering of hot observables. On the other
hand,
there is the discussion if cold observables should be repeatable, as you
posted in an earlier issue.

It would be nice to see how RxJava treats those different
observables/streams but one option is to have a protocol that will tell
you
if a stream is hot or cold so you can act accordingly.

For example, when group_by starts, you invoke this new protocol that will
let you know if it is hot or cold. If it is hot, you will start receiving
messages and you will redirect them to the published keys (or ignore if
there are no keys).

If cold, you treat as a normal data structure and guarantee repeatability
as you would get with Enum. This implies though the data structure would
be
traversed multiple times (one per key to be more precise).

Another advantage of having a protocol is that you could specialize hot
observables on merge too. Today, if you pass a gen event on merge, you
have
the caller process, the collector process, the consumer process per stream
and gen event process.

If you know what is hot and cold, you can get hot streams to send messages
directly to the collector, skipping the consumer one.

Another approach would be to have different functions with different
names.
group_by starts one stream per key, the other functions starts the stream
once and publish it per key. This may be the simplest approach.

This is way too much fun. :)

On Monday, July 21, 2014, hamiltop notifications@github.com wrote:

After mulling over this, here's the implementation I'm leaning towards.

GroupBy.start_link(stream) spawns a GroupBy process (probably GenServer
based).

Enumeration is done by sending a message to the GroupBy process
requesting
data for key. The response can either be a value or :done.

When the GroupBy process receives a request for data it first looks to
see
if there is any buffered data. If there is, then the first element is
removed and returned. If no data is available, data is pulled from the
source stream until it matches the key. The matching data is sent back
to
the enumeration process. The "burned through" data is placed into the
data
buffers to be returned if that key is enumerated at a later point.

The use case this is meant to solve is where both A and B are keys being
enumerated. When B is consuming the source stream, it may come across
values that should be in A. Rather than lose them, it stores them away
so A
can fetch them.

This will solve the overall functionality, but at a fairly high cost.
The
buffered data may never be consumed. There are many options in dealing
with
this. We could set a cap. We could not save any buffered data. We could
selective save buffered data (only those keys in active enumeration, or
only the keys specified as options to start_link). The above solution
could
be adapted to any of those scenarios fairly easily.

@josevalim https://github.com/josevalim What are your thoughts on
this?
Are we ok with GroupBy being such a different beast? Most of Stream
mirrors
Enum, but here we depart quite a bit in usage.

—
Reply to this email directly or view it on GitHub
#3 (comment).

José Valimwww.plataformatec.com.br
http://www.plataformatec.com.br/Founder and Lead Developer

—
Reply to this email directly or view it on GitHub
#3 (comment).

[josevalim]

So it sounds like for cold streams that group by is nothing more than

steam |> Enum.filter &( group_fun.(&1) == key )

Good point.

And hot streams are as I described, but with selective buffering. We would still want to buffer subscribed keys.

In my opinion there is no buffering, if something is subscribed to the key, it will receive it, if not, it won't.

I guess if you don't want to lose any data in a hot stream, make sure you
have comprehension subscription. I'm thinking about the case of a network
stream

It depends on how you design the stream. For a network, you could start the acceptor loop just once someone starts consuming the stream, this way there is no loss of messages. That's what we do for files, but we can't do it for all streams.

[hamiltop]

Either we are ok with huge potential losses or we have to buffer somewhere.

Consider a range and grouping by even/odd and each has a process
enumerating. i.e. something is subscribed to :even and something is
subscribed to :odd. All elements should be processed and 50% of elements
fall into each camp.

What happens if :odd consumes 10 elements during the time it takes :even to
consume 2. What do we do with the other 8 :even elements? Either we buffer
them (explicitly or via a process's inbox) or we drop them.

On Mon, Jul 21, 2014 at 2:48 PM, José Valim notifications@github.com
wrote:

So it sounds like for cold streams that group by is nothing more than

steam |> Enum.filter &( group_fun.(&1) == key )

Good point.

And hot streams are as I described, but with selective buffering. We would
still want to buffer subscribed keys.

In my opinion there is no buffering, if something is subscribed to the
key, it will receive it, if not, it won't.

I guess if you don't want to lose any data in a hot stream, make sure you
have comprehension subscription. I'm thinking about the case of a network
stream

It depends on how you design the stream. For a network, you could start
the acceptor loop just once someone starts consuming the stream, this way
there is no loss of messages. That's what we do for files, but we can't do
it for all streams.

—
Reply to this email directly or view it on GitHub
#3 (comment).

[josevalim]

When you say range, do you mean the range literal x..y? If so, it is a cold observable and I thought the conclusion was to not use group_by? There is probably no problem in buffering cold streams but I think it could be very dangerous for hot ones.

I think this discussion is really pointing out that we cannot have one operation that will really handle both types nicely. So we need pick one side and go with it. :)

[hamiltop]

I mean a hot stream producing data that looks like a range in that it is
distributed between even and odd.

My intention with this library is to build things that may or may not be
useful. I think I need to go ahead and build it. My suspicion us that
group_by as we have discussed will be too trivial for cold streams and too
lossy for hot streams.

But that shouldn't stop me from building it and seeing how it behaves. I'll
give a lossy version a go and start playing around with it.
On Jul 21, 2014 11:21 PM, "José Valim" notifications@github.com wrote:

When you say range, do you mean the range literal x..y? If so, it is a
cold observable and I thought the conclusion was to not use group_by? There
is probably no problem in buffering cold streams but I think it could be
very dangerous for hot ones.

I think this discussion is really pointing out that we cannot have one
operation that will really handle both types nicely. So we need pick one
side and go with it. :)

—
Reply to this email directly or view it on GitHub
#3 (comment).

[hamiltop]

Just for reference: http://netflix.github.io/RxJava/javadoc/rx/observables/GroupedObservable.html

Note: A GroupedObservable will cache the items it is to emit until such time as it is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those GroupedObservables that do not concern you. Instead, you can signal to them that they may discard their buffers by applying an operator like Observable.take(int)(0) to them.

[josevalim]

Awesome. :D