JEP 453: Structured Concurrency (Preview)

Аннотация

Упростить конкурентное программирование, добавив API для Structured Concurrency. Structured Concurrency рассматривает группы связанных задач, выполняемых в разных потоках, как единую единицу работы. Это упрощает обработку ошибок и отмену, повышает надёжность и улучшает наблюдаемость. Это API в статусе Preview.

История

Structured Concurrency был предложен в JEP 428 и поставлен в JDK 19 как API в статусе Incubator (инкубационный модуль). Повторно в статусе Incubator он был выпущен в JEP 437 в JDK 20 с небольшим обновлением — наследованием Scoped Values (значения с ограниченной областью видимости), JEP 429.

Здесь мы предлагаем сделать Structured Concurrency Preview-API в пакете java.util.concurrent. Единственное существенное изменение состоит в том, что метод StructuredTaskScope::fork(...) возвращает Subtask, а не Future, как описано ниже.

Цели

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

• Улучшить наблюдаемость конкурентного кода.

Что не является целью

• Цель не в том, чтобы заменить какие-либо конструкции конкурентности из пакета java.util.concurrent, такие как ExecutorService и Future.

• Цель не в том, чтобы определить окончательный API Structured Concurrency для платформы Java. Другие конструкции Structured Concurrency могут быть определены сторонними библиотеками или в будущих выпусках JDK.

• Цель не в том, чтобы определить способ обмена потоками данных между потоками (т. е. каналы). Возможно, мы предложим это в будущем.

• Цель не в том, чтобы заменить существующий механизм прерывания потоков новым механизмом отмены потоков. Возможно, мы предложим это в будущем.

Мотивация

Разработчики справляются со сложностью, разбивая задачи на несколько подзадач. В обычном однопоточном коде подзадачи выполняются последовательно. Однако если подзадачи достаточно независимы друг от друга и если аппаратных ресурсов достаточно, то всю задачу можно выполнить быстрее (т. е. с меньшей задержкой), выполняя подзадачи конкурентно. Например, задача, которая объединяет результаты нескольких операций ввода-вывода, будет выполняться быстрее, если каждая операция ввода-вывода выполняется конкурентно в собственном потоке. Virtual Threads (виртуальные потоки), JEP 444, делают экономически оправданным выделение отдельного потока под каждую такую операцию ввода-вывода, но управлять огромным числом потоков, которое может в результате появиться, по-прежнему сложно.

Неструктурированная конкурентность с ExecutorService

API java.util.concurrent.ExecutorService, появившийся в Java 5, помогает разработчикам выполнять подзадачи конкурентно.

Например, вот метод handle(), который представляет задачу в серверном приложении. Он обрабатывает входящий запрос, передавая две подзадачи в ExecutorService. Одна подзадача выполняет метод findUser(), а другая — метод fetchOrder(). ExecutorService сразу возвращает Future для каждой подзадачи и выполняет подзадачи конкурентно в соответствии с политикой планирования Executor. Метод handle() ожидает результатов подзадач с помощью блокирующих вызовов методов get() их объектов future, поэтому говорят, что задача присоединяет свои подзадачи.

Response handle() throws ExecutionException, InterruptedException {
    Future<String>  user  = esvc.submit(() -> findUser());
    Future<Integer> order = esvc.submit(() -> fetchOrder());
    String theUser  = user.get();   // Join findUser
    int    theOrder = order.get();  // Join fetchOrder
    return new Response(theUser, theOrder);
}

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

• Если findUser() выбрасывает исключение, то handle() выбросит исключение при вызове user.get(), но fetchOrder() продолжит выполняться в собственном потоке. Это утечка потоков, которая в лучшем случае впустую расходует ресурсы; в худшем случае поток fetchOrder() будет мешать другим задачам.

• Если поток, выполняющий handle(), прерывается, прерывание не распространяется на подзадачи. Оба потока, findUser() и fetchOrder(), утекут и продолжат выполняться даже после сбоя handle().

