pytorch

1
# Owner(s): ["oncall: distributed"]
2

3
import random
4
import sys
5
import unittest
6
from collections import OrderedDict
7
from dataclasses import dataclass
8
from typing import List
9

10
import torch
11
import torch.nn as nn
12
from torch import distributed as dist
13
from torch.distributed.utils import _apply_to_tensors, _replace_by_prefix
14
from torch.testing._internal.common_utils import (
15
    instantiate_parametrized_tests,
16
    parametrize,
17
    run_tests,
18
    subtest,
19
    TEST_WITH_DEV_DBG_ASAN,
20
    TestCase,
21
)
22

23
if not dist.is_available():
24
    print("Distributed not available, skipping tests", file=sys.stderr)
25
    sys.exit(0)
26

27
if TEST_WITH_DEV_DBG_ASAN:
28
    print(
29
        "Skip dev-asan as torch + multiprocessing spawn have known issues",
30
        file=sys.stderr,
31
    )
32
    sys.exit(0)
33

34

35
class TestUtils(TestCase):
36
    @parametrize(
37
        "devices", [["cpu"], ["cuda"], subtest(["cpu", "cuda"], name="cpu_cuda")]
38
    )
39
    def test_apply_to_tensors(self, devices):
40
        if "cuda" in devices and (
41
            not torch.cuda.is_available() or torch.cuda.device_count() < 1
42
        ):
43
            raise unittest.SkipTest("Skipped due to lack of GPU")
44

45
        expected = 0
46

47
        def get_a_tensor():
48
            """Return a random tensor on random device."""
49
            dev = random.choice(devices)
50
            shape = random.choice(((1), (2, 3), (4, 5, 6), (7, 8, 9, 10)))
51
            t = torch.rand(shape).to(dev)
52
            nonlocal expected
53
            expected += t.numel()
54
            return t
55

56
        @dataclass
57
        class SomeDataClass:
58
            some_key: str
59
            some_float: float
60
            some_tensor: List[torch.Tensor]
61

62
        # create a mixed bag of data.
63
        data = [1, "str"]
64
        data.append({"key1": get_a_tensor(), "key2": {1: get_a_tensor()}, "key3": 3})
65
        data.insert(0, {"x", get_a_tensor(), get_a_tensor()})
66
        data.append(([1], get_a_tensor(), (1), [get_a_tensor()], {1, 2}))
67
        data.append({"abc": SomeDataClass("some_key", 1.0, [get_a_tensor()])})
68
        od = OrderedDict()
69
        od["k"] = "value"
70
        data.append(od)
71

72
        total = 0
73

74
        def fn(t):
75
            nonlocal total
76
            total += t.numel()
77
            return t
78

79
        new_data = _apply_to_tensors(fn, data)
80
        self.assertEqual(total, expected)
81
        for i, v in enumerate(data):
82
            self.assertEqual(type(new_data[i]), type(v))
83

84
    def test_replace_by_prefix(self):
85
        state_dict = {
86
            "layer.a": torch.tensor(1),
87
            "abc.layer.def": torch.tensor(2),
88
            "layer.b": torch.tensor(3),
89
        }
90
        original_state_dict = state_dict.copy()
91
        _replace_by_prefix(state_dict, "layer.", "module.layer.")
92
        assert state_dict == {
93
            "module.layer.a": torch.tensor(1),
94
            "abc.layer.def": torch.tensor(2),
95
            "module.layer.b": torch.tensor(3),
96
        }
97
        _replace_by_prefix(state_dict, "module.layer.", "layer.")
98
        assert state_dict == original_state_dict
99

100
    def test_packed_sequence(self):
101
        """Test to ensure RNN packed sequences are modified correctly."""
102
        rnn = nn.RNN(5, 5)
103

104
        x = torch.rand((5, 1, 5), dtype=torch.float)
105
        seq_length = torch.tensor([4], dtype=torch.int)
106

107
        def fill_fn(x):
108
            x.fill_(0)
109

110
        x = nn.utils.rnn.pack_padded_sequence(x, seq_length)
111
        x, h = rnn(x)
112
        x = _apply_to_tensors(fill_fn, x)
113
        x, _ = nn.utils.rnn.pad_packed_sequence(x)
114
        self.assertEqual(torch.sum(x), 0)
115

116

117
instantiate_parametrized_tests(TestUtils)
118

119
if __name__ == "__main__":
120
    run_tests()
121