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) – функция Callable, которая принимает один аргумент — ссылку на этот Future.

Примечание

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

Пример::

>>> def callback(fut):
...     print(f"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 завершен. 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 уже завершён, заданный обратный вызов будет выполнен немедленно.

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

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

Параметры:

callback (Callable) – функция 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() может завершиться ошибкой.

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

Возвращает:

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

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

T

wait() [source]

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

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

Возвращает:

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

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

T

torch.futures.collect_all(futures) [source]

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

Параметры:

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

Возвращает:

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

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

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 выбросит ошибку.

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

Список