[tune] [rllib] Centralized driver logging (#1208)

* logger v2

* add logger

* lint

* todo

* viskit works now

* doc

* remove none check

* fix timeout

* Missing Numpy for Sigmoid data

python/ray/tune/trial.py

@@ -2,12 +2,14 @@ from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function
 
+import tempfile
 import traceback
 import ray
 import os
 
 from collections import namedtuple
 from ray.rllib.agent import get_agent_class
+from ray.tune.logger import NoopLogger, UnifiedLogger
 
 
 class Resources(
@@ -92,6 +94,8 @@ class Trial(object):
         self.agent = None
         self.status = Trial.PENDING
         self.location = None
+        self.logdir = None
+        self.result_logger = None
 
     def start(self):
         """Starts this trial.
@@ -104,7 +108,7 @@ class Trial(object):
         if self._checkpoint_path:
             self.restore_from_path(path=self._checkpoint_path)
 
-    def stop(self, error=False):
+    def stop(self, error=False, stop_logger=True):
         """Stops this trial.
 
         Stops this trial, releasing all allocating resources. If stopping the
@@ -126,10 +130,11 @@ class Trial(object):
                 stop_tasks.append(self.agent.stop.remote())
                 stop_tasks.append(self.agent.__ray_terminate__.remote(
                     self.agent._ray_actor_id.id()))
+                # TODO(ekl)  seems like wait hangs when killing actors
                 _, unfinished = ray.wait(
-                        stop_tasks, num_returns=2, timeout=10000)
+                        stop_tasks, num_returns=2, timeout=250)
                 if unfinished:
-                    print(("Stopping %s Actor was unsuccessful, "
+                    print(("Stopping %s Actor timed out, "
                            "but moving on...") % self)
         except Exception:
             print("Error stopping agent:", traceback.format_exc())
@@ -137,6 +142,10 @@ class Trial(object):
         finally:
             self.agent = None
 
+        if stop_logger and self.result_logger:
+            self.result_logger.close()
+            self.result_logger = None
+
     def pause(self):
         """We want to release resources (specifically GPUs) when pausing an
         experiment. This results in a state similar to TERMINATED."""
@@ -144,7 +153,7 @@ class Trial(object):
         assert self.status == Trial.RUNNING, self.status
         try:
             self.checkpoint()
-            self.stop()
+            self.stop(stop_logger=False)
             self.status = Trial.PAUSED
         except Exception:
             print("Error pausing agent:", traceback.format_exc())
@@ -250,9 +259,19 @@ class Trial(object):
         cls = ray.remote(
             num_cpus=self.resources.driver_cpu_limit,
             num_gpus=self.resources.driver_gpu_limit)(agent_cls)
+        if not self.result_logger:
+            if not os.path.exists(self.local_dir):
+                os.makedirs(self.local_dir)
+            self.logdir = tempfile.mkdtemp(
+                prefix=str(self), dir=self.local_dir)
+            self.result_logger = UnifiedLogger(
+                self.config, self.logdir, self.upload_dir)
+        remote_logdir = self.logdir
+        # Logging for trials is handled centrally by TrialRunner, so
+        # configure the remote agent to use a noop-logger.
         self.agent = cls.remote(
-            self.env_creator, self.config, self.local_dir, self.upload_dir,
-            experiment_tag=self.experiment_tag)
+            self.env_creator, self.config,
+            lambda config: NoopLogger(config, remote_logdir))
 
     def __str__(self):
         identifier = '{}_{}'.format(self.alg, self.env_name)