# 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.

import functools

import logging

import socket

import time

import os

import signal

import copy

import sys

import six

import subprocess

from contextlib import closing

import socket

from paddle.fluid import core

from paddle.distributed.fleet.launch_utils import get_backend_by_compile_flag

from distutils.util import strtobool

from paddle.fluid.layer_helper import LayerHelper

from paddle.fluid.framework import _non_static_mode

from paddle.fluid.data_feeder import check_variable_and_dtype

from paddle import _C_ops

__all__ = [ #noqa

'get_host_name_ip',

'Trainer',

'get_cluster',

'start_local_trainers',

'watch_local_trainers',

'find_free_ports',

'JobServer',

'Cluster',

'Pod',

'Hdfs',

'add_arguments',

'terminate_local_procs',

'TrainerProc',

'get_logger',

'pull_worker_log',

'global_scatter',

'global_gather',

]

def global_scatter(x,

local_count,

global_count,

group=None,

use_calc_stream=True):

"""

The global_scatter operator distributes the data of x to n_expert * world_size experts according to local_count,

and then receives data according to global_count. The expert refers to a user-defined expert network,

n_expert refers to the number of expert networks owned by each card, and world_size refers to the number of graphics cards running the network.

As shown below, the value of the world size is 2, n_expert 2, the batch size of the x 4 and local_count is [2, 0, 2, 0].

The global_count of the rank 0 is [2, 0, , ], rank 1 is [2, 0, ,](Due to the limited space, only the data calculated on rank 0 is shown here).

In the global_scatter operator, local_count[i] represents sending local_count[i] data to the (i % n_expert)th expert of the (i // n_expert)th card,

global_count[i] represents receiving global_count[i] data from the (i // n_expert)th card to the (i % n_expert)th expert of this card. The rank in the

figure respresent the rank of the current card in all cards.

The process of global_scatter sending data is as follows:

local_count[0] represents taking out 2 batches from x and sending 2 batches to the 0th expert of the 0th card;

local_count[1] represents taking out 0 batches from x and sending 0 batches to the 1th expert of the 0th card;

local_count[2] represents taking out 2 batches from x and sending 2 batches to the 0th expert of the 1th card;

local_count[3] represents taking out 0 batches from x and sending 0 batches to the 1th expert of the 1th card;

Therefore, the global_count[0] of the 0th card is equal to 2, which means that 2 batches of data are received from the 0th card to the 0th expert;

the global_count[1] of the 0th card is equal to 0, which means that 0 batches of data are received from the 0th card to the 1th expert;

the global_count[0] of the 1th card is equal to 2, which means that 2 batches of data are received from the 0th card to the 0th expert;

the global_count[1] of the 1th card is equal to 0, which means that 0 batches of data are received from the 0th card to the 1th expert.

.. image:: https://githubraw.cdn.bcebos.com/PaddlePaddle/docs/develop/docs/api/paddle/distributed/img/global_scatter_gather.png

:width: 800

:alt: global_scatter_gather

:align: center

Args:

x (Tensor): Tensor. The tensor data type should be float16, float32, float64, int32 or int64.

local_count (Tensor): Tensor which have n_expert * world_size elements that indicates

how many data needed to be sent. The tensor data type should be int64.

global_count (Tensor): Tensor which have n_expert * world_size elements that indicates

how many data needed to be received. The tensor data type should be int64.

group (Group, optional): The group instance return by new_group or None for global default group. Default: None.

use_calc_stream (bool, optional): Wether to use calculation stream (True) or communication stream. Default: True.

Returns:

out (Tensor): The data received from all experts.

Examples:

.. code-block:: python

# required: distributed

import numpy as np

import paddle

from paddle.distributed import init_parallel_env

init_parallel_env()

n_expert = 2

world_size = 2

d_model = 2

in_feat = d_model

local_input_buf = np.array([[1, 2],[3, 4],[5, 6],[7, 8],[9, 10]], \

dtype=np.float32)

if paddle.distributed.ParallelEnv().local_rank == 0:

local_count = np.array([2, 1, 1, 1])

global_count = np.array([2, 1, 1, 1])

else:

local_count = np.array([1, 1, 2, 1])

global_count = np.array([1, 1, 2, 1])

local_input_buf = paddle.to_tensor(local_input_buf, dtype="float32", stop_gradient=False)

local_count = paddle.to_tensor(local_count, dtype="int64")

global_count = paddle.to_tensor(global_count, dtype="int64")

a = paddle.distributed.utils.global_scatter(local_input_buf, \

local_count, global_count)

a.stop_gradient = False

print(a)

# out for rank 0: [[1, 2], [3, 4], [1, 2], [5, 6], [3, 4]]

# out for rank 1: [[7, 8], [5, 6], [7, 8], [9, 10], [9, 10]]

# backward test

c = a * a

c.backward()

print("local_input_buf.grad: ", local_input_buf.grad)

# out for rank 0: [[2, 4], [6, 8], [10, 12], [14, 16], [18, 20]]

# out for rank 1: [[2, 4], [6, 8], [10, 12], [14, 16], [18, 20]]

"""

