diff --git a/paddle/fluid/imperative/CMakeLists.txt b/paddle/fluid/imperative/CMakeLists.txt index fc08c1a8b8dfbb..3cc768d05509c6 100644 --- a/paddle/fluid/imperative/CMakeLists.txt +++ b/paddle/fluid/imperative/CMakeLists.txt @@ -159,11 +159,11 @@ if(NOT WIN32) SRCS heter_ccl_context.cc DEPS phi tensor var_type_traits) endif() - cc_library( - data_loader - SRCS data_loader.cc - DEPS phi common) endif() +cc_library( + data_loader + SRCS data_loader.cc + DEPS phi common) if(WITH_GLOO) cc_library( imperative_gloo_context diff --git a/paddle/fluid/imperative/data_loader.cc b/paddle/fluid/imperative/data_loader.cc index 1ab4a620eb3e6f..77408fb6916268 100644 --- a/paddle/fluid/imperative/data_loader.cc +++ b/paddle/fluid/imperative/data_loader.cc @@ -12,12 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -#ifndef _WIN32 - #include "paddle/fluid/imperative/data_loader.h" -#include -#include #include #include @@ -26,6 +22,14 @@ #include "paddle/fluid/platform/enforce.h" #include "paddle/phi/core/memory/allocation/mmap_allocator.h" +#ifdef _WIN32 +#include +#include +#else +#include +#include +#endif + namespace paddle::imperative { static std::map> load_process_pids; @@ -49,6 +53,7 @@ void EraseLoadProcessPIDs(int64_t key) { } } +#ifndef _WIN32 // sigaction doc: http://man7.org/linux/man-pages/man2/sigaction.2.html // sigemptyset doc: https://linux.die.net/man/3/sigemptyset // siginfo_t doc: https://www.mkssoftware.com/docs/man5/siginfo_t.5.asp @@ -117,16 +122,89 @@ static inline void setSignalHandler(int signal, "An error occurred while setting handler for %s.", strsignal(signal))); } } +#endif + +#ifdef _WIN32 +// Last-chance exception filter: logs crash info before OS terminates the +// process. +static LONG CALLBACK paddle_crash_filter(EXCEPTION_POINTERS *ep) { + fprintf(stderr, + "[PADDLE_CRASH] PID=%lu code=0x%08lX addr=%p\n", + GetCurrentProcessId(), + ep->ExceptionRecord->ExceptionCode, + ep->ExceptionRecord->ExceptionAddress); + fflush(stderr); + return EXCEPTION_EXECUTE_HANDLER; +} +#endif -// Note: maybe need to add other signal handler void SetLoadProcessSignalHandler() { +#ifdef _WIN32 + // Last-chance filter: logs crash details before OS terminates the process. + // This is the ONLY handler on Windows -- VEH is removed because it could + // deadlock if MemoryMapFdSet::Clear() is called while the mutex is held. + SetUnhandledExceptionFilter(&paddle_crash_filter); + VLOG(3) << "DataLoader: crash handler registered"; +#else setSignalHandler(SIGSEGV, &SIGSEGV_handler, nullptr); setSignalHandler(SIGBUS, &SIGBUS_handler, nullptr); setSignalHandler(SIGFPE, &SIGFPE_handler, nullptr); setSignalHandler(SIGTERM, &SIGTERM_handler, nullptr); +#endif } void ThrowErrorIfLoadProcessFailed() { +#ifdef _WIN32 + // On Windows, use WaitForSingleObject + GetExitCodeProcess + // to check for crashed workers. This is called periodically from Python. + int error = 0; + std::set *pids_set = nullptr; + pid_t process_pid = 0; + + for (auto &p : load_process_pids) { + pids_set = &(p.second); + for (auto pid_it = pids_set->begin(); pid_it != pids_set->end(); ++pid_it) { + process_pid = *pid_it; + VLOG(3) << "DataLoader: monitor loader child process " << process_pid; + + HANDLE hProcess = OpenProcess(PROCESS_QUERY_INFORMATION | SYNCHRONIZE, + FALSE, + static_cast(process_pid)); + if (hProcess == NULL) { + // Process may have already exited; GetLastError() == + // ERROR_INVALID_PARAMETER + continue; + } + + DWORD exit_code = STILL_ACTIVE; + if (WaitForSingleObject(hProcess, 0) == WAIT_OBJECT_0) { + // Process has exited + GetExitCodeProcess(hProcess, &exit_code); + CloseHandle(hProcess); + + if (exit_code != 0 && exit_code != STILL_ACTIVE) { + // Exited with error (non-zero exit code) + fprintf(stderr, + "[DL_WORKER_EXIT] pid=%lu exit_code=0x%08lX (%lu)\n", + process_pid, + exit_code, + exit_code); + fflush(stderr); + pids_set->clear(); + PADDLE_THROW(common::errors::Fatal( + "DataLoader process (pid %ld) exited unexpectedly with code " + "0x%lX (%lu). " + "Rerunning with num_workers=0 may give better error trace.", + process_pid, + exit_code, + exit_code)); + } + } else { + CloseHandle(hProcess); + } + } + } +#else int error = 0; std::set *pids_set = nullptr; pid_t process_pid = 0; @@ -190,8 +268,7 @@ void ThrowErrorIfLoadProcessFailed() { } } } +#endif } } // namespace paddle::imperative - -#endif diff --git a/paddle/fluid/imperative/data_loader.h b/paddle/fluid/imperative/data_loader.h index e66a3b9edc3ffd..43f8113859be98 100644 --- a/paddle/fluid/imperative/data_loader.h +++ b/paddle/fluid/imperative/data_loader.h @@ -14,9 +14,15 @@ #pragma once -#ifndef _WIN32 - +#ifdef _WIN32 +#include +// pid_t is not defined on Windows; use int as cross-platform alternative. +// This is compatible with both getpid() (POSIX) and GetCurrentProcessId() +// (Win32). +using pid_t = int; +#else #include +#endif #include #include @@ -31,5 +37,3 @@ extern void ThrowErrorIfLoadProcessFailed(); } // namespace imperative } // namespace paddle - -#endif diff --git a/paddle/fluid/pybind/CMakeLists.txt b/paddle/fluid/pybind/CMakeLists.txt index 620ba61b5a3d03..f533f0637fbe9d 100755 --- a/paddle/fluid/pybind/CMakeLists.txt +++ b/paddle/fluid/pybind/CMakeLists.txt @@ -88,8 +88,9 @@ if(WITH_CUSTOM_DEVICE) endif() endif() +set(PYBIND_DEPS ${PYBIND_DEPS} data_loader) + if(NOT WIN32) - set(PYBIND_DEPS ${PYBIND_DEPS} data_loader) if(WITH_NCCL OR WITH_RCCL) set(PYBIND_DEPS ${PYBIND_DEPS} nccl_context) set(PYBIND_DEPS ${PYBIND_DEPS} heter_ccl_context) diff --git a/paddle/fluid/pybind/imperative.cc b/paddle/fluid/pybind/imperative.cc index 77ce400672033f..b7072fff57c597 100644 --- a/paddle/fluid/pybind/imperative.cc +++ b/paddle/fluid/pybind/imperative.cc @@ -478,7 +478,6 @@ static void VarBaseCopy(std::shared_ptr &src, // NOLINT void BindImperative(py::module *m_ptr) { auto &m = *m_ptr; -#ifndef _WIN32 // Dygraph DataLoader signal handler m.def("_set_process_pids", [](int64_t key, py::object &obj) { PADDLE_ENFORCE_EQ( @@ -611,7 +610,6 @@ void BindImperative(py::module *m_ptr) { memory::allocation::MemoryMapAllocationPool::Instance().SetMaxPoolSize( size); }); -#endif m.def("start_imperative_gperf_profiler", []() { imperative::StartProfile(); }); diff --git a/paddle/fluid/pybind/tensor.cc b/paddle/fluid/pybind/tensor.cc index e3948afd67f6e4..2edb86ff4d9cf9 100644 --- a/paddle/fluid/pybind/tensor.cc +++ b/paddle/fluid/pybind/tensor.cc @@ -1310,7 +1310,7 @@ void BindTensor(pybind11::module &m) { // NOLINT if (find_id != -1) { handle = memory::allocation::MemoryMapAllocationPool::Instance().GetById(find_id).file_name_; // NOLINT } - int shared_fd = -1; + intptr_t shared_fd = -1; auto shared_holder = memory::allocation::AllocateRefcountedMemoryMapAllocation( handle, shared_fd, flags, data_size, find_id); @@ -1361,7 +1361,7 @@ void BindTensor(pybind11::module &m) { // NOLINT // 2. Rebuild Allocation const std::string &ipc_name = t[0].cast(); - const int shared_fd = t[1].cast(); + const intptr_t shared_fd = t[1].cast(); const bool use_file_descriptor = t[6].cast(); size_t size = t[2].cast(); @@ -1559,6 +1559,137 @@ void BindTensor(pybind11::module &m) { // NOLINT .def("indices", [](const phi::SparseCooTensor &self) -> DenseTensor { return self.indices(); }); + + // Module-level wrappers for _share_filename etc. + // These bypass a pybind11 class method registration issue on Windows. + m.def("_share_filename", [](DenseTensor &self, bool use_file_descriptor) { + if (!self.IsInitialized() || self.numel() == 0) + throw std::runtime_error( + "Tensor not initialized or numel is 0. could not pass to " + "shared memory. "); + + auto holder = self.Holder(); + PADDLE_ENFORCE_EQ(phi::is_cpu_place(holder->place()) || + phi::is_cuda_pinned_place(holder->place()), + true, + common::errors::InvalidArgument( + "Tensor is not on CPU. share_filename only " + "support CPU Tensor.")); + + auto *mmap_allocation = + dynamic_cast( + holder.get()); + if (mmap_allocation == nullptr) { + void *data_ptr = self.data(); + size_t data_size = + self.numel() * + framework::SizeOfType(framework::TransToProtoVarType(self.type())); + + int flags = memory::allocation::MAPPED_SHAREDMEM | + memory::allocation::MAPPED_EXCLUSIVE; + if (use_file_descriptor) { + flags = flags | memory::allocation::MAPPED_KEEPFD | + memory::allocation::MAPPED_UNLINK; + } + std::string handle = memory::allocation::GetIPCName(); + int find_id = -1; + if (FLAGS_use_shm_cache) { + find_id = memory::allocation::MemoryMapAllocationPool::Instance() + .FindFromCache(flags, data_size); // NOLINT + } + if (find_id != -1) { + handle = memory::allocation::MemoryMapAllocationPool::Instance() + .GetById(find_id) + .file_name_; // NOLINT + } + intptr_t shared_fd = -1; + auto shared_holder = + memory::allocation::AllocateRefcountedMemoryMapAllocation( + handle, shared_fd, flags, data_size, find_id); + + // On Windows, GPUPinnedPlace copy is not needed (non-CUDA memory sharing) +#ifndef _WIN32 + if (phi::is_cuda_pinned_place(holder->place())) { +#ifdef PADDLE_WITH_CUDA + memory::Copy(CPUPlace(), + shared_holder->ptr(), + phi::GPUPinnedPlace(), + data_ptr, + data_size); +#endif + } else { + memory::Copy( + CPUPlace(), shared_holder->ptr(), CPUPlace(), data_ptr, data_size); + } +#else + memory::Copy( + CPUPlace(), shared_holder->ptr(), CPUPlace(), data_ptr, data_size); +#endif + self.ResetHolder(shared_holder); + mmap_allocation = shared_holder.get(); + } + int type_idx = static_cast(self.type()); + + return py::make_tuple(mmap_allocation->ipc_name(), + mmap_allocation->shared_fd(), + mmap_allocation->size(), + type_idx, + common::vectorize(self.dims()), + self.lod(), + use_file_descriptor); + }); + + m.def("_new_shared_filename", [](py::tuple t) { + if (t.size() != 7) + throw std::runtime_error("Invalid Tensor meta info state!"); + + DenseTensor tensor; + + const std::string &ipc_name = t[0].cast(); + const intptr_t shared_fd = t[1].cast(); + const bool use_file_descriptor = t[6].cast(); + + size_t size = t[2].cast(); + int flags = memory::allocation::MAPPED_SHAREDMEM | + memory::allocation::MAPPED_NOCREATE; + if (use_file_descriptor) { + flags = flags | memory::allocation::MAPPED_KEEPFD | + memory::allocation::MAPPED_UNLINK; + } + int find_id = -1; + if (FLAGS_use_shm_cache) { + find_id = + memory::allocation::MemoryMapAllocationPool::Instance().FindFromCache( + flags, size, ipc_name, false); // NOLINT + } + auto shared_holder = + memory::allocation::AllocateRefcountedMemoryMapAllocation( + ipc_name, shared_fd, flags, size, find_id); + + tensor.ResetHolderWithType(shared_holder, + static_cast(t[3].cast())); + tensor.Resize(common::make_ddim(t[4].cast>())); + + return tensor; + }); + + m.def("_shared_incref", [](DenseTensor &self) { + auto *mmap_allocation = + dynamic_cast( + self.Holder().get()); + if (mmap_allocation) { + mmap_allocation->incref(); + } + }); + + m.def("_shared_decref", [](DenseTensor &self) { + auto *mmap_allocation = + dynamic_cast( + self.Holder().get()); + if (mmap_allocation) { + mmap_allocation->decref(); + } + }); } } // namespace paddle::pybind diff --git a/paddle/phi/core/memory/allocation/CMakeLists.txt b/paddle/phi/core/memory/allocation/CMakeLists.txt index ad327fa68feeed..cbddc59239cd88 100644 --- a/paddle/phi/core/memory/allocation/CMakeLists.txt +++ b/paddle/phi/core/memory/allocation/CMakeLists.txt @@ -36,7 +36,6 @@ if(CUDA_VERSION VERSION_GREATER_EQUAL 10.2) endif() if(NOT WIN32) - list(APPEND ALLOCATOR_SRCS mmap_allocator.cc) if(WITH_GPU) list(APPEND ALLOCATOR_SRCS cuda_ipc_allocator.cc) endif() @@ -45,6 +44,8 @@ if(NOT WIN32) endif() endif() +list(APPEND ALLOCATOR_SRCS mmap_allocator.cc) + if(WITH_CUSTOM_DEVICE) list(APPEND ALLOCATOR_SRCS custom_allocator.cc stream_safe_custom_device_allocator.cc) diff --git a/paddle/phi/core/memory/allocation/mmap_allocator.cc b/paddle/phi/core/memory/allocation/mmap_allocator.cc index 252a4aa7479bdc..0ee97258f1c8f9 100644 --- a/paddle/phi/core/memory/allocation/mmap_allocator.cc +++ b/paddle/phi/core/memory/allocation/mmap_allocator.cc @@ -12,12 +12,16 @@ // See the License for the specific language governing permissions and // limitations under the License. -#ifndef _WIN32 - #include "paddle/phi/core/memory/allocation/mmap_allocator.h" -#include -#include +#include + +#ifdef _WIN32 +#include +#else +#include +#endif + #include #include @@ -28,6 +32,19 @@ #include "paddle/common/flags.h" #include "paddle/phi/core/enforce.h" +#ifdef _WIN32 +static inline int GetPid() { return static_cast(GetCurrentProcessId()); } +#else +static inline int GetPid() { return getpid(); } +#endif + +#ifdef _WIN32 +#include +#else +#include +#include +#endif + COMMON_DECLARE_bool(use_shm_cache); namespace paddle::memory::allocation { @@ -39,7 +56,7 @@ std::string GetIPCName() { #ifdef _WIN32 handle += std::to_string(GetCurrentProcessId()); #else - handle += std::to_string(getpid()); + handle += std::to_string(GetPid()); #endif handle += "_"; handle += std::to_string(counter.fetch_add(1)); @@ -52,12 +69,112 @@ struct CountInfo { std::atomic refcount; }; -void AllocateMemoryMap(std::string filename, - int *shared_fd, +void AllocateMemoryMap(std::string *fname_ptr, + intptr_t *shared_fd, int flags, size_t size, void **map_ptr_) { - // TODO(@ZHUI): support win32 + std::string &fname = *fname_ptr; +#ifdef _WIN32 + DWORD protect = (flags & MAPPED_SHAREDMEM) ? PAGE_READWRITE : PAGE_READONLY; + // For reader path (MAPPED_NOCREATE): open existing section, never create. + if (flags & MAPPED_NOCREATE) { + HANDLE hMap = OpenFileMappingA(FILE_MAP_ALL_ACCESS, FALSE, fname.c_str()); + PADDLE_ENFORCE_NE(hMap, + nullptr, + common::errors::Unavailable( + "OpenFileMappingA failed for %s, error: %lu", + fname.c_str(), + GetLastError())); + void *ptr = MapViewOfFile(hMap, FILE_MAP_ALL_ACCESS, 0, 0, size); + if (ptr == nullptr) { + DWORD err = GetLastError(); + CloseHandle(hMap); + PADDLE_THROW(common::errors::Unavailable( + "MapViewOfFile for reader %s failed, error: %lu", + fname.c_str(), + err)); + } + *shared_fd = reinterpret_cast(hMap); + *map_ptr_ = ptr; + return; + } + // Retry loop for exclusive mode: on Windows, CreateFileMapping silently + // returns the existing mapping if the name already exists, unlike Linux's + // shm_open + O_EXCL which fails atomically. We must check explicitly. + for (int attempt = 0; attempt < 100; attempt++) { + HANDLE hMap = CreateFileMappingA(INVALID_HANDLE_VALUE, + NULL, + protect, + static_cast(size >> 32), + static_cast(size & 0xffffffffULL), + fname.c_str()); + + if (hMap == NULL) { + DWORD err = GetLastError(); + // Write diagnostic to stderr before throwing + fprintf(stderr, + "[PADDLE_MMAP] PID=%lu CreateFileMapping FAILED " + "name=%s size=%zu flags=0x%x attempt=%d err=%lu\n", + GetCurrentProcessId(), + fname.c_str(), + size, + flags, + attempt, + err); + fflush(stderr); + PADDLE_THROW(common::errors::Unavailable( + "CreateFileMapping failed for %s, error: %lu", fname.c_str(), err)); + } + + if ((flags & MAPPED_EXCLUSIVE) && GetLastError() == ERROR_ALREADY_EXISTS) { + CloseHandle(hMap); + VLOG(3) << "[PADDLE_MMAP] PID=" << GetCurrentProcessId() + << " name collision, retrying attempt=" << attempt + << " name=" << fname; + fname = GetIPCName(); // name collision; retry with fresh name + continue; + } + + DWORD access = FILE_MAP_ALL_ACCESS; + *map_ptr_ = MapViewOfFile(hMap, access, 0, 0, size); + if (*map_ptr_ == nullptr) { + DWORD err = GetLastError(); + CloseHandle(hMap); + fprintf(stderr, + "[PADDLE_MMAP] PID=%lu MapViewOfFile FAILED " + "name=%s size=%zu err=%lu\n", + GetCurrentProcessId(), + fname.c_str(), + size, + err); + fflush(stderr); + PADDLE_THROW(common::errors::Unavailable( + "MapViewOfFile failed for %s, error: %lu", fname.c_str(), err)); + } + + // On Windows, always keep the HANDLE so the section stays alive + // after the caller's MapViewOfFile. The handle is closed in + // RefcountedMemoryMapAllocation::close(). Without this, the section + // would be destroyed when the last view is unmapped (i.e. when the + // worker's tensor is GC'd), before the reader opens it -- this is + // the root cause of "Blocking queue is killed" with large data. + // Linux doesn't have this issue because munmap never destroys the + // shared memory file (only shm_unlink does). + *shared_fd = reinterpret_cast(hMap); + + if (flags & MAPPED_UNLINK) { + VLOG(6) << "CreateFileMapping (unlink mode): " << fname; + } + + // Caller (AllocateRefcountedMemoryMapAllocation) handles Insert + return; + } + + PADDLE_THROW(common::errors::Unavailable( + "Failed to allocate exclusive shared memory after 100 retries")); +#else + // Linux implementation using shm_open + mmap int file_flags = 0; int fd = *shared_fd; if (flags & MAPPED_SHAREDMEM) { @@ -74,15 +191,15 @@ void AllocateMemoryMap(std::string filename, if (!(flags & MAPPED_FROMFD) && fd == -1) { if (flags & MAPPED_SHAREDMEM) { - fd = shm_open(filename.c_str(), file_flags, (mode_t)0600); + fd = shm_open(fname.c_str(), file_flags, (mode_t)0600); PADDLE_ENFORCE_NE( fd, -1, common::errors::Unavailable( "File descriptor %s open failed, unable in read-write mode", - filename.c_str())); - VLOG(6) << "shm_open: " << filename; - MemoryMapFdSet::Instance().Insert(filename); + fname.c_str())); + VLOG(6) << "shm_open: " << fname; + MemoryMapFdSet::Instance().Insert(fname); } } @@ -98,8 +215,8 @@ void AllocateMemoryMap(std::string filename, } if (flags & MAPPED_UNLINK) { - VLOG(6) << "shm_unlink: " << filename; - shm_unlink(filename.c_str()); + VLOG(6) << "shm_unlink: " << fname; + shm_unlink(fname.c_str()); } PADDLE_ENFORCE_NE(*map_ptr_, @@ -113,38 +230,40 @@ void AllocateMemoryMap(std::string filename, PADDLE_ENFORCE_NE(::close(fd), -1, common::errors::Unavailable( - "Error closing memory mapped file %s", filename)); - + "Error closing memory mapped file %s", fname)); *shared_fd = -1; } +#endif } std::shared_ptr -AllocateRefcountedMemoryMapAllocation(std::string filename, - int shared_fd, +AllocateRefcountedMemoryMapAllocation(std::string fname, + intptr_t shared_fd, int flags, size_t size, int buffer_id) { - int fd = shared_fd; + intptr_t fd = shared_fd; void *base_ptr = nullptr; if (buffer_id == -1) { - AllocateMemoryMap(filename, &fd, flags, size + mmap_alignment, &base_ptr); - VLOG(4) << "Create and mmap a new shm: " << filename; + AllocateMemoryMap(&fname, &fd, flags, size + mmap_alignment, &base_ptr); + VLOG(4) << "Create and mmap a new shm: " << fname; } else { base_ptr = MemoryMapAllocationPool::Instance().GetById(buffer_id).mmap_ptr_; - VLOG(4) << "Get a cached shm " << filename; + fd = shared_fd; + VLOG(4) << "Get a cached shm " << fname; } void *aligned_base_ptr = static_cast(static_cast(base_ptr) + mmap_alignment); + MemoryMapFdSet::Instance().Insert(fname); return std::make_shared( - aligned_base_ptr, size, filename, fd, flags, buffer_id); + aligned_base_ptr, size, fname, fd, flags, buffer_id); } RefcountedMemoryMapAllocation::RefcountedMemoryMapAllocation( void *ptr, size_t size, std::string ipc_name, - int fd, + intptr_t fd, int flags, int buffer_id) : MemoryMapAllocation(ptr, size, ipc_name, fd, flags) { @@ -159,12 +278,21 @@ RefcountedMemoryMapAllocation::RefcountedMemoryMapAllocation( void MemoryMapAllocation::close() { if (!closed_fd_) { closed_fd_ = true; +#ifdef _WIN32 + // On Windows, the HANDLE from CreateFileMapping is always kept + // (never closed in AllocateMemoryMap). Close it unconditionally + // to prevent handle leaks. + if (fd_ != -1) { + CloseHandle(reinterpret_cast(fd_)); + } +#else if (flags_ & MAPPED_KEEPFD) { PADDLE_ENFORCE_NE(::close(fd_), -1, common::errors::Unavailable( "Error closing file descriptor <%d>", fd_)); } +#endif } if (closed_) { return; @@ -209,16 +337,13 @@ void RefcountedMemoryMapAllocation::close() { void *data = map_ptr_; CountInfo *info = reinterpret_cast(data); --info->refcount; - if (flags_ & MAPPED_KEEPFD) { - closed_fd_ = true; - PADDLE_ENFORCE_NE( - ::close(fd_), - -1, - common::errors::Unavailable("Error closing file descriptor <%d>", fd_)); - VLOG(6) << "close fd: " << fd_; - } if (FLAGS_use_shm_cache && buffer_id_ != -1) { +#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; +#endif return; } else { if (FLAGS_use_shm_cache && @@ -226,31 +351,109 @@ void RefcountedMemoryMapAllocation::close() { static_cast( MemoryMapAllocationPool::Instance().MaxPoolSize())) { MemoryMapAllocationPool::Instance().Insert(MemoryMapInfo( - flags_, map_size_ - mmap_alignment, ipc_name_, map_ptr_)); + flags_, + map_size_ - mmap_alignment, + ipc_name_, + map_ptr_ +#ifdef _WIN32 + , + fd_ +#endif + )); +#ifdef _WIN32 + // Prevent base class destructor from closing the HANDLE. + // When the buffer is later retrieved, a new RefcountedMMap is + // constructed from the same map_ptr_ and increments refcount — the + // HANDLE must stay alive so the named section survives until then. + closed_fd_ = true; +#endif } else { if (info->refcount == 0) { +#ifdef _WIN32 + // Only close the handle AND unmap when refcount reaches 0. + // On Windows, CloseHandle destroys the named file mapping section. + // Closing it early (while refcount > 0) would delete the section + // before the reader process has a chance to OpenFileMappingA it. + if (fd_ != -1 && !closed_fd_) { + closed_fd_ = true; + CloseHandle(reinterpret_cast(fd_)); + VLOG(6) << "close handle: " << fd_; + } + VLOG(6) << "UnmapViewOfFile: " << ipc_name_; + UnmapViewOfFile(map_ptr_); +#else + if (flags_ & MAPPED_KEEPFD) { + closed_fd_ = true; + PADDLE_ENFORCE_NE(::close(fd_), + -1, + common::errors::Unavailable( + "Error closing file descriptor <%d>", fd_)); + VLOG(6) << "close fd: " << fd_; + } shm_unlink(ipc_name_.c_str()); VLOG(6) << "shm_unlink file: " << ipc_name_; +#endif + } else { +#ifdef _WIN32 + if (info->refcount > 1) { + // Refcount > 1 means the reader has already opened this section + // (initializeRefercount was called after MapViewOfFile). The + // reader's mapping keeps the section alive, so we can safely + // CloseHandle immediately — no accumulation needed. + VLOG(6) << "close handle (reader already opened): " << ipc_name_; + CloseHandle(reinterpret_cast(fd_)); + fd_ = -1; + closed_fd_ = true; + VLOG(6) << "UnmapViewOfFile: " << ipc_name_; + UnmapViewOfFile(map_ptr_); + } else { + // Refcount == 1: reader hasn't opened yet. Transfer both HANDLE + // and map_ptr_ to WindowsHandleKeeper so the named section stays + // alive until the reader opens it. The keeper will UnmapViewOfFile + // + CloseHandle when refcount reaches 0 (via SweepClosedMappings). + WindowsHandleKeeper::Instance().Insert( + ipc_name_, fd_, map_ptr_, map_size_); + fd_ = -1; + closed_fd_ = true; + } +#endif } - +#ifndef _WIN32 + // On Linux, munmap is always safe since it only unmaps virtual memory; + // the shared memory file in /dev/shm persists until shm_unlink. PADDLE_ENFORCE_NE(munmap(map_ptr_, map_size_), -1, common::errors::Unavailable( "could not unmap the shared memory file: %s (%d)", strerror(errno), errno)); +#endif } } } MemoryMapWriterAllocation::~MemoryMapWriterAllocation() { +#ifdef _WIN32 + UnmapViewOfFile(this->ptr()); + if (fd_ != -1) { + CloseHandle(reinterpret_cast(fd_)); + } +#else if (munmap(this->ptr(), this->size()) == -1) { common::errors::Unavailable("could not unmap the shared memory file %s", this->ipc_name()); } +#endif } MemoryMapReaderAllocation::~MemoryMapReaderAllocation() { +#ifdef _WIN32 + UnmapViewOfFile(this->ptr()); + // On Windows, named file mapping is auto-destroyed when final handle is + // closed. + MemoryMapFdSet::Instance().Remove(this->ipc_name()); + VLOG(3) << "~MemoryMapReaderAllocation: " << this->ipc_name(); +#else if (munmap(this->ptr(), this->size()) == -1) { common::errors::Unavailable("could not unmap the shared memory file %s", this->ipc_name()); @@ -276,11 +479,41 @@ MemoryMapReaderAllocation::~MemoryMapReaderAllocation() { shm_unlink(this->ipc_name().c_str()); MemoryMapFdSet::Instance().Remove(this->ipc_name()); VLOG(3) << "~MemoryMapReaderAllocation: " << this->ipc_name(); +#endif } std::shared_ptr AllocateMemoryMapWriterAllocation( size_t size) { const std::string &ipc_name = GetIPCName(); +#ifdef _WIN32 + HANDLE hMap = CreateFileMappingA(INVALID_HANDLE_VALUE, + NULL, + PAGE_READWRITE, + static_cast(size >> 32), + static_cast(size & 0xffffffffULL), + ipc_name.c_str()); + PADDLE_ENFORCE_NE( + hMap, + nullptr, + common::errors::Unavailable("CreateFileMapping for writer %s failed", + ipc_name.c_str())); + + void *ptr = MapViewOfFile(hMap, FILE_MAP_ALL_ACCESS, 0, 0, size); + if (ptr == nullptr) { + DWORD err = GetLastError(); + CloseHandle(hMap); + PADDLE_THROW(common::errors::Unavailable( + "MapViewOfFile for writer %s failed, error: %lu", + ipc_name.c_str(), + err)); + } + // Keep the HANDLE open until ~MemoryMapWriterAllocation (where + // UnmapViewOfFile runs first, then CloseHandle). Closing it here + // would destroy the named section as soon as the writer unmaps, + // before the reader has a chance to OpenFileMappingA it. + return std::make_shared( + ptr, size, ipc_name, reinterpret_cast(hMap)); +#else int flags = O_RDWR | O_CREAT; int fd = shm_open(ipc_name.c_str(), flags, 0600); @@ -301,10 +534,32 @@ std::shared_ptr AllocateMemoryMapWriterAllocation( close(fd); return std::make_shared(ptr, size, ipc_name); +#endif } std::shared_ptr RebuildMemoryMapReaderAllocation( const std::string &ipc_name, size_t size) { +#ifdef _WIN32 + HANDLE hMap = OpenFileMappingA(FILE_MAP_ALL_ACCESS, FALSE, ipc_name.c_str()); + PADDLE_ENFORCE_NE(hMap, + nullptr, + common::errors::Unavailable( + "OpenFileMappingA for reader %s failed, error: %lu", + ipc_name.c_str(), + GetLastError())); + + void *ptr = MapViewOfFile(hMap, FILE_MAP_ALL_ACCESS, 0, 0, size); + if (ptr == nullptr) { + DWORD err = GetLastError(); + CloseHandle(hMap); + PADDLE_THROW(common::errors::Unavailable( + "MapViewOfFile for reader %s failed, error: %lu", + ipc_name.c_str(), + err)); + } + CloseHandle(hMap); + return std::make_shared(ptr, size, ipc_name); +#else int flags = O_RDWR | O_CREAT; flags &= ~O_CREAT; int fd = shm_open(ipc_name.c_str(), flags, 0600); @@ -319,6 +574,7 @@ std::shared_ptr RebuildMemoryMapReaderAllocation( "Memory map failed when rebuild shared memory.")); close(fd); return std::make_shared(ptr, size, ipc_name); +#endif } MemoryMapFdSet &MemoryMapFdSet::Instance() { // NOLINT @@ -329,34 +585,46 @@ MemoryMapFdSet &MemoryMapFdSet::Instance() { // NOLINT void MemoryMapFdSet::Insert(const std::string &ipc_name) { std::lock_guard guard(mtx_); fd_set_.emplace(ipc_name); - VLOG(3) << "PID: " << getpid() << ", MemoryMapFdSet: insert " << ipc_name + VLOG(3) << "PID: " << GetPid() << ", MemoryMapFdSet: insert " << ipc_name << ", set size: " << fd_set_.size(); } void MemoryMapFdSet::Remove(const std::string &ipc_name) { std::lock_guard guard(mtx_); fd_set_.erase(ipc_name); - VLOG(3) << "PID: " << getpid() << ", MemoryMapFdSet: erase " << ipc_name + VLOG(3) << "PID: " << GetPid() << ", MemoryMapFdSet: erase " << ipc_name << ", set size: " << fd_set_.size(); } void MemoryMapFdSet::Clear() { - VLOG(7) << "PID: " << getpid() << ", MemoryMapFdSet: set size - " + VLOG(7) << "PID: " << GetPid() << ", MemoryMapFdSet: set size - " << fd_set_.size(); std::lock_guard guard(mtx_); for (auto const &fd : fd_set_) { +#ifdef _WIN32 + // On Windows, named file mappings are auto-destroyed when the last view + // is unmapped. Nothing to unlink explicitly. + VLOG(7) << "PID: " << GetPid() << ", MemoryMapFdSet: clear " << fd; +#else int rlt = shm_unlink(fd.c_str()); if (rlt == 0) { - VLOG(7) << "PID: " << getpid() << ", MemoryMapFdSet: clear " << fd; + VLOG(7) << "PID: " << GetPid() << ", MemoryMapFdSet: clear " << fd; } +#endif } fd_set_.clear(); +#ifdef _WIN32 + // NOTE: Do NOT CloseAll() here. The keeper's SweepClosedMappings + // reclaims entries whose refcount has reached 0 during normal operation. + // HANDLEs still in the keeper at process exit are reclaimed by the OS. + // Calling CloseAll() here while batches are still in-flight in the + // multiprocessing queue would destroy named sections before the reader + // has a chance to OpenFileMappingA them. +#endif } MemoryMapFdSet::~MemoryMapFdSet() { Clear(); } -MemoryMapAllocationPool *MemoryMapAllocationPool::pool_ = nullptr; - void MemoryMapAllocationPool::Insert(const MemoryMapInfo &memory_map) { std::lock_guard guard(mtx_); memory_map_allocations_.push_back(memory_map); @@ -399,6 +667,13 @@ void MemoryMapAllocationPool::SetMaxPoolSize(const int &size) { void MemoryMapAllocationPool::Clear() { std::lock_guard guard(mtx_); for (auto const &mmap : memory_map_allocations_) { +#ifdef _WIN32 + VLOG(4) << "MemoryMapAllocationPool: clear " << mmap.file_name_; + UnmapViewOfFile(mmap.mmap_ptr_); + if (mmap.fd_ != -1) { + CloseHandle(reinterpret_cast(mmap.fd_)); + } +#else int rlt = shm_unlink(mmap.file_name_.c_str()); if (rlt == 0) { VLOG(4) << "MemoryMapAllocationPool: clear " << mmap.file_name_; @@ -409,12 +684,67 @@ void MemoryMapAllocationPool::Clear() { "could not unmap the shared memory file: %s (%d)", strerror(errno), errno)); +#endif } memory_map_allocations_.clear(); } MemoryMapAllocationPool::~MemoryMapAllocationPool() { Clear(); } // NOLINT -} // namespace paddle::memory::allocation +#ifdef _WIN32 +WindowsHandleKeeper &WindowsHandleKeeper::Instance() { + // Leaky singleton: heap-allocated, never destructed, to avoid static + // destruction ordering conflicts with MemoryMapFdSet. Normal cleanup + // is driven by MemoryMapFdSet::Clear() / _cleanup_mmap_fds(); on + // abnormal process exit the OS reclaims the memory. + static auto *keeper = new WindowsHandleKeeper(); + return *keeper; +} + +void WindowsHandleKeeper::Insert(const std::string &ipc_name, + intptr_t fd, + void *map_ptr, + size_t map_size) { + std::lock_guard lock(mtx_); + SweepClosedMappingsLocked(); + handles_[ipc_name] = {fd, map_ptr}; +} +void WindowsHandleKeeper::SweepClosedMappingsLocked() { + for (auto it = handles_.begin(); it != handles_.end();) { + // CountInfo::refcount is the first field at map_ptr. A value of 0 means + // all references (including the reader's) have been released — the + // section is no longer in use and can be cleaned up. + auto *refcnt = + static_cast *>(it->second.map_ptr); + if (refcnt->load(std::memory_order_acquire) == 0) { + VLOG(6) << "WindowsHandleKeeper sweeping: " << it->first; + UnmapViewOfFile(it->second.map_ptr); + CloseHandle(reinterpret_cast(it->second.fd)); + it = handles_.erase(it); + } else { + ++it; + } + } +} + +void WindowsHandleKeeper::SweepClosedMappings() { + std::lock_guard lock(mtx_); + SweepClosedMappingsLocked(); +} + +void WindowsHandleKeeper::CloseAll() { + std::lock_guard lock(mtx_); + for (auto &pair : handles_) { + if (pair.second.fd != -1) { + UnmapViewOfFile(pair.second.map_ptr); + CloseHandle(reinterpret_cast(pair.second.fd)); + } + } + handles_.clear(); +} + +WindowsHandleKeeper::~WindowsHandleKeeper() { CloseAll(); } // NOLINT #endif + +} // namespace paddle::memory::allocation diff --git a/paddle/phi/core/memory/allocation/mmap_allocator.h b/paddle/phi/core/memory/allocation/mmap_allocator.h index cb330eff8eb6e2..83db302d07b91f 100644 --- a/paddle/phi/core/memory/allocation/mmap_allocator.h +++ b/paddle/phi/core/memory/allocation/mmap_allocator.h @@ -14,22 +14,23 @@ #pragma once -#ifndef _WIN32 - #include +#include #include #include #include +#include #include #include +#include "paddle/phi/api/include/dll_decl.h" #include "paddle/phi/core/memory/allocation/allocator.h" namespace paddle { namespace memory { namespace allocation { -std::string GetIPCName(); +PADDLE_API std::string GetIPCName(); static constexpr int64_t mmap_alignment = 64; @@ -42,19 +43,19 @@ enum MappedModes { MAPPED_UNLINK = 32 }; -class MemoryMapAllocation : public Allocation { +class PADDLE_API MemoryMapAllocation : public Allocation { public: explicit MemoryMapAllocation(void *ptr, size_t size, std::string ipc_name, - int fd) + intptr_t fd) : Allocation(ptr, size, CPUPlace()), ipc_name_(std::move(ipc_name)), fd_(fd), map_ptr_(ptr), map_size_(size) {} explicit MemoryMapAllocation( - void *ptr, size_t size, std::string ipc_name, int fd, int flags) + void *ptr, size_t size, std::string ipc_name, intptr_t fd, int flags) : Allocation(ptr, size, CPUPlace()), ipc_name_(std::move(ipc_name)), fd_(fd), @@ -63,7 +64,7 @@ class MemoryMapAllocation : public Allocation { map_size_(size) {} inline const std::string &ipc_name() const { return ipc_name_; } - inline int shared_fd() const { return fd_; } + inline intptr_t shared_fd() const { return fd_; } virtual void close(); @@ -71,7 +72,7 @@ class MemoryMapAllocation : public Allocation { protected: std::string ipc_name_; - int fd_ = -1; + intptr_t fd_ = -1; int flags_ = 0; void *map_ptr_ = nullptr; size_t map_size_ = 0; @@ -79,13 +80,13 @@ class MemoryMapAllocation : public Allocation { bool closed_fd_ = false; }; -class RefcountedMemoryMapAllocation : public MemoryMapAllocation { +class PADDLE_API RefcountedMemoryMapAllocation : public MemoryMapAllocation { public: RefcountedMemoryMapAllocation(void *ptr, size_t size, std::string ipc_name, + intptr_t fd, int flags, - int fd, int buffer_id = -1); void incref(); @@ -99,37 +100,44 @@ class RefcountedMemoryMapAllocation : public MemoryMapAllocation { void resetBaseptr(); }; -void AllocateMemoryMap(std::string filename, - int *shared_fd, - int flags, - size_t size, - void **base_ptr_); +PADDLE_API void AllocateMemoryMap(std::string *filename, + intptr_t *shared_fd, + int flags, + size_t size, + void **base_ptr_); -std::shared_ptr +PADDLE_API std::shared_ptr AllocateRefcountedMemoryMapAllocation(std::string filename, - int shared_fd, + intptr_t shared_fd, int flags, size_t size, int buffer_id = -1); -class MemoryMapWriterAllocation : public Allocation { +class PADDLE_API MemoryMapWriterAllocation : public Allocation { public: explicit MemoryMapWriterAllocation(void *ptr, size_t size, std::string ipc_name) : Allocation(ptr, size, CPUPlace()), ipc_name_(std::move(ipc_name)) {} + explicit MemoryMapWriterAllocation(void *ptr, + size_t size, + std::string ipc_name, + intptr_t fd) + : Allocation(ptr, size, CPUPlace()), + ipc_name_(std::move(ipc_name)), + fd_(fd) {} inline const std::string &ipc_name() const { return ipc_name_; } - inline int shared_fd() const { return fd_; } + inline intptr_t shared_fd() const { return fd_; } ~MemoryMapWriterAllocation() override; private: std::string ipc_name_; - int fd_ = -1; + intptr_t fd_ = -1; }; -class MemoryMapReaderAllocation : public Allocation { +class PADDLE_API MemoryMapReaderAllocation : public Allocation { public: explicit MemoryMapReaderAllocation(void *ptr, size_t size, @@ -146,13 +154,13 @@ class MemoryMapReaderAllocation : public Allocation { int fd_ = -1; }; -std::shared_ptr AllocateMemoryMapWriterAllocation( - size_t size); +PADDLE_API std::shared_ptr +AllocateMemoryMapWriterAllocation(size_t size); -std::shared_ptr RebuildMemoryMapReaderAllocation( - const std::string &ipc_name, size_t size); +PADDLE_API std::shared_ptr +RebuildMemoryMapReaderAllocation(const std::string &ipc_name, size_t size); -class MemoryMapFdSet { +class PADDLE_API MemoryMapFdSet { public: static MemoryMapFdSet &Instance(); // NOLINT @@ -171,21 +179,67 @@ class MemoryMapFdSet { std::mutex mtx_; }; +#ifdef _WIN32 +// Tracks HANDLEs and mapped views from CreateFileMappingA that must stay +// open (refcount == 1 at writer close time) to keep the named section alive +// for readers. SweepClosedMappings is called on each Insert() and reclaims +// entries whose refcount has reached 0. Remaining entries at process exit +// are cleaned up by the OS (handle closure on process termination). +class PADDLE_API WindowsHandleKeeper { + public: + static WindowsHandleKeeper &Instance(); // NOLINT + + struct PendingMapping { + intptr_t fd; + void *map_ptr; + }; + + void Insert(const std::string &ipc_name, + intptr_t fd, + void *map_ptr, + size_t map_size); + void SweepClosedMappings(); + void CloseAll(); + ~WindowsHandleKeeper(); + + private: + WindowsHandleKeeper() = default; + // Caller must hold mtx_ before calling this. + void SweepClosedMappingsLocked(); + std::unordered_map handles_; + std::mutex mtx_; +}; +#endif + class MemoryMapInfo { public: explicit MemoryMapInfo(int flags, size_t data_size, std::string file_name, - void *mmap_ptr) + void *mmap_ptr +#ifdef _WIN32 + , + intptr_t fd = -1 +#endif + ) : flags_(flags), data_size_(data_size), file_name_(file_name), - mmap_ptr_(mmap_ptr) {} + mmap_ptr_(mmap_ptr) +#ifdef _WIN32 + , + fd_(fd) +#endif + { + } int flags_ = 0; size_t data_size_ = 0; std::string file_name_; void *mmap_ptr_ = nullptr; +#ifdef _WIN32 + intptr_t fd_ = -1; +#endif }; /* Note(zhangbo): @@ -197,13 +251,11 @@ created by the _share_filename process will be cached and reused according to the data_size of shm, thus eliminating the problem of munmap blocking other threads */ -class MemoryMapAllocationPool { +class PADDLE_API MemoryMapAllocationPool { public: static MemoryMapAllocationPool &Instance() { - if (pool_ == nullptr) { - pool_ = new MemoryMapAllocationPool(); - } - return *pool_; + static MemoryMapAllocationPool pool; + return pool; } void Insert(const MemoryMapInfo &memory_map); @@ -227,7 +279,8 @@ class MemoryMapAllocationPool { private: MemoryMapAllocationPool() = default; - static MemoryMapAllocationPool *pool_; + MemoryMapAllocationPool(const MemoryMapAllocationPool &) = delete; + MemoryMapAllocationPool &operator=(const MemoryMapAllocationPool &) = delete; std::vector memory_map_allocations_; int max_pool_size_ = 0; std::mutex mtx_; @@ -236,5 +289,3 @@ class MemoryMapAllocationPool { } // namespace allocation } // namespace memory } // namespace paddle - -#endif diff --git a/python/paddle/base/core.py b/python/paddle/base/core.py index c728f498c775a7..813cc0678b4d67 100644 --- a/python/paddle/base/core.py +++ b/python/paddle/base/core.py @@ -362,18 +362,21 @@ def to_list(s): # type promotion # isort: on - if sys.platform != 'win32': - from .libpaddle import ( # noqa: F401 - _array_to_share_memory_tensor, - _cleanup_mmap_fds, - _convert_to_tensor_list, - _erase_process_pids, - _remove_tensor_list_mmap_fds, - _set_max_memory_map_allocation_pool_size, - _set_process_pids, - _set_process_signal_handler, - _throw_error_if_process_failed, - ) + from .libpaddle import ( # noqa: F401 + _array_to_share_memory_tensor, + _cleanup_mmap_fds, + _convert_to_tensor_list, + _erase_process_pids, + _new_shared_filename, + _remove_tensor_list_mmap_fds, + _set_max_memory_map_allocation_pool_size, + _set_process_pids, + _set_process_signal_handler, + _share_filename, + _shared_decref, + _shared_incref, + _throw_error_if_process_failed, + ) except Exception as e: if has_paddle_dy_lib: diff --git a/python/paddle/incubate/multiprocessing/reductions.py b/python/paddle/incubate/multiprocessing/reductions.py index c6fde04319ba2a..b4e8c30d14df92 100644 --- a/python/paddle/incubate/multiprocessing/reductions.py +++ b/python/paddle/incubate/multiprocessing/reductions.py @@ -14,17 +14,18 @@ import copy import multiprocessing +import sys # TODO: check the hooks of tensor # TODO: check serializing named tensor # TODO: check influence on autograd -import sys import threading from collections import OrderedDict from multiprocessing.reduction import ForkingPickler from multiprocessing.util import register_after_fork import paddle +from paddle.base import core def _supported_check(): @@ -135,7 +136,7 @@ def _rebuild_lodtensor_filename( lod, dataloader_use_file_descriptor, ): - lodtensor = cls._new_shared_filename( + lodtensor = core._new_shared_filename( ( ipc_name, shared_fd, @@ -146,7 +147,7 @@ def _rebuild_lodtensor_filename( dataloader_use_file_descriptor, ) ) - lodtensor._shared_decref() + core._shared_decref(lodtensor) return lodtensor @@ -161,7 +162,7 @@ def _rebuild_lodtensor_filedescriptor( dataloader_use_file_descriptor, ): shared_fd = shared_fd.detach() - lodtensor = cls._new_shared_filename( + lodtensor = core._new_shared_filename( ( ipc_name, shared_fd, @@ -172,7 +173,7 @@ def _rebuild_lodtensor_filedescriptor( dataloader_use_file_descriptor, ) ) - lodtensor._shared_decref() + core._shared_decref(lodtensor) return lodtensor @@ -241,9 +242,18 @@ def _reduce_lodtensor(lodtensor): "FLAGS_dataloader_use_file_descriptor" ] # Default use share filename strategy - metadata = lodtensor._share_filename( - dataloader_use_file_descriptor - ) # ipc_name, fd, size, type_idx, dims, lod + try: + metadata = core._share_filename( + lodtensor, dataloader_use_file_descriptor + ) # ipc_name, fd, size, type_idx, dims, lod + except Exception as e: + # Log and re-raise. This exception happens in the Queue feeder + # thread; logging here helps diagnose intermittent crashes. + sys.stderr.write( + f"[REDUCE] _share_filename failed: {type(e).__name__}: {e}\n" + ) + sys.stderr.flush() + raise if dataloader_use_file_descriptor: metalist = list(metadata) @@ -252,8 +262,7 @@ def _reduce_lodtensor(lodtensor): rebuild = _rebuild_lodtensor_filedescriptor else: rebuild = _rebuild_lodtensor_filename - lodtensor._shared_incref() - # TODO, maintain reference for lodtensor + core._shared_incref(lodtensor) elif lodtensor._place().is_gpu_place(): prev_id = paddle.base.core.get_cuda_current_device_id() cur_id = lodtensor._place().gpu_device_id() @@ -280,9 +289,6 @@ def _reduce_lodtensor(lodtensor): def init_reductions() -> None: - if not _supported_check(): - return - ForkingPickler.register(paddle.Tensor, _reduce_tensor) ForkingPickler.register(paddle.base.core.eager.Tensor, _reduce_tensor) ForkingPickler.register( diff --git a/python/paddle/io/dataloader/dataloader_iter.py b/python/paddle/io/dataloader/dataloader_iter.py index 7a5b3c5d04015f..356cf7de4b54d6 100644 --- a/python/paddle/io/dataloader/dataloader_iter.py +++ b/python/paddle/io/dataloader/dataloader_iter.py @@ -407,7 +407,7 @@ def __init__(self, loader): # see _try_put_indices self._thread_lock = threading.Lock() - self._base_seed = np.random.randint(low=0, high=sys.maxsize) + self._base_seed = np.random.randint(low=0, high=np.iinfo(np.int32).max) # Note(zhangbo): shm_buffer_size is used for MemoryMapAllocationPool. # MemoryMapAllocationPool is used to cache and reuse shm, thus reducing munmap in dataloader. @@ -482,10 +482,13 @@ def _init_workers(self): self._use_shared_memory, self._base_seed, self._worker_shm_buffer_size, + os.getpid(), ), ) worker.daemon = True worker.start() + if sys.platform == 'win32' and self._num_workers > 4: + time.sleep(0.05) self._workers.append(worker) self._worker_status.append(True) @@ -722,6 +725,12 @@ def _get_data(self): continue # check failed workers + if sys.platform == 'win32': + try: + core._throw_error_if_process_failed() + except Exception: + self._exit_thread_unexpectedly() + raise failed_workers = [] for i, w in enumerate(self._workers): if self._worker_status[i] and not w.is_alive(): diff --git a/python/paddle/io/dataloader/worker.py b/python/paddle/io/dataloader/worker.py index 94013b4c3ec5b7..c2000dd98b3a21 100644 --- a/python/paddle/io/dataloader/worker.py +++ b/python/paddle/io/dataloader/worker.py @@ -66,13 +66,75 @@ def create_fetcher( class ParentWatchDog: - def __init__(self): - self._parent_pid = os.getppid() + def __init__(self, parent_pid=None): + self._parent_pid = ( + parent_pid if parent_pid is not None else os.getppid() + ) self._parent_alive = True + # On Windows, os.getppid() can return a stale PID from the spawn + # bootstrap process. Use OpenProcess + GetExitCodeProcess instead + # of PPID comparison to reliably detect parent death. + if sys.platform == 'win32': + self._parent_alive = self._parent_alive_win32() + else: + self._parent_alive = True + + def _parent_alive_win32(self): + """Check if parent process is alive using Windows API.""" + import ctypes + from ctypes import wintypes + + kernel32 = ctypes.windll.kernel32 + # Set proper argument/return types for Win64 HANDLE compatibility + 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 + + PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + STILL_ACTIVE = 259 + 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: + err = ctypes.windll.kernel32.GetLastError() + # ERROR_INVALID_PARAMETER (87) means PID doesn't exist. + # ERROR_ACCESS_DENIED (5) means process may exist but access denied. + if err == 5: # ERROR_ACCESS_DENIED + return True + return False + exit_code = wintypes.DWORD(0) + ctypes.windll.kernel32.GetExitCodeProcess( + handle, ctypes.byref(exit_code) + ) + ctypes.windll.kernel32.CloseHandle(handle) + return exit_code.value == STILL_ACTIVE def is_alive(self): if self._parent_alive: - self._parent_alive = os.getppid() == self._parent_pid + if sys.platform == 'win32': + self._parent_alive = self._parent_alive_win32() + else: + self._parent_alive = os.getppid() == self._parent_pid return self._parent_alive @@ -290,8 +352,17 @@ def _worker_loop( use_shared_memory, base_seed, shm_cache_size=0, + parent_pid=None, ): try: + import faulthandler + + faulthandler.enable() + # Register ForkingPickler handlers so DenseTensor can be serialized + from paddle.incubate.multiprocessing import init_reductions + + init_reductions() + # NOTE: [ mmap files clear ] When the child process exits unexpectedly, # some shared memory objects may have been applied for but have not yet # been put into the inter-process Queue. This part of the object needs @@ -334,8 +405,13 @@ def _worker_loop( except: init_exception = _WorkerException(worker_id) + def numpy2lodtensor(arr): + lodtensor = core.DenseTensor() + lodtensor.set(arr, core.CPUPlace()) + return lodtensor + iterator_drained = False - parent_watch_dog = ParentWatchDog() + parent_watch_dog = ParentWatchDog(parent_pid) while parent_watch_dog.is_alive(): try: @@ -343,36 +419,36 @@ def _worker_loop( except queue.Empty: continue - if isinstance(data, _ResumeIteration): - out_queue.put((data, None, None)) - iterator_drained = False - fetcher = _DatasetKind.create_fetcher( - dataset_kind, dataset, auto_collate_batch, collate_fn, True - ) - continue - - # None as poison piil, so worker event should be set - if data is None: - assert done_event.is_set() or iterator_drained, ( - "get None when worker done_event set" - ) - break - # If worker done event is set but get still get data in - # indices_queue, remaining data should be get and skipped. - if done_event.is_set() or iterator_drained: - continue - - idx, indices = data try: + idx = None + if isinstance(data, _ResumeIteration): + out_queue.put((data, None, None)) + iterator_drained = False + fetcher = _DatasetKind.create_fetcher( + dataset_kind, + dataset, + auto_collate_batch, + collate_fn, + True, + ) + continue + + # None as poison piil, so worker event should be set + if data is None: + assert done_event.is_set() or iterator_drained, ( + "get None when worker done_event set" + ) + break + # If worker done event is set but get still get data in + # indices_queue, remaining data should be get and skipped. + if done_event.is_set() or iterator_drained: + continue + + idx, indices = data if init_exception is not None: batch = init_exception init_exception = None else: - # NOTE: GPU tensor operation is not supported in sub-process - # but default device is GPU in paddle-gpu version, which - # may copy CPU tensor to GPU even if users want to use - # CPU tensor operation, so we add CPUPlace guard here - # to make sure tensor will be operated only on CPU with paddle.base.dygraph.guard(place=paddle.CPUPlace()): batch = fetcher.fetch(indices) except Exception as e: @@ -389,12 +465,6 @@ def _worker_loop( out_queue.put((idx, batch, None)) batch, structure = _flatten_batch(batch) if use_shared_memory: - - def numpy2lodtensor(arr): - lodtensor = core.DenseTensor() - lodtensor.set(arr, core.CPUPlace()) - return lodtensor - tensor_list = [ ( numpy2lodtensor(b) @@ -410,10 +480,18 @@ def numpy2lodtensor(arr): # NOTE: Main process will raise KeyboardInterrupt anyways, ignore it in child process pass except: + # Write to stderr (visible in parent console) + sys.stderr.write( + f"[WORKER pid={os.getpid()}] UNCAUGHT EXCEPTION\n" + f"{traceback.format_exc()}\n" + ) + sys.stderr.flush() raise finally: if use_shared_memory: _cleanup_mmap() - if done_event.is_set(): - out_queue.cancel_join_thread() - out_queue.close() + # On Windows (spawn), skip queue cleanup that may fail during shutdown + if not sys.platform.startswith('win'): + if done_event.is_set(): + out_queue.cancel_join_thread() + out_queue.close() diff --git a/python/paddle/io/multiprocess_utils.py b/python/paddle/io/multiprocess_utils.py index 6870832b408b0d..236fd3a98db1ce 100644 --- a/python/paddle/io/multiprocess_utils.py +++ b/python/paddle/io/multiprocess_utils.py @@ -44,13 +44,19 @@ def _cleanup(): # NOTE: inter-process Queue shared memory objects clear function _clear_multiprocess_queue_set() # NOTE: main process memory map files clear function - core._cleanup_mmap_fds() + try: + core._cleanup_mmap_fds() + except Exception: + pass # NOTE: for child process clear function at exit def _cleanup_mmap(): # clear memory map files in child process - core._cleanup_mmap_fds() + try: + core._cleanup_mmap_fds() + except Exception: + pass # NOTE used for register a function to be executed at interpreter exit. @@ -109,9 +115,7 @@ def _signal_register(signals): # BlockingQueue) may not be completely released, resulting in the corresponding # memory-mapped file remaining on the disk (/dev/shm), so register this function # to clean up shared memory objects in these two queues before the python interpreter exits. -# NOTE: Currently multi-process DataLoader only supports Linux platform -if not (sys.platform == 'darwin' or sys.platform == 'win32'): - CleanupFuncRegistrar.register(_cleanup) +CleanupFuncRegistrar.register(_cleanup) # ------------ SIGCHLD handler setting -------------- _SIGCHLD_handler_set = False @@ -121,6 +125,11 @@ def _set_SIGCHLD_handler(): global _SIGCHLD_handler_set if _SIGCHLD_handler_set: return + if sys.platform == 'win32': + # Windows does not have SIGCHLD; worker failure is detected + # via periodic polling in ThrowErrorIfLoadProcessFailed. + _SIGCHLD_handler_set = True + return current_handler = signal.getsignal(signal.SIGCHLD) if not callable(current_handler): diff --git a/python/paddle/io/reader.py b/python/paddle/io/reader.py index 25daa0fa317a06..0d01542f9b9ab9 100644 --- a/python/paddle/io/reader.py +++ b/python/paddle/io/reader.py @@ -497,11 +497,9 @@ def __init__( self.places = _convert_places(places) assert num_workers >= 0, "num_workers should be a non-negative value" - if num_workers > 0 and ( - sys.platform == 'darwin' or sys.platform == 'win32' - ): + if num_workers > 0 and sys.platform == 'darwin': warnings.warn( - "DataLoader with multi-process mode is not supported on MacOs and Windows currently." + "DataLoader with multi-process mode is not supported on MacOs currently." " Please use single-process mode with num_workers = 0 instead" ) num_workers = 0 diff --git a/test/legacy_test/test_dataloader_autotune.py b/test/legacy_test/test_dataloader_autotune.py index e6ff899eeb7745..427da807378feb 100755 --- a/test/legacy_test/test_dataloader_autotune.py +++ b/test/legacy_test/test_dataloader_autotune.py @@ -76,7 +76,7 @@ def test_dataloader_disable_autotune(self): loader = DataLoader( self.dataset, batch_size=self.batch_size, num_workers=2 ) - if sys.platform == 'darwin' or sys.platform == 'win32': + if sys.platform == 'darwin': self.assertEqual(loader.num_workers, 0) else: self.assertEqual(loader.num_workers, 2) diff --git a/test/legacy_test/test_dataloader_multiprocess.py b/test/legacy_test/test_dataloader_multiprocess.py new file mode 100644 index 00000000000000..6170992e5da473 --- /dev/null +++ b/test/legacy_test/test_dataloader_multiprocess.py @@ -0,0 +1,181 @@ +# Copyright (c) 2026 PaddlePaddle Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import unittest + +import numpy as np + +import paddle +from paddle.io import DataLoader, Dataset, IterableDataset + + +class RandomDataset(Dataset): + def __init__(self, sample_num, feature_dim=10): + self.sample_num = sample_num + self.feature_dim = feature_dim + self.data = np.random.rand(sample_num, feature_dim).astype('float32') + self.label = np.random.randint(0, 2, size=(sample_num, 1)).astype( + 'int64' + ) + + def __getitem__(self, idx): + return ( + paddle.to_tensor(self.data[idx]), + paddle.to_tensor(self.label[idx]), + ) + + def __len__(self): + return self.sample_num + + +class RandomIterableDataset(IterableDataset): + def __init__(self, sample_num, feature_dim=10): + self.sample_num = sample_num + self.feature_dim = feature_dim + + def __iter__(self): + for i in range(self.sample_num): + yield paddle.rand([self.feature_dim]) + + +def collate_batch(batch): + data = paddle.stack([item[0] for item in batch]) + label = paddle.stack([item[1] for item in batch]) + return data, label + + +class TestDataLoaderWindowsMultiprocess(unittest.TestCase): + def setUp(self): + self.sample_num = 100 + self.batch_size = 8 + self.feature_dim = 10 + self.epoch_num = 3 + + def run_simple_net(self, num_workers, use_shared_memory=False): + paddle.seed(2026) + np.random.seed(2026) + + dataset = RandomDataset(self.sample_num, self.feature_dim) + loader = DataLoader( + dataset, + places=paddle.CPUPlace(), + batch_size=self.batch_size, + shuffle=True, + num_workers=num_workers, + use_shared_memory=use_shared_memory, + ) + + losses = [] + for epoch in range(self.epoch_num): + for data, label in loader(): + # Simple linear layer + pred = paddle.mean(data, axis=1, keepdim=True) + loss = paddle.nn.functional.binary_cross_entropy_with_logits( + pred, paddle.cast(label, 'float32') + ) + losses.append(loss.numpy()) + return np.mean(losses) + + def test_multiprocess_singleprocess_loss_close(self): + """ + Test that multi-process (num_workers=2) and single-process (num_workers=0) + produce similar loss values. + """ + loss_single = self.run_simple_net(num_workers=0) + loss_multi = self.run_simple_net(num_workers=2) + diff = np.abs(loss_single - loss_multi) / np.abs(loss_single) + self.assertLess( + diff, + 1e-2, + f"Loss difference too large: single={loss_single}, multi={loss_multi}, diff={diff}", + ) + + def test_multiprocess_with_shared_memory(self): + """ + Test multi-process DataLoader with use_shared_memory=True. + """ + loss_single = self.run_simple_net(num_workers=0) + loss_multi = self.run_simple_net(num_workers=2, use_shared_memory=True) + diff = np.abs(loss_single - loss_multi) / np.abs(loss_single) + self.assertLess( + diff, + 1e-2, + f"Loss difference too large with shared memory: single={loss_single}, multi={loss_multi}, diff={diff}", + ) + + def test_multiprocess_more_workers(self): + """ + Test with num_workers=4 to stress-test the worker pool. + """ + loss_single = self.run_simple_net(num_workers=0) + loss_multi = self.run_simple_net(num_workers=4) + diff = np.abs(loss_single - loss_multi) / np.abs(loss_single) + self.assertLess( + diff, + 1e-2, + f"Loss difference too large (4 workers): single={loss_single}, multi={loss_multi}, diff={diff}", + ) + + def test_multiprocess_persistent_workers(self): + """ + Test with persistent_workers=True to ensure worker reuse works. + """ + paddle.seed(2026) + np.random.seed(2026) + dataset = RandomDataset(self.sample_num, self.feature_dim) + loader = DataLoader( + dataset, + places=paddle.CPUPlace(), + batch_size=self.batch_size, + shuffle=True, + num_workers=2, + use_shared_memory=False, + persistent_workers=True, + ) + losses = [] + for epoch in range(self.epoch_num): + for data, label in loader(): + pred = paddle.mean(data, axis=1, keepdim=True) + loss = paddle.nn.functional.binary_cross_entropy_with_logits( + pred, paddle.cast(label, 'float32') + ) + losses.append(loss.numpy()) + loss_multi = np.mean(losses) + + loss_single = self.run_simple_net(num_workers=0) + diff = np.abs(loss_single - loss_multi) / np.abs(loss_single) + self.assertLess( + diff, + 1e-2, + f"Loss difference too large (persistent workers): single={loss_single}, multi={loss_multi}, diff={diff}", + ) + + def test_multiprocess_iterable_dataset(self): + """ + Test multi-process DataLoader with IterableDataset. + """ + dataset = RandomIterableDataset(50, self.feature_dim) + loader = DataLoader( + dataset, + batch_size=self.batch_size, + num_workers=2, + ) + count = 0 + for batch in loader(): + count += 1 + self.assertGreater(count, 0) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/legacy_test/test_modelaverage.py b/test/legacy_test/test_modelaverage.py index 06f9a1b51ee517..f4148f75a9ce35 100644 --- a/test/legacy_test/test_modelaverage.py +++ b/test/legacy_test/test_modelaverage.py @@ -29,6 +29,21 @@ def get_value_by_name(name, ops): return value +class RandomDataset(paddle.io.Dataset): + def __init__(self, num_samples, image_size=784, class_num=10): + self.num_samples = num_samples + self.image_size = image_size + self.class_num = class_num + + def __getitem__(self, idx): + image = np.random.random([self.image_size]).astype('float32') + label = np.random.randint(0, self.class_num - 1, (1,)).astype('int64') + return image, label + + def __len__(self): + return self.num_samples + + class TestModelAverage(unittest.TestCase): def test_model_average_static(self): paddle.enable_static() @@ -146,21 +161,6 @@ def test_model_average_dygraph(self): IMAGE_SIZE = 784 CLASS_NUM = 10 - # define a random dataset - class RandomDataset(paddle.io.Dataset): - def __init__(self, num_samples): - self.num_samples = num_samples - - def __getitem__(self, idx): - image = np.random.random([IMAGE_SIZE]).astype('float32') - label = np.random.randint(0, CLASS_NUM - 1, (1,)).astype( - 'int64' - ) - return image, label - - def __len__(self): - return self.num_samples - class LinearNet(nn.Layer): def __init__(self): super().__init__() diff --git a/test/legacy_test/test_multiprocess_dataloader_iterable_dataset_split.py b/test/legacy_test/test_multiprocess_dataloader_iterable_dataset_split.py index a8fe137523058b..a148cb1e951e51 100644 --- a/test/legacy_test/test_multiprocess_dataloader_iterable_dataset_split.py +++ b/test/legacy_test/test_multiprocess_dataloader_iterable_dataset_split.py @@ -75,26 +75,27 @@ def __iter__(self): yield np.array([i]) +def worker_splitter(worker_id): + worker_info = get_worker_info() + + dataset = worker_info.dataset + start = dataset.start + end = dataset.end + num_per_worker = int( + math.ceil((end - start) / float(worker_info.num_workers)) + ) + + worker_id = worker_info.id + dataset.start = start + worker_id * num_per_worker + dataset.end = min(dataset.start + num_per_worker, end) + + class TestDynamicDataLoaderIterInitFuncSplit(unittest.TestCase): def test_main(self): place = base.CPUPlace() with base.dygraph.guard(place): dataset = RangeIterableDataset(0, 10) - def worker_splitter(worker_id): - worker_info = get_worker_info() - - dataset = worker_info.dataset - start = dataset.start - end = dataset.end - num_per_worker = int( - math.ceil((end - start) / float(worker_info.num_workers)) - ) - - worker_id = worker_info.id - dataset.start = start + worker_id * num_per_worker - dataset.end = min(dataset.start + num_per_worker, end) - dataloader = DataLoader( dataset, places=place,