torch.futures

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

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

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

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

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

add_done_callback(callback) [source]

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

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

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

Параметры

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

Примечание

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

Пример::

>>> 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() [source]

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

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

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

bool

set_exception(result) [source]

Установить исключение для этого 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) [source]

Установить результат для этого 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) [source]

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

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

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

Параметры

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

Возвращает

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

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

Future[S]

Примечание

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

Пример::

>>> 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() [source]

Получить значение уже завершённого будущего.

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

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

Возвращает

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

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

T

wait() [source]

Ожидание, пока значение этого Future будет готово.

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

Возвращает

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

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

T

torch.futures.collect_all(futures) [source]

Собирает предоставленные объекты Future в один объединённый объект Future, который завершается, когда все подзадачи завершены.

Параметры

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

Возвращает

Возвращает объект Future со списком переданных объектов Futures.

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

Future[Список[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) [source]

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

Параметры

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

Возвращает

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

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

Список