[Environment Adaptation] Support Windows DataLoader multiprocess (num_workers > 0)#79417
[Environment Adaptation] Support Windows DataLoader multiprocess (num_workers > 0)#79417PlumBlossomMaid wants to merge 32 commits into
Conversation
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
PlumBlossomMaid
changed the title
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已完成本轮复查。当前仍有需要阻塞合入的问题，主要是最新 head 中 mmap_allocator 的函数签名/调用不一致会直接导致编译失败，以及 DataLoader worker 外层异常被吞后可能以 0 退出。详细建议已放在行级评论中。
另外，已有其它 review 线程覆盖了 Windows shared memory 生命周期、HANDLE 宽度、getpid() 和 ParentWatchDog 父进程 PID 等阻塞点，建议一起修复后再重新触发 CI。
| }; | ||
|
|
||
| void AllocateMemoryMap(std::string filename, | ||
| void AllocateMemoryMap(std::string *filename, |
There was a problem hiding this comment.
这里把 AllocateMemoryMap 改成接收 std::string *，但函数体仍按对象使用 filename.c_str()，并在冲突重试时执行 filename = GetIPCName()；std::string* 没有这些接口，也不能从 std::string 赋值，所以当前 head 会直接编译失败。请把声明/定义恢复为引用，并把调用点从取地址改回直接传入，或者把函数体所有访问都改成正确的指针解引用；前者更符合这里需要回写文件名的语义。
// mmap_allocator.h / .cc
void AllocateMemoryMap(std::string &filename,
int *shared_fd,
int flags,
size_t size,
void **base_ptr_);
// AllocateRefcountedMemoryMapAllocation
AllocateMemoryMap(filename, &fd, flags, size + mmap_alignment, &base_ptr);There was a problem hiding this comment.
更新：这一处在新提交里只修了部分 filename->... 访问，但签名/类型问题仍然会阻塞构建。当前 .h 声明的是 AllocateMemoryMap(std::string *, intptr_t *)，.cc 定义仍是 int *shared_fd；AllocateRefcountedMemoryMapAllocation 也在头文件声明为 intptr_t shared_fd、实现为 int shared_fd。调用方会按头文件生成 intptr_t 版本符号，当前实现没有对应定义，链接会失败。另外内部重新声明了局部 int fd，外层传给 RefcountedMemoryMapAllocation 的 fd 没有被更新，Windows HANDLE 仍会丢失。
请把声明、定义、构造函数和 pybind 元数据类型一次性统一到 intptr_t（或专门的 Windows HANDLE 封装），并避免在 if 块内 shadow 外层 fd：
intptr_t fd = shared_fd;
...
AllocateMemoryMap(&filename, &fd, flags, size + mmap_alignment, &base_ptr);
...
return std::make_shared<RefcountedMemoryMapAllocation>(
aligned_base_ptr, size, filename, fd, flags, buffer_id);There was a problem hiding this comment.
更新：最新提交修掉了 shared_fd 的部分声明/定义不一致，但这个线程里的构建阻塞点仍未完全消除。当前 AllocateMemoryMap 已改成 std::string *filename，Windows 分支冲突重试处仍是 filename = GetIPCName()，POSIX 分支也还在用 filename.c_str() / shm_unlink(filename.c_str())；std::string* 没有这些成员，当前代码仍会直接编译失败。
另外，AllocateRefcountedMemoryMapAllocation 现在内部使用 intptr_t fd，但 RefcountedMemoryMapAllocation 的声明和定义仍接收 int fd，Windows HANDLE 进入对象前还是会被窄化。请把文件名访问和 fd/HANDLE 类型一次性统一，例如保留指针签名时至少需要按下面的形态修正：
// collision retry
*filename = GetIPCName();
// POSIX branch
fd = shm_open(filename->c_str(), file_flags, static_cast<mode_t>(0600));
...
shm_unlink(filename->c_str());
// constructor declaration/definition
RefcountedMemoryMapAllocation(void* ptr,
size_t size,
std::string ipc_name,
intptr_t fd,
int flags,
int buffer_id = -1);There was a problem hiding this comment.
更新到当前 head 后，前面提到的 std::string* 访问和 int 构造参数窄化已经修掉，但这条 mmap 分配链路仍然会阻塞构建/共享内存路径。
当前 mmap_allocator.h:84-88 声明是 (int flags, intptr_t fd, ...)，而 mmap_allocator.cc:262-267 定义是 (intptr_t fd, int flags, ...)，Linux CI 报 no declaration matches，Windows CI 也在 mmap_allocator.cc.obj 编译时报 C2511。同时 AllocateRefcountedMemoryMapAllocation 里的 intptr_t fd = 0 会把正常传入的 shared_fd == -1 丢掉，POSIX 分支会跳过 shm_open 并对 fd 0 执行 ftruncate/mmap，使 DataLoader 共享内存路径回归。
请把构造函数声明、定义和 std::make_shared<RefcountedMemoryMapAllocation>(...) 的实参顺序统一，并恢复 fd 初始值，例如：
intptr_t fd = shared_fd;
RefcountedMemoryMapAllocation(void* ptr,
size_t size,
std::string ipc_name,
intptr_t fd,
int flags,
int buffer_id = -1);There was a problem hiding this comment.
更新：最新提交已修复构造函数参数顺序和 fd = shared_fd，但同一条 Windows HANDLE 宽度链路还没有完全收口。当前 MemoryMapAllocation::fd_ 已是 intptr_t，但 mmap_allocator.h:66 的 shared_fd() 仍返回 int；随后新增的模块级 _share_filename 会把这个值放进 Python metadata，_new_shared_filename 又用 t[1].cast<int>() 读回。64 位 Windows 上启用 FLAGS_dataloader_use_file_descriptor=True 时，CreateFileMappingA 返回的 HANDLE 仍可能在 metadata 边界被截断，导致 reader 侧拿到无效句柄或关闭错误句柄。
请把共享句柄的 metadata 类型也保持为指针宽度，或在 Windows 上明确禁用 file-descriptor 模式。修复形态建议如下：
// mmap_allocator.h
inline intptr_t shared_fd() const { return fd_; }
// tensor.cc: 类方法和新增的模块级 wrapper 都保持一致
intptr_t shared_fd = -1;
const intptr_t shared_fd = t[1].cast<intptr_t>();There was a problem hiding this comment.
更新：当前提交已经把 MemoryMapAllocation::shared_fd() 改成 intptr_t，这一部分是正确推进；但 metadata 边界还没有完全收口。paddle/fluid/pybind/tensor.cc 里类方法和模块级 wrapper 的 _new_shared_filename 仍然用 const int shared_fd = t[1].cast<int>()，并且新建共享 allocation 时的临时 shared_fd 仍声明为 int。这样 Python tuple 里的 64 位 Windows HANDLE 读回 C++ 时仍可能被截断。
请把这两个 pybind 路径同步改为指针宽度，例如：
intptr_t shared_fd = -1;
...
const intptr_t shared_fd = t[1].cast<intptr_t>();| # Write to stderr (visible in parent console) | ||
| sys.stderr.write( | ||
| f"[WORKER pid={os.getpid()}] UNCAUGHT EXCEPTION\n{traceback.format_exc()}\n" | ||
| ) | ||
| sys.stderr.flush() |
There was a problem hiding this comment.
这个最外层 except 现在只写 stderr 后继续走 finally 并返回，worker 进程会以 0 退出。这样 _worker_loop 初始化阶段或队列循环外抛出的异常不会被 ThrowErrorIfLoadProcessFailed 识别为 worker failure，主进程可能只看到 DataLoader 卡住/提前结束，尤其 Windows 分支只按非零 exit code 报错。请保留诊断日志后重新抛出，或者显式以非零状态退出，让父进程能可靠感知失败。
| # Write to stderr (visible in parent console) | |
| sys.stderr.write( | |
| f"[WORKER pid={os.getpid()}] UNCAUGHT EXCEPTION\n{traceback.format_exc()}\n" | |
| ) | |
| sys.stderr.flush() | |
| # Write to stderr (visible in parent console) | |
| sys.stderr.write( | |
| f"[WORKER pid={os.getpid()}] UNCAUGHT EXCEPTION\n{traceback.format_exc()}\n" | |
| ) | |
| sys.stderr.flush() | |
| raise |
…coding, ParentWatchDog PID
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查新提交。部分 bot 线程里的 Windows shared memory 问题已有推进，但仍有会阻塞合入的问题：既有 Nyanpasu 线程补充了 mmap_allocator 签名/句柄类型不一致的新证据，另外新的行级评论覆盖了非 Windows PID helper 递归、_worker_loop 调用点兼容性、以及 Windows ctypes 句柄宽度问题。
上轮指出的 worker 外层异常被吞问题仍在既有线程中覆盖，本次不重复开新线程。
| #ifdef _WIN32 | ||
| static inline int GetPid() { return static_cast<int>(GetCurrentProcessId()); } | ||
| #else | ||
| static inline int GetPid() { return GetPid(); } |
There was a problem hiding this comment.
这里的非 Windows 分支会递归调用 GetPid() 自身。当前 GetIPCName() 和 MemoryMapFdSet 日志都会走这个 helper，Linux DataLoader 共享内存路径一进入 GetIPCName() 就可能栈溢出，而不是生成 IPC 名称。请改回调用 POSIX getpid()（上面已经在非 Windows 分支 include 了 <unistd.h>）。
| static inline int GetPid() { return GetPid(); } | |
| static inline int GetPid() { return getpid(); } |
| handle = ctypes.windll.kernel32.OpenProcess( | ||
| PROCESS_QUERY_LIMITED_INFORMATION, False, self._parent_pid | ||
| ) | ||
| if not handle: | ||
| # Fallback: try PROCESS_QUERY_INFORMATION (might fail on some systems) | ||
| PROCESS_QUERY_INFORMATION = 0x0400 | ||
| handle = ctypes.windll.kernel32.OpenProcess( | ||
| PROCESS_QUERY_INFORMATION, False, self._parent_pid | ||
| ) | ||
| if not handle: | ||
| return True # Can't check, assume alive | ||
| exit_code = wintypes.DWORD(0) | ||
| ctypes.windll.kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)) | ||
| ctypes.windll.kernel32.CloseHandle(handle) |
There was a problem hiding this comment.
ctypes.windll.kernel32.OpenProcess 默认 restype 是 c_int，64 位 Windows 上会把 HANDLE 截断成 32 位；随后 GetExitCodeProcess/CloseHandle 可能拿到无效句柄，ParentWatchDog 会误判父进程已退出或泄漏/关闭错误句柄。请给这些 WinAPI 设置 argtypes/restype，用 wintypes.HANDLE 保存句柄后再查询退出码。
| handle = ctypes.windll.kernel32.OpenProcess( | |
| PROCESS_QUERY_LIMITED_INFORMATION, False, self._parent_pid | |
| ) | |
| if not handle: | |
| # Fallback: try PROCESS_QUERY_INFORMATION (might fail on some systems) | |
| PROCESS_QUERY_INFORMATION = 0x0400 | |
| handle = ctypes.windll.kernel32.OpenProcess( | |
| PROCESS_QUERY_INFORMATION, False, self._parent_pid | |
| ) | |
| if not handle: | |
| return True # Can't check, assume alive | |
| exit_code = wintypes.DWORD(0) | |
| ctypes.windll.kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)) | |
| ctypes.windll.kernel32.CloseHandle(handle) | |
| kernel32 = ctypes.windll.kernel32 | |
| kernel32.OpenProcess.argtypes = [ | |
| wintypes.DWORD, | |
| wintypes.BOOL, | |
| wintypes.DWORD, | |
| ] | |
| kernel32.OpenProcess.restype = wintypes.HANDLE | |
| kernel32.GetExitCodeProcess.argtypes = [ | |
| wintypes.HANDLE, | |
| ctypes.POINTER(wintypes.DWORD), | |
| ] | |
| kernel32.GetExitCodeProcess.restype = wintypes.BOOL | |
| kernel32.CloseHandle.argtypes = [wintypes.HANDLE] | |
| kernel32.CloseHandle.restype = wintypes.BOOL | |
| handle = kernel32.OpenProcess( | |
| PROCESS_QUERY_LIMITED_INFORMATION, | |
| False, | |
| wintypes.DWORD(self._parent_pid), | |
| ) | |
| if not handle: | |
| # Fallback: try PROCESS_QUERY_INFORMATION (might fail on some systems) | |
| PROCESS_QUERY_INFORMATION = 0x0400 | |
| handle = kernel32.OpenProcess( | |
| PROCESS_QUERY_INFORMATION, | |
| False, | |
| wintypes.DWORD(self._parent_pid), | |
| ) | |
| if not handle: | |
| return True # Can't check, assume alive | |
| exit_code = wintypes.DWORD(0) | |
| try: | |
| if not kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)): | |
| return True | |
| finally: | |
| kernel32.CloseHandle(handle) |
|
|
||
|
|
||
| def _worker_loop( | ||
| parent_pid, |
There was a problem hiding this comment.
这里把 parent_pid 插到 _worker_loop 的第一个必选位置，但仓库里 test/legacy_test/test_multiprocess_dataloader_exception.py 仍有两处直接按旧签名调用 _worker_loop(...)。这些调用现在会把 dataset 当成 parent_pid，并因为缺少 base_seed 直接 TypeError/参数错位，导致既有 DataLoader 异常测试失败。请同步更新所有直接调用，或把新参数放到末尾并保留默认值，避免破坏旧的内部调用形态。
def _worker_loop(
dataset,
dataset_kind,
indices_queue,
out_queue,
done_event,
auto_collate_batch,
collate_fn,
drop_last,
init_fn,
worker_id,
num_workers,
use_shared_memory,
base_seed,
shm_cache_size=0,
parent_pid=None,
):
...
parent_watch_dog = ParentWatchDog(parent_pid)
# dataloader_iter.py 中再把 os.getpid() 作为最后一个参数传入。
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
…ch, worker parent_pid, except raise, ctypes HANDLE
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。worker 父进程 PID 传参、外层异常重新抛出、GetPid() 递归和 WinAPI HANDLE 的 ctypes 声明等前序问题看起来已经修复。
不过 mmap_allocator 这一条链路仍有会阻塞构建/Windows 句柄正确性的残留问题，我已在既有线程补充了当前 head 的具体证据。请先处理该线程后再合入。
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。前序 std::string* 访问、HANDLE 构造参数窄化、worker/PID/ctypes 相关问题看起来已有修复，Pre Commit 也已通过。
但 mmap_allocator 仍有阻塞构建/共享内存路径的问题，我已在既有 Nyanpasu 线程补充当前 head 和 CI 日志证据。请先处理该线程后再合入。
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。mmap_allocator 的构造函数参数顺序和 fd 初始值问题已经修复，前序 worker/PID/ctypes 相关问题也仍看起来已处理。
不过 Windows 共享句柄在 Python metadata 边界仍会从 intptr_t 收窄到 int，我已在既有 Nyanpasu 线程补充当前 head 的证据和修复方向。该问题会影响 file-descriptor 模式下 64 位 Windows 句柄正确性，建议修复后再合入。
当前仍需以最终 CI 和审批结果为准。
CI报告基于以下代码生成（30分钟更新一次）: 1 Required任务 : 37/48 通过
2 失败详情🔴 Coverage build — 环境问题（置信度: 高）分析器: 通用分析(fallback) 失败用例:
关键日志:
深度状态显示 修复建议:
关联变更: 未发现与 PR 修改直接相关的代码证据 🔴 Coverage build — 环境问题（置信度: 高）分析器: 通用分析(fallback) 失败用例:
关键日志:
失败发生在读取 develop 分支 coverage build size 基线时，BOS 地址 443 端口连接超时后脚本退出。该错误属于外部存储/网络访问异常，与 PR 代码变更无直接关系。 修复建议:
关联变更: 未发现与 PR 修改直接相关的代码证据 🔴 Linux-DCU / Test — PR问题（置信度: 高）分析器: 通用分析(fallback) 失败用例:
关键日志:
PR 新增 修复建议:
关联变更: 🔴 PR-CI-Inference / Build — 环境问题（置信度: 高）分析器: 通用分析(fallback) 失败用例:
关键日志:
快速摘要明确为 self-hosted runner lost communication；深度状态也显示构建未正常结束，后续上传和清理步骤未执行。当前日志没有编译错误或测试断言失败证据。 修复建议:
关联变更: 未发现与 PR 修改直接相关的代码证据 🔴 Check approval — 需要 Approval（置信度: 高）该 Job 需要人工 Approval，完成审批后 CI 才会继续执行。 修复建议:
|
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。CloseHandle 早关的问题有继续推进，但当前 refcounted mmap 的 Windows HANDLE/view 生命周期仍有新的阻塞风险，细节已放在行级评论中。
前序 intptr_t 到 int 的 Python metadata 收窄问题仍由既有 Nyanpasu 线程覆盖，本轮不重复开新线程。当前 Pre Commit 还因 test_modelaverage.py 的 ruff format 未通过，也需要同步处理。
| @@ -229,28 +349,64 @@ void RefcountedMemoryMapAllocation::close() { | |||
| flags_, map_size_ - mmap_alignment, ipc_name_, map_ptr_)); | |||
| } else { | |||
| if (info->refcount == 0) { | |||
There was a problem hiding this comment.
这里把 Windows 的 CloseHandle 推迟到 info->refcount == 0，但 RefcountedMemoryMapAllocation 析构时会先调用这个 override，随后基类 MemoryMapAllocation::~MemoryMapAllocation() 还会调用基类 close()；基类 close() 在检查 closed_ 之前执行 if (!closed_fd_) CloseHandle(...)。因此当 writer tensor 发送后 refcount 仍未归零时，这里不会设置 closed_fd_，基类析构仍会把 section HANDLE 提前关掉，reader 还没 OpenFileMappingA 的竞态仍存在；同时当前对象自己的 view 也没有 UnmapViewOfFile，会在 worker 持续出 batch 时泄漏映射。
请把 refcounted mapping 的 HANDLE/view 生命周期从基类 close() 中拆出来：refcount 未归零时不能让基类析构关闭该 HANDLE；若需要保留 HANDLE 让 reader 打开，也要转交给一个可在最终 refcount 归零或 worker 清理时关闭/Unmap 的 owner。示意：
const bool last_ref = (--info->refcount == 0);
#ifdef _WIN32
if (last_ref) {
if (fd_ != -1 && !closed_fd_) {
CloseHandle(reinterpret_cast<HANDLE>(fd_));
closed_fd_ = true;
}
UnmapViewOfFile(map_ptr_);
} else {
// Transfer fd_/map_ptr_ to a keeper that is cleaned when the reader
// side has opened/finished, and prevent the base destructor from
// closing this HANDLE in the meantime.
TransferPendingMapping(ipc_name_, fd_, map_ptr_, map_size_);
fd_ = -1;
closed_fd_ = true;
}
#endifThere was a problem hiding this comment.
更新：这次补丁在 refcount > 0 路径调用了 UnmapViewOfFile，所以前面提到的 mapped view 按 batch 泄漏已有推进；但 HANDLE 生命周期仍未收口。当前新分支只是设置 closed_fd_ = true 来阻止基类析构 CloseHandle，没有关闭 fd_，也没有把这个 HANDLE 转交给任何后续可清理的 owner。RefcountedMemoryMapAllocation 析构后这个 HANDLE 值随对象消失，只能等 worker 进程退出由 OS 清理；长时间 DataLoader 仍会按 batch 积累 CreateFileMapping HANDLE/section 对象。
请不要把进程退出作为正常释放路径。建议把 refcount > 0 时必须保留的 Windows HANDLE 放入一个明确的 keeper，并在 reader 已确认打开后或 worker 的 mmap cleanup 中关闭；或者改为可安全传递/复制的 HANDLE 方案，避免 sender allocation 析构后留下不可回收句柄。修复形态应类似：
if (info->refcount > 0) {
WindowsMappingKeeper::Instance().Insert(ipc_name_, fd_);
fd_ = -1;
closed_fd_ = true;
UnmapViewOfFile(map_ptr_);
}
// reader 已打开或 worker cleanup 时：
WindowsMappingKeeper::Instance().Close(ipc_name_);There was a problem hiding this comment.
这里我不同意“与 Linux 对齐”的判断，关键差异在当前代码里可以直接看到：
- Linux 默认不设置
MAPPED_KEEPFD时并不会把 fd 泄漏到进程退出。POSIX 分支在mmap_allocator.cc:226-235里只有MAPPED_KEEPFD才保留 fd，否则mmap后立刻::close(fd)并把shared_fd设为-1。后续保持对象可打开的是 shm 名称和shm_unlink语义，不是 fd。 - Windows 分支这次是主动把
CreateFileMappingA返回的HANDLE保存到fd_，因为当前注释也说明了它用于让 named section 在 readerOpenFileMappingA前保持存活。当前refcount > 0分支只设置closed_fd_ = true并UnmapViewOfFile，没有CloseHandle(fd_)，也没有把fd_转交给任何后续清理 owner；对象析构后这个 HANDLE 值就丢失，只能等 worker 进程退出。 - 因此长期运行或 persistent worker 场景下，Windows 会按 batch 累积
CreateFileMappingHANDLE/section owner；这不是 Linux 默认路径的行为。
WindowsMappingKeeper 不是必须的具体类名，但需要有等价的明确生命周期 owner：例如 reader 成功打开后能通知 sender 关闭保留 HANDLE，或 worker cleanup 能遍历并 CloseHandle 这些 pending handles。当前仅依赖进程退出回收，仍然不够。
There was a problem hiding this comment.
更新：当前提交只在 info->refcount > 1 时立即 CloseHandle，但这没有覆盖主要泄漏路径。发送端 _reduce_lodtensor() 会先 _shared_incref()；如果发送端 holder 在 reader rebuild 后释放，reader 侧 _new_shared_filename() 随即又调用 _shared_decref()，所以发送端 close() 里先 --info->refcount 后通常仍会落到 refcount == 1，进入 WindowsHandleKeeper::Insert()。如果 reader 还没打开，也同样是 refcount == 1。而当前 keeper 只有 Insert()/CloseAll()，没有在 reader 打开或 refcount 归零后按 ipc_name 关闭的路径；worker.py 的发送循环在 out_queue.put() 后也不会调用 _remove_tensor_list_mmap_fds，只在 finally 里 _cleanup_mmap()。因此长 epoch / persistent worker 下，落入 refcount == 1 的 HANDLE 仍会按 batch 保留到 worker 退出，前面讨论的资源线性增长没有闭环。
建议把 refcount == 1 转交 keeper 的条目做成可回收状态，而不是只保存 HANDLE 到退出。例如 keeper 同时保存 base view 并在每次插入/周期性 cleanup 时回收 refcount == 0 的条目：
struct PendingMapping {
intptr_t fd;
void* map_ptr;
size_t map_size;
};
void WindowsHandleKeeper::Insert(const std::string& ipc_name,
intptr_t fd,
void* map_ptr,
size_t map_size) {
SweepClosedMappings();
handles_[ipc_name] = {fd, map_ptr, map_size};
}
void WindowsHandleKeeper::SweepClosedMappings() {
for (auto it = handles_.begin(); it != handles_.end();) {
auto* info = reinterpret_cast<CountInfo*>(it->second.map_ptr);
if (info->refcount == 0) {
UnmapViewOfFile(it->second.map_ptr);
CloseHandle(reinterpret_cast<HANDLE>(it->second.fd));
it = handles_.erase(it);
} else {
++it;
}
}
}对应地，refcount == 1 分支不要先 UnmapViewOfFile(map_ptr_) 后只保存 fd_；否则 keeper 后续无法判断 shared refcount，也无法在正常训练过程中释放这些 pending HANDLE。
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
…k); ruff format test_modelaverage.py
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。test_modelaverage.py 的 ruff format 已修复；Windows refcount > 0 分支也已经补上 UnmapViewOfFile，前序 mapped view 泄漏问题有推进。
不过 refcounted mmap 的 Windows HANDLE 生命周期仍未完全收口：当前补丁会阻止基类析构关闭 HANDLE，但没有把该 HANDLE 交给后续可清理的 owner，细节已在既有 Nyanpasu 行级线程补充。前序 intptr_t 到 int 的 metadata 收窄问题也仍由既有线程覆盖，本轮不重复开新线程。
当前 Pre Commit 还因 mmap_allocator.cc 新增 else 的 cpplint brace 规则未通过，需同步处理。
|
@risemeup1111 关于 WindowsMappingKeeper 的分析：不需要。 理由：
|
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。新增的 Windows worker 失败轮询入口方向是对的，但当前抛错路径没有先关闭/kill blocking queue，仍可能让主线程阻塞在读取端；细节已放在行级评论中。
前序 idx 初始化问题已修复；shared_fd metadata 宽度问题仍由既有 Nyanpasu 线程覆盖，本轮不重复展开。当前还需以最终 CI 和审批结果为准。
| if sys.platform == 'win32': | ||
| core._throw_error_if_process_failed() |
There was a problem hiding this comment.
这里直接调用 core._throw_error_if_process_failed() 后，如果 C++ 侧发现 Windows worker 非零退出并抛出 Fatal，异常会从 _get_data() 直接冒出到 reader 线程。_thread_loop() 在调用 _get_data() 的地方没有外层 try，因此不会执行 _exit_thread_unexpectedly() 去 kill() blocking queue；主线程仍可能阻塞在 read_next() 上，只在后台线程日志里看到异常，达不到新增轮询的诊断目的。请在重新抛出前先走现有的异常退出路径，让 blocking queue 唤醒主线程。
| if sys.platform == 'win32': | |
| core._throw_error_if_process_failed() | |
| if sys.platform == 'win32': | |
| try: | |
| core._throw_error_if_process_failed() | |
| except Exception: | |
| self._exit_thread_unexpectedly() | |
| raise |
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。上一轮 Windows worker 失败轮询抛错前未关闭/kill blocking queue 的问题已按建议修复：现在 core._throw_error_if_process_failed() 抛错时会先调用 _exit_thread_unexpectedly() 再重新抛出，可唤醒主线程读取端。
本轮没有新增行级评论。前序 shared_fd metadata 从 intptr_t 收窄到 int 的问题仍由既有 Nyanpasu 线程覆盖，本轮不重复展开；当前还需以最终 CI 和审批结果为准。
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
@PaddlePaddle-bot F2 那个意见是假警报——test/legacy_test 下 753/753 个测试文件都有 if name == 'main'，这是 Paddle 测试的标准写法。 |
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。MemoryMapAllocation::shared_fd() 返回类型已改为 intptr_t，但 shared_fd metadata 在 pybind 读写边界仍有 int 收窄，剩余问题已在既有 Nyanpasu 线程中补充说明，本轮不重复开新的行级评论。
前序 Windows worker 失败轮询和 idx 初始化问题仍看起来已修复；当前还需以最终 CI 和审批结果为准。
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。shared_fd metadata 的 pybind 读写边界已同步改为 intptr_t，前序 Windows HANDLE 宽度收窄问题在当前代码中看起来已修复。
本轮没有新增行级评论。前序 Windows worker 失败轮询和 idx 初始化问题仍看起来已修复；当前还需以最终 CI 和审批结果为准。
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
… keeper when reader hasn't opened (ref==1)
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。refcount > 1 时立即关闭 HANDLE 是一个改进，但 refcount == 1 进入 keeper 的条目仍没有正常训练过程中的回收路径，长时间运行仍可能累积 HANDLE/section；细节已补充在既有 Nyanpasu 行级线程中，本轮不重复开新行级评论。
前序 shared_fd metadata 宽度、Windows worker 失败轮询和 idx 初始化问题仍看起来已修复；当前还需以最终 CI 和审批结果为准。
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
…eper sweep for refcount==1 entries
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
liuhao2638
left a comment
liuhao2638
left a comment
There was a problem hiding this comment.
已复查最新提交。前序 Windows HANDLE 生命周期问题已有推进，keeper 现在保存了 mapping view 以便按 refcount 回收；但这次新增的 keeper 插入路径存在自锁死风险，细节已放在新的行级评论中，需要修复后再合入。
前序 shared_fd metadata 宽度、Windows worker 失败轮询和 idx 初始化问题仍看起来已修复；当前还需以最终 CI 和审批结果为准。
| std::lock_guard<std::mutex> lock(mtx_); | ||
| SweepClosedMappings(); |
There was a problem hiding this comment.
这里会在第一次需要把 refcount == 1 的 Windows mapping 转交给 keeper 时自锁死：Insert() 已持有 mtx_，随后调用的 SweepClosedMappings() 开头又用同一个 std::mutex 构造 lock_guard。std::mutex 不可重入，worker 会卡在这条 HANDLE 保活路径上，后续 batch 也无法送到主进程。
请把 sweep 放到加锁前，或拆出一个要求调用方已持锁的 SweepClosedMappingsLocked()；不要在同一线程重复锁同一把 mutex。局部可先改成：
| std::lock_guard<std::mutex> lock(mtx_); | |
| SweepClosedMappings(); | |
| SweepClosedMappings(); | |
| std::lock_guard<std::mutex> lock(mtx_); |
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。前序 WindowsHandleKeeper::Insert() 自锁问题已通过 locked helper 修复；refcount == 1 的 Windows HANDLE keeper 路径在当前代码中也已有可回收 owner。
本轮没有新增行级评论，未发现新的需要阻塞合入的问题；仍需以最终 CI 和审批结果为准。
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
…destructor from closing HANDLE
|
@luotao1 方便安排人过来看看吗 我自己和写代码的Agent还有这几个评审Agent已经尽可能的把代码弄的好用好看了 |
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。FLAGS_use_shm_cache 的 cache 分支已经避免基类析构提前关闭 HANDLE，但当前补丁没有把该 HANDLE 转交给后续 owner，仍会在 worker 长时间运行时累积到进程退出；细节已放在新的行级评论中。
前序 WindowsHandleKeeper::Insert() 自锁问题仍看起来已修复；当前还需以最终 CI 和审批结果为准。
| #ifdef _WIN32 | ||
| // Prevent base class MemoryMapAllocation::close() from closing the HANDLE | ||
| // (which would destroy the named section before the reader opens it). | ||
| closed_fd_ = true; |
There was a problem hiding this comment.
这里阻止了基类析构关闭 HANDLE，但没有把 fd_ 转交给任何后续 owner。MemoryMapAllocationPool::Insert() 当前只保存 mmap_ptr_/name/size，MemoryMapAllocationPool::Clear() 在 Windows 下也只 UnmapViewOfFile(mmap.mmap_ptr_)，不会 CloseHandle(fd_)；而这个 allocation 析构后 fd_ 值就丢失了。这样虽然修掉了 reader 打开前被提前 CloseHandle 的竞态，但开启 FLAGS_use_shm_cache 后每个被放入 pool 的 mapping HANDLE 会保留到进程退出，长时间 worker 仍会累积内核 HANDLE。
请让 cache 分支也有明确的 HANDLE owner，例如把 fd_ 一并存进 MemoryMapInfo 并在 MemoryMapAllocationPool::Clear() 里关闭，或把该条目转交给 WindowsHandleKeeper/等价结构。修复形态可以是：
struct MemoryMapInfo {
// ...
#ifdef _WIN32
intptr_t fd_ = -1;
#endif
};
// pool 插入时保存 fd_，然后阻止基类重复关闭
MemoryMapAllocationPool::Instance().Insert(
MemoryMapInfo(flags_, map_size_ - mmap_alignment, ipc_name_, map_ptr_, fd_));
fd_ = -1;
closed_fd_ = true;
// pool Clear():
UnmapViewOfFile(mmap.mmap_ptr_);
if (mmap.fd_ != -1) {
CloseHandle(reinterpret_cast<HANDLE>(mmap.fd_));
}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
risemeup1111
left a comment
risemeup1111
left a comment
There was a problem hiding this comment.
已复查最新提交。FLAGS_use_shm_cache 的 cache 分支现在会把 Windows mapping HANDLE 随 MemoryMapInfo 交给 MemoryMapAllocationPool 持有，并在 pool cleanup 时关闭；MemoryMapFdSet::Clear() 也已移除会提前关闭 keeper 条目的 CloseAll()，前序 cache-path HANDLE owner/cleanup 竞态问题看起来已修复。
本轮没有新增行级评论，未发现新的需要阻塞合入的问题；仍需以最终 CI 和审批结果为准。
PaddlePaddle-bot
left a comment
PaddlePaddle-bot
left a comment
There was a problem hiding this comment.
🤖 Paddle-CI-Agent | pr_review |
2026-07-06 19:15:09
📋 Review 摘要
PR 概述：支持 Windows 平台 paddle.io.DataLoader(num_workers > 0)，补齐 mmap allocator、pybind、worker 监控和测试。
变更范围：paddle/phi/core/memory/allocation、paddle/fluid/imperative、paddle/fluid/pybind、python/paddle/io、DataLoader 测试。
影响面 Tag：[Environment Adaptation] [Execute Infrastructure]
问题
未发现新的阻塞性问题。PR 规范问题在下面章节报,不要在这里重复
历史 Findings 修复情况
| Finding | 问题 | 状态 |
|---|---|---|
| F1 | test/** 新增 if __name__ == '__main__' 命中测试环境依赖信号。 |
|
| F2 | Windows worker cleanup 会提前关闭仍等待 reader 打开的共享内存。 | ✅ 已修复 |
📝 PR 规范检查
符合规范；历史标题 Tag 建议已修复为 [Environment Adaptation] ...，描述模板和精度变化字段齐全。
总体评价
本轮按风险优先审查了共享内存生命周期、Windows worker 监控、pybind 导出、Python multiprocessing reductions 和新增测试。当前 diff 已针对历史 F2 增加 WindowsHandleKeeper，并避免 _cleanup_mmap_fds() 在 Windows 上直接关闭仍可能被 reader 打开的 mapping；未全量验证 Windows CI 运行结果。
|
你的PR提交成功，感谢你对开源项目的贡献! |
4 participants
PR Category
Environment Adaptation
PR Types
New features
Description
背景
Windows 平台上 DataLoader 的
num_workers > 0长期不支持。当用户在 Windows 上设置num_workers > 0时，框架会发出警告并将num_workers强制设为 0，退化为单进程模式。对于数据增强复杂、IO 密集的场景，这会导致严重的数据加载性能瓶颈。改动内容
本 PR 通过 3 层修复实现了 Windows 上的多进程 DataLoader：
1. Windows 共享内存生命周期修复（核心 bug）
Windows 上
UnmapViewOfFile会在最后一个引用释放时销毁 named file mapping section。当 worker 的 tensor 被 GC 后，RefcountedMemoryMapAllocation::close()调用UnmapViewOfFile销毁了 section，而此时 reader（主进程）可能还没通过 Queue 拿到 IPC 名称来调用_new_shared_filename。reader 随后打开了一个空 section，读到零值数据。Linux 上无此问题：
munmap不会销毁/dev/shm中的共享内存文件（只有shm_unlink才删除名字，且数据保留至所有映射解除）。修复方法：
AllocateMemoryMap（Windows）：不关闭CreateFileMapping返回的 HANDLE，始终返回给 callerRefcountedMemoryMapAllocation::close()（Windows）：只在info->refcount == 0时才调用UnmapViewOfFile。refcount > 0 时保持 view 映射，让 section 存活2. ParentWatchDog 误判修复
Windows spawn 模式下，
os.getppid()可能返回临时引导进程的 PID 而非真正的父进程 PID。这导致 worker 中的ParentWatchDog误判父进程已死，worker 提前正常退出（exit code 0）。修复方法： 将
os.getppid()比对替换为 Windows APIOpenProcess(PROCESS_QUERY_LIMITED_INFORMATION)+GetExitCodeProcess，直接查询父进程是否存活。3. CUDA 初始化竞争缓解
Windows spawn 模式下，每个 worker 独立加载 Paddle 并初始化 CUDA。当 12+ 个 worker 同时启动时，NVIDIA 驱动初始化可能触发
STATUS_STACK_BUFFER_OVERRUN (0xC0000409)崩溃（由/GS安全检查检测，通过__fastfail直接杀进程，用户态异常处理器无法捕获）。缓解措施： Windows 上
num_workers > 4时，worker 启动间隔 0.05s（stagger），错开 CUDA 初始化窗口。其他改动：
SetUnhandledExceptionFiltercrash 日志记录器（Windows），用于 worker 崩溃诊断mmap_allocator、data_loader等文件中的#ifndef _WIN32守卫PADDLE_API导出标记_share_filename等），解决 MSVC 上类方法注册问题test_dataloader_multiprocess.py（5 个测试用例）修改文件清单
C++（6 个文件）：
paddle/phi/core/memory/allocation/mmap_allocator.{h,cc}— Windows 共享内存实现和生命周期修复paddle/fluid/imperative/data_loader.{h,cc}— Windows 进程监控（OpenProcess + GetExitCodeProcess + WaitForSingleObject）paddle/fluid/pybind/imperative.cc— 无条件的绑定注册paddle/fluid/pybind/tensor.cc— 模块级 pybind 包装（Windows 兼容性）CMake（3 个文件）：
paddle/phi/core/memory/allocation/CMakeLists.txt— 无条件编译 mmap_allocatorpaddle/fluid/imperative/CMakeLists.txt— 无条件编译 data_loaderpaddle/fluid/pybind/CMakeLists.txt— 无条件链接 data_loaderPython（5 个文件）：
python/paddle/io/reader.py— 移除 Windows 限制，保留 macOS 限制python/paddle/io/dataloader/dataloader_iter.py— Worker 启动 staggerpython/paddle/io/dataloader/worker.py— ParentWatchDog 修复、faulthandler、ForkingPickler 注册python/paddle/io/multiprocess_utils.py— 移除平台守卫python/paddle/incubate/multiprocessing/reductions.py— Windows 使用模块级函数测试（1 个文件）：
test/legacy_test/test_dataloader_multiprocess.py— 5 个多进程 DataLoader 测试用例是否引起精度变化
否