• Если findUser() выполняется долго, а fetchOrder() тем временем завершается сбоем, то handle() будет без необходимости ждать findUser(), блокируясь на user.get(), вместо того чтобы отменить эту подзадачу. Только после того как findUser() завершится и user.get() вернёт управление, order.get() выбросит исключение, из-за чего handle() завершится сбоем.

В каждом случае проблема в том, что логически наша программа структурирована отношениями «задача — подзадача», но эти отношения существуют только в голове разработчика.

Из-за этого не только больше возможностей для ошибок, но и сложнее диагностировать и устранять такие ошибки. Например, средства наблюдения, такие как дампы потоков, покажут handle(), findUser() и fetchOrder() в стеках вызовов не связанных между собой потоков, без какого-либо указания на отношение «задача — подзадача».

Можно попытаться сделать лучше, явно отменяя другие подзадачи при возникновении ошибки, например оборачивая задачи в try-finally и вызывая методы cancel(boolean) объектов future других задач в блоке catch для задачи, завершившейся сбоем. Нам также понадобилось бы использовать ExecutorService внутри оператора try-with-resources, как показано в примерах в JEP 425, потому что Future не даёт способа дождаться задачи, которая была отменена. Но всё это бывает очень сложно сделать правильно, и из-за этого логический замысел кода часто становится труднее понять. Отслеживать отношения между задачами и вручную добавлять обратно необходимые связи для отмены между задачами — слишком большое требование к разработчикам.

Необходимость вручную координировать время жизни связана с тем, что ExecutorService и Future допускают неограниченные схемы конкурентности. На участвующие потоки не накладывается никаких ограничений и никакого порядка. Один поток может создать ExecutorService, второй поток может передать ему работу, а потоки, которые выполняют эту работу, никак не связаны ни с первым, ни со вторым потоком. Более того, после того как поток передал работу, результатов выполнения может ожидать совершенно другой поток. Любой код со ссылкой на Future может присоединить его (т. е. дождаться его результата, вызвав get()), даже код в потоке, отличном от того, который получил Future. По сути, подзадача, запущенная одной задачей, не обязана возвращаться к задаче, которая её передала. Она может вернуться к любой из множества задач — или вообще ни к какой.

Поскольку ExecutorService и Future допускают такое неструктурированное использование, они не обеспечивают и даже не отслеживают отношения между задачами и подзадачами, хотя такие отношения распространены и полезны. Поэтому, даже когда подзадачи передаются и присоединяются в одной и той же задаче, сбой одной подзадачи не может автоматически вызвать отмену другой: в приведённом выше методе handle() сбой fetchOrder() не может автоматически вызвать отмену findUser(). Объект future для fetchOrder() не связан с объектом future для findUser(), и ни один из них не связан с потоком, который в конечном счёте присоединит его через метод get(). Вместо того чтобы просить разработчиков управлять такой отменой вручную, мы хотим надёжно её автоматизировать.

Структура задач должна отражать структуру кода

В отличие от вольного набора потоков при использовании ExecutorService, выполнение однопоточного кода всегда обеспечивает иерархию задач и подзадач. Блок тела метода {...} соответствует задаче, а методы, вызываемые внутри блока, соответствуют подзадачам. Вызванный метод должен либо вернуть управление вызвавшему его методу, либо выбросить в него исключение. Он не может пережить вызвавший его метод, а также не может вернуть управление или выбросить исключение в другой метод. Таким образом, все подзадачи завершаются раньше задачи, каждая подзадача является дочерней по отношению к своей родительской задаче, а время жизни каждой подзадачи относительно других подзадач и задачи определяется синтаксической блочной структурой кода.

Например, в этой однопоточной версии handle() отношение «задача — подзадача» очевидно из синтаксической структуры:

Response handle() throws IOException {
    String theUser  = findUser();
    int    theOrder = fetchOrder();
    return new Response(theUser, theOrder);
}

