torch.futures

Создано: 12 июня 2025 | Последнее обновление: 12 июня 2025

Этот пакет предоставляет тип Future, инкапсулирующий асинхронное выполнение, а также набор вспомогательных функций для упрощения работы с объектами Future. В настоящее время тип Future используется главным образом в распределённой RPC-инфраструктуре.

class torch.futures.Future(*, devices=None)

Обёртка над torch._C.Future, инкапсулирующая асинхронное выполнение вызываемого объекта, например rpc_async(). Также предоставляет набор API для добавления функций обратного вызова и установки результатов.

Предупреждение

Поддержка GPU — это бета-функция, которая может измениться.

add_done_callback(callback) [исходный код]

Добавляет указанную функцию обратного вызова в этот Future. Она будет выполнена после завершения Future. К одному Future можно добавить несколько функций обратного вызова, однако порядок их выполнения не гарантируется. Функция обратного вызова должна принимать один аргумент — ссылку на этот Future. Чтобы получить значение, функция обратного вызова может использовать метод value(). Обратите внимание: если этот Future уже завершён, указанная функция обратного вызова будет выполнена сразу в текущем потоке.

Рекомендуем использовать метод then(), поскольку он позволяет синхронизироваться после завершения функции обратного вызова. add_done_callback может быть эффективнее, если функция обратного вызова ничего не возвращает. Однако и then(), и add_done_callback используют под капотом один и тот же API регистрации функций обратного вызова.

Для тензоров GPU этот метод работает так же, как then().

Параметры:

callback (Future) – функция обратного вызова Callable, принимающая один аргумент — ссылку на этот Future.

Примечание

Обратите внимание: если функция обратного вызова вызывает исключение — из-за того, что исходный future завершается с исключением и вызывает fut.wait(), либо из-за другого кода в функции обратного вызова, — обработку ошибки необходимо тщательно продумать. Например, если эта функция обратного вызова впоследствии завершает другие future, эти future не будут помечены как завершённые с ошибкой, и пользователь должен самостоятельно обработать их завершение и ожидание.

Пример:

>>> def callback(fut):
...     print("This will run after the future has finished.")
...     print(fut.wait())
>>> fut = torch.futures.Future()
>>> fut.add_done_callback(callback)
>>> fut.set_result(5)
This will run after the future has finished.
5

done() [исходный код]

Возвращает True, если этот Future завершён. Future считается завершённым, если у него есть результат или исключение.

Если значение содержит тензоры, расположенные на GPU, Future.done() вернёт True, даже если асинхронные ядра, заполняющие эти тензоры, ещё не завершили работу на устройстве, поскольку на этом этапе результат уже можно использовать при условии выполнения необходимых синхронизаций (см. wait()).

Тип возвращаемого значения:

bool

set_exception(result) [исходный код]

Устанавливает исключение для этого Future, помечая его Future как завершённый с ошибкой и вызывая все зарегистрированные функции обратного вызова. Обратите внимание: при вызове wait()/value() для этого Future установленное здесь исключение будет немедленно выброшено.

Параметры:

result (BaseException) – исключение для этого Future.

Пример:

>>> fut = torch.futures.Future()
>>> fut.set_exception(ValueError("foo"))
>>> fut.wait()
Traceback (most recent call last):
...
ValueError: foo

set_result(result) [исходный код]

Устанавливает результат для этого Future, помечая его Future как завершённый и вызывая все зарегистрированные функции обратного вызова. Обратите внимание: Future нельзя пометить как завершённый дважды.

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

Параметры:

result (object) – объект результата этого Future.

Пример:

>>> import threading
>>> import time
>>> def slow_set_future(fut, value):
...     time.sleep(0.5)
...     fut.set_result(value)
>>> fut = torch.futures.Future()
>>> t = threading.Thread(
...     target=slow_set_future,
...     args=(fut, torch.ones(2) * 3)
... )
>>> t.start()
>>> print(fut.wait())
tensor([3., 3.])
>>> t.join()

then(callback) [исходный код]

Добавляет указанную функцию обратного вызова в этот Future. Она будет выполнена после завершения Future. К одному Future можно добавить несколько функций обратного вызова, однако порядок их выполнения не гарантируется (чтобы задать определённый порядок, используйте цепочку вызовов: fut.then(cb1).then(cb2)). Функция обратного вызова должна принимать один аргумент — ссылку на этот Future. Чтобы получить значение, функция обратного вызова может использовать метод value(). Обратите внимание: если этот Future уже завершён, указанная функция обратного вызова будет немедленно выполнена в текущем потоке.

