Menjalankan dan menunda tugas

Saat Anda menjalankan skrip biasa, instruksi dieksekusi secara berurutan. Menggunakan Accelerate untuk mendeploy skrip Anda di beberapa GPU secara bersamaan memperkenalkan komplikasi: meskipun setiap proses mengeksekusi semua instruksi secara berurutan, beberapa mungkin lebih cepat daripada yang lain.

Anda mungkin perlu menunggu hingga semua proses mencapai titik tertentu sebelum menjalankan instruksi tertentu. Misalnya, Anda tidak boleh menyimpan model sebelum yakin bahwa setiap proses telah selesai melakukan pelatihan, dan Anda tidak ingin melanjutkan pelatihan sebelum semua bobot model dimuat. Untuk melakukan ini, cukup tulis baris berikut di kode Anda:

accelerator.wait_for_everyone()

Instruksi ini akan memblokir semua proses yang tiba lebih dulu hingga semua proses lain mencapai titik tersebut (jika Anda menjalankan skrip Anda hanya di satu GPU atau CPU, ini tidak akan melakukan apa-apa).

Beberapa contoh kasus penggunaan utilitas ini diuraikan di bawah ini:

Beberapa dari ini digunakan dengan context manager main_process_first(), yang memanfaatkan wait_for_everyone() untuk menjalankan set kode tertentu pada proses utama terlebih dahulu sebelum memicu dan meluncurkan proses lainnya

Mengunduh Dataset

Saat mengunduh dataset, Anda harus mengunduhnya terlebih dahulu pada proses utama dan kemudian memuat dataset yang sudah di-cache

load_dataset akan melakukan penguncian di balik layar untuk menghentikan beberapa unduhan yang terjadi bersamaan, tetapi jika Anda mengunduh sesuatu tanpa menggunakan pustaka ini, Anda sebaiknya menggunakan metode ini.

with accelerator.main_process_first():
    datasets = load_dataset("glue", "mrpc")

Di balik layar, ini sama dengan memanggil:

# First do something on the main process
if accelerator.is_main_process:
    datasets = load_dataset("glue", "mrpc")
else:
    accelerator.wait_for_everyone()

# And then send it to the rest of them
if not accelerator.is_main_process:
    datasets = load_dataset("glue", "mrpc")
else:
    accelerator.wait_for_everyone()

Menyimpan state_dict

Saat menyimpan state_dict dari model, karena biasanya Anda menyimpan satu file hanya pada proses utama, Anda harus menyatakannya:

if accelerator.is_main_process:
    model = accelerator.unwrap_model(model)
    torch.save(model.state_dict(), "weights.pth")

Memuat dalam state_dict

Saat memuat state_dict ke model, optimizer, atau scheduler, Anda harus menunggu hingga semua worker selesai memuat bobot sebelum melanjutkan ke pelatihan

with accelerator.main_process_first():
    state = torch.load("weights.pth")
    model.load_state_dict(state)

Menerapkan operasi CPU multi-worker

Menerapkan operasi map() pada beberapa worker, seperti tokenisasi, harus dilakukan terlebih dahulu pada proses utama, lalu disebarkan ke masing-masing worker.

datasets = load_dataset("glue", "mrpc")

with accelerator.main_process_first():
    tokenized_datasets = datasets.map(
        tokenize_function,
        batched=True,
        remove_columns=["idx", "sentence1", "sentence2"],
    )

Menerapkan pemeriksaan seperti Early Stopping

Untuk memiliki pemeriksaan yang bekerja dengan flag yang diset oleh proses tertentu, API set_trigger dan check_trigger harus digunakan. Contoh yang berguna untuk melakukannya dapat mencakup situasi seperti penggunaan early stopping dan pemantauan loss (karena setiap loss sedikit berbeda pada setiap proses).

Panggil Accelerator.set_trigger() ketika kondisi Anda telah terpenuhi, dan Accelerator.check_trigger() saat memeriksa apakah kondisi tersebut telah terpenuhi di proses mana pun:

for (x,y) in data_loader:
    logits = model(x)
    loss = loss_func(logits, y)
    # Assume `should_do_early_stopping` is a custom defined function that returns a conditional
    if should_do_early_stopping(loss):
        accelerator.set_trigger()

    # Later in the training script when we need to check for the breakpoint
    if accelerator.check_trigger():
        break