Мы не запускаем подзадачу fetchOrder(), пока не завершится подзадача findUser(), успешно или сбоем. Если findUser() завершается сбоем, мы вообще не запускаем fetchOrder(), и задача handle() неявно завершается сбоем. Важно то, что подзадача может вернуться только к своей родительской задаче: это означает, что родительская задача может неявно считать сбой одной подзадачи сигналом отменить другие незавершённые подзадачи, а затем сама завершиться сбоем.

В однопоточном коде иерархия «задача — подзадача» материализуется в стеке вызовов во время выполнения. Таким образом, мы бесплатно получаем соответствующие отношения «родитель — потомок», которые управляют распространением ошибок. При наблюдении за одним потоком иерархическое отношение очевидно: findUser() (а затем fetchOrder()) выглядят подчинёнными handle(). Это позволяет легко ответить на вопрос: «над чем сейчас работает handle()?»

Конкурентное программирование было бы проще, надёжнее и удобнее для наблюдения, если бы отношения «родитель — потомок» между задачами и их подзадачами были видны из синтаксической структуры кода, а также материализовались во время выполнения — так же, как в однопоточном коде. Синтаксическая структура очерчивала бы время жизни подзадач и позволяла бы представить во время выполнения иерархию между потоками, аналогичную стеку вызовов внутри потока. Такое представление обеспечило бы распространение ошибок и отмену, а также осмысленное наблюдение за конкурентной программой.

(На платформе Java уже есть API для наложения структуры на конкурентные задачи, а именно java.util.concurrent.ForkJoinPool — механизм выполнения, лежащий в основе параллельных потоков данных (parallel streams). Однако этот API рассчитан на задачи с интенсивными вычислениями, а не на задачи, связанные с вводом-выводом.)

Structured Concurrency

Structured Concurrency — это подход к конкурентному программированию, сохраняющий естественную связь между задачами и подзадачами, благодаря чему конкурентный код становится более читаемым, сопровождаемым и надёжным. Термин «Structured Concurrency» ввёл Martin Sústrik, а популяризировал Nathaniel J. Smith. На проектирование обработки ошибок в Structured Concurrency повлияли идеи из других языков, например иерархические супервизоры Erlang.

Structured Concurrency исходит из простого принципа:

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

В Structured Concurrency подзадачи работают от имени задачи. Задача ожидает результатов подзадач и отслеживает их сбои. Как и в приёмах структурного программирования для кода в одном потоке, сила Structured Concurrency для нескольких потоков основана на двух идеях: (1) чётко определённые точки входа и выхода для хода выполнения через блок кода и (2) строгая вложенность времени жизни операций, отражающая их синтаксическую вложенность в коде.

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

Structured Concurrency отлично сочетается с Virtual Threads — легковесными потоками, реализованными в JDK. Многие потоки Virtual Threads используют один и тот же поток операционной системы, что позволяет иметь очень большое количество потоков Virtual Threads. Помимо того что потоки Virtual Threads многочисленны, они достаточно дёшевы, чтобы представлять любую конкурентную единицу поведения, даже поведение, связанное с вводом-выводом. Это означает, что серверное приложение может использовать Structured Concurrency для одновременной обработки тысяч или миллионов входящих запросов: оно может выделять новый поток Virtual Threads под задачу обработки каждого запроса, а когда задача разветвляется, передавая подзадачи на конкурентное выполнение, оно может выделять новый поток Virtual Threads под каждую подзадачу. За кулисами отношение «задача — подзадача» материализуется в дерево за счёт того, что каждый поток Virtual Threads хранит ссылку на своего единственного родителя, подобно тому как кадр стека вызовов ссылается на своего единственного вызывающего.

Итак, Virtual Threads дают потоки в изобилии. Structured Concurrency может правильно и надёжно координировать их и позволяет инструментам наблюдения отображать потоки так, как их понимает разработчик. Наличие API Structured Concurrency в JDK упростит создание удобных в сопровождении, надёжных и наблюдаемых серверных приложений.

