mapConcurrent() with InterruptedException

[Jige Yu]

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
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://mail.openjdk.org/pipermail/core-libs-dev/attachments/20250103/61480109/attachment-0001.htm>