Интерфейс StructuredTaskScope<T, R, R_X extends Throwable>

StructuredTaskScope определяет статический метод open() для создания и открытия новой StructuredTaskScope. Он определяет метод close() для её закрытия. API предназначен для использования с оператором try-with-resources, в котором StructuredTaskScope открывается как ресурс, а затем закрывается автоматически. Код внутри блока try использует метод fork(Callable), чтобы запускать подзадачи. Каждый вызов метода fork(Callable) запускает новый Thread (обычно виртуальный поток) для выполнения подзадачи в виде метода, возвращающего значение. Подзадача выполняется одновременно с кодом внутри блока try и с другими подзадачами, запущенными в области. Запустив все подзадачи, код внутри блока использует метод join(), чтобы дождаться завершения всех подзадач (или другого исхода) как единой операции. Код после метода join() обрабатывает результат. Выполнение не продолжается за пределами блока try (или метода close), пока не завершатся все потоки, запущенные в области для выполнения подзадач.

Для обеспечения корректного использования методы fork(Callable), join() и close() могут вызываться только потоком-владельцем (потоком, открывшим StructuredTaskScope); метод fork(Callable) нельзя вызывать после join(); метод join() необходимо вызвать, чтобы получить результат после запуска подзадач; а метод close() после закрытия выбрасывает исключение, если владелец не вызвал метод join() после запуска подзадач.

Для начала рассмотрим пример: задача «main» разбивается на две подзадачи, которые параллельно получают значения от двух удалённых служб. Главная задача объединяет результаты обеих подзадач. В примере вызывается fork(Callable) для запуска двух подзадач. Каждый вызов fork(Callable) возвращает Subtask в качестве дескриптора запущенной подзадачи. Обе подзадачи могут завершиться успешно, одна подзадача может завершиться успешно, а другая — с ошибкой, или обе подзадачи могут завершиться с ошибкой.

Главная задача в примере заинтересована в успешном результате обеих подзадач. В методе join() она ожидает успешного завершения обеих подзадач или сбоя любой из них. Если обе подзадачи завершатся успешно, метод join() завершится нормально, и задача использует метод Subtask.get()^PREVIEW, чтобы получить результат каждой подзадачи. Если одна из подзадач завершится с ошибкой, другая подзадача будет отменена, а метод join() выбросит ExecutionException, указав исключение от сбойной подзадачи в качестве причины.

   try (var scope = StructuredTaskScope.open()) {

       Subtask<String> subtask1 = scope.fork(() -> fetchFromRemoteService1());
       Subtask<Integer> subtask2 = scope.fork(() -> fetchFromRemoteService2());

       // throws ExecutionException if either subtask fails
       scope.join();

       // both subtasks completed successfully
       var result = new MyResult(subtask1.get(), subtask2.get());

   } // close

Метод close() всегда ожидает завершения потоков, выполняющих подзадачи, даже если метод join() выбрасывает исключение, поэтому выполнение не может продолжиться за пределами метода close(), пока прерванные потоки не завершатся.

Чтобы обеспечить возможность отмены, подзадачи должны быть написаны так, чтобы при прерывании они завершались как можно скорее. Подзадачи, которые не реагируют на прерывание, например блокируются на методах, не поддерживающих прерывание, могут задержать выполнение метода close() на неопределённое время.

В примере подзадачи возвращают результаты разных типов (String и Integer). В других случаях все подзадачи могут возвращать результаты одного типа. Если бы в примере для открытия области использовался StructuredTaskScope.<String>open(), то его можно было бы использовать только для запуска подзадач, возвращающих результат String.

Объединители

В приведённом выше примере метод join() завершается нормально и возвращает null, если все подзадачи выполнены успешно. Если какая-либо подзадача завершается с ошибкой, он выбрасывает ExecutionException. Другие политики и результаты можно получить, создав StructuredTaskScope с помощью Joiner^PREVIEW, реализующего требуемую политику и результат. Joiner обрабатывает подзадачи при их запуске и завершении и формирует результат для метода join(). Вместо null, Joiner может привести к тому, что join() вернёт результат конкретной подзадачи, коллекцию результатов или объект, созданный на основе результатов некоторых или всех подзадач. Если результатом является исключение, Joiner может привести к тому, что join() выбросит исключение, отличное от ExecutionException.

Joiner может отменить область (иногда это называют «досрочным завершением»), когда выполнено некоторое условие, например сбой подзадачи, и результат остальных выполняющихся подзадач уже не требуется. Отмена области не позволяет запускать в ней новые потоки, отменяет подзадачи, выполнение которых ещё не завершено, и пробуждает метод join() с результатом (значением или исключением). В приведённом выше примере результатом является нормальное завершение join(), когда все подзадачи выполнены успешно. Область отменяется, если любая подзадача завершается с ошибкой, а join() выбрасывает ExecutionException, указывая исключение от сбойной подзадачи в качестве причины. Другие реализации Joiner могут отменять область по другим причинам и приводить к тому, что join() выбрасывает другое исключение, если результатом является исключение.