Описание

Основной класс API Structured Concurrency — StructuredTaskScope в пакете java.util.concurrent. Этот класс позволяет разработчикам структурировать задачу как семейство конкурентных подзадач и координировать их как единое целое. Подзадачи выполняются в собственных потоках: их по отдельности ответвляют, а затем присоединяют как единое целое и, возможно, отменяют как единое целое. Успешные результаты или исключения подзадач собираются и обрабатываются родительской задачей. StructuredTaskScope ограничивает время жизни подзадач чёткой лексической областью видимости, в которой происходит всё взаимодействие задачи с её подзадачами — ответвление, присоединение, отмена, обработка ошибок и объединение результатов.

Вот приведённый ранее пример handle(), переписанный с использованием StructuredTaskScope (ShutdownOnFailure объясняется ниже):

Response handle() throws ExecutionException, InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        Supplier<String>  user  = scope.fork(() -> findUser());
        Supplier<Integer> order = scope.fork(() -> fetchOrder());

        scope.join()            // Join both subtasks
             .throwIfFailed();  // ... and propagate errors

        // Here, both subtasks have succeeded, so compose their results
        return new Response(user.get(), order.get());
    }
}

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

• Обработка ошибок с коротким замыканием — если любая из подзадач, findUser() или fetchOrder(), завершается сбоем, другая отменяется, если она ещё не завершилась. (Этим управляет политика завершения, реализованная в ShutdownOnFailure; возможны и другие политики.)

• Распространение отмены — если поток, выполняющий handle(), прерывается до или во время вызова join(), обе подзадачи автоматически отменяются, когда поток выходит из области видимости.

• Ясность — у приведённого кода чёткая структура: подготовить подзадачи, дождаться, пока они завершатся или будут отменены, а затем решить, завершиться ли успешно (и обработать результаты дочерних задач, которые уже закончены) или со сбоем (подзадачи уже закончены, так что очищать больше нечего).

• Наблюдаемость — дамп потоков, описанный ниже, наглядно показывает иерархию задач: потоки, выполняющие findUser() и fetchOrder(), отображаются как дочерние элементы области видимости.

StructuredTaskScope — это API в статусе Preview, по умолчанию отключённый

Чтобы использовать API StructuredTaskScope, нужно включить Preview-API следующим образом:

• Скомпилируйте программу с javac --release 21 --enable-preview Main.java и запускайте её с java --enable-preview Main; или

• При использовании средства запуска исходного кода запускайте программу с java --source 21 --enable-preview Main.java; или

• При использовании jshell запускайте его с jshell --enable-preview.

Использование StructuredTaskScope

API StructuredTaskScope таков:

public class StructuredTaskScope<T> implements AutoCloseable {

    public <U extends T> Subtask<U> fork(Callable<? extends U> task);
    public void shutdown();

    public StructuredTaskScope<T> join() throws InterruptedException;
    public StructuredTaskScope<T> joinUntil(Instant deadline)
        throws InterruptedException, TimeoutException;
    public void close();

    protected void handleComplete(Subtask<? extends T> handle);
    protected final void ensureOwnerAndJoined();

}

Общий порядок работы кода с StructuredTaskScope такой:

1. Создайте область видимости. Поток, который создаёт область видимости, является её владельцем.

2. Используйте метод fork(Callable), чтобы ответвлять подзадачи в этой области видимости.

3. В любой момент любая из подзадач или владелец области видимости может вызвать метод shutdown() этой области видимости, чтобы отменить незавершённые подзадачи и запретить ответвление новых подзадач.

4. Владелец области присоединяется к области (join), то есть ко всем её подзадачам, как к единому целому. Владелец может вызвать метод области join(), чтобы дождаться, пока все подзадачи либо завершатся (успешно или нет), либо будут отменены через shutdown(). Либо же он может вызвать метод области joinUntil(java.time.Instant), чтобы ждать не дольше заданного крайнего срока.

