elf_symbolizer: Use a process for max 4000 lookups and then restart

build/android/pylib/symbols/elf_symbolizer.py

 # Copyright 2014 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import collections
 import datetime
 import logging
 import multiprocessing
 import os
 import posixpath
 import Queue
 import re
 import subprocess
 import sys
 import threading
 
 
+# addr2line builds a possibly infinite memory cache that can exhaust
+# the computer's memory if allowed to grow for too long. This constant
+# controls how many lookups we do before restarting the process. 4000
+# gives near peak performance without extreme memory usage.
+ADDR2LINE_RECYCLE_LIMIT = 4000
+
+
 class ELFSymbolizer(object):
   """An uber-fast (multiprocessing, pipelined and asynchronous) ELF symbolizer.
 
   This class is a frontend for addr2line (part of GNU binutils), designed to
   symbolize batches of large numbers of symbols for a given ELF file. It
   supports sharding symbolization against many addr2line instances and
   pipelining of multiple requests per each instance (in order to hide addr2line
   internals and OS pipe latencies).
 
   The interface exhibited by this class is a very simple asynchronous interface,
@@ skipping 82 matching lines @@
 
     Args:
       addr: address to symbolize.
       callback_arg: optional argument which will be passed to the |callback|."""
     assert(isinstance(addr, int))
 
     # Process all the symbols that have been resolved in the meanwhile.
     # Essentially, this drains all the addr2line(s) out queues.
     for a2l_to_purge in self._a2l_instances:
       a2l_to_purge.ProcessAllResolvedSymbolsInQueue()
+      a2l_to_purge.RecycleIfNecessary()
 
     # Find the best instance according to this logic:
     # 1. Find an existing instance with the shortest queue.
     # 2. If all of instances' queues are full, but there is room in the pool,
     #    (i.e. < |max_concurrent_jobs|) create a new instance.
     # 3. If there were already |max_concurrent_jobs| instances and all of them
     #    had full queues, make back-pressure.
 
     # 1.
     def _SortByQueueSizeAndReqID(a2l):
@@ skipping 47 matching lines @@
       self._lib_file_name = posixpath.basename(symbolizer.elf_file_path)
 
       # The request queue (i.e. addresses pushed to addr2line's stdin and not
       # yet retrieved on stdout)
       self._request_queue = collections.deque()
 
       # This is essentially len(self._request_queue). It has been optimized to a
       # separate field because turned out to be a perf hot-spot.
       self.queue_size = 0
 
+      # Keep track of the number of symbols a process has processed to
+      # avoid a single process growing too big and using all the memory.
+      self._processed_symbols_count = 0
+
       # Objects required to handle the addr2line subprocess.
       self._proc = None  # Subprocess.Popen(...) instance.
       self._thread = None  # Threading.thread instance.
       self._out_queue = None  # Queue.Queue instance (for buffering a2l stdout).
       self._RestartAddr2LineProcess()
 
     def EnqueueRequest(self, addr, callback_arg):
       """Pushes an address to addr2line's stdin (and keeps track of it)."""
       self._symbolizer.requests_counter += 1  # For global "age" of requests.
       req_idx = self._symbolizer.requests_counter
@@ skipping 47 matching lines @@
       """Consumes all the addr2line output lines produced (without blocking)."""
       if not self.queue_size:
         return
       while True:
         try:
           lines = self._out_queue.get_nowait()
         except Queue.Empty:
           break
         self._ProcessSymbolOutput(lines)
 
+    def RecycleIfNecessary(self):
+      """Restarts the process if it has been used for too long.
+
+      A long running addr2line process will consume excessive amounts
+      of memory without any gain in performance."""
+      if self._processed_symbols_count >= ADDR2LINE_RECYCLE_LIMIT:
+        self._RestartAddr2LineProcess()
+
+
     def Terminate(self):
       """Kills the underlying addr2line process.
 
       The poller |_thread| will terminate as well due to the broken pipe."""
       try:
         self._proc.kill()
         self._proc.communicate()  # Essentially wait() without risking deadlock.
       except Exception:  # An exception while terminating? How interesting.
         pass
       self._proc = None
@@ skipping 26 matching lines @@
               source_line = int(m.group(2))
         else:
           logging.warning('Got invalid symbol path from addr2line: %s' % line2)
 
         sym_info = ELFSymbolInfo(name, source_path, source_line)
         if prev_sym_info:
           prev_sym_info.inlined_by = sym_info
         if not innermost_sym_info:
           innermost_sym_info = sym_info
 
+      self._processed_symbols_count += 1
       self._symbolizer.callback(innermost_sym_info, callback_arg)
 
     def _RestartAddr2LineProcess(self):
       if self._proc:
         self.Terminate()
 
       # The only reason of existence of this Queue (and the corresponding
       # Thread below) is the lack of a subprocess.stdout.poll_avail_lines().
       # Essentially this is a pipe able to extract a couple of lines atomically.
       self._out_queue = Queue.Queue()
 
       # Start the underlying addr2line process in line buffered mode.
 
       cmd = [self._symbolizer.addr2line_path, '--functions', '--demangle',
           '--exe=' + self._symbolizer.elf_file_path]
       if self._symbolizer.inlines:
         cmd += ['--inlines']
       self._proc = subprocess.Popen(cmd, bufsize=1, stdout=subprocess.PIPE,
           stdin=subprocess.PIPE, stderr=sys.stderr, close_fds=True)
 
       # Start the poller thread, which simply moves atomically the lines read
       # from the addr2line's stdout to the |_out_queue|.
       self._thread = threading.Thread(
           target=ELFSymbolizer.Addr2Line.StdoutReaderThread,
           args=(self._proc.stdout, self._out_queue, self._symbolizer.inlines))
       self._thread.daemon = True  # Don't prevent early process exit.
       self._thread.start()
 
+      self._processed_symbols_count = 0
+
       # Replay the pending requests on the new process (only for the case
       # of a hung addr2line timing out during the game).
       for (addr, _, _) in self._request_queue:
         self._WriteToA2lStdin(addr)
 
     @staticmethod
     def StdoutReaderThread(process_pipe, queue, inlines):
       """The poller thread fn, which moves the addr2line stdout to the |queue|.
 
       This is the only piece of code not running on the main thread. It merely
@@ skipping 36 matching lines @@
     self.name = name
     self.source_path = source_path
     self.source_line = source_line
     # In the case of |inlines|=True, the |inlined_by| points to the outer
     # function inlining the current one (and so on, to form a chain).
     self.inlined_by = None
 
   def __str__(self):
     return '%s [%s:%d]' % (
         self.name or '??', self.source_path or '??', self.source_line or 0)