Structured Concurrency yet again

[forax at univ-mlv.fr]

----- Original Message -----
> From: "Ron Pressler" <ron.pressler at oracle.com>
> To: "Remi Forax" <forax at univ-mlv.fr>
> Cc: "Alan Bateman" <alan.bateman at oracle.com>, "loom-dev" <loom-dev at openjdk.java.net>
> Sent: Thursday, May 11, 2023 3:55:13 PM
> Subject: Re: Structured Concurrency yet again

>> On 11 May 2023, at 09:24, forax at univ-mlv.fr wrote:
>> 
>> 
>> Errgroup is a special construct that is the equivalent of STS.ShutdownOnFailure
>> (you can also implement ShutdownOnSucess because there is a method cancel()
>> (the equivalent of shutdown())).
>> Errgroup does not have to cover all scenarios because in Go you can always
>> fallback on using channels. Errgroup is a nice structured concurrency API for
>> some special cases while channels is the more general API.
> 
> I’m not sure I see how channels generalise SC, but we’d like to have channels
> too, and in the meantime we have all of the existing classing in j.u.c at our
> disposal. The real low-level mechanism underneath STS is the ThreadFlock — as
> that’s what integrates with observability — and we’d like to expose that, too,
> eventually.

Channels do not generalize SC, they generalize errgroup but channels are not a SC API.

> 
>> 
>> In term of typing, Errgroup is not better than STS.ShutdownOnFailure because
>> both erase the type of the error/exception. Note that errgroup what developed
>> before go had generics.
> 
> What do you mean by *the* type? In many (most?) fan-out situations, tasks even
> have different *return* types.

The type of the "exception" is erased to Throwable
(see https://docs.oracle.com/en/java/javase/20/docs/api/jdk.incubator.concurrent/jdk/incubator/concurrent/StructuredTaskScope.ShutdownOnFailure.html#exception())

> 
>> 
>> I believe STS should not focus only on the special cases STS.ShutdownOnFailure
>> and STS.ShutdownOnSucess, but also provides a thread safe way to implement most
>> of the general cases.
> 
> Again, before proposing something, I need to understand what problem you’re
> trying to solve. Without articulating the problem it’s very hard to evaluate
> solutions for it.
> 
>> What i am proposing cover all scenarios that could be implemented by a channel
>> with multiple writers but only one reader. For those cases, I believe that the
>> a join(Function<Stream,T>) is a better API than the channel API because it is a
>> structured concurrency API and it's also better than letting users to write a
>> poor man implementation of channels as a subclass of STS.
> 
> 
>> what i'm saying is not that we need channels, as you said it's an orthogonal
>> feature, what i'm saying is that there is a category of problem that could be
>> solved by a channel with multiple writers and only one reader, for those
>> problem, we can also provide a structured concurrency API.
> 
> Isn’t join(Function<Stream,T>), why t that something that a subclass of STS
> couldn’t offer? Isn’t it a matter of identifying a third common use-case (that
> solves a problem we’ve yet to articulate) beyond SutdownOnFailure/Success and
> offering a third built-in policy?

yes, the idea of join(Function<Stream,T>) (and it's variant with a deadline) is to offer a third policy.

The problems it solves are
- filtering to exclude/transform the results (filter, flatMap, mapMulti)
- reducing all succeeding results to one value either in a functional way or as mutable collection (reduce, collect)
- aggregating several errors (by example as suppressed exceptions) instead of keeping only the first error (reduce, collect)
- canceling/short-circuiting before all results are computed (anyMatch, noneMatch, findFirst, takeWhile)
- a combination of the problems listed above

All of them are implementable by subclassing STS (it's what's my implementation does) but it easier if we provide a builtin policy.

> 
> — Ron

Rémi