if group is not None and not group.is_member():

return

ring_id = 0 if group is None else group.id

if _non_static_mode():

return _C_ops.global_scatter(x, local_count, \

global_count, \

'use_calc_stream', use_calc_stream, \

'ring_id', ring_id)

else:

op_type = 'global_scatter'

check_variable_and_dtype(

x, 'x', ['float16', 'float32', 'float64', 'int32', 'int64'],

'global_scatter')

check_variable_and_dtype(local_count, 'local_count', ['int64'],

'global_scatter')

check_variable_and_dtype(global_count, 'global_count', ['int64'],

'global_scatter')

helper = LayerHelper(op_type, **locals())

out = helper.create_variable_for_type_inference(dtype=x.dtype)

helper.append_op(type=op_type,

inputs={

'X': [x],

'local_count': [local_count],

'global_count': [global_count],

},

outputs={'Out': [out]},

attrs={

'ring_id': ring_id,

'use_calc_stream': use_calc_stream

})

return out

def global_gather(x,

local_count,

global_count,

group=None,

use_calc_stream=True):

"""

The global_gather operator gathers the data of x into n_expert * world_size experts according to global_count, and then receives data according to local_count.

The expert refers to a user-defined expert network, n_expert refers to the number of expert networks owned by each card, and world_size refers to the number of graphics cards running the network.

As shown below, the value of the world size is 2, n_expert 2, the batch size of the x 4 and local_count is [2, 0, 2, 0].

The global_count of the rank 0 is [2, 0, , ], rank 1 is [2, 0, ,](Due to the limited space, only the data calculated on rank 0 is shown here).

In the global_gather operator, the meaning of the global_count and local_count is opposed to global_scatter, global_count[i] represents sending global_count[i] data to the (i % n_expert)th expert of the (i // n_expert)th card,

local_count[i] represents receiving local_count[i] data from the (i // n_expert)th card to the (i % n_expert)th expert of this card. The data sent will be arranged according to the experts of each card.

The rank in the figure respresent the rank of the current card in all cards.

The process of global_gather sending data is as follows:

The global_count[0] of the 0th card represents sending 2 data to the 0th expert of the 0th card;

The global_count[1] of the 0th card represents sending 0 data to the 1th expert of the 0th card;

The global_count[0] of the 1th card represents sending 2 data to the 0th expert of the 0th card;

The global_count[1] of the 1th card represents sending 0 data to the 1th expert of the 0th card.

.. image:: https://githubraw.cdn.bcebos.com/PaddlePaddle/docs/develop/docs/api/paddle/distributed/img/global_scatter_gather.png

:width: 800

:alt: global_scatter_gather

:align: center

Args:

x (Tensor): Tensor. Tensor whose data type should be float16, float32, float64, int32 or int64.

local_count (Tensor): Tensor which have n_expert * world_size elements that indicates

how many data needed to be received. Tensor data type should be int64.

global_count (Tensor): Tensor which have n_expert * world_size elements that indicates

how many data needed to be sent. Tensor data type should be int64.

group (Group, optional): The group instance return by new_group or None for global default group. Default: None.

use_calc_stream (bool, optional): Wether to use calculation stream (True) or communication stream. Default: True.

Returns:

out (Tensor): The data received from all experts.

Examples:

.. code-block:: python

# required: distributed

import numpy as np

import paddle

from paddle.distributed import init_parallel_env

init_parallel_env()

n_expert = 2

world_size = 2

d_model = 2

in_feat = d_model

local_input_buf = np.array([[1, 2],[3, 4],[5, 6],[7, 8],[9, 10]],\

dtype=np.float32)

if paddle.distributed.ParallelEnv().local_rank == 0:

local_count = np.array([2, 1, 1, 1])

global_count = np.array([2, 1, 1, 1])

else:

local_count = np.array([1, 1, 2, 1])

global_count = np.array([1, 1, 2, 1])

local_input_buf = paddle.to_tensor(local_input_buf, dtype="float32", stop_gradient=False)

local_count = paddle.to_tensor(local_count, dtype="int64")

global_count = paddle.to_tensor(global_count, dtype="int64")

a = paddle.distributed.utils.global_gather(local_input_buf, local_count, global_count)

print(a)

# out for rank 0: [[1, 2], [3, 4], [7, 8], [1, 2], [7, 8]]

# out for rank 1: [[5, 6], [9, 10], [3, 4], [5, 6], [9, 10]]

a.stop_gradient = False

c = a * a

c.backward()

print("local_input_buf.grad", local_input_buf.grad)

# out for rank 0: [[2, 4], [6, 8], [10, 12], [14, 16], [18, 20]]

# out for rank 1: [[2, 4], [6, 8], [10, 12], [14, 16], [18, 20]]

"""