Интерфейс Joiner^PREVIEW определяет статические фабричные методы для создания Joiner для ряда распространённых случаев. Если требуется более сложная или пользовательская политика, можно реализовать этот интерфейс. Joiner, возвращающий результат, отличный от null, может избавить от необходимости вести учёт и хранить ссылки на объекты Subtask, возвращаемые методом fork(Callable).

Рассмотрим другой пример, в котором главная задача разбивается на две подзадачи. В этом примере каждая подзадача возвращает результат String, а главной задаче нужен только результат первой успешно завершившейся подзадачи. В примере используется Joiner.anySuccessfulOrThrow()^PREVIEW для создания Joiner, возвращающего результат любой успешно завершившейся подзадачи.

   try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulOrThrow())) {

       scope.fork(callable1);
       scope.fork(callable2);

       // throws ExecutionException if both subtasks fail
       String firstResult = scope.join();

   } // close

В примере задача запускает две подзадачи, а затем в методе join() ожидает успешного завершения любой из них или сбоя обеих. Если одна из подзадач завершится успешно, Joiner вызовет отмену другой подзадачи (поток, выполняющий её, будет прерван), а метод join() вернёт результат успешно завершившейся подзадачи. Отмена другой подзадачи позволяет задаче не ожидать результат, который ей не нужен. Если обе подзадачи завершатся с ошибкой, метод join() выбросит ExecutionException, указав исключение одной из подзадач в качестве причины. Joiner.anySuccessfulOrThrow(Function)^PREVIEW можно использовать с функцией, которая при сбое всех подзадач создаёт исключение, отличное от ExecutionException.

Конфигурация

Методы open() и open(Joiner) создают StructuredTaskScope с конфигурацией по умолчанию. Конфигурация по умолчанию содержит ThreadFactory, создающий безымянные виртуальные потоки; область не получает имени, и время ожидания не задаётся.

Методы open(UnaryOperator) и open(Joiner, UnaryOperator) можно использовать для создания StructuredTaskScope с другим ThreadFactory, именем для мониторинга и управления или временем ожидания, при истечении которого область отменяется, если это происходит до или во время ожидания завершения подзадач. Методы open принимают оператор, применяемый к конфигурации по умолчанию и возвращающий Configuration^PREVIEW для создаваемой StructuredTaskScope.

В следующем примере открывается новая StructuredTaskScope с ThreadFactory, создающей виртуальные потоки с именами "duke-0", "duke-1" ...

    ThreadFactory factory = Thread.ofVirtual().name("duke-", 0).factory();

    try (var scope = StructuredTaskScope.open(cf -> cf.withThreadFactory(factory))) {

        var subtask1 = scope.fork( .. );   // runs in a virtual thread with name "duke-0"
        var subtask2 = scope.fork( .. );   // runs in a virtual thread with name "duke-1"

        scope.join();

        var result = new MyResult(subtask1.get(), subtask2.get());

     }

Во втором примере задаётся время ожидания, представленное объектом Duration. Отсчёт времени ожидания начинается при открытии новой области. Если время ожидания истечёт до или во время ожидания в методе join(), область будет отменена (потоки, выполняющие незавершённые подзадачи, будут прерваны), а метод join() выбросит ExecutionException, указав в качестве причины CancelledByTimeoutException^PREVIEW.

   Duration timeout = Duration.ofSeconds(10);

   try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow(),
                                             cf -> cf.withTimeout(timeout))) {

       scope.fork(callable1);   // subtask takes a really long time
       scope.fork(callable2);

       // throws ExecutionException with CancelledByTimeoutException as cause
       List<String> results = scope.join();

   }

Обработка исключений

Результатом метода join() является значение или исключение. Если результатом является исключение, его причиной обычно будет исключение от сбойной подзадачи или CancelledByTimeoutException^PREVIEW, если было задано время ожидания.

В некоторых случаях может быть полезно добавить блок catch к оператору try-with-resources для обработки исключения. В следующем примере метод open(UnaryOperator) используется для открытия области с заданным временем ожидания. Метод join() в этом примере выбрасывает ExecutionException, если какая-либо подзадача завершится с ошибкой или истечёт время ожидания. Причиной исключения будет исключение от сбойной подзадачи или CancelledByTimeoutException. В примере оператор switch используется для выбора действия в зависимости от причины.

   try (var scope = StructuredTaskScope.open(cf -> cf.withTimeout(timeout))) {

       ..

   } catch (ExecutionException e) {
       switch (e.getCause()) {
           case CancelledByTimeoutException ->
           case IOException ioe -> ..
           default -> ..
       }
   }