5. После join обработайте ошибки в подзадачах и обработайте их результаты.

6. Закройте область, обычно неявно через try-with-resources. При этом область завершается, если она ещё не завершена, и ожидается завершение всех подзадач, которые были отменены, но ещё не завершились.

Каждый вызов fork(...) запускает новый поток для выполнения подзадачи; по умолчанию это поток Virtual Threads. Подзадача может создать собственную вложенную StructuredTaskScope, чтобы порождать свои подзадачи, и так образуется иерархия. Эта иерархия отражена в блочной структуре кода, которая ограничивает время жизни подзадач: гарантируется, что после закрытия области все потоки подзадач завершены, и после выхода из блока не остаётся ни одного потока.

Любая подзадача в области, любые подзадачи второго уровня во вложенной области и владелец области могут в любой момент вызвать метод области shutdown(), чтобы обозначить, что задача завершена, — даже пока другие подзадачи ещё выполняются. Метод shutdown() прерывает потоки, которые ещё выполняют подзадачи, и заставляет метод join() или joinUntil(Instant) вернуть управление. Поэтому все подзадачи следует писать так, чтобы они реагировали на прерывание. Новая подзадача, порождённая после вызова shutdown(), будет в состоянии UNAVAILABLE и не будет запущена. По сути, shutdown() — это конкурентный аналог оператора break в последовательном коде.

Вызов join() или joinUntil(Instant) внутри области обязателен. Если выход из блока области происходит до join, область дождётся завершения всех подзадач, а затем выбросит исключение.

Поток-владелец области может быть прерван либо до join, либо во время него. Например, он может быть подзадачей объемлющей области, которая была завершена. В этом случае join() и joinUntil(Instant) выбросят исключение, поскольку продолжать нет смысла. Затем оператор try-with-resources завершит область, что отменит все подзадачи и дождётся их завершения. В результате отмена задачи автоматически распространяется на её подзадачи. Если крайний срок метода joinUntil(Instant) истекает раньше, чем подзадачи завершатся или будет вызван shutdown(), метод выбросит исключение, и оператор try-with-resources снова завершит область.

Когда join() завершается успешно, каждая из подзадач либо завершилась успешно, либо завершилась с ошибкой, либо была отменена, потому что область была завершена.

После join владелец области обрабатывает подзадачи, завершившиеся с ошибкой, и обрабатывает результаты подзадач, завершившихся успешно; обычно это делает политика завершения (см. ниже. Результат успешно завершившейся задачи можно получить методом Subtask.get(). Метод get() никогда не блокируется; он выбрасывает IllegalStateException, если его по ошибке вызвали до join или если подзадача не завершилась успешно.

Подзадачи, порождённые в области, наследуют привязки ScopedValue (JEP 446). Если владелец области читает значение из привязанного ScopedValue, то каждая подзадача прочитает то же значение.

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

Структурированное использование StructuredTaskScope обеспечивается во время выполнения. Например, попытки вызвать fork(Callable) из потока, который не входит в древовидную иерархию области — то есть не является владельцем, подзадачей или подзадачей во вложенных областях (подзадачей второго уровня), — завершатся исключением. Использование области вне блока try-with-resources и возврат без вызова close() или без соблюдения правильной вложенности вызовов close() могут привести к тому, что методы области выбросят StructureViolationException.

StructuredTaskScope навязывает конкурентным операциям структуру и порядок. Поэтому он не реализует интерфейсы ExecutorService и Executor, так как экземпляры этих интерфейсов обычно используются неструктурированно (см. ниже). Однако код, который использует ExecutorService, но выиграл бы от структуры, несложно перевести на StructuredTaskScope.

На практике в большинстве случаев StructuredTaskScope будет использоваться не через класс StructuredTaskScope напрямую, а через один из двух подклассов, описанных в следующем разделе, которые реализуют политики завершения. В других сценариях пользователи, скорее всего, будут писать собственные подклассы для реализации своих политик завершения.

Политики завершения

При работе с конкурентными подзадачами часто используют шаблоны с коротким замыканием, чтобы не выполнять лишнюю работу. Например, иногда имеет смысл отменить все подзадачи, если одна из них завершилась с ошибкой (то есть invoke all), или, наоборот, если одна из них завершилась успешно (то есть invoke any). Два подкласса StructuredTaskScope, ShutdownOnFailure и ShutdownOnSuccess, поддерживают эти шаблоны с помощью политик, которые завершают область, когда первая подзадача завершается с ошибкой или успешно соответственно.

Кроме того, политики завершения предоставляют централизованные методы для обработки исключений и, возможно, успешных результатов. Это соответствует духу Structured Concurrency, согласно которому вся область рассматривается как единое целое.

Вот StructuredTaskScope с политикой завершения при ошибке (она также используется в примере handle() выше), который конкурентно выполняет набор задач и завершается с ошибкой, если хотя бы одна из них завершилась с ошибкой:

<T> List<T> runAll(List<Callable<T>> tasks) 
        throws InterruptedException, ExecutionException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        List<? extends Supplier<T>> suppliers = tasks.stream().map(scope::fork).toList();
        scope.join()
             .throwIfFailed();  // Propagate exception if any subtask fails
        // Here, all tasks have succeeded, so compose their results
        return suppliers.stream().map(Supplier::get).toList();
    }
}