if group is not None and not group.is_member():

return

ring_id = 0 if group is None else group.id

if _non_static_mode():

return _C_ops.global_gather(x, local_count, \

global_count, \

'use_calc_stream', use_calc_stream, \

'ring_id', ring_id)

else:

op_type = 'global_gather'

check_variable_and_dtype(

x, 'x', ['float16', 'float32', 'float64', 'int32', 'int64'],

'global_gather')

check_variable_and_dtype(local_count, 'local_count', ['int64'],

'global_gather')

check_variable_and_dtype(global_count, 'global_count', ['int64'],

'global_gather')

helper = LayerHelper(op_type, **locals())

out = helper.create_variable_for_type_inference(dtype=x.dtype)

helper.append_op(type=op_type,

inputs={

'X': [x],

'local_count': [local_count],

'global_count': [global_count]

},

outputs={'Out': [out]},

attrs={

'ring_id': group,

'use_calc_stream': use_calc_stream,

})

return out

logger = logging.getLogger("root")

logger.propagate = False

def get_cluster_from_args(args, selected_gpus):

node_ips = [x.strip() for x in args.cluster_node_ips.split(',')]

node_ip = args.node_ip

node_rank = node_ips.index(node_ip)

logger.debug("parsed from args:node_ips:{} node_ip:{} node_rank:{}".format(

node_ips, node_ip, node_rank))

free_ports = None

if not args.use_paddlecloud and len(

node_ips) <= 1 and args.started_port is None:

free_ports = find_free_ports(len(selected_gpus))

if free_ports is not None:

free_ports = list(free_ports)

else:

started_port = 6070

if args.started_port is not None:

started_port = args.started_port

free_ports = [

x for x in range(started_port, started_port + len(selected_gpus))

]

trainer_endpoints = []

for ip in node_ips:

trainer_endpoints.append(["%s:%d" % (ip, port) for port in free_ports])

return get_cluster(node_ips, node_ip, trainer_endpoints, selected_gpus)

def get_gpus(selected_gpus):

if selected_gpus is None:

from paddle.fluid import core

gpus_num = core.get_cuda_device_count()

gpus = [str(x) for x in range(0, gpus_num)]

else:

cuda_visible_devices = os.getenv("CUDA_VISIBLE_DEVICES")

if cuda_visible_devices is None or cuda_visible_devices == "":

gpus = [x.strip() for x in selected_gpus.split(',')]

else:

# change selected_gpus into relative values

# e.g. CUDA_VISIBLE_DEVICES=4,5,6,7; args.selected_gpus=4,5,6,7;

# therefore selected_gpus=0,1,2,3

cuda_visible_devices_list = cuda_visible_devices.split(',')

for x in selected_gpus.split(','):

assert x in cuda_visible_devices_list, "Can't find "\

"your selected_gpus %s in CUDA_VISIBLE_DEVICES[%s]."\

% (x, cuda_visible_devices)