В других случаях перехватывать исключение может быть нецелесообразно: его можно передать настроенному обработчику неперехваченных исключений для журналирования.

Если конкретное исключение должно приводить к использованию результата по умолчанию, может быть целесообразнее обработать его в самой подзадаче, а не допускать сбой подзадачи с последующей обработкой исключения владельцем области.

Метод join() выбрасывает InterruptedException, если поток прерван до или во время ожидания в методе join(). Раздел Thread Interruption спецификации Thread содержит рекомендации по обработке этого исключения.

Наследование привязок ограниченных значений

При использовании вместе с StructuredTaskScope, ScopedValue также может безопасно и эффективно передавать значение методам, выполняемым подзадачами, запущенными в области. Когда объект ScopedValue привязан к значению в потоке, выполняющем главную задачу, эта привязка наследуется потоками, созданными для выполнения подзадач. Поток, выполняющий главную задачу, не продолжает выполнение за пределами метода close(), пока не завершатся все потоки, выполняющие подзадачи. Это гарантирует, что ScopedValue не станет непривязанным (или не вернётся к предыдущему значению), пока выполняются подзадачи. Помимо безопасного и эффективного способа наследования значения подзадачами, это наследование позволяет преобразовать последовательный код, использующий ScopedValue, в код со структурированной конкурентностью.

Для обеспечения корректности при открытии новой StructuredTaskScope сохраняются привязки ограниченных значений текущего потока. Именно эти привязки ограниченных значений наследуются потоками, созданными для выполнения подзадач в области. При запуске подзадачи проверяется, совпадают ли действующие в этот момент привязки с привязками на момент создания StructuredTaskScope. Эта проверка гарантирует, что подзадача не унаследует привязку, которая будет отменена в главной задаче до завершения подзадачи.

ScopedValue, совместно используемое потоками, требует, чтобы значение было неизменяемым объектом или чтобы все обращения к нему были должным образом синхронизированы.

В следующем примере демонстрируется наследование привязок ограниченных значений. Ограниченное значение USERNAME привязано к значению "duke" на ограниченный период выполнения лямбда-выражения потоком, который его выполняет. Код в блоке открывает StructuredTaskScope и запускает две подзадачи, а затем ожидает в методе join() и объединяет результаты обеих подзадач. Если код, выполняемый потоками, в которых запущены subtask1 и subtask2, использует ScopedValue.get() для получения значения USERNAME, то будет возвращено значение "duke".

    private static final ScopedValue<String> USERNAME = ScopedValue.newInstance();

    MyResult result = ScopedValue.where(USERNAME, "duke").call(() -> {

        try (var scope = StructuredTaskScope.open()) {

            Subtask<String> subtask1 = scope.fork( .. );    // inherits binding
            Subtask<Integer> subtask2 = scope.fork( .. );   // inherits binding

            scope.join();
            return new MyResult(subtask1.get(), subtask2.get());
        }

    });

Ограниченное значение, унаследованное подзадачей, может быть повторно привязано к новому значению в подзадаче на время выполнения некоторого метода в ней. После завершения метода значение ScopedValue возвращается к предыдущему значению — значению, унаследованному от потока, выполняющего главную задачу.

Подзадача может выполнять код, который сам открывает новую StructuredTaskScope. Главная задача, выполняющаяся в потоке T1, открывает StructuredTaskScope и запускает подзадачу, выполняющуюся в потоке T2. Привязки ограниченных значений, сохранённые при открытии области потоком T1, наследуются потоком T2. Подзадача (в потоке T2) выполняет код, который открывает новую StructuredTaskScope и запускает (под-)подзадачу, выполняющуюся в потоке T3. Привязки ограниченных значений, сохранённые при открытии области потоком T2, наследуются потоком T3. В их число входят привязки, унаследованные от T1. Таким образом, ограниченные значения наследуются по дереву подзадач, а не только на один уровень вложенности.

Гарантии согласованности памяти

Действия в потоке-владельце StructuredTaskScope, выполненные до запуска подзадачи, происходят-перед любыми действиями потока, выполняющего подзадачу; в свою очередь, они происходят-перед действиями любого потока, который успешно получает результат подзадачи с помощью Subtask.get()^PREVIEW или Subtask.exception()^PREVIEW. Если результат подзадачи влияет на значение или исключение, возвращаемое методом join(), то любые действия потока, выполняющего эту подзадачу, происходят-перед возвратом потока-владельца из join() с результатом.

Общие исключения

Если не указано иное, передача аргумента null методу этого класса приводит к выбрасыванию NullPointerException.