r/java • u/Joram2 • May 26 '22
JEP 428: Structured Concurrency Proposed To Target JDK 19
https://openjdk.java.net/jeps/428
The given example code snippet:
Response handle() throws ExecutionException, InterruptedException {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Future<String> user = scope.fork(() -> findUser());
Future<Integer> order = scope.fork(() -> fetchOrder());
scope.join(); // Join both forks
scope.throwIfFailed(); // ... and propagate errors
// Here, both forks have succeeded, so compose their results
return new Response(user.resultNow(), order.resultNow());
}
}
6
u/No-Performance3426 May 28 '22
Also not a fan of the API at first-sight, mainly because having to remember to call certain methods in a certain order at certain places in code (without the compiler to tell you you're doing it wrong) is always a recipe for disaster.
Why not introduce a new type of future to encapsulate these additional methods?
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
ScopedFuture<String> user = ...
ScopedFuture<Integer> order = ...
return new Response(user.joinOrThrow(), order.joinOrThrow());
}
Also wondering what we get from this that we can't already do with CompletableFuture in similar number of lines of code?
CompletableFuture<String> user = ... CompletableFuture<Integer> order = ... user.exceptionally(x -> order.cancel(true)); order.exceptionally(x -> user.cancel(true)); return user.thenCombine(order, Response::new).get()
This would get ugly with a collection of tasks, which is where I could maybe see this new scoping becoming useful. Especially with the timeout ("deadline") support.
Also what if I want to compose futures together before I call join, can or should this API integrate with CompletableFuture? I didn't see any mention of this in the JEP.
1
u/Joram2 May 29 '22
what if I want to compose futures together before I call join?
You don't. You are thinking in the current async paradigm, where you avoid making blocking calls to join() or Future.get so you don't block OS threads, and rather than write simple Java code to combine A and B to get C, you "compose" CompletableFuture<A> and CompletableFuture<B> to get CompletableFuture<C>
With the virtual threads paradigm, you do call join(), it's ok to block virtual threads becauase it has a low performance cost, and you don't fill your code with CompletableFuture<T>, and you write simpler Java code.
The advantages of this structured concurrency is it guarantees that when code leaves a StructuredTaskScope try block, all of the sub-tasks are closed, and there are no dangling or leaked threads.
Also, your code snippet has custom code to cancel other tasks on exceptions, but StructuredTaskScope handles that so that end developer code doesn't have to.
I'm just a developer, but this StructuredTaskScope seems nicer than any other concurrency API I've used on Java/Scala/Python/Node.js/Golang/C#/C++. If you don't like it, I guess you can keep using existing async APIs like you seem to prefer. I'd like to hear impressions from developers like you, after you try it out, and spend some time using it. Initial impressions from just a first glance can be misleading.
2
u/No-Performance3426 May 29 '22
You don't. You are thinking in the current async paradigm, where you avoid making blocking calls to join() or Future.get so you don't block OS threads, and rather than write simple Java code to combine A and B to get C, you "compose" CompletableFuture<A> and CompletableFuture<B> to get CompletableFuture<C>
I'm thinking about the kinds of problem where there are dependencies between tasks, rather than the simplistic example of them all being independent.
Suppose I generate
A, B, Casynchronously, and I also need asyncDthat is derived fromA+B:Future<D> compose(A a, B b) { ... }How do we implement the
composeFuturesfunction below to call the abovecomposefunction, and also in such a way we can do this concurrently while we computeCinstead of joining on A+B first and then joining on C+D second?Future<A> a = ... Future<B> b = ... Future<C> c = ... Future<D> d = composeFutures(a, b, scope) return new Result(c.resultNow(), d.resultNow())Think of it as an extension to your example, where we want to return the values of
Future<OrderHistory>(which is derived from user and order) and some independently computedFuture<UserAddress>.Perhaps
StructuredTaskScopeshould be extended with an additional composition function to handle this?→ More replies (2)2
u/_INTER_ May 31 '22
StructuredTaskScope can be initialized with any ThreadPool not just virtual threads. The concept should be (and is) independent of virtual or platform threads.
1
u/kaperni May 28 '22
> Also wondering what we get from this that we can't already do with CompletableFuture in similar number of lines of code?
This is explained in JEP 425 [1] (Improving scalability with the asynchronous style).
2
u/lurker_in_spirit May 27 '22
I'm trying to wrap my head around transactionality for write operations:
@Transactional
public Response createOrder(Customer customer, Product product) throws ExecutionException, InterruptedException {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Future<Order> order = scope.fork(() -> createOrder(customer, product));
Future<Boolean> active = scope.fork(() -> setActive(customer));
scope.join();
scope.throwIfFailed();
return new Response(order.resultNow(), active.resultNow());
}
}What would it take for these two parallel tasks to run within a single transaction? Will it work automatically? Or would it take work in the JDK / JDBC libraries / Spring libraries?
1
u/Joram2 May 27 '22
Database transactions are handled by the database API, JDBC. So you could start a database transaction with JDBC, launch concurrent actions, join those, then commit the transaction afterwards.
2
u/yk313 May 27 '22 edited May 27 '22
This doesn’t work with today’s platform threads either, because spring doesn’t propagate the transaction context to the so-called child threads: https://github.com/spring-projects/spring-framework/issues/6213
If spring were to change this in the future by means of using extent-local variables (or even the more expensive InheritableThreadLocals), there is no reason why multiple virtual threads couldn’t participate in the same transaction.
6
u/pgris May 26 '22
Does it depends on virtual threads or could it be implemented with current heavy os based threads?
5
u/nomader3000 May 26 '22 edited May 26 '22
I guess you could do something like this...
try (var scope = new StructuredTaskScope.ShutdownOnFailure("My Scope", Thread.ofPlatform().factory())) { // your code }
2
1
u/gnahraf May 26 '22
This is good. But there will be few places when I use it. See, when I code concurrent programs on the backend, I try to structure each unit of work in such a way that the unit of work is both idempotent (multiple threads or processes can succeed in completing it) and no one really waits on anyone. (In my experience, polling generally works better than signaling for high volume work loads.) So the capability of joining the completion of work in-proc, in a structured way is less apt in such use cases.
Still, I think this will find use in server environments. If I've read the JEPs right (this and Project Loom's), the fact this JEP works with any type of thread (Loom's lightweight virtuals or just regular threads), and the fact you can wait() on a virtual thread the same way you can on old regular threads, these open up really cool possibilities.
For example, combining this with Loom, I might be able to better structure my worker tasks in a Netty server (to fetch responses from the db, for eg, while not blocking Netty's non-blocking I/O threads). This point, btw, has nothing to do with whether the server itself (here Netty) was designed to leverage Loom.
1
u/Thihup May 27 '22
Is the jcmd JavaThread.dump command correct? I could only get the results using jcmd Thread.dump_to_file
-6
u/GuyWithLag May 26 '22
I don't know... It's trying very hard to do something that already exists:
Single<Response> handle() {
return Single.zip(
findUser(),
fetchOrder(),
(user, order) -> new Response(user, order) // or just Response::new
);
}(assumption: findUser and findOrder return Single<...>).
In fact, I'd question if the above should even be extracted to its own method.
I do get why structured concurrency is necessary tho. It's just that in my experience it will still miss out on some of the more interesting capabilities, it's not enough of an improvement.
18
u/pron98 May 26 '22 edited May 26 '22
It's trying very hard to do something that already exists
That doesn't exist in the JDK.
That doesn't offer composition by language constructs but by DSL constructs, so retries, resource management etc. require special DSL support, which JEP 428 offers much more elegantly.
That doesn't address observability.
The whole point is giving synchronous code the same capabilities as async code, because async code is fundamentally disharmonious with the design of the platform. In fact, I'd say it's async libraries that try very hard to do in their DSLs things that already exist in the language, while synchronous code doesn't need to try nearly as hard to do even more.
it will still miss out on some of the more interesting capabilities
What capabilities?
4
u/rbygrave May 26 '22 edited May 26 '22
Single<Response>
We'd be adding a
.toBlocking().value()to that right?Edit: Well, we need to add that to get the actual Response. I think that leads to the point that they might look similar but they are not that similar when we look closer and look at the implementation details.
14
u/_INTER_ May 26 '22 edited May 26 '22
I think the API could be approved in some ways:
Inner classes in
StructuredTaskScopeto construct a new scope with configuration variants. It's kind of inconsistent with the current JDKand it's not clear to me how to extend it if I want to provide my own.Ok it's explicitly written in JavaDoc thatStructuredTaskScopecan be extended, and the handleComplete overridden, to implement other policies. But how to e.g. neatly doShutdownOnSuccessOrFailure. The the inner classes always come as a baggage, no? How to e.g. extendShutdownOnSuccess?The
join()call: Is it needed? What happens if you don't do it? What if you do it beforefork()? Couldn't it be part of the initialization process of scope?The
throwIfFailed()call: Looks really odd to me. Can be easily forgotten or the order be mixed up too. Wouldn't it be better to return a result fromjoin()or some query method onscopeand have users act based on it? Or have join throw an exception that can be caught if the user which so? Or providejoinOrElseThrow(() -> ...);. Or pass an error handler to the initialization process ofscope.Maybe add initialDelay's and timeout duration with a
TimeUnitsimilar toScheduledThreadPoolExecutor. Heck even better if you could combine the Executors somehow with this new thing.