make memoizer into an interface class and impls

by this time, all the checkpointing and memoization code bother
needs to be moved out of the DFK

Author: benclifford

parsl/dataflow/dflow.py

@@ -29,7 +29,7 @@
 from parsl.dataflow.dependency_resolvers import SHALLOW_DEPENDENCY_RESOLVER
 from parsl.dataflow.errors import DependencyError, JoinError
 from parsl.dataflow.futures import AppFuture
-from parsl.dataflow.memoization import Memoizer
+from parsl.dataflow.memoization import BasicMemoizer, Memoizer
 from parsl.dataflow.rundirs import make_rundir
 from parsl.dataflow.states import FINAL_FAILURE_STATES, FINAL_STATES, States
 from parsl.dataflow.taskrecord import TaskRecord
@@ -167,11 +167,12 @@ def __init__(self, config: Config) -> None:
 
         # TODO: the parameters that remain here should be parameters that are going to be configured by
         # the user as part of checkpoint/memo configuration object.
-        self.memoizer = Memoizer(memoize=config.app_cache,
-                                 checkpoint_mode=config.checkpoint_mode,
-                                 checkpoint_files=config.checkpoint_files,
-                                 checkpoint_period=config.checkpoint_period)
+        self.memoizer: Memoizer = BasicMemoizer(memoize=config.app_cache,
+                                                checkpoint_mode=config.checkpoint_mode,
+                                                checkpoint_files=config.checkpoint_files,
+                                                checkpoint_period=config.checkpoint_period)
         self.memoizer.run_dir = self.run_dir
+
         self.memoizer.start()
 
         self._modify_checkpointable_tasks_lock = threading.Lock()
@@ -1234,6 +1235,10 @@ def cleanup(self) -> None:
         # should still see it.
         logger.info("DFK cleanup complete")
 
+    # TODO: this should maybe go away: manual explicit checkponting is
+    # a property of the (upcoming) BasicMemoizer, not of a memoisation
+    # plugin in general -- configure a BasicMemoizer separately from the
+    # DFK and call checkpoint on that...
     def checkpoint(self) -> None:
         with self._modify_checkpointable_tasks_lock:
             self.memoizer.checkpoint()

parsl/dataflow/memoization.py

@@ -119,6 +119,23 @@ def id_for_memo_function(f: types.FunctionType, output_ref: bool = False) -> byt
 
 
 class Memoizer:
+    def update_memo_exception(self, task: TaskRecord, e: BaseException) -> None:
+        raise NotImplementedError
+
+    def update_memo_result(self, task: TaskRecord, r: Any) -> None:
+        raise NotImplementedError
+
+    def checkpoint(self, *, task: Optional[TaskRecord] = None) -> None:
+        raise NotImplementedError
+
+    def check_memo(self, task: TaskRecord) -> Optional[Future[Any]]:
+        raise NotImplementedError
+
+    def close(self) -> None:
+        raise NotImplementedError
+
+
+class BasicMemoizer(Memoizer):
     """Memoizer is responsible for ensuring that identical work is not repeated.
 
     When a task is repeated, i.e., the same function is called with the same exact arguments, the