Changed to start two threads (stdout and stderr) and removed polling logic per @andy31415 suggestion

Author: shaoltan-amazon

scripts/tests/linux/log_line_processing.py

@@ -14,10 +14,8 @@
 
 import logging
 import queue
-import select
 import subprocess
 import threading
-import time
 from typing import List
 
 
@@ -48,43 +46,48 @@ def __init__(self, command: List[str], output_path: str):
         self.output_path = output_path
         self.output_lines = queue.Queue()
         self.process = None
-        self.io_thread = None
+        self.stdout_thread = None
+        self.stderr_thread = None
+        self.lock = threading.Lock()
         self.done = False
 
-    def _io_thread(self):
-        """Reads process lines and writes them to an output file.
+    def _write_to_file(self, line: str, is_error_line=False):
+        """Writes a line to an output file in a thread-safe manner.
+
+        Opens file in append-text mode ('at') with line buffering 
+        for immediate output to ensure all writes are appended.
+        """
+        with self.lock:
+            with open(self.output_path, "at", buffering=1) as f:
+                if is_error_line:
+                    f.write(f"!!STDERR!! : {line}")
+                else:
+                    f.write(line)
+                f.flush()
+
+    def _stdout_thread(self):
+        """Reads stdout process lines and writes them to an output file.
 
         It also sends the output lines to `self.output_lines` for later
-        reading
+        reading.
         """
-        out_wait = select.poll()
-        out_wait.register(self.process.stdout, select.POLLIN | select.POLLHUP)
-
-        err_wait = select.poll()
-        err_wait.register(self.process.stderr, select.POLLIN | select.POLLHUP)
-
-        with open(self.output_path, "wt", buffering=1) as f:  # Enable line buffering for immediate output
-            f.write("PROCESS START: %s\n" % time.ctime())
-            f.flush()
-            while not self.done:
-                out_changes = out_wait.poll(0.1)
-                if self.process.stdout.fileno() in [fd for (fd, _) in out_changes]:
-                    while True:
-                        out_line = self.process.stdout.readline()
-                        if not out_line:
-                            break
-                        f.write(out_line)
-                        f.flush()
-                        self.output_lines.put(out_line)
-
-                err_changes = err_wait.poll(0)
-                if self.process.stderr.fileno() in [fd for (fd, _) in err_changes]:
-                    err_line = self.process.stderr.readline()
-                    if err_line:
-                        f.write(f"!!STDERR!! : {err_line}")
-                        f.flush()
-            f.write("PROCESS END: %s\n" % time.ctime())
-            f.flush()
+        while not self.done:
+            out_line = self.process.stdout.readline()
+            if not out_line:
+                break
+            self._write_to_file(out_line)
+            self.output_lines.put(out_line)
+
+    def _stderr_thread(self):
+        """Reads stderr process lines and writes them to an output file.
+
+        The lines are marked as error lines when written to the file.
+        """
+        while not self.done:
+            err_line = self.process.stderr.readline()
+            if not err_line:
+                break
+            self._write_to_file(err_line, is_error_line=True)
 
     def __enter__(self):
         self.done = False
@@ -96,8 +99,10 @@ def __enter__(self):
             text=True,
             bufsize=1,  # Enable line buffering for immediate output from subprocess
         )
-        self.io_thread = threading.Thread(target=self._io_thread)
-        self.io_thread.start()
+        self.stdout_thread = threading.Thread(target=self._stdout_thread)
+        self.stderr_thread = threading.Thread(target=self._stderr_thread)
+        self.stdout_thread.start()
+        self.stderr_thread.start()
         return self
 
     def __exit__(self, exception_type, exception_value, traceback):
@@ -106,8 +111,11 @@ def __exit__(self, exception_type, exception_value, traceback):
             self.process.terminate()
             self.process.wait()
 
-        if self.io_thread:
-            self.io_thread.join()
+        if self.stdout_thread:
+            self.stdout_thread.join()
+
+        if self.stderr_thread:
+            self.stderr_thread.join()
 
         if exception_value:
             # When we fail because of an exception, report the entire log content