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

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

В качестве первого примера рассмотрим задачу, которая разделяется на две подзадачи для одновременного получения ресурсов из двух URL-адресов — «left» и «right». Обе подзадачи могут завершиться успешно, одна подзадача может завершиться успешно, а другая — неудачно, или обе подзадачи могут завершиться неудачно. В этом примере задаче нужны успешные результаты обеих подзадач. Метод join ожидает успешного завершения обеих подзадач или сбоя любой из них.

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

       Subtask<String> subtask1 = scope.fork(() -> query(left));
       Subtask<Integer> subtask2 = scope.fork(() -> query(right));

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

       // both subtasks completed successfully
       return new MyResult(subtask1.get(), subtask2.get());

   } // close

Если обе подзадачи завершаются успешно, метод join завершается нормально, и задача использует метод Subtask.get()^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ}, чтобы получить результат каждой подзадачи. Если одна из подзадач завершается неудачно, другая подзадача отменяется (это приведёт к прерыванию потока, выполняющего другую подзадачу), а метод join выбрасывает StructuredTaskScope.FailedException^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ}, указав исключение сбойной подзадачи в качестве причины.

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

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

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

В приведённом выше примере задача завершается неудачно, если сбой происходит в любой подзадаче. Если все подзадачи завершаются успешно, метод join завершается нормально. Другие политики и исходы поддерживаются с помощью создания StructuredTaskScope с объектом StructuredTaskScope.Joiner^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ}, реализующим нужную политику. Joiner обрабатывает завершение подзадач и формирует исход для метода join. В приведённом выше примере join возвращает null. В зависимости от Joiner, join может возвращать результат, поток элементов или другой объект. Интерфейс Joiner определяет фабричные методы для создания Joiner в некоторых распространённых случаях.

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

Рассмотрим другой пример, в котором задача разделяется на две подзадачи. В этом примере каждая подзадача создаёт результат типа String, и задаче нужен только результат первой успешно завершившейся подзадачи. В примере используется Joiner.anySuccessfulResultOrThrow()^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ} для создания Joiner, предоставляющего результат первой успешно завершившейся подзадачи. Параметр типа в примере — «String», поэтому можно запускать только подзадачи, возвращающие String.

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

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

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

   }

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

Использовать ли кодом значение Subtask, возвращённое из fork, зависит от Joiner и сценария использования. Некоторые реализации Joiner подходят для подзадач, возвращающих результаты одного типа, и случаев, когда метод join возвращает результат, используемый задачей. Код, запускающий подзадачи с результатами разных типов и использующий Joiner, например Joiner.awaitAllSuccessfulOrThrow(), который не возвращает результат, после объединения использует Subtask.get()^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ}.

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

StructuredTaskScope открывается с объектом Joiner^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ}, который обрабатывает завершение подзадач и формирует исход для метода join. В некоторых случаях исходом будет результат, в других — исключение. Если исходом является исключение, метод join выбрасывает StructuredTaskScope.FailedException^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ}, указав исключение в качестве причины. Во многих реализациях Joiner это будет исключение, выброшенное сбойной подзадачей. Например, в случае allSuccessfulOrThrow^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ} и awaitAllSuccessfulOrThrow^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ} исключение выбрасывает первая подзадача, завершившаяся неудачно.

Многие детали обработки исключений зависят от сценария использования. В некоторых случаях полезно добавить блок catch в оператор try-with-resources, чтобы перехватывать FailedException. Для обработки определённых причин в обработке исключений может использоваться instanceof с сопоставлением с образцом.

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

       ..

   } catch (StructuredTaskScope.FailedException e) {

       Throwable cause = e.getCause();
       switch (cause) {
           case IOException ioe -> ..
           default -> ..
       }

   }

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

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

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

Метод open с двумя аргументами можно использовать для создания StructuredTaskScope, который использует другой ThreadFactory, имеет имя для мониторинга и управления или задаёт время ожидания, отменяющее область, если срок истекает до ожидания завершения подзадач или во время него. Метод open вызывается с функцией, которая применяется к конфигурации по умолчанию и возвращает Configuration^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ} для создаваемой StructuredTaskScope.

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

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

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

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

        scope.join();

     }

Во втором примере задаётся время ожидания, представленное объектом Duration. Отсчёт времени ожидания начинается при открытии новой области. Если срок истекает до завершения метода join, область отменяется. Это прерывает потоки, выполняющие две подзадачи, и приводит к тому, что метод join выбрасывает StructuredTaskScope.TimeoutException^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ}.

   Duration timeout = Duration.ofSeconds(10);

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

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

       List<String> result = scope.join()
                                  .map(Subtask::get)
                                  .toList();

  }

Наследование привязок значений области видимости

При совместном использовании с 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, предшествующие запуску подзадачи, происходят до любых действий, выполняемых этой подзадачей; в свою очередь, они происходят до получения^{ПРЕДВАРИТЕЛЬНАЯ ВЕРСИЯ} результата подзадачи.

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

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