[Environment Adaptation] Support Windows DataLoader multiprocess (num_workers > 0)

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

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 <sys/wait.h>
-#include <unistd.h>
 #include <cstdlib>
 
 #include <csignal>
@@ -26,6 +22,14 @@
 #include "paddle/fluid/platform/enforce.h"
 #include "paddle/phi/core/memory/allocation/mmap_allocator.h"
 
+#ifdef _WIN32
+#include <processthreadsapi.h>
+#include <tlhelp32.h>
+#else
+#include <sys/wait.h>
+#include <unistd.h>
+#endif
+
 namespace paddle::imperative {
 
 static std::map<int64_t, std::set<pid_t>> 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<pid_t> *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<DWORD>(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<pid_t> *pids_set = nullptr;
   pid_t process_pid = 0;
@@ -190,8 +268,7 @@ void ThrowErrorIfLoadProcessFailed() {
       }
     }
   }
+#endif
 }
 
 }  // namespace paddle::imperative
-
-#endif

paddle/fluid/imperative/data_loader.h

@@ -14,9 +14,15 @@
 
 #pragma once
 
-#ifndef _WIN32
-
+#ifdef _WIN32
+#include <Windows.h>
+// 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 <unistd.h>
+#endif
 
 #include <cstdint>
 #include <set>
@@ -31,5 +37,3 @@ extern void ThrowErrorIfLoadProcessFailed();
 
 }  // namespace imperative
 }  // namespace paddle
-
-#endif

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)

paddle/fluid/pybind/imperative.cc

@@ -478,7 +478,6 @@ static void VarBaseCopy(std::shared_ptr<imperative::VarBase> &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(); });

paddle/fluid/pybind/tensor.cc

@@ -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<memory::allocation::RefcountedMemoryMapAllocation *>(
+            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
+      }
+      int 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<int>(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<std::string>();
+    const int shared_fd = t[1].cast<int>();
+    const bool use_file_descriptor = t[6].cast<bool>();
+
+    size_t size = t[2].cast<size_t>();
+    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<DataType>(t[3].cast<int>()));
+    tensor.Resize(common::make_ddim(t[4].cast<std::vector<int>>()));
+
+    return tensor;
+  });
+
+  m.def("_shared_incref", [](DenseTensor &self) {
+    auto *mmap_allocation =
+        dynamic_cast<memory::allocation::RefcountedMemoryMapAllocation *>(
+            self.Holder().get());
+    if (mmap_allocation) {
+      mmap_allocation->incref();
+    }
+  });
+
+  m.def("_shared_decref", [](DenseTensor &self) {
+    auto *mmap_allocation =
+        dynamic_cast<memory::allocation::RefcountedMemoryMapAllocation *>(
+            self.Holder().get());
+    if (mmap_allocation) {
+      mmap_allocation->decref();
+    }
+  });
 }
 
 }  // namespace paddle::pybind

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)