8284199: Implementation of Structured Concurrency (Incubator)

Co-authored-by: Ron Pressler <rpressler@openjdk.org>
Co-authored-by: Alan Bateman <alanb@openjdk.org>
Co-authored-by: Brian Goetz <briangoetz@openjdk.org>
Co-authored-by: Paul Sandoz <psandoz@openjdk.org>
Reviewed-by: psandoz, mcimadamore, darcy

make/conf/docs-modules.conf

@@ -42,6 +42,7 @@ DOCS_MODULES= \
     jdk.hotspot.agent \
     jdk.httpserver \
     jdk.jpackage \
+    jdk.incubator.concurrent \
     jdk.incubator.vector \
     jdk.jartool \
     jdk.javadoc \

make/conf/module-loader-map.conf

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2014, 2021, Oracle and/or its affiliates. All rights reserved.
+# Copyright (c) 2014, 2022, Oracle and/or its affiliates. All rights reserved.
 # DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 #
 # This code is free software; you can redistribute it and/or modify it
@@ -43,6 +43,7 @@ BOOT_MODULES= \
     java.rmi \
     java.security.sasl \
     java.xml \
+    jdk.incubator.concurrent \
     jdk.incubator.vector \
     jdk.internal.vm.ci \
     jdk.jfr \

src/java.base/share/classes/module-info.java

@@ -202,6 +202,7 @@ module java.base {
         jdk.charsets,
         jdk.compiler,
         jdk.crypto.cryptoki,
+        jdk.incubator.concurrent,
         jdk.incubator.vector,
         jdk.jfr,
         jdk.jshell,

src/jdk.incubator.concurrent/share/classes/jdk/incubator/concurrent/StructureViolationException.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) 2021, 2022, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package jdk.incubator.concurrent;
+
+/**
+ * Thrown when a structure violation is detected.
+ *
+ * @see StructuredTaskScope#fork(Callable)
+ * @see StructuredTaskScope#close()
+ *
+ * @since 19
+ */
+public final class StructureViolationException extends RuntimeException {
+    @java.io.Serial
+    private static final long serialVersionUID = -7705327650798235468L;
+
+    /**
+     * Constructs a {@code StructureViolationException} with no detail message.
+     */
+    public StructureViolationException() {
+        super();
+    }
+
+    /**
+     * Constructs a {@code StructureViolationException} with the specified
+     * detail message.
+     *
+     * @param  message the detail message, can be null
+     */
+    public StructureViolationException(String message) {
+        super(message);
+    }
+}

src/jdk.incubator.concurrent/share/classes/jdk/incubator/concurrent/package-info.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2022, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+/**
+ * Defines non-final APIs for concurrent programming.
+ * {@Incubating}
+ */
+package jdk.incubator.concurrent;

src/jdk.incubator.concurrent/share/classes/module-info.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2022, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+/**
+ * Defines non-final APIs for concurrent programming.
+ * {@Incubating}
+ *
+ * @moduleGraph
+ */
+module jdk.incubator.concurrent {
+    exports jdk.incubator.concurrent;
+}
+

test/jdk/TEST.groups

@@ -274,6 +274,7 @@ jdk_loom = \
     java/util/concurrent \
     java/net/vthread \
     java/nio/channels/vthread \
+    jdk/incubator/concurrent \
     jdk/internal/misc/ThreadFlock \
     jdk/internal/vm/Continuation \
     jdk/jfr/threading
@@ -313,6 +314,7 @@ jdk_other = \
     javax/xml \
     -javax/xml/crypto \
     jdk/dynalink \
+    jdk/incubator/concurrent \
     jdk/internal/jline \
     com/sun/jndi \
     lib/testlibrary

test/jdk/jdk/incubator/concurrent/StructuredTaskScope/PreviewFeaturesNotEnabled.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2022, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+/**
+ * @test
+ * @bug 8284199
+ * @summary Test StructuredTaskScope without --enable-preview
+ * @modules jdk.incubator.concurrent
+ * @run testng/othervm PreviewFeaturesNotEnabled
+ */
+
+import jdk.incubator.concurrent.StructuredTaskScope;
+import org.testng.annotations.Test;
+import static org.testng.Assert.*;
+
+public class PreviewFeaturesNotEnabled {
+    /**
+     * One-arg constructor needs --enable-preview.
+     */
+    @Test
+    public void testUnsupportedOperationException() {
+        assertThrows(UnsupportedOperationException.class, StructuredTaskScope::new);
+    }
+
+    /**
+     * Two-arg constructor does not need --enable-preview.
+     */
+    @Test
+    public void testNoUnsupportedOperationException() {
+        try (var scope = new StructuredTaskScope<Object>(null, Thread::new)) {
+        }
+    }
+}

test/jdk/jdk/incubator/concurrent/StructuredTaskScope/StructuredThreadDumpTest.java

