Don't setpgid() on actors (#3347)

python/ray/actor.py

@@ -6,8 +6,7 @@ import copy
 import hashlib
 import inspect
 import logging
-import os
-import signal
+import sys
 import traceback
 
 import ray.cloudpickle as pickle
@@ -783,9 +782,7 @@ def make_actor(cls, num_cpus, num_gpus, resources, actor_method_cpus,
                 # this is so that when the worker kills itself below, the local
                 # scheduler won't push an error message to the driver.
                 worker.local_scheduler_client.disconnect()
-                # Kill the process group. We will get the SIGTERM and
-                # eventually call sys.exit(0) in worker.py as a result of this.
-                os.killpg(os.getpgid(os.getpid()), signal.SIGTERM)
+                sys.exit(0)
                 assert False, "This process should have terminated."
 
         def __ray_save_checkpoint__(self):

python/ray/rllib/test/test_env_with_subprocess.py

@@ -29,12 +29,13 @@ class EnvWithSubprocess(gym.Env):
         self.action_space = Discrete(2)
         self.observation_space = Discrete(2)
         # Subprocess that should be cleaned up
-        self.subproc = subprocess.Popen(UNIQUE_CMD, shell=True)
+        self.subproc = subprocess.Popen(UNIQUE_CMD.split(" "), shell=False)
         # Exit handler should be called
         if config.worker_index == 0:
             atexit.register(lambda: os.unlink(UNIQUE_FILE_0))
         else:
             atexit.register(lambda: os.unlink(UNIQUE_FILE_1))
+        atexit.register(lambda: self.subproc.kill())
 
     def reset(self):
         return [0]

python/ray/worker.py

@@ -905,10 +905,6 @@ class Worker(object):
         self.actor_id = task.actor_creation_id().id()
         class_id = arguments[0]
 
-        # Set the process group id. This ensures that child processes spawned
-        # by this actor can be killed with this actor easily on termination.
-        os.setpgid(os.getpid(), os.getpid())
-
         key = b"ActorClass:" + class_id
 
         # Wait for the actor class key to have been imported by the import
@@ -976,7 +972,7 @@ class Worker(object):
             driver_id, function_id.id()) == execution_info.max_calls)
         if reached_max_executions:
             self.local_scheduler_client.disconnect()
-            os._exit(0)
+            sys.exit(0)
 
     def _get_next_task_from_local_scheduler(self):
         """Get the next task from the local scheduler.