Stream logs to driver by default.

python/ray/experimental/state.py

@@ -401,34 +401,6 @@ def client_table(self):
 
         return parse_client_table(self.redis_client)
 
-    def log_files(self):
-        """Fetch and return a dictionary of log file names to outputs.
-
-        Returns:
-            IP address to log file name to log file contents mappings.
-        """
-        relevant_files = self.redis_client.keys("LOGFILE*")
-
-        ip_filename_file = {}
-
-        for filename in relevant_files:
-            filename = decode(filename)
-            filename_components = filename.split(":")
-            ip_addr = filename_components[1]
-
-            file = self.redis_client.lrange(filename, 0, -1)
-            file_str = []
-            for x in file:
-                y = decode(x)
-                file_str.append(y)
-
-            if ip_addr not in ip_filename_file:
-                ip_filename_file[ip_addr] = {}
-
-            ip_filename_file[ip_addr][filename] = file_str
-
-        return ip_filename_file
-
     def _profile_table(self, batch_id):
         """Get the profile events for a given batch of profile events.
 
@@ -869,6 +841,10 @@ def available_resources(self):
             for resource_id, num_available in available_resources.items():
                 total_available_resources[resource_id] += num_available
 
+        # Close the pubsub clients to avoid leaking file descriptors.
+        for subscribe_client in subscribe_clients:
+            subscribe_client.close()
+
         return dict(total_available_resources)
 
     def _error_messages(self, job_id):

python/ray/gcs_utils.py

@@ -27,6 +27,7 @@
 ]
 
 FUNCTION_PREFIX = "RemoteFunction:"
+LOG_FILE_CHANNEL = "RAY_LOG_CHANNEL"
 
 # xray heartbeats
 XRAY_HEARTBEAT_CHANNEL = str(TablePubsub.HEARTBEAT).encode("ascii")

python/ray/import_thread.py

@@ -5,8 +5,6 @@
 import threading
 import traceback
 
-import redis
-
 import ray
 from ray import ray_constants
 from ray import cloudpickle as pickle
@@ -17,29 +15,35 @@
 class ImportThread(object):
     """A thread used to import exports from the driver or other workers.
 
-    Note:
-    The driver also has an import thread, which is used only to
-    import custom class definitions from calls to register_custom_serializer
-    that happen under the hood on workers.
+    Note: The driver also has an import thread, which is used only to import
+    custom class definitions from calls to register_custom_serializer that
+    happen under the hood on workers.
 
     Attributes:
         worker: the worker object in this process.
         mode: worker mode
         redis_client: the redis client used to query exports.
+        threads_stopped (threading.Event): A threading event used to signal to
+            the thread that it should exit.
     """
 
-    def __init__(self, worker, mode):
+    def __init__(self, worker, mode, threads_stopped):
         self.worker = worker
         self.mode = mode
         self.redis_client = worker.redis_client
+        self.threads_stopped = threads_stopped
 
     def start(self):
         """Start the import thread."""
-        t = threading.Thread(target=self._run, name="ray_import_thread")
+        self.t = threading.Thread(target=self._run, name="ray_import_thread")
         # Making the thread a daemon causes it to exit
         # when the main thread exits.
-        t.daemon = True
-        t.start()
+        self.t.daemon = True
+        self.t.start()
+
+    def join_import_thread(self):
+        """Wait for the thread to exit."""
+        self.t.join()
 
     def _run(self):
         import_pubsub_client = self.redis_client.pubsub()
@@ -50,14 +54,24 @@ def _run(self):
         # Keep track of the number of imports that we've imported.
         num_imported = 0
 
-        # Get the exports that occurred before the call to subscribe.
-        with self.worker.lock:
-            export_keys = self.redis_client.lrange("Exports", 0, -1)
-            for key in export_keys:
-                num_imported += 1
-                self._process_key(key)
         try:
-            for msg in import_pubsub_client.listen():
+            # Get the exports that occurred before the call to subscribe.
+            with self.worker.lock:
+                export_keys = self.redis_client.lrange("Exports", 0, -1)
+                for key in export_keys:
+                    num_imported += 1
+                    self._process_key(key)
+
+            while True:
+                # Exit if we received a signal that we should stop.
+                if self.threads_stopped.is_set():
+                    return
+
+                msg = import_pubsub_client.get_message()
+                if msg is None:
+                    self.threads_stopped.wait(timeout=0.01)
+                    continue
+
                 with self.worker.lock:
                     if msg["type"] == "subscribe":
                         continue
@@ -68,10 +82,9 @@ def _run(self):
                         num_imported += 1
                         key = self.redis_client.lindex("Exports", i)
                         self._process_key(key)
-        except redis.ConnectionError:
-            # When Redis terminates the listen call will throw a
-            # ConnectionError, which we catch here.
-            pass
+        finally:
+            # Close the pubsub client to avoid leaking file descriptors.
+            import_pubsub_client.close()
 
     def _process_key(self, key):
         """Process the given export key from redis."""