[rllib] Try to call close on envs on stop (#4521)

python/ray/rllib/agents/agent.py

@@ -148,7 +148,7 @@ COMMON_CONFIG = {
     # If using num_envs_per_worker > 1, whether to create those new envs in
     # remote processes instead of in the same worker. This adds overheads, but
     # can make sense if your envs can take much time to step / reset
-    # (e.g., for StarCraft)
+    # (e.g., for StarCraft). Use this cautiously; overheads are significant.
     "remote_worker_envs": False,
     # Timeout that remote workers are waiting when polling environments.
     # 0 (continue when at least one env is ready) is a reasonable default,

python/ray/rllib/env/base_env.py

@@ -186,7 +186,10 @@ class BaseEnv(object):
     @PublicAPI
     def stop(self):
         """Releases all resources used."""
-        pass
+
+        for env in self.get_unwrapped():
+            if hasattr(env, "close"):
+                env.close()
 
 
 # Fixed agent identifier when there is only the single agent in the env

python/ray/rllib/tests/test_env_with_subprocess.py

@@ -20,6 +20,8 @@ from ray.tune.registry import register_env
 UNIQUE_CMD = "sleep {}".format(str(time.time()))
 _, UNIQUE_FILE_0 = tempfile.mkstemp("test_env_with_subprocess")
 _, UNIQUE_FILE_1 = tempfile.mkstemp("test_env_with_subprocess")
+_, UNIQUE_FILE_2 = tempfile.mkstemp("test_env_with_subprocess")
+_, UNIQUE_FILE_3 = tempfile.mkstemp("test_env_with_subprocess")
 
 
 class EnvWithSubprocess(gym.Env):
@@ -30,6 +32,7 @@ class EnvWithSubprocess(gym.Env):
         self.observation_space = Discrete(2)
         # Subprocess that should be cleaned up
         self.subproc = subprocess.Popen(UNIQUE_CMD.split(" "), shell=False)
+        self.config = config
         # Exit handler should be called
         if config.worker_index == 0:
             atexit.register(lambda: os.unlink(UNIQUE_FILE_0))
@@ -37,6 +40,12 @@ class EnvWithSubprocess(gym.Env):
             atexit.register(lambda: os.unlink(UNIQUE_FILE_1))
         atexit.register(lambda: self.subproc.kill())
 
+    def close(self):
+        if self.config.worker_index == 0:
+            os.unlink(UNIQUE_FILE_2)
+        else:
+            os.unlink(UNIQUE_FILE_3)
+
     def reset(self):
         return 0
 
@@ -75,4 +84,6 @@ if __name__ == "__main__":
     assert not leaked, "LEAKED PROCESSES: {}".format(leaked)
     assert not os.path.exists(UNIQUE_FILE_0), "atexit handler not called"
     assert not os.path.exists(UNIQUE_FILE_1), "atexit handler not called"
+    assert not os.path.exists(UNIQUE_FILE_2), "close not called"
+    assert not os.path.exists(UNIQUE_FILE_3), "close not called"
     print("OK")