Вот StructuredTaskScope с политикой завершения при успехе, который возвращает результат первой успешной подзадачи:

<T> T race(List<Callable<T>> tasks, Instant deadline) 
        throws InterruptedException, ExecutionException, TimeoutException {
    try (var scope = new StructuredTaskScope.ShutdownOnSuccess<T>()) {
        for (var task : tasks) {
            scope.fork(task);
        }
        return scope.joinUntil(deadline)
                    .result();  // Throws if none of the subtasks completed successfully
    }
}

Как только одна подзадача завершается успешно, эта область автоматически завершается, отменяя незавершённые подзадачи. Задача завершается с ошибкой, если все подзадачи завершились с ошибкой или если истёк заданный крайний срок. Этот шаблон может пригодиться, например, в серверных приложениях, которым нужен результат от любого из набора дублирующих друг друга сервисов.

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

Обработка результатов

После join и централизованной обработки исключений политикой завершения (например, с помощью ShutdownOnFailure::throwIfFailed) владелец области может обработать результаты подзадач, используя объекты Subtask, возвращённые вызовами fork(...), если эти результаты не обрабатывает сама политика (например, ShutdownOnSuccess::result()).

Как правило, единственный метод Subtask, который вызывает владелец области, — это метод get(). Все остальные методы Subtask обычно используются только в реализации метода handleComplete(...) собственных политик завершения (см. ниже). Более того, мы рекомендуем объявлять переменные, ссылающиеся на Subtask, возвращённый fork(...), с типом, например, Supplier<String>, а не Subtask<String> (если, конечно, вы не решили использовать var). Если политика завершения сама обрабатывает результаты подзадач — как в случае ShutdownOnSuccess, — то объектов Subtask, возвращаемых fork(...), следует вообще избегать, а метод fork(...) рассматривать так, будто он возвращает void. Подзадачи должны возвращать в качестве результата всю информацию, которую владелец области должен обработать после централизованной обработки исключений политикой.

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

<T> List<Future<T>> executeAll(List<Callable<T>> tasks)
        throws InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    	  List<? extends Supplier<Future<T>>> futures = tasks.stream()
    	      .map(task -> asFuture(task))
     	      .map(scope::fork)
     	      .toList();
    	  scope.join();
    	  return futures.stream().map(Supplier::get).toList();
    }
}

static <T> Callable<Future<T>> asFuture(Callable<T> task) {
   return () -> {
       try {
           return CompletableFuture.completedFuture(task.call());
       } catch (Exception ex) {
           return CompletableFuture.failedFuture(ex);
       }
   };
}

