1 /*
2 * Copyright (c) 2021, 2023, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation.
8 *
9 * This code is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
12 * version 2 for more details (a copy is included in the LICENSE file that
13 * accompanied this code).
14 *
15 * You should have received a copy of the GNU General Public License version
16 * 2 along with this work; if not, write to the Free Software Foundation,
17 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18 *
19 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
20 * or visit www.oracle.com if you need additional information or have any
21 * questions.
22 */
23
24 /*
25 * @test id=platform
26 * @bug 8284199 8296779 8306647
27 * @summary Basic tests for StructuredTaskScope
28 * @enablePreview
29 * @run junit/othervm -DthreadFactory=platform StructuredTaskScopeTest
30 */
31
32 /*
33 * @test id=virtual
34 * @enablePreview
35 * @run junit/othervm -DthreadFactory=virtual StructuredTaskScopeTest
36 */
37
38 import java.time.Duration;
39 import java.io.IOException;
40 import java.time.Instant;
41 import java.util.Arrays;
42 import java.util.ArrayList;
43 import java.util.List;
44 import java.util.Set;
45 import java.util.concurrent.Callable;
46 import java.util.concurrent.ConcurrentHashMap;
47 import java.util.concurrent.CountDownLatch;
48 import java.util.concurrent.Executors;
49 import java.util.concurrent.ExecutionException;
50 import java.util.concurrent.Future;
51 import java.util.concurrent.ThreadFactory;
52 import java.util.concurrent.TimeoutException;
53 import java.util.concurrent.TimeUnit;
54 import java.util.concurrent.RejectedExecutionException;
55 import java.util.concurrent.ScheduledExecutorService;
56 import java.util.concurrent.StructuredTaskScope;
57 import java.util.concurrent.StructuredTaskScope.Subtask;
58 import java.util.concurrent.StructuredTaskScope.ShutdownOnSuccess;
59 import java.util.concurrent.StructuredTaskScope.ShutdownOnFailure;
60 import java.util.concurrent.StructureViolationException;
61 import java.util.concurrent.atomic.AtomicBoolean;
62 import java.util.concurrent.atomic.AtomicInteger;
63 import java.util.concurrent.atomic.AtomicReference;
64 import java.util.function.Supplier;
65 import java.util.stream.Stream;
66 import static java.lang.Thread.State.*;
67
68 import org.junit.jupiter.api.Test;
69 import org.junit.jupiter.api.BeforeAll;
70 import org.junit.jupiter.api.AfterAll;
71 import org.junit.jupiter.params.ParameterizedTest;
72 import org.junit.jupiter.params.provider.MethodSource;
73 import static org.junit.jupiter.api.Assertions.*;
74
75 class StructuredTaskScopeTest {
76 private static ScheduledExecutorService scheduler;
77 private static List<ThreadFactory> threadFactories;
78
79 @BeforeAll
80 static void setup() throws Exception {
81 scheduler = Executors.newSingleThreadScheduledExecutor();
82
83 // thread factories
84 String value = System.getProperty("threadFactory");
85 List<ThreadFactory> list = new ArrayList<>();
86 if (value == null || value.equals("platform"))
87 list.add(Thread.ofPlatform().factory());
88 if (value == null || value.equals("virtual"))
89 list.add(Thread.ofVirtual().factory());
90 assertTrue(list.size() > 0, "No thread factories for tests");
91 threadFactories = list;
92 }
93
94 @AfterAll
95 static void shutdown() {
96 scheduler.shutdown();
97 }
98
99 private static Stream<ThreadFactory> factories() {
100 return threadFactories.stream();
101 }
102
103 /**
104 * Test that fork creates a new thread for each task.
105 */
106 @ParameterizedTest
107 @MethodSource("factories")
108 void testForkCreatesThread(ThreadFactory factory) throws Exception {
109 Set<Long> tids = ConcurrentHashMap.newKeySet();
110 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
111 for (int i = 0; i < 100; i++) {
112 scope.fork(() -> {
113 tids.add(Thread.currentThread().threadId());
114 return null;
115 });
116 }
117 scope.join();
118 }
119 assertEquals(100, tids.size());
120 }
121
122 /**
123 * Test that fork creates a new virtual thread for each task.
124 */
125 @Test
126 void testForkCreateVirtualThread() throws Exception {
127 Set<Thread> threads = ConcurrentHashMap.newKeySet();
128 try (var scope = new StructuredTaskScope<Object>()) {
129 for (int i = 0; i < 100; i++) {
130 scope.fork(() -> {
131 threads.add(Thread.currentThread());
132 return null;
133 });
134 }
135 scope.join();
136 }
137 assertEquals(100, threads.size());
138 threads.forEach(t -> assertTrue(t.isVirtual()));
139 }
140
141 /**
142 * Test that fork creates a new thread with the given thread factory.
143 */
144 @ParameterizedTest
145 @MethodSource("factories")
146 void testForkUsesFactory(ThreadFactory factory) throws Exception {
147 var count = new AtomicInteger();
148 ThreadFactory countingFactory = task -> {
149 count.incrementAndGet();
150 return factory.newThread(task);
151 };
152 try (var scope = new StructuredTaskScope<Object>(null, countingFactory)) {
153 for (int i = 0; i < 100; i++) {
154 scope.fork(() -> null);
155 }
156 scope.join();
157 }
158 assertEquals(100, count.get());
159 }
160
161 /**
162 * Test fork is confined to threads in the scope "tree".
163 */
164 @ParameterizedTest
165 @MethodSource("factories")
166 void testForkConfined(ThreadFactory factory) throws Exception {
167 try (var scope1 = new StructuredTaskScope<Boolean>();
168 var scope2 = new StructuredTaskScope<Boolean>()) {
169
170 // thread in scope1 cannot fork thread in scope2
171 Subtask<Boolean> subtask1 = scope1.fork(() -> {
172 assertThrows(WrongThreadException.class, () -> {
173 scope2.fork(() -> null);
174 });
175 return true;
176 });
177
178 // thread in scope2 can fork thread in scope1
179 Subtask<Boolean> subtask2 = scope2.fork(() -> {
180 scope1.fork(() -> null);
181 return true;
182 });
183
184 scope2.join();
185 scope1.join();
186
187 assertTrue(subtask1.get());
188 assertTrue(subtask2.get());
189
190 // random thread cannot fork
191 try (var pool = Executors.newSingleThreadExecutor()) {
192 Future<Void> future = pool.submit(() -> {
193 assertThrows(WrongThreadException.class, () -> {
194 scope1.fork(() -> null);
195 });
196 assertThrows(WrongThreadException.class, () -> {
197 scope2.fork(() -> null);
198 });
199 return null;
200 });
201 future.get();
202 }
203 }
204 }
205
206 /**
207 * Test fork after join completes.
208 */
209 @ParameterizedTest
210 @MethodSource("factories")
211 void testForkAfterJoin(ThreadFactory factory) throws Exception {
212 try (var scope = new StructuredTaskScope<String>(null, factory)) {
213 // round 1
214 var subtask1 = scope.fork(() -> "foo");
215 assertThrows(IllegalStateException.class, subtask1::get);
216 scope.join();
217 assertEquals("foo", subtask1.get());
218
219 // round 2
220 var subtask2 = scope.fork(() -> "bar");
221 assertEquals("foo", subtask1.get());
222 assertThrows(IllegalStateException.class, subtask2::get);
223 scope.join();
224 assertEquals("foo", subtask1.get());
225 assertEquals("bar", subtask2.get());
226
227 // round 3
228 var subtask3 = scope.fork(() -> "baz");
229 assertEquals("foo", subtask1.get());
230 assertEquals("bar", subtask2.get());
231 assertThrows(IllegalStateException.class, subtask3::get);
232 scope.join();
233 assertEquals("foo", subtask1.get());
234 assertEquals("bar", subtask2.get());
235 assertEquals("baz", subtask3.get());
236 }
237 }
238
239 /**
240 * Test fork after join throws.
241 */
242 @ParameterizedTest
243 @MethodSource("factories")
244 void testForkAfterJoinThrows(ThreadFactory factory) throws Exception {
245 try (var scope = new StructuredTaskScope<String>(null, factory)) {
246 var latch = new CountDownLatch(1);
247 var subtask1 = scope.fork(() -> {
248 latch.await();
249 return "foo";
250 });
251
252 // join throws
253 Thread.currentThread().interrupt();
254 assertThrows(InterruptedException.class, scope::join);
255
256 // allow subtask1 to finish
257 latch.countDown();
258
259 // continue to fork
260 var subtask2 = scope.fork(() -> "bar");
261 assertThrows(IllegalStateException.class, subtask1::get);
262 assertThrows(IllegalStateException.class, subtask2::get);
263 scope.join();
264 assertEquals("foo", subtask1.get());
265 assertEquals("bar", subtask2.get());
266 }
267 }
268
269 /**
270 * Test fork after scope is shutdown.
271 */
272 @ParameterizedTest
273 @MethodSource("factories")
274 void testForkAfterShutdown(ThreadFactory factory) throws Exception {
275 var executed = new AtomicBoolean();
276 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
277 scope.shutdown();
278 Subtask<String> subtask = scope.fork(() -> {
279 executed.set(true);
280 return null;
281 });
282 scope.join();
283 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
284 assertThrows(IllegalStateException.class, subtask::get);
285 assertThrows(IllegalStateException.class, subtask::exception);
286 }
287 assertFalse(executed.get());
288 }
289
290 /**
291 * Test fork after scope is closed.
292 */
293 @ParameterizedTest
294 @MethodSource("factories")
295 void testForkAfterClose(ThreadFactory factory) throws Exception {
296 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
297 scope.close();
298 assertThrows(IllegalStateException.class, () -> scope.fork(() -> null));
299 }
300 }
301
302 /**
303 * Test fork when the thread factory rejects creating a thread.
304 */
305 @Test
306 void testForkRejectedExecutionException() throws Exception {
307 ThreadFactory factory = task -> null;
308 try (var scope = new StructuredTaskScope(null, factory)) {
309 assertThrows(RejectedExecutionException.class, () -> scope.fork(() -> null));
310 scope.join();
311 }
312 }
313
314 /**
315 * Test join with no subtasks.
316 */
317 @Test
318 void testJoinWithNoSubtasks() throws Exception {
319 try (var scope = new StructuredTaskScope()) {
320 scope.join();
321 }
322 }
323
324 /**
325 * Test join with unfinished subtasks.
326 */
327 @ParameterizedTest
328 @MethodSource("factories")
329 void testJoinWithSubtasks(ThreadFactory factory) throws Exception {
330 try (var scope = new StructuredTaskScope(null, factory)) {
331 Subtask<String> subtask = scope.fork(() -> {
332 Thread.sleep(Duration.ofMillis(50));
333 return "foo";
334 });
335 scope.join();
336 assertEquals("foo", subtask.get());
337 }
338 }
339
340 /**
341 * Test join is owner confined.
342 */
343 @ParameterizedTest
344 @MethodSource("factories")
345 void testJoinConfined(ThreadFactory factory) throws Exception {
346 try (var scope = new StructuredTaskScope<Boolean>()) {
347
348 // thread in scope cannot join
349 Subtask<Boolean> subtask = scope.fork(() -> {
350 assertThrows(WrongThreadException.class, () -> { scope.join(); });
351 return true;
352 });
353
354 scope.join();
355
356 assertTrue(subtask.get());
357
358 // random thread cannot join
359 try (var pool = Executors.newSingleThreadExecutor()) {
360 Future<Void> future = pool.submit(() -> {
361 assertThrows(WrongThreadException.class, scope::join);
362 return null;
363 });
364 future.get();
365 }
366 }
367 }
368
369 /**
370 * Test join with interrupt status set.
371 */
372 @ParameterizedTest
373 @MethodSource("factories")
374 void testInterruptJoin1(ThreadFactory factory) throws Exception {
375 try (var scope = new StructuredTaskScope(null, factory)) {
376 var latch = new CountDownLatch(1);
377
378 Subtask<String> subtask = scope.fork(() -> {
379 latch.await();
380 return "foo";
381 });
382
383 // join should throw
384 Thread.currentThread().interrupt();
385 try {
386 scope.join();
387 fail("join did not throw");
388 } catch (InterruptedException expected) {
389 assertFalse(Thread.interrupted()); // interrupt status should be clear
390 } finally {
391 // let task continue
392 latch.countDown();
393 }
394
395 // join should complete
396 scope.join();
397 assertEquals("foo", subtask.get());
398 }
399 }
400
401 /**
402 * Test interrupt of thread blocked in join.
403 */
404 @ParameterizedTest
405 @MethodSource("factories")
406 void testInterruptJoin2(ThreadFactory factory) throws Exception {
407 try (var scope = new StructuredTaskScope(null, factory)) {
408 var latch = new CountDownLatch(1);
409 Subtask<String> subtask = scope.fork(() -> {
410 latch.await();
411 return "foo";
412 });
413
414 // join should throw
415 scheduleInterruptAt("java.util.concurrent.StructuredTaskScope.join");
416 try {
417 scope.join();
418 fail("join did not throw");
419 } catch (InterruptedException expected) {
420 assertFalse(Thread.interrupted()); // interrupt status should be clear
421 } finally {
422 // let task continue
423 latch.countDown();
424 }
425
426 // join should complete
427 scope.join();
428 assertEquals("foo", subtask.get());
429 }
430 }
431
432 /**
433 * Test join when scope is shutdown.
434 */
435 @ParameterizedTest
436 @MethodSource("factories")
437 void testJoinWithShutdown1(ThreadFactory factory) throws Exception {
438 try (var scope = new StructuredTaskScope<String>(null, factory)) {
439 var interrupted = new CountDownLatch(1);
440 var finish = new CountDownLatch(1);
441
442 Subtask<String> subtask = scope.fork(() -> {
443 try {
444 Thread.sleep(Duration.ofDays(1));
445 } catch (InterruptedException e) {
446 interrupted.countDown();
447 }
448 finish.await();
449 return "foo";
450 });
451
452 scope.shutdown(); // should interrupt task
453
454 interrupted.await();
455
456 scope.join();
457
458 // signal task to finish
459 finish.countDown();
460 }
461 }
462
463 /**
464 * Test shutdown when owner is blocked in join.
465 */
466 @ParameterizedTest
467 @MethodSource("factories")
468 void testJoinWithShutdown2(ThreadFactory factory) throws Exception {
469 class MyScope<T> extends StructuredTaskScope<T> {
470 MyScope(ThreadFactory factory) {
471 super(null, factory);
472 }
473 @Override
474 protected void handleComplete(Subtask<? extends T> subtask) {
475 shutdown();
476 }
477 }
478
479 try (var scope = new MyScope<String>(factory)) {
480 Subtask<String> subtask1 = scope.fork(() -> {
481 Thread.sleep(Duration.ofMillis(50));
482 return "foo";
483 });
484 Subtask<String> subtask2 = scope.fork(() -> {
485 Thread.sleep(Duration.ofDays(1));
486 return "bar";
487 });
488
489 // join should wakeup when shutdown is called
490 scope.join();
491
492 // task1 should have completed successfully
493 assertEquals(Subtask.State.SUCCESS, subtask1.state());
494 assertEquals("foo", subtask1.get());
495 assertThrows(IllegalStateException.class, subtask1::exception);
496
497 // task2 result/exception not available
498 assertEquals(Subtask.State.UNAVAILABLE, subtask2.state());
499 assertThrows(IllegalStateException.class, subtask2::get);
500 assertThrows(IllegalStateException.class, subtask2::exception);
501 }
502 }
503
504 /**
505 * Test join after scope is closed.
506 */
507 @Test
508 void testJoinAfterClose() throws Exception {
509 try (var scope = new StructuredTaskScope()) {
510 scope.join();
511 scope.close();
512 assertThrows(IllegalStateException.class, () -> scope.join());
513 assertThrows(IllegalStateException.class, () -> scope.joinUntil(Instant.now()));
514 }
515 }
516
517 /**
518 * Test joinUntil, subtasks finish before deadline expires.
519 */
520 @ParameterizedTest
521 @MethodSource("factories")
522 void testJoinUntil1(ThreadFactory factory) throws Exception {
523 try (var scope = new StructuredTaskScope<String>(null, factory)) {
524 Subtask<String> subtask = scope.fork(() -> {
525 try {
526 Thread.sleep(Duration.ofSeconds(2));
527 } catch (InterruptedException e) { }
528 return "foo";
529 });
530
531 long startMillis = millisTime();
532 scope.joinUntil(Instant.now().plusSeconds(30));
533 expectDuration(startMillis, /*min*/1900, /*max*/20_000);
534 assertEquals("foo", subtask.get());
535 }
536 }
537
538 /**
539 * Test joinUntil, deadline expires before subtasks finish.
540 */
541 @ParameterizedTest
542 @MethodSource("factories")
543 void testJoinUntil2(ThreadFactory factory) throws Exception {
544 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
545 Subtask<Void> subtask = scope.fork(() -> {
546 Thread.sleep(Duration.ofDays(1));
547 return null;
548 });
549
550 long startMillis = millisTime();
551 try {
552 scope.joinUntil(Instant.now().plusSeconds(2));
553 } catch (TimeoutException e) {
554 expectDuration(startMillis, /*min*/1900, /*max*/20_000);
555 }
556 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
557 }
558 }
559
560 /**
561 * Test joinUntil many times.
562 */
563 @ParameterizedTest
564 @MethodSource("factories")
565 void testJoinUntil3(ThreadFactory factory) throws Exception {
566 try (var scope = new StructuredTaskScope<String>(null, factory)) {
567 Subtask<String> subtask = scope.fork(() -> {
568 Thread.sleep(Duration.ofDays(1));
569 return null;
570 });
571
572 for (int i = 0; i < 3; i++) {
573 try {
574 scope.joinUntil(Instant.now().plusMillis(50));
575 fail("joinUntil did not throw");
576 } catch (TimeoutException expected) {
577 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
578 }
579 }
580 }
581 }
582
583 /**
584 * Test joinUntil with a deadline that has already expired.
585 */
586 @ParameterizedTest
587 @MethodSource("factories")
588 void testJoinUntil4(ThreadFactory factory) throws Exception {
589 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
590 Subtask<Void> subtask = scope.fork(() -> {
591 Thread.sleep(Duration.ofDays(1));
592 return null;
593 });
594
595 // now
596 try {
597 scope.joinUntil(Instant.now());
598 fail("joinUntil did not throw");
599 } catch (TimeoutException expected) {
600 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
601 }
602
603 // in the past
604 try {
605 scope.joinUntil(Instant.now().minusSeconds(1));
606 fail("joinUntil did not throw");
607 } catch (TimeoutException expected) {
608 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
609 }
610 }
611 }
612
613 /**
614 * Test joinUntil with interrupt status set.
615 */
616 @ParameterizedTest
617 @MethodSource("factories")
618 void testInterruptJoinUntil1(ThreadFactory factory) throws Exception {
619 try (var scope = new StructuredTaskScope<String>(null, factory)) {
620 var latch = new CountDownLatch(1);
621
622 Subtask<String> subtask = scope.fork(() -> {
623 latch.await();
624 return "foo";
625 });
626
627 // joinUntil should throw
628 Thread.currentThread().interrupt();
629 try {
630 scope.joinUntil(Instant.now().plusSeconds(30));
631 fail("joinUntil did not throw");
632 } catch (InterruptedException expected) {
633 assertFalse(Thread.interrupted()); // interrupt status should be clear
634 } finally {
635 // let task continue
636 latch.countDown();
637 }
638
639 // join should complete
640 scope.join();
641 assertEquals("foo", subtask.get());
642 }
643 }
644
645 /**
646 * Test interrupt of thread blocked in joinUntil.
647 */
648 @ParameterizedTest
649 @MethodSource("factories")
650 void testInterruptJoinUntil2(ThreadFactory factory) throws Exception {
651 try (var scope = new StructuredTaskScope(null, factory)) {
652 var latch = new CountDownLatch(1);
653
654 Subtask<String> subtask = scope.fork(() -> {
655 latch.await();
656 return "foo";
657 });
658
659 // joinUntil should throw
660 scheduleInterruptAt("java.util.concurrent.StructuredTaskScope.joinUntil");
661 try {
662 scope.joinUntil(Instant.now().plusSeconds(30));
663 fail("joinUntil did not throw");
664 } catch (InterruptedException expected) {
665 assertFalse(Thread.interrupted()); // interrupt status should be clear
666 } finally {
667 // let task continue
668 latch.countDown();
669 }
670
671 // join should complete
672 scope.join();
673 assertEquals("foo", subtask.get());
674 }
675 }
676
677 /**
678 * Test that shutdown interrupts unfinished subtasks.
679 */
680 @ParameterizedTest
681 @MethodSource("factories")
682 void testShutdownInterruptsThreads1(ThreadFactory factory) throws Exception {
683 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
684 var interrupted = new AtomicBoolean();
685 var latch = new CountDownLatch(1);
686 var subtask = scope.fork(() -> {
687 try {
688 Thread.sleep(Duration.ofDays(1));
689 } catch (InterruptedException e) {
690 interrupted.set(true);
691 } finally {
692 latch.countDown();
693 }
694 return null;
695 });
696
697 scope.shutdown();
698
699 // wait for task to complete
700 latch.await();
701 assertTrue(interrupted.get());
702
703 scope.join();
704
705 // subtask result/exception not available
706 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
707 assertThrows(IllegalStateException.class, subtask::get);
708 assertThrows(IllegalStateException.class, subtask::exception);
709 }
710 }
711
712 /**
713 * Test that shutdown does not interrupt current thread.
714 */
715 @ParameterizedTest
716 @MethodSource("factories")
717 void testShutdownInterruptsThreads2(ThreadFactory factory) throws Exception {
718 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
719 var interrupted = new AtomicBoolean();
720 var latch = new CountDownLatch(1);
721 var subtask = scope.fork(() -> {
722 try {
723 scope.shutdown();
724 interrupted.set(Thread.currentThread().isInterrupted());
725 } finally {
726 latch.countDown();
727 }
728 return null;
729 });
730
731 // wait for task to complete
732 latch.await();
733 assertFalse(interrupted.get());
734
735 scope.join();
736 }
737 }
738
739 /**
740 * Test shutdown wakes join.
741 */
742 @ParameterizedTest
743 @MethodSource("factories")
744 void testShutdownWakesJoin(ThreadFactory factory) throws Exception {
745 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
746 var latch = new CountDownLatch(1);
747 scope.fork(() -> {
748 Thread.sleep(Duration.ofMillis(100)); // give time for join to block
749 scope.shutdown();
750 latch.await();
751 return null;
752 });
753
754 scope.join();
755
756 // join woke up, allow task to complete
757 latch.countDown();
758 }
759 }
760
761 /**
762 * Test shutdown after scope is closed.
763 */
764 @Test
765 void testShutdownAfterClose() throws Exception {
766 try (var scope = new StructuredTaskScope<Object>()) {
767 scope.join();
768 scope.close();
769 assertThrows(IllegalStateException.class, scope::shutdown);
770 }
771 }
772
773 /**
774 * Test shutdown is confined to threads in the scope "tree".
775 */
776 @ParameterizedTest
777 @MethodSource("factories")
778 void testShutdownConfined(ThreadFactory factory) throws Exception {
779 try (var scope1 = new StructuredTaskScope<Boolean>();
780 var scope2 = new StructuredTaskScope<Boolean>()) {
781
782 // thread in scope1 cannot shutdown scope2
783 Subtask<Boolean> subtask1 = scope1.fork(() -> {
784 assertThrows(WrongThreadException.class, scope2::shutdown);
785 return true;
786 });
787
788 // wait for task in scope1 to complete to avoid racing with task in scope2
789 while (subtask1.state() == Subtask.State.UNAVAILABLE) {
790 Thread.sleep(10);
791 }
792
793 // thread in scope2 shutdown scope1
794 Subtask<Boolean> subtask2 = scope2.fork(() -> {
795 scope1.shutdown();
796 return true;
797 });
798
799 scope2.join();
800 scope1.join();
801
802 assertTrue(subtask1.get());
803 assertTrue(subtask1.get());
804
805 // random thread cannot shutdown
806 try (var pool = Executors.newSingleThreadExecutor()) {
807 Future<Void> future = pool.submit(() -> {
808 assertThrows(WrongThreadException.class, scope1::shutdown);
809 assertThrows(WrongThreadException.class, scope2::shutdown);
810 return null;
811 });
812 future.get();
813 }
814 }
815 }
816
817 /**
818 * Test isShutdown.
819 */
820 @Test
821 void testIsShutdown() {
822 try (var scope = new StructuredTaskScope<Object>()) {
823 assertFalse(scope.isShutdown()); // before shutdown
824 scope.shutdown();
825 assertTrue(scope.isShutdown()); // after shutdown
826 scope.close();
827 assertTrue(scope.isShutdown()); // after cose
828 }
829 }
830
831 /**
832 * Test close without join, no subtasks forked.
833 */
834 @Test
835 void testCloseWithoutJoin1() {
836 try (var scope = new StructuredTaskScope<Object>()) {
837 // do nothing
838 }
839 }
840
841 /**
842 * Test close without join, unfinished subtasks.
843 */
844 @ParameterizedTest
845 @MethodSource("factories")
846 void testCloseWithoutJoin2(ThreadFactory factory) {
847 try (var scope = new StructuredTaskScope<String>(null, factory)) {
848 Subtask<String> subtask = scope.fork(() -> {
849 Thread.sleep(Duration.ofDays(1));
850 return null;
851 });
852 assertThrows(IllegalStateException.class, scope::close);
853
854 // subtask result/exception not available
855 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
856 assertThrows(IllegalStateException.class, subtask::get);
857 assertThrows(IllegalStateException.class, subtask::exception);
858 }
859 }
860
861 /**
862 * Test close without join, unfinished subtasks forked after join.
863 */
864 @ParameterizedTest
865 @MethodSource("factories")
866 void testCloseWithoutJoin3(ThreadFactory factory) throws Exception {
867 try (var scope = new StructuredTaskScope(null, factory)) {
868 scope.fork(() -> "foo");
869 scope.join();
870
871 Subtask<String> subtask = scope.fork(() -> {
872 Thread.sleep(Duration.ofDays(1));
873 return null;
874 });
875 assertThrows(IllegalStateException.class, scope::close);
876
877 // subtask result/exception not available
878 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
879 assertThrows(IllegalStateException.class, subtask::get);
880 assertThrows(IllegalStateException.class, subtask::exception);
881 }
882 }
883
884 /**
885 * Test close after join throws. Close should not throw as join attempted.
886 */
887 @ParameterizedTest
888 @MethodSource("factories")
889 void testCloseAfterJoinThrows(ThreadFactory factory) throws Exception {
890 try (var scope = new StructuredTaskScope<Object>()) {
891 var subtask = scope.fork(() -> {
892 Thread.sleep(Duration.ofDays(1));
893 return null;
894 });
895
896 // join throws
897 Thread.currentThread().interrupt();
898 assertThrows(InterruptedException.class, scope::join);
899 assertThrows(IllegalStateException.class, subtask::get);
900 }
901 }
902
903 /**
904 * Test close after joinUntil throws. Close should not throw as join attempted.
905 */
906 @ParameterizedTest
907 @MethodSource("factories")
908 void testCloseAfterJoinUntilThrows(ThreadFactory factory) throws Exception {
909 try (var scope = new StructuredTaskScope<Object>()) {
910 var subtask = scope.fork(() -> {
911 Thread.sleep(Duration.ofDays(1));
912 return null;
913 });
914
915 // joinUntil throws
916 assertThrows(TimeoutException.class, () -> scope.joinUntil(Instant.now()));
917 assertThrows(IllegalStateException.class, subtask::get);
918 }
919 }
920
921 /**
922 * Test close is owner confined.
923 */
924 @ParameterizedTest
925 @MethodSource("factories")
926 void testCloseConfined(ThreadFactory factory) throws Exception {
927 try (var scope = new StructuredTaskScope<Boolean>()) {
928
929 // attempt to close from thread in scope
930 Subtask<Boolean> subtask = scope.fork(() -> {
931 assertThrows(WrongThreadException.class, scope::close);
932 return true;
933 });
934
935 scope.join();
936 assertTrue(subtask.get());
937
938 // random thread cannot close scope
939 try (var pool = Executors.newCachedThreadPool(factory)) {
940 Future<Boolean> future = pool.submit(() -> {
941 assertThrows(WrongThreadException.class, scope::close);
942 return null;
943 });
944 future.get();
945 }
946 }
947 }
948
949 /**
950 * Test close with interrupt status set.
951 */
952 @ParameterizedTest
953 @MethodSource("factories")
954 void testInterruptClose1(ThreadFactory factory) throws Exception {
955 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
956 var done = new AtomicBoolean();
957 scope.fork(() -> {
958 try {
959 Thread.sleep(Duration.ofDays(1));
960 } catch (InterruptedException e) {
961 // interrupted by shutdown, expected
962 }
963 Thread.sleep(Duration.ofMillis(100)); // force close to wait
964 done.set(true);
965 return null;
966 });
967
968 scope.shutdown();
969 scope.join();
970
971 // invoke close with interrupt status set
972 Thread.currentThread().interrupt();
973 try {
974 scope.close();
975 } finally {
976 assertTrue(Thread.interrupted()); // clear interrupt status
977 assertTrue(done.get());
978 }
979 }
980 }
981
982 /**
983 * Test interrupting thread waiting in close.
984 */
985 @ParameterizedTest
986 @MethodSource("factories")
987 void testInterruptClose2(ThreadFactory factory) throws Exception {
988 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
989 var done = new AtomicBoolean();
990 Thread mainThread = Thread.currentThread();
991 scope.fork(() -> {
992 try {
993 Thread.sleep(Duration.ofDays(1));
994 } catch (InterruptedException e) {
995 // interrupted by shutdown, expected
996 }
997
998 // interrupt main thread when it blocks in close
999 interruptThreadAt(mainThread, "java.util.concurrent.StructuredTaskScope.close");
1000
1001 Thread.sleep(Duration.ofMillis(100)); // force close to wait
1002 done.set(true);
1003 return null;
1004 });
1005
1006 scope.shutdown(); // interrupts task
1007 scope.join();
1008 try {
1009 scope.close();
1010 } finally {
1011 assertTrue(Thread.interrupted()); // clear interrupt status
1012 assertTrue(done.get());
1013 }
1014 }
1015 }
1016
1017 /**
1018 * Test that closing an enclosing scope closes the thread flock of a nested scope.
1019 */
1020 @Test
1021 void testCloseThrowsStructureViolation() throws Exception {
1022 try (var scope1 = new StructuredTaskScope<Object>()) {
1023 try (var scope2 = new StructuredTaskScope<Object>()) {
1024
1025 // join + close enclosing scope
1026 scope1.join();
1027 try {
1028 scope1.close();
1029 fail("close did not throw");
1030 } catch (StructureViolationException expected) { }
1031
1032 // underlying flock should be closed, fork should return a cancelled task
1033 var executed = new AtomicBoolean();
1034 Subtask<Void> subtask = scope2.fork(() -> {
1035 executed.set(true);
1036 return null;
1037 });
1038 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1039 scope2.join();
1040 assertFalse(executed.get());
1041 }
1042 }
1043 }
1044
1045 /**
1046 * A StructuredTaskScope that collects the subtasks notified to the handleComplete method.
1047 */
1048 private static class CollectAll<T> extends StructuredTaskScope<T> {
1049 private final Set<Subtask<? extends T>> subtasks = ConcurrentHashMap.newKeySet();
1050
1051 CollectAll(ThreadFactory factory) {
1052 super(null, factory);
1053 }
1054
1055 @Override
1056 protected void handleComplete(Subtask<? extends T> subtask) {
1057 subtasks.add(subtask);
1058 }
1059
1060 Set<Subtask<? extends T>> subtasks() {
1061 return subtasks;
1062 }
1063
1064 Subtask<? extends T> find(Callable<T> task) {
1065 return subtasks.stream()
1066 .filter(h -> task.equals(h.task()))
1067 .findAny()
1068 .orElseThrow();
1069 }
1070 }
1071
1072 /**
1073 * Test that handleComplete method is invoked for tasks that complete before shutdown.
1074 */
1075 @ParameterizedTest
1076 @MethodSource("factories")
1077 void testHandleCompleteBeforeShutdown(ThreadFactory factory) throws Exception {
1078 try (var scope = new CollectAll<String>(factory)) {
1079 Callable<String> task1 = () -> "foo";
1080 Callable<String> task2 = () -> { throw new FooException(); };
1081 scope.fork(task1);
1082 scope.fork(task2);
1083 scope.join();
1084
1085 var subtask1 = scope.find(task1);
1086 assertEquals("foo", subtask1.get());
1087
1088 var subtask2 = scope.find(task2);
1089 assertTrue(subtask2.exception() instanceof FooException);
1090 }
1091 }
1092
1093 /**
1094 * Test that handleComplete method is not invoked for tasks that finish after shutdown
1095 * or are forked after shutdown.
1096 */
1097 @ParameterizedTest
1098 @MethodSource("factories")
1099 void testHandleCompleteAfterShutdown(ThreadFactory factory) throws Exception {
1100 try (var scope = new CollectAll<String>(factory)) {
1101 Callable<String> task1 = () -> {
1102 try {
1103 Thread.sleep(Duration.ofDays(1));
1104 } catch (InterruptedException ignore) { }
1105 return "foo";
1106 };
1107 Callable<String> task2 = () -> {
1108 Thread.sleep(Duration.ofDays(1));
1109 return "bar";
1110 };
1111 Callable<String> task3 = () -> "baz";
1112
1113 // forked before shutdown, will complete after shutdown
1114 var subtask1 = scope.fork(task1);
1115 var subtask2 = scope.fork(task2);
1116
1117 scope.shutdown();
1118
1119 // forked after shutdown
1120 var subtask3 = scope.fork(task3);
1121
1122 scope.join();
1123
1124 // handleComplete should not be called
1125 for (int i = 0; i < 3; i++) {
1126 assertEquals(0, scope.subtasks().size());
1127 Thread.sleep(20);
1128 }
1129
1130 assertEquals(Subtask.State.UNAVAILABLE, subtask1.state());
1131 assertEquals(Subtask.State.UNAVAILABLE, subtask2.state());
1132 assertEquals(Subtask.State.UNAVAILABLE, subtask3.state());
1133 }
1134 }
1135
1136 /**
1137 * Test that the default handleComplete throws IllegalArgumentException if called
1138 * with a running task.
1139 */
1140 @Test
1141 void testHandleCompleteThrows() throws Exception {
1142 class TestScope<T> extends StructuredTaskScope<T> {
1143 protected void handleComplete(Subtask<? extends T> subtask) {
1144 super.handleComplete(subtask);
1145 }
1146 }
1147
1148 try (var scope = new TestScope<String>()) {
1149 var subtask = scope.fork(() -> {
1150 Thread.sleep(Duration.ofDays(1));
1151 return "foo";
1152 });
1153
1154 // running task
1155 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1156 assertThrows(IllegalArgumentException.class, () -> scope.handleComplete(subtask));
1157 scope.shutdown();
1158
1159 // null task
1160 assertThrows(NullPointerException.class, () -> scope.handleComplete(null));
1161
1162 scope.join();
1163 }
1164 }
1165
1166 /**
1167 * Test ensureOwnerAndJoined.
1168 */
1169 @ParameterizedTest
1170 @MethodSource("factories")
1171 void testEnsureOwnerAndJoined(ThreadFactory factory) throws Exception {
1172 class MyScope<T> extends StructuredTaskScope<T> {
1173 MyScope(ThreadFactory factory) {
1174 super(null, factory);
1175 }
1176 void invokeEnsureOwnerAndJoined() {
1177 super.ensureOwnerAndJoined();
1178 }
1179 }
1180
1181 try (var scope = new MyScope<Boolean>(factory)) {
1182 // owner thread, before join
1183 scope.fork(() -> true);
1184 assertThrows(IllegalStateException.class, () -> {
1185 scope.invokeEnsureOwnerAndJoined();
1186 });
1187
1188 // owner thread, after join
1189 scope.join();
1190 scope.invokeEnsureOwnerAndJoined();
1191
1192 // thread in scope cannot invoke ensureOwnerAndJoined
1193 Subtask<Boolean> subtask = scope.fork(() -> {
1194 assertThrows(WrongThreadException.class, () -> {
1195 scope.invokeEnsureOwnerAndJoined();
1196 });
1197 return true;
1198 });
1199 scope.join();
1200 assertTrue(subtask.get());
1201
1202 // random thread cannot invoke ensureOwnerAndJoined
1203 try (var pool = Executors.newSingleThreadExecutor()) {
1204 Future<Void> future = pool.submit(() -> {
1205 assertThrows(WrongThreadException.class, () -> {
1206 scope.invokeEnsureOwnerAndJoined();
1207 });
1208 return null;
1209 });
1210 future.get();
1211 }
1212 }
1213 }
1214
1215 /**
1216 * Test ensureOwnerAndJoined after the task scope has been closed.
1217 */
1218 @ParameterizedTest
1219 @MethodSource("factories")
1220 void testEnsureOwnerAndJoinedAfterClose(ThreadFactory factory) throws Exception {
1221 class MyScope<T> extends StructuredTaskScope<T> {
1222 MyScope(ThreadFactory factory) {
1223 super(null, factory);
1224 }
1225 public void invokeEnsureOwnerAndJoined() {
1226 super.ensureOwnerAndJoined();
1227 }
1228 }
1229
1230 // ensureOwnerAndJoined after close, join invoked
1231 try (var scope = new MyScope<String>(factory)) {
1232 scope.fork(() -> "foo");
1233 scope.join();
1234 scope.close();
1235 scope.invokeEnsureOwnerAndJoined(); // should not throw
1236 }
1237
1238 // ensureOwnerAndJoined after close, join not invoked
1239 try (var scope = new MyScope<String>(factory)) {
1240 scope.fork(() -> "foo");
1241 assertThrows(IllegalStateException.class, scope::close);
1242 scope.invokeEnsureOwnerAndJoined(); // should not throw
1243 }
1244 }
1245
1246
1247 /**
1248 * Test toString.
1249 */
1250 @Test
1251 void testToString() throws Exception {
1252 ThreadFactory factory = Thread.ofVirtual().factory();
1253 try (var scope = new StructuredTaskScope<Object>("duke", factory)) {
1254 // open
1255 assertTrue(scope.toString().contains("duke"));
1256
1257 // shutdown
1258 scope.shutdown();
1259 assertTrue(scope.toString().contains("duke"));
1260
1261 // closed
1262 scope.join();
1263 scope.close();
1264 assertTrue(scope.toString().contains("duke"));
1265 }
1266 }
1267
1268 /**
1269 * Test Subtask with task that completes successfully.
1270 */
1271 @ParameterizedTest
1272 @MethodSource("factories")
1273 void testSubtaskWhenSuccess(ThreadFactory factory) throws Exception {
1274 try (var scope = new StructuredTaskScope<String>(null, factory)) {
1275 Callable<String> task = () -> "foo";
1276 Subtask<String> subtask = scope.fork(task);
1277
1278 // before join, owner thread
1279 assertEquals(task, subtask.task());
1280 assertThrows(IllegalStateException.class, subtask::get);
1281 assertThrows(IllegalStateException.class, subtask::exception);
1282
1283 scope.join();
1284
1285 // after join
1286 assertEquals(task, subtask.task());
1287 assertEquals(Subtask.State.SUCCESS, subtask.state());
1288 assertEquals("foo", subtask.get());
1289 assertThrows(IllegalStateException.class, subtask::exception);
1290 }
1291 }
1292
1293 /**
1294 * Test Subtask with task that fails.
1295 */
1296 @ParameterizedTest
1297 @MethodSource("factories")
1298 void testSubtaskWhenFailed(ThreadFactory factory) throws Exception {
1299 try (var scope = new StructuredTaskScope<String>(null, factory)) {
1300 Callable<String> task = () -> { throw new FooException(); };
1301 Subtask<String> subtask = scope.fork(task);
1302
1303 // before join, owner thread
1304 assertEquals(task, subtask.task());
1305 assertThrows(IllegalStateException.class, subtask::get);
1306 assertThrows(IllegalStateException.class, subtask::exception);
1307
1308 scope.join();
1309
1310 // after join
1311 assertEquals(task, subtask.task());
1312 assertEquals(Subtask.State.FAILED, subtask.state());
1313 assertThrows(IllegalStateException.class, subtask::get);
1314 assertTrue(subtask.exception() instanceof FooException);
1315 }
1316 }
1317
1318 /**
1319 * Test Subtask with a task that has not completed.
1320 */
1321 @ParameterizedTest
1322 @MethodSource("factories")
1323 void testSubtaskWhenNotCompleted(ThreadFactory factory) throws Exception {
1324 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
1325 Callable<Void> task = () -> {
1326 Thread.sleep(Duration.ofDays(1));
1327 return null;
1328 };
1329 Subtask<Void> subtask = scope.fork(task);
1330
1331 // before join
1332 assertEquals(task, subtask.task());
1333 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1334 assertThrows(IllegalStateException.class, subtask::get);
1335 assertThrows(IllegalStateException.class, subtask::exception);
1336
1337 // attempt join, join throws
1338 Thread.currentThread().interrupt();
1339 assertThrows(InterruptedException.class, scope::join);
1340
1341 // after join
1342 assertEquals(task, subtask.task());
1343 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1344 assertThrows(IllegalStateException.class, subtask::get);
1345 assertThrows(IllegalStateException.class, subtask::exception);
1346 }
1347 }
1348
1349 /**
1350 * Test Subtask when forked after shutdown.
1351 */
1352 @ParameterizedTest
1353 @MethodSource("factories")
1354 void testSubtaskWhenShutdown(ThreadFactory factory) throws Exception {
1355 try (var scope = new StructuredTaskScope<Object>(null, factory)) {
1356 Callable<Void> task = () -> {
1357 Thread.sleep(Duration.ofDays(1));
1358 return null;
1359 };
1360
1361 scope.shutdown();
1362
1363 // fork after shutdown
1364 Subtask<Void> subtask = scope.fork(task);
1365 scope.join();
1366 assertEquals(task, subtask.task());
1367 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1368 assertThrows(IllegalStateException.class, subtask::get);
1369 assertThrows(IllegalStateException.class, subtask::exception);
1370 }
1371 }
1372
1373 /**
1374 * Test Subtask::toString.
1375 */
1376 @Test
1377 void testSubtaskToString() throws Exception {
1378 try (var scope = new StructuredTaskScope<Object>()) {
1379 // success
1380 var subtask1 = scope.fork(() -> "foo");
1381 scope.join();
1382 assertTrue(subtask1.toString().contains("Completed successfully"));
1383
1384 // failed
1385 var subtask2 = scope.fork(() -> { throw new FooException(); });
1386 scope.join();
1387 assertTrue(subtask2.toString().contains("Failed"));
1388
1389 // not completed
1390 Callable<Void> sleepForDay = () -> {
1391 Thread.sleep(Duration.ofDays(1));
1392 return null;
1393 };
1394 var subtask3 = scope.fork(sleepForDay);
1395 assertTrue(subtask3.toString().contains("Unavailable"));
1396
1397 scope.shutdown();
1398
1399 // forked after shutdown
1400 var subtask4 = scope.fork(sleepForDay);
1401 assertTrue(subtask4.toString().contains("Unavailable"));
1402
1403 scope.join();
1404 }
1405 }
1406
1407 /**
1408 * Test ShutdownOnSuccess with no completed tasks.
1409 */
1410 @Test
1411 void testShutdownOnSuccess1() throws Exception {
1412 try (var scope = new ShutdownOnSuccess<Object>()) {
1413 assertThrows(IllegalStateException.class, () -> scope.result());
1414 assertThrows(IllegalStateException.class, () -> scope.result(e -> null));
1415 }
1416 }
1417
1418 /**
1419 * Test ShutdownOnSuccess with tasks that complete successfully.
1420 */
1421 @ParameterizedTest
1422 @MethodSource("factories")
1423 void testShutdownOnSuccess2(ThreadFactory factory) throws Exception {
1424 try (var scope = new ShutdownOnSuccess<String>(null, factory)) {
1425 scope.fork(() -> "foo");
1426 scope.join(); // ensures foo completes first
1427 scope.fork(() -> "bar");
1428 scope.join();
1429 assertEquals("foo", scope.result());
1430 assertEquals("foo", scope.result(e -> null));
1431 }
1432 }
1433
1434 /**
1435 * Test ShutdownOnSuccess with a task that completes successfully with a null result.
1436 */
1437 @ParameterizedTest
1438 @MethodSource("factories")
1439 void testShutdownOnSuccess3(ThreadFactory factory) throws Exception {
1440 try (var scope = new ShutdownOnSuccess<Object>(null, factory)) {
1441 scope.fork(() -> null);
1442 scope.join();
1443 assertNull(scope.result());
1444 assertNull(scope.result(e -> null));
1445 }
1446 }
1447
1448 /**
1449 * Test ShutdownOnSuccess with tasks that complete succcessfully and tasks that fail.
1450 */
1451 @ParameterizedTest
1452 @MethodSource("factories")
1453 void testShutdownOnSuccess4(ThreadFactory factory) throws Exception {
1454 try (var scope = new ShutdownOnSuccess<String>(null, factory)) {
1455 scope.fork(() -> "foo");
1456 scope.fork(() -> { throw new ArithmeticException(); });
1457 scope.join();
1458 assertEquals("foo", scope.result());
1459 assertEquals("foo", scope.result(e -> null));
1460 }
1461 }
1462
1463 /**
1464 * Test ShutdownOnSuccess with a task that fails.
1465 */
1466 @ParameterizedTest
1467 @MethodSource("factories")
1468 void testShutdownOnSuccess5(ThreadFactory factory) throws Exception {
1469 try (var scope = new ShutdownOnSuccess<Object>(null, factory)) {
1470 scope.fork(() -> { throw new ArithmeticException(); });
1471 scope.join();
1472 Throwable ex = assertThrows(ExecutionException.class, () -> scope.result());
1473 assertTrue(ex.getCause() instanceof ArithmeticException);
1474 ex = assertThrows(FooException.class, () -> scope.result(e -> new FooException(e)));
1475 assertTrue(ex.getCause() instanceof ArithmeticException);
1476 }
1477 }
1478
1479 /**
1480 * Test ShutdownOnSuccess methods are confined to the owner.
1481 */
1482 @ParameterizedTest
1483 @MethodSource("factories")
1484 void testShutdownOnSuccessConfined(ThreadFactory factory) throws Exception {
1485 // owner before join
1486 try (var scope = new ShutdownOnSuccess<Boolean>(null, factory)) {
1487 scope.fork(() -> { throw new FooException(); });
1488 assertThrows(IllegalStateException.class, scope::result);
1489 assertThrows(IllegalStateException.class, () -> {
1490 scope.result(e -> new RuntimeException(e));
1491 });
1492 scope.join();
1493 }
1494
1495 // non-owner
1496 try (var scope = new ShutdownOnSuccess<Boolean>(null, factory)) {
1497 Subtask<Boolean> subtask = scope.fork(() -> {
1498 assertThrows(WrongThreadException.class, scope::result);
1499 assertThrows(WrongThreadException.class, () -> {
1500 scope.result(e -> new RuntimeException(e));
1501 });
1502 return true;
1503 });
1504 scope.join();
1505 assertTrue(subtask.get());
1506 }
1507 }
1508
1509 /**
1510 * Test ShutdownOnFailure with no completed tasks.
1511 */
1512 @Test
1513 void testShutdownOnFailure1() throws Throwable {
1514 try (var scope = new ShutdownOnFailure()) {
1515 assertTrue(scope.exception().isEmpty());
1516 scope.throwIfFailed();
1517 scope.throwIfFailed(e -> new FooException(e));
1518 }
1519 }
1520
1521 /**
1522 * Test ShutdownOnFailure with tasks that complete successfully.
1523 */
1524 @ParameterizedTest
1525 @MethodSource("factories")
1526 void testShutdownOnFailure2(ThreadFactory factory) throws Throwable {
1527 try (var scope = new ShutdownOnFailure(null, factory)) {
1528 scope.fork(() -> "foo");
1529 scope.fork(() -> "bar");
1530 scope.join();
1531
1532 // no exception
1533 assertTrue(scope.exception().isEmpty());
1534 scope.throwIfFailed();
1535 scope.throwIfFailed(e -> new FooException(e));
1536 }
1537 }
1538
1539 /**
1540 * Test ShutdownOnFailure with tasks that complete succcessfully and tasks that fail.
1541 */
1542 @ParameterizedTest
1543 @MethodSource("factories")
1544 void testShutdownOnFailure3(ThreadFactory factory) throws Throwable {
1545 try (var scope = new ShutdownOnFailure(null, factory)) {
1546
1547 // one task completes successfully, the other fails
1548 scope.fork(() -> "foo");
1549 scope.fork(() -> { throw new ArithmeticException(); });
1550 scope.join();
1551
1552 Throwable ex = scope.exception().orElse(null);
1553 assertTrue(ex instanceof ArithmeticException);
1554
1555 ex = assertThrows(ExecutionException.class, () -> scope.throwIfFailed());
1556 assertTrue(ex.getCause() instanceof ArithmeticException);
1557
1558 ex = assertThrows(FooException.class,
1559 () -> scope.throwIfFailed(e -> new FooException(e)));
1560 assertTrue(ex.getCause() instanceof ArithmeticException);
1561 }
1562 }
1563
1564 /**
1565 * Test ShutdownOnFailure methods are confined to the owner.
1566 */
1567 @ParameterizedTest
1568 @MethodSource("factories")
1569 void testShutdownOnFailureConfined(ThreadFactory factory) throws Exception {
1570 // owner before join
1571 try (var scope = new ShutdownOnFailure(null, factory)) {
1572 scope.fork(() -> "foo");
1573 assertThrows(IllegalStateException.class, scope::exception);
1574 assertThrows(IllegalStateException.class, scope::throwIfFailed);
1575 assertThrows(IllegalStateException.class, () -> {
1576 scope.throwIfFailed(e -> new RuntimeException(e));
1577 });
1578 scope.join();
1579 }
1580
1581 // non-owner
1582 try (var scope = new ShutdownOnFailure(null, factory)) {
1583 Subtask<Boolean> subtask = scope.fork(() -> {
1584 assertThrows(WrongThreadException.class, scope::exception);
1585 assertThrows(WrongThreadException.class, scope::throwIfFailed);
1586 assertThrows(WrongThreadException.class, () -> {
1587 scope.throwIfFailed(e -> new RuntimeException(e));
1588 });
1589 return true;
1590 });
1591 scope.join();
1592 assertTrue(subtask.get());
1593 }
1594 }
1595
1596 /**
1597 * Test for NullPointerException.
1598 */
1599 @Test
1600 void testNulls() throws Exception {
1601 assertThrows(NullPointerException.class, () -> new StructuredTaskScope("", null));
1602 try (var scope = new StructuredTaskScope<Object>()) {
1603 assertThrows(NullPointerException.class, () -> scope.fork(null));
1604 assertThrows(NullPointerException.class, () -> scope.joinUntil(null));
1605 }
1606
1607 assertThrows(NullPointerException.class, () -> new ShutdownOnSuccess<Object>("", null));
1608 try (var scope = new ShutdownOnSuccess<Object>()) {
1609 assertThrows(NullPointerException.class, () -> scope.fork(null));
1610 assertThrows(NullPointerException.class, () -> scope.joinUntil(null));
1611 assertThrows(NullPointerException.class, () -> scope.result(null));
1612 }
1613
1614 assertThrows(NullPointerException.class, () -> new ShutdownOnFailure("", null));
1615 try (var scope = new ShutdownOnFailure()) {
1616 assertThrows(NullPointerException.class, () -> scope.fork(null));
1617 assertThrows(NullPointerException.class, () -> scope.joinUntil(null));
1618 assertThrows(NullPointerException.class, () -> scope.throwIfFailed(null));
1619 }
1620 }
1621
1622 /**
1623 * A runtime exception for tests.
1624 */
1625 private static class FooException extends RuntimeException {
1626 FooException() { }
1627 FooException(Throwable cause) { super(cause); }
1628 }
1629
1630 /**
1631 * Returns the current time in milliseconds.
1632 */
1633 private long millisTime() {
1634 long now = System.nanoTime();
1635 return TimeUnit.MILLISECONDS.convert(now, TimeUnit.NANOSECONDS);
1636 }
1637
1638 /**
1639 * Check the duration of a task
1640 * @param start start time, in milliseconds
1641 * @param min minimum expected duration, in milliseconds
1642 * @param max maximum expected duration, in milliseconds
1643 * @return the duration (now - start), in milliseconds
1644 */
1645 private long expectDuration(long start, long min, long max) {
1646 long duration = millisTime() - start;
1647 assertTrue(duration >= min,
1648 "Duration " + duration + "ms, expected >= " + min + "ms");
1649 assertTrue(duration <= max,
1650 "Duration " + duration + "ms, expected <= " + max + "ms");
1651 return duration;
1652 }
1653
1654 /**
1655 * Interrupts a thread when it waits (timed or untimed) at location "{@code c.m}".
1656 * {@code c} is the fully qualified class name and {@code m} is the method name.
1657 */
1658 private void interruptThreadAt(Thread target, String location) throws InterruptedException {
1659 int index = location.lastIndexOf('.');
1660 String className = location.substring(0, index);
1661 String methodName = location.substring(index + 1);
1662
1663 boolean found = false;
1664 while (!found) {
1665 Thread.State state = target.getState();
1666 assertTrue(state != TERMINATED);
1667 if ((state == WAITING || state == TIMED_WAITING)
1668 && contains(target.getStackTrace(), className, methodName)) {
1669 found = true;
1670 } else {
1671 Thread.sleep(20);
1672 }
1673 }
|
1 /*
2 * Copyright (c) 2021, 2025, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation.
8 *
9 * This code is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
12 * version 2 for more details (a copy is included in the LICENSE file that
13 * accompanied this code).
14 *
15 * You should have received a copy of the GNU General Public License version
16 * 2 along with this work; if not, write to the Free Software Foundation,
17 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18 *
19 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
20 * or visit www.oracle.com if you need additional information or have any
21 * questions.
22 */
23
24 /*
25 * @test id=platform
26 * @bug 8284199 8296779 8306647
27 * @summary Basic tests for StructuredTaskScope
28 * @enablePreview
29 * @run junit/othervm -DthreadFactory=platform StructuredTaskScopeTest
30 */
31
32 /*
33 * @test id=virtual
34 * @enablePreview
35 * @run junit/othervm -DthreadFactory=virtual StructuredTaskScopeTest
36 */
37
38 import java.time.Duration;
39 import java.util.Arrays;
40 import java.util.ArrayList;
41 import java.util.List;
42 import java.util.Set;
43 import java.util.NoSuchElementException;
44 import java.util.concurrent.Callable;
45 import java.util.concurrent.ConcurrentHashMap;
46 import java.util.concurrent.CountDownLatch;
47 import java.util.concurrent.Executors;
48 import java.util.concurrent.Future;
49 import java.util.concurrent.LinkedTransferQueue;
50 import java.util.concurrent.ThreadFactory;
51 import java.util.concurrent.TimeUnit;
52 import java.util.concurrent.RejectedExecutionException;
53 import java.util.concurrent.ScheduledExecutorService;
54 import java.util.concurrent.StructuredTaskScope;
55 import java.util.concurrent.StructuredTaskScope.TimeoutException;
56 import java.util.concurrent.StructuredTaskScope.Configuration;
57 import java.util.concurrent.StructuredTaskScope.FailedException;
58 import java.util.concurrent.StructuredTaskScope.Joiner;
59 import java.util.concurrent.StructuredTaskScope.Subtask;
60 import java.util.concurrent.StructureViolationException;
61 import java.util.concurrent.atomic.AtomicBoolean;
62 import java.util.concurrent.atomic.AtomicInteger;
63 import java.util.concurrent.atomic.AtomicReference;
64 import java.util.function.Function;
65 import java.util.function.Predicate;
66 import java.util.stream.Stream;
67 import static java.lang.Thread.State.*;
68
69 import org.junit.jupiter.api.Test;
70 import org.junit.jupiter.api.BeforeAll;
71 import org.junit.jupiter.api.AfterAll;
72 import org.junit.jupiter.params.ParameterizedTest;
73 import org.junit.jupiter.params.provider.MethodSource;
74 import static org.junit.jupiter.api.Assertions.*;
75
76 class StructuredTaskScopeTest {
77 private static ScheduledExecutorService scheduler;
78 private static List<ThreadFactory> threadFactories;
79
80 @BeforeAll
81 static void setup() throws Exception {
82 scheduler = Executors.newSingleThreadScheduledExecutor();
83
84 // thread factories
85 String value = System.getProperty("threadFactory");
86 List<ThreadFactory> list = new ArrayList<>();
87 if (value == null || value.equals("platform"))
88 list.add(Thread.ofPlatform().factory());
89 if (value == null || value.equals("virtual"))
90 list.add(Thread.ofVirtual().factory());
91 assertTrue(list.size() > 0, "No thread factories for tests");
92 threadFactories = list;
93 }
94
95 @AfterAll
96 static void shutdown() {
97 scheduler.shutdown();
98 }
99
100 private static Stream<ThreadFactory> factories() {
101 return threadFactories.stream();
102 }
103
104 /**
105 * Test that fork creates virtual threads when no ThreadFactory is configured.
106 */
107 @Test
108 void testForkCreatesVirtualThread() throws Exception {
109 Set<Thread> threads = ConcurrentHashMap.newKeySet();
110 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
111 for (int i = 0; i < 50; i++) {
112 // runnable
113 scope.fork(() -> {
114 threads.add(Thread.currentThread());
115 });
116
117 // callable
118 scope.fork(() -> {
119 threads.add(Thread.currentThread());
120 return null;
121 });
122 }
123 scope.join();
124 }
125 assertEquals(100, threads.size());
126 threads.forEach(t -> assertTrue(t.isVirtual()));
127 }
128
129 /**
130 * Test that fork create threads with the configured ThreadFactory.
131 */
132 @ParameterizedTest
133 @MethodSource("factories")
134 void testForkUsesThreadFactory(ThreadFactory factory) throws Exception {
135 // TheadFactory that keeps reference to all threads it creates
136 class RecordingThreadFactory implements ThreadFactory {
137 final ThreadFactory delegate;
138 final Set<Thread> threads = ConcurrentHashMap.newKeySet();
139 RecordingThreadFactory(ThreadFactory delegate) {
140 this.delegate = delegate;
141 }
142 @Override
143 public Thread newThread(Runnable task) {
144 Thread thread = delegate.newThread(task);
145 threads.add(thread);
146 return thread;
147 }
148 Set<Thread> threads() {
149 return threads;
150 }
151 }
152 var recordingThreadFactory = new RecordingThreadFactory(factory);
153 Set<Thread> threads = ConcurrentHashMap.newKeySet();
154 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
155 cf -> cf.withThreadFactory(recordingThreadFactory))) {
156
157 for (int i = 0; i < 50; i++) {
158 // runnable
159 scope.fork(() -> {
160 threads.add(Thread.currentThread());
161 });
162
163 // callable
164 scope.fork(() -> {
165 threads.add(Thread.currentThread());
166 return null;
167 });
168 }
169 scope.join();
170 }
171 assertEquals(100, threads.size());
172 assertEquals(recordingThreadFactory.threads(), threads);
173 }
174
175 /**
176 * Test fork method is owner confined.
177 */
178 @ParameterizedTest
179 @MethodSource("factories")
180 void testForkConfined(ThreadFactory factory) throws Exception {
181 try (var scope = StructuredTaskScope.open(Joiner.<Boolean>awaitAll(),
182 cf -> cf.withThreadFactory(factory))) {
183
184 // random thread cannot fork
185 try (var pool = Executors.newSingleThreadExecutor()) {
186 Future<Void> future = pool.submit(() -> {
187 assertThrows(WrongThreadException.class, () -> {
188 scope.fork(() -> null);
189 });
190 return null;
191 });
192 future.get();
193 }
194
195 // subtask cannot fork
196 Subtask<Boolean> subtask = scope.fork(() -> {
197 assertThrows(WrongThreadException.class, () -> {
198 scope.fork(() -> null);
199 });
200 return true;
201 });
202 scope.join();
203 assertTrue(subtask.get());
204 }
205 }
206
207 /**
208 * Test fork after join, no subtasks forked before join.
209 */
210 @ParameterizedTest
211 @MethodSource("factories")
212 void testForkAfterJoin1(ThreadFactory factory) throws Exception {
213 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
214 cf -> cf.withThreadFactory(factory))) {
215 scope.join();
216 assertThrows(IllegalStateException.class, () -> scope.fork(() -> "bar"));
217 }
218 }
219
220 /**
221 * Test fork after join, subtasks forked before join.
222 */
223 @ParameterizedTest
224 @MethodSource("factories")
225 void testForkAfterJoin2(ThreadFactory factory) throws Exception {
226 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
227 cf -> cf.withThreadFactory(factory))) {
228 scope.fork(() -> "foo");
229 scope.join();
230 assertThrows(IllegalStateException.class, () -> scope.fork(() -> "bar"));
231 }
232 }
233
234 /**
235 * Test fork after join throws.
236 */
237 @ParameterizedTest
238 @MethodSource("factories")
239 void testForkAfterJoinThrows(ThreadFactory factory) throws Exception {
240 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
241 cf -> cf.withThreadFactory(factory))) {
242 var latch = new CountDownLatch(1);
243 var subtask1 = scope.fork(() -> {
244 latch.await();
245 return "foo";
246 });
247
248 // join throws
249 Thread.currentThread().interrupt();
250 assertThrows(InterruptedException.class, scope::join);
251
252 // fork should throw
253 assertThrows(IllegalStateException.class, () -> scope.fork(() -> "bar"));
254 }
255 }
256
257 /**
258 * Test fork after task scope is cancelled. This test uses a custom Joiner to
259 * cancel execution.
260 */
261 @ParameterizedTest
262 @MethodSource("factories")
263 void testForkAfterCancel2(ThreadFactory factory) throws Exception {
264 var countingThreadFactory = new CountingThreadFactory(factory);
265 var testJoiner = new CancelAfterOneJoiner<String>();
266
267 try (var scope = StructuredTaskScope.open(testJoiner,
268 cf -> cf.withThreadFactory(countingThreadFactory))) {
269
270 // fork subtask, the scope should be cancelled when the subtask completes
271 var subtask1 = scope.fork(() -> "foo");
272 awaitCancelled(scope);
273
274 assertEquals(1, countingThreadFactory.threadCount());
275 assertEquals(1, testJoiner.onForkCount());
276 assertEquals(1, testJoiner.onCompleteCount());
277
278 // fork second subtask, it should not run
279 var subtask2 = scope.fork(() -> "bar");
280
281 // onFork should be invoked, newThread and onComplete should not be invoked
282 assertEquals(1, countingThreadFactory.threadCount());
283 assertEquals(2, testJoiner.onForkCount());
284 assertEquals(1, testJoiner.onCompleteCount());
285
286 scope.join();
287
288 assertEquals(1, countingThreadFactory.threadCount());
289 assertEquals(2, testJoiner.onForkCount());
290 assertEquals(1, testJoiner.onCompleteCount());
291 assertEquals("foo", subtask1.get());
292 assertEquals(Subtask.State.UNAVAILABLE, subtask2.state());
293 }
294 }
295
296 /**
297 * Test fork after task scope is closed.
298 */
299 @Test
300 void testForkAfterClose() {
301 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
302 scope.close();
303 assertThrows(IllegalStateException.class, () -> scope.fork(() -> null));
304 }
305 }
306
307 /**
308 * Test fork with a ThreadFactory that rejects creating a thread.
309 */
310 @Test
311 void testForkRejectedExecutionException() {
312 ThreadFactory factory = task -> null;
313 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
314 cf -> cf.withThreadFactory(factory))) {
315 assertThrows(RejectedExecutionException.class, () -> scope.fork(() -> null));
316 }
317 }
318
319 /**
320 * Test join with no subtasks.
321 */
322 @Test
323 void testJoinWithNoSubtasks() throws Exception {
324 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
325 scope.join();
326 }
327 }
328
329 /**
330 * Test join with a remaining subtask.
331 */
332 @ParameterizedTest
333 @MethodSource("factories")
334 void testJoinWithRemainingSubtasks(ThreadFactory factory) throws Exception {
335 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
336 cf -> cf.withThreadFactory(factory))) {
337 Subtask<String> subtask = scope.fork(() -> {
338 Thread.sleep(Duration.ofMillis(100));
339 return "foo";
340 });
341 scope.join();
342 assertEquals("foo", subtask.get());
343 }
344 }
345
346 /**
347 * Test join after join completed with a result.
348 */
349 @Test
350 void testJoinAfterJoin1() throws Exception {
351 var results = new LinkedTransferQueue<>(List.of("foo", "bar", "baz"));
352 Joiner<Object, String> joiner = results::take;
353 try (var scope = StructuredTaskScope.open(joiner)) {
354 scope.fork(() -> "foo");
355 assertEquals("foo", scope.join());
356
357 // join already called
358 for (int i = 0 ; i < 3; i++) {
359 assertThrows(IllegalStateException.class, scope::join);
360 }
361 }
362 }
363
364 /**
365 * Test join after join completed with an exception.
366 */
367 @Test
368 void testJoinAfterJoin2() throws Exception {
369 try (var scope = StructuredTaskScope.open(Joiner.anySuccessfulResultOrThrow())) {
370 scope.fork(() -> { throw new FooException(); });
371 Throwable ex = assertThrows(FailedException.class, scope::join);
372 assertTrue(ex.getCause() instanceof FooException);
373
374 // join already called
375 for (int i = 0 ; i < 3; i++) {
376 assertThrows(IllegalStateException.class, scope::join);
377 }
378 }
379 }
380
381 /**
382 * Test join after join completed with a timeout.
383 */
384 @Test
385 void testJoinAfterJoin3() throws Exception {
386 try (var scope = StructuredTaskScope.open(Joiner.anySuccessfulResultOrThrow(),
387 cf -> cf.withTimeout(Duration.ofMillis(100)))) {
388 // wait for scope to be cancelled by timeout
389 awaitCancelled(scope);
390 assertThrows(TimeoutException.class, scope::join);
391
392 // join already called
393 for (int i = 0 ; i < 3; i++) {
394 assertThrows(IllegalStateException.class, scope::join);
395 }
396 }
397 }
398
399 /**
400 * Test join method is owner confined.
401 */
402 @ParameterizedTest
403 @MethodSource("factories")
404 void testJoinConfined(ThreadFactory factory) throws Exception {
405 try (var scope = StructuredTaskScope.open(Joiner.<Boolean>awaitAll(),
406 cf -> cf.withThreadFactory(factory))) {
407
408 // random thread cannot join
409 try (var pool = Executors.newSingleThreadExecutor()) {
410 Future<Void> future = pool.submit(() -> {
411 assertThrows(WrongThreadException.class, scope::join);
412 return null;
413 });
414 future.get();
415 }
416
417 // subtask cannot join
418 Subtask<Boolean> subtask = scope.fork(() -> {
419 assertThrows(WrongThreadException.class, () -> { scope.join(); });
420 return true;
421 });
422 scope.join();
423 assertTrue(subtask.get());
424 }
425 }
426
427 /**
428 * Test join with interrupt status set.
429 */
430 @ParameterizedTest
431 @MethodSource("factories")
432 void testInterruptJoin1(ThreadFactory factory) throws Exception {
433 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
434 cf -> cf.withThreadFactory(factory))) {
435
436 Subtask<String> subtask = scope.fork(() -> {
437 Thread.sleep(60_000);
438 return "foo";
439 });
440
441 // join should throw
442 Thread.currentThread().interrupt();
443 try {
444 scope.join();
445 fail("join did not throw");
446 } catch (InterruptedException expected) {
447 assertFalse(Thread.interrupted()); // interrupt status should be cleared
448 }
449 }
450 }
451
452 /**
453 * Test interrupt of thread blocked in join.
454 */
455 @ParameterizedTest
456 @MethodSource("factories")
457 void testInterruptJoin2(ThreadFactory factory) throws Exception {
458 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
459 cf -> cf.withThreadFactory(factory))) {
460
461 var latch = new CountDownLatch(1);
462 Subtask<String> subtask = scope.fork(() -> {
463 Thread.sleep(60_000);
464 return "foo";
465 });
466
467 // interrupt main thread when it blocks in join
468 scheduleInterruptAt("java.util.concurrent.StructuredTaskScopeImpl.join");
469 try {
470 scope.join();
471 fail("join did not throw");
472 } catch (InterruptedException expected) {
473 assertFalse(Thread.interrupted()); // interrupt status should be clear
474 }
475 }
476 }
477
478 /**
479 * Test join when scope is cancelled.
480 */
481 @ParameterizedTest
482 @MethodSource("factories")
483 void testJoinWhenCancelled(ThreadFactory factory) throws Exception {
484 var countingThreadFactory = new CountingThreadFactory(factory);
485 var testJoiner = new CancelAfterOneJoiner<String>();
486
487 try (var scope = StructuredTaskScope.open(testJoiner,
488 cf -> cf.withThreadFactory(countingThreadFactory))) {
489
490 // fork subtask, the scope should be cancelled when the subtask completes
491 var subtask1 = scope.fork(() -> "foo");
492 awaitCancelled(scope);
493
494 // fork second subtask, it should not run
495 var subtask2 = scope.fork(() -> {
496 Thread.sleep(Duration.ofDays(1));
497 return "bar";
498 });
499
500 scope.join();
501
502 assertEquals("foo", subtask1.get());
503 assertEquals(Subtask.State.UNAVAILABLE, subtask2.state());
504 }
505 }
506
507 /**
508 * Test join after scope is closed.
509 */
510 @Test
511 void testJoinAfterClose() throws Exception {
512 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
513 scope.close();
514 assertThrows(IllegalStateException.class, () -> scope.join());
515 }
516 }
517
518 /**
519 * Test join with timeout, subtasks finish before timeout expires.
520 */
521 @ParameterizedTest
522 @MethodSource("factories")
523 void testJoinWithTimeout1(ThreadFactory factory) throws Exception {
524 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
525 cf -> cf.withThreadFactory(factory)
526 .withTimeout(Duration.ofDays(1)))) {
527
528 Subtask<String> subtask = scope.fork(() -> {
529 Thread.sleep(Duration.ofSeconds(1));
530 return "foo";
531 });
532
533 scope.join();
534
535 assertFalse(scope.isCancelled());
536 assertEquals("foo", subtask.get());
537 }
538 }
539
540 /**
541 * Test join with timeout, timeout expires before subtasks finish.
542 */
543 @ParameterizedTest
544 @MethodSource("factories")
545 void testJoinWithTimeout2(ThreadFactory factory) throws Exception {
546 long startMillis = millisTime();
547 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
548 cf -> cf.withThreadFactory(factory)
549 .withTimeout(Duration.ofSeconds(2)))) {
550
551 Subtask<Void> subtask = scope.fork(() -> {
552 Thread.sleep(Duration.ofDays(1));
553 return null;
554 });
555
556 assertThrows(TimeoutException.class, scope::join);
557 expectDuration(startMillis, /*min*/1900, /*max*/20_000);
558
559 assertTrue(scope.isCancelled());
560 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
561 }
562 }
563
564 /**
565 * Test join with timeout that has already expired.
566 */
567 @ParameterizedTest
568 @MethodSource("factories")
569 void testJoinWithTimeout3(ThreadFactory factory) throws Exception {
570 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
571 cf -> cf.withThreadFactory(factory)
572 .withTimeout(Duration.ofSeconds(-1)))) {
573
574 Subtask<Void> subtask = scope.fork(() -> {
575 Thread.sleep(Duration.ofDays(1));
576 return null;
577 });
578
579 assertThrows(TimeoutException.class, scope::join);
580
581 assertTrue(scope.isCancelled());
582 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
583 }
584 }
585
586 /**
587 * Test that cancelling execution interrupts unfinished threads. This test uses
588 * a custom Joiner to cancel execution.
589 */
590 @ParameterizedTest
591 @MethodSource("factories")
592 void testCancelInterruptsThreads2(ThreadFactory factory) throws Exception {
593 var testJoiner = new CancelAfterOneJoiner<String>();
594
595 try (var scope = StructuredTaskScope.open(testJoiner,
596 cf -> cf.withThreadFactory(factory))) {
597
598 // fork subtask1 that runs for a long time
599 var started = new CountDownLatch(1);
600 var interrupted = new CountDownLatch(1);
601 var subtask1 = scope.fork(() -> {
602 started.countDown();
603 try {
604 Thread.sleep(Duration.ofDays(1));
605 } catch (InterruptedException e) {
606 interrupted.countDown();
607 }
608 });
609 started.await();
610
611 // fork subtask2, the scope should be cancelled when the subtask completes
612 var subtask2 = scope.fork(() -> "bar");
613 awaitCancelled(scope);
614
615 // subtask1 should be interrupted
616 interrupted.await();
617
618 scope.join();
619 assertEquals(Subtask.State.UNAVAILABLE, subtask1.state());
620 assertEquals("bar", subtask2.get());
621 }
622 }
623
624 /**
625 * Test that timeout interrupts unfinished threads.
626 */
627 @ParameterizedTest
628 @MethodSource("factories")
629 void testTimeoutInterruptsThreads(ThreadFactory factory) throws Exception {
630 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
631 cf -> cf.withThreadFactory(factory)
632 .withTimeout(Duration.ofSeconds(2)))) {
633
634 var started = new AtomicBoolean();
635 var interrupted = new CountDownLatch(1);
636 Subtask<Void> subtask = scope.fork(() -> {
637 started.set(true);
638 try {
639 Thread.sleep(Duration.ofDays(1));
640 } catch (InterruptedException e) {
641 interrupted.countDown();
642 }
643 return null;
644 });
645
646 // wait for scope to be cancelled by timeout
647 awaitCancelled(scope);
648
649 // if subtask started then it should be interrupted
650 if (started.get()) {
651 interrupted.await();
652 }
653
654 assertThrows(TimeoutException.class, scope::join);
655
656 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
657 }
658 }
659
660 /**
661 * Test close without join, no subtasks forked.
662 */
663 @Test
664 void testCloseWithoutJoin1() {
665 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
666 // do nothing
667 }
668 }
669
670 /**
671 * Test close without join, subtasks forked.
672 */
673 @ParameterizedTest
674 @MethodSource("factories")
675 void testCloseWithoutJoin2(ThreadFactory factory) {
676 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
677 cf -> cf.withThreadFactory(factory))) {
678 Subtask<String> subtask = scope.fork(() -> {
679 Thread.sleep(Duration.ofDays(1));
680 return null;
681 });
682
683 // first call to close should throw
684 assertThrows(IllegalStateException.class, scope::close);
685
686 // subsequent calls to close should not throw
687 for (int i = 0; i < 3; i++) {
688 scope.close();
689 }
690
691 // subtask result/exception not available
692 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
693 assertThrows(IllegalStateException.class, subtask::get);
694 assertThrows(IllegalStateException.class, subtask::exception);
695 }
696 }
697
698 /**
699 * Test close after join throws. Close should not throw as join attempted.
700 */
701 @ParameterizedTest
702 @MethodSource("factories")
703 void testCloseAfterJoinThrows(ThreadFactory factory) throws Exception {
704 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
705 cf -> cf.withThreadFactory(factory))) {
706 var subtask = scope.fork(() -> {
707 Thread.sleep(Duration.ofDays(1));
708 return null;
709 });
710
711 // join throws
712 Thread.currentThread().interrupt();
713 assertThrows(InterruptedException.class, scope::join);
714 assertThrows(IllegalStateException.class, subtask::get);
715
716 } // close should not throw
717 }
718
719 /**
720 * Test close method is owner confined.
721 */
722 @ParameterizedTest
723 @MethodSource("factories")
724 void testCloseConfined(ThreadFactory factory) throws Exception {
725 try (var scope = StructuredTaskScope.open(Joiner.<Boolean>awaitAll(),
726 cf -> cf.withThreadFactory(factory))) {
727
728 // random thread cannot close scope
729 try (var pool = Executors.newCachedThreadPool(factory)) {
730 Future<Boolean> future = pool.submit(() -> {
731 assertThrows(WrongThreadException.class, scope::close);
732 return null;
733 });
734 future.get();
735 }
736
737 // subtask cannot close
738 Subtask<Boolean> subtask = scope.fork(() -> {
739 assertThrows(WrongThreadException.class, scope::close);
740 return true;
741 });
742 scope.join();
743 assertTrue(subtask.get());
744 }
745 }
746
747 /**
748 * Test close with interrupt status set.
749 */
750 @ParameterizedTest
751 @MethodSource("factories")
752 void testInterruptClose1(ThreadFactory factory) throws Exception {
753 var testJoiner = new CancelAfterOneJoiner<String>();
754 try (var scope = StructuredTaskScope.open(testJoiner,
755 cf -> cf.withThreadFactory(factory))) {
756
757 // fork first subtask, a straggler as it continues after being interrupted
758 var started = new CountDownLatch(1);
759 var done = new AtomicBoolean();
760 scope.fork(() -> {
761 started.countDown();
762 try {
763 Thread.sleep(Duration.ofDays(1));
764 } catch (InterruptedException e) {
765 // interrupted by cancel, expected
766 }
767 Thread.sleep(Duration.ofMillis(100)); // force close to wait
768 done.set(true);
769 return null;
770 });
771 started.await();
772
773 // fork second subtask, the scope should be cancelled when this subtask completes
774 scope.fork(() -> "bar");
775 awaitCancelled(scope);
776
777 scope.join();
778
779 // invoke close with interrupt status set
780 Thread.currentThread().interrupt();
781 try {
782 scope.close();
783 } finally {
784 assertTrue(Thread.interrupted()); // clear interrupt status
785 assertTrue(done.get());
786 }
787 }
788 }
789
790 /**
791 * Test interrupting thread waiting in close.
792 */
793 @ParameterizedTest
794 @MethodSource("factories")
795 void testInterruptClose2(ThreadFactory factory) throws Exception {
796 var testJoiner = new CancelAfterOneJoiner<String>();
797 try (var scope = StructuredTaskScope.open(testJoiner,
798 cf -> cf.withThreadFactory(factory))) {
799
800 Thread mainThread = Thread.currentThread();
801
802 // fork first subtask, a straggler as it continues after being interrupted
803 var started = new CountDownLatch(1);
804 var done = new AtomicBoolean();
805 scope.fork(() -> {
806 started.countDown();
807 try {
808 Thread.sleep(Duration.ofDays(1));
809 } catch (InterruptedException e) {
810 // interrupted by cancel, expected
811 }
812
813 // interrupt main thread when it blocks in close
814 interruptThreadAt(mainThread, "java.util.concurrent.StructuredTaskScopeImpl.close");
815
816 Thread.sleep(Duration.ofMillis(100)); // force close to wait
817 done.set(true);
818 return null;
819 });
820 started.await();
821
822 // fork second subtask, the scope should be cancelled when this subtask completes
823 scope.fork(() -> "bar");
824 awaitCancelled(scope);
825
826 scope.join();
827
828 // main thread will be interrupted while blocked in close
829 try {
830 scope.close();
831 } finally {
832 assertTrue(Thread.interrupted()); // clear interrupt status
833 assertTrue(done.get());
834 }
835 }
836 }
837
838 /**
839 * Test that closing an enclosing scope closes the thread flock of a nested scope.
840 */
841 @Test
842 void testCloseThrowsStructureViolation() throws Exception {
843 try (var scope1 = StructuredTaskScope.open(Joiner.awaitAll())) {
844 try (var scope2 = StructuredTaskScope.open(Joiner.awaitAll())) {
845
846 // close enclosing scope
847 try {
848 scope1.close();
849 fail("close did not throw");
850 } catch (StructureViolationException expected) { }
851
852 // underlying flock should be closed
853 var executed = new AtomicBoolean();
854 Subtask<?> subtask = scope2.fork(() -> executed.set(true));
855 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
856 scope2.join();
857 assertFalse(executed.get());
858 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
859 }
860 }
861 }
862
863 /**
864 * Test that isCancelled returns true after close.
865 */
866 @Test
867 void testIsCancelledAfterClose() throws Exception {
868 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
869 assertFalse(scope.isCancelled());
870 scope.close();
871 assertTrue(scope.isCancelled());
872 }
873 }
874
875 /**
876 * Test Joiner.onFork throwing exception.
877 */
878 @Test
879 void testOnForkThrows() throws Exception {
880 var joiner = new Joiner<String, Void>() {
881 @Override
882 public boolean onFork(Subtask<? extends String> subtask) {
883 throw new FooException();
884 }
885 @Override
886 public Void result() {
887 return null;
888 }
889 };
890 try (var scope = StructuredTaskScope.open(joiner)) {
891 assertThrows(FooException.class, () -> scope.fork(() -> "foo"));
892 }
893 }
894
895 /**
896 * Test Joiner.onFork returning true to cancel execution.
897 */
898 @Test
899 void testOnForkCancelsExecution() throws Exception {
900 var joiner = new Joiner<String, Void>() {
901 @Override
902 public boolean onFork(Subtask<? extends String> subtask) {
903 return true;
904 }
905 @Override
906 public Void result() {
907 return null;
908 }
909 };
910 try (var scope = StructuredTaskScope.open(joiner)) {
911 assertFalse(scope.isCancelled());
912 scope.fork(() -> "foo");
913 assertTrue(scope.isCancelled());
914 scope.join();
915 }
916 }
917
918 /**
919 * Test Joiner.onComplete throwing exception causes UHE to be invoked.
920 */
921 @Test
922 void testOnCompleteThrows() throws Exception {
923 var joiner = new Joiner<String, Void>() {
924 @Override
925 public boolean onComplete(Subtask<? extends String> subtask) {
926 throw new FooException();
927 }
928 @Override
929 public Void result() {
930 return null;
931 }
932 };
933 var excRef = new AtomicReference<Throwable>();
934 Thread.UncaughtExceptionHandler uhe = (t, e) -> excRef.set(e);
935 ThreadFactory factory = Thread.ofVirtual()
936 .uncaughtExceptionHandler(uhe)
937 .factory();
938 try (var scope = StructuredTaskScope.open(joiner, cf -> cf.withThreadFactory(factory))) {
939 scope.fork(() -> "foo");
940 scope.join();
941 assertInstanceOf(FooException.class, excRef.get());
942 }
943 }
944
945 /**
946 * Test Joiner.onComplete returning true to cancel execution.
947 */
948 @Test
949 void testOnCompleteCancelsExecution() throws Exception {
950 var joiner = new Joiner<String, Void>() {
951 @Override
952 public boolean onComplete(Subtask<? extends String> subtask) {
953 return true;
954 }
955 @Override
956 public Void result() {
957 return null;
958 }
959 };
960 try (var scope = StructuredTaskScope.open(joiner)) {
961 assertFalse(scope.isCancelled());
962 scope.fork(() -> "foo");
963 awaitCancelled(scope);
964 scope.join();
965 }
966 }
967
968 /**
969 * Test toString.
970 */
971 @Test
972 void testToString() throws Exception {
973 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
974 cf -> cf.withName("duke"))) {
975
976 // open
977 assertTrue(scope.toString().contains("duke"));
978
979 // closed
980 scope.close();
981 assertTrue(scope.toString().contains("duke"));
982 }
983 }
984
985 /**
986 * Test Subtask with task that completes successfully.
987 */
988 @ParameterizedTest
989 @MethodSource("factories")
990 void testSubtaskWhenSuccess(ThreadFactory factory) throws Exception {
991 try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAll(),
992 cf -> cf.withThreadFactory(factory))) {
993
994 Subtask<String> subtask = scope.fork(() -> "foo");
995
996 // before join
997 assertThrows(IllegalStateException.class, subtask::get);
998 assertThrows(IllegalStateException.class, subtask::exception);
999
1000 scope.join();
1001
1002 // after join
1003 assertEquals(Subtask.State.SUCCESS, subtask.state());
1004 assertEquals("foo", subtask.get());
1005 assertThrows(IllegalStateException.class, subtask::exception);
1006 }
1007 }
1008
1009 /**
1010 * Test Subtask with task that fails.
1011 */
1012 @ParameterizedTest
1013 @MethodSource("factories")
1014 void testSubtaskWhenFailed(ThreadFactory factory) throws Exception {
1015 try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAll(),
1016 cf -> cf.withThreadFactory(factory))) {
1017
1018 Subtask<String> subtask = scope.fork(() -> { throw new FooException(); });
1019
1020 // before join
1021 assertThrows(IllegalStateException.class, subtask::get);
1022 assertThrows(IllegalStateException.class, subtask::exception);
1023
1024 scope.join();
1025
1026 // after join
1027 assertEquals(Subtask.State.FAILED, subtask.state());
1028 assertThrows(IllegalStateException.class, subtask::get);
1029 assertTrue(subtask.exception() instanceof FooException);
1030 }
1031 }
1032
1033 /**
1034 * Test Subtask with a task that has not completed.
1035 */
1036 @ParameterizedTest
1037 @MethodSource("factories")
1038 void testSubtaskWhenNotCompleted(ThreadFactory factory) throws Exception {
1039 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(),
1040 cf -> cf.withThreadFactory(factory))) {
1041 Subtask<Void> subtask = scope.fork(() -> {
1042 Thread.sleep(Duration.ofDays(1));
1043 return null;
1044 });
1045
1046 // before join
1047 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1048 assertThrows(IllegalStateException.class, subtask::get);
1049 assertThrows(IllegalStateException.class, subtask::exception);
1050
1051 // attempt join, join throws
1052 Thread.currentThread().interrupt();
1053 assertThrows(InterruptedException.class, scope::join);
1054
1055 // after join
1056 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1057 assertThrows(IllegalStateException.class, subtask::get);
1058 assertThrows(IllegalStateException.class, subtask::exception);
1059 }
1060 }
1061
1062 /**
1063 * Test Subtask forked after execution cancelled.
1064 */
1065 @ParameterizedTest
1066 @MethodSource("factories")
1067 void testSubtaskWhenCancelled(ThreadFactory factory) throws Exception {
1068 try (var scope = StructuredTaskScope.open(new CancelAfterOneJoiner<String>())) {
1069 scope.fork(() -> "foo");
1070 awaitCancelled(scope);
1071
1072 var subtask = scope.fork(() -> "foo");
1073
1074 // before join
1075 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1076 assertThrows(IllegalStateException.class, subtask::get);
1077 assertThrows(IllegalStateException.class, subtask::exception);
1078
1079 scope.join();
1080
1081 // after join
1082 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1083 assertThrows(IllegalStateException.class, subtask::get);
1084 assertThrows(IllegalStateException.class, subtask::exception);
1085 }
1086 }
1087
1088 /**
1089 * Test Subtask::toString.
1090 */
1091 @Test
1092 void testSubtaskToString() throws Exception {
1093 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
1094 var latch = new CountDownLatch(1);
1095 var subtask1 = scope.fork(() -> {
1096 latch.await();
1097 return "foo";
1098 });
1099 var subtask2 = scope.fork(() -> { throw new FooException(); });
1100
1101 // subtask1 result is unavailable
1102 assertTrue(subtask1.toString().contains("Unavailable"));
1103 latch.countDown();
1104
1105 scope.join();
1106
1107 assertTrue(subtask1.toString().contains("Completed successfully"));
1108 assertTrue(subtask2.toString().contains("Failed"));
1109 }
1110 }
1111
1112 /**
1113 * Test Joiner.allSuccessfulOrThrow() with no subtasks.
1114 */
1115 @Test
1116 void testAllSuccessfulOrThrow1() throws Throwable {
1117 try (var scope = StructuredTaskScope.open(Joiner.allSuccessfulOrThrow())) {
1118 var subtasks = scope.join().toList();
1119 assertTrue(subtasks.isEmpty());
1120 }
1121 }
1122
1123 /**
1124 * Test Joiner.allSuccessfulOrThrow() with subtasks that complete successfully.
1125 */
1126 @ParameterizedTest
1127 @MethodSource("factories")
1128 void testAllSuccessfulOrThrow2(ThreadFactory factory) throws Throwable {
1129 try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow(),
1130 cf -> cf.withThreadFactory(factory))) {
1131 var subtask1 = scope.fork(() -> "foo");
1132 var subtask2 = scope.fork(() -> "bar");
1133 var subtasks = scope.join().toList();
1134 assertEquals(List.of(subtask1, subtask2), subtasks);
1135 assertEquals("foo", subtask1.get());
1136 assertEquals("bar", subtask2.get());
1137 }
1138 }
1139
1140 /**
1141 * Test Joiner.allSuccessfulOrThrow() with a subtask that complete successfully and
1142 * a subtask that fails.
1143 */
1144 @ParameterizedTest
1145 @MethodSource("factories")
1146 void testAllSuccessfulOrThrow3(ThreadFactory factory) throws Throwable {
1147 try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow(),
1148 cf -> cf.withThreadFactory(factory))) {
1149 scope.fork(() -> "foo");
1150 scope.fork(() -> { throw new FooException(); });
1151 try {
1152 scope.join();
1153 } catch (FailedException e) {
1154 assertTrue(e.getCause() instanceof FooException);
1155 }
1156 }
1157 }
1158
1159 /**
1160 * Test Joiner.anySuccessfulResultOrThrow() with no subtasks.
1161 */
1162 @Test
1163 void testAnySuccessfulResultOrThrow1() throws Exception {
1164 try (var scope = StructuredTaskScope.open(Joiner.anySuccessfulResultOrThrow())) {
1165 try {
1166 scope.join();
1167 } catch (FailedException e) {
1168 assertTrue(e.getCause() instanceof NoSuchElementException);
1169 }
1170 }
1171 }
1172
1173 /**
1174 * Test Joiner.anySuccessfulResultOrThrow() with a subtask that completes successfully.
1175 */
1176 @ParameterizedTest
1177 @MethodSource("factories")
1178 void testAnySuccessfulResultOrThrow2(ThreadFactory factory) throws Exception {
1179 try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulResultOrThrow(),
1180 cf -> cf.withThreadFactory(factory))) {
1181 scope.fork(() -> "foo");
1182 String result = scope.join();
1183 assertEquals("foo", result);
1184 }
1185 }
1186
1187 /**
1188 * Test Joiner.anySuccessfulResultOrThrow() with a subtask that completes successfully
1189 * with a null result.
1190 */
1191 @ParameterizedTest
1192 @MethodSource("factories")
1193 void testAnySuccessfulResultOrThrow3(ThreadFactory factory) throws Exception {
1194 try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulResultOrThrow(),
1195 cf -> cf.withThreadFactory(factory))) {
1196 scope.fork(() -> null);
1197 String result = scope.join();
1198 assertNull(result);
1199 }
1200 }
1201
1202 /**
1203 * Test Joiner.anySuccessfulResultOrThrow() with a subtask that complete succcessfully
1204 * and a subtask that fails.
1205 */
1206 @ParameterizedTest
1207 @MethodSource("factories")
1208 void testAnySuccessfulResultOrThrow4(ThreadFactory factory) throws Exception {
1209 try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulResultOrThrow(),
1210 cf -> cf.withThreadFactory(factory))) {
1211 scope.fork(() -> "foo");
1212 scope.fork(() -> { throw new FooException(); });
1213 String first = scope.join();
1214 assertEquals("foo", first);
1215 }
1216 }
1217
1218 /**
1219 * Test Joiner.anySuccessfulResultOrThrow() with a subtask that fails.
1220 */
1221 @ParameterizedTest
1222 @MethodSource("factories")
1223 void testAnySuccessfulResultOrThrow5(ThreadFactory factory) throws Exception {
1224 try (var scope = StructuredTaskScope.open(Joiner.anySuccessfulResultOrThrow(),
1225 cf -> cf.withThreadFactory(factory))) {
1226 scope.fork(() -> { throw new FooException(); });
1227 Throwable ex = assertThrows(FailedException.class, scope::join);
1228 assertTrue(ex.getCause() instanceof FooException);
1229 }
1230 }
1231
1232 /**
1233 * Test Joiner.awaitAllSuccessfulOrThrow() with no subtasks.
1234 */
1235 @Test
1236 void testAwaitSuccessfulOrThrow1() throws Throwable {
1237 try (var scope = StructuredTaskScope.open(Joiner.awaitAllSuccessfulOrThrow())) {
1238 var result = scope.join();
1239 assertNull(result);
1240 }
1241 }
1242
1243 /**
1244 * Test Joiner.awaitAllSuccessfulOrThrow() with subtasks that complete successfully.
1245 */
1246 @ParameterizedTest
1247 @MethodSource("factories")
1248 void testAwaitSuccessfulOrThrow2(ThreadFactory factory) throws Throwable {
1249 try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow(),
1250 cf -> cf.withThreadFactory(factory))) {
1251 var subtask1 = scope.fork(() -> "foo");
1252 var subtask2 = scope.fork(() -> "bar");
1253 var result = scope.join();
1254 assertNull(result);
1255 assertEquals("foo", subtask1.get());
1256 assertEquals("bar", subtask2.get());
1257 }
1258 }
1259
1260 /**
1261 * Test Joiner.awaitAllSuccessfulOrThrow() with a subtask that complete successfully and
1262 * a subtask that fails.
1263 */
1264 @ParameterizedTest
1265 @MethodSource("factories")
1266 void testAwaitSuccessfulOrThrow3(ThreadFactory factory) throws Throwable {
1267 try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow(),
1268 cf -> cf.withThreadFactory(factory))) {
1269 scope.fork(() -> "foo");
1270 scope.fork(() -> { throw new FooException(); });
1271 try {
1272 scope.join();
1273 } catch (FailedException e) {
1274 assertTrue(e.getCause() instanceof FooException);
1275 }
1276 }
1277 }
1278
1279 /**
1280 * Test Joiner.awaitAll() with no subtasks.
1281 */
1282 @Test
1283 void testAwaitAll1() throws Throwable {
1284 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
1285 var result = scope.join();
1286 assertNull(result);
1287 }
1288 }
1289
1290 /**
1291 * Test Joiner.awaitAll() with subtasks that complete successfully.
1292 */
1293 @ParameterizedTest
1294 @MethodSource("factories")
1295 void testAwaitAll2(ThreadFactory factory) throws Throwable {
1296 try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAll(),
1297 cf -> cf.withThreadFactory(factory))) {
1298 var subtask1 = scope.fork(() -> "foo");
1299 var subtask2 = scope.fork(() -> "bar");
1300 var result = scope.join();
1301 assertNull(result);
1302 assertEquals("foo", subtask1.get());
1303 assertEquals("bar", subtask2.get());
1304 }
1305 }
1306
1307 /**
1308 * Test Joiner.awaitAll() with a subtask that complete successfully and a subtask
1309 * that fails.
1310 */
1311 @ParameterizedTest
1312 @MethodSource("factories")
1313 void testAwaitAll3(ThreadFactory factory) throws Throwable {
1314 try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAll(),
1315 cf -> cf.withThreadFactory(factory))) {
1316 var subtask1 = scope.fork(() -> "foo");
1317 var subtask2 = scope.fork(() -> { throw new FooException(); });
1318 var result = scope.join();
1319 assertNull(result);
1320 assertEquals("foo", subtask1.get());
1321 assertTrue(subtask2.exception() instanceof FooException);
1322 }
1323 }
1324
1325 /**
1326 * Test Joiner.allUntil(Predicate) with no subtasks.
1327 */
1328 @Test
1329 void testAllUntil1() throws Throwable {
1330 try (var scope = StructuredTaskScope.open(Joiner.allUntil(s -> false))) {
1331 var subtasks = scope.join();
1332 assertEquals(0, subtasks.count());
1333 }
1334 }
1335
1336 /**
1337 * Test Joiner.allUntil(Predicate) with no cancellation.
1338 */
1339 @ParameterizedTest
1340 @MethodSource("factories")
1341 void testAllUntil2(ThreadFactory factory) throws Exception {
1342 try (var scope = StructuredTaskScope.open(Joiner.<String>allUntil(s -> false),
1343 cf -> cf.withThreadFactory(factory))) {
1344
1345 var subtask1 = scope.fork(() -> "foo");
1346 var subtask2 = scope.fork(() -> { throw new FooException(); });
1347
1348 var subtasks = scope.join().toList();
1349 assertEquals(2, subtasks.size());
1350
1351 assertSame(subtask1, subtasks.get(0));
1352 assertSame(subtask2, subtasks.get(1));
1353 assertEquals("foo", subtask1.get());
1354 assertTrue(subtask2.exception() instanceof FooException);
1355 }
1356 }
1357
1358 /**
1359 * Test Joiner.allUntil(Predicate) with cancellation after one subtask completes.
1360 */
1361 @ParameterizedTest
1362 @MethodSource("factories")
1363 void testAllUntil3(ThreadFactory factory) throws Exception {
1364 try (var scope = StructuredTaskScope.open(Joiner.<String>allUntil(s -> true),
1365 cf -> cf.withThreadFactory(factory))) {
1366
1367 var subtask1 = scope.fork(() -> "foo");
1368 var subtask2 = scope.fork(() -> {
1369 Thread.sleep(Duration.ofDays(1));
1370 return "bar";
1371 });
1372
1373 var subtasks = scope.join().toList();
1374
1375 assertEquals(2, subtasks.size());
1376 assertSame(subtask1, subtasks.get(0));
1377 assertSame(subtask2, subtasks.get(1));
1378 assertEquals("foo", subtask1.get());
1379 assertEquals(Subtask.State.UNAVAILABLE, subtask2.state());
1380 }
1381 }
1382
1383 /**
1384 * Test Joiner.allUntil(Predicate) with cancellation after serveral subtasks complete.
1385 */
1386 @ParameterizedTest
1387 @MethodSource("factories")
1388 void testAllUntil4(ThreadFactory factory) throws Exception {
1389
1390 // cancel execution after two or more failures
1391 class CancelAfterTwoFailures<T> implements Predicate<Subtask<? extends T>> {
1392 final AtomicInteger failedCount = new AtomicInteger();
1393 @Override
1394 public boolean test(Subtask<? extends T> subtask) {
1395 return subtask.state() == Subtask.State.FAILED
1396 && failedCount.incrementAndGet() >= 2;
1397 }
1398 }
1399 var joiner = Joiner.allUntil(new CancelAfterTwoFailures<String>());
1400
1401 try (var scope = StructuredTaskScope.open(joiner)) {
1402 int forkCount = 0;
1403
1404 // fork subtasks until execution cancelled
1405 while (!scope.isCancelled()) {
1406 scope.fork(() -> "foo");
1407 scope.fork(() -> { throw new FooException(); });
1408 forkCount += 2;
1409 Thread.sleep(Duration.ofMillis(20));
1410 }
1411
1412 var subtasks = scope.join().toList();
1413 assertEquals(forkCount, subtasks.size());
1414
1415 long failedCount = subtasks.stream()
1416 .filter(s -> s.state() == Subtask.State.FAILED)
1417 .count();
1418 assertTrue(failedCount >= 2);
1419 }
1420 }
1421
1422 /**
1423 * Test Test Joiner.allUntil(Predicate) where the Predicate's test method throws.
1424 */
1425 @Test
1426 void testAllUntil5() throws Exception {
1427 var joiner = Joiner.allUntil(_ -> { throw new FooException(); });
1428 var excRef = new AtomicReference<Throwable>();
1429 Thread.UncaughtExceptionHandler uhe = (t, e) -> excRef.set(e);
1430 ThreadFactory factory = Thread.ofVirtual()
1431 .uncaughtExceptionHandler(uhe)
1432 .factory();
1433 try (var scope = StructuredTaskScope.open(joiner, cf -> cf.withThreadFactory(factory))) {
1434 scope.fork(() -> "foo");
1435 scope.join();
1436 assertInstanceOf(FooException.class, excRef.get());
1437 }
1438 }
1439
1440 /**
1441 * Test Joiner default methods.
1442 */
1443 @Test
1444 void testJoinerDefaultMethods() throws Exception {
1445 try (var scope = StructuredTaskScope.open(new CancelAfterOneJoiner<String>())) {
1446
1447 // need subtasks to test default methods
1448 var subtask1 = scope.fork(() -> "foo");
1449 awaitCancelled(scope);
1450 var subtask2 = scope.fork(() -> "bar");
1451 scope.join();
1452
1453 assertEquals(Subtask.State.SUCCESS, subtask1.state());
1454 assertEquals(Subtask.State.UNAVAILABLE, subtask2.state());
1455
1456 // Joiner that does not override default methods
1457 Joiner<Object, Void> joiner = () -> null;
1458 assertThrows(NullPointerException.class, () -> joiner.onFork(null));
1459 assertThrows(NullPointerException.class, () -> joiner.onComplete(null));
1460 assertThrows(IllegalArgumentException.class, () -> joiner.onFork(subtask1));
1461 assertFalse(joiner.onFork(subtask2));
1462 assertFalse(joiner.onComplete(subtask1));
1463 assertThrows(IllegalArgumentException.class, () -> joiner.onComplete(subtask2));
1464 }
1465 }
1466
1467 /**
1468 * Test Joiners onFork/onComplete methods with a subtask in an unexpected state.
1469 */
1470 @Test
1471 void testJoinersWithUnavailableResult() throws Exception {
1472 try (var scope = StructuredTaskScope.open()) {
1473 var done = new CountDownLatch(1);
1474 var subtask = scope.fork(() -> {
1475 done.await();
1476 return null;
1477 });
1478
1479 // onComplete with uncompleted task should throw IAE
1480 assertEquals(Subtask.State.UNAVAILABLE, subtask.state());
1481 assertThrows(IllegalArgumentException.class,
1482 () -> Joiner.allSuccessfulOrThrow().onComplete(subtask));
1483 assertThrows(IllegalArgumentException.class,
1484 () -> Joiner.anySuccessfulResultOrThrow().onComplete(subtask));
1485 assertThrows(IllegalArgumentException.class,
1486 () -> Joiner.awaitAllSuccessfulOrThrow().onComplete(subtask));
1487 assertThrows(IllegalArgumentException.class,
1488 () -> Joiner.awaitAll().onComplete(subtask));
1489 assertThrows(IllegalArgumentException.class,
1490 () -> Joiner.allUntil(_ -> false).onComplete(subtask));
1491
1492 done.countDown();
1493 scope.join();
1494
1495 // onFork with completed task should throw IAE
1496 assertEquals(Subtask.State.SUCCESS, subtask.state());
1497 assertThrows(IllegalArgumentException.class,
1498 () -> Joiner.allSuccessfulOrThrow().onFork(subtask));
1499 assertThrows(IllegalArgumentException.class,
1500 () -> Joiner.anySuccessfulResultOrThrow().onFork(subtask));
1501 assertThrows(IllegalArgumentException.class,
1502 () -> Joiner.awaitAllSuccessfulOrThrow().onFork(subtask));
1503 assertThrows(IllegalArgumentException.class,
1504 () -> Joiner.awaitAll().onFork(subtask));
1505 assertThrows(IllegalArgumentException.class,
1506 () -> Joiner.allUntil(_ -> false).onFork(subtask));
1507 }
1508
1509 }
1510
1511 /**
1512 * Test the Configuration function apply method throwing an exception.
1513 */
1514 @Test
1515 void testConfigFunctionThrows() throws Exception {
1516 assertThrows(FooException.class,
1517 () -> StructuredTaskScope.open(Joiner.awaitAll(),
1518 cf -> { throw new FooException(); }));
1519 }
1520
1521 /**
1522 * Test Configuration equals/hashCode/toString
1523 */
1524 @Test
1525 void testConfigMethods() throws Exception {
1526 Function<Configuration, Configuration> testConfig = cf -> {
1527 var name = "duke";
1528 var threadFactory = Thread.ofPlatform().factory();
1529 var timeout = Duration.ofSeconds(10);
1530
1531 assertEquals(cf, cf);
1532 assertEquals(cf.withName(name), cf.withName(name));
1533 assertEquals(cf.withThreadFactory(threadFactory), cf.withThreadFactory(threadFactory));
1534 assertEquals(cf.withTimeout(timeout), cf.withTimeout(timeout));
1535
1536 assertNotEquals(cf, cf.withName(name));
1537 assertNotEquals(cf, cf.withThreadFactory(threadFactory));
1538 assertNotEquals(cf, cf.withTimeout(timeout));
1539
1540 assertEquals(cf.withName(name).hashCode(), cf.withName(name).hashCode());
1541 assertEquals(cf.withThreadFactory(threadFactory).hashCode(),
1542 cf.withThreadFactory(threadFactory).hashCode());
1543 assertEquals(cf.withTimeout(timeout).hashCode(), cf.withTimeout(timeout).hashCode());
1544
1545 assertTrue(cf.withName(name).toString().contains(name));
1546 assertTrue(cf.withThreadFactory(threadFactory).toString().contains(threadFactory.toString()));
1547 assertTrue(cf.withTimeout(timeout).toString().contains(timeout.toString()));
1548
1549 return cf;
1550 };
1551 try (var scope = StructuredTaskScope.open(Joiner.awaitAll(), testConfig)) {
1552 // do nothing
1553 }
1554 }
1555
1556 /**
1557 * Test for NullPointerException.
1558 */
1559 @Test
1560 void testNulls() throws Exception {
1561 assertThrows(NullPointerException.class,
1562 () -> StructuredTaskScope.open(null));
1563 assertThrows(NullPointerException.class,
1564 () -> StructuredTaskScope.open(null, cf -> cf));
1565 assertThrows(NullPointerException.class,
1566 () -> StructuredTaskScope.open(Joiner.awaitAll(), null));
1567
1568 assertThrows(NullPointerException.class, () -> Joiner.allUntil(null));
1569
1570 // fork
1571 try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) {
1572 assertThrows(NullPointerException.class, () -> scope.fork((Callable<Object>) null));
1573 assertThrows(NullPointerException.class, () -> scope.fork((Runnable) null));
1574 }
1575
1576 // Configuration and withXXX methods
1577 assertThrows(NullPointerException.class,
1578 () -> StructuredTaskScope.open(Joiner.awaitAll(), cf -> null));
1579 assertThrows(NullPointerException.class,
1580 () -> StructuredTaskScope.open(Joiner.awaitAll(), cf -> cf.withName(null)));
1581 assertThrows(NullPointerException.class,
1582 () -> StructuredTaskScope.open(Joiner.awaitAll(), cf -> cf.withThreadFactory(null)));
1583 assertThrows(NullPointerException.class,
1584 () -> StructuredTaskScope.open(Joiner.awaitAll(), cf -> cf.withTimeout(null)));
1585
1586 // Joiner.onFork/onComplete
1587 assertThrows(NullPointerException.class,
1588 () -> Joiner.awaitAllSuccessfulOrThrow().onFork(null));
1589 assertThrows(NullPointerException.class,
1590 () -> Joiner.awaitAllSuccessfulOrThrow().onComplete(null));
1591 assertThrows(NullPointerException.class,
1592 () -> Joiner.awaitAll().onFork(null));
1593 assertThrows(NullPointerException.class,
1594 () -> Joiner.awaitAll().onComplete(null));
1595 assertThrows(NullPointerException.class,
1596 () -> Joiner.allSuccessfulOrThrow().onFork(null));
1597 assertThrows(NullPointerException.class,
1598 () -> Joiner.allSuccessfulOrThrow().onComplete(null));
1599 assertThrows(NullPointerException.class,
1600 () -> Joiner.anySuccessfulResultOrThrow().onFork(null));
1601 assertThrows(NullPointerException.class,
1602 () -> Joiner.anySuccessfulResultOrThrow().onComplete(null));
1603 }
1604
1605 /**
1606 * ThreadFactory that counts usage.
1607 */
1608 private static class CountingThreadFactory implements ThreadFactory {
1609 final ThreadFactory delegate;
1610 final AtomicInteger threadCount = new AtomicInteger();
1611 CountingThreadFactory(ThreadFactory delegate) {
1612 this.delegate = delegate;
1613 }
1614 @Override
1615 public Thread newThread(Runnable task) {
1616 threadCount.incrementAndGet();
1617 return delegate.newThread(task);
1618 }
1619 int threadCount() {
1620 return threadCount.get();
1621 }
1622 }
1623
1624 /**
1625 * A joiner that counts that counts the number of subtasks that are forked and the
1626 * number of subtasks that complete.
1627 */
1628 private static class CountingJoiner<T> implements Joiner<T, Void> {
1629 final AtomicInteger onForkCount = new AtomicInteger();
1630 final AtomicInteger onCompleteCount = new AtomicInteger();
1631 @Override
1632 public boolean onFork(Subtask<? extends T> subtask) {
1633 onForkCount.incrementAndGet();
1634 return false;
1635 }
1636 @Override
1637 public boolean onComplete(Subtask<? extends T> subtask) {
1638 onCompleteCount.incrementAndGet();
1639 return false;
1640 }
1641 @Override
1642 public Void result() {
1643 return null;
1644 }
1645 int onForkCount() {
1646 return onForkCount.get();
1647 }
1648 int onCompleteCount() {
1649 return onCompleteCount.get();
1650 }
1651 }
1652
1653 /**
1654 * A joiner that cancels execution when a subtask completes. It also keeps a count
1655 * of the number of subtasks that are forked and the number of subtasks that complete.
1656 */
1657 private static class CancelAfterOneJoiner<T> implements Joiner<T, Void> {
1658 final AtomicInteger onForkCount = new AtomicInteger();
1659 final AtomicInteger onCompleteCount = new AtomicInteger();
1660 @Override
1661 public boolean onFork(Subtask<? extends T> subtask) {
1662 onForkCount.incrementAndGet();
1663 return false;
1664 }
1665 @Override
1666 public boolean onComplete(Subtask<? extends T> subtask) {
1667 onCompleteCount.incrementAndGet();
1668 return true;
1669 }
1670 @Override
1671 public Void result() {
1672 return null;
1673 }
1674 int onForkCount() {
1675 return onForkCount.get();
1676 }
1677 int onCompleteCount() {
1678 return onCompleteCount.get();
1679 }
1680 }
1681
1682 /**
1683 * A runtime exception for tests.
1684 */
1685 private static class FooException extends RuntimeException {
1686 FooException() { }
1687 FooException(Throwable cause) { super(cause); }
1688 }
1689
1690 /**
1691 * Returns the current time in milliseconds.
1692 */
1693 private long millisTime() {
1694 long now = System.nanoTime();
1695 return TimeUnit.MILLISECONDS.convert(now, TimeUnit.NANOSECONDS);
1696 }
1697
1698 /**
1699 * Check the duration of a task
1700 * @param start start time, in milliseconds
1701 * @param min minimum expected duration, in milliseconds
1702 * @param max maximum expected duration, in milliseconds
1703 * @return the duration (now - start), in milliseconds
1704 */
1705 private long expectDuration(long start, long min, long max) {
1706 long duration = millisTime() - start;
1707 assertTrue(duration >= min,
1708 "Duration " + duration + "ms, expected >= " + min + "ms");
1709 assertTrue(duration <= max,
1710 "Duration " + duration + "ms, expected <= " + max + "ms");
1711 return duration;
1712 }
1713
1714 /**
1715 * Wait for the given scope to be cancelled.
1716 */
1717 private static void awaitCancelled(StructuredTaskScope<?, ?> scope) throws InterruptedException {
1718 while (!scope.isCancelled()) {
1719 Thread.sleep(Duration.ofMillis(20));
1720 }
1721 }
1722
1723 /**
1724 * Interrupts a thread when it waits (timed or untimed) at location "{@code c.m}".
1725 * {@code c} is the fully qualified class name and {@code m} is the method name.
1726 */
1727 private void interruptThreadAt(Thread target, String location) throws InterruptedException {
1728 int index = location.lastIndexOf('.');
1729 String className = location.substring(0, index);
1730 String methodName = location.substring(index + 1);
1731
1732 boolean found = false;
1733 while (!found) {
1734 Thread.State state = target.getState();
1735 assertTrue(state != TERMINATED);
1736 if ((state == WAITING || state == TIMED_WAITING)
1737 && contains(target.getStackTrace(), className, methodName)) {
1738 found = true;
1739 } else {
1740 Thread.sleep(20);
1741 }
1742 }
|