colossalai

1
import torch
2
from torch.fx import symbolic_trace
3

4
from colossalai.fx.passes.adding_split_node_pass import (
5
    balanced_split_pass,
6
    balanced_split_pass_v2,
7
    split_with_split_nodes_pass,
8
    uniform_split_pass,
9
)
10
from colossalai.testing import clear_cache_before_run
11

12
MODEL_DIM = 16
13
BATCH_SIZE = 8
14
PIPELINE_SIZE = 2
15

16

17
class MLP(torch.nn.Module):
18
    def __init__(self, dim: int):
19
        super().__init__()
20
        self.linear1 = torch.nn.Linear(dim, dim)
21
        self.linear2 = torch.nn.Linear(dim, dim)
22
        self.linear3 = torch.nn.Linear(dim, dim)
23
        self.linear4 = torch.nn.Linear(dim, dim)
24

25
    def forward(self, x):
26
        x = self.linear1(x)
27
        x = self.linear2(x)
28
        x = self.linear3(x)
29
        x = self.linear4(x)
30
        return x
31

32

33
def pipeline_pass_test_helper(model, data, pass_func):
34
    origin_output = model(data)
35
    symbolic_traced = symbolic_trace(model)
36
    annotated_model = pass_func(symbolic_traced, PIPELINE_SIZE)
37
    split_model, split_submodules = split_with_split_nodes_pass(annotated_model)
38
    output = split_model(data)
39
    assert output.equal(origin_output)
40

41

42
@clear_cache_before_run()
43
def test_pipeline_passes():
44
    model = MLP(MODEL_DIM)
45
    data = torch.rand(BATCH_SIZE, MODEL_DIM)
46
    pipeline_pass_test_helper(model, data, balanced_split_pass)
47
    pipeline_pass_test_helper(model, data, balanced_split_pass_v2)
48
    pipeline_pass_test_helper(model, data, uniform_split_pass)
49

50

51
if __name__ == "__main__":
52
    test_pipeline_passes()
53