| 文件 | 最后提交记录 | 最后更新时间 |
|---|---|---|
[refractor] NodeManager代码微重构，提升代码可维护性 Co-authored-by: 吕有辉<lvyouhui@huawei.com> # message auto-generated for no-merge-commit merge: !620 merge refractor/node_manager into master [refractor] NodeManager代码微重构，提升代码可维护性 Created-by: codeDogPro Commit-by: 吕有辉 Merged-by: tobking Description: ## 1. 合入背景 ISSUE：https://gitcode.com/Ascend/MindIE-Motor/issues/368 ## 2. 修改内容 ### 2.1 Service Registry 重构 **文件**: motor/node_manager/core/services/registry.py 1. 用 typing.cast 消除 get_preparable() 中的 # type: ignore[arg-type,return-value]。 2. _MODULE_MAP 从模块级静态 dict 迁移到 _ServiceRegistry 实例属性，新增 add_discovery_path() API 支持动态注册后端模块路径。 3. 新增 get_active_sorted() 方法，将排序逻辑从 daemon.py 内联哨兵模式收归 registry。 4. _ServiceRegistration.is_active 改为 _is_active(active_backends) 方法，接受解析后的后端列表参数。 5. discover() 支持逗号分隔的多后端服务列表（如 "engine,memcache"），替代原来的单值匹配。 6. 删除 __slots__ 限制，2-3 个实例的内存节省可忽略。 7. 新增 tests/node_manager/core/services/test_registry.py，20 个单元测试覆盖 register、get_active 过滤、get_preparable 排序、discover 模块导入、重复注册、add_discovery_path、线程安全。 ### 2.2 服务协议提取 **文件**: motor/node_manager/core/services/protocols.py（新增） 1. DaemonService 和 PreparableService 从 registry.py 提取到 services/protocols.py。 2. daemon.py 直接从 protocols.py 导入，Protocol 定义与注册中心解耦。 ### 2.3 memcache LocalService 重构 **文件**: motor/node_manager/core/services/memcache/（新增目录） 1. **目录拆分**： services/memcache/ __init__.py worker.py ← LocalService 子进程入口（DistributedObjectStore().init(0)） lifecycle.py ← daemon 侧生命周期管理（@register_service、pull/stop/health_check） 2. **拉起方式改进**：pull() 使用 sys.executable -m motor.node_manager.core.services.memcache.worker 替代内联 -c 字符串，取消 PYTHON_EXEC_PATH 依赖。保持 subprocess.Popen(env=...) 确保与 Engine 子进程的 MMC_LOCAL_CONFIG_PATH 隔离。 3. **消除重复检查**：提取 _can_launch property，统一 should_launch() / pull() / prepare() 中的 enable、backend、mode 条件判断。 4. **简化 mark_dead()**：用 poll() 替代 wait(timeout=0) + 三重异常捕获。 5. **移除未使用的 _endpoints_count**：仅在日志中使用，改为局部变量。 ### 2.4 Engine 解耦 —— 服务配置化 **文件**: motor/node_manager/core/services/engine.py, motor/node_manager/core/daemon.py, motor/node_manager/main.py, motor/config/node_manager.py 1. @register_service(SERVICE_ENGINE) 新增 backend="engine"，从 backend=None（始终激活）改为按配置激活。 2. KVCacheStoreConfig 新增 mode 字段（"combined" / "separated"），通过 user_config.json 控制： json // Engine + KV 合体 Pod（默认） { "kv_cache_store_config": { "backend": "memcache", "mode": "combined" } } // KV 分离 Pod（只起 LocalService，不拉 Engine，不注册/心跳） { "kv_cache_store_config": { "backend": "memcache", "mode": "separated" } } // Engine only Pod {} 3. Daemon 新增 has_engine 属性，main.py 据此条件初始化 EngineManager / HeartbeatManager。 ### 2.5 main.py 重构 —— Application 基类 **文件**: motor/common/app/application.py（新增）, motor/node_manager/node_manager.py（新增）, motor/node_manager/main.py 1. **Application 基类** — 封装四个组件共享的 boilerplate： - 模块管理（add_module / get_module / stop_all_modules） - 配置热更新传播（on_config_updated → 先刷新自身间隔 _refresh_check_interval，再传播给所有带 update_config 的模块） - 可配置的 daemon loop 间隔（check_interval 参数，默认 1s，子类从 config 读取） - 信号处理（SIGINT / SIGTERM → threading.Event） - select-based daemon loop（stdin 读取 + stop_event.wait） - run() 模板方法：banner → init_modules → start_modules → config_watcher → daemon_loop → shutdown 2. **NodeManager(Application)**： - __init__ 传入 check_interval=config.basic_config.daemon_loop_interval（默认 5s，可在 user_config.json 中配置） - _refresh_check_interval()：配置热更新时同步刷新间隔 - init_modules()：根据 daemon.has_engine 动态注册模块 - _on_daemon_tick()：每 tick 检查 HeartbeatManager 自杀标志 - exit_code：自杀时返回 -1（pod rescheduling） 3. **main.py 瘦身**：从 186 行 → 37 行 thin wrapper： python def main() -> int: config = NodeManagerConfig.from_json() reconfigure_logging(config.logging_config) run_port_setup_or_exit(apply_node_manager_ports, config) nm = NodeManager(config) return nm.run() 删除所有模块级全局变量（modules、_should_exit、config、config_watcher）和 7 个模块级函数。 ### 2.6 测试重构 1. 测试目录镜像源码结构： tests/node_manager/ __init__.py conftest.py test_config.py core/ __init__.py test_daemon.py test_engine_manager.py test_heartbeat_manager.py test_fault_reporter.py test_api_ready_event.py services/ __init__.py test_registry.py memcache/ __init__.py test_lifecycle.py 2. test_main_process_title.py 从 NodeManager 和 EngineServer 各一份合并为 tests/common/utils/test_process_title.py，NodeManager 用例适配新 NodeManager 类 API。 ### 2.7 改动文件清单 | 文件 | 改动类型 | |------|----------| | motor/common/app/__init__.py | 新增 | | motor/common/app/application.py | 新增 | | motor/node_manager/node_manager.py | 新增 | | motor/node_manager/core/services/protocols.py | 新增 | | motor/node_manager/core/services/memcache/__init__.py | 新增 | | motor/node_manager/core/services/memcache/worker.py | 新增 | | motor/node_manager/core/services/memcache/lifecycle.py | 重命名自 local_service.py | | tests/node_manager/core/__init__.py | 新增 | | tests/node_manager/core/services/__init__.py | 新增 | | tests/node_manager/core/services/memcache/__init__.py | 新增 | | tests/node_manager/core/services/test_registry.py | 新增 | | tests/node_manager/core/services/memcache/test_lifecycle.py | 重命名 | | tests/common/utils/test_process_title.py | 合并自两份拷贝 | | motor/node_manager/core/services/registry.py | 重构 | | motor/node_manager/core/daemon.py | 重构 | | motor/node_manager/core/services/engine.py | 改动 | | motor/node_manager/main.py | 瘦身 | | motor/config/node_manager.py | 改动 | | motor/node_manager/core/__init__.py | 删除多余版权声明 | | motor/node_manager/core/services/__init__.py | 删除多余版权声明 | | motor/node_manager/__init__.py | 删除多余版权声明 | ## 3. 资料变更 不涉及。 ## 4. 接口变更 不涉及（所有改动为内部重构，对外接口不变）。 ## 5. 测试结果 python -m pytest tests/node_manager/ tests/common/utils/test_process_title.py tests/engine_server/ -q 529 passed in 1.47s 测试覆盖： - **registry**：注册、过滤、排序、发现、线程安全（20 个用例） - **memcache lifecycle**：should_launch、prepare、pull、stop、health_check（11 个用例） - **daemon**：engine pull、参数校验、D2D peer、signal handler（13 个用例） - **heartbeat manager**：状态上报、端点管理、自杀检测 - **engine manager**：注册、re-register、ranktable、snapshot - **config**：配置解析、验证、热加载 - **process_title**：NodeManager + EngineServer 标题设置（4 个用例） ## 6. CheckList - [x] 代码注释完备 - [x] 正确记录维测日志 - [x] 是否有UT用例（新增 24 个用例，全量 529 passed） - [x] 若涉及多线程场景，考虑了并发场景，不存在死锁问题 - _ServiceRegistry 使用 threading.Lock 保护 _registrations 和 _module_map - daemon.py 和 engine.py 的锁均为短临界区非嵌套使用 - Application 的信号处理仅设 threading.Event，清理在主线程执行 See merge request: Ascend/MindIE-Motor!620 | 10 天前 | |
[refractor] NodeManager代码微重构，提升代码可维护性 Co-authored-by: 吕有辉<lvyouhui@huawei.com> # message auto-generated for no-merge-commit merge: !620 merge refractor/node_manager into master [refractor] NodeManager代码微重构，提升代码可维护性 Created-by: codeDogPro Commit-by: 吕有辉 Merged-by: tobking Description: ## 1. 合入背景 ISSUE：https://gitcode.com/Ascend/MindIE-Motor/issues/368 ## 2. 修改内容 ### 2.1 Service Registry 重构 **文件**: motor/node_manager/core/services/registry.py 1. 用 typing.cast 消除 get_preparable() 中的 # type: ignore[arg-type,return-value]。 2. _MODULE_MAP 从模块级静态 dict 迁移到 _ServiceRegistry 实例属性，新增 add_discovery_path() API 支持动态注册后端模块路径。 3. 新增 get_active_sorted() 方法，将排序逻辑从 daemon.py 内联哨兵模式收归 registry。 4. _ServiceRegistration.is_active 改为 _is_active(active_backends) 方法，接受解析后的后端列表参数。 5. discover() 支持逗号分隔的多后端服务列表（如 "engine,memcache"），替代原来的单值匹配。 6. 删除 __slots__ 限制，2-3 个实例的内存节省可忽略。 7. 新增 tests/node_manager/core/services/test_registry.py，20 个单元测试覆盖 register、get_active 过滤、get_preparable 排序、discover 模块导入、重复注册、add_discovery_path、线程安全。 ### 2.2 服务协议提取 **文件**: motor/node_manager/core/services/protocols.py（新增） 1. DaemonService 和 PreparableService 从 registry.py 提取到 services/protocols.py。 2. daemon.py 直接从 protocols.py 导入，Protocol 定义与注册中心解耦。 ### 2.3 memcache LocalService 重构 **文件**: motor/node_manager/core/services/memcache/（新增目录） 1. **目录拆分**： services/memcache/ __init__.py worker.py ← LocalService 子进程入口（DistributedObjectStore().init(0)） lifecycle.py ← daemon 侧生命周期管理（@register_service、pull/stop/health_check） 2. **拉起方式改进**：pull() 使用 sys.executable -m motor.node_manager.core.services.memcache.worker 替代内联 -c 字符串，取消 PYTHON_EXEC_PATH 依赖。保持 subprocess.Popen(env=...) 确保与 Engine 子进程的 MMC_LOCAL_CONFIG_PATH 隔离。 3. **消除重复检查**：提取 _can_launch property，统一 should_launch() / pull() / prepare() 中的 enable、backend、mode 条件判断。 4. **简化 mark_dead()**：用 poll() 替代 wait(timeout=0) + 三重异常捕获。 5. **移除未使用的 _endpoints_count**：仅在日志中使用，改为局部变量。 ### 2.4 Engine 解耦 —— 服务配置化 **文件**: motor/node_manager/core/services/engine.py, motor/node_manager/core/daemon.py, motor/node_manager/main.py, motor/config/node_manager.py 1. @register_service(SERVICE_ENGINE) 新增 backend="engine"，从 backend=None（始终激活）改为按配置激活。 2. KVCacheStoreConfig 新增 mode 字段（"combined" / "separated"），通过 user_config.json 控制： json // Engine + KV 合体 Pod（默认） { "kv_cache_store_config": { "backend": "memcache", "mode": "combined" } } // KV 分离 Pod（只起 LocalService，不拉 Engine，不注册/心跳） { "kv_cache_store_config": { "backend": "memcache", "mode": "separated" } } // Engine only Pod {} 3. Daemon 新增 has_engine 属性，main.py 据此条件初始化 EngineManager / HeartbeatManager。 ### 2.5 main.py 重构 —— Application 基类 **文件**: motor/common/app/application.py（新增）, motor/node_manager/node_manager.py（新增）, motor/node_manager/main.py 1. **Application 基类** — 封装四个组件共享的 boilerplate： - 模块管理（add_module / get_module / stop_all_modules） - 配置热更新传播（on_config_updated → 先刷新自身间隔 _refresh_check_interval，再传播给所有带 update_config 的模块） - 可配置的 daemon loop 间隔（check_interval 参数，默认 1s，子类从 config 读取） - 信号处理（SIGINT / SIGTERM → threading.Event） - select-based daemon loop（stdin 读取 + stop_event.wait） - run() 模板方法：banner → init_modules → start_modules → config_watcher → daemon_loop → shutdown 2. **NodeManager(Application)**： - __init__ 传入 check_interval=config.basic_config.daemon_loop_interval（默认 5s，可在 user_config.json 中配置） - _refresh_check_interval()：配置热更新时同步刷新间隔 - init_modules()：根据 daemon.has_engine 动态注册模块 - _on_daemon_tick()：每 tick 检查 HeartbeatManager 自杀标志 - exit_code：自杀时返回 -1（pod rescheduling） 3. **main.py 瘦身**：从 186 行 → 37 行 thin wrapper： python def main() -> int: config = NodeManagerConfig.from_json() reconfigure_logging(config.logging_config) run_port_setup_or_exit(apply_node_manager_ports, config) nm = NodeManager(config) return nm.run() 删除所有模块级全局变量（modules、_should_exit、config、config_watcher）和 7 个模块级函数。 ### 2.6 测试重构 1. 测试目录镜像源码结构： tests/node_manager/ __init__.py conftest.py test_config.py core/ __init__.py test_daemon.py test_engine_manager.py test_heartbeat_manager.py test_fault_reporter.py test_api_ready_event.py services/ __init__.py test_registry.py memcache/ __init__.py test_lifecycle.py 2. test_main_process_title.py 从 NodeManager 和 EngineServer 各一份合并为 tests/common/utils/test_process_title.py，NodeManager 用例适配新 NodeManager 类 API。 ### 2.7 改动文件清单 | 文件 | 改动类型 | |------|----------| | motor/common/app/__init__.py | 新增 | | motor/common/app/application.py | 新增 | | motor/node_manager/node_manager.py | 新增 | | motor/node_manager/core/services/protocols.py | 新增 | | motor/node_manager/core/services/memcache/__init__.py | 新增 | | motor/node_manager/core/services/memcache/worker.py | 新增 | | motor/node_manager/core/services/memcache/lifecycle.py | 重命名自 local_service.py | | tests/node_manager/core/__init__.py | 新增 | | tests/node_manager/core/services/__init__.py | 新增 | | tests/node_manager/core/services/memcache/__init__.py | 新增 | | tests/node_manager/core/services/test_registry.py | 新增 | | tests/node_manager/core/services/memcache/test_lifecycle.py | 重命名 | | tests/common/utils/test_process_title.py | 合并自两份拷贝 | | motor/node_manager/core/services/registry.py | 重构 | | motor/node_manager/core/daemon.py | 重构 | | motor/node_manager/core/services/engine.py | 改动 | | motor/node_manager/main.py | 瘦身 | | motor/config/node_manager.py | 改动 | | motor/node_manager/core/__init__.py | 删除多余版权声明 | | motor/node_manager/core/services/__init__.py | 删除多余版权声明 | | motor/node_manager/__init__.py | 删除多余版权声明 | ## 3. 资料变更 不涉及。 ## 4. 接口变更 不涉及（所有改动为内部重构，对外接口不变）。 ## 5. 测试结果 python -m pytest tests/node_manager/ tests/common/utils/test_process_title.py tests/engine_server/ -q 529 passed in 1.47s 测试覆盖： - **registry**：注册、过滤、排序、发现、线程安全（20 个用例） - **memcache lifecycle**：should_launch、prepare、pull、stop、health_check（11 个用例） - **daemon**：engine pull、参数校验、D2D peer、signal handler（13 个用例） - **heartbeat manager**：状态上报、端点管理、自杀检测 - **engine manager**：注册、re-register、ranktable、snapshot - **config**：配置解析、验证、热加载 - **process_title**：NodeManager + EngineServer 标题设置（4 个用例） ## 6. CheckList - [x] 代码注释完备 - [x] 正确记录维测日志 - [x] 是否有UT用例（新增 24 个用例，全量 529 passed） - [x] 若涉及多线程场景，考虑了并发场景，不存在死锁问题 - _ServiceRegistry 使用 threading.Lock 保护 _registrations 和 _module_map - daemon.py 和 engine.py 的锁均为短临界区非嵌套使用 - Application 的信号处理仅设 threading.Event，清理在主线程执行 See merge request: Ascend/MindIE-Motor!620 | 10 天前 | |
[refractor] NodeManager代码微重构，提升代码可维护性 Co-authored-by: 吕有辉<lvyouhui@huawei.com> # message auto-generated for no-merge-commit merge: !620 merge refractor/node_manager into master [refractor] NodeManager代码微重构，提升代码可维护性 Created-by: codeDogPro Commit-by: 吕有辉 Merged-by: tobking Description: ## 1. 合入背景 ISSUE：https://gitcode.com/Ascend/MindIE-Motor/issues/368 ## 2. 修改内容 ### 2.1 Service Registry 重构 **文件**: motor/node_manager/core/services/registry.py 1. 用 typing.cast 消除 get_preparable() 中的 # type: ignore[arg-type,return-value]。 2. _MODULE_MAP 从模块级静态 dict 迁移到 _ServiceRegistry 实例属性，新增 add_discovery_path() API 支持动态注册后端模块路径。 3. 新增 get_active_sorted() 方法，将排序逻辑从 daemon.py 内联哨兵模式收归 registry。 4. _ServiceRegistration.is_active 改为 _is_active(active_backends) 方法，接受解析后的后端列表参数。 5. discover() 支持逗号分隔的多后端服务列表（如 "engine,memcache"），替代原来的单值匹配。 6. 删除 __slots__ 限制，2-3 个实例的内存节省可忽略。 7. 新增 tests/node_manager/core/services/test_registry.py，20 个单元测试覆盖 register、get_active 过滤、get_preparable 排序、discover 模块导入、重复注册、add_discovery_path、线程安全。 ### 2.2 服务协议提取 **文件**: motor/node_manager/core/services/protocols.py（新增） 1. DaemonService 和 PreparableService 从 registry.py 提取到 services/protocols.py。 2. daemon.py 直接从 protocols.py 导入，Protocol 定义与注册中心解耦。 ### 2.3 memcache LocalService 重构 **文件**: motor/node_manager/core/services/memcache/（新增目录） 1. **目录拆分**： services/memcache/ __init__.py worker.py ← LocalService 子进程入口（DistributedObjectStore().init(0)） lifecycle.py ← daemon 侧生命周期管理（@register_service、pull/stop/health_check） 2. **拉起方式改进**：pull() 使用 sys.executable -m motor.node_manager.core.services.memcache.worker 替代内联 -c 字符串，取消 PYTHON_EXEC_PATH 依赖。保持 subprocess.Popen(env=...) 确保与 Engine 子进程的 MMC_LOCAL_CONFIG_PATH 隔离。 3. **消除重复检查**：提取 _can_launch property，统一 should_launch() / pull() / prepare() 中的 enable、backend、mode 条件判断。 4. **简化 mark_dead()**：用 poll() 替代 wait(timeout=0) + 三重异常捕获。 5. **移除未使用的 _endpoints_count**：仅在日志中使用，改为局部变量。 ### 2.4 Engine 解耦 —— 服务配置化 **文件**: motor/node_manager/core/services/engine.py, motor/node_manager/core/daemon.py, motor/node_manager/main.py, motor/config/node_manager.py 1. @register_service(SERVICE_ENGINE) 新增 backend="engine"，从 backend=None（始终激活）改为按配置激活。 2. KVCacheStoreConfig 新增 mode 字段（"combined" / "separated"），通过 user_config.json 控制： json // Engine + KV 合体 Pod（默认） { "kv_cache_store_config": { "backend": "memcache", "mode": "combined" } } // KV 分离 Pod（只起 LocalService，不拉 Engine，不注册/心跳） { "kv_cache_store_config": { "backend": "memcache", "mode": "separated" } } // Engine only Pod {} 3. Daemon 新增 has_engine 属性，main.py 据此条件初始化 EngineManager / HeartbeatManager。 ### 2.5 main.py 重构 —— Application 基类 **文件**: motor/common/app/application.py（新增）, motor/node_manager/node_manager.py（新增）, motor/node_manager/main.py 1. **Application 基类** — 封装四个组件共享的 boilerplate： - 模块管理（add_module / get_module / stop_all_modules） - 配置热更新传播（on_config_updated → 先刷新自身间隔 _refresh_check_interval，再传播给所有带 update_config 的模块） - 可配置的 daemon loop 间隔（check_interval 参数，默认 1s，子类从 config 读取） - 信号处理（SIGINT / SIGTERM → threading.Event） - select-based daemon loop（stdin 读取 + stop_event.wait） - run() 模板方法：banner → init_modules → start_modules → config_watcher → daemon_loop → shutdown 2. **NodeManager(Application)**： - __init__ 传入 check_interval=config.basic_config.daemon_loop_interval（默认 5s，可在 user_config.json 中配置） - _refresh_check_interval()：配置热更新时同步刷新间隔 - init_modules()：根据 daemon.has_engine 动态注册模块 - _on_daemon_tick()：每 tick 检查 HeartbeatManager 自杀标志 - exit_code：自杀时返回 -1（pod rescheduling） 3. **main.py 瘦身**：从 186 行 → 37 行 thin wrapper： python def main() -> int: config = NodeManagerConfig.from_json() reconfigure_logging(config.logging_config) run_port_setup_or_exit(apply_node_manager_ports, config) nm = NodeManager(config) return nm.run() 删除所有模块级全局变量（modules、_should_exit、config、config_watcher）和 7 个模块级函数。 ### 2.6 测试重构 1. 测试目录镜像源码结构： tests/node_manager/ __init__.py conftest.py test_config.py core/ __init__.py test_daemon.py test_engine_manager.py test_heartbeat_manager.py test_fault_reporter.py test_api_ready_event.py services/ __init__.py test_registry.py memcache/ __init__.py test_lifecycle.py 2. test_main_process_title.py 从 NodeManager 和 EngineServer 各一份合并为 tests/common/utils/test_process_title.py，NodeManager 用例适配新 NodeManager 类 API。 ### 2.7 改动文件清单 | 文件 | 改动类型 | |------|----------| | motor/common/app/__init__.py | 新增 | | motor/common/app/application.py | 新增 | | motor/node_manager/node_manager.py | 新增 | | motor/node_manager/core/services/protocols.py | 新增 | | motor/node_manager/core/services/memcache/__init__.py | 新增 | | motor/node_manager/core/services/memcache/worker.py | 新增 | | motor/node_manager/core/services/memcache/lifecycle.py | 重命名自 local_service.py | | tests/node_manager/core/__init__.py | 新增 | | tests/node_manager/core/services/__init__.py | 新增 | | tests/node_manager/core/services/memcache/__init__.py | 新增 | | tests/node_manager/core/services/test_registry.py | 新增 | | tests/node_manager/core/services/memcache/test_lifecycle.py | 重命名 | | tests/common/utils/test_process_title.py | 合并自两份拷贝 | | motor/node_manager/core/services/registry.py | 重构 | | motor/node_manager/core/daemon.py | 重构 | | motor/node_manager/core/services/engine.py | 改动 | | motor/node_manager/main.py | 瘦身 | | motor/config/node_manager.py | 改动 | | motor/node_manager/core/__init__.py | 删除多余版权声明 | | motor/node_manager/core/services/__init__.py | 删除多余版权声明 | | motor/node_manager/__init__.py | 删除多余版权声明 | ## 3. 资料变更 不涉及。 ## 4. 接口变更 不涉及（所有改动为内部重构，对外接口不变）。 ## 5. 测试结果 python -m pytest tests/node_manager/ tests/common/utils/test_process_title.py tests/engine_server/ -q 529 passed in 1.47s 测试覆盖： - **registry**：注册、过滤、排序、发现、线程安全（20 个用例） - **memcache lifecycle**：should_launch、prepare、pull、stop、health_check（11 个用例） - **daemon**：engine pull、参数校验、D2D peer、signal handler（13 个用例） - **heartbeat manager**：状态上报、端点管理、自杀检测 - **engine manager**：注册、re-register、ranktable、snapshot - **config**：配置解析、验证、热加载 - **process_title**：NodeManager + EngineServer 标题设置（4 个用例） ## 6. CheckList - [x] 代码注释完备 - [x] 正确记录维测日志 - [x] 是否有UT用例（新增 24 个用例，全量 529 passed） - [x] 若涉及多线程场景，考虑了并发场景，不存在死锁问题 - _ServiceRegistry 使用 threading.Lock 保护 _registrations 和 _module_map - daemon.py 和 engine.py 的锁均为短临界区非嵌套使用 - Application 的信号处理仅设 threading.Event，清理在主线程执行 See merge request: Ascend/MindIE-Motor!620 | 10 天前 | |
[refractor] NodeManager代码微重构，提升代码可维护性 Co-authored-by: 吕有辉<lvyouhui@huawei.com> # message auto-generated for no-merge-commit merge: !620 merge refractor/node_manager into master [refractor] NodeManager代码微重构，提升代码可维护性 Created-by: codeDogPro Commit-by: 吕有辉 Merged-by: tobking Description: ## 1. 合入背景 ISSUE：https://gitcode.com/Ascend/MindIE-Motor/issues/368 ## 2. 修改内容 ### 2.1 Service Registry 重构 **文件**: motor/node_manager/core/services/registry.py 1. 用 typing.cast 消除 get_preparable() 中的 # type: ignore[arg-type,return-value]。 2. _MODULE_MAP 从模块级静态 dict 迁移到 _ServiceRegistry 实例属性，新增 add_discovery_path() API 支持动态注册后端模块路径。 3. 新增 get_active_sorted() 方法，将排序逻辑从 daemon.py 内联哨兵模式收归 registry。 4. _ServiceRegistration.is_active 改为 _is_active(active_backends) 方法，接受解析后的后端列表参数。 5. discover() 支持逗号分隔的多后端服务列表（如 "engine,memcache"），替代原来的单值匹配。 6. 删除 __slots__ 限制，2-3 个实例的内存节省可忽略。 7. 新增 tests/node_manager/core/services/test_registry.py，20 个单元测试覆盖 register、get_active 过滤、get_preparable 排序、discover 模块导入、重复注册、add_discovery_path、线程安全。 ### 2.2 服务协议提取 **文件**: motor/node_manager/core/services/protocols.py（新增） 1. DaemonService 和 PreparableService 从 registry.py 提取到 services/protocols.py。 2. daemon.py 直接从 protocols.py 导入，Protocol 定义与注册中心解耦。 ### 2.3 memcache LocalService 重构 **文件**: motor/node_manager/core/services/memcache/（新增目录） 1. **目录拆分**： services/memcache/ __init__.py worker.py ← LocalService 子进程入口（DistributedObjectStore().init(0)） lifecycle.py ← daemon 侧生命周期管理（@register_service、pull/stop/health_check） 2. **拉起方式改进**：pull() 使用 sys.executable -m motor.node_manager.core.services.memcache.worker 替代内联 -c 字符串，取消 PYTHON_EXEC_PATH 依赖。保持 subprocess.Popen(env=...) 确保与 Engine 子进程的 MMC_LOCAL_CONFIG_PATH 隔离。 3. **消除重复检查**：提取 _can_launch property，统一 should_launch() / pull() / prepare() 中的 enable、backend、mode 条件判断。 4. **简化 mark_dead()**：用 poll() 替代 wait(timeout=0) + 三重异常捕获。 5. **移除未使用的 _endpoints_count**：仅在日志中使用，改为局部变量。 ### 2.4 Engine 解耦 —— 服务配置化 **文件**: motor/node_manager/core/services/engine.py, motor/node_manager/core/daemon.py, motor/node_manager/main.py, motor/config/node_manager.py 1. @register_service(SERVICE_ENGINE) 新增 backend="engine"，从 backend=None（始终激活）改为按配置激活。 2. KVCacheStoreConfig 新增 mode 字段（"combined" / "separated"），通过 user_config.json 控制： json // Engine + KV 合体 Pod（默认） { "kv_cache_store_config": { "backend": "memcache", "mode": "combined" } } // KV 分离 Pod（只起 LocalService，不拉 Engine，不注册/心跳） { "kv_cache_store_config": { "backend": "memcache", "mode": "separated" } } // Engine only Pod {} 3. Daemon 新增 has_engine 属性，main.py 据此条件初始化 EngineManager / HeartbeatManager。 ### 2.5 main.py 重构 —— Application 基类 **文件**: motor/common/app/application.py（新增）, motor/node_manager/node_manager.py（新增）, motor/node_manager/main.py 1. **Application 基类** — 封装四个组件共享的 boilerplate： - 模块管理（add_module / get_module / stop_all_modules） - 配置热更新传播（on_config_updated → 先刷新自身间隔 _refresh_check_interval，再传播给所有带 update_config 的模块） - 可配置的 daemon loop 间隔（check_interval 参数，默认 1s，子类从 config 读取） - 信号处理（SIGINT / SIGTERM → threading.Event） - select-based daemon loop（stdin 读取 + stop_event.wait） - run() 模板方法：banner → init_modules → start_modules → config_watcher → daemon_loop → shutdown 2. **NodeManager(Application)**： - __init__ 传入 check_interval=config.basic_config.daemon_loop_interval（默认 5s，可在 user_config.json 中配置） - _refresh_check_interval()：配置热更新时同步刷新间隔 - init_modules()：根据 daemon.has_engine 动态注册模块 - _on_daemon_tick()：每 tick 检查 HeartbeatManager 自杀标志 - exit_code：自杀时返回 -1（pod rescheduling） 3. **main.py 瘦身**：从 186 行 → 37 行 thin wrapper： python def main() -> int: config = NodeManagerConfig.from_json() reconfigure_logging(config.logging_config) run_port_setup_or_exit(apply_node_manager_ports, config) nm = NodeManager(config) return nm.run() 删除所有模块级全局变量（modules、_should_exit、config、config_watcher）和 7 个模块级函数。 ### 2.6 测试重构 1. 测试目录镜像源码结构： tests/node_manager/ __init__.py conftest.py test_config.py core/ __init__.py test_daemon.py test_engine_manager.py test_heartbeat_manager.py test_fault_reporter.py test_api_ready_event.py services/ __init__.py test_registry.py memcache/ __init__.py test_lifecycle.py 2. test_main_process_title.py 从 NodeManager 和 EngineServer 各一份合并为 tests/common/utils/test_process_title.py，NodeManager 用例适配新 NodeManager 类 API。 ### 2.7 改动文件清单 | 文件 | 改动类型 | |------|----------| | motor/common/app/__init__.py | 新增 | | motor/common/app/application.py | 新增 | | motor/node_manager/node_manager.py | 新增 | | motor/node_manager/core/services/protocols.py | 新增 | | motor/node_manager/core/services/memcache/__init__.py | 新增 | | motor/node_manager/core/services/memcache/worker.py | 新增 | | motor/node_manager/core/services/memcache/lifecycle.py | 重命名自 local_service.py | | tests/node_manager/core/__init__.py | 新增 | | tests/node_manager/core/services/__init__.py | 新增 | | tests/node_manager/core/services/memcache/__init__.py | 新增 | | tests/node_manager/core/services/test_registry.py | 新增 | | tests/node_manager/core/services/memcache/test_lifecycle.py | 重命名 | | tests/common/utils/test_process_title.py | 合并自两份拷贝 | | motor/node_manager/core/services/registry.py | 重构 | | motor/node_manager/core/daemon.py | 重构 | | motor/node_manager/core/services/engine.py | 改动 | | motor/node_manager/main.py | 瘦身 | | motor/config/node_manager.py | 改动 | | motor/node_manager/core/__init__.py | 删除多余版权声明 | | motor/node_manager/core/services/__init__.py | 删除多余版权声明 | | motor/node_manager/__init__.py | 删除多余版权声明 | ## 3. 资料变更 不涉及。 ## 4. 接口变更 不涉及（所有改动为内部重构，对外接口不变）。 ## 5. 测试结果 python -m pytest tests/node_manager/ tests/common/utils/test_process_title.py tests/engine_server/ -q 529 passed in 1.47s 测试覆盖： - **registry**：注册、过滤、排序、发现、线程安全（20 个用例） - **memcache lifecycle**：should_launch、prepare、pull、stop、health_check（11 个用例） - **daemon**：engine pull、参数校验、D2D peer、signal handler（13 个用例） - **heartbeat manager**：状态上报、端点管理、自杀检测 - **engine manager**：注册、re-register、ranktable、snapshot - **config**：配置解析、验证、热加载 - **process_title**：NodeManager + EngineServer 标题设置（4 个用例） ## 6. CheckList - [x] 代码注释完备 - [x] 正确记录维测日志 - [x] 是否有UT用例（新增 24 个用例，全量 529 passed） - [x] 若涉及多线程场景，考虑了并发场景，不存在死锁问题 - _ServiceRegistry 使用 threading.Lock 保护 _registrations 和 _module_map - daemon.py 和 engine.py 的锁均为短临界区非嵌套使用 - Application 的信号处理仅设 threading.Event，清理在主线程执行 See merge request: Ascend/MindIE-Motor!620 | 10 天前 | |
[refractor] NodeManager代码微重构，提升代码可维护性 Co-authored-by: 吕有辉<lvyouhui@huawei.com> # message auto-generated for no-merge-commit merge: !620 merge refractor/node_manager into master [refractor] NodeManager代码微重构，提升代码可维护性 Created-by: codeDogPro Commit-by: 吕有辉 Merged-by: tobking Description: ## 1. 合入背景 ISSUE：https://gitcode.com/Ascend/MindIE-Motor/issues/368 ## 2. 修改内容 ### 2.1 Service Registry 重构 **文件**: motor/node_manager/core/services/registry.py 1. 用 typing.cast 消除 get_preparable() 中的 # type: ignore[arg-type,return-value]。 2. _MODULE_MAP 从模块级静态 dict 迁移到 _ServiceRegistry 实例属性，新增 add_discovery_path() API 支持动态注册后端模块路径。 3. 新增 get_active_sorted() 方法，将排序逻辑从 daemon.py 内联哨兵模式收归 registry。 4. _ServiceRegistration.is_active 改为 _is_active(active_backends) 方法，接受解析后的后端列表参数。 5. discover() 支持逗号分隔的多后端服务列表（如 "engine,memcache"），替代原来的单值匹配。 6. 删除 __slots__ 限制，2-3 个实例的内存节省可忽略。 7. 新增 tests/node_manager/core/services/test_registry.py，20 个单元测试覆盖 register、get_active 过滤、get_preparable 排序、discover 模块导入、重复注册、add_discovery_path、线程安全。 ### 2.2 服务协议提取 **文件**: motor/node_manager/core/services/protocols.py（新增） 1. DaemonService 和 PreparableService 从 registry.py 提取到 services/protocols.py。 2. daemon.py 直接从 protocols.py 导入，Protocol 定义与注册中心解耦。 ### 2.3 memcache LocalService 重构 **文件**: motor/node_manager/core/services/memcache/（新增目录） 1. **目录拆分**： services/memcache/ __init__.py worker.py ← LocalService 子进程入口（DistributedObjectStore().init(0)） lifecycle.py ← daemon 侧生命周期管理（@register_service、pull/stop/health_check） 2. **拉起方式改进**：pull() 使用 sys.executable -m motor.node_manager.core.services.memcache.worker 替代内联 -c 字符串，取消 PYTHON_EXEC_PATH 依赖。保持 subprocess.Popen(env=...) 确保与 Engine 子进程的 MMC_LOCAL_CONFIG_PATH 隔离。 3. **消除重复检查**：提取 _can_launch property，统一 should_launch() / pull() / prepare() 中的 enable、backend、mode 条件判断。 4. **简化 mark_dead()**：用 poll() 替代 wait(timeout=0) + 三重异常捕获。 5. **移除未使用的 _endpoints_count**：仅在日志中使用，改为局部变量。 ### 2.4 Engine 解耦 —— 服务配置化 **文件**: motor/node_manager/core/services/engine.py, motor/node_manager/core/daemon.py, motor/node_manager/main.py, motor/config/node_manager.py 1. @register_service(SERVICE_ENGINE) 新增 backend="engine"，从 backend=None（始终激活）改为按配置激活。 2. KVCacheStoreConfig 新增 mode 字段（"combined" / "separated"），通过 user_config.json 控制： json // Engine + KV 合体 Pod（默认） { "kv_cache_store_config": { "backend": "memcache", "mode": "combined" } } // KV 分离 Pod（只起 LocalService，不拉 Engine，不注册/心跳） { "kv_cache_store_config": { "backend": "memcache", "mode": "separated" } } // Engine only Pod {} 3. Daemon 新增 has_engine 属性，main.py 据此条件初始化 EngineManager / HeartbeatManager。 ### 2.5 main.py 重构 —— Application 基类 **文件**: motor/common/app/application.py（新增）, motor/node_manager/node_manager.py（新增）, motor/node_manager/main.py 1. **Application 基类** — 封装四个组件共享的 boilerplate： - 模块管理（add_module / get_module / stop_all_modules） - 配置热更新传播（on_config_updated → 先刷新自身间隔 _refresh_check_interval，再传播给所有带 update_config 的模块） - 可配置的 daemon loop 间隔（check_interval 参数，默认 1s，子类从 config 读取） - 信号处理（SIGINT / SIGTERM → threading.Event） - select-based daemon loop（stdin 读取 + stop_event.wait） - run() 模板方法：banner → init_modules → start_modules → config_watcher → daemon_loop → shutdown 2. **NodeManager(Application)**： - __init__ 传入 check_interval=config.basic_config.daemon_loop_interval（默认 5s，可在 user_config.json 中配置） - _refresh_check_interval()：配置热更新时同步刷新间隔 - init_modules()：根据 daemon.has_engine 动态注册模块 - _on_daemon_tick()：每 tick 检查 HeartbeatManager 自杀标志 - exit_code：自杀时返回 -1（pod rescheduling） 3. **main.py 瘦身**：从 186 行 → 37 行 thin wrapper： python def main() -> int: config = NodeManagerConfig.from_json() reconfigure_logging(config.logging_config) run_port_setup_or_exit(apply_node_manager_ports, config) nm = NodeManager(config) return nm.run() 删除所有模块级全局变量（modules、_should_exit、config、config_watcher）和 7 个模块级函数。 ### 2.6 测试重构 1. 测试目录镜像源码结构： tests/node_manager/ __init__.py conftest.py test_config.py core/ __init__.py test_daemon.py test_engine_manager.py test_heartbeat_manager.py test_fault_reporter.py test_api_ready_event.py services/ __init__.py test_registry.py memcache/ __init__.py test_lifecycle.py 2. test_main_process_title.py 从 NodeManager 和 EngineServer 各一份合并为 tests/common/utils/test_process_title.py，NodeManager 用例适配新 NodeManager 类 API。 ### 2.7 改动文件清单 | 文件 | 改动类型 | |------|----------| | motor/common/app/__init__.py | 新增 | | motor/common/app/application.py | 新增 | | motor/node_manager/node_manager.py | 新增 | | motor/node_manager/core/services/protocols.py | 新增 | | motor/node_manager/core/services/memcache/__init__.py | 新增 | | motor/node_manager/core/services/memcache/worker.py | 新增 | | motor/node_manager/core/services/memcache/lifecycle.py | 重命名自 local_service.py | | tests/node_manager/core/__init__.py | 新增 | | tests/node_manager/core/services/__init__.py | 新增 | | tests/node_manager/core/services/memcache/__init__.py | 新增 | | tests/node_manager/core/services/test_registry.py | 新增 | | tests/node_manager/core/services/memcache/test_lifecycle.py | 重命名 | | tests/common/utils/test_process_title.py | 合并自两份拷贝 | | motor/node_manager/core/services/registry.py | 重构 | | motor/node_manager/core/daemon.py | 重构 | | motor/node_manager/core/services/engine.py | 改动 | | motor/node_manager/main.py | 瘦身 | | motor/config/node_manager.py | 改动 | | motor/node_manager/core/__init__.py | 删除多余版权声明 | | motor/node_manager/core/services/__init__.py | 删除多余版权声明 | | motor/node_manager/__init__.py | 删除多余版权声明 | ## 3. 资料变更 不涉及。 ## 4. 接口变更 不涉及（所有改动为内部重构，对外接口不变）。 ## 5. 测试结果 python -m pytest tests/node_manager/ tests/common/utils/test_process_title.py tests/engine_server/ -q 529 passed in 1.47s 测试覆盖： - **registry**：注册、过滤、排序、发现、线程安全（20 个用例） - **memcache lifecycle**：should_launch、prepare、pull、stop、health_check（11 个用例） - **daemon**：engine pull、参数校验、D2D peer、signal handler（13 个用例） - **heartbeat manager**：状态上报、端点管理、自杀检测 - **engine manager**：注册、re-register、ranktable、snapshot - **config**：配置解析、验证、热加载 - **process_title**：NodeManager + EngineServer 标题设置（4 个用例） ## 6. CheckList - [x] 代码注释完备 - [x] 正确记录维测日志 - [x] 是否有UT用例（新增 24 个用例，全量 529 passed） - [x] 若涉及多线程场景，考虑了并发场景，不存在死锁问题 - _ServiceRegistry 使用 threading.Lock 保护 _registrations 和 _module_map - daemon.py 和 engine.py 的锁均为短临界区非嵌套使用 - Application 的信号处理仅设 threading.Event，清理在主线程执行 See merge request: Ascend/MindIE-Motor!620 | 10 天前 | |
[refractor] NodeManager代码微重构，提升代码可维护性 Co-authored-by: 吕有辉<lvyouhui@huawei.com> # message auto-generated for no-merge-commit merge: !620 merge refractor/node_manager into master [refractor] NodeManager代码微重构，提升代码可维护性 Created-by: codeDogPro Commit-by: 吕有辉 Merged-by: tobking Description: ## 1. 合入背景 ISSUE：https://gitcode.com/Ascend/MindIE-Motor/issues/368 ## 2. 修改内容 ### 2.1 Service Registry 重构 **文件**: motor/node_manager/core/services/registry.py 1. 用 typing.cast 消除 get_preparable() 中的 # type: ignore[arg-type,return-value]。 2. _MODULE_MAP 从模块级静态 dict 迁移到 _ServiceRegistry 实例属性，新增 add_discovery_path() API 支持动态注册后端模块路径。 3. 新增 get_active_sorted() 方法，将排序逻辑从 daemon.py 内联哨兵模式收归 registry。 4. _ServiceRegistration.is_active 改为 _is_active(active_backends) 方法，接受解析后的后端列表参数。 5. discover() 支持逗号分隔的多后端服务列表（如 "engine,memcache"），替代原来的单值匹配。 6. 删除 __slots__ 限制，2-3 个实例的内存节省可忽略。 7. 新增 tests/node_manager/core/services/test_registry.py，20 个单元测试覆盖 register、get_active 过滤、get_preparable 排序、discover 模块导入、重复注册、add_discovery_path、线程安全。 ### 2.2 服务协议提取 **文件**: motor/node_manager/core/services/protocols.py（新增） 1. DaemonService 和 PreparableService 从 registry.py 提取到 services/protocols.py。 2. daemon.py 直接从 protocols.py 导入，Protocol 定义与注册中心解耦。 ### 2.3 memcache LocalService 重构 **文件**: motor/node_manager/core/services/memcache/（新增目录） 1. **目录拆分**： services/memcache/ __init__.py worker.py ← LocalService 子进程入口（DistributedObjectStore().init(0)） lifecycle.py ← daemon 侧生命周期管理（@register_service、pull/stop/health_check） 2. **拉起方式改进**：pull() 使用 sys.executable -m motor.node_manager.core.services.memcache.worker 替代内联 -c 字符串，取消 PYTHON_EXEC_PATH 依赖。保持 subprocess.Popen(env=...) 确保与 Engine 子进程的 MMC_LOCAL_CONFIG_PATH 隔离。 3. **消除重复检查**：提取 _can_launch property，统一 should_launch() / pull() / prepare() 中的 enable、backend、mode 条件判断。 4. **简化 mark_dead()**：用 poll() 替代 wait(timeout=0) + 三重异常捕获。 5. **移除未使用的 _endpoints_count**：仅在日志中使用，改为局部变量。 ### 2.4 Engine 解耦 —— 服务配置化 **文件**: motor/node_manager/core/services/engine.py, motor/node_manager/core/daemon.py, motor/node_manager/main.py, motor/config/node_manager.py 1. @register_service(SERVICE_ENGINE) 新增 backend="engine"，从 backend=None（始终激活）改为按配置激活。 2. KVCacheStoreConfig 新增 mode 字段（"combined" / "separated"），通过 user_config.json 控制： json // Engine + KV 合体 Pod（默认） { "kv_cache_store_config": { "backend": "memcache", "mode": "combined" } } // KV 分离 Pod（只起 LocalService，不拉 Engine，不注册/心跳） { "kv_cache_store_config": { "backend": "memcache", "mode": "separated" } } // Engine only Pod {} 3. Daemon 新增 has_engine 属性，main.py 据此条件初始化 EngineManager / HeartbeatManager。 ### 2.5 main.py 重构 —— Application 基类 **文件**: motor/common/app/application.py（新增）, motor/node_manager/node_manager.py（新增）, motor/node_manager/main.py 1. **Application 基类** — 封装四个组件共享的 boilerplate： - 模块管理（add_module / get_module / stop_all_modules） - 配置热更新传播（on_config_updated → 先刷新自身间隔 _refresh_check_interval，再传播给所有带 update_config 的模块） - 可配置的 daemon loop 间隔（check_interval 参数，默认 1s，子类从 config 读取） - 信号处理（SIGINT / SIGTERM → threading.Event） - select-based daemon loop（stdin 读取 + stop_event.wait） - run() 模板方法：banner → init_modules → start_modules → config_watcher → daemon_loop → shutdown 2. **NodeManager(Application)**： - __init__ 传入 check_interval=config.basic_config.daemon_loop_interval（默认 5s，可在 user_config.json 中配置） - _refresh_check_interval()：配置热更新时同步刷新间隔 - init_modules()：根据 daemon.has_engine 动态注册模块 - _on_daemon_tick()：每 tick 检查 HeartbeatManager 自杀标志 - exit_code：自杀时返回 -1（pod rescheduling） 3. **main.py 瘦身**：从 186 行 → 37 行 thin wrapper： python def main() -> int: config = NodeManagerConfig.from_json() reconfigure_logging(config.logging_config) run_port_setup_or_exit(apply_node_manager_ports, config) nm = NodeManager(config) return nm.run() 删除所有模块级全局变量（modules、_should_exit、config、config_watcher）和 7 个模块级函数。 ### 2.6 测试重构 1. 测试目录镜像源码结构： tests/node_manager/ __init__.py conftest.py test_config.py core/ __init__.py test_daemon.py test_engine_manager.py test_heartbeat_manager.py test_fault_reporter.py test_api_ready_event.py services/ __init__.py test_registry.py memcache/ __init__.py test_lifecycle.py 2. test_main_process_title.py 从 NodeManager 和 EngineServer 各一份合并为 tests/common/utils/test_process_title.py，NodeManager 用例适配新 NodeManager 类 API。 ### 2.7 改动文件清单 | 文件 | 改动类型 | |------|----------| | motor/common/app/__init__.py | 新增 | | motor/common/app/application.py | 新增 | | motor/node_manager/node_manager.py | 新增 | | motor/node_manager/core/services/protocols.py | 新增 | | motor/node_manager/core/services/memcache/__init__.py | 新增 | | motor/node_manager/core/services/memcache/worker.py | 新增 | | motor/node_manager/core/services/memcache/lifecycle.py | 重命名自 local_service.py | | tests/node_manager/core/__init__.py | 新增 | | tests/node_manager/core/services/__init__.py | 新增 | | tests/node_manager/core/services/memcache/__init__.py | 新增 | | tests/node_manager/core/services/test_registry.py | 新增 | | tests/node_manager/core/services/memcache/test_lifecycle.py | 重命名 | | tests/common/utils/test_process_title.py | 合并自两份拷贝 | | motor/node_manager/core/services/registry.py | 重构 | | motor/node_manager/core/daemon.py | 重构 | | motor/node_manager/core/services/engine.py | 改动 | | motor/node_manager/main.py | 瘦身 | | motor/config/node_manager.py | 改动 | | motor/node_manager/core/__init__.py | 删除多余版权声明 | | motor/node_manager/core/services/__init__.py | 删除多余版权声明 | | motor/node_manager/__init__.py | 删除多余版权声明 | ## 3. 资料变更 不涉及。 ## 4. 接口变更 不涉及（所有改动为内部重构，对外接口不变）。 ## 5. 测试结果 python -m pytest tests/node_manager/ tests/common/utils/test_process_title.py tests/engine_server/ -q 529 passed in 1.47s 测试覆盖： - **registry**：注册、过滤、排序、发现、线程安全（20 个用例） - **memcache lifecycle**：should_launch、prepare、pull、stop、health_check（11 个用例） - **daemon**：engine pull、参数校验、D2D peer、signal handler（13 个用例） - **heartbeat manager**：状态上报、端点管理、自杀检测 - **engine manager**：注册、re-register、ranktable、snapshot - **config**：配置解析、验证、热加载 - **process_title**：NodeManager + EngineServer 标题设置（4 个用例） ## 6. CheckList - [x] 代码注释完备 - [x] 正确记录维测日志 - [x] 是否有UT用例（新增 24 个用例，全量 529 passed） - [x] 若涉及多线程场景，考虑了并发场景，不存在死锁问题 - _ServiceRegistry 使用 threading.Lock 保护 _registrations 和 _module_map - daemon.py 和 engine.py 的锁均为短临界区非嵌套使用 - Application 的信号处理仅设 threading.Event，清理在主线程执行 See merge request: Ascend/MindIE-Motor!620 | 10 天前 | |
[refractor] NodeManager代码微重构，提升代码可维护性 Co-authored-by: 吕有辉<lvyouhui@huawei.com> # message auto-generated for no-merge-commit merge: !620 merge refractor/node_manager into master [refractor] NodeManager代码微重构，提升代码可维护性 Created-by: codeDogPro Commit-by: 吕有辉 Merged-by: tobking Description: ## 1. 合入背景 ISSUE：https://gitcode.com/Ascend/MindIE-Motor/issues/368 ## 2. 修改内容 ### 2.1 Service Registry 重构 **文件**: motor/node_manager/core/services/registry.py 1. 用 typing.cast 消除 get_preparable() 中的 # type: ignore[arg-type,return-value]。 2. _MODULE_MAP 从模块级静态 dict 迁移到 _ServiceRegistry 实例属性，新增 add_discovery_path() API 支持动态注册后端模块路径。 3. 新增 get_active_sorted() 方法，将排序逻辑从 daemon.py 内联哨兵模式收归 registry。 4. _ServiceRegistration.is_active 改为 _is_active(active_backends) 方法，接受解析后的后端列表参数。 5. discover() 支持逗号分隔的多后端服务列表（如 "engine,memcache"），替代原来的单值匹配。 6. 删除 __slots__ 限制，2-3 个实例的内存节省可忽略。 7. 新增 tests/node_manager/core/services/test_registry.py，20 个单元测试覆盖 register、get_active 过滤、get_preparable 排序、discover 模块导入、重复注册、add_discovery_path、线程安全。 ### 2.2 服务协议提取 **文件**: motor/node_manager/core/services/protocols.py（新增） 1. DaemonService 和 PreparableService 从 registry.py 提取到 services/protocols.py。 2. daemon.py 直接从 protocols.py 导入，Protocol 定义与注册中心解耦。 ### 2.3 memcache LocalService 重构 **文件**: motor/node_manager/core/services/memcache/（新增目录） 1. **目录拆分**： services/memcache/ __init__.py worker.py ← LocalService 子进程入口（DistributedObjectStore().init(0)） lifecycle.py ← daemon 侧生命周期管理（@register_service、pull/stop/health_check） 2. **拉起方式改进**：pull() 使用 sys.executable -m motor.node_manager.core.services.memcache.worker 替代内联 -c 字符串，取消 PYTHON_EXEC_PATH 依赖。保持 subprocess.Popen(env=...) 确保与 Engine 子进程的 MMC_LOCAL_CONFIG_PATH 隔离。 3. **消除重复检查**：提取 _can_launch property，统一 should_launch() / pull() / prepare() 中的 enable、backend、mode 条件判断。 4. **简化 mark_dead()**：用 poll() 替代 wait(timeout=0) + 三重异常捕获。 5. **移除未使用的 _endpoints_count**：仅在日志中使用，改为局部变量。 ### 2.4 Engine 解耦 —— 服务配置化 **文件**: motor/node_manager/core/services/engine.py, motor/node_manager/core/daemon.py, motor/node_manager/main.py, motor/config/node_manager.py 1. @register_service(SERVICE_ENGINE) 新增 backend="engine"，从 backend=None（始终激活）改为按配置激活。 2. KVCacheStoreConfig 新增 mode 字段（"combined" / "separated"），通过 user_config.json 控制： json // Engine + KV 合体 Pod（默认） { "kv_cache_store_config": { "backend": "memcache", "mode": "combined" } } // KV 分离 Pod（只起 LocalService，不拉 Engine，不注册/心跳） { "kv_cache_store_config": { "backend": "memcache", "mode": "separated" } } // Engine only Pod {} 3. Daemon 新增 has_engine 属性，main.py 据此条件初始化 EngineManager / HeartbeatManager。 ### 2.5 main.py 重构 —— Application 基类 **文件**: motor/common/app/application.py（新增）, motor/node_manager/node_manager.py（新增）, motor/node_manager/main.py 1. **Application 基类** — 封装四个组件共享的 boilerplate： - 模块管理（add_module / get_module / stop_all_modules） - 配置热更新传播（on_config_updated → 先刷新自身间隔 _refresh_check_interval，再传播给所有带 update_config 的模块） - 可配置的 daemon loop 间隔（check_interval 参数，默认 1s，子类从 config 读取） - 信号处理（SIGINT / SIGTERM → threading.Event） - select-based daemon loop（stdin 读取 + stop_event.wait） - run() 模板方法：banner → init_modules → start_modules → config_watcher → daemon_loop → shutdown 2. **NodeManager(Application)**： - __init__ 传入 check_interval=config.basic_config.daemon_loop_interval（默认 5s，可在 user_config.json 中配置） - _refresh_check_interval()：配置热更新时同步刷新间隔 - init_modules()：根据 daemon.has_engine 动态注册模块 - _on_daemon_tick()：每 tick 检查 HeartbeatManager 自杀标志 - exit_code：自杀时返回 -1（pod rescheduling） 3. **main.py 瘦身**：从 186 行 → 37 行 thin wrapper： python def main() -> int: config = NodeManagerConfig.from_json() reconfigure_logging(config.logging_config) run_port_setup_or_exit(apply_node_manager_ports, config) nm = NodeManager(config) return nm.run() 删除所有模块级全局变量（modules、_should_exit、config、config_watcher）和 7 个模块级函数。 ### 2.6 测试重构 1. 测试目录镜像源码结构： tests/node_manager/ __init__.py conftest.py test_config.py core/ __init__.py test_daemon.py test_engine_manager.py test_heartbeat_manager.py test_fault_reporter.py test_api_ready_event.py services/ __init__.py test_registry.py memcache/ __init__.py test_lifecycle.py 2. test_main_process_title.py 从 NodeManager 和 EngineServer 各一份合并为 tests/common/utils/test_process_title.py，NodeManager 用例适配新 NodeManager 类 API。 ### 2.7 改动文件清单 | 文件 | 改动类型 | |------|----------| | motor/common/app/__init__.py | 新增 | | motor/common/app/application.py | 新增 | | motor/node_manager/node_manager.py | 新增 | | motor/node_manager/core/services/protocols.py | 新增 | | motor/node_manager/core/services/memcache/__init__.py | 新增 | | motor/node_manager/core/services/memcache/worker.py | 新增 | | motor/node_manager/core/services/memcache/lifecycle.py | 重命名自 local_service.py | | tests/node_manager/core/__init__.py | 新增 | | tests/node_manager/core/services/__init__.py | 新增 | | tests/node_manager/core/services/memcache/__init__.py | 新增 | | tests/node_manager/core/services/test_registry.py | 新增 | | tests/node_manager/core/services/memcache/test_lifecycle.py | 重命名 | | tests/common/utils/test_process_title.py | 合并自两份拷贝 | | motor/node_manager/core/services/registry.py | 重构 | | motor/node_manager/core/daemon.py | 重构 | | motor/node_manager/core/services/engine.py | 改动 | | motor/node_manager/main.py | 瘦身 | | motor/config/node_manager.py | 改动 | | motor/node_manager/core/__init__.py | 删除多余版权声明 | | motor/node_manager/core/services/__init__.py | 删除多余版权声明 | | motor/node_manager/__init__.py | 删除多余版权声明 | ## 3. 资料变更 不涉及。 ## 4. 接口变更 不涉及（所有改动为内部重构，对外接口不变）。 ## 5. 测试结果 python -m pytest tests/node_manager/ tests/common/utils/test_process_title.py tests/engine_server/ -q 529 passed in 1.47s 测试覆盖： - **registry**：注册、过滤、排序、发现、线程安全（20 个用例） - **memcache lifecycle**：should_launch、prepare、pull、stop、health_check（11 个用例） - **daemon**：engine pull、参数校验、D2D peer、signal handler（13 个用例） - **heartbeat manager**：状态上报、端点管理、自杀检测 - **engine manager**：注册、re-register、ranktable、snapshot - **config**：配置解析、验证、热加载 - **process_title**：NodeManager + EngineServer 标题设置（4 个用例） ## 6. CheckList - [x] 代码注释完备 - [x] 正确记录维测日志 - [x] 是否有UT用例（新增 24 个用例，全量 529 passed） - [x] 若涉及多线程场景，考虑了并发场景，不存在死锁问题 - _ServiceRegistry 使用 threading.Lock 保护 _registrations 和 _module_map - daemon.py 和 engine.py 的锁均为短临界区非嵌套使用 - Application 的信号处理仅设 threading.Event，清理在主线程执行 See merge request: Ascend/MindIE-Motor!620 | 10 天前 |
| 文件 | 最后提交记录 | 最后更新时间 |
|---|---|---|
| 10 天前 | ||
| 10 天前 | ||
| 10 天前 | ||
| 10 天前 | ||
| 10 天前 | ||
| 10 天前 | ||
| 10 天前 |