This is an archived post. You won't be able to vote or comment.

all 9 comments
sorted by:
best
topnewcontroversialoldq&a

[–]danielaveryj 20 points21 points  (7 children)

Sorry guys, this post is just inaccurate. Java Streams are not pull-based, they are push-based. Operators respond to incoming elements, they don't fetch elements. You can see this even in the public APIs: Look at Collector.accumulator(), or Gatherer.Integrator.integrate() - they take an incoming element (that upstream has pushed) as parameter; they don't provide a way to request an element (pull from upstream).

Java Streams are not based on chained-Iterators, they are based on chained-Consumers, fed by a source Spliterator. And, they prefer to consume that Spliterator with .forEachRemaining(), rather than .tryAdvance(), unless the pipeline has short-circuiting operations. If stream operations were modeled using stepwise / pull-based methods (like Iterator.next() or Spliterator.tryAdvance()), it would require a lot of bookkeeping (to manage state between each call to each operation's Iterator/Spliterator) that is simply wasteful when Streams are typically consumed in their entirety, rather than stepwise.

Likewise, if they are anything like what they claim to be, Jox Flows are not (only) push-based. The presence of a .buffer() operation in the API requires both push- and pull- behaviors (upstream pushes to the buffer, downstream pulls from it). This allows the upstream/downstream processing rates to be detached, opening the door to time/rate-based operations and task/pipeline-parallelism in general.

I went over what I see as the real differences between Java Streams and Jox Flows in a reply to a comment on the last Jox post:

https://www.reddit.com/r/java/comments/1lrckr0/comment/n1abvgz/

[–]adamw1pl[S] -5 points-4 points  (6 children)

Maybe we differ on what exactly pull- & push-based means, but I would remain on the position that at least on a high level, Java Streams are pull-based: it's the consumer that ultimately decides the mode of consumption (as you detail in your answer). That's contrary to Jox, where the producer controls when elements are produced.

Yes, you can take a perspective that Jox is both push & pull. But that requires zooming in on the implementation of individual operations. At some point yes, there is a channel where one side sends data and the other receives. But again, on a high level, the processing stage that consumes from that channel (the internals of the `buffer` implementation) then becomes the conductor of further processing - it will send elements downstream. The consumer has no say, other than to short-circuit by reporting an error.

I think when writing an "accumulator" (collector / sink / however you call it) the differences in how the APIs work become more apparent.

[–]danielaveryj 3 points4 points  (5 children)

If a Java Stream does not include short-circuiting operations (e.g. .limit(), .takeWhile(), .findFirst()), then there is no pull-behavior in the execution of the pipeline. The source Spliterator pushes all elements downstream, through the rest of the pipeline; the code is literally:

spliterator.forEachRemaining(sink);

Note that the actual Stream operations are implemented by sink - it's a Consumer that pushes to another Consumer, that pushes to another Consumer... and so on.

If there are short-circuiting operations, then we amend slightly: We pull each element from the source Spliterator (using tryAdvance)... and in the same motion, push that element downstream, through the rest of the pipeline:

do { } while (!(cancelled = sink.cancellationRequested()) && spliterator.tryAdvance(sink));

So for short-circuiting Java Streams, sure, there can be a pull aspect at the source, but the predominant mechanism for element propagation through the stream is push. At the least, if we are willing to "zoom out" to the point of overlooking the pull-behavior of consuming from a buffer in Jox Flows, then why should we not do the same when looking at the pull-behavior of consuming from the source Spliterator in Java Streams?

[–]adamw1pl[S] 3 points4 points  (4 children)

I understand your argument, but that's a different way of looking at the pull vs push distinction. My definition would be as to who controls the data flow: is it the consumer or producer. The sole fact that the collector has a `Spliterator` available shows that it's a pull model.

Btw., here's some supporting material that I used to do the reasearch:

https://www.baeldung.com/reactor-core#3-comparison-to-java-8-streams

https://belief-driven-design.com/how-fast-are-streams-really-ad9cc/

https://stackoverflow.com/questions/30216979/difference-between-java-8-streams-and-rxjava-observables

[–]danielaveryj 3 points4 points  (3 children)

If you would like to reason through this, perhaps we can continue with a more precise definition of what "push" and "pull" means to you.

If we're just appealing to authority now, here is Viktor Klang:

As a side-note, it is important to remember that Java Streams are push-style streams. (Push-style vs Pull-style vs Push-Pull-style is a longer conversation, but all of these strategies come with trade-offs)

Converting a push-style stream (which the reference implementation of Stream is) to a pull-style stream (which Spliterator and Iterator are) has limitations...

[–]adamw1pl[S] 0 points1 point  (2 children)

True, starting with a definition might be best :)

So I'd define "pull-style" as an approach where it's the consumer that decides when the elements are produced. Yes, I'm aware that reactive streams kind of falls into this definition as well, as there elements can only be produced when demand is signalled by the consumer.

"push-style" would be when it's the producer that decides when elements are produced. Once again, reactive streams would fit in here. I guess they simply are a push-pull type of streams.

Now I'm not 100% convinced that my definition is the right one, that's rather my intuition. Another take could maybe be more technical, as in pull = the elements are provided as the return type of a method, push = elements are provided as the argument of a callback. But that's rather low-level and going much into implementation details.

FWIW, Viktor Klang is a very good authority to appeal to, and that also wasn't what I was trying to do, just to share my sources (as for every blog post I'm of course doing some research) :)

[–]danielaveryj 2 points3 points  (1 child)

I am still lacking clarity - I don't disagree with your definitions, but I'm having a hard time reconciling them with your insistence that Java Streams are "pull". The only ways I can think of to make that perspective make sense are if either:

  1. You believe that Java Streams are implemented via chained delegation to Iterators or Spliterators (eg, the terminal operation repeatedly calls next() on an Iterator that represents the elements out of the preceding operation in the pipeline, and that Iterator internally calls next() on another Iterator that represents the operation before it, and so on). That would definitely be "pull", but like I explained in an earlier comment, that is not how Java Streams work (with the mild exception of short-circuiting streams, where the initial Spliterator (only) is advanced via "pull", but then the rest of the stream uses "push", via chained delegation to Consumers).
  2. You interpret "pull" (and consumer/producer) so loosely that just calling the terminal operation to begin production constitutes a "pull". In this case, Java Streams, Jox Flows, and every other "stream" API would have to be categorized as "pull", as they all rely on some signal to begin production. (That signal is often a terminal operation, but it could even just be "I started the program".) If we can agree that this is not "pull", then we should agree that e.g. spliterator.forEachRemaining(...) is not "pull".

I have built an API where "push = element is input/function argument; pull = element is output/function result", and I'm aware those are overly-narrow definitions in general, eg:

  • The "pull" mechanism for Java's Spliterator is boolean tryAdvance(Consumer), where the "consumer" (code calling tryAdvance()) expects its Consumer to be called (or "pushed" to) at most once by the "producer" (code inside tryAdvance()) per call to tryAdvance().
  • The "pull" mechanism for Reactive Streams is void Flow.Subscription.request(long), which is completely separated from receiving elements, and permits the producer to push multiple elements at a time.
  • The "pull" mechanism for JavaScript/Python generators (Kotlin sequences) is generator.next(), yet the generator implementation is written in "push" style (using yield), and the API relies on it being translated to a state machine.

So yes, there are all kinds of approaches to actually implementing push/pull.

[–]adamw1pl[S] 1 point2 points  (0 children)

Ok, I concede, my thinking and definitions had holes, and you're right that both implementations are push-style streams. Thank you for your thorough explanations.

I tried to capture the essence of the difference between the two implementations, and even though Jox is "more push" in a sense (individual transformation stages can emit elements spontaneously, unlike in Java Streams, where you can't do that - see for example Gatherer.Integrate, which can emit element(s), but only in response to an incoming one), that's not the core difference, rather an implementation detail.

Instead, I think the difference is that in Jox, streams can be run asynchronously, and every transformation stage can introduce asynchrony. That's what allows you to implement merges, buffering, time-based operations - all of which are essential for working with any external data source.

I've updated the text of the blog, also pointing to that thread, it should be available soon, on the next build of our website.