tor-browser

expensive_line_transformer.py (7538B)

---

# Copyright 2023 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from abc import ABC, abstractmethod
import logging
import subprocess
import threading
import time
import uuid

from devil.utils import reraiser_thread


class ExpensiveLineTransformer(ABC):
 def __init__(self, process_start_timeout, minimum_timeout, per_line_timeout):
   self._process_start_timeout = process_start_timeout
   self._minimum_timeout = minimum_timeout
   self._per_line_timeout = per_line_timeout
   self._started = False
   # Allow only one thread to call TransformLines() at a time.
   self._lock = threading.Lock()
   # Ensure that only one thread attempts to kill self._proc in Close().
   self._close_lock = threading.Lock()
   self._closed_called = False
   # Assign to None so that attribute exists if Popen() throws.
   self._proc = None
   # Start process eagerly to hide start-up latency.
   self._proc_start_time = None

 def start(self):
   # delay the start of the process, to allow the initialization of the
   # descendant classes first.
   if self._started:
     logging.error('%s: Trying to start an already started command', self.name)
     return

   # Start process eagerly to hide start-up latency.
   self._proc_start_time = time.time()

   if not self.command:
     logging.error('%s: No command available', self.name)
     return

   self._proc = subprocess.Popen(self.command,
                                 bufsize=1,
                                 stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE,
                                 universal_newlines=True,
                                 close_fds=True)
   self._started = True

 def IsClosed(self):
   return (not self._started or self._closed_called
           or self._proc.returncode is not None)

 def IsBusy(self):
   return self._lock.locked()

 def IsReady(self):
   return self._started and not self.IsClosed() and not self.IsBusy()

 def TransformLines(self, lines):
   """Symbolizes names found in the given lines.

   If anything goes wrong (process crashes, timeout, etc), returns |lines|.

   Args:
     lines: A list of strings without trailing newlines.

   Returns:
     A list of strings without trailing newlines.
   """
   if not lines:
     return []

   # symbolized output contain more lines than the input, as the symbolized
   # stacktraces will be added. To account for the extra output lines, keep
   # reading until this eof_line token is reached. Using a format that will
   # be considered a "useful line" without modifying its output by
   # third_party/android_platform/development/scripts/stack_core.py
   eof_line = self.getEofLine()
   out_lines = []

   def _reader():
     while True:
       line = self._proc.stdout.readline()
       # Return an empty string at EOF (when stdin is closed).
       if not line:
         break
       line = line[:-1]
       if line == eof_line:
         break
       out_lines.append(line)

   if self.IsBusy():
     logging.warning('%s: Having to wait for transformation.', self.name)

   # Allow only one thread to operate at a time.
   with self._lock:
     if self.IsClosed():
       if self._started and not self._closed_called:
         logging.warning('%s: Process exited with code=%d.', self.name,
                         self._proc.returncode)
         self.Close()
       return lines

     reader_thread = reraiser_thread.ReraiserThread(_reader)
     reader_thread.start()

     try:
       self._proc.stdin.write('\n'.join(lines))
       self._proc.stdin.write('\n{}\n'.format(eof_line))
       self._proc.stdin.flush()
       time_since_proc_start = time.time() - self._proc_start_time
       timeout = (max(0, self._process_start_timeout - time_since_proc_start) +
                  max(self._minimum_timeout,
                      len(lines) * self._per_line_timeout))
       reader_thread.join(timeout)
       if self.IsClosed():
         logging.warning('%s: Close() called by another thread during join().',
                         self.name)
         return lines
       if reader_thread.is_alive():
         logging.error('%s: Timed out after %f seconds with input:', self.name,
                       timeout)
         for l in lines:
           logging.error(l)
         logging.error(eof_line)
         logging.error('%s: End of timed out input.', self.name)
         logging.error('%s: Timed out output was:', self.name)
         for l in out_lines:
           logging.error(l)
         logging.error('%s: End of timed out output.', self.name)
         self.Close()
         return lines
       return out_lines
     except IOError:
       logging.exception('%s: Exception during transformation', self.name)
       self.Close()
       return lines

 def Close(self):
   with self._close_lock:
     needs_closing = not self.IsClosed()
     self._closed_called = True

   if needs_closing:
     self._proc.stdin.close()
     self._proc.kill()
     self._proc.wait()

 def __del__(self):
   # self._proc is None when Popen() fails.
   if not self._closed_called and self._proc:
     logging.error('%s: Forgot to Close()', self.name)
     self.Close()

 @property
 @abstractmethod
 def name(self):
   ...

 @property
 @abstractmethod
 def command(self):
   ...

 @staticmethod
 def getEofLine():
   # Use a format that will be considered a "useful line" without modifying its
   # output by third_party/android_platform/development/scripts/stack_core.py
   return "Generic useful log header: \'{}\'".format(uuid.uuid4().hex)


class ExpensiveLineTransformerPool(ABC):
 def __init__(self, max_restarts, pool_size, passthrough_on_failure):
   self._max_restarts = max_restarts
   self._pool = [self.CreateTransformer() for _ in range(pool_size)]
   self._passthrough_on_failure = passthrough_on_failure
   # Allow only one thread to select from the pool at a time.
   self._lock = threading.Lock()
   self._num_restarts = 0

 def __enter__(self):
   pass

 def __exit__(self, *args):
   self.Close()

 def TransformLines(self, lines):
   with self._lock:
     assert self._pool, 'TransformLines() called on a closed Pool.'

     # transformation is broken.
     if self._num_restarts == self._max_restarts:
       if self._passthrough_on_failure:
         return lines
       raise Exception('%s is broken.' % self.name)

     # Restart any closed transformer.
     for i, d in enumerate(self._pool):
       if d.IsClosed():
         logging.warning('%s: Restarting closed instance.', self.name)
         self._pool[i] = self.CreateTransformer()
         self._num_restarts += 1
         if self._num_restarts == self._max_restarts:
           logging.warning('%s: MAX_RESTARTS reached.', self.name)
           if self._passthrough_on_failure:
             return lines
           raise Exception('%s is broken.' % self.name)

     selected = next((x for x in self._pool if x.IsReady()), self._pool[0])
     # Rotate the order so that next caller will not choose the same one.
     self._pool.remove(selected)
     self._pool.append(selected)

   return selected.TransformLines(lines)

 def Close(self):
   with self._lock:
     for d in self._pool:
       d.Close()
     self._pool = None

 @abstractmethod
 def CreateTransformer(self):
   ...

 @property
 @abstractmethod
 def name(self):
   ...