mirror of
https://github.com/hpcaitech/ColossalAI.git
synced 2025-04-28 11:45:23 +00:00
5a6fd71f90
* support p2p communication with any type of object | pass test * reconstruct pipeline schedule with p2p_v2.py(support communication with List[Any]) | pass test * [engin/schedule] use p2p_v2 to recontruct pipeline_schedule * [pipeline/rpc] implement a demo for PP with cuda rpc framework * [pipeline/rpc] support interleaving | fix checkpoint bug | change logic when dispatch data in work_list to ensure steady 1F1B * [pipeline/rpc] implement distributed optimizer | test with assert_close * [pipeline/rpc] implement distributed optimizer | test with assert_close * [pipeline/rpc] update outstanding mechanism | optimize dispatching strategy * [pipeline/rpc] update outstanding mechanism | optimize dispatching strategy * [pipeline/rpc] update outstanding mechanism | optimize dispatching strategy
45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
import torch
|
|
from torch import nn
|
|
|
|
from colossalai.pipeline.rpc.PipelineBase import FillDrainPipelineEngine, OneFOneBPipelineEngine
|
|
from rpc_test_utils import rpc_run, parse_args, RpcTestModel
|
|
|
|
|
|
def run_master(args):
|
|
torch.manual_seed(100)
|
|
|
|
epoch = args.epoch
|
|
device = args.device
|
|
stage_num = args.world_size
|
|
chunk = args.chunk
|
|
num_microbatches = args.num_microbatches
|
|
actual_stage_num = stage_num * chunk
|
|
use_checkpoint = args.use_checkpoint
|
|
|
|
sample_num = 1024
|
|
feat_num = 10
|
|
h = 10
|
|
batch_size = 1024
|
|
|
|
assert sample_num % batch_size == 0
|
|
batch_num = sample_num // batch_size
|
|
|
|
input_sample = torch.randn((sample_num, feat_num), device=device)
|
|
|
|
module_partitions = [RpcTestModel(pp_rank, actual_stage_num, feat_num, h) for pp_rank in range(actual_stage_num)]
|
|
|
|
engine = OneFOneBPipelineEngine(module_partitions=module_partitions,
|
|
stage_num=stage_num,
|
|
num_microbatches=num_microbatches,
|
|
device=device,
|
|
chunk=chunk,
|
|
checkpoint=use_checkpoint)
|
|
|
|
for _ in range(epoch):
|
|
_ = engine.forward_backward(input_sample, forward_only=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
args = parse_args()
|
|
rpc_run(args, run_master)
|