Fix a process-wide global singleton

Author: shunping

.github/trigger_files/beam_PostCommit_Python_Versions.json

@@ -0,0 +1,4 @@
+{
+  "comment": "Modify this file in a trivial way to cause this test suite to run",
+  "revision": 1
+}

sdks/python/apache_beam/runners/portability/prism_runner_test.py

@@ -37,6 +37,7 @@
 
 import apache_beam as beam
 from apache_beam.options.pipeline_options import DebugOptions
+from apache_beam.options.pipeline_options import PipelineOptions
 from apache_beam.options.pipeline_options import PortableOptions
 from apache_beam.options.pipeline_options import StandardOptions
 from apache_beam.options.pipeline_options import TypeOptions
@@ -49,6 +50,7 @@
 from apache_beam.transforms import trigger
 from apache_beam.transforms import window
 from apache_beam.utils import shared
+from apache_beam.utils import subprocess_server
 
 # Run as
 #

sdks/python/apache_beam/transforms/external_test.py

@@ -799,7 +799,13 @@ def test_implicit_builder_with_constructor_method(self):
 
 class JavaJarExpansionServiceTest(unittest.TestCase):
   def setUp(self):
-    SubprocessServer._cache._live_owners = set()
+    SubprocessServer._cache._purge_buffer.append({
+        'owner': 'reset', 'pid': os.getpid(), 'live_owners': []
+    })
+    patcher = mock.patch.object(
+        SubprocessServer._cache, '_live_owners', new=set())
+    patcher.start()
+    self.addCleanup(patcher.stop)
 
   def test_classpath(self):
     with tempfile.TemporaryDirectory() as temp_dir:

sdks/python/apache_beam/utils/subprocess_server.py

@@ -76,25 +76,47 @@ def __init__(self, constructor, destructor):
     self._cache = {}
     self._lock = threading.Lock()
     self._counter = 0
+    self._purge_buffer = []
+    self._owner2_buffer = []
 
   def _next_id(self):
-    with self._lock:
-      self._counter += 1
-      return self._counter
+    # Caller must hold self._lock.
+    self._counter += 1
+    return self._counter
 
   def register(self):
-    owner = self._next_id()
-    self._live_owners.add(owner)
+    with self._lock:
+      owner = self._next_id()
+      self._live_owners.add(owner)
+      if owner == 2:
+        self._owner2_buffer.append({
+            'action': 'register',
+            'pid': os.getpid(),
+            'live_owners': list(self._live_owners)
+        })
     return owner
 
   def purge(self, owner):
     to_delete = []
     with self._lock:
+      self._purge_buffer.append({
+          'owner': owner,
+          'pid': os.getpid(),
+          'live_owners': ("%s" % self._live_owners),
+      })
+      if owner == 2:
+        self._owner2_buffer.append({
+            'action': 'purge',
+            'pid': os.getpid(),
+            'live_owners': list(self._live_owners)
+        })
       if owner not in self._live_owners:
         _LOGGER.warning(
             "Subprocess owner %s already purged. If this occurs during atexit "
-            "shutdown, the subprocess was already cleaned up earlier.",
-            owner)
+            "shutdown, the subprocess was already cleaned up earlier. Purge buffer: %s. Owner 2 buffer: %s",
+            owner,
+            self._purge_buffer,
+            self._owner2_buffer)
         return
       self._live_owners.remove(owner)
       for key, entry in list(self._cache.items()):
@@ -103,6 +125,11 @@ def purge(self, owner):
         if not entry.owners:
           to_delete.append(entry.obj)
           del self._cache[key]
+      if to_delete:
+        _LOGGER.warning(
+            "Purging subprocess. Purge buffer: %s. Owner 2 buffer: %s",
+            self._purge_buffer,
+            self._owner2_buffer)
     # Actually call the destructors outside of the lock.
     for value in to_delete:
       self._destructor(value)

sdks/python/pytest.ini

@@ -17,6 +17,8 @@
 
 [pytest]
 junit_family = xunit2
+log_cli = true
+log_cli_level = WARNING
 filterwarnings =
   ignore:Deprecated call to `pkg_resources.declare_namespace\('.*'\):DeprecationWarning
   ignore::DeprecationWarning:google.rpc