Structured Concurrency vs the Streams API

[Remi Forax]

----- Original Message -----
> From: "Arkadiusz Gasiński" <jigga at jigga.pl>
> To: "Ron Pressler" <ron.pressler at oracle.com>
> Cc: "loom-dev" <loom-dev at openjdk.java.net>
> Sent: Sunday, May 8, 2022 12:14:45 AM
> Subject: Re: Structured Concurrency vs the Streams API

> Thanks Ron.
> 
> I'm actually working on the "Structured Concurrency in Java" presentation
> that I plan to give at a local JUG meeting at the end of May and after
> playing a bit with the SC APIs in the EA Loom builds, I wonder if Java's
> intent is purely about containing the threads in a well defined scope or
> will there be more to it?
> 
> Practically all presentations on the subject I've seen so far start with
> going down the "goto statement considered harmful" memory lane and are
> trying to prove that thread spawning, futures, promises, etc. are "not even
> the modern domesticated goto, but the old-testament fire-and-brimstone goto"
> of concurrency. Do you agree with this?
> 
> Also, one of Dijkstra's remarks in the "Case against the GO TO statement"
> was that we should do our best to shorten the gap between how the program
> is spread out in text and its runtime (process) representation. Do you
> think this is feasible thing to achieve in structured Concurrency?
> 
> One last thing - I made an attempt at implementing a rather dummy (don't
> take IPv4 vs IPv6 into consideration) version of the Happy Eyeballs
> algorithm with the StructuredTaskScope API (
> https://github.com/jigga/classes/blob/main/HappyEyeballs.java) - similar to
> what N. J. Smith did with his Trio library in this (
> https://youtu.be/oLkfnc_UMcE) talk. Would you be so kind to have a look and
> give some feedback? While it works (starts subsequent task when the
> previous one times out or fails and finishes as soon the first succeeds), I
> think it's still lacking in cleanliness. Right now I'm thinking if it
> wouldn't be clearer/more idiomatic to encapsulate the delay between tasks
> submissions in some custom policy (e.g.
> StructuredTaskScope.ShutdownOnSuccessForkWithDelay), where you'd fork all
> tasks immediately (ideally a call to fork would return instantaneously and
> not after timeout passes) and then just join like in other use cases. Any
> thoughts?

Hi,
I've slightly refactor your code to make it more readable and i find your solution quite elegant.

public class HappyEyeballs {
  private static Socket connect(String hostname) throws InterruptedException, IOException {
    var addresses = InetAddress.getAllByName(hostname);
    try (var scope = new StructuredTaskScope.ShutdownOnSuccess<Socket>()) {
      var prev = scope.fork(task(addresses[0]));
      for (var i = 1; i < addresses.length; i++) {
        try {
          prev.get(250, TimeUnit.MILLISECONDS);
          break;
        } catch (ExecutionException | TimeoutException e) {
          System.out.println("Got " + e + " when connecting to " + addresses[i - 1]);
          prev = scope.fork(task(addresses[i]));
        } catch (CancellationException e) {
          System.out.println("Cancel with " + e + " when connecting to " + addresses[i - 1]);
          break;
        }
      }
      scope.join();
      return scope.result(e -> new IOException("Could not connect to " + hostname));
    }
  }

  private static Callable<Socket> task(InetAddress address) {
    return () -> {
      System.out.println(Thread.currentThread() + " - connecting to: " + address);
      Thread.sleep(address instanceof Inet4Address ? 300 : 100);
      var socket = new Socket(address, 443);
      System.out.println("Successfully connected to: " + address);
      return socket;
    };
  }

  public static void main(String[] args) throws Exception {
    var hostname = args[0];
    System.out.println(connect(hostname));
  }
}

> 
> Thanks,
> Arek

Rémi

> 
> On Fri, May 6, 2022, 12:09 PM Ron Pressler <ron.pressler at oracle.com> wrote:
> 
>> Hi.
>>
>> Parallel streams are indeed structured, but they’re (currently) focused on
>> pure processing and data parallelism, rather than “concurrency,” which
>> often involves a lot of I/O, failure handling, and cancellation.
>>
>> It is also true that the Stream API could be modified to implement some
>> flavour of structued concurrency, and it is an idea that we’re thinking
>> about.
>>
>> — Ron
>>
>> > On 6 May 2022, at 09:53, Arkadiusz Gasiński <jigga at jigga.pl> wrote:
>> >
>> > Hi all,
>> >
>> > Just a random thought of mine that I'd like loom-dev to comment on.
>> >
>> > Do you guys think it is a valid claim to call the (parallel) stream API
>> an
>> > implementation of the Structured Concurrency paradigm?
>> >
>> > The more I think about it the more I'm convinced that this is indeed the
>> > case, but would like to hear some more opinions.
>> >
>> > Thanks,
>> > Arek
>>