Resolve comments

Author: AMOOOMA

sdks/python/apache_beam/utils/multi_process_shared.py

@@ -186,7 +186,6 @@ def release_singleton(self, tag, obj):
 
   def unsafe_hard_delete_singleton(self, tag):
     self.entries[tag].unsafe_hard_delete()
-    self._hard_delete_callback()
 
 
 _process_level_singleton_manager = _SingletonManager()
@@ -236,13 +235,7 @@ def get_auto_proxy_object(self):
     return self._proxyObject
 
   def unsafe_hard_delete(self):
-    try:
-      self._proxyObject.unsafe_hard_delete()
-    except (EOFError, ConnectionResetError, BrokenPipeError):
-      pass
-    except Exception as e:
-      logging.warning(
-          "Exception %s when trying to hard delete shared object proxy", e)
+    self._proxyObject.unsafe_hard_delete()
 
 
 def _run_server_process(address_file, tag, constructor, authkey):
@@ -259,8 +252,8 @@ def cleanup_files():
         os.remove(address_file)
       if os.path.exists(address_file + ".error"):
         os.remove(address_file + ".error")
-    except Exception:
-      pass
+    except Exception as e:
+      logging.warning('Failed to cleanup files for tag %s: %s', tag, e)
 
   def handle_unsafe_hard_delete():
     cleanup_files()
@@ -270,6 +263,9 @@ def _monitor_parent():
     """Checks if parent is alive every second."""
     while True:
       try:
+        # Sends a check to see if parent_pid is still alive,
+        # this call will fail with OSError if the parent has died
+        # and no-op if alive.
         os.kill(parent_pid, 0)
       except OSError:
         logging.warning(
@@ -284,7 +280,6 @@ def _monitor_parent():
 
   try:
     t = threading.Thread(target=_monitor_parent, daemon=True)
-    t.start()
 
     logging.getLogger().setLevel(logging.INFO)
     multiprocessing.current_process().authkey = authkey
@@ -298,6 +293,9 @@ def _monitor_parent():
         tag,
         initialize_eagerly=True,
         hard_delete_callback=handle_unsafe_hard_delete)
+    # Start monitoring parent after initialisation is done to avoid
+    # potential race conditions.
+    t.start()
 
     server = serving_manager.get_server()
     logging.info(
@@ -422,7 +420,8 @@ def acquire(self):
     # Trigger a sweep of zombie processes.
     # calling active_children() has the side-effect of joining any finished
     # processes, effectively reaping zombies from previous unsafe_hard_deletes.
-    if self._spawn_process: multiprocessing.active_children()
+    if self._spawn_process:
+      multiprocessing.active_children()
     return _AutoProxyWrapper(singleton)
 
   def release(self, obj):
@@ -437,15 +436,7 @@ def unsafe_hard_delete(self):
       to this object exist, or (b) you are ok with all existing references to
       this object throwing strange errors when derefrenced.
     """
-    try:
-      self._get_manager().unsafe_hard_delete_singleton(self._tag)
-    except (EOFError, ConnectionResetError, BrokenPipeError):
-      pass
-    except Exception as e:
-      logging.warning(
-          "Exception %s when trying to hard delete shared object %s",
-          e,
-          self._tag)
+    self._get_manager().unsafe_hard_delete_singleton(self._tag)
 
   def _create_server(self, address_file):
     if self._spawn_process:
@@ -477,8 +468,11 @@ def cleanup_process():
             os.remove(address_file)
           if os.path.exists(error_file):
             os.remove(error_file)
-        except Exception:
-          pass
+        except Exception as e:
+          logging.warning(
+              'Failed to cleanup files for tag %s in atexit handler: %s',
+              self._tag,
+              e)
 
       atexit.register(cleanup_process)
 

sdks/python/apache_beam/utils/multi_process_shared_test.py

@@ -189,8 +189,11 @@ def test_unsafe_hard_delete(self):
     self.assertEqual(counter1.increment(), 1)
     self.assertEqual(counter2.increment(), 2)
 
-    multi_process_shared.MultiProcessShared(
-        Counter, tag='test_unsafe_hard_delete').unsafe_hard_delete()
+    try:
+      multi_process_shared.MultiProcessShared(
+          Counter, tag='test_unsafe_hard_delete').unsafe_hard_delete()
+    except Exception:
+      pass
 
     with self.assertRaises(Exception):
       counter1.get()
@@ -219,7 +222,10 @@ def test_unsafe_hard_delete_autoproxywrapper(self):
     self.assertEqual(counter1.increment(), 1)
     self.assertEqual(counter2.increment(), 2)
 
-    counter2.unsafe_hard_delete()
+    try:
+      counter2.unsafe_hard_delete()
+    except Exception:
+      pass
 
     with self.assertRaises(Exception):
       counter1.get()
@@ -243,8 +249,11 @@ def test_unsafe_hard_delete_no_op(self):
     self.assertEqual(counter1.increment(), 1)
     self.assertEqual(counter2.increment(), 2)
 
-    multi_process_shared.MultiProcessShared(
-        Counter, tag='no_tag_to_delete').unsafe_hard_delete()
+    try:
+      multi_process_shared.MultiProcessShared(
+          Counter, tag='no_tag_to_delete').unsafe_hard_delete()
+    except Exception:
+      pass
 
     self.assertEqual(counter1.increment(), 3)
     self.assertEqual(counter2.increment(), 4)
@@ -298,8 +307,9 @@ def setUp(self):
                 'main',
                 'to_delete',
                 'mix1',
-                'mix2'
-                'test_process_exit']:
+                'mix2',
+                'test_process_exit',
+                'thundering_herd_test']:
       for ext in ['', '.address', '.address.error']:
         try:
           os.remove(os.path.join(tempdir, tag + ext))
@@ -326,7 +336,10 @@ def test_proxy_on_proxy(self):
     instance = shared1.acquire()
     proxy_instance = instance.make_proxy(spawn_process=True)
     self.assertEqual(proxy_instance.increment(), 1)
-    proxy_instance.unsafe_hard_delete()
+    try:
+      proxy_instance.unsafe_hard_delete()
+    except Exception:
+      pass
 
     proxy_instance2 = instance.make_proxy(tag='proxy_2', spawn_process=True)
     self.assertEqual(proxy_instance2.increment(), 1)
@@ -344,7 +357,10 @@ def test_unsafe_hard_delete_autoproxywrapper(self):
     self.assertEqual(counter1.increment(), 1)
     self.assertEqual(counter2.increment(), 2)
 
-    counter2.unsafe_hard_delete()
+    try:
+      counter2.unsafe_hard_delete()
+    except Exception:
+      pass
 
     with self.assertRaises(Exception):
       counter1.get()
@@ -385,7 +401,10 @@ def test_process_exits_on_unsafe_hard_delete(self):
 
     self.assertIsNotNone(
         server_process, "Could not find spawned server process")
-    obj.unsafe_hard_delete()
+    try:
+      obj.unsafe_hard_delete()
+    except Exception:
+      pass
     server_process.join(timeout=5)
 
     self.assertFalse(
@@ -413,7 +432,10 @@ def test_process_exits_on_unsafe_hard_delete_with_manager(self):
 
     self.assertIsNotNone(
         server_process, "Could not find spawned server process")
-    shared.unsafe_hard_delete()
+    try:
+      shared.unsafe_hard_delete()
+    except Exception:
+      pass
     server_process.join(timeout=5)
 
     self.assertFalse(
@@ -434,7 +456,10 @@ def test_zombie_reaping_on_acquire(self):
     server_pid = next(
         p.pid for p in children if p.is_alive() and p.pid != os.getpid())
 
-    obj.unsafe_hard_delete()
+    try:
+      obj.unsafe_hard_delete()
+    except Exception:
+      pass
 
     try:
       os.kill(server_pid, 0)
@@ -458,7 +483,10 @@ def test_zombie_reaping_on_acquire(self):
     self.assertFalse(
         pid_exists,
         f"Old server process {server_pid} was not reaped by acquire() sweep")
-    shared2.unsafe_hard_delete()
+    try:
+      shared2.unsafe_hard_delete()
+    except Exception:
+      pass
 
 
 if __name__ == '__main__':