到期計時器

到期計時器設定在與代理程式相同的處理序上，並從您的腳本中使用以處理卡住的工作器。當您進入有可能卡住的程式碼區塊時，您可以取得一個到期計時器，它會指示計時器伺服器在未能在自行設定的到期期限前釋放計時器的情況下終止處理序。

用法

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 秒的時間來完成，則工作器處理序將被終止，並且代理程式將重試工作器群組。

用戶端方法

torch.distributed.elastic.timer.configure(timer_client)

設定計時器用戶端。必須在使用 expires 之前呼叫。

torch.distributed.elastic.timer.expires(after, scope=None, client=None)

取得一個倒計時計時器，該計時器將在從現在開始的 after 秒後到期，除非它包裝的程式碼區塊在時間範圍內完成。當計時器到期時，此工作器將有資格被回收。 「回收」的確切含義取決於用戶端實作。在大多數情況下，回收意味著終止工作器處理序。請注意，工作器不保證在 time.now() + after 時被回收，而是工作器「有資格」被回收，並且用戶端與之通話的 TimerServer 將最終決定何時以及如何回收具有過期計時器的工作器。

用法

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)

與 LocalTimerClient 一起使用的伺服器。用戶端應該是執行此伺服器的父處理序的子處理序。作業中的每個主機都應該在本地端啟動自己的計時器伺服器，並且每個伺服器執行個體都管理本地端工作器（在同一個主機上的處理序上執行）的計時器。

class torch.distributed.elastic.timer.LocalTimerClient(mp_queue)

LocalTimerServer 的用戶端。此用戶端設計用於與 LocalTimerServer 運行在相同的主機上，並使用 pid 來唯一識別工作者。這在主機上有多個 GPU 裝置，並為每個 GPU 產生一個子進程（訓練器）的情況下特別有用。

以下是以命名管道為基礎實作的另一組計時器伺服器和用戶端。

class torch.distributed.elastic.timer.FileTimerServer(file_path, run_id, max_interval=10, daemon=True, log_event=None)

與 FileTimerClient 搭配使用的伺服器。用戶端應與運行此伺服器的進程在同一台主機上運行。作業中的每個主機都應在本地啟動自己的計時器伺服器，並且每個伺服器實例都管理本地工作者（在同一台主機上的進程上運行）的計時器。

參數

• file_path (str) – 字串，要建立的 FIFO 特殊檔案的路徑。

• max_interval (float) – 浮點數，每個看門狗迴圈的最大間隔（以秒為單位）。

• daemon (bool) – 布林值，以守護進程模式運行看門狗執行緒或不運行。守護進程執行緒不會阻止進程停止。

• log_event (Optional[Callable[[str, Optional[FileTimerRequest]], None]]) – Callable[[Dict[str, str]], None]，一個用於以 JSON 格式記錄事件的可選回調函數。

class torch.distributed.elastic.timer.FileTimerClient(file_path, signal=Signals.SIGKILL)

FileTimerServer 的用戶端。此用戶端設計用於與 FileTimerServer 運行在相同的主機上，並使用 pid 來唯一識別工作者。此用戶端使用命名管道將計時器請求發送到 FileTimerServer。此用戶端是生產者，而 FileTimerServer 是消費者。多個用戶端可以使用同一個 FileTimerServer。

參數

• file_path (str) – 字串，FIFO 特殊檔案的路徑。FileTimerServer 必須透過呼叫 os.mkfifo() 建立它。

• signal – 訊號，用於終止進程的訊號。使用負數或零訊號不會終止進程。

撰寫自定義計時器伺服器/用戶端

若要撰寫您自己的計時器伺服器和用戶端，請針對伺服器擴展 torch.distributed.elastic.timer.TimerServer，並針對用戶端擴展 torch.distributed.elastic.timer.TimerClient。TimerRequest 物件用於在伺服器和用戶端之間傳遞訊息。

class torch.distributed.elastic.timer.TimerRequest(worker_id, scope_id, expiration_time)

代表在 TimerClient 和 TimerServer 之間使用的倒數計時器取得和釋放的資料物件。負數 expiration_time 應被解釋為「釋放」請求。

注意

worker_id 的類型因實作而異。它是 TimerServer 和 TimerClient 實作中用於唯一識別工作者的任何內容。

class torch.distributed.elastic.timer.TimerServer(request_queue, max_interval, daemon=True)

監控活動計時器並及時使其過期的實體。此伺服器負責回收已過期計時器的工作者。

abstract clear_timers(worker_ids)

清除給定 worker_ids 的所有計時器。

abstract get_expired_timers(deadline)

傳回每個 worker_id 的所有過期計時器。過期計時器是過期時間小於或等於提供的截止時間的計時器。

傳回類型

Dict[str, List[TimerRequest]]

abstract register_timers(timer_requests)

處理傳入的計時器請求，並將其註冊到伺服器。計時器請求可以是取得計時器或釋放計時器請求。過期時間為負數的計時器請求應被解釋為釋放計時器請求。

class torch.distributed.elastic.timer.TimerClient

透過與 TimerServer 通訊來取得和釋放倒數計時器的用戶端程式庫。

abstract acquire(scope_id, expiration_time)

為持有此用戶端物件的工作者取得計時器，給定 scope_id 和 expiration_time。通常會向 TimerServer 註冊計時器。

abstract release(scope_id)

釋放此用戶端代表的工作者上 scope_id 的計時器。呼叫此方法後，範圍上的倒數計時器將不再有效。

除錯資訊記錄

torch.distributed.elastic.timer.debug_info_logging.log_debug_info_for_expired_timers(run_id, expired_timers)