gpus = [

cuda_visible_devices_list.index(x.strip())

for x in selected_gpus.split(',')

]

logger.info("Change selected_gpus into reletive values. --ips:{} "

"will change into relative_ips:{} according to your "

"CUDA_VISIBLE_DEVICES:{}".format(

selected_gpus, gpus, cuda_visible_devices_list))

return gpus

def _print_arguments(args):

print("----------- Configuration Arguments -----------")

for arg, value in sorted(six.iteritems(vars(args))):

print("%s: %s" % (arg, value))

print("------------------------------------------------")

class Hdfs(object):

def __init__(self):

self.hdfs_ugi = None

self.hdfs_name = None

self.hdfs_path = None

def is_valid(self):

return self.hdfs_ugi is not None and \

self.hdfs_name is not None and \

self.hdfs_path is not None

def __str__(self):

return "hdfs_ugi:{} hdfs_name:{} hdfs_path{}".format(

self.hdfs_ugi, self.hdfs_name, self.hdfs_path)

def __eq__(self, n):

return self.hdfs_ugi == n.hdfs_ugi and \

self.hdfs_name == n.hdfs_name and \

self.hdfs_path == n.hdfs_path

def __ne__(self, n):

return not self == n

class Cluster(object):

def __init__(self, hdfs):

self.job_server = None

self.pods = []

self.hdfs = None

self.job_stage_flag = None

def __str__(self):

return "job_server:{} pods:{} job_stage_flag:{} hdfs:{}".format(

self.job_server, [str(pod) for pod in self.pods],

self.job_stage_flag, self.hdfs)

def __eq__(self, cluster):

if len(self.pods) != len(cluster.pods):

return False

for a, b in zip(self.pods, cluster.pods):

if a != b:

return False

if self.job_stage_flag != cluster.job_stage_flag:

return False

return True

def __ne__(self, cluster):

return not self.__eq__(cluster)

def update_pods(self, cluster):

self.pods = copy.copy(cluster.pods)

def trainers_nranks(self):

return len(self.trainers_endpoints())

def pods_nranks(self):

return len(self.pods)

def trainers_endpoints(self):

r = []

for pod in self.pods:

for t in pod.trainers:

r.append(t.endpoint)

return r

def pods_endpoints(self):

r = []

for pod in self.pods:

ep = "{}:{}".format(pod.addr, pod.port)

assert pod.port != None and pod.addr != None, "{} not a valid endpoint".format(

ep)

r.append(ep)

return r

def get_pod_by_id(self, pod_id):

for pod in self.pods:

if str(pod_id) == str(pod.id):

return pod

return None

class JobServer(object):

def __init__(self):

self.endpoint = None

def __str__(self):

return "{}".format(self.endpoint)

def __eq__(self, j):

return self.endpint == j.endpoint

def __ne__(self, j):

return not self == j

class Trainer(object):

def __init__(self):

self.gpus = []

self.endpoint = None

self.rank = None

def __str__(self):

return "gpu:{} endpoint:{} rank:{}".format(self.gpus, self.endpoint,

self.rank)

def __eq__(self, t):

if len(self.gpus) != len(t.gpus):

return False

if self.endpoint != t.endpoint or \

self.rank != t.rank:

return False

for a, b in zip(self.gpus, t.gpus):

if a != b:

return False

return True

def __ne__(self, t):

return not self == t

def get_rank(self):

return self.rank

class Pod(object):

def __init__(self):

self.rank = None

self.id = None

self.addr = None

self.port = None

self.trainers = []

self.gpus = []

def __str__(self):

return "rank:{} id:{} addr:{} port:{} visible_gpu:{} trainers:{}".format(

self.rank, self.id, self.addr, self.port, self.gpus,

[str(t) for t in self.trainers])

def __eq__(self, pod):

if self.rank != pod.rank or \

self.id != pod.id or \

self.addr != pod.addr or \

self.port != pod.port:

logger.debug("pod {} != {}".format(self, pod))

return False

if len(self.trainers) != len(pod.trainers):

logger.debug("trainers {} != {}".format(self.trainers,

pod.trainers))

return False

for i in range(len(self.trainers)):

if self.trainers[i] != pod.trainers[i]:

logger.debug("trainer {} != {}".format(self.trainers[i],

pod.trainers[i]))

