[Object Spilling] Name IO workers properly. (#11097)

* Fix.

* Addressed code review.

python/ray/ray_constants.py

@@ -184,6 +184,9 @@ PROCESS_TYPE_REDIS_SERVER = "redis_server"
 PROCESS_TYPE_WEB_UI = "web_ui"
 PROCESS_TYPE_GCS_SERVER = "gcs_server"
 
+WORKER_PROCESS_TYPE_IDLE_WORKER = "ray::IDLE"
+WORKER_PROCESS_TYPE_IO_WORKER = "ray::IOWorker"
+
 LOG_MONITOR_MAX_OPEN_FILES = 200
 
 # A constant used as object metadata to indicate the object is cross language.

python/ray/tests/test_object_spilling.py

@@ -4,6 +4,7 @@ import time
 
 import numpy as np
 import pytest
+import psutil
 import ray
 
 
@@ -39,6 +40,16 @@ def test_spill_objects_manually(shutdown_only):
                 ray.experimental.force_spill_objects([ref_to_spill])
                 spilled_objects.add(ref_to_spill)
 
+    def is_worker(cmdline):
+        return cmdline and cmdline[0].startswith("ray::")
+
+    # Make sure io workers are spawned with proper name.
+    processes = [
+        x.cmdline()[0] for x in psutil.process_iter(attrs=["cmdline"])
+        if is_worker(x.info["cmdline"])
+    ]
+    assert ray.ray_constants.WORKER_PROCESS_TYPE_IO_WORKER in processes
+
     # Spill 2 more objects so we will always have enough space for
     # restoring objects back.
     refs_to_spill = (pinned_objects.pop(), pinned_objects.pop())

python/ray/worker.py

@@ -1164,7 +1164,10 @@ def connect(node,
             job_id).binary()
 
     if mode is not SCRIPT_MODE and setproctitle:
-        setproctitle.setproctitle("ray::IDLE")
+        process_name = ray_constants.WORKER_PROCESS_TYPE_IDLE_WORKER
+        if mode is IO_WORKER_MODE:
+            process_name = ray_constants.WORKER_PROCESS_TYPE_IO_WORKER
+        setproctitle.setproctitle(process_name)
 
     if not isinstance(job_id, JobID):
         raise TypeError("The type of given job id must be JobID.")