Собственные политики завершения

Можно расширить StructuredTaskScope и переопределить его защищённый метод handleComplete(...), чтобы реализовать политики, отличные от политик ShutdownOnSuccess и ShutdownOnFailure. Подкласс может, например:

• собирать результаты подзадач, завершившихся успешно, и игнорировать подзадачи, завершившиеся с ошибкой,
• собирать исключения, когда подзадачи завершаются с ошибкой, или
• вызывать метод shutdown(), чтобы завершить область и пробудить join(), когда возникает некоторое условие.

Когда подзадача завершается, даже после вызова shutdown(), она передаётся методу handleComplete(...) как Subtask:

public sealed interface Subtask<T> extends Supplier<T> {
    enum State { SUCCESS, FAILED, UNAVAILABLE }

    State state();
    Callable<? extends T> task();
    T get();
    Throwable exception();
}

Метод handleComplete(...) вызывается для подзадач, которые завершились успешно (состояние SUCCESS) или неуспешно (состояние FAILED) до вызова shutdown(). Метод get() можно вызывать, только если подзадача находится в состоянии SUCCESS, а метод exception() — только если подзадача находится в состоянии FAILED; вызов get() или exception() в других ситуациях приведёт к тому, что они выбросят IllegalStateException. Состояние UNAVAILABLE означает одно из следующего: (1) подзадача была порождена, но ещё не завершилась; (2) подзадача завершилась после завершения области или (3) подзадача была порождена после завершения области и поэтому не запускалась. Метод handleComplete(...) никогда не вызывается для подзадачи в состоянии UNAVAILABLE.

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

Вот пример подкласса StructuredTaskScope, который собирает результаты подзадач, завершившихся успешно. Он определяет метод results(), который основная задача использует для получения результатов.

class MyScope<T> extends StructuredTaskScope<T> {

    private final Queue<T> results = new ConcurrentLinkedQueue<>();

    MyScope() { super(null, Thread.ofVirtual().factory()); }

    @Override
    protected void handleComplete(Subtask<? extends T> subtask) {
        if (subtask.state() == Subtask.State.SUCCESS)
            results.add(subtask.get());
    }

    @Override
    public MyScope<T> join() throws InterruptedException {
        super.join();
        return this;
    }

    // Returns a stream of results from the subtasks that completed successfully
    public Stream<T> results() {
        super.ensureOwnerAndJoined();
        return results.stream();
    }

}

Эту собственную политику можно использовать так:

<T> List<T> allSuccessful(List<Callable<T>> tasks) throws InterruptedException {
    try (var scope = new MyScope<T>()) {
        for (var task : tasks) scope.fork(task);
        return scope.join()
                    .results().toList();
    }
}

Сценарии fan-in

Примеры выше были посвящены сценариям fan-out, в которых обслуживается несколько конкурентных исходящих операций ввода-вывода. StructuredTaskScope полезен и в сценариях fan-in, в которых обслуживается несколько конкурентных входящих операций ввода-вывода. В таких сценариях мы обычно создаём заранее неизвестное число подзадач в ответ на входящие запросы.

Вот пример сервера, который порождает подзадачи для обработки входящих соединений внутри StructuredTaskScope:

void serve(ServerSocket serverSocket) throws IOException, InterruptedException {
    try (var scope = new StructuredTaskScope<Void>()) {
        try {
            while (true) {
                var socket = serverSocket.accept();
                scope.fork(() -> handle(socket));
            }
        } finally {
            // If there's been an error or we're interrupted, we stop accepting
            scope.shutdown();  // Close all active connections
            scope.join();
        }
    }
}

С точки зрения конкурентности этот сценарий отличается не столько направлением запросов, сколько длительностью и количеством задач. Здесь, в отличие от предыдущих примеров, время работы владельца области не ограничено — он остановится, только когда его прервут. Число подзадач тоже неизвестно, поскольку они порождаются динамически в ответ на внешние события.