return False

return True

def __ne__(self, pod):

return not self == pod

def parse_response(self, res_pods):

pass

def get_visible_gpus(self):

r = ""

for g in self.gpus:

r += "{},".format(g)

assert r != "", "this pod {} can't see any gpus".format(self)

r = r[:-1]

return r

def get_logger(log_level, name="root"):

logger = logging.getLogger(name)

# Avoid printing multiple logs

if not logger.handlers:

logger.setLevel(log_level)

log_handler = logging.StreamHandler()

log_format = logging.Formatter(

'%(levelname)s %(asctime)s %(filename)s:%(lineno)d] %(message)s')

log_handler.setFormatter(log_format)

logger.addHandler(log_handler)

return logger

def get_cluster(node_ips, node_ip, trainer_endpoints, selected_gpus):

assert type(trainer_endpoints) is list, "trainer_endpoints must be list"

cluster = Cluster(hdfs=None)

trainer_rank = 0

for node_rank, ip in enumerate(node_ips):

pod = Pod()

pod.rank = node_rank

pod.addr = ip

cur_node_endpoints = trainer_endpoints[node_rank]

# when use paddlecloud, endpoints may > selected_gpus(user_defined)

assert len(cur_node_endpoints) >= len(

selected_gpus

), "current trainer_endpoints size should be greater equal than selected_gpus size."

for i in range(len(selected_gpus)):

trainer = Trainer()

trainer.gpus.append(selected_gpus[i])

trainer.endpoint = "%s" % (cur_node_endpoints[i])

trainer.rank = trainer_rank

trainer_rank += 1

pod.trainers.append(trainer)

cluster.pods.append(pod)

pod_rank = node_ips.index(node_ip)

return cluster, cluster.pods[pod_rank]

def terminate_local_procs(procs):

for p in procs:

if p.proc.poll() is None:

p.proc.terminate()

if p.log_fn:

p.log_fn.close()

logger.debug("terminate process id:{}".format(p.proc.pid))

#wait all process terminiated

time.sleep(3)

for step in range(0, 50):

alive = False

for p in procs:

if p.proc.poll() is None: # not termniate

os.kill(p.proc.pid, signal.SIGKILL)

alive = True

if not alive:

logger.info("terminate all the procs")

return

time.sleep(3)

logger.fatal("can't kill all process and exit")

exit(1)

def get_host_name_ip():

try:

host_name = socket.gethostname()

host_ip = socket.gethostbyname(host_name)

return host_name, host_ip

except:

return None

def add_arguments(argname, type, default, help, argparser, **kwargs):

"""Add argparse's argument.

Usage:

.. code-block:: python

parser = argparse.ArgumentParser()

add_argument("name", str, "Jonh", "User name.", parser)

args = parser.parse_args()

"""

type = strtobool if type == bool else type

argparser.add_argument("--" + argname,

default=default,

type=type,

help=help + ' Default: %(default)s.',

**kwargs)

def find_free_ports(num):

def __free_port():

with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:

s.bind(('', 0))

return s.getsockname()[1]

port_set = set()

step = 0

while True:

port = __free_port()

if port not in port_set:

port_set.add(port)

if len(port_set) >= num:

return port_set

step += 1

if step > 100:

print(

"can't find avilable port and use the specified static port now!"

)

return None

return None

def _prepare_trainer_env(cluster, trainer, backend=None):

if backend is None:

backend = get_backend_by_compile_flag() # for compatibility

if backend == 'bkcl':

proc_env = {

"FLAGS_selected_xpus":

"%s" % ",".join([str(g) for g in trainer.gpus]),

"PADDLE_TRAINER_ID": "%d" % trainer.rank,

"PADDLE_CURRENT_ENDPOINT": "%s" % trainer.endpoint,

"PADDLE_TRAINERS_NUM": "%d" % cluster.trainers_nranks(),

"PADDLE_TRAINER_ENDPOINTS": ",".join(cluster.trainers_endpoints())

}

elif backend == 'nccl':

proc_env = {

"FLAGS_selected_gpus":

"%s" % ",".join([str(g) for g in trainer.gpus]),

"PADDLE_TRAINER_ID": "%d" % trainer.rank,

"PADDLE_CURRENT_ENDPOINT": "%s" % trainer.endpoint,

"PADDLE_TRAINERS_NUM": "%d" % cluster.trainers_nranks(),

"PADDLE_TRAINER_ENDPOINTS": ",".join(cluster.trainers_endpoints())

}

elif backend == 'cncl':

proc_env = {

"FLAGS_selected_mlus":

"%s" % ",".join([str(g) for g in trainer.gpus]),

"PADDLE_TRAINER_ID": "%d" % trainer.rank,

"PADDLE_CURRENT_ENDPOINT": "%s" % trainer.endpoint,

"PADDLE_TRAINERS_NUM": "%d" % cluster.trainers_nranks(),

"PADDLE_TRAINER_ENDPOINTS": ",".join(cluster.trainers_endpoints())

}

elif backend == 'gloo':

# NOTE (xiongkun) default fall back into cpu only

proc_env = {

"PADDLE_TRAINER_ID": "%d" % trainer.rank,

"PADDLE_CURRENT_ENDPOINT": "%s" % trainer.endpoint,

"PADDLE_TRAINERS_NUM": "%d" % cluster.trainers_nranks(),

"PADDLE_TRAINER_ENDPOINTS": ",".join(cluster.trainers_endpoints()),

"PADDLE_DISTRI_BACKEND":

backend, # only add here, other will be auto

}

else:

raise ValueError("backend must be one of 'gloo, nccl, bkcl'")

return proc_env

class TrainerProc(object):

def __init__(self):

self.proc = None

self.log_fn = None

self.log_offset = None

self.rank = None

self.local_rank = None

self.cmd = None

def start_local_trainers(cluster,

pod,

training_script,

training_script_args,

log_dir=None):

current_env = copy.copy(os.environ.copy())

#paddle broadcast ncclUniqueId use socket, and

#proxy maybe make trainers unreachable, so delete them.

#if we set them to "", grpc will log error message "bad uri"

#so just delete them.

current_env.pop("http_proxy", None)

current_env.pop("https_proxy", None)

procs = []

for idx, t in enumerate(pod.trainers):

proc_env = _prepare_trainer_env(cluster, t)

current_env.update(proc_env)

logger.debug("trainer proc env:{}".format(current_env))

cmd = [sys.executable, "-u", training_script] + training_script_args

logger.info("start trainer proc:{} env:{}".format(cmd, proc_env))

fn = None

if log_dir is not None:

os.system("mkdir -p {}".format(log_dir))

fn = open("%s/workerlog.%d" % (log_dir, idx), "a")

proc = subprocess.Popen(cmd, env=current_env, stdout=fn, stderr=fn)

else:

proc = subprocess.Popen(cmd, env=current_env)

tp = TrainerProc()

tp.proc = proc

tp.rank = t.rank

tp.local_rank = idx

tp.log_fn = fn

tp.log_offset = fn.tell() if fn else None

tp.cmd = cmd

procs.append(tp)

return procs

def pull_worker_log(tp):

if tp.log_fn:

with open(tp.log_fn.name, 'r') as fin:

fin.seek(tp.log_offset, 0)

for line in fin:

try:

sys.stdout.write(line)

except UnicodeEncodeError:

sys.stdout.write(

'UnicodeEncodeError occurs at this line. '

'Please refer to the original log file "%s"\n' %

tp.log_fn.name)

tp.log_offset = fin.tell()

def watch_local_trainers(procs, nranks):

try:

error = False

error_rank = []

# wait all process finish or one error

alive = False

for p in procs:

if p.log_fn and p.local_rank == 0:

pull_worker_log(p)

ret = p.proc.poll()

if ret is None:

alive = True

elif ret != 0:

error = True

error_rank.append(p.rank)

if error:

terminate_local_procs(procs)

exit(1)

except KeyboardInterrupt:

logger.warning("KeyboardInterrupt, exit")

terminate_local_procs(procs)

raise

except SystemExit:

logger.error(

"ABORT!!! Out of all {} trainers, the trainer process with rank={} was aborted. Please check its log."

.format(nranks, error_rank))

terminate_local_procs(procs)

raise

except:

logger.error(

"ABORT!!! Out of all {} trainers, the trainer process with rank={} was aborted. Please check its log."

.format(nranks, error_rank))

terminate_local_procs(procs)

raise

return alive