@@ -0,0 +1,200 @@
+/*
+ * Copyright (c) 2022, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+/*
+ * @test
+ * @bug 8284199
+ * @summary Test thread dumps with StructuredTaskScope
+ * @enablePreview
+ * @modules jdk.incubator.concurrent
+ * @library /test/lib
+ * @run testng/othervm StructuredThreadDumpTest
+ */
+
+import jdk.incubator.concurrent.StructuredTaskScope;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
+import java.util.stream.Stream;
+import com.sun.management.HotSpotDiagnosticMXBean;
+import com.sun.management.HotSpotDiagnosticMXBean.ThreadDumpFormat;
+import jdk.test.lib.threaddump.ThreadDump;
+import org.testng.annotations.Test;
+import static org.testng.Assert.*;
+
+public class StructuredThreadDumpTest {
+
+    /**
+     * Test that a thread dump with a tree of task scopes contains a thread grouping for
+     * each task scope.
+     */
+    @Test
+    public void testTree() throws Exception {
+        ThreadFactory factory = Thread.ofVirtual().factory();
+        try (var scope = new StructuredTaskScope<>("scope", factory)) {
+            Thread thread1 = fork(scope, "child-scope-A");
+            Thread thread2 = fork(scope, "child-scope-B");
+            try {
+                ThreadDump threadDump = threadDump();
+
+                // thread dump should have a thread container for each scope
+                var rootContainer = threadDump.rootThreadContainer();
+                var container1 = threadDump.findThreadContainer("scope").orElseThrow();
+                var container2 = threadDump.findThreadContainer("child-scope-A").orElseThrow();
+                var container3 = threadDump.findThreadContainer("child-scope-B").orElseThrow();
+
+                // check parents
+                assertFalse(rootContainer.parent().isPresent());
+                assertTrue(container1.parent().get() == rootContainer);
+                assertTrue(container2.parent().get() == container1);
+                assertTrue(container3.parent().get() == container1);
+
+                // check owners
+                assertFalse(rootContainer.owner().isPresent());
+                assertTrue(container1.owner().getAsLong() == Thread.currentThread().threadId());
+                assertTrue(container2.owner().getAsLong() == thread1.threadId());
+                assertTrue(container3.owner().getAsLong() == thread2.threadId());
+
+                // thread1 and threads2 should be in threads array of "scope"
+                container1.findThread(thread1.threadId()).orElseThrow();
+                container1.findThread(thread2.threadId()).orElseThrow();
+
+            } finally {
+                scope.shutdown();
+                scope.join();
+            }
+        }
+    }
+
+    /**
+     * Test that a thread dump with nested tasks scopes contains a thread grouping for
+     * each task scope.
+     */
+    @Test
+    public void testNested() throws Exception {
+        ThreadFactory factory = Thread.ofVirtual().factory();
+        try (var scope1 = new StructuredTaskScope<>("scope-A", factory)) {
+            Thread thread1 = fork(scope1);
+
+            try (var scope2 = new StructuredTaskScope<>("scope-B", factory)) {
+                Thread thread2 = fork(scope2);
+                try {
+                    ThreadDump threadDump = threadDump();
+
+                    // thread dump should have a thread container for both scopes
+                    var rootContainer = threadDump.rootThreadContainer();
+                    var container1 = threadDump.findThreadContainer("scope-A").orElseThrow();
+                    var container2 = threadDump.findThreadContainer("scope-B").orElseThrow();
+
+                    // check parents
+                    assertFalse(rootContainer.parent().isPresent());
+                    assertTrue(container1.parent().get() == rootContainer);
+                    assertTrue(container2.parent().get() == container1);
+
+                    // check owners
+                    long tid = Thread.currentThread().threadId();
+                    assertFalse(rootContainer.owner().isPresent());
+                    assertTrue(container1.owner().getAsLong() == tid);
+                    assertTrue(container2.owner().getAsLong() == tid);
+
+                    // thread1 should be in threads array of "scope-A"
+                    container1.findThread(thread1.threadId()).orElseThrow();
+
+                    // thread2 should be in threads array of "scope-B"
+                    container2.findThread(thread2.threadId()).orElseThrow();
+
+                } finally {
+                    scope2.shutdown();
+                    scope2.join();
+                }
+            } finally {
+                scope1.shutdown();
+                scope1.join();
+            }
+        }
+    }
+
+    /**
+     * Generates a JSON formatted thread dump to a temporary file, prints it to standard
+     * output, parses the JSON text and returns a ThreadDump object for the thread dump.
+     */
+    private static ThreadDump threadDump() throws IOException {
+        Path dir = Path.of(".").toAbsolutePath();
+        Path file = Files.createTempFile(dir, "threadump", "json");
+        Files.delete(file);
+        ManagementFactory.getPlatformMXBean(HotSpotDiagnosticMXBean.class)
+                .dumpThreads(file.toString(), ThreadDumpFormat.JSON);
+
+        try (Stream<String> stream = Files.lines(file)) {
+            stream.forEach(System.out::println);
+        }
+
+        String jsonText = Files.readString(file);
+        return ThreadDump.parse(jsonText);
+    }
+
+    /**
+     * Forks a subtask in the given scope that parks, returning the Thread that executes
+     * the subtask.
+     */
+    private static Thread fork(StructuredTaskScope<Object> scope) throws Exception {
+        var ref = new AtomicReference<Thread>();
+        scope.fork(() -> {
+            ref.set(Thread.currentThread());
+            LockSupport.park();
+            return null;
+        });
+        Thread thread;
+        while ((thread = ref.get()) == null) {
+            Thread.sleep(10);
+        }
+        return thread;
+    }
+
+    /**
+     * Forks a subtask in the given scope. The subtask creates a new child scope with
+     * the given name, then parks. This method returns Thread that executes the subtask.
+     */
+    private static Thread fork(StructuredTaskScope<Object> scope,
+                               String childScopeName) throws Exception {
+        var ref = new AtomicReference<Thread>();
+        scope.fork(() -> {
+            ThreadFactory factory = Thread.ofVirtual().factory();
+            try (var childScope = new StructuredTaskScope<Object>(childScopeName, factory)) {
+                ref.set(Thread.currentThread());
+                LockSupport.park();
+            }
+            return null;
+        });
+        Thread thread;
+        while ((thread = ref.get()) == null) {
+            Thread.sleep(10);
+        }
+        return thread;
+    }
+
+}