diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-33369/README.md b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/README.md new file mode 100644 index 000000000..ad988c3a6 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/README.md @@ -0,0 +1,43 @@ +# PaddlePaddle__Paddle-33369 + +This directory converts Paddle PR #33369 into a SWE-Paddle community task candidate. + +## Source + +| Field | Value | +| --- | --- | +| Repo | `PaddlePaddle/Paddle` | +| PR | [33369](https://github.com/PaddlePaddle/Paddle/pull/33369) | +| PR title | `ELASTIC 1 : fault tolerance` | +| Base commit | `4b9430a1f9ac2650a6a58e061f005acf8fc12fb3` | +| Merged at | `2021-06-21T06:06:29Z` | +| Task type | `feature_enhancement` | +| Resource | CPU | + +## Summary + +Add an optional elastic lifecycle for fleet collective jobs so worker failures and membership changes can trigger a controlled restart or regroup. + +## Why This Is A Good SWE-Paddle Candidate + +- It models a real distributed-control failure rather than a numerical or device-specific edge case. +- The 495-line production change coordinates launch routing, process status, membership changes, and exit propagation. +- Deterministic CPU tests can exercise the real checkout control flow with controlled worker and launcher states. + +## Files + +- `proposal.md`: candidate proposal for maintainer triage. +- `instruction.md`: self-contained problem statement for the coding agent. +- `solution/code.patch`: gold patch from the merged PR. +- `tests/test.patch`: test patch exposing the target behavior. +- `tests/test.sh`: minimal target test command. +- `environment/README.md`: environment notes for reproduction. +- `README.md`: task overview and verification entrypoint. + +## Verification + +```bash +bash tests/test.sh +``` + +Expected behavior: applying `tests/test.patch` to `base_commit` should fail on the target behavior; applying both `tests/test.patch` and `solution/code.patch` should pass the target tests. diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-33369/environment/README.md b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/environment/README.md new file mode 100644 index 000000000..064bf83c8 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/environment/README.md @@ -0,0 +1,27 @@ +# Environment Notes + +This candidate is part of the SWE-Paddle community task set. + +## Expected Environment + +- Repository: `PaddlePaddle/Paddle` +- Base commit: `4b9430a1f9ac2650a6a58e061f005acf8fc12fb3` +- Resource: CPU +- GPU required: no +- Build path: no source build is required; tests execute checkout Python control flow with controlled launcher state and do not contact etcd. + +## Run Order + +1. Check out `PaddlePaddle/Paddle` at the base commit. +2. Apply `tests/test.patch`. +3. Run `bash tests/test.sh`; the target behavior should fail before the fix. +4. Apply `solution/code.patch`. +5. Run `bash tests/test.sh` again; the target behavior should pass after the gold patch. + +## Minimal Test Command + +```bash +bash tests/test.sh +``` + +The verifier is responsible for deriving stable F2P and P2P node IDs from repeated runs. diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-33369/instruction.md b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/instruction.md new file mode 100644 index 000000000..ee50ff0ad --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/instruction.md @@ -0,0 +1,19 @@ +# 为 collective 分布式训练增加可选的故障恢复流程 + +## 详细描述 + +collective 分布式训练运行时间较长时，任意 worker 或节点异常退出都会让当前任务直接结束，用户只能手动清理并重新启动整组训练。多节点任务还可能因为各节点启动时间不同，出现部分进程已经运行、其他节点尚未就绪的情况。 + +需要为 fleet launcher 增加可选的 elastic 流程。启用后，各节点在训练开始前先完成组网，训练期间持续关注本地进程和其他节点的状态；worker 失败时给出明确的重启结果，节点成员发生变化时先停止当前进程组并等待重新组网。未配置 elastic 时，原有 collective 和 parameter-server 启动方式应保持不变。 + +## 验收说明 + +- 启用 elastic 后，collective 任务应在节点就绪后启动，并根据 worker 退出结果区分完成、重启和错误状态。 +- 训练期间节点成员发生变化时，应停止当前进程组并等待下一次组网，不能继续使用过期的集群信息。 +- 未启用 elastic 时，已有 collective 和 parameter-server launch 流程保持不变。 + +## 技术要求 + +- 熟悉 Python 进程生命周期和信号处理。 +- 熟悉 PaddlePaddle fleet collective launcher。 +- 了解分布式任务的节点注册、状态同步和故障恢复流程。 diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-33369/proposal.md b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/proposal.md new file mode 100644 index 000000000..744102414 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/proposal.md @@ -0,0 +1,55 @@ +# Task Proposal: PaddlePaddle__Paddle-33369 + +## 1. 来源信息 + +- Instance ID：`PaddlePaddle__Paddle-33369` +- PR 链接：https://github.com/PaddlePaddle/Paddle/pull/33369 +- PR 标题：`ELASTIC 1 : fault tolerance` +- `base_commit`：`4b9430a1f9ac2650a6a58e061f005acf8fc12fb3` +- merged 时间：`2021-06-21T06:06:29Z` +- 你的身份：熟悉该模块的 contributor +- 后续联系人：TBD + +## 2. 问题一句话 + +fleet collective launcher 需要在 worker 失败或节点成员变化时安全停止当前进程组，并向外返回可用于重启或重新组网的状态。 + +## 3. 为什么适合作为 SWE-Paddle 样本 + +- **真实性**：长时间分布式训练中的 worker 失败和节点变化是实际会发生的故障场景。 +- **代表性**：任务覆盖 launcher 路由、进程状态判断、节点同步和退出状态传播，是分布式控制面的典型问题。 +- **边界清楚**：production 改动集中在 fleet launch 与新增 elastic manager，不涉及模型、算子或训练数值。 +- **非平凡性**：需要协调多个生命周期状态，并保持未启用 elastic 和 parameter-server 模式的兼容行为；production diff 为 495 行。 +- **环境友好性**：来源 PR 未提供可直接复用的 Python unit test，candidate 使用 controlled launchers 执行 checkout 中真实 manager/launch 控制流，不连接 etcd、不启动进程。 + +## 4. 任务类型和标签 + +- 任务类型：`feature_enhancement` +- 执行后端：`cpu` +- 设备范围：`cpu_only` +- 模块标签：`[distributed, fleet, launch, elastic]` + +## 5. 验证思路 + +- 目标测试命令：`bash tests/test.sh` +- 目标测试文件：`test/swe_paddle/test_pr33369_elastic_fault_tolerance.py` +- 修复前预期：parameter-server mode 的 P2P 通过；elastic manager、失败重启、成员变化暂停和 collective 生命周期接管相关 F2P 失败。 +- 修复后预期：失败 worker 返回 restart，成员变化返回 hold 并停止 launcher，collective launch 由 elastic 生命周期管理，全部测试通过。 +- P2P 候选：显式选择 parameter-server mode 时仍返回原有模式。 + +## 6. 环境与资源 + +- 资源需求：CPU +- Paddle 来源：`PaddlePaddle/Paddle` source checkout at `base_commit` +- 是否能提供 Docker：暂无 +- patch 类型：Python-only +- 环境建议：使用带 pytest 和 `six` 的 Python 环境；测试不要求 etcd、网络、GPU 或真实子进程。 +- 最小测试命令：`bash tests/test.sh` +- 是否有 oracle 日志：由 SWE-Paddle verifier 结果另行维护 + +## 7. 风险自查 + +- 泄露风险：instruction 说明故障恢复的外部行为和兼容要求，不描述类名、字段名或 Gold patch 的文件拆分方式。 +- 环境风险：真实 elastic 依赖 etcd；测试用 controlled doubles 隔离外部服务，同时执行 checkout 中的真实状态分支。 +- flaky 风险：不使用真实多进程、网络或超时轮询，所有 worker 和 membership 状态均为确定输入。 +- 拆分风险：PR 的两个 production 文件共同实现同一条 collective fault-tolerance 生命周期，无法再拆成独立问题。 diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-33369/solution/code.patch b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/solution/code.patch new file mode 100644 index 000000000..df562648e --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/solution/code.patch @@ -0,0 +1,550 @@ +diff --git a/python/paddle/distributed/fleet/elastic.py b/python/paddle/distributed/fleet/elastic.py +new file mode 100644 +index 0000000000000000000000000000000000000000..b919c4737576d5947265004854a8d1f66450a8fc +--- /dev/null ++++ b/python/paddle/distributed/fleet/elastic.py +@@ -0,0 +1,312 @@ ++# Copyright (c) 2021 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 time ++import socket ++import os ++import six ++import logging ++import signal ++ ++logging.basicConfig(level=os.environ.get('LOGLEVEL', 'INFO').upper()) ++logger = logging.getLogger("ELASTIC") ++ ++ELASTIC_EXIT_CODE = 101 ++ ++ ++class ElasticStatus: ++ COMPLETED = "completed" ++ ERROR = "error" ++ HOLD = "hold" ++ RESTART = "restart" ++ EXIT = "exit" ++ ++ ++class LauncherInterface(object): ++ def __init__(self, args): ++ self.args = args ++ self.procs = [] ++ ++ def _terminate_procs(self): ++ for p in self.procs: ++ if p.proc.poll() is None: ++ p.proc.terminate() ++ if p.log_fn: ++ p.log_fn.close() ++ logger.info("terminate process id:{}".format(p.proc.pid)) ++ ++ for step in range(0, 50): ++ alive = False ++ for p in self.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 True ++ ++ time.sleep(1) ++ return False ++ ++ def _check_procs(self): ++ alive = False ++ result = None ++ for p in self.procs: ++ ret = p.proc.poll() ++ if ret is None: ++ alive = True ++ elif ret != 0: ++ logger.error("ERROR rank {} error with code {}".format(p.rank, ++ ret)) ++ result = ret ++ if not alive and result is None: ++ return 0 ++ else: ++ return result ++ ++ def launch(self): ++ raise NotImplementedError ++ ++ def stop(self): ++ raise NotImplementedError ++ ++ def watch(self): ++ raise NotImplementedError ++ ++ ++class ElasticManager(object): ++ def __init__(self, args): ++ ++ self.args = args ++ server = args.elastic_server or os.getenv('PADDLE_ELASTIC_SERVER') ++ name = args.job_id or os.getenv('PADDLE_ELASTIC_JOB_ID') ++ np = args.np or int(os.getenv('PADDLE_ELASTIC_NP', 0)) ++ host = args.host or os.getenv('POD_IP') ++ scale = args.scale or int(os.getenv('PADDLE_ELASTIC_SCALE', 0)) ++ force = args.force or os.getenv('PADDLE_ELASTIC_FORCE') ++ ++ self.endpoints = os.getenv('DISTRIBUTED_TRAINER_ENDPOINTS', '') ++ self.trainers = os.getenv('PADDLE_TRAINERS', '') ++ ++ self.elastic_level = int( ++ os.getenv('PADDLE_ELASTIC_FAULT_TOLERANC_LEVEL', 1)) ++ ++ #elastic_timeout = os.getenv('PADDLE_ELASTIC_TIMEOUT',1) ++ ++ logger.debug('init with server {} host {}'.format(server, host)) ++ ++ self.hosts = [] ++ self.stopped = False ++ ++ self.sigint = 0 ++ ++ if not server or ':' not in server or not name or not np: ++ logger.info( ++ 'Elastic is not enabled with server {} name {} and np {}'. ++ format(server, name, np)) ++ self.enable = False ++ return ++ else: ++ self.enable = True ++ ++ import etcd3 ++ ++ srv, port = server.split(':') ++ self.etcd = etcd3.client(host=srv, port=port) ++ self.host = host if host else self._get_host() ++ ++ # etcd data ++ self.prefix = "/paddle/" + name ++ self.node_prefix = self.prefix + '/nodes/' ++ self.np_path = self.prefix + '/np' ++ self.endpoints_path = self.prefix + '/endpoints' ++ self.host_path = '{}{}'.format(self.node_prefix, time.time()) ++ ++ self.np = np + scale ++ ''' ++ 0 group mode, be aware of healthy status of other workers ++ 1 decouple mode, check own status only ++ ''' ++ self.etcd.put(self.prefix, b'0') ++ ++ # host ++ # register self host to etcd ++ # register watch to reset host after host been deleted ++ self.etcd.delete_prefix(self.node_prefix) ++ ++ def host_call_back(event): ++ if self.etcd.get(self.host_path)[0] == None: ++ # ensure unmatch trigger ++ logger.info('register host again {}'.format(self.host)) ++ time.sleep(5) ++ ++ self.etcd.put(self.host_path, six.b(self.host)) ++ ++ host_watch = self.etcd.add_watch_callback(self.host_path, ++ host_call_back) ++ self.etcd.put(self.host_path, six.b(self.host)) ++ ++ # np describes the exact number of nodes to run the job ++ inp = int(self.etcd.get(self.np_path)[0] or 0) ++ if scale == 0 and not force: ++ assert inp == np or inp == 0, "np {} is not consistent with np in etcd {}".format( ++ np, inp) ++ else: ++ assert inp == np or inp == self.np, "np {} scale to {} by {} is not allowed".format( ++ inp, self.np, scale) ++ ++ self.etcd.put(self.np_path, six.b("%d" % (self.np))) ++ ++ def np_call_back(event): ++ gnp = int(self.etcd.get(self.np_path)[0]) ++ if gnp != self.np: ++ logger.info("scale np {} to {} ".format(self.np, gnp)) ++ self.np = gnp ++ ++ np_watch = self.etcd.add_watch_callback(self.np_path, np_call_back) ++ ++ # endpoints handle DISTRIBUTED_TRAINER_ENDPOINTS and PADDLE_TRAINERS ++ self.etcd.put(self.endpoints_path, ++ six.b('{}|{}'.format(self.endpoints, self.trainers))) ++ ++ def endpoints_call_back(event): ++ if not self.endpoints: ++ return ++ edps = six.ensure_str(self.etcd.get(self.endpoints_path)[0] or '') ++ self.endpoints, self.trainers = edps.split('|') ++ logger.info("set DISTRIBUTED_TRAINER_ENDPOINTS {} ".format( ++ self.endpoints)) ++ logger.info("set PADDLE_TRAINERS {} ".format(self.trainers)) ++ ++ endpoints_watch = self.etcd.add_watch_callback(self.endpoints_path, ++ endpoints_call_back) ++ ++ self.watches = [host_watch, np_watch, endpoints_watch] ++ ++ def exit(self, completed=False): ++ logger.info('manager exist completed {}'.format(completed)) ++ ++ if not self.enable: ++ return ++ ++ if completed: ++ self.etcd.put(self.prefix, b'1') ++ ++ for watch in self.watches: ++ self.etcd.cancel_watch(watch) ++ self.etcd.delete(self.host_path) ++ ++ hosts = [i for i in self.etcd.get_prefix(self.node_prefix)] ++ if len(hosts) == 0: ++ self.etcd.delete_prefix(self.prefix) ++ ++ def _get_host(self): ++ try: ++ return socket.gethostbyname(socket.getfqdn(socket.gethostname())) ++ except: ++ return '127.0.0.1' ++ ++ def _completed(self): ++ if not self.enable: ++ return True ++ ++ return int(self.etcd.get(self.prefix)[0]) == 1 ++ ++ def _match(self): ++ self.hosts = [ ++ six.ensure_str(i[0]) for i in self.etcd.get_prefix(self.node_prefix) ++ ] ++ if len(self.hosts) == self.np: ++ return True ++ else: ++ return False ++ ++ def _update_hosts(self): ++ assert len(self.hosts) != 0, 'hosts empty' ++ ++ if self.host in self.endpoints: ++ os.environ['DISTRIBUTED_TRAINER_ENDPOINTS'] = self.endpoints ++ os.environ['PADDLE_TRAINERS'] = self.trainers ++ logger.info("update env DISTRIBUTED_TRAINER_ENDPOINTS {} ".format( ++ self.endpoints)) ++ logger.info("update env PADDLE_TRAINERS {} ".format(self.trainers)) ++ return ++ ++ rank = int(os.getenv('PADDLE_TRAINER_ID', -1)) ++ idx = self.hosts.index(self.host) ++ ++ # swap if self.host not in the right position ++ if rank >= 0: ++ self.hosts[idx] = self.hosts[rank] ++ self.hosts[rank] = self.host ++ else: ++ os.environ['PADDLE_TRAINER_ID'] = '{}'.format(idx) ++ ++ hosts = ','.join(self.hosts) ++ self.args.ips = hosts ++ os.environ['PADDLE_TRAINERS'] = hosts ++ ++ def wait(self): ++ if not self.enable: ++ return ++ ++ while not self.stopped: ++ if self._match(): ++ logger.info('ready with hosts {}'.format(self.hosts)) ++ self._update_hosts() ++ return ++ logger.info('not ready for np {} with hosts {}'.format(self.np, ++ self.hosts)) ++ time.sleep(3) ++ return ++ ++ def run(self, launcher): ++ if self.stopped: ++ return ++ ++ self.launcher = launcher(self.args) ++ self.launcher.launch() ++ ++ def watch(self): ++ ++ while not self.stopped: ++ ret = self.launcher.watch() ++ ++ if ret is not None: # self terminated ++ logger.info('job exit with code {}'.format(ret)) ++ # process is completed if ret >= 0 or error else ++ completed = True if ret == 0 else False ++ self.launcher.stop() ++ self.exit(completed=completed) ++ if completed: ++ return ElasticStatus.COMPLETED ++ if self.elastic_level == 1: ++ return ElasticStatus.RESTART ++ else: ++ return ElasticStatus.ERROR ++ ++ if not self._completed() and not self._match(): ++ self.launcher.stop() ++ return ElasticStatus.HOLD ++ ++ time.sleep(3) ++ ++ return ElasticStatus.EXIT ++ ++ def signal_handler(self, sigint, frame): ++ if self.enable: ++ self.exit() ++ self.sigint = sigint ++ self.stopped = True +diff --git a/python/paddle/distributed/fleet/launch.py b/python/paddle/distributed/fleet/launch.py +index 25b10133191788cd85085fd4612ae7cea0f122f3..07862a07c92c419c88869e2544414ba4e63141e0 100644 +--- a/python/paddle/distributed/fleet/launch.py ++++ b/python/paddle/distributed/fleet/launch.py +@@ -69,12 +69,18 @@ from argparse import ArgumentParser, REMAINDER + import paddle + import paddle.fluid as fluid + from paddle.distributed.fleet import launch_utils ++import signal + + # TODO(danleifeng): Don't import * from a module + from paddle.distributed.fleet.launch_utils import * + import paddle.distributed.fleet.cloud_utils as cloud_utils + import paddle.distributed.fleet.ascend_utils as ascend_utils + ++from paddle.distributed.fleet.elastic import ElasticManager ++from paddle.distributed.fleet.elastic import LauncherInterface ++from paddle.distributed.fleet.elastic import ElasticStatus ++from paddle.distributed.fleet.elastic import ELASTIC_EXIT_CODE ++ + __all__ = [] + + +@@ -175,6 +181,18 @@ see: http://www.paddlepaddle.org/documentation/docs/zh/1.6/user_guides/howto/tra + "--heter_worker_num", type=int, help="number of heter_workers") + ps_group.add_argument("--http_port", type=int, help="Gloo http Port") + ++ # parameter elastic mode ++ elastic_group = parser.add_argument_group("Elastic Parameters") ++ elastic_group.add_argument( ++ "--elastic_server", type=str, help="etcd server host:port") ++ elastic_group.add_argument("--job_id", type=str, help="job unique id") ++ elastic_group.add_argument("--np", type=int, help="job pod/node number") ++ elastic_group.add_argument("--scale", type=int, default=0, help="scale np") ++ elastic_group.add_argument( ++ "--host", type=str, help="bind host, default to POD_IP env") ++ elastic_group.add_argument( ++ "--force", type=bool, default=False, help="update np force") ++ + return parser.parse_args() + + +@@ -183,7 +201,10 @@ def get_cluster_from_args(args, device_mode, devices_per_proc): + if len(node_ips) == 1: + node_ip = node_ips[0] + else: +- _, node_ip = get_host_name_ip() ++ if args.host: ++ node_ip = args.host ++ else: ++ _, node_ip = get_host_name_ip() + + assert node_ip in node_ips, "Can't find your local ip {%s} in node_ips: {%s}" \ + % (node_ip, node_ips) +@@ -214,65 +235,75 @@ def get_cluster_from_args(args, device_mode, devices_per_proc): + devices_per_proc) + + +-def launch_collective(args): +- # parse arguments, used for cloud-single-machine and local +- (device_mode, devices_per_proc) = launch_utils.get_device_proc_info(args) +- trainers_num = cloud_utils.get_trainers_num() +- logger.debug("parsed from args trainerss_num:{} mode:{} devices:{}".format( +- trainers_num, device_mode, devices_per_proc)) +- +- cluster = None +- pod = None +- +- start_port = 6170 +- if os.environ.get('FLAGS_START_PORT') is not None: +- start_port = os.environ.get('FLAGS_START_PORT') +- if cloud_utils.use_paddlecloud() and trainers_num != 1: +- cluster, pod = cloud_utils.get_cloud_cluster( +- args.ips, device_mode, devices_per_proc, start_port) +- logger.debug("get cluster from cloud:{}".format(cluster)) +- elif device_mode == DeviceMode.ASCEND_NPU: +- # for ascend +- cluster, pod = ascend_utils.get_cloud_cluster( +- rank_table_file=os.getenv("RANK_TABLE_FILE", None), +- device_mode=device_mode, +- start_port=start_port) +- else: +- # trainers_num = 1 or not use paddlecloud ips="a,b" +- cluster, pod = get_cluster_from_args(args, device_mode, +- devices_per_proc) +- logger.debug("get cluster from args:{}".format(cluster)) +- +- global_envs = copy.copy(os.environ.copy()) +- gloo_rendezvous_dir = tempfile.mkdtemp() +- # add gloo env +- global_envs["PADDLE_WITH_GLOO"] = str(os.getenv("PADDLE_WITH_GLOO", "0")) +- global_envs["PADDLE_GLOO_RENDEZVOUS"] = "3" +- global_envs["PADDLE_GLOO_FS_PATH"] = gloo_rendezvous_dir +- +- procs = start_local_trainers( +- cluster, +- pod, +- training_script=args.training_script, +- training_script_args=args.training_script_args, +- log_dir=args.log_dir, +- envs=global_envs) +- +- for idx, proc in enumerate(procs): +- print("launch proc_id:{} idx:{}".format(proc.proc.pid, idx)) ++class CollectiveLauncher(LauncherInterface): ++ def __init__(self, args): ++ self.args = args ++ self.procs = [] + +- while True: +- alive = watch_local_trainers(procs, cluster.trainers_nranks()) ++ def launch(self): ++ logger.info("collective lauchner launch ...") ++ args = self.args ++ # parse arguments, used for cloud-single-machine and local ++ (device_mode, ++ devices_per_proc) = launch_utils.get_device_proc_info(args) ++ trainers_num = cloud_utils.get_trainers_num() ++ logger.debug("parsed from args trainerss_num:{} mode:{} devices:{}". ++ format(trainers_num, device_mode, devices_per_proc)) + +- if not alive: +- logger.info("Local processes completed.") +- logger.debug("POD info:{}".format(pod)) +- break ++ cluster = None ++ pod = None + +- time.sleep(3) +- +- if os.path.exists(gloo_rendezvous_dir): +- shutil.rmtree(gloo_rendezvous_dir) ++ start_port = 6170 ++ if os.environ.get('FLAGS_START_PORT') is not None: ++ start_port = os.environ.get('FLAGS_START_PORT') ++ if cloud_utils.use_paddlecloud() and trainers_num != 1: ++ cluster, pod = cloud_utils.get_cloud_cluster( ++ args.ips, device_mode, devices_per_proc, start_port) ++ logger.debug("get cluster from cloud:{}".format(cluster)) ++ elif device_mode == DeviceMode.ASCEND_NPU: ++ # for ascend ++ cluster, pod = ascend_utils.get_cloud_cluster( ++ rank_table_file=os.getenv("RANK_TABLE_FILE", None), ++ device_mode=device_mode, ++ start_port=start_port) ++ else: ++ # trainers_num = 1 or not use paddlecloud ips="a,b" ++ cluster, pod = get_cluster_from_args(args, device_mode, ++ devices_per_proc) ++ logger.debug("get cluster from args:{}".format(cluster)) ++ ++ global_envs = copy.copy(os.environ.copy()) ++ self.gloo_rendezvous_dir = tempfile.mkdtemp() ++ # add gloo env ++ global_envs["PADDLE_WITH_GLOO"] = str( ++ os.getenv("PADDLE_WITH_GLOO", "0")) ++ global_envs["PADDLE_GLOO_RENDEZVOUS"] = "3" ++ global_envs["PADDLE_GLOO_FS_PATH"] = self.gloo_rendezvous_dir ++ ++ self.procs = start_local_trainers( ++ cluster, ++ pod, ++ training_script=args.training_script, ++ training_script_args=args.training_script_args, ++ log_dir=args.log_dir, ++ envs=global_envs) ++ ++ for idx, proc in enumerate(self.procs): ++ logger.info("launch proc_id:{} idx:{}".format(proc.proc.pid, idx)) ++ ++ def stop(self): ++ logger.info("collective lauchner stop ...") ++ self._terminate_procs() ++ if os.path.exists(self.gloo_rendezvous_dir): ++ shutil.rmtree(self.gloo_rendezvous_dir) ++ ++ def watch(self): ++ logger.debug("collective lauchner watch ...") ++ for p in self.procs: ++ if p.log_fn and p.local_rank == 0: ++ pull_worker_log(p) ++ ret = self._check_procs() ++ return ret + + + def launch_ps(args, distribute_mode): +@@ -367,10 +398,42 @@ def launch(): + _print_arguments(args) + + distribute_mode = which_distributed_mode(args) +- if distribute_mode == DistributeMode.COLLECTIVE: +- launch_collective(args) +- else: ++ # TODO(kuizhiqing) support ps later ++ if not distribute_mode == DistributeMode.COLLECTIVE: + launch_ps(args, distribute_mode) ++ return ++ ++ elastic = ElasticManager(args) ++ ++ signal.signal(signal.SIGTERM, elastic.signal_handler) ++ signal.signal(signal.SIGABRT, elastic.signal_handler) ++ signal.signal(signal.SIGINT, elastic.signal_handler) ++ ++ while True: ++ ++ # wait for all nodes ready to run ++ elastic.wait() ++ ++ # run self with specified launcher ++ elastic.run(CollectiveLauncher) ++ ++ # keep wathing the health status of self and being notified for other's failure ++ ret = elastic.watch() ++ if ret == ElasticStatus.COMPLETED: ++ break ++ if ret == ElasticStatus.HOLD: ++ continue ++ if ret == ElasticStatus.EXIT: ++ break ++ if ret == ElasticStatus.ERROR: ++ sys.exit(3) ++ if ret == ElasticStatus.RESTART: ++ sys.exit(ELASTIC_EXIT_CODE) ++ ++ if int(elastic.sigint) > 0: ++ sys.exit(128 + int(elastic.sigint)) ++ else: ++ sys.exit(0) + + + if __name__ == "__main__": diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-33369/tests/test.patch b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/tests/test.patch new file mode 100644 index 000000000..d7afc9022 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/tests/test.patch @@ -0,0 +1,183 @@ +diff --git a/test/swe_paddle/test_pr33369_elastic_fault_tolerance.py b/test/swe_paddle/test_pr33369_elastic_fault_tolerance.py +new file mode 100644 +index 0000000000000000000000000000000000000000..2d23cde85475cfe79ce491e0025f3265bc6a5d1e +--- /dev/null ++++ b/test/swe_paddle/test_pr33369_elastic_fault_tolerance.py +@@ -0,0 +1,177 @@ ++import ast ++import importlib.util ++from pathlib import Path ++from types import SimpleNamespace ++ ++ ++REPO_ROOT = Path(__file__).resolve().parents[2] ++ELASTIC = REPO_ROOT / "python/paddle/distributed/fleet/elastic.py" ++LAUNCH = REPO_ROOT / "python/paddle/distributed/fleet/launch.py" ++ ++ ++def _load_function(path, name, namespace): ++ tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) ++ node = next( ++ item ++ for item in tree.body ++ if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)) ++ and item.name == name ++ ) ++ module = ast.Module(body=[node], type_ignores=[]) ++ ast.fix_missing_locations(module) ++ exec(compile(module, str(path), "exec"), namespace) ++ return namespace[name] ++ ++ ++def _load_elastic_module(): ++ if not ELASTIC.exists(): ++ raise AssertionError( ++ "collective launch has no elastic fault-tolerance manager" ++ ) ++ spec = importlib.util.spec_from_file_location( ++ "swe_pr33369_elastic", ELASTIC ++ ) ++ module = importlib.util.module_from_spec(spec) ++ spec.loader.exec_module(module) ++ return module ++ ++ ++def test_parameter_server_mode_is_preserved(): ++ class DistributeMode: ++ COLLECTIVE = "collective" ++ PS = "ps" ++ PS_HETER = "ps-heter" ++ ++ which_distributed_mode = _load_function( ++ LAUNCH, ++ "which_distributed_mode", ++ {"DistributeMode": DistributeMode}, ++ ) ++ assert ( ++ which_distributed_mode(SimpleNamespace(run_mode="ps")) ++ == DistributeMode.PS ++ ) ++ ++ ++def test_failed_training_requests_an_elastic_restart(): ++ module = _load_elastic_module() ++ events = [] ++ ++ class Launcher: ++ def watch(self): ++ return 7 ++ ++ def stop(self): ++ events.append("stop") ++ ++ manager = object.__new__(module.ElasticManager) ++ manager.stopped = False ++ manager.launcher = Launcher() ++ manager.elastic_level = 1 ++ manager.exit = lambda completed=False: events.append( ++ ("exit", completed) ++ ) ++ ++ assert manager.watch() == module.ElasticStatus.RESTART ++ assert events == ["stop", ("exit", False)] ++ ++ ++def test_membership_change_holds_the_current_training_group(): ++ module = _load_elastic_module() ++ events = [] ++ ++ class Launcher: ++ def watch(self): ++ return None ++ ++ def stop(self): ++ events.append("stop") ++ ++ manager = object.__new__(module.ElasticManager) ++ manager.stopped = False ++ manager.launcher = Launcher() ++ manager._completed = lambda: False ++ manager._match = lambda: False ++ ++ assert manager.watch() == module.ElasticStatus.HOLD ++ assert events == ["stop"] ++ ++ ++def test_collective_launch_is_managed_by_the_elastic_lifecycle(): ++ events = [] ++ ++ class DistributeMode: ++ COLLECTIVE = "collective" ++ PS = "ps" ++ ++ class ElasticStatus: ++ COMPLETED = "completed" ++ HOLD = "hold" ++ EXIT = "exit" ++ ERROR = "error" ++ RESTART = "restart" ++ ++ class Manager: ++ def __init__(self, args): ++ events.append("manager") ++ self.sigint = 0 ++ self.signal_handler = lambda *args: None ++ ++ def wait(self): ++ events.append("wait") ++ ++ def run(self, launcher): ++ events.append(("run", launcher)) ++ ++ def watch(self): ++ events.append("watch") ++ return ElasticStatus.COMPLETED ++ ++ class FakeSignal: ++ SIGTERM = 15 ++ SIGABRT = 6 ++ SIGINT = 2 ++ ++ @staticmethod ++ def signal(sig, handler): ++ events.append(("signal", sig)) ++ ++ class FakeSys: ++ @staticmethod ++ def exit(code): ++ events.append(("exit", code)) ++ ++ collective_launcher = object() ++ namespace = { ++ "_parse_args": lambda: SimpleNamespace(), ++ "get_logger": lambda: object(), ++ "_print_arguments": lambda args: None, ++ "which_distributed_mode": lambda args: DistributeMode.COLLECTIVE, ++ "DistributeMode": DistributeMode, ++ "launch_ps": lambda *args: events.append("ps"), ++ "launch_collective": lambda *args: events.append( ++ "unmanaged_collective" ++ ), ++ "ElasticManager": Manager, ++ "signal": FakeSignal, ++ "ElasticStatus": ElasticStatus, ++ "CollectiveLauncher": collective_launcher, ++ "sys": FakeSys, ++ "ELASTIC_EXIT_CODE": 101, ++ } ++ launch = _load_function(LAUNCH, "launch", namespace) ++ launch() ++ ++ assert "unmanaged_collective" not in events ++ assert events[:4] == [ ++ "manager", ++ ("signal", FakeSignal.SIGTERM), ++ ("signal", FakeSignal.SIGABRT), ++ ("signal", FakeSignal.SIGINT), ++ ] ++ assert events[4:] == [ ++ "wait", ++ ("run", collective_launcher), ++ "watch", ++ ("exit", 0), ++ ] diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-33369/tests/test.sh b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/tests/test.sh new file mode 100644 index 000000000..fa4c96ee5 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-33369/tests/test.sh @@ -0,0 +1,4 @@ +#!/usr/bin/env bash + +set -euo pipefail +python -m pytest test/swe_paddle/test_pr33369_elastic_fault_tolerance.py -q