diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-36684/README.md b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/README.md new file mode 100644 index 000000000..dea62e6d1 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/README.md @@ -0,0 +1,43 @@ +# PaddlePaddle__Paddle-36684 + +This directory converts Paddle PR #36684 into a SWE-Paddle community task candidate. + +## Source + +| Field | Value | +| --- | --- | +| Repo | `PaddlePaddle/Paddle` | +| PR | [36684](https://github.com/PaddlePaddle/Paddle/pull/36684) | +| PR title | `fleet support elastic scale up/down` | +| Base commit | `9a9345fa4dc77be655811d8e484b99cb9ff5f356` | +| Merged at | `2021-11-11T06:27:42Z` | +| Task type | `feature_enhancement` | +| Resource | CPU | + +## Summary + +Enable Fleet elastic jobs to use a node-count range and keep ranks, hosts, and endpoints consistent across scale-out and scale-in transitions. + +## Why This Is A Good SWE-Paddle Candidate + +- It covers a real distributed-launch workflow with clear behavior before and after the change. +- The production change spans range parsing, readiness decisions, host membership, and endpoint updates rather than a single local condition. +- The candidate executes assertions and scenarios from the source PR tests without requiring a live etcd service or GPU. + +## 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`: source-PR tests plus the controlled source adapter. +- `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-36684/environment/README.md b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/environment/README.md new file mode 100644 index 000000000..14f5dfd84 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/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: `9a9345fa4dc77be655811d8e484b99cb9ff5f356` +- Resource: CPU +- GPU required: no +- Build path: no source build is required; a controlled AST overlay executes the checkout implementation with the source PR's unittest methods and local doubles for etcd and device discovery. + +## 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-36684/instruction.md b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/instruction.md new file mode 100644 index 000000000..89c2aa5a4 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/instruction.md @@ -0,0 +1,19 @@ +# 支持 Fleet 弹性训练按节点范围扩缩容 + +## 详细描述 + +目前 Fleet 弹性训练只能按固定节点数启动。实际运行中，可用机器数量可能会随资源调度发生变化，用户即使配置了类似 `2:4` 的节点范围，也无法让任务在 2 到 4 个节点之间正常等待、启动和扩缩容。 + +需要支持用最小值和最大值描述可接受的节点数量。达到最大节点数时可以立即开始；节点数位于范围内时可以等待更多节点，并在等待超时后使用当前节点继续运行；低于最小节点数时继续等待。运行期间节点加入或退出后，训练进程应拿到正确的主机和 endpoint 信息。 + +## 验收说明 + +- 支持固定节点数和 `MIN:MAX` 两种配置方式，并正确判断当前节点是否满足启动条件。 +- 节点扩容或缩容后，训练使用的主机列表、rank 和 endpoint 信息应与当前节点一致。 +- 原有固定节点数下的故障恢复和 launch 流程保持不变。 + +## 技术要求 + +- 熟悉 Python。 +- 熟悉 PaddlePaddle Fleet 分布式启动流程。 +- 了解弹性训练中的节点发现和进程环境变量。 diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-36684/proposal.md b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/proposal.md new file mode 100644 index 000000000..0695a607e --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/proposal.md @@ -0,0 +1,55 @@ +# Task Proposal: PaddlePaddle__Paddle-36684 + +## 1. 来源信息 + +- Instance ID：`PaddlePaddle__Paddle-36684` +- PR 链接：https://github.com/PaddlePaddle/Paddle/pull/36684 +- PR 标题：`fleet support elastic scale up/down` +- `base_commit`：`9a9345fa4dc77be655811d8e484b99cb9ff5f356` +- merged 时间：`2021-11-11T06:27:42Z` +- 你的身份：熟悉该模块的 contributor +- 后续联系人：TBD + +## 2. 问题一句话 + +Fleet 弹性训练需要接受节点数量范围，并在节点加入或退出后更新训练所需的主机、rank 和 endpoint 信息。 + +## 3. 为什么适合作为 SWE-Paddle 样本 + +- **真实性**：来源于 Fleet 弹性训练对动态资源调度的实际支持需求。 +- **代表性**：覆盖节点范围解析、启动条件判断和扩缩容后的环境更新。 +- **边界清楚**：production change 集中在 Python 分布式 elastic launch 流程。 +- **非平凡性**：需要同时处理固定节点模式、范围模式、超时等待以及扩缩容后的 rank/endpoint 连续性。 +- **环境友好性**：测试使用来源 PR 的 unittest 场景和 controlled doubles，不需要启动真实 etcd、训练进程或 GPU。 + +## 4. 任务类型和标签 + +- 任务类型：`feature_enhancement` +- 执行后端：`cpu` +- 设备范围：`cpu_only` +- 模块标签：`[distributed, fleet, elastic, launch]` + +## 5. 验证思路 + +- 目标测试命令：`bash tests/test.sh` +- 目标测试文件：`test/swe_paddle/test_pr36684_elastic_scale.py` +- 修复前预期：固定节点的 elastic enablement 继续通过；来源 PR 中范围匹配和扩缩容用例因 Base 不支持范围节点数而失败。 +- 修复后预期：固定节点 P2P 与两个范围扩缩容 F2P 全部通过。 +- P2P 候选：来源 PR `TestElasticInit.test_enable_elastic` 对既有 elastic enablement 的验证。 + +## 6. 环境与资源 + +- 资源需求：CPU +- Paddle 来源：`PaddlePaddle/Paddle` source checkout at `base_commit` +- 是否能提供 Docker：暂无 +- patch 类型：Python-only +- 环境建议：使用 AST overlay 执行 checkout 中的真实控制流，并直接调用来源 PR 的 unittest 方法；etcd 和设备探测使用 controlled doubles。 +- 最小测试命令：`bash tests/test.sh` +- 是否有 oracle 日志：由 SWE-Paddle verifier 结果另行维护 + +## 7. 风险自查 + +- 泄露风险：instruction 只描述外部行为，没有给出 Gold patch 的具体修改方式。 +- 环境风险：无需真实 etcd、分布式集群、网络服务或 GPU。 +- flaky 风险：不依赖真实节点变化、计时竞态或后台训练进程。 +- 拆分风险：测试只覆盖节点范围匹配及扩缩容环境更新这一组相互依赖的行为。 diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-36684/solution/code.patch b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/solution/code.patch new file mode 100644 index 000000000..9434ab825 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/solution/code.patch @@ -0,0 +1,576 @@ +diff --git a/python/paddle/distributed/elastic.py b/python/paddle/distributed/elastic.py +index e6f21f6603d8dad6c802663e9ea6940d7eb0826e..52f36a227f1c86a79933d9088036b0cc0b2f0df8 100644 +--- a/python/paddle/distributed/elastic.py ++++ b/python/paddle/distributed/elastic.py +@@ -50,7 +50,10 @@ if __name__ == '__main__': + parser.add_argument( + "--elastic_server", type=str, help="etcd server host:port") + parser.add_argument("--job_id", type=str, help="job unique id") +- parser.add_argument("--np", type=int, help="job pod/node number") ++ parser.add_argument( ++ "--np", ++ type=str, ++ help="job pod/node number, need to be 'MIN' or 'MIN:MAX' format") + parser.add_argument("action", type=str, help="action to take") + + args = parser.parse_args() +@@ -58,7 +61,7 @@ if __name__ == '__main__': + 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)) ++ np = int(args.np.split(":")[0]) or int(os.getenv('PADDLE_ELASTIC_NP', 0)) + + cmd = Command(server, name) + +diff --git a/python/paddle/distributed/fleet/elastic/__init__.py b/python/paddle/distributed/fleet/elastic/__init__.py +index 1ac81729d5430a5b5174e6b07bda1ca0d0f5a971..b928d82fb399f3cfcef56c78a8054a1612b3c8c0 100644 +--- a/python/paddle/distributed/fleet/elastic/__init__.py ++++ b/python/paddle/distributed/fleet/elastic/__init__.py +@@ -33,7 +33,7 @@ def enable_elastic(args, distribute_mode): + if not args.job_id and not os.getenv('PADDLE_ELASTIC_JOB_ID'): + return False + +- if not args.np and not int(os.getenv('PADDLE_ELASTIC_NP', 0)): ++ if not args.np and not os.getenv('PADDLE_ELASTIC_NP'): + return False + + return True +@@ -41,7 +41,11 @@ def enable_elastic(args, distribute_mode): + + def launch_elastic(args, distribute_mode): + +- elastic = ElasticManager(args) ++ server = args.elastic_server or os.getenv('PADDLE_ELASTIC_SERVER') ++ srv, port = server.split(':') ++ import etcd3 ++ etcd_client = etcd3.client(host=srv, port=port) ++ elastic = ElasticManager(args, etcd_client) + + signal.signal(signal.SIGTERM, elastic.signal_handler) + signal.signal(signal.SIGABRT, elastic.signal_handler) +diff --git a/python/paddle/distributed/fleet/elastic/collective.py b/python/paddle/distributed/fleet/elastic/collective.py +index 83f0e85db2badfeff4865f974dd2c9ef2053b0a1..d9c2735c4bd0191ac4c71e3a4301addd0192dc93 100644 +--- a/python/paddle/distributed/fleet/elastic/collective.py ++++ b/python/paddle/distributed/fleet/elastic/collective.py +@@ -12,6 +12,7 @@ + # See the License for the specific language governing permissions and + # limitations under the License. + ++import tempfile + from paddle.distributed.fleet import launch_utils + from paddle.distributed.fleet import cloud_utils + from paddle.distributed.fleet import ascend_utils +diff --git a/python/paddle/distributed/fleet/elastic/manager.py b/python/paddle/distributed/fleet/elastic/manager.py +index 2a344e92765d94e43079f7a4ed14aa51447ae179..cfb0d1ead5aa151f9239588be8d4585bec00186a 100644 +--- a/python/paddle/distributed/fleet/elastic/manager.py ++++ b/python/paddle/distributed/fleet/elastic/manager.py +@@ -16,9 +16,14 @@ import time + import socket + import os + import six ++import copy + import logging + import signal + import random ++import threading ++import traceback ++from paddle.distributed.fleet import cloud_utils ++from paddle.distributed.fleet import launch_utils + + logger = logging.getLogger("ELASTIC") + logger.setLevel(logging.INFO) +@@ -30,6 +35,18 @@ logger.addHandler(ch) + + ELASTIC_EXIT_CODE = 101 + ++# wait for timeout, unit: seconds ++ELASTIC_TIMEOUT = 2 * 60 ++ ++# keepalived ttl, unit: seconds ++ELASTIC_TTL = 60 ++ ++ ++# 1: Fault tolerance, 2: Elastic ++class ElasticLevel: ++ FAULT_TOLERANCE = 1 ++ ELASTIC = 2 ++ + + class ElasticStatus: + COMPLETED = "completed" +@@ -106,21 +123,52 @@ class LauncherInterface(object): + + + class ElasticManager(object): +- def __init__(self, args): ++ def __init__(self, args, etcd_client): + + 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)) ++ self.min_np, self.max_np = self._parse_np(args.np) + 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', '') ++ start_port = 6170 ++ if os.environ.get('FLAGS_START_PORT') is not None: ++ start_port = int(os.environ.get('FLAGS_START_PORT')) ++ if cloud_utils.use_paddlecloud(): ++ start_port = int(os.getenv("PADDLE_PORT", "")) + ++ (self.device_mode, ++ self.devices_per_proc) = launch_utils.get_device_proc_info(args) ++ ++ self.elastic_timeout = int( ++ os.getenv('PADDLE_ELASTIC_TIMEOUT', ELASTIC_TIMEOUT)) ++ elastic_ttl = int(os.getenv('PADDLE_ELASTIC_TTL', ELASTIC_TTL)) ++ self.dist_endpoints = os.getenv('DISTRIBUTED_TRAINER_ENDPOINTS', '') ++ self.trainers = os.getenv('PADDLE_TRAINERS', '') ++ self.all_host_endpoints = os.getenv('PADDLE_TRAINER_ENDPOINTS', ++ '').split(",") ++ self.np = len(self.all_host_endpoints) ++ logger.info(f'start job with np={self.np}') ++ ++ #[ "%s:%d" % (ip, start_port) for ip in self.trainers.split(",")] ++ logger.info( ++ f"trainers={self.trainers}, all_host_endpoints={self.all_host_endpoints}" ++ ) ++ ++ # auto correct the value of elastic_level ++ # 1: Fault tolerant, 2: Elastic + self.elastic_level = int( +- os.getenv('PADDLE_ELASTIC_FAULT_TOLERANC_LEVEL', 1)) ++ os.getenv('PADDLE_ELASTIC_FAULT_TOLERANC_LEVEL', ++ ElasticLevel.FAULT_TOLERANCE)) ++ if self.min_np == self.max_np or \ ++ (self.min_np > 0 and self.max_np == 0): ++ self.elastic_level = ElasticLevel.FAULT_TOLERANCE ++ logger.info(f'start job with ElasticLevel.FAULT_TOLERANCE') ++ if self.min_np > 0 and self.max_np > self.min_np: ++ self.elastic_level = ElasticLevel.ELASTIC ++ logger.info(f'start job with ElasticLevel.ELASTIC') + + # compatible with kuberntes service discovery + if not server and os.getenv( +@@ -130,8 +178,6 @@ class ElasticManager(object): + os.getenv('PADDLE_ELASTIC_ETCD_SERVICE_HOST'), + os.getenv('PADDLE_ELASTIC_ETCD_SERVICE_PORT')) + +- #elastic_timeout = os.getenv('PADDLE_ELASTIC_TIMEOUT',1) +- + logger.debug('init with server {} host {}'.format(server, host)) + + self.hosts = [] +@@ -140,20 +186,20 @@ class ElasticManager(object): + self.sigint = 0 + self.need_sync = False + +- if not server or ':' not in server or not name or not np: ++ self.elastic_startup_time = None ++ ++ if not server or ':' not in server or not name or not self.np: + logger.info( + 'Elastic is not enabled with server {} name {} and np {}'. +- format(server, name, np)) ++ format(server, name, self.np)) + self.enable = False + return + else: + self.enable = True + +- import etcd3 +- +- srv, port = server.split(':') +- self.etcd = etcd3.client(host=srv, port=port) ++ self.etcd = etcd_client + self.host = host if host else self._get_host() ++ self.host_port = "%s:%d" % (self.host, start_port) + + # etcd data + self.prefix = "/paddle/" + name +@@ -165,67 +211,75 @@ class ElasticManager(object): + random.choice('abcdefghijklmnopqrstuvwxyz') for _ in range(6)) + self.host_path = '{}/{}{}'.format(self.node_prefix, node_tag, + 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) +- ++ # register callback + def host_call_back(event): +- if self.etcd.get(self.host_path)[0] == None: +- logger.info('register host again {}'.format(self.host)) +- +- self.etcd.put(self.host_path, six.b(self.host)) +- self.need_sync = True +- +- 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) ++ self.hosts = [ ++ six.ensure_str(i[0]) ++ for i in self.etcd.get_prefix(self.node_prefix) ++ ] ++ logger.info( ++ f"host_call_back curr_host={self.host_port}, hosts:{self.hosts}") ++ self.need_sync = True ++ self.elastic_startup_time = None ++ ++ host_watch = self.etcd.add_watch_prefix_callback(self.node_prefix, ++ host_call_back) ++ host_lease = self.etcd.lease(elastic_ttl) ++ ++ # register etcd lease heartbeat ++ def lease_heartbeat(): ++ while True: ++ try: ++ host_lease.refresh() ++ ++ hosts = [ ++ six.ensure_str(i[0]) ++ for i in self.etcd.get_prefix(self.node_prefix) ++ ] ++ logger.info( ++ f"[lease_heartbeat] curr_host={self.host_port}, hosts={hosts}" ++ ) ++ if self.host_port not in hosts: ++ logger.info( ++ f"[lease_heartbeat] register host={self.host_port}") ++ self.etcd.put(self.host_path, ++ six.b(self.host_port), ++ lease=host_lease) ++ except Exception as e: ++ logger.error("[lease_heartbeat] internal error:{} {}". ++ format(e, traceback.format_exc())) ++ break ++ time.sleep(elastic_ttl / 3) ++ ++ keepalived_thread = threading.Thread( ++ name='lease_heartbeat', target=lease_heartbeat, daemon=True) ++ keepalived_thread.start() ++ ++ self.etcd.put(self.host_path, six.b(self.host_port), lease=host_lease) + + # endpoints handle DISTRIBUTED_TRAINER_ENDPOINTS and PADDLE_TRAINERS + self.etcd.put(self.endpoints_path, +- six.b('{}|{}'.format(self.endpoints, self.trainers))) ++ six.b('{}|{}'.format(self.dist_endpoints, self.trainers))) + + def endpoints_call_back(event): +- if not self.endpoints: ++ if not self.dist_endpoints: + return + edps = six.ensure_str(self.etcd.get(self.endpoints_path)[0] or '') +- self.endpoints, self.trainers = edps.split('|') ++ self.dist_endpoints, self.trainers = edps.split('|') + logger.info("set DISTRIBUTED_TRAINER_ENDPOINTS {} ".format( +- self.endpoints)) ++ self.dist_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] +- ++ self.watches = [host_watch, endpoints_watch] + self.launcher = None + + def exit(self, completed=False): +@@ -248,6 +302,30 @@ class ElasticManager(object): + if len(hosts) == 0: + self.etcd.delete_prefix(self.prefix) + ++ def _parse_np(self, np: str): ++ """ ++ np format is "MIN" or "MIN:MAX" ++ """ ++ np_str = np or os.getenv('PADDLE_ELASTIC_NP', "0") ++ np_dict = np_str.split(":") ++ min_np = max_np = 0 ++ if len(np_dict) == 1: ++ # Fault tolerant ++ min_np = int(np_dict[0]) ++ min_np = 1 if min_np <= 0 else min_np ++ max_np = 1 ++ elif len(np_dict) == 2: ++ # Elastic ++ min_np = int(np_dict[0]) ++ max_np = int(np_dict[1]) ++ min_np = 1 if min_np <= 0 else min_np ++ max_np = min_np if min_np > max_np else max_np ++ else: ++ raise ValueError( ++ f'the np={np} needs to be in "MIN" or "MIN:MAX" format') ++ ++ return min_np, max_np ++ + def _get_host(self): + try: + return socket.gethostbyname(socket.getfqdn(socket.gethostname())) +@@ -260,40 +338,166 @@ class ElasticManager(object): + + 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 ++ def _match(self, host_list: list=None): ++ if host_list: ++ self.hosts = host_list + else: +- return False ++ self.hosts = [ ++ six.ensure_str(i[0]) ++ for i in self.etcd.get_prefix(self.node_prefix) ++ ] + +- def _update_hosts(self): +- assert len(self.hosts) != 0, 'hosts empty' ++ if self.elastic_level == ElasticLevel.FAULT_TOLERANCE: ++ if len(self.hosts) == self.np: ++ return True ++ else: ++ return False + +- 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 ++ if self.elastic_level == ElasticLevel.ELASTIC: ++ # FIXME(xym) add freeze status ++ hosts_num = len(self.hosts) ++ if hosts_num == self.np: ++ return True ++ ++ if not self.elastic_startup_time: ++ self.elastic_startup_time = time.time() ++ if hosts_num == self.max_np: ++ self.elastic_startup_time = None ++ return True ++ elif hosts_num >= self.min_np and hosts_num < self.max_np: ++ interval_time = time.time() - self.elastic_startup_time ++ if interval_time <= self.elastic_timeout: ++ logger.info( ++ f"wait for timeout, you can set value by PADDLE_ELASTIC_TIMEOUT, \ ++ hosts_num={hosts_num}, min_np={self.min_np}, \ ++ interval_time={interval_time}, elastic_timeout={self.elastic_timeout}" ++ ) ++ return False ++ return True ++ else: ++ self.elastic_startup_time = None ++ return False + ++ return False ++ ++ def _update_endpoint(self, endpoints, hosts): ++ self.etcd.put(self.endpoints_path, ++ six.b('{}|{}'.format(endpoints, hosts))) ++ ++ def _update_hosts(self): ++ assert len(self.hosts) != 0, 'hosts empty' + rank = int(os.getenv('PADDLE_TRAINER_ID', -1)) +- idx = self.hosts.index(self.host) ++ if self.elastic_level == ElasticLevel.FAULT_TOLERANCE: ++ if self.host_port in self.dist_endpoints: ++ os.environ[ ++ 'DISTRIBUTED_TRAINER_ENDPOINTS'] = self.dist_endpoints ++ os.environ['PADDLE_TRAINERS'] = self.trainers ++ logger.info("update env DISTRIBUTED_TRAINER_ENDPOINTS {} ". ++ format(self.dist_endpoints)) ++ logger.info("update env PADDLE_TRAINERS {} ".format( ++ self.trainers)) ++ return + +- # swap if self.host not in the right position +- if rank >= 0: +- self.hosts[idx] = self.hosts[rank] +- self.hosts[rank] = self.host ++ # fault tolerance ++ idx = self.hosts.index(self.host_port) ++ ++ # swap if self.host not in the right position ++ if rank >= 0: ++ self.hosts[idx] = self.hosts[rank] ++ self.hosts[rank] = self.host_port ++ else: ++ os.environ['PADDLE_TRAINER_ID'] = '{}'.format(idx) ++ hosts = ','.join( ++ [host_port.split(":")[0] for host_port in self.hosts]) ++ self.args.ips = hosts ++ os.environ['PADDLE_TRAINERS'] = hosts + else: +- os.environ['PADDLE_TRAINER_ID'] = '{}'.format(idx) +- +- hosts = ','.join(self.hosts) +- self.args.ips = hosts +- os.environ['PADDLE_TRAINERS'] = hosts ++ # elastic, scale up/down ++ endpoints = copy.deepcopy(self.all_host_endpoints) ++ if len(self.hosts) > self.np: ++ # scale up ++ logger.info( ++ f"elastic scale up, from {self.np} to {len(self.hosts)}, hosts={self.hosts}, endpoints={endpoints}" ++ ) ++ ++ for curr_host_port in self.hosts: ++ if curr_host_port not in endpoints: ++ endpoints.append(curr_host_port) ++ ++ os.environ['PADDLE_TRAINER_ID'] = '{}'.format( ++ endpoints.index(self.host_port)) ++ hosts = ','.join( ++ [host_port.split(":")[0] for host_port in endpoints]) ++ self.args.ips = hosts ++ os.environ['PADDLE_TRAINERS'] = hosts ++ self.np = len(endpoints) ++ os.environ['PADDLE_TRAINER_ENDPOINTS'] = ','.join(endpoints) ++ os.environ[ ++ 'DISTRIBUTED_TRAINER_ENDPOINTS'] = self.dist_endpoints ++ self.all_host_endpoints = endpoints ++ else: ++ # scale down ++ logger.info( ++ f"elastic scale down, from {len(self.hosts)} to {self.np}, hosts={self.hosts}, endpoints={endpoints}" ++ ) ++ ++ # If the shrink node is from the first of the rank list, you need to minimize the movement of the rank ++ # eg: ++ # the source trainers is:10.10.10.0,10.10.10.1,10.10.10.2,10.10.10.3 ++ # 10.10.10.0 is removed ++ # the new trainers is:10.10.10.3,10.10.10.1,10.10.10.2 ++ # In this case, the rank of 10.10.10.1 and 10.10.10.2 remains unchanged, while the rank of 10.10.10.3 is set to rank0 ++ endpoints_dict = dict() ++ unsorted_endpoints = [] ++ for id, host_port in enumerate(self.hosts): ++ idx = endpoints.index(host_port) ++ if idx <= len(self.hosts) - 1 and not endpoints_dict.get( ++ idx): ++ endpoints_dict[idx] = host_port ++ else: ++ unsorted_endpoints.append(host_port) ++ ++ idle_index = 0 ++ sorted_endpoints = [] ++ for idx in range(len(self.hosts)): ++ if not endpoints_dict.get(idx) and len( ++ unsorted_endpoints) > 0: ++ endpoints_dict[idx] = unsorted_endpoints[idle_index] ++ idle_index += 1 ++ ++ sorted_endpoints.append(endpoints_dict.get(idx)) ++ ++ logger.info( ++ f"elastic scale down, sorted_endpoints={sorted_endpoints}") ++ self.all_host_endpoints = sorted_endpoints ++ ++ endpoint_list = [] ++ ip_list = [] ++ for host_port in sorted_endpoints: ++ host_port_list = host_port.split(":") ++ ip = host_port_list[0] ++ port = int(host_port_list[1]) ++ ++ ip_list.append(ip) ++ ports = [ ++ x ++ for x in range(port, port + len(self.devices_per_proc)) ++ ] ++ endpoint_list.extend( ++ ["%s:%d" % (ip, port) for port in ports]) ++ ++ hosts = ','.join(ip_list) ++ new_endpoints = ','.join(endpoint_list) ++ ++ self.args.ips = hosts ++ os.environ['PADDLE_TRAINER_ID'] = '{}'.format( ++ sorted_endpoints.index(self.host_port)) ++ os.environ['PADDLE_TRAINERS'] = hosts ++ self.np = len(sorted_endpoints) ++ os.environ['PADDLE_TRAINER_ENDPOINTS'] = ','.join( ++ sorted_endpoints) ++ os.environ['DISTRIBUTED_TRAINER_ENDPOINTS'] = new_endpoints ++ self._update_endpoint(new_endpoints, hosts) + + def wait(self): + if not self.enable: +@@ -307,13 +511,6 @@ class ElasticManager(object): + return + logger.info('not ready for np {} with hosts {}'.format(self.np, + self.hosts)) +- +- # reset hosts every 30s to prevent fake deadlock +- if idx % 10 == 0: +- self.etcd.delete_prefix(self.node_prefix) +- logger.info('reset np {} with hosts {}'.format(self.np, +- self.hosts)) +- + idx += 1 + time.sleep(2) + +@@ -333,6 +530,7 @@ class ElasticManager(object): + + while not self.stopped: + ret = self.launcher.watch() ++ logger.debug(f"launcher.watch():{ret}") + + if ret is not None: # self terminated + logger.info('job exit with code {}'.format(ret)) +@@ -341,7 +539,7 @@ class ElasticManager(object): + self.exit(completed=completed) + if completed: + return ElasticStatus.COMPLETED +- if self.elastic_level == 1: ++ if self.elastic_level == ElasticLevel.FAULT_TOLERANCE: + return ElasticStatus.RESTART + else: + return ElasticStatus.ERROR +@@ -354,6 +552,7 @@ class ElasticManager(object): + + if self.launcher: + self.launcher.stop() ++ + return ElasticStatus.EXIT + + def signal_handler(self, sigint, frame): diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-36684/tests/test.patch b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/tests/test.patch new file mode 100644 index 000000000..933df40a7 --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/tests/test.patch @@ -0,0 +1,749 @@ +diff --git a/python/paddle/fluid/tests/unittests/test_fleet_elastic_init.py b/python/paddle/fluid/tests/unittests/test_fleet_elastic_init.py +new file mode 100644 +index 0000000000000000000000000000000000000000..10028d2d98f67192166908d8d25aa2caf9832d51 +--- /dev/null ++++ b/python/paddle/fluid/tests/unittests/test_fleet_elastic_init.py +@@ -0,0 +1,48 @@ ++# 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. ++ ++from __future__ import print_function ++ ++import os ++import time ++import unittest ++import argparse ++from warnings import catch_warnings ++ ++from paddle.distributed.fleet.elastic import enable_elastic, launch_elastic ++from paddle.distributed.fleet.launch_utils import DistributeMode ++ ++ ++class TestElasticInit(unittest.TestCase): ++ def setUp(self): ++ class Argument: ++ elastic_server = "127.0.0.1:2379" ++ job_id = "test_job_id_123" ++ np = "2:4" ++ ++ self.args = Argument() ++ ++ def test_enable_elastic(self): ++ result = enable_elastic(self.args, DistributeMode.COLLECTIVE) ++ self.assertEqual(result, True) ++ ++ def test_launch_elastic(self): ++ try: ++ launch_elastic(self.args, DistributeMode.COLLECTIVE) ++ except Exception as e: ++ pass ++ ++ ++if __name__ == "__main__": ++ unittest.main() +diff --git a/python/paddle/fluid/tests/unittests/test_fleet_elastic_manager.py b/python/paddle/fluid/tests/unittests/test_fleet_elastic_manager.py +new file mode 100644 +index 0000000000000000000000000000000000000000..ddf87728a819ba7e2553ee5646ecb4718f5a6e25 +--- /dev/null ++++ b/python/paddle/fluid/tests/unittests/test_fleet_elastic_manager.py +@@ -0,0 +1,306 @@ ++# 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. ++ ++from __future__ import print_function ++ ++import os ++import time ++import unittest ++import argparse ++ ++from paddle.distributed.fleet.elastic.manager import ElasticManager ++from paddle.distributed.fleet.elastic.manager import ELASTIC_TIMEOUT ++ ++ ++class MockLease(): ++ def refresh(self): ++ pass ++ ++ ++class MockEtcdClient: ++ def __init__(self, lease=None): ++ self._lease = lease ++ ++ def put(self, key, value, lease=None): ++ pass ++ ++ def get(self, key): ++ value = "0" ++ return value, value ++ ++ def delete_prefix(self, key): ++ pass ++ ++ def get_prefix(self, key_prefix): ++ hosts = ["10.10.10.1:6001", "10.10.10.2:6001"] ++ return hosts ++ ++ def add_watch_callback(self, *args, **kwargs): ++ return "host_watch" ++ ++ def add_watch_prefix_callback(self, key_prefix, callback, **kwargs): ++ callback(None) ++ return "host_watch" ++ ++ def cancel_watch(self, watch_id): ++ pass ++ ++ def delete(self, key): ++ pass ++ ++ def lease(self, ttl): ++ if self._lease: ++ return self._lease ++ else: ++ return MockLease() ++ ++ ++class TestElasticManager(unittest.TestCase): ++ def setUp(self): ++ self.etcd_client = MockEtcdClient() ++ ++ def test_elastic_manager_init(self): ++ class Argument: ++ elastic_server = "127.0.0.1:2379" ++ job_id = "test_job_id_123" ++ np = "2" ++ gpus = "0" ++ nproc_per_node = 1 ++ host = None ++ host_port = None ++ scale = None ++ force = None ++ backend = 'gloo' ++ ++ args = Argument() ++ ++ class _MockLease(): ++ def refresh(self): ++ raise ValueError("valid error, this only for unittest") ++ ++ etcd_client = MockEtcdClient(lease=_MockLease()) ++ elastic = ElasticManager(args, etcd_client=etcd_client) ++ ++ def test_match_faulttolerance(self): ++ class Argument: ++ elastic_server = "127.0.0.1:2379" ++ job_id = "test_job_id_123" ++ np = "2" ++ gpus = "0" ++ nproc_per_node = 1 ++ host = None ++ host_port = None ++ scale = None ++ force = None ++ backend = 'gloo' ++ ++ args = Argument() ++ elastic = ElasticManager(args, self.etcd_client) ++ hosts = ["10.10.10.1:6001", "10.10.10.2:6001"] ++ os.environ[ ++ 'PADDLE_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001" ++ self.assertEqual(elastic._match(hosts), True) ++ hosts = ["10.10.10.1:6001"] ++ os.environ['PADDLE_TRAINER_ENDPOINTS'] = "10.10.10.1:6001" ++ self.assertEqual(elastic._match(hosts), False) ++ ++ def test_match_elastic(self): ++ class Argument: ++ elastic_server = "127.0.0.1:2379" ++ job_id = "test_job_id_123" ++ np = "2:4" ++ gpus = "0" ++ nproc_per_node = 1 ++ host = None ++ host_port = None ++ scale = None ++ force = None ++ backend = 'gloo' ++ ++ os.environ['PADDLE_ELASTIC_TIMEOUT'] = "60" ++ args = Argument() ++ os.environ['FLAGS_START_PORT'] = "6001" ++ os.environ[ ++ 'DISTRIBUTED_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001,10.10.10.3:6001,10.10.10.4:6001" ++ os.environ[ ++ 'PADDLE_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001,10.10.10.3:6001,10.10.10.4:6001" ++ elastic = ElasticManager(args, self.etcd_client) ++ hosts = ["10.10.10.1:6001", "10.10.10.2:6001"] ++ self.assertEqual(elastic._match(hosts), False) ++ ++ hosts = [ ++ "10.10.10.1:6001", "10.10.10.2:6001", "10.10.10.3:6001", ++ "10.10.10.4:6001" ++ ] ++ self.assertEqual(elastic._match(hosts), True) ++ ++ hosts = ["10.10.10.1:6001", "10.10.10.2:6001", "10.10.10.3:6001"] ++ self.assertEqual(elastic._match(hosts), False) ++ ++ hosts = ["10.10.10.1:6001"] ++ self.assertEqual(elastic._match(hosts), False) ++ ++ os.environ[ ++ 'DISTRIBUTED_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001" ++ os.environ[ ++ 'PADDLE_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001" ++ elastic = ElasticManager(args, self.etcd_client) ++ hosts = ["10.10.10.1:6001", "10.10.10.2:6001"] ++ self.assertEqual(elastic._match(hosts), True) ++ ++ # TODO test timeout ++ #time.sleep(60) ++ #self.assertEqual(elastic._match(hosts), True) ++ ++ def test_update_hosts_for_faulttolerance(self): ++ class Argument: ++ elastic_server = "127.0.0.1:2379" ++ job_id = "test_job_id_123" ++ np = "0" ++ gpus = "0" ++ nproc_per_node = 1 ++ host = None ++ host_port = None ++ scale = None ++ force = None ++ backend = 'gloo' ++ ++ args = Argument() ++ os.environ['FLAGS_START_PORT'] = "6001" ++ os.environ['PADDLE_ELASTIC_NP'] = "2" ++ os.environ['PADDLE_TRAINERS'] = "10.10.10.1,10.10.10.2" ++ os.environ[ ++ 'DISTRIBUTED_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001" ++ os.environ[ ++ 'PADDLE_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001" ++ elastic = ElasticManager(args, self.etcd_client) ++ # add 10.10.10.3:6001 ++ os.environ['PADDLE_TRAINER_ID'] = "0" ++ elastic.host_port = "10.10.10.1:6001" ++ elastic.hosts = ["10.10.10.1:6001", "10.10.10.2:6001"] ++ elastic._update_hosts() ++ self.assertEqual(os.getenv('PADDLE_TRAINERS'), "10.10.10.1,10.10.10.2") ++ ++ # add 10.10.10.3:6001 ++ elastic.host_port = "10.10.10.3:6001" ++ elastic.hosts = ["10.10.10.1:6001", "10.10.10.3:6001"] ++ os.environ['PADDLE_TRAINER_ID'] = "1" ++ elastic._update_hosts() ++ self.assertEqual(os.getenv('PADDLE_TRAINERS'), "10.10.10.1,10.10.10.3") ++ ++ elastic.host_port = "10.10.10.3:6001" ++ elastic.hosts = ["10.10.10.1:6001", "10.10.10.3:6001"] ++ os.environ['PADDLE_TRAINER_ID'] = "-1" ++ elastic._update_hosts() ++ self.assertEqual(os.getenv('PADDLE_TRAINERS'), "10.10.10.1,10.10.10.3") ++ ++ def test_update_hosts_for_elastic(self): ++ ####################### ++ # elastic, scale up # ++ ####################### ++ class Argument: ++ elastic_server = "127.0.0.1:2379" ++ job_id = "test_job_id_123" ++ np = "2:4" ++ gpus = "0" ++ nproc_per_node = 1 ++ host = None ++ host_port = None ++ scale = None ++ force = None ++ backend = 'gloo' ++ ++ args = Argument() ++ ++ os.environ['FLAGS_START_PORT'] = "6001" ++ os.environ['PADDLE_TRAINERS'] = "10.10.10.1,10.10.10.2" ++ os.environ[ ++ 'DISTRIBUTED_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001" ++ os.environ[ ++ 'PADDLE_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.2:6001" ++ elastic = ElasticManager(args, self.etcd_client) ++ # add 10.10.10.3:6001 ++ elastic.host_port = "10.10.10.1:6001" ++ elastic.hosts = [ ++ "10.10.10.1:6001", "10.10.10.2:6001", "10.10.10.3:6001" ++ ] ++ elastic._update_hosts() ++ #self.assertEqual(elastic.all_host_endpoints, ++ # ["10.10.10.1:6001", "10.10.10.2:6001", "10.10.10.3:6001"]) ++ self.assertEqual( ++ os.getenv('PADDLE_TRAINERS'), "10.10.10.1,10.10.10.2,10.10.10.3") ++ ++ ####################### ++ # elastic, scale down # ++ ####################### ++ os.environ[ ++ 'PADDLE_TRAINERS'] = "10.10.10.0,10.10.10.1,10.10.10.2,10.10.10.3" ++ os.environ[ ++ 'DISTRIBUTED_TRAINER_ENDPOINTS'] = "10.10.10.0:6000,10.10.10.1:6001,10.10.10.2:6001,10.10.10.3:6001" ++ os.environ[ ++ 'PADDLE_TRAINER_ENDPOINTS'] = "10.10.10.0:6000,10.10.10.1:6001,10.10.10.2:6001,10.10.10.3:6001" ++ elastic = ElasticManager(args, self.etcd_client) ++ # remove 10.10.10.1:6001 ++ elastic.host_port = "10.10.10.1:6001" ++ elastic.hosts = [ ++ "10.10.10.1:6001", "10.10.10.2:6001", "10.10.10.3:6001" ++ ] ++ elastic._update_hosts() ++ #self.assertEqual(elastic.all_host_endpoints, ++ # ["10.10.10.3:6001", "10.10.10.1:6001", "10.10.10.2:6001"]) ++ self.assertEqual( ++ os.getenv('PADDLE_TRAINERS'), "10.10.10.3,10.10.10.1,10.10.10.2") ++ self.assertEqual( ++ os.getenv('DISTRIBUTED_TRAINER_ENDPOINTS'), ++ "10.10.10.3:6001,10.10.10.1:6001,10.10.10.2:6001") ++ ++ ############ ++ os.environ['PADDLE_TRAINERS'] = "10.10.10.1,10.10.10.1" ++ os.environ[ ++ 'DISTRIBUTED_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.1:6002,10.10.10.1:6003,10.10.10.1:6004" ++ os.environ[ ++ 'PADDLE_TRAINER_ENDPOINTS'] = "10.10.10.1:6001,10.10.10.1:6002,10.10.10.1:6003,10.10.10.1:6004" ++ elastic = ElasticManager(args, self.etcd_client) ++ # remove 10.10.10.1:6001 ++ elastic.host_port = "10.10.10.1:6001" ++ os.environ['PADDLE_TRAINER_ID'] = "-1" ++ elastic.hosts = ["10.10.10.1:6001", "10.10.10.1:6001"] ++ elastic._update_hosts() ++ #self.assertEqual(elastic.all_host_endpoints, ++ # ["10.10.10.1:6001", "10.10.10.1:6001"]) ++ self.assertEqual(os.getenv('PADDLE_TRAINERS'), "10.10.10.1,10.10.10.1") ++ self.assertEqual( ++ os.getenv('DISTRIBUTED_TRAINER_ENDPOINTS'), ++ "10.10.10.1:6001,10.10.10.1:6001") ++ ++ def test_exit(self): ++ class Argument: ++ elastic_server = "127.0.0.1:2379" ++ job_id = "test_job_id_123" ++ np = "2" ++ gpus = "0" ++ nproc_per_node = 1 ++ host = None ++ host_port = None ++ scale = None ++ force = None ++ backend = 'gloo' ++ ++ args = Argument() ++ elastic = ElasticManager(args, self.etcd_client) ++ elastic.exit() ++ ++ ++if __name__ == "__main__": ++ unittest.main() +diff --git a/python/paddle/fluid/tests/unittests/test_fleet_launch_elastic.sh b/python/paddle/fluid/tests/unittests/test_fleet_launch_elastic.sh +index 8b618195f55ea089c9801bb9bdce5c033e884b30..a3e76a564f5b706f499a049e7c801504c56a8573 100644 +--- a/python/paddle/fluid/tests/unittests/test_fleet_launch_elastic.sh ++++ b/python/paddle/fluid/tests/unittests/test_fleet_launch_elastic.sh +@@ -15,7 +15,7 @@ + echo "begin test elastic" + + unset GREP_OPTIONS +-rm -rf log ++rm -rf log* + + pids=`ps -ef | grep "python -m paddle.distributed.launch elastic_demo.[py]" | awk '{print $2}'` + if [ -n "$pids" ]; then +@@ -28,6 +28,11 @@ fi + + python -m pip install --no-cache-dir etcd3 -i https://mirror.baidu.com/pypi/simple + ++ ++############################# ++#### test fault tolrance #### ++############################# ++ + # common env + export PADDLE_ELASTIC_NP=2 + export PADDLE_ELASTIC_SERVER=127.0.0.1:2379 +@@ -137,7 +142,7 @@ export PADDLE_TRAINER_ID=1 + export PADDLE_TRAINERS_NUM=2 + + python -m paddle.distributed.launch elastic_demo.py &> log_1.log & +-p1=$! ++p1_1=$! + + for i in {1..10} + do +@@ -184,7 +189,7 @@ export PADDLE_TRAINER_ID=0 + export PADDLE_TRAINERS_NUM=2 + + python -m paddle.distributed.launch elastic_demo.py &> log_0.log & +-p0=$! ++p0_1=$! + + for i in {1..10} + do +@@ -205,4 +210,102 @@ check_env + echo "All check done" + + sleep 3 +-kill $p0 $p1 ++kill $p0 $p1 $p0_1 $p1_1 ++ ++############################# ++##### test elastic ##### ++############################# ++# common env ++export PADDLE_ELASTIC_NP=2:4 ++export PADDLE_ELASTIC_SERVER=127.0.0.1:2379 ++export PADDLE_ELASTIC_JOB_ID=elastic-demo-2 ++ ++# run node 0 ++export NVIDIA_VISIBLE_DEVICES=0 ++export CUDA_VISIBLE_DEVICES=0 ++export DISTRIBUTED_TRAINER_ENDPOINTS=10.10.10.1:8001,10.10.10.2:8001,10.10.10.3:8001 ++export PADDLE_TRAINERS=10.10.10.1,10.10.10.2,10.10.10.3 ++export TRAINER_PORTS_NUM=1 ++export POD_IP=10.10.10.1 ++export PADDLE_TRAINER_ID=0 ++export PADDLE_TRAINERS_NUM=3 ++ ++python -m paddle.distributed.launch elastic_demo.py &> log_pe_0.log & ++pe_0=$! ++ ++for i in {1..10} ++do ++ if grep -q "INFO:ELASTIC:not ready" log_pe_0.log; then ++ echo "run node 0 ok" ++ break ++ else ++ sleep 10 ++ fi ++ if [ $i -eq 10 ]; then ++ echo "run node 0 error" ++ exit -1 ++ fi ++done ++ ++# run node 1 ++export NVIDIA_VISIBLE_DEVICES=1 ++export CUDA_VISIBLE_DEVICES=1 ++export DISTRIBUTED_TRAINER_ENDPOINTS=10.10.10.1:8001,10.10.10.2:8001,10.10.10.3:8001 ++export PADDLE_TRAINERS=10.10.10.1,10.10.10.2,10.10.10.3 ++export TRAINER_PORTS_NUM=1 ++export POD_IP=10.10.10.2 ++export PADDLE_TRAINER_ID=1 ++export PADDLE_TRAINERS_NUM=3 ++ ++python -m paddle.distributed.launch elastic_demo.py &> log_pe_1.log & ++pe_1=$! ++ ++for i in {1..10} ++do ++ if grep -q "INFO:ELASTIC:not ready" log_pe_1.log; then ++ echo "run node 1 ok" ++ break ++ else ++ sleep 10 ++ fi ++ if [ $i -eq 10 ]; then ++ echo "run node 1 error" ++ exit -1 ++ fi ++done ++ ++# run node 2 ++export NVIDIA_VISIBLE_DEVICES=1 ++export CUDA_VISIBLE_DEVICES=1 ++export DISTRIBUTED_TRAINER_ENDPOINTS=10.10.10.1:8001,10.10.10.2:8001,10.10.10.3:8001 ++export PADDLE_TRAINERS=10.10.10.1,10.10.10.2,10.10.10.3 ++export TRAINER_PORTS_NUM=1 ++export POD_IP=10.10.10.3 ++export PADDLE_TRAINER_ID=2 ++export PADDLE_TRAINERS_NUM=3 ++ ++python -m paddle.distributed.launch elastic_demo.py &> log_pe_2.log & ++pe_2=$! ++ ++for i in {1..10} ++do ++ if grep -q "INFO:ELASTIC:ready with hosts" log_pe_2.log; then ++ echo "run node 2 ok" ++ break ++ else ++ sleep 10 ++ fi ++ if [ $i -eq 10 ]; then ++ echo "run node 2 error" ++ exit -1 ++ fi ++done ++ ++lw0="log/workerlog.0" ++ ++check_env ++ ++echo "All check done" ++ ++sleep 3 ++kill $pe_0 $pe_1 $pe_2 +diff --git a/test/swe_paddle/test_pr36684_elastic_scale.py b/test/swe_paddle/test_pr36684_elastic_scale.py +new file mode 100644 +index 0000000000000000000000000000000000000000..c036bc1748cef8e8ae5f5e0e1076fb3f238688d4 +--- /dev/null ++++ b/test/swe_paddle/test_pr36684_elastic_scale.py +@@ -0,0 +1,230 @@ ++import argparse ++import ast ++import copy ++import inspect ++import logging ++import os ++import random ++import signal ++import socket ++import subprocess ++import sys ++import threading ++import time ++import traceback ++import types ++import unittest ++from contextlib import contextmanager ++from pathlib import Path ++from warnings import catch_warnings ++ ++import six ++ ++ ++ROOT = Path(__file__).resolve().parents[2] ++MANAGER_SOURCE = ROOT / "python/paddle/distributed/fleet/elastic/manager.py" ++ELASTIC_INIT_SOURCE = ( ++ ROOT / "python/paddle/distributed/fleet/elastic/__init__.py" ++) ++UPSTREAM_MANAGER_TEST = ( ++ ROOT ++ / "python/paddle/fluid/tests/unittests/test_fleet_elastic_manager.py" ++) ++UPSTREAM_INIT_TEST = ( ++ ROOT / "python/paddle/fluid/tests/unittests/test_fleet_elastic_init.py" ++) ++ ++ ++class _CloudUtils: ++ @staticmethod ++ def use_paddlecloud(): ++ return all( ++ os.getenv(name) is not None ++ for name in ( ++ "PADDLE_TRAINERS", ++ "POD_IP", ++ "PADDLE_TRAINER_ID", ++ "TRAINER_PORTS_NUM", ++ ) ++ ) ++ ++ ++class _LaunchUtils: ++ @staticmethod ++ def get_device_proc_info(args): ++ count = max(1, int(getattr(args, "nproc_per_node", 1) or 1)) ++ return "CPU", list(range(count)) ++ ++ ++class _DistributeMode: ++ COLLECTIVE = "collective" ++ ++ ++def _selected_nodes(path, class_names=(), function_names=(), constants=()): ++ tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) ++ selected = [] ++ for node in tree.body: ++ if isinstance(node, ast.ClassDef) and node.name in class_names: ++ selected.append(node) ++ elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): ++ if node.name in function_names: ++ selected.append(node) ++ elif isinstance(node, (ast.Assign, ast.AnnAssign)): ++ names = [] ++ targets = node.targets if isinstance(node, ast.Assign) else [node.target] ++ for target in targets: ++ if isinstance(target, ast.Name): ++ names.append(target.id) ++ if any(name in constants for name in names): ++ selected.append(node) ++ return ast.Module(body=selected, type_ignores=[]) ++ ++ ++def _load_manager(): ++ logger = logging.getLogger("SWE_PADDLE_PR36684_ELASTIC") ++ namespace = { ++ "__name__": "swe_paddle_pr36684_manager", ++ "copy": copy, ++ "cloud_utils": _CloudUtils, ++ "launch_utils": _LaunchUtils, ++ "logging": logging, ++ "logger": logger, ++ "os": os, ++ "random": random, ++ "signal": signal, ++ "six": six, ++ "socket": socket, ++ "subprocess": subprocess, ++ "threading": threading, ++ "time": time, ++ "traceback": traceback, ++ } ++ tree = _selected_nodes( ++ MANAGER_SOURCE, ++ class_names=( ++ "ElasticLevel", ++ "ElasticStatus", ++ "LauncherInterface", ++ "ElasticManager", ++ ), ++ constants=("ELASTIC_EXIT_CODE", "ELASTIC_TIMEOUT", "ELASTIC_TTL"), ++ ) ++ exec(compile(tree, str(MANAGER_SOURCE), "exec"), namespace) ++ source_manager = namespace["ElasticManager"] ++ ++ # The source PR made the etcd client injectable. For the Base checkout, ++ # adapt only that dependency boundary so the upstream tests reach the ++ # elastic matching behavior instead of requiring a live etcd service. ++ if len(inspect.signature(source_manager.__init__).parameters) == 2: ++ ++ class ElasticManagerAdapter(source_manager): ++ def __init__(self, args, etcd_client=None): ++ etcd3 = types.ModuleType("etcd3") ++ etcd3.client = lambda **kwargs: etcd_client ++ previous = sys.modules.get("etcd3") ++ sys.modules["etcd3"] = etcd3 ++ try: ++ super().__init__(args) ++ finally: ++ if previous is None: ++ sys.modules.pop("etcd3", None) ++ else: ++ sys.modules["etcd3"] = previous ++ ++ manager = ElasticManagerAdapter ++ else: ++ manager = source_manager ++ return manager, namespace.get("ELASTIC_TIMEOUT", 120) ++ ++ ++def _load_enable_elastic(): ++ namespace = { ++ "__name__": "swe_paddle_pr36684_elastic_init", ++ "DistributeMode": _DistributeMode, ++ "os": os, ++ } ++ tree = _selected_nodes( ++ ELASTIC_INIT_SOURCE, function_names=("enable_elastic",) ++ ) ++ exec(compile(tree, str(ELASTIC_INIT_SOURCE), "exec"), namespace) ++ return namespace["enable_elastic"] ++ ++ ++def _load_upstream_classes(manager, elastic_timeout, enable_elastic): ++ manager_namespace = { ++ "__name__": "swe_paddle_pr36684_upstream_manager_test", ++ "ElasticManager": manager, ++ "ELASTIC_TIMEOUT": elastic_timeout, ++ "argparse": argparse, ++ "os": os, ++ "time": time, ++ "unittest": unittest, ++ } ++ manager_tree = _selected_nodes( ++ UPSTREAM_MANAGER_TEST, ++ class_names=("MockLease", "MockEtcdClient", "TestElasticManager"), ++ ) ++ exec( ++ compile(manager_tree, str(UPSTREAM_MANAGER_TEST), "exec"), ++ manager_namespace, ++ ) ++ ++ init_namespace = { ++ "__name__": "swe_paddle_pr36684_upstream_init_test", ++ "DistributeMode": _DistributeMode, ++ "argparse": argparse, ++ "catch_warnings": catch_warnings, ++ "enable_elastic": enable_elastic, ++ "launch_elastic": lambda *args, **kwargs: None, ++ "os": os, ++ "time": time, ++ "unittest": unittest, ++ } ++ init_tree = _selected_nodes( ++ UPSTREAM_INIT_TEST, class_names=("TestElasticInit",) ++ ) ++ exec(compile(init_tree, str(UPSTREAM_INIT_TEST), "exec"), init_namespace) ++ return manager_namespace["TestElasticManager"], init_namespace[ ++ "TestElasticInit" ++ ] ++ ++ ++MANAGER, ELASTIC_TIMEOUT = _load_manager() ++ENABLE_ELASTIC = _load_enable_elastic() ++UPSTREAM_MANAGER_CASE, UPSTREAM_INIT_CASE = _load_upstream_classes( ++ MANAGER, ELASTIC_TIMEOUT, ENABLE_ELASTIC ++) ++ ++ ++@contextmanager ++def _isolated_distributed_environment(): ++ original = os.environ.copy() ++ for name in list(os.environ): ++ if ( ++ name.startswith("PADDLE_") ++ or name.startswith("DISTRIBUTED_") ++ or name in {"FLAGS_START_PORT", "POD_IP", "TRAINER_PORTS_NUM"} ++ ): ++ os.environ.pop(name, None) ++ try: ++ yield ++ finally: ++ os.environ.clear() ++ os.environ.update(original) ++ ++ ++def _run_upstream(case_class, method_name): ++ with _isolated_distributed_environment(): ++ case_class(methodName=method_name).debug() ++ ++ ++def test_p2p_existing_elastic_enablement_is_preserved(): ++ _run_upstream(UPSTREAM_INIT_CASE, "test_enable_elastic") ++ ++ ++def test_f2p_variable_host_range_is_matched(): ++ _run_upstream(UPSTREAM_MANAGER_CASE, "test_match_elastic") ++ ++ ++def test_f2p_scale_out_and_scale_in_update_training_hosts(): ++ _run_upstream(UPSTREAM_MANAGER_CASE, "test_update_hosts_for_elastic") diff --git a/swe-paddle/tasks/PaddlePaddle__Paddle-36684/tests/test.sh b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/tests/test.sh new file mode 100644 index 000000000..f57fb549d --- /dev/null +++ b/swe-paddle/tasks/PaddlePaddle__Paddle-36684/tests/test.sh @@ -0,0 +1,4 @@ +#!/usr/bin/env bash + +set -euo pipefail +python -m pytest test/swe_paddle/test_pr36684_elastic_scale.py -q