Improve fork() handling (#37)

* For Python3.7+, improve fork() handling by stopping/starting the registry using 'os.register_at_fork()'

* Move to bionic for Travis tests

Co-authored-by: JSJ <[email protected]>

Author: jsjeannotte

.travis.yml

@@ -11,7 +11,7 @@ matrix:
     - python: 3.8
       env: TOXENV=py38
 
-dist: xenial
+dist: bionic
 
 install: pip install tox
 

README.md

@@ -9,6 +9,11 @@ See the Spectator [documentation] for an overview of core concepts and details o
 
 Supports Python >= 3.5, which is the oldest system Python 3 available on our commonly used OSes.
 
+Note that there is a risk of deadlock if you are running Python 3.6 or lower and using 
+`os.fork()` or using a library that will fork the process 
+(see [this section](#concurrent-usage-under-older-python) for workarounds),
+so **we recommend using Python >= 3.7**
+
 [Spectator]: https://github.com/Netflix/spectator/
 [documentation]: https://netflix.github.io/atlas-docs/spectator/
 [usage]: https://netflix.github.io/atlas-docs/spectator/lang/py/usage/
@@ -54,7 +59,9 @@ from spectator import GlobalRegistry
 
 Once the `GlobalRegistry` is imported, it is used to create and manage Meters.
 
-### Concurrent Usage
+### Concurrent Usage Under Older Python
+
+> :warning: **Use Python 3.7+ if possible**: But if you can't, here's a workaround to prevent deadlocks
 
 There is a known issue in Python where forking a process after a thread is started can lead to
 deadlocks. This is commonly seen when using the `multiprocessing` module with default settings.
@@ -84,6 +91,8 @@ WSGI_GUNICORN_PRELOAD = undef
 
 #### Task Worker Forking
 
+> :warning: **Use Python 3.7+ if possible**: But if you can't, here's a workaround to prevent deadlocks
+
 For other pre-fork worker processing frameworks, such as [huey], you need to be careful about how
 and when you start the `GlobalRegistry` to avoid deadlocks in the background publish thread. You
 should set the `SPECTATOR_PY_DISABLE_AUTO_START_GLOBAL` environment variable to disable automatic
@@ -120,6 +129,8 @@ workers, to help ensure that it is not started when the module is loaded.
 
 #### Generic Multiprocessing
 
+> :warning: **Use Python 3.7+ if possible**: But if you can't, here's a workaround to prevent deadlocks
+
 In Python 3, you can configure the start method to `spawn` for `multiprocessing`. This will cause
 the module to do a `fork()` followed by an `execve()` to start a brand new Python process.
 
@@ -130,7 +141,7 @@ from multiprocessing import set_start_method
 set_start_method("spawn")
 ```
 
-To configure thid option within a context:
+To configure this option within a context:
 
 ```python
 from multiprocessing import get_context

spectator/__init__.py

@@ -16,6 +16,14 @@
     if auto_start_global():
         GlobalRegistry.start()
         atexit.register(GlobalRegistry.stop)
+        try:
+            from os import register_at_fork
+            register_at_fork(before=GlobalRegistry.stop_without_publish,
+                             after_in_parent=GlobalRegistry.start,
+                             after_in_child=GlobalRegistry.clear_meters_and_start)
+        except ImportError:
+            pass
+
     else:
         logger.debug("module spectatorconfig auto-start is disabled - GlobalRegistry will not start")
 except ImportError:

spectator/registry.py

@@ -101,12 +101,29 @@ def start(self, config=None):
             logger.debug("registry started with config: %s", config)
             return RegistryStopper(self)
 
-    def stop(self):
+    def clear_meters_and_start(self):
+        """
+        This is called after a fork in the child process
+        to clear the cloned `_meters` and prevent duplicates
+        (the `_meters` are copied with the process
+        during the forking)
+        """
+        self._meters = {}
+        self.start()
+
+    def stop_without_publish(self):
+        """
+        This is called before a fork to prevent a potential deadlock.
+        It cancels the background timer thread. After the fork, the timer
+        thread is restarted in the main and cloned processes.
+        """
         if self._started:
-            logger.info("stopping registry")
+            logger.debug("stopping log registry")
             self._timer.cancel()
             self._started = False
 
+    def stop(self):
+        self.stop_without_publish()
         # Even if not started, attempt to flush data to minimize risk
         # of data loss
         self._publish()