Все подзадачи, обрабатывающие соединения, создаются внутри области, поэтому их назначение легко увидеть в дампе потоков, где они будут показаны как дочерние по отношению к владельцу области. Кроме того, весь сервис легко завершить как единое целое.

Наблюдаемость

Мы расширяем новый формат дампа потоков в JSON, добавленный в JEP 444, чтобы показать группировку потоков в иерархию, которую задаёт StructuredTaskScope:

$ jcmd <pid> Thread.dump_to_file -format=json <file>

JSON-объект каждой области содержит массив потоков, порождённых в этой области, вместе с их трассировками стека. Поток-владелец области обычно заблокирован в методе join и ждёт завершения подзадач. Дамп потоков показывает древовидную иерархию, которую задаёт Structured Concurrency, поэтому по нему легко увидеть, что делают потоки подзадач. В JSON-объекте области есть и ссылка на родительскую область, так что по дампу можно восстановить структуру программы.

API com.sun.management.HotSpotDiagnosticsMXBean также можно использовать для создания таких дампов потоков — напрямую или косвенно, через платформенный MBeanServer и локальный или удалённый инструмент JMX.

Почему fork(...) не возвращает Future?

Когда API StructuredTaskScope был в статусе Incubator, метод fork(...) возвращал Future. Это создавало ощущение привычности: fork(...) походил на существующий метод ExecutorService::submit. Однако StructuredTaskScope предназначен для использования не так, как ExecutorService, а структурно, как описано выше. Поэтому использование Future больше запутывало, чем проясняло.

• Привычный способ использовать Future — вызвать его метод get(), который блокирует выполнение, пока не будет доступен результат. Но в контексте StructuredTaskScope такое использование Future не просто не рекомендуется, а контрпродуктивно. Структурированные объекты Future следует опрашивать только после возврата из join(), когда уже известно, что они завершены или отменены. И использовать при этом следует не привычный метод get(), а новый метод resultNow(), который никогда не блокирует выполнение.

• Некоторые разработчики спрашивали, почему fork(...) не возвращает вместо этого более функциональные объекты CompletableFuture. Поскольку Future, который возвращает fork(...), следует использовать только тогда, когда уже известно, что он завершён, CompletableFuture не дал бы никакого преимущества: его расширенные возможности полезны только для незавершённых future-объектов. Кроме того, CompletableFuture рассчитан на парадигму асинхронного программирования, а StructuredTaskScope поощряет блокирующую парадигму.

  Словом, Future и CompletableFuture спроектированы так, чтобы давать степени свободы, которые в Structured Concurrency контрпродуктивны.

• Суть Structured Concurrency в том, чтобы рассматривать несколько задач, выполняемых в разных потоках, как единую единицу работы, тогда как Future полезен в основном тогда, когда несколько задач рассматриваются как отдельные задачи. Область должна блокироваться только один раз, чтобы дождаться результатов подзадач, а затем централизованно обработать исключения. Поэтому в подавляющем большинстве случаев единственным методом, который следовало вызывать у Future, возвращённого из fork(...), был resultNow(). Это заметно отличалось от обычного использования Future, и интерфейс Future лишь отвлекал от правильного использования в этом контексте.

В текущем API Subtask::get() ведёт себя в точности так же, как вёл себя Future::resultNow(), когда API был в статусе Incubator.

Альтернативы

• Улучшить интерфейс ExecutorService. Мы создали прототип реализации этого интерфейса, которая всегда обеспечивает структурность и ограничивает, какие потоки могут отправлять задачи. Однако мы сочли этот подход проблематичным, потому что большинство случаев использования ExecutorService (и его родительского интерфейса Executor) в JDK и в экосистеме не являются структурированными. Повторное использование того же API для гораздо более ограниченной концепции неизбежно приведёт к путанице. Например, передача структурированного экземпляра ExecutorService в существующие методы, принимающие этот тип, почти наверняка приводила бы к исключениям в большинстве ситуаций.