Add support to dill serializer

billiard/common.py

@@ -6,6 +6,7 @@
 import signal
 import sys
 
+import dill
 import pickle
 
 from .exceptions import RestartFreqExceeded
@@ -14,6 +15,9 @@
 pickle_load = pickle.load
 pickle_loads = pickle.loads
 
+dill_load = dill.load
+dill_loads = dill.loads
+
 # cPickle.loads does not support buffer() objects,
 # but we can just create a StringIO and use load.
 from io import BytesIO

billiard/pool.py

@@ -27,7 +27,7 @@
 from . import cpu_count, get_context
 from . import util
 from .common import (
-    TERM_SIGNAL, human_status, pickle_loads, reset_signals, restart_state,
+    TERM_SIGNAL, human_status, dill_loads, reset_signals, restart_state,
 )
 from .compat import get_errno, mem_rss, send_offset
 from .einfo import ExceptionInfo
@@ -441,7 +441,7 @@ def _make_recv_method(self, conn):
             if hasattr(conn, 'get_payload') and conn.get_payload:
                 get_payload = conn.get_payload
 
-                def _recv(timeout, loads=pickle_loads):
+                def _recv(timeout, loads=dill_loads):
                     return True, loads(get_payload())
             else:
                 def _recv(timeout):  # noqa
@@ -456,7 +456,7 @@ def _recv(timeout):  # noqa
                     return False, None
         return _recv
 
-    def _make_child_methods(self, loads=pickle_loads):
+    def _make_child_methods(self, loads=dill_loads):
         self.wait_for_job = self._make_protected_receive(self.inq)
         self.wait_for_syn = (self._make_protected_receive(self.synq)
                              if self.synq else None)

requirements/default.txt

@@ -0,0 +1 @@
+dill>=0.3.8

setup.py

@@ -166,6 +166,42 @@ def _is_build_command(argv=sys.argv, cmds=('install', 'build', 'bdist')):
             return arg
 
 
+def _strip_comments(l):
+    return l.split('#', 1)[0].strip()
+
+
+def _pip_requirement(req):
+    if req.startswith('-r '):
+        _, path = req.split()
+        return reqs(*path.split('/'))
+    return [req]
+
+
+def _reqs(*f):
+    return [
+        _pip_requirement(r) for r in (
+            _strip_comments(l) for l in open(
+                os.path.join(os.getcwd(), 'requirements', *f)).readlines()
+        ) if r]
+
+
+def reqs(*f):
+    """Parse requirement file.
+
+    Example:
+        reqs('default.txt')          # requirements/default.txt
+        reqs('extras', 'redis.txt')  # requirements/extras/redis.txt
+    Returns:
+        List[str]: list of requirements specified in the file.
+    """
+    return [req for subreq in _reqs(*f) for req in subreq]
+
+
+def install_requires():
+    """Get list of requirements required for installation."""
+    return reqs('default.txt')
+
+
 def run_setup(with_extensions=True):
     extensions = []
     if with_extensions:
@@ -204,6 +240,7 @@ def run_setup(with_extensions=True):
         maintainer=meta['maintainer'],
         maintainer_email=meta['contact'],
         url=meta['homepage'],
+        install_requires=install_requires(),
         zip_safe=False,
         license='BSD',
         python_requires='>=3.7',