diff --git a/python/ray/ray_process_reaper.py b/python/ray/ray_process_reaper.py index 79c754207..cccd1389e 100644 --- a/python/ray/ray_process_reaper.py +++ b/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(): diff --git a/python/ray/reporter.py b/python/ray/reporter.py index b78ac1784..d01cb7a2a 100644 --- a/python/ray/reporter.py +++ b/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 diff --git a/python/ray/services.py b/python/ray/services.py index eb51bdb0d..b3a4a28a2 100644 --- a/python/ray/services.py +++ b/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 diff --git a/python/ray/utils.py b/python/ray/utils.py index af1936276..73aecc46f 100644 --- a/python/ray/utils.py +++ b/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 ; 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 diff --git a/src/ray/gcs/gcs_client/service_based_gcs_client.cc b/src/ray/gcs/gcs_client/service_based_gcs_client.cc index edd4e353f..6b8f1eecb 100644 --- a/src/ray/gcs/gcs_client/service_based_gcs_client.cc +++ b/src/ray/gcs/gcs_client/service_based_gcs_client.cc @@ -77,7 +77,7 @@ void ServiceBasedGcsClient::GetGcsServerAddressFromRedis( redisReply *reply = nullptr; while (num_attempts < RayConfig::instance().gcs_service_connect_retries()) { reply = reinterpret_cast(redisCommand(context, "GET GcsServerAddress")); - if (reply->type != REDIS_REPLY_NIL) { + if (reply && reply->type != REDIS_REPLY_NIL) { break; } @@ -88,6 +88,7 @@ void ServiceBasedGcsClient::GetGcsServerAddressFromRedis( } RAY_CHECK(num_attempts < RayConfig::instance().gcs_service_connect_retries()) << "No entry found for GcsServerAddress"; + RAY_CHECK(reply) << "Redis did not reply to GcsServerAddress. Is redis running?"; RAY_CHECK(reply->type == REDIS_REPLY_STRING) << "Expected string, found Redis type " << reply->type << " for GcsServerAddress"; std::string result(reply->str);