Hi Ben,
Thanks for raising these questions—getting feedback is crucial in the Preview
stage of features.
I wrote a reply to the Reddit thread so I'll just summarize here:
It is important to note that mapConcurrent() is not a part of the Structured
Concurrency JEPs, so it is not designed to join SC scopes.
I'm currently experimenting with ignoring-but-restoring interrupts on the
"calling thread" for mapConcurrent(), as well as capping work-in-progress to
maxConcurrency (not only capping the concurrency but also the amount of
completed-but-yet-to-be-pushed work). Both of these adjustments should increase
predictability of behavior in the face of blocking operations with variable
delays.
Another adjustment I'm looking at right now is to harden/improve the cleanup to
wait for concurrent tasks to acknowledge cancellation, so that once the
finisher is done executing the VTs are known to have terminated.
As for not preserving the encounter order, that would be a completely different
thing, and I'd encourage you to experiment with that if that functionality
would be interesting for your use-case(s).
Again, thanks for your feedback!
Cheers,
√
Viktor Klang
Software Architect, Java Platform Group
Oracle
________________________________
From: core-libs-dev <core-libs-dev-r...@openjdk.org> on behalf of Jige Yu
<yuj...@gmail.com>
Sent: Friday, 3 January 2025 17:53
To: core-libs-dev@openjdk.org <core-libs-dev@openjdk.org>
Subject: mapConcurrent() with InterruptedException
Hi Java Experts,
I sent this email incorrectly to loom-dev@ and was told on Reddit that
core-libs-dev is the right list.
The question is about the behavior of mapConcurrent() when the thread is
interrupted.
Currently mapConcurrent()'s finisher phase will re-interrupt the thread, then
stop at whatever element that has already been processed and return.
This strikes me as a surprising behavior, because for example if I'm running:
Stream.of(1, 2, 3)
.gather(mapConcurrent(i -> i * 2))
.toList()
and the thread is being interrupted, the result could be any of [2], [2, 4] or
[2, 4, 6].
Since thread interruption is cooperative, there is no guarantee that the thread
being interrupted will just abort. It's quite possible that it'll keep going
and then will use for example [2] as the result of doubling the list of [1, 2,
3], which is imho incorrect.
In the
Reddit<https://www.reddit.com/r/java/comments/1hr8xyu/observations_of_gatherersmapconcurrent/?utm_source=share&utm_medium=web3x&utm_name=web3xcss&utm_term=1&utm_content=share_button>
thread, someone argued that interruption rarely happens so it's more of a
theoretical issue. But interruption can easily happen in Structured Concurrency
or in mapConcurrent() itself if any subtask has failed in order to
cancel/interrupt the other ongoing tasks.
There had been discussion about alternative strategies:
1. Don't respond to interruption and just keep running to completion.
2. Re-interrupt thread and wrap the InterruptedException in a standard
unchecked exception (StructuredConcurrencyInterruptedException?).
I have concerns with option 1 because it disables cancellation propagation when
mapConcurrent() itself is used in a subtask of a parent mapConcurrent() or in a
StructuredConcurrencyScope.
Both equivalent Future-composition async code, or C++'s fiber trees support
cancellation propagation and imho it's a critical feature or else it's possible
that a zombie thread is still sending RPCs long after the main thread has
exited (failed, or falled back to some default action).
My arguments for option 2:
1. InterruptedException is more error prone than traditional checked
exceptions for users to catch and handle. They can forget to re-interrupt the
thread. It's so confusing that even seasoned programmers may not know they are
supposed to re-interrupt the thread.
2. With Stream API using functional interfaces like Supplier, Function, the
option of just tacking on "throws IE" isn't available to many users.
3. With Virtual Threads, it will be more acceptable, or even become common
to do blocking calls from a stream operation (including but exclusive to
mapConcurrent()). So the chance users are forced to deal with IE will become
substantially higher.
4. Other APIs such as the Structured Concurrency API have already started
wrapping system checked exceptions like ExecutionException, TimeoutException in
unchecked exceptions (
join()<https://download.java.net/java/early_access/loom/docs/api/java.base/java/util/concurrent/StructuredTaskScope.html#join()>
for example).
5. Imho, exceptions that we'd rather users not catch and handle but instead
should mostly just propagate up as is, should be unchecked.
There is also a side
discussion<https://www.reddit.com/r/java/comments/1hr8xyu/comment/m4z4f8c/?utm_source=share&utm_medium=web3x&utm_name=web3xcss&utm_term=1&utm_content=share_button>
about whether mapConcurrent() is better off preserving input order or push to
downstream as soon as an element is computed. I'd love to discuss that topic
too but maybe it's better to start a separate thread?
Thank you and cheers!
Ben Yu
