CU-8699qg2tg: Hopefully fix multiprocessing stalling by removing the … (#36)

* CU-8699qg2tg: Hopefully fix multiprocessing stalling by removing the dunder del method from the usage monitor for multiprocessing

* CU-8699qg2tg: Fix minor oversight in changes

* CU-8699qg2tg: Hopefully fix usage monitoring with no dunder del method for good

Author: mart-r

medcat-v2/medcat/cat.py

@@ -5,6 +5,7 @@
 from datetime import date
 from concurrent.futures import ProcessPoolExecutor, as_completed, Future
 import itertools
+from contextlib import contextmanager
 
 import shutil
 import logging
@@ -30,7 +31,7 @@
 from medcat.utils.defaults import avoid_legacy_conversion
 from medcat.utils.defaults import doing_legacy_conversion_message
 from medcat.utils.defaults import LegacyConversionDisabledError
-from medcat.utils.usage_monitoring import UsageMonitor
+from medcat.utils.usage_monitoring import UsageMonitor, _NoDelUM
 from medcat.utils.import_utils import MissingDependenciesError
 
 
@@ -357,6 +358,28 @@ def get_entities_multi_texts(
                     yield text_index, result
             return
 
+        with self._no_usage_monitor_exit_flushing():
+            yield from self._multiprocess(n_process, batch_iter)
+
+    @contextmanager
+    def _no_usage_monitor_exit_flushing(self):
+        # NOTE: the `UsageMonitor.__del__` method can cause
+        #       multiprocessing to stall while it waits for it to be
+        #       called. So here we remove the method.
+        #       However, due to the object being pickled for multiprocessing
+        #       purposes, the class'es `__del__` method will be used anyway.
+        #       So we need to trick it into using a different class.
+        original_cls = self.usage_monitor.__class__
+        self.usage_monitor.__class__ = _NoDelUM
+        try:
+            yield
+        finally:
+            self.usage_monitor.__class__ = original_cls
+
+    def _multiprocess(
+            self, n_process: int,
+            batch_iter: Iterator[list[tuple[str, str, bool]]]
+            ) -> Iterator[tuple[str, Union[dict, Entities, OnlyCUIEntities]]]:
         external_processes = n_process - 1
         with ProcessPoolExecutor(max_workers=external_processes) as executor:
             yield from self._mp_one_batch_per_process(

medcat-v2/medcat/utils/usage_monitoring.py

@@ -121,6 +121,12 @@ def __del__(self):
                 raise
 
 
+class _NoDelUM(UsageMonitor):
+
+    def __del__(self):
+        pass
+
+
 # NOTE: at termination time, open is not available
 #       while this isn't great, there's nothing we can do
 def _says_open_not_available(err: NameError) -> bool: