Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 12 additions & 6 deletions python/paddle/distributed/fleet/launch.py
Original file line number Diff line number Diff line change
Expand Up @@ -279,14 +279,20 @@ def launch_collective(args):
print("launch proc_id:{} idx:{}".format(proc.proc.pid, idx))

while True:
alive = watch_local_trainers(procs, cluster.trainers_nranks())
try:
alive = watch_local_trainers(procs, cluster.trainers_nranks())

if not alive:
logger.info("Local processes completed.")
logger.debug("POD info:{}".format(pod))
break
if not alive:
logger.info("Local processes completed.")
logger.debug("POD info:{}".format(pod))
break

time.sleep(3)
time.sleep(3)

except:
logger.warning("Terminating... exit")
terminate_local_procs(procs)
exit(1)

if os.path.exists(gloo_rendezvous_dir):
shutil.rmtree(gloo_rendezvous_dir)
Expand Down
17 changes: 14 additions & 3 deletions python/paddle/distributed/fleet/launch_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -307,6 +307,17 @@ def get_cluster(node_ips, node_ip, trainer_endpoints, device_mode,


def terminate_local_procs(procs):
# try to terminate process by group, this happend in multiprocess senario in user process
if os.name != 'nt':
for p in procs:
if p.proc.poll() is None:
os.killpg(os.getpgid(p.proc.pid), signal.SIGTERM)
if p.log_fn:
p.log_fn.close()
logger.info("terminate process group gid:{}".format(p.proc.pid))

time.sleep(1)

for p in procs:
if p.proc.poll() is None:
p.proc.terminate()
Expand Down Expand Up @@ -583,19 +594,19 @@ def watch_local_trainers(procs, nranks):
except KeyboardInterrupt:
logger.warning("KeyboardInterrupt, exit")
terminate_local_procs(procs)
raise
return
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
return
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

return alive

Expand Down