TODO: Revisit these changes

* switched from dill to cloudpickle due to conflicts (cloudpickle needed for dask) * made netcdfplus objects picklable (for immutable info) ... should check whether this is even necessary * moved setting the delayed interrupt the last moment, but that might be a performance risk for toy models
dwhswenson · Jul 21, 2020 · Jul 21, 2020 · Jul 21, 2020 · Jul 22, 2020 · Jul 22, 2020
commit 0198b1d50a3cea1a1e229ac5dc4987c7b3b82385
diff --git a/openpathsampling/engines/delayedinterrupt.py b/openpathsampling/engines/delayedinterrupt.py
@@ -1,5 +1,6 @@
 import signal
 import logging
+import threading
 
 
 # class based on: http://stackoverflow.com/a/21919644/487556
@@ -39,3 +40,12 @@ def __enter__(self):
 
     def __exit__(self, type, value, traceback):
         pass
+
+
+def get_interrupter():
+    if isinstance(threading.currentThread(), threading._MainThread):
+        interrupter = DelayedInterrupt
+    else:
+        interrupter = EmptyContext
+    return interrupter
+
diff --git a/openpathsampling/engines/dynamics_engine.py b/openpathsampling/engines/dynamics_engine.py
@@ -4,17 +4,17 @@
 @author JDC Chodera
 @author: JH Prinz
 """
-
 import logging
 import sys
+import threading
 
 from openpathsampling.netcdfplus import StorableNamedObject
 from openpathsampling.integration_tools import is_simtk_unit_type
 
 from .snapshot import BaseSnapshot
 from .trajectory import Trajectory
 
-from .delayedinterrupt import DelayedInterrupt
+from .delayedinterrupt import get_interrupter
 
 logger = logging.getLogger(__name__)
 
@@ -152,7 +152,8 @@ def __init__(self, options=None, descriptor=None):
 
         self.descriptor = descriptor
         self._check_options(options)
-        self.interrupter = DelayedInterrupt
+
+        self.interrupter = get_interrupter()
 
     @property
     def current_snapshot(self):
@@ -546,7 +547,8 @@ def iter_generate(self, initial, running=None, direction=+1,
                 snapshot = None
 
                 try:
-                    with self.interrupter():
+                    interrupter = get_interrupter()
+                    with interrupter():
                         snapshot = self.generate_next_frame()
 
                         # if self.on_nan != 'ignore' and \

diff --git a/openpathsampling/experimental/storage/callable_codec.py b/openpathsampling/experimental/storage/callable_codec.py
@@ -1,6 +1,7 @@
 import types
 import dis
-import dill
+# import dill
+import cloudpickle as cp
 from .serialization_helpers import has_uuid, do_import
 
 GLOBALS_ERROR_MESSAGE="""
@@ -107,7 +108,11 @@ def default(self, obj):
 
             # Case 2: arbitrary function
             if obj.__module__ == "__main__":
-                all_globals = dill.detect.globalvars(obj)
+                pass
+                # all_globals = dill.detect.globalvars(obj)
+                # TODO: this is a serious oversight (should fail tests)
+                # problems getting dill & cloudpickle to play nice
+                all_globals = {}
             else:
                 all_globals = {}
             errors += self._error_message(GLOBALS_ERROR_MESSAGE,
@@ -127,7 +132,7 @@ def default(self, obj):
 
             return {
                 '__callable_name__': obj.__name__,
-                '_dilled': dill.dumps(obj)
+                '_dilled': cp.dumps(obj)
             }
         return obj
 
@@ -138,7 +143,7 @@ def object_hook(self, dct):
             elif '__module__' in dct:
                 func = do_import(dct['__module__'], dct['__callable_name__'])
             elif '_dilled' in dct:
-                func =  dill.loads(dct['_dilled'])
+                func =  cp.loads(dct['_dilled'])
             else:  # pragma: no cover
                 raise RuntimeError("Error reloading ",
                                    dct['__callable_name__'])

diff --git a/openpathsampling/netcdfplus/base.py b/openpathsampling/netcdfplus/base.py
@@ -294,6 +294,9 @@ def to_dict(self):
             key: self.__dict__[key] for key in keys_to_store
         }
 
+    def __reduce__(self):
+        return (self.from_dict, (self.to_dict(),))
+
     @classmethod
     def from_dict(cls, dct):
         """