超时计时器¶
超时计时器在与代理相同的进程上设置,并从您的脚本中使用,以处理卡住的 worker。当您进入可能卡住的代码块时,您可以获取一个超时计时器,该计时器指示计时器服务器在进程未在自行设定的超时截止日期前释放计时器时终止该进程。
用法
import torchelastic.timer as timer
import torchelastic.agent.server as agent
def main():
start_method = "spawn"
message_queue = mp.get_context(start_method).Queue()
server = timer.LocalTimerServer(message, max_interval=0.01)
server.start() # non-blocking
spec = WorkerSpec(
fn=trainer_func,
args=(message_queue,),
...<OTHER_PARAMS...>)
agent = agent.LocalElasticAgent(spec, start_method)
agent.run()
def trainer_func(message_queue):
timer.configure(timer.LocalTimerClient(message_queue))
with timer.expires(after=60): # 60 second expiry
# do some work
在上面的示例中,如果 trainer_func
耗时超过 60 秒才能完成,则 worker 进程将被终止,并且代理会重试 worker 组。
客户端方法¶
- torch.distributed.elastic.timer.expires(after, scope=None, client=None)[source]¶
获取一个倒计时计时器,该计时器将在
after
秒后过期,除非它包装的代码块在此时间范围内完成。当计时器过期时,此 worker 有资格被回收。 “回收”的确切含义取决于客户端的实现。在大多数情况下,回收意味着终止 worker 进程。请注意,worker 不会保证在 exactlytime.now() + after
时被回收,而是 worker “有资格”被回收,并且客户端所联系的TimerServer
最终将决定何时以及如何回收具有过期计时器的 worker。用法
torch.distributed.elastic.timer.configure(LocalTimerClient()) with expires(after=10): torch.distributed.all_reduce(...)
服务器/客户端实现¶
以下是 torchelastic 提供的计时器服务器和客户端对。
注意
计时器服务器和客户端总是必须成对实现和使用,因为服务器和客户端之间存在消息传递协议。
下面是一对基于 multiprocess.Queue
实现的计时器服务器和客户端。
- class torch.distributed.elastic.timer.LocalTimerServer(mp_queue, max_interval=60, daemon=True)[source]¶
与
LocalTimerClient
配合使用的服务器。客户端预计是运行此服务器的父进程的子进程。作业中的每个主机都预计会本地启动自己的计时器服务器,并且每个服务器实例都管理本地 worker(在同一主机上的进程上运行)的计时器。
- class torch.distributed.elastic.timer.LocalTimerClient(mp_queue)[source]¶
LocalTimerServer
的客户端。此客户端旨在与LocalTimerServer
运行在同一主机上,并使用 pid 来唯一标识 worker。这在每个主机上有多个 GPU 设备的情况下,每个 GPU 生成一个子进程(训练器)时特别有用。
下面是另一对基于命名管道实现的计时器服务器和客户端。
- class torch.distributed.elastic.timer.FileTimerServer(file_path, run_id, max_interval=10, daemon=True, log_event=None)[source]¶
与
FileTimerClient
协同工作的服务器。客户端预计运行在与运行此服务器的进程相同的主机上。作业中的每个主机都应该在本地启动自己的计时器服务器,并且每个服务器实例管理本地工作器(运行在同一主机上的进程)的计时器。
- class torch.distributed.elastic.timer.FileTimerClient(file_path, signal=Signals.SIGKILL)[source]¶
FileTimerServer
的客户端。此客户端旨在用于与FileTimerServer
运行在同一主机上,并使用 pid 来唯一标识工作器。此客户端使用命名管道将计时器请求发送到FileTimerServer
。此客户端是生产者,而FileTimerServer
是消费者。多个客户端可以与同一个FileTimerServer
协同工作。- 参数
file_path (str) – str,FIFO 特殊文件的路径。
FileTimerServer
必须通过调用 os.mkfifo() 创建它。signal – signal,用于杀死进程的信号。使用负数或零信号不会杀死进程。
编写自定义计时器服务器/客户端¶
要编写您自己的计时器服务器和客户端,请扩展 torch.distributed.elastic.timer.TimerServer
用于服务器,并扩展 torch.distributed.elastic.timer.TimerClient
用于客户端。TimerRequest
对象用于在服务器和客户端之间传递消息。
- class torch.distributed.elastic.timer.TimerRequest(worker_id, scope_id, expiration_time)[source]¶
表示倒计时器获取和释放的数据对象,用于
TimerClient
和TimerServer
之间。负expiration_time
应解释为“释放”请求。注意
worker_id
的类型是特定于实现的。它是 TimerServer 和 TimerClient 实现用来唯一标识工作器的任何内容。
- class torch.distributed.elastic.timer.TimerServer(request_queue, max_interval, daemon=True)[source]¶
监控活动计时器并及时使其过期的实体。此服务器负责回收计时器已过期的工作器。