Thanks for the response. And thanks for the explanation on push/pull, I understand it a little better now.
As for the termination logic, that also makes good sense now. And I agree, it would be important to communicate termination. I do hope that that idea of a ClosableQueue eventually sees the light of GA. It really would make a lot of fundamental problems that I keep running into disappear. Thanks for the discussion on this, it was enlightening. On Mon, Aug 31, 2026 at 4:36 AM Viktor Klang <[email protected]> wrote: > >You said that streams are push-style. Is that not what I just described? > >[...] > >I always thought the difference of push vs pull was who was deciding when > work is to be done. Push being that the producer says do it now, only > stopping when there is not a worker available. Whereas pull is the workers > choosing when they become available to do work, stopping or delaying for > any reason they want. > > No, unless I misunderstood something. When you have a queue in front, the > downstream pulls in work by consuming from the queue—if there are no > elements in the queue, the workers will not have anything to do. You > _could_ argue that it is a "hybrid/push-pull" since as long as there are > workers waiting for an element, any new elements added to the queue will be > processed as a result of pushing an element to the queue. However, since > there is a bi-directional dependency, you need to have a strategy for > managing the relative progress between producers and consumers—hence the > enqueuing strategies like the "block, drop-oldest, drop-newest, drop-all". > > With that said, I tend to view solutions like the one I've described above > as at least a "primarily-pull" solution, as downstream is polling for > elements produced by the upstream. > > >For worker termination, can you give me an example of when a worker would > want to terminate? For me, I see workers purely as threads -- probably from > the FJP or WSP. I don't really see a situation where them "terminating" > makes sense, but that's likely lack of creativity from me. > > A worker may want to terminate because its operation has short-circuited. > It may want to terminate due to a non-recoverable error/exception. It may > want to terminate due to a resizing operation of its pool. (...and so on) > On 2026-08-28 16:32, David Alayachew wrote: > > Then maybe I am confused -- where does the difference lie between push and > pull? You said that streams are push-style. Is that not what I just > described? If there are no more threads in the ForkJoinPool, then the only > recourse besides waiting > ZjQcmQRYFpfptBannerStart > Then maybe I am confused -- where does the difference lie between push and > pull? > > You said that streams are push-style. Is that not what I just described? > If there are no more threads in the ForkJoinPool, then the only recourse > besides waiting is for the producer thread itself to jump in and start > doing the work too. Other than that, I am not seeing the difference. And if > that is the only difference, I am fine with doing that instead. > > I always thought the difference of push vs pull was who was deciding when > work is to be done. Push being that the producer says do it now, only > stopping when there is not a worker available. Whereas pull is the workers > choosing when they become available to do work, stopping or delaying for > any reason they want. > > Please do respond to the above, as I think my lack of understanding is > definitely part of the problem here, so I want to understand. That said, > your ClosableBlockingQueue idea has become more interesting to me. > > For me, dropping doesn't make sense, at least as a DIRECT option. But, > what might make more sense would be an overloaded constructor/factory that > provides a Consumer<T> overflowConsumer, so the user can decide what they > want to do upon filling up, whereas the original constructor/factory is for > blocking. > > For worker termination, can you give me an example of when a worker would > want to terminate? For me, I see workers purely as threads -- probably from > the FJP or WSP. I don't really see a situation where them "terminating" > makes sense, but that's likely lack of creativity from me. > > At best, a worker should be able to communicate that it completed the task > at hand, but that's a given by being a thread. > > As for FIFO, I think that first depends on what we actually want to make > here. > > If this is going to be another branch of the collections framework, like > TransferQueue, then this would make sense to do as a separate > implementation under the same interface. Call that parent interface a > ClosableQueue, where the producer can close the queue once the queue has > been emptied, and simultaneously, all workers have finished their tasks. > From there, the idea of priority could be introduced, to help decide order, > similar to how the PriorityQueue does it. > > But if that is too much, and this should be a single implementation (maybe > even without a new parent interface, just existing parent interfaces), then > providing an overloaded constructor/factory that adds order would do the > job. The non-comparator overloads must communicate that they don't do > priority, but tbh, multiple implementations make more sense at that point > imo. Less error-prone for the developer, lest they assume that no > comparator means natural order. > > On Fri, Aug 28, 2026 at 5:32 AM Viktor Klang <[email protected]> > wrote: > >> >I want to announce that there is more work, then immediately hand off >> the task to a worker until I run out of workers, which I believe is known >> as backpressure. In which case, requests will queue up until a worker is >> free. Am I understanding this correctly? >> >> This is by its very nature an async hand-off between a producer and >> consumers. With a push-style stream the availability of a new element >> causes the immediate processing of said element. It isn't the availability >> of workers pulling in available work. >> >> On the topic of "closable" BlockingQueues, it is something which I've >> thought about and experimented with a lot, and there's a ton of nuance >> there: what available strategies for dealing with a full queue should >> exist, block, drop-oldest, drop-newest, drop-all, etc. Also, should workers >> be able to signal termination such that the producer doesn't end up >> producing to a queue where all workers have terminated? Is it always >> first-come-first-served in terms of consumption or does it support >> "broadcast"? >> On 2026-08-27 22:42, David Alayachew wrote: >> >> Hmmm, this looks more like async-style, where it's pub-sub. That's not >> really what I am looking for. >> >> What I am asking for is inherently push-based, assuming that I understand >> the terminology (basing myself off of this -- >> https://stackoverflow.com/questions/51254117/what-is-difference-between-push-based-and-pull-based-structures-like-ienumerable >> <https://urldefense.com/v3/__https://stackoverflow.com/questions/51254117/what-is-difference-between-push-based-and-pull-based-structures-like-ienumerable__;!!ACWV5N9M2RV99hQ!MRNMRsm0f3rRv6WFvB4G2dQcwa_aPT_GGxQ0l0IMKow3rwkzSfQpE7rfQ3bkH2FKjOM2GoSS7daDbQD07Wdq3LBPomQ$>). >> I want to announce that there is more work, then immediately hand off the >> task to a worker until I run out of workers, which I believe is known as >> backpressure. In which case, requests will queue up until a worker is free. >> Am I understanding this correctly? >> >> And let me ask a way more fundamental question -- why can't I use the >> Stream framework for an upstream that changes its contents? For me, if I >> could do that -- toss items onto my upstream collection as I am iterating, >> and then signal that I am done, then this entire problem goes away. I guess >> let's start with that, as I feel like me not understanding why that is in >> place prevents me from understanding. >> >> And I am not at all implying that every collection should be able to do >> that. Certainly not. But if there was even one collection implementation >> that had a stream method, and that method allowed me to communicate that >> there are no more elements, and specifically, that there won't be anymore >> elements, then this entire problem goes away. >> >> >> On Thu, Aug 27, 2026 at 4:58 AM Viktor Klang <[email protected]> >> wrote: >> >>> Thanks for your kind words, David! >>> >>> Perhaps I focused too much on the "recursive"-side of your question? >>> >>> What you describe seems more amendable to a pull-style stream with an >>> async hand-off (allowing the upstream to progress independently, up to a >>> limit) in order to allow to hide the latency of generating/retriving the >>> upstream values. This specific scenario was exactly the motivating reason >>> behind *java.util.concurrent.Flow*. There are inherent trade-offs >>> between *push*-style streams (*java.util.stream.Stream*) and *pull*-style >>> streams (*java.util.concurrent.Flow*) and sometimes either of those >>> approaches are preferable given the desired runtime characteristics. >>> >>> What you could experiment with is the style I shared previously and pair >>> it with Gatherers.mapConcurrent() for the subsequent processing of each >>> element (Path), or you could try using mapConcurrent as inspiration for how >>> to implement flatMapConcurrent to concurrently (up to a limit) traverse the >>> hierarchy? >>> >>> Let me know if that is more aligned with what you're thinking here. >>> On 2026-08-27 05:00, David Alayachew wrote: >>> >>> Lol, that is so creative. I never would have thought to use anonymous >>> classes on the java. util. function interfaces lol. That's like choosing >>> the horse when you have a motorcycle right in front of you. But lo and >>> behold, it genuinely is the >>> Lol, that is so creative. >>> >>> I never would have thought to use anonymous classes on the >>> java.util.function interfaces lol. That's like choosing the horse when you >>> have a motorcycle right in front of you. But lo and behold, it genuinely is >>> the tersest way forward using normal Java. >>> >>> How did you come up with this? lol >>> >>> Anyways, creativeness aside, this one had about the same performance as >>> the Gatherer that I came up with. I am wondering if there might be too much >>> hand off, or something performance wise that I might be doing wrong here. >>> Long story short is that, I want to gather the files as fast as possible, >>> but I don't see a way to get the full extent of the parallelism out without >>> keeping it all on one pipeline. At the end of the day, the solution that >>> you made seems to be making all of these instances of Stream over and over >>> again, and I worry if that might be making things slower than needed. The >>> instance of Gatherer I made doesn't have this problem, but that's because >>> it has the opposite problem of not being able to parallelize easily. Yours >>> looks like parallelizes easily enough, but also does a lot of work to make >>> that happen. >>> >>> Maybe this is imaginative of me, but I am thinking of some sort of >>> Collection type, which doesn't suffer from ConcurrentModificationException, >>> where the Stream can just keep pulling from it until it gets told that no >>> more elements are coming through, and it can pull and process the elements >>> from that Collection type in parallel. This is my dream object, if you >>> will. You just keep appending to the end of it as you come across >>> directories, and you keep pulling from the start of it until the collection >>> object finally becomes empty, in which case, you are finally done. >>> >>> Does any of that make sense to you? >>> >>> >>> On Wed, Aug 26, 2026, 11:26 AM Viktor Klang <[email protected]> >>> wrote: >>> >>>> Hello David, >>>> >>>> You mean something like this? >>>> >>>> void main() throws Exception { >>>> final Path root = Path.of(System.getProperty("user.home")); >>>> >>>> IO.println(root); >>>> IO.println(root.toAbsolutePath()); >>>> >>>> Stream >>>> .of(root) >>>> .flatMap(new Function<Path, Stream<Path>>() { >>>> @Override public Stream<Path> apply(Path p) { >>>> if (Files.isRegularFile(p)) { >>>> return Stream.of(p); >>>> } else { >>>> try { >>>> return Files.list(p).flatMap(this::apply); >>>> } catch (IOException _) { >>>> return Stream.empty(); >>>> } >>>> } >>>> } >>>> }) >>>> .limit(5) >>>> .forEach(IO::println); >>>> } >>>> >>>> On 2026-08-26 16:24, David Alayachew wrote: >>>> > Hello Viktor and Rémi, >>>> > >>>> > First off, sorry for the horrifically delayed response. Juggling >>>> disasters >>>> > and emergencies. >>>> > >>>> > Rémi, thanks for the example with mapMulti. That has been my temporary >>>> > workaround for now, and while it is still not ideal, it is better than >>>> > where I was before. >>>> > >>>> > Viktor, sure, here is a simple example -- traversing a directory >>>> tree, and >>>> > only passing the files down the stream. >>>> > >>>> > I actually made a post on StackOverflow -- >>>> > https://softwareengineering.stackexchange.com/questions/461442/ >>>> <https://urldefense.com/v3/__https://softwareengineering.stackexchange.com/questions/461442/__;!!ACWV5N9M2RV99hQ!JsL3fQJ1ult_TN0D6BCiRDGewOWQs2G_97ksDE0txV_GsNtGvjC1XLvjm8-rVEgGNQKIOX_5Xz7VRezKpGpwi9bbnPg$> >>>> > >>>> > But anyways, when attempting to do this with streams, this is where I >>>> > started. >>>> > >>>> > >>>> > import module java.base; >>>> > >>>> > void main() throws Exception >>>> > { >>>> > >>>> > final Path root = Path.of(System.getProperty("user.home")); >>>> > >>>> > IO.println(root); >>>> > IO.println(root.toAbsolutePath()); >>>> > >>>> > Stream >>>> > .of(root) >>>> > .mapMulti(this::recursiveDescent) >>>> > .limit(5) >>>> > .forEach(IO::println) >>>> > ; >>>> > >>>> > } >>>> > >>>> > private void recursiveDescent(final Path rootPath, final >>>> Consumer<Path> >>>> > downstream) >>>> > { >>>> > >>>> > final Stack<Path> stack = new Stack<>(); >>>> > stack.push(rootPath); >>>> > >>>> > while (!stack.empty()) >>>> > { >>>> > >>>> > final Path path = stack.pop(); >>>> > >>>> > if (Files.isRegularFile(path)) >>>> > { >>>> > >>>> > downstream.accept(path); >>>> > >>>> > } >>>> > >>>> > else >>>> > { >>>> > >>>> > try (final Stream<Path> folderContents = Files.list(path)) >>>> > { >>>> > >>>> > folderContents.forEach(stack::push); >>>> > >>>> > } >>>> > >>>> > catch (final Exception exception) >>>> > { >>>> > >>>> > throw new IllegalStateException("Failed for " + path, >>>> > exception); >>>> > >>>> > } >>>> > >>>> > } >>>> > >>>> > } >>>> > >>>> > } >>>> > >>>> > But this has at least 2 major downsides. >>>> > >>>> > 1 - This is not easy to turn parallel (comparatively). >>>> > >>>> > 2 - This does not short-circuit when the downstream no longer accepts >>>> > elements. >>>> > >>>> > Ok, I can at least solve problem 2 by becoming a Gatherer instead. >>>> > >>>> > Here is my gatherer attempt. >>>> > >>>> > >>>> > import module java.base; >>>> > >>>> > void main() throws Exception >>>> > { >>>> > >>>> > final Path root = Path.of(System.getProperty("user.home")); >>>> > >>>> > IO.println(root); >>>> > IO.println(root.toAbsolutePath()); >>>> > >>>> > final Gatherer<Path, Stack<Path>, Path> gatherer = >>>> > Gatherer >>>> > .of >>>> > ( >>>> > Stack<Path>::new, >>>> > (stack, rootPath, downstream) -> >>>> > { >>>> > >>>> > stack.push(rootPath); >>>> > >>>> > while (!stack.isEmpty()) >>>> > { >>>> > >>>> > final Path path = stack.pop(); >>>> > >>>> > if (Files.isRegularFile(path)) >>>> > { >>>> > >>>> > final boolean acceptingMoreElements = >>>> > downstream.push(path); >>>> > >>>> > if (!acceptingMoreElements) >>>> > { >>>> > >>>> > return false; >>>> > >>>> > } >>>> > >>>> > } >>>> > >>>> > else >>>> > { >>>> > >>>> > try (final Stream<Path> folderContents = >>>> > Files.list(path)) >>>> > { >>>> > >>>> > folderContents.forEach(stack::push); >>>> > >>>> > } >>>> > >>>> > catch (final Exception exception) >>>> > { >>>> > >>>> > throw new IllegalStateException("Failed for >>>> " + >>>> > path, exception); >>>> > >>>> > } >>>> > >>>> > } >>>> > } >>>> > >>>> > return true; >>>> > >>>> > }, >>>> > (s1, s2) -> >>>> > { >>>> > >>>> > s1.addAll(s2); >>>> > return s1; >>>> > >>>> > }, >>>> > (stack, downstream) -> >>>> > { >>>> > >>>> > for (final Path path : stack) >>>> > { >>>> > >>>> > if (!downstream.push(path)) >>>> > { >>>> > >>>> > return; >>>> > >>>> > } >>>> > >>>> > } >>>> > >>>> > } >>>> > ) >>>> > ; >>>> > >>>> > Stream >>>> > .of(root) >>>> > .gather(gatherer) >>>> > .limit(5) >>>> > .forEach(IO::println) >>>> > ; >>>> > >>>> > } >>>> > >>>> > So, problem 2 is solved, but problem 1 is not really. Sure, I could >>>> turn my >>>> > stream parallel, but the actual meat of the processing is sequential >>>> when >>>> > it really doesn't need to be. >>>> > >>>> > Let me know if this makes more sense. And sorry, you might have to >>>> scroll >>>> > to read earlier emails in this thread to get the context. I know it >>>> was >>>> > several months back. >>>> > >>>> > >>>> > On Fri, Nov 14, 2025 at 5:42 AM Remi Forax <[email protected]> wrote: >>>> > >>>> >> Hi David, >>>> >> You can always transform an imperative code to a stream by pushing >>>> the >>>> >> element through a consumer. >>>> >> Internally, a stream uses a push iterator (see >>>> >> Spliterator.tryAdvance(consumer)). >>>> >> >>>> >> As a silly example, this is a way to write fibonacci (the recursive >>>> form) >>>> >> with a stream right in the middle. >>>> >> >>>> >> >>>> >> static void fibo(int n, IntConsumer consumer) { >>>> >> if (n < 2) { >>>> >> consumer.accept(n); >>>> >> return; >>>> >> } >>>> >> var result = Stream.of("") >>>> >> .mapMultiToInt((_, consumer2) -> { >>>> >> fibo(n - 1, consumer2); >>>> >> fibo(n - 2, consumer2); >>>> >> }) >>>> >> .sum(); >>>> >> consumer.accept(result); >>>> >> } >>>> >> >>>> >> static void main() { >>>> >> fibo(7, IO::println); >>>> >> } >>>> >> >>>> >> >>>> >> Here, I use mapMulti() to convert the imperative code to a Stream >>>> >> (there is no factory method on Stream that takes a consumer of >>>> consumer). >>>> >> >>>> >> If you also want to short-circuit, you can use a gatherer instead of >>>> >> mapMulti but short-circuiting the recursive code will require to use >>>> an >>>> >> exception as control flow (it will not be pretty). >>>> >> >>>> >> regards, >>>> >> Rémi >>>> >> >>>> >> >>>> >> ------------------------------ >>>> >> >>>> >> *From: *"David Alayachew" <[email protected]> >>>> >> *To: *"core-libs-dev" <[email protected]> >>>> >> *Sent: *Tuesday, November 11, 2025 4:36:29 AM >>>> >> *Subject: *Difficulties of recursion with Streams >>>> >> >>>> >> Hello @core-libs-dev <[email protected]>, >>>> >> >>>> >> When working with streams, I often run into situations where I have >>>> to >>>> >> "demote" back to imperative code because I am trying to solve a >>>> problem >>>> >> best solved by recursion. >>>> >> >>>> >> Consider the common use case of cycling through permutations to find >>>> all >>>> >> permutations that satisfy some condition. With recursion, the answer >>>> is >>>> >> incredibly simple -- just grab an element from the set, then call the >>>> >> recursive method with a copy of the set minus the grabbed element. >>>> Once you >>>> >> reach the empty set, you've reached your terminal condition. >>>> >> >>>> >> Use cases like that are not only incredibly common, but usually, >>>> >> embarrassingly parallel. The example above of cycling through >>>> permutations >>>> >> is only a few lines of imperative code, but I struggle to imagine >>>> how I >>>> >> would do this with Streams. >>>> >> >>>> >> I guess let me start by asking -- are there any good ways currently >>>> to >>>> >> accomplish the above permutation example with Streams? And if not, >>>> should >>>> >> there be? >>>> >> >>>> >> Thank you for your time and consideration. >>>> >> David Alayachew >>>> >> >>>> >> >>>> -- >>>> Cheers, >>>> √ >>>> >>>> >>>> Viktor Klang >>>> Software Architect, Java Platform Group >>>> Oracle >>>> >>>> -- >>> Cheers, >>> √ >>> >>> >>> Viktor Klang >>> Software Architect, Java Platform Group >>> Oracle >>> >>> -- >> Cheers, >> √ >> >> >> Viktor Klang >> Software Architect, Java Platform Group >> Oracle >> >> -- > Cheers, > √ > > > Viktor Klang > Software Architect, Java Platform Group > Oracle > >
