Parallel

Patch the parallel models so they work with RNNs

DataParallel.reset[source]

DataParallel.reset()

class ParallelTrainer[source]

ParallelTrainer(device_ids) :: Callback

Basic class handling tweaks of the training loop by changing a Learner in various events

Learner.to_parallel[source]

Learner.to_parallel(device_ids=None)

Learner.detach_parallel[source]

Learner.detach_parallel()

Remove ParallelTrainer callback from Learner.

Learner.parallel_ctx[source]

Learner.parallel_ctx(device_ids=None)

A context manager to adapt a learner to train in data parallel mode.

Distributed

Patch the parallel models so they work with RNNs

DistributedDataParallel.reset[source]

DistributedDataParallel.reset()

Convenience functions to set up/tear down torch distributed data parallel mode.

setup_distrib[source]

setup_distrib(gpu=None)

teardown_distrib[source]

teardown_distrib()

DataLoader

We need to change the dataloaders so that they only get one part of the batch each (otherwise there is no point in using distributed training).

class DistributedDL[source]

DistributedDL(dl, rank, world_size) :: TfmdDL

Transformed DataLoader

dl = TfmdDL(list(range(50)), bs=12, num_workers=2)
for i in range(4):
    dl1 = DistributedDL(dl, i, 4)
    test_eq(list(dl1), (torch.arange(i*13, i*13+12)%50,torch.tensor([i*13+12])%50))

dl = TfmdDL(list(zip(range(50),range(100,150))), bs=12, num_workers=4)
for i in range(4):
    dl1 = DistributedDL(dl, i, 4)
    test_eq(list(dl1), [(torch.arange(i*13, i*13+12)%50,100+torch.arange(i*13, i*13+12)%50),
                        ((torch.tensor([i*13+12])%50),100+torch.tensor([i*13+12])%50)])

dl = TfmdDL(list(range(50)), bs=12, num_workers=2,drop_last=True)
for i in range(4):
    dl1 = DistributedDL(dl, i, 4)
    test_eq(list(dl1), [torch.arange(i*13, i*13+12)%50])

dl = TfmdDL(list(zip(range(12),range(100,112))), bs=12, num_workers=4)
res,dls = [],[]
for i in range(5): dls.append(DistributedDL(dl, i, 5))
for b in zip(*dls):
    for r in range(5):
        d=L(dls[r].to_detach(b[r]))
        test_eq(d.map(lambda x:(x!=-100).sum().item()),(3,3) if r!=4 else (0,0))

dl = TfmdDL(list(range(10)), bs=4, num_workers=2, shuffle=True)
res = []
for i in range(3):
    dl1 = DistributedDL(dl, i, 3)
    b  = list(dl1)[0]
    bd = dl1.to_detach(b)
    test_eq(b[:None if i<2 else 2],bd[4*i:4*(i+1)])

from fastai.callback.data import WeightedDL

dl = WeightedDL(list(range(50)), bs=16, num_workers=2, shuffle=True,wgts=list(np.arange(50)>=25))
res = []
for i in range(4):
    dl1 = DistributedDL(dl, i, 4)
    res += list(dl1)[0].tolist()
test(res,[25]*len(res),operator.ge)        # all res >=25
test(res,[25]*len(res),lambda a,b: ~(a<b)) # all res NOT < 25

class DistributedTrainer[source]

DistributedTrainer(cuda_id=0, sync_bn=True) :: Callback

Basic class handling tweaks of the training loop by changing a Learner in various events

Attach, remove a callback which adapts the model to use DistributedDL to train in distributed data parallel mode.

Learner.to_distributed[source]

Learner.to_distributed(cuda_id, sync_bn=True)

Learner.detach_distributed[source]

Learner.detach_distributed()

Learner.distrib_ctx[source]

Learner.distrib_ctx(cuda_id=None, sync_bn=True)

A context manager to adapt a learner to train in distributed data parallel mode.

distrib_ctx context manager

distrib_ctx(cuda_id) prepares a learner to train in distributed data parallel mode. It assumes these environment variables have all been setup properly, such as those launched by python -m fastai.launch.

Typical usage:

with learn.distrib_ctx(): learn.fit(.....)

It attaches a DistributedTrainer callback and DistributedDL data loader to the learner, then executes learn.fit(.....). Upon exiting the context, it removes the DistributedTrainer and DistributedDL, and destroys any locally created distributed process group. The process is still attached to the GPU though.

rank0_first[source]

rank0_first(func)

Execute func in the Rank-0 process first, then in other ranks in parallel.

rank0_first(f) calls f() in rank-0 process first, then in parallel on the rest, in distributed training mode. In single process, non-distributed training mode, f() is called only once as expected.

One application of rank0_first() is to make fresh downloads via untar_data() safe in distributed training scripts launched by python -m fastai.launch <script>:

path = untar_data(URLs.IMDB)

becomes:> path = rank0_first(lambda: untar_data(URLs.IMDB))

Some learner factory methods may use untar_data() to download pretrained models by default:

learn = text_classifier_learner(dls, AWD_LSTM, drop_mult=0.5, metrics=accuracy)

becomes:> learn = rank0_first(lambda: text_classifier_learner(dls, AWD_LSTM, drop_mult=0.5, metrics=accuracy)) Otherwise, multiple processes will download at the same time and corrupt the data.