Если значение Future содержит тензоры, расположенные на GPU, функция обратного вызова может быть вызвана до того, как асинхронные ядра, заполняющие эти тензоры, завершат выполнение на устройстве. Однако функция обратного вызова будет вызвана при установленных в качестве текущих выделенных потоках (полученных из глобального пула), синхронизированных с этими ядрами. Поэтому любые операции, выполняемые функцией обратного вызова над этими тензорами, будут запланированы на устройстве после завершения ядер. Иными словами, функция обратного вызова может безопасно работать с результатом без дополнительной синхронизации, если она не переключает потоки. Это аналогично неблокирующему поведению метода wait().

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

Параметры:

callback (Callable) – функция обратного вызова Callable, принимающая этот Future в качестве единственного аргумента.

Возвращает:

Новый объект Future, содержащий возвращаемое значение callback и помечаемый как завершённый после завершения указанной callback.

Тип возвращаемого значения:

Future[S]

Примечание

Обратите внимание: если функция обратного вызова вызывает исключение — из-за того, что исходный future завершается с исключением и вызывает fut.wait(), либо из-за другого кода в функции обратного вызова, — future, возвращённый вызовом then, будет соответствующим образом помечен обнаруженной ошибкой. Однако если эта функция обратного вызова впоследствии завершает другие future, эти future не будут помечены как завершённые с ошибкой, и пользователь должен самостоятельно обработать их завершение и ожидание.

Пример:

>>> def callback(fut):
...     print(f"RPC return value is {fut.wait()}.")
>>> fut = torch.futures.Future()
>>> # The inserted callback will print the return value when
>>> # receiving the response from "worker1"
>>> cb_fut = fut.then(callback)
>>> chain_cb_fut = cb_fut.then(
...     lambda x : print(f"Chained cb done. {x.wait()}")
... )
>>> fut.set_result(5)
RPC return value is 5.
Chained cb done. None

value() [исходный код]

Получает значение уже завершённого future.

Этот метод следует вызывать только после завершения вызова wait() или внутри функции обратного вызова, переданной в then(). В остальных случаях этот Future может ещё не содержать значения, и вызов value() может завершиться ошибкой.

Если значение содержит тензоры, расположенные на GPU, этот метод не выполняет дополнительную синхронизацию. Её необходимо выполнить заранее отдельно с помощью вызова wait() (за исключением функций обратного вызова, где синхронизация уже обеспечивается методом then()).

Возвращает:

Значение, хранящееся в этом Future. Если функция (функция обратного вызова или RPC), создавшая значение, вызвала ошибку, этот метод value() также вызовет ошибку.

Тип возвращаемого значения:

T

wait() [исходный код]

Блокирует выполнение до готовности значения этого Future.

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

Возвращает:

Значение, хранящееся в этом Future. Если функция (функция обратного вызова или RPC), создавшая значение, вызвала ошибку, этот метод wait также вызовет ошибку.

Тип возвращаемого значения:

T

torch.futures.collect_all(futures) [исходный код]

Собирает переданные объекты Future в один общий объект Future, который завершается после завершения всех вложенных future.

Параметры:

futures (list) – список объектов Future.

Возвращает:

Объект Future, содержащий список переданных объектов Future.

Тип возвращаемого значения:

Future[list[Future]]

Пример::

>>> fut0 = torch.futures.Future()
>>> fut1 = torch.futures.Future()
>>> fut = torch.futures.collect_all([fut0, fut1])
>>> fut0.set_result(0)
>>> fut1.set_result(1)
>>> fut_list = fut.wait()
>>> print(f"fut0 result = {fut_list[0].wait()}")
fut0 result = 0
>>> print(f"fut1 result = {fut_list[1].wait()}")
fut1 result = 1

torch.futures.wait_all(futures) [исходный код]

Ожидает завершения всех переданных future и возвращает список завершённых значений. Если в одном из future возникает ошибка, метод немедленно завершается и сообщает об ошибке, не дожидаясь завершения остальных future.

Параметры:

futures (list) – список объектов Future.

Возвращает:

Список результатов завершённых объектов Future. Этот метод вызовет ошибку, если wait любого объекта Future вызовет ошибку.

Тип возвращаемого значения:

list