parallel: fix the leak of pymp-xxx temp directories

multiprocessing.managers.SyncManager spawns and controls a manager
process. Upon exit, SyncManager instructs the process to shutdown and
wait for it to join with a timeout of 0.2s. After hitting the timeout,
it then proceed to terminate the process. The timeout is too low and
the process is often killed before shutting down properly. This causes
temporary directories (pymp-xxx) not being cleaned up.

This CL bypasses the problem in the core library by using a subclass
of SyncManager which sets the default timeout to 1s.

BUG=chromium:348714
TEST=Ran `cbuildbot/run_tests` and checked no /tmp/pymp-xxx directories

Change-Id: I04b06676f449cab0a62f4139daffa0b84f253b87
Reviewed-on: https://chromium-review.googlesource.com/202544
Reviewed-by: Mike Frysinger <vapier@chromium.org>
Commit-Queue: Yu-Ju Hong <yjhong@chromium.org>
Tested-by: Yu-Ju Hong <yjhong@chromium.org>

1 files changed, 29 insertions, 1 deletions

diff --git a/lib/parallel.py b/lib/parallel.py
index 3229e1ead..26c2e59b0 100644
--- a/lib/parallel.py
+++ b/lib/parallel.py
@@ -24,6 +24,8 @@ import tempfile
 import time
 import traceback
 
+from multiprocessing.managers import SyncManager
+
 from chromite.cbuildbot import cbuildbot_failures as failures_lib
 from chromite.cbuildbot import cbuildbot_results as results_lib
 from chromite.lib import cros_build_lib
@@ -35,6 +37,30 @@ _BUFSIZE = 1024
 logger = logging.getLogger(__name__)
 
 
+class HackTimeoutSyncManager(SyncManager):
+  """Increase the process join timeout in SyncManager.
+
+  The timeout for the manager process to join in the core library is
+  too low. The process is often killed before shutting down properly,
+  resulting in temporary directories (pymp-xxx) not being cleaned
+  up. This class increases the default timeout.
+  """
+
+  @staticmethod
+  def _finalize_manager(process, *args, **kwargs):
+    """Shutdown the manager process."""
+
+    def _join(functor, *args, **kwargs):
+      timeout = kwargs.get('timeout')
+      if not timeout is None and timeout < 1:
+        kwargs['timeout'] = 1
+
+      functor(*args, **kwargs)
+
+    process.join = functools.partial(_join, process.join)
+    SyncManager._finalize_manager(process, *args, **kwargs)
+
+
 def Manager():
   """Create a background process for managing interprocess communication.
 
@@ -53,7 +79,9 @@ def Manager():
   """
   old_tempdir_value, old_tempdir_env = osutils.SetGlobalTempDir('/tmp')
   try:
-    return multiprocessing.Manager()
+    m = HackTimeoutSyncManager()
+    m.start()
+    return m
   finally:
     osutils.SetGlobalTempDir(old_tempdir_value, old_tempdir_env)