Fix excessive buffering of worker stdout/stderr. (#4094)

* Start workers with 'python -u' to prevent buffering of prints.

* Set sys.stdout and sys.stderr.

* Add comment.

python/ray/node.py

@@ -223,7 +223,7 @@ class Node(object):
             suffix=".out", prefix=name, directory_name=self._logs_dir)
         log_stderr = self._make_inc_temp(
             suffix=".err", prefix=name, directory_name=self._logs_dir)
-        # Line-buffer the output (mode 1)
+        # Line-buffer the output (mode 1).
         log_stdout_file = open(log_stdout, "a", buffering=1)
         log_stderr_file = open(log_stderr, "a", buffering=1)
         return log_stdout_file, log_stderr_file

python/ray/worker.py

@@ -1877,6 +1877,12 @@ def connect(info,
             # be redirected.
             os.dup2(log_stdout_file.fileno(), sys.stdout.fileno())
             os.dup2(log_stderr_file.fileno(), sys.stderr.fileno())
+            # We also manually set sys.stdout and sys.stderr because that seems
+            # to have an affect on the output buffering. Without doing this,
+            # stdout and stderr are heavily buffered resulting in seemingly
+            # lost logging statements.
+            sys.stdout = log_stdout_file
+            sys.stderr = log_stderr_file
             # This should always be the first message to appear in the worker's
             # stdout and stderr log files. The string "Ray worker pid:" is
             # parsed in the log monitor process.

test/runtest.py

@@ -2523,11 +2523,11 @@ def test_logging_to_driver(shutdown_only):
 
     @ray.remote
     def f():
+        # It's important to make sure that these print statements occur even
+        # without calling sys.stdout.flush() and sys.stderr.flush().
         for i in range(100):
             print(i)
             print(100 + i, file=sys.stderr)
-            sys.stdout.flush()
-            sys.stderr.flush()
 
     captured = {}
     with CaptureOutputAndError(captured):