role_maker.py 6.5 KB
Newer Older
D
dongdaxiang 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
#   Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
14
import sys
D
dongdaxiang 已提交
15 16 17


class RoleMakerBase(object):
18 19 20 21 22 23 24
    """
    RoleMakerBase is a base class for assigning a role to current process
    in distributed training.
    A paddle developer can implement RoleMakerBase to design a role maker
    for worker or pserver assignment.
    """

D
dongdaxiang 已提交
25 26 27 28
    def __init__(self):
        self.role_maker_name_ = ""
        self.trainer_endpoints_ = []
        self.pserver_endpoints_ = []
29
        self.role_is_generated_ = False
D
dongdaxiang 已提交
30 31

    def is_worker(self):
32 33 34
        """
        return is_worker() of current process
        """
D
dongdaxiang 已提交
35 36 37
        raise NotImplementedError("Please implement this method in child class")

    def is_server(self):
38 39 40
        """
        return is_server() of current process
        """
D
dongdaxiang 已提交
41 42 43
        raise NotImplementedError("Please implement this method in child class")

    def get_local_ip(self):
44 45 46
        """
        return get local ip
        """
D
dongdaxiang 已提交
47 48 49 50 51
        import socket
        self.ip_ = socket.gethostbyname(socket.gethostname())
        return self.ip_

    def get_trainer_endpoints(self):
52 53 54
        """
        return trainer endpoints
        """
D
dongdaxiang 已提交
55 56 57
        return self.trainer_endpoints_

    def get_pserver_endpoints(self):
58 59 60
        """
        return pserver endpoints
        """
D
dongdaxiang 已提交
61 62 63
        return self.pserver_endpoints_

    def generate_role(self):
64 65 66
        """
        generate_role() should be called to identify current process's role
        """
D
dongdaxiang 已提交
67 68 69 70
        raise NotImplementedError("Please implement this method in child class")


class MPIRoleMaker(RoleMakerBase):
71 72 73 74 75
    """
    MPIRoleMaker is a MPI-API based role maker which is a counter-part of K8SRoleMaker
    mpi4py will be used if a developer inherits MPIRoleMaker
    """

D
dongdaxiang 已提交
76
    def __init__(self):
X
xujiaqi01 已提交
77
        super(MPIRoleMaker, self).__init__()
D
dongdaxiang 已提交
78 79 80
        from mpi4py import MPI
        self.comm_ = MPI.COMM_WORLD
        self.MPI = MPI
D
dongdaxiang 已提交
81
        self.ips_ = None
D
dongdaxiang 已提交
82 83

    def get_rank(self):
84 85 86
        """
        return rank
        """
D
dongdaxiang 已提交
87 88 89 90
        self.rank_ = self.comm_.Get_rank()
        return self.rank_

    def get_size(self):
91 92 93
        """
        return size
        """
D
dongdaxiang 已提交
94 95 96 97
        self.size_ = self.comm_.Get_size()
        return self.size_

    def all_gather(self, obj):
98 99 100
        """
        all_gather(obj) will call MPI's allgather function
        """
D
dongdaxiang 已提交
101 102 103 104
        self.barrier_all()
        return self.comm_.allgather(obj)

    def barrier_all(self):
105 106 107
        """
        barrier_all() will call MPI's barrier_all function
        """
D
dongdaxiang 已提交
108 109 110
        self.comm_.barrier()

    def get_ips(self):
111 112 113
        """
        collect current distributed job's ip list
        """
D
dongdaxiang 已提交
114 115 116 117 118
        if self.ips_ == None:
            self.ips_ = self.comm_.allgather(self.get_local_ip())
        return self.ips_

    def finalize(self):
119 120 121
        """
        finalize the current MPI instance.
        """
D
dongdaxiang 已提交
122 123 124 125
        self.comm_.finalize()


class MPISymetricRoleMaker(MPIRoleMaker):
126 127 128 129 130 131
    """
    MPISymetricRoleMaker is designed for worker and server assignment
    under MPI. Typically, a worker and a server node will be appointed
    on each physical node. This role maker can be only used under MPI.
    """

D
dongdaxiang 已提交
132 133 134 135 136
    def __init__(self):
        super(MPISymetricRoleMaker, self).__init__()
        self.node_type_ = None
        self.proc_per_node_ = 2

137 138 139 140 141 142 143
    def _check_role_generation(self):
        if not self.role_is_generated_:
            sys.stderr.write("generate_role() should be called first")
            sys.exit(-1)
            return False
        return True

D
dongdaxiang 已提交
144
    def is_first_worker(self):
145 146 147 148 149 150
        """
        return whether current process is the first worker assigned by role maker
        """
        if self._check_role_generation():
            return self.is_worker() and 0 == self.worker_index()
        return False
D
dongdaxiang 已提交
151 152

    def is_worker(self):
153 154 155 156 157 158
        """
        return whether current process is worker assigned by role maker
        """
        if self._check_role_generation():
            return self.node_type_ == 1
        return False
D
dongdaxiang 已提交
159 160

    def is_server(self):
161 162 163 164 165 166
        """
        return whether current process is server assigned by role maker
        """
        if self._check_role_generation():
            return self.node_type_ == 0
        return False
D
dongdaxiang 已提交
167 168

    def worker_num(self):
169 170 171 172 173
        """
        return the current number of worker
        """
        if self._check_role_generation():
            if self.is_worker():
174
                return self.get_size() / 2;
175
        return 0
D
dongdaxiang 已提交
176 177

    def server_num(self):
178 179 180 181 182
        """
        return the current number of server
        """
        if self._check_role_generation():
            if self.is_server():
183
                return self.get_size() / 2;
184
        return 0
D
dongdaxiang 已提交
185 186

    def worker_index(self):
187 188 189 190 191 192
        """
        return the index of worker
        """
        if self._check_role_generation():
            return self.rank_ / self.proc_per_node_
        return 0
D
dongdaxiang 已提交
193 194

    def server_index(self):
195 196 197 198 199 200
        """
        return the index of server
        """
        if self._check_role_generation():
            return self.rank_ / self.proc_per_node_
        return 0
D
dongdaxiang 已提交
201 202

    def barrier_worker(self):
203 204 205 206 207 208
        """
        barrier all workers in current distributed job
        """
        if self._check_role_generation():
            if self.is_worker():
                self.node_type_comm_.barrier()
D
dongdaxiang 已提交
209 210

    def barrier_server(self):
211 212 213 214 215 216
        """
        barrier all servers in current distributed job
        """
        if self._check_role_generation():
            if self.is_server():
                self.node_type_comm_.barrier()
D
dongdaxiang 已提交
217 218

    def generate_role(self):
219 220 221 222 223 224 225 226 227 228 229 230 231 232
        """
        generate currently process's role
        """
        if not self.role_is_generated_:
            # TODO(guru4elephant): only allow to be called once
            self.trainer_endpoints_ = self.get_ips()
            self.pserver_endpoints_ = self.get_ips()

            if 0 == self.get_rank() % self.proc_per_node_ % 2:
                self.node_type_ = 0
            else:
                self.node_type_ = 1
            self.node_type_comm_ = self.comm_.Split(self.node_type_)
            self.role_is_generated_ = True