Windows bug fixes (#7740)

python/ray/ray_process_reaper.py

@@ -1,3 +1,4 @@
+import atexit
 import os
 import signal
 import sys
@@ -23,13 +24,23 @@ def reap_process_group(*args):
         # Give a one-second grace period for other processes to clean up.
         time.sleep(SIGTERM_GRACE_PERIOD_SECONDS)
         # SIGKILL the pgroup (including ourselves) as a last-resort.
-        os.killpg(0, signal.SIGKILL)
+        if sys.platform == "win32":
+            atexit.unregister(sigterm_handler)
+            os.kill(0, signal.CTRL_BREAK_EVENT)
+        else:
+            os.killpg(0, signal.SIGKILL)
 
     # Set a SIGTERM handler to handle SIGTERMing ourselves with the group.
-    signal.signal(signal.SIGTERM, sigterm_handler)
+    if sys.platform == "win32":
+        atexit.register(sigterm_handler)
+    else:
+        signal.signal(signal.SIGTERM, sigterm_handler)
 
     # Our parent must have died, SIGTERM the group (including ourselves).
-    os.killpg(0, signal.SIGTERM)
+    if sys.platform == "win32":
+        os.kill(0, signal.CTRL_C_EVENT)
+    else:
+        os.killpg(0, signal.SIGTERM)
 
 
 def main():

python/ray/reporter.py

@@ -6,6 +6,7 @@ import traceback
 import time
 import datetime
 import grpc
+import socket
 import subprocess
 import sys
 from concurrent import futures
@@ -91,7 +92,7 @@ class Reporter:
         """Initialize the reporter object."""
         self.cpu_counts = (psutil.cpu_count(), psutil.cpu_count(logical=False))
         self.ip = ray.services.get_node_ip_address()
-        self.hostname = os.uname().nodename
+        self.hostname = socket.gethostname()
 
         _ = psutil.cpu_percent()  # For initialization
 
@@ -149,7 +150,11 @@ class Reporter:
         ]
 
     def get_load_avg(self):
-        load = os.getloadavg()
+        if sys.platform == "win32":
+            cpu_percent = psutil.cpu_percent()
+            load = (cpu_percent, cpu_percent, cpu_percent)
+        else:
+            load = os.getloadavg()
         per_cpu_load = tuple((round(x / self.cpu_counts[0], 2) for x in load))
         return load, per_cpu_load
 

python/ray/services.py

@@ -633,9 +633,10 @@ def start_reaper(fate_share=None):
     # up other ray processes without killing the process group of the
     # process that started us.
     try:
-        os.setpgrp()
-    except (AttributeError, OSError) as e:
-        errcode = e.errno if isinstance(e, OSError) else None
+        if sys.platform != "win32":
+            os.setpgrp()
+    except OSError as e:
+        errcode = e.errno
         if errcode == errno.EPERM and os.getpgrp() == os.getpid():
             # Nothing to do; we're already a session leader.
             pass

python/ray/utils.py

@@ -517,6 +517,8 @@ def detect_fate_sharing_support_win32():
             kernel32.SetInformationJobObject.restype = BOOL
             kernel32.AssignProcessToJobObject.argtypes = (HANDLE, HANDLE)
             kernel32.AssignProcessToJobObject.restype = BOOL
+            kernel32.IsDebuggerPresent.argtypes = ()
+            kernel32.IsDebuggerPresent.restype = BOOL
         except (AttributeError, TypeError, ImportError):
             kernel32 = None
         job = kernel32.CreateJobObjectW(None, None) if kernel32 else None
@@ -558,6 +560,8 @@ def detect_fate_sharing_support_win32():
                     ("PeakJobMemoryUsed", ctypes.c_size_t),
                 ]
 
+            debug = kernel32.IsDebuggerPresent()
+
             # Defined in <WinNT.h>; also available here:
             # https://docs.microsoft.com/en-us/windows/win32/api/jobapi2/nf-jobapi2-setinformationjobobject
             JobObjectExtendedLimitInformation = 9
@@ -566,7 +570,7 @@ def detect_fate_sharing_support_win32():
             JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x00002000
             buf = JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
             buf.BasicLimitInformation.LimitFlags = (
-                JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
+                (0 if debug else JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE)
                 | JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION
                 | JOB_OBJECT_LIMIT_BREAKAWAY_OK)
             infoclass = JobObjectExtendedLimitInformation