torch.futures

这个包提供了一个 Future 类型，封装了异步执行和一组简化操作的实用函数 在 Future 对象上。目前， Future 类型主要由 分布式RPC框架 使用。

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

围绕一个torch._C.Future进行封装，该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()相同。

Parameters

• 回调函数 (Future) – 一个 Callable，它接受一个参数,

• 是对这个Future的引用。 (which) –

注意

注意，如果回调函数抛出异常，无论是通过原始future完成时带有异常并调用fut.wait()，还是通过回调中的其他代码，都必须小心处理错误。例如，如果这个回调稍后完成了额外的futures，这些futures不会被标记为带有错误的完成，用户需要独立处理和等待这些futures。

Example::

>>> 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()

返回 True 如果此 Future 已完成。如果它 有结果或异常，则 Future 已完成。

如果值包含位于GPU上的张量，即使填充这些张量的异步内核尚未在设备上完成运行， Future.done() 仍将返回 True， 因为在该阶段结果已经是可用的，只要进行适当的同步操作（参见wait()）。

set_exception(result)

为这个 Future 设置一个异常，这将把这个 Future 标记为带有错误的完成，并触发所有附加的回调函数。请注意，在调用此 Future 的 wait()/value() 时，此处设置的异常将被内联抛出。

Parameters

结果 (BaseException) – 此 Future 的异常。

Example::

>>> 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的所有消费者的正确调度。

Parameters

结果 (对象) – 本次Future的结果对象。

Example::

>>> 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 上的张量的值， 即使生成这些张量的内核仍在设备上运行，它也可以这样做， 只要回调函数在执行期间没有更改流。如果需要更改流， 必须小心将其重新与原始流同步，即那些在回调被调用时当前的流。

Parameters

回调函数 (Callable) – 一个 Callable，它将此 Future 作为唯一的参数。

Returns

一个新的 Future 对象，用于保存 callback 的返回值，并在给定的 callback 完成时标记为已完成。

注意

注意，如果回调函数抛出异常，无论是通过原始的future由于异常完成并调用fut.wait()，还是通过回调中的其他代码，由then返回的future都会被适当地标记为遇到的错误。然而，如果这个回调稍后完成了额外的futures，这些futures不会被标记为由于错误完成，并且用户需要独立处理/等待这些futures。

Example::

>>> 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()

获取一个已完成的未来的值。

此方法应在调用wait()完成后，或在传递给then()的回调函数内调用。在其他情况下，此Future可能尚无值，调用value()可能会失败。

如果值包含位于GPU上的张量，则此方法将不会执行任何额外的同步操作。应在调用wait()之前单独进行（在回调函数内除外，因为已经在then()中处理了）。

Returns

该值由此 Future 持有。如果创建该值的函数（回调或远程过程调用）抛出了错误，此 value() 方法也将抛出错误。

wait()

阻塞直到该 Future 的值准备就绪。

如果值包含位于GPU上的张量，则会执行额外的与内核（在设备上运行）同步操作，这些内核可能会异步填充这些张量。这种同步是非阻塞的，这意味着wait()会在当前流中插入必要的指令，以确保进一步在此流中排队的操作将在异步内核之后正确调度，但一旦完成，wait()将返回，即使那些内核仍在运行。在访问和使用这些值时，只要不更改流，就不需要进一步的同步。

Returns

该值由此 Future 持有。如果创建该值的函数（回调或远程过程调用）抛出了错误，此 wait 方法也将抛出错误。

torch.futures.collect_all(futures)

收集提供的 Future 对象到一个单一的 组合 Future，当所有的子任务完成后即完成。

Parameters

期货 (列表) – 一个包含 Future 对象的列表。

Returns

返回一个Future对象到传入的Future列表中。

Example::

>>> 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)

等待所有提供的未来任务完成，并返回已完成值的列表。如果有任何一个未来任务遇到错误，该方法会提前退出并报告错误，而不会等待其他任务完成。

Parameters

期货 (列表) – 一个包含 Future 对象的列表。

Returns

已完成的Future结果列表。如果任何 Future 抛出错误，此方法将抛出错误。