configurable memoizer instance

TODO: rewrite documentation that refers to any example checkpointing or app caching or memoization configuration
with the documentation ready to receive documentation for the SQL plugin

this PR is the biggest user facing change because it changes the configuration interface for anyone configuring
anything to do with app caching/memoization/checkpointing.

Author: benclifford

parsl/config.py

@@ -5,6 +5,7 @@
 from typing_extensions import Literal
 
 from parsl.dataflow.dependency_resolvers import DependencyResolver
+from parsl.dataflow.memoization import Memoizer
 from parsl.dataflow.taskrecord import TaskRecord
 from parsl.errors import ConfigurationError
 from parsl.executors.base import ParslExecutor
@@ -27,17 +28,6 @@ class Config(RepresentationMixin, UsageInformation):
     executors : sequence of ParslExecutor, optional
         List (or other iterable) of `ParslExecutor` instances to use for executing tasks.
         Default is (:class:`~parsl.executors.threads.ThreadPoolExecutor()`,).
-    app_cache : bool, optional
-        Enable app caching. Default is True.
-    checkpoint_files : sequence of str, optional
-        List of paths to checkpoint files. See :func:`parsl.utils.get_all_checkpoints` and
-        :func:`parsl.utils.get_last_checkpoint` for helpers. Default is None.
-    checkpoint_mode : str, optional
-        Checkpoint mode to use, can be ``'dfk_exit'``, ``'task_exit'``, ``'periodic'`` or ``'manual'``.
-        If set to `None`, checkpointing will be disabled. Default is None.
-    checkpoint_period : str, optional
-        Time interval (in "HH:MM:SS") at which to checkpoint completed tasks. Only has an effect if
-        ``checkpoint_mode='periodic'``.
     dependency_resolver: plugin point for custom dependency resolvers. Default: only resolve Futures,
         using the `SHALLOW_DEPENDENCY_RESOLVER`.
     exit_mode: str, optional
@@ -100,14 +90,7 @@ class Config(RepresentationMixin, UsageInformation):
     @typeguard.typechecked
     def __init__(self,
                  executors: Optional[Iterable[ParslExecutor]] = None,
-                 app_cache: bool = True,
-                 checkpoint_files: Optional[Sequence[str]] = None,
-                 checkpoint_mode: Union[None,
-                                        Literal['task_exit'],
-                                        Literal['periodic'],
-                                        Literal['dfk_exit'],
-                                        Literal['manual']] = None,
-                 checkpoint_period: Optional[str] = None,
+                 memoizer: Optional[Memoizer] = None,
                  dependency_resolver: Optional[DependencyResolver] = None,
                  exit_mode: Literal['cleanup', 'skip', 'wait'] = 'cleanup',
                  garbage_collect: bool = True,
@@ -131,21 +114,7 @@ def __init__(self,
         self._executors: Sequence[ParslExecutor] = executors
         self._validate_executors()
 
-        self.app_cache = app_cache
-        self.checkpoint_files = checkpoint_files
-        self.checkpoint_mode = checkpoint_mode
-        if checkpoint_period is not None:
-            if checkpoint_mode is None:
-                logger.debug('The requested `checkpoint_period={}` will have no effect because `checkpoint_mode=None`'.format(
-                    checkpoint_period)
-                )
-            elif checkpoint_mode != 'periodic':
-                logger.debug("Requested checkpoint period of {} only has an effect with checkpoint_mode='periodic'".format(
-                    checkpoint_period)
-                )
-        if checkpoint_mode == 'periodic' and checkpoint_period is None:
-            checkpoint_period = "00:30:00"
-        self.checkpoint_period = checkpoint_period
+        self.memoizer = memoizer
         self.dependency_resolver = dependency_resolver
         self.exit_mode = exit_mode
         self.garbage_collect = garbage_collect

parsl/configs/ASPIRE1.py

@@ -1,5 +1,6 @@
 from parsl.addresses import address_by_interface
 from parsl.config import Config
+from parsl.dataflow.memoization import BasicMemoizer
 from parsl.executors import HighThroughputExecutor
 from parsl.launchers import MpiRunLauncher
 from parsl.monitoring.monitoring import MonitoringHub
@@ -38,7 +39,6 @@
         ),
         strategy='simple',
         retries=3,
-        app_cache=True,
-        checkpoint_mode='task_exit',
+        memoizer=BasicMemoizer(checkpoint_mode='task_exit'),
         usage_tracking=LEVEL_1,
 )

parsl/dataflow/dflow.py

@@ -165,15 +165,8 @@ def __init__(self, config: Config) -> None:
             self.monitoring_radio.send((MessageType.WORKFLOW_INFO,
                                        workflow_info))
 
-        # 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 = 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.memoizer: Memoizer = config.memoizer if config.memoizer is not None else BasicMemoizer()
+        self.memoizer.start(run_dir=self.run_dir)
 
         self._modify_checkpointable_tasks_lock = threading.Lock()
 

parsl/dataflow/memoization.py

@@ -125,6 +125,9 @@ def update_memo_exception(self, task: TaskRecord, e: BaseException) -> None:
     def update_memo_result(self, task: TaskRecord, r: Any) -> None:
         raise NotImplementedError
 
+    def start(self, *, run_dir: str) -> None:
+        raise NotImplementedError
+
     def checkpoint(self, *, task: Optional[TaskRecord] = None) -> None:
         raise NotImplementedError
 
@@ -169,18 +172,17 @@ class BasicMemoizer(Memoizer):
     run_dir: str
 
     def __init__(self, *,
-                 memoize: bool = True,
-                 checkpoint_files: Sequence[str] | None,
-                 checkpoint_period: Optional[str],
-                 checkpoint_mode: Literal['task_exit', 'periodic', 'dfk_exit', 'manual'] | None):
+                 checkpoint_files: Sequence[str] | None = None,
+                 checkpoint_period: Optional[str] = None,
+                 checkpoint_mode: Literal['task_exit', 'periodic', 'dfk_exit', 'manual'] | None = None,
+                 memoize: bool = True):  # TODO: unlikely to need to set this to false, but it was in config API before...
         """Initialize the memoizer.
 
         KWargs:
             - memoize (Bool): enable memoization or not.
             - checkpoint (Dict): A checkpoint loaded as a dict.
+            TODO: update
         """
-        self.memoize = memoize
-
         self.checkpointed_tasks = 0
 
         self.checkpoint_lock = threading.Lock()
@@ -192,8 +194,12 @@ def __init__(self, *,
         self.checkpointable_tasks: List[TaskRecord] = []
 
         self._checkpoint_timer: Timer | None = None
+        self.memoize = memoize
+
+    def start(self, *, run_dir: str) -> None:
+
+        self.run_dir = run_dir
 
-    def start(self) -> None:
         if self.checkpoint_files is not None:
             checkpoint_files = self.checkpoint_files
         elif self.checkpoint_files is None and self.checkpoint_mode is not None:

parsl/tests/configs/local_threads_checkpoint_dfk_exit.py

@@ -1,4 +1,5 @@
 from parsl.config import Config
+from parsl.dataflow.memoization import BasicMemoizer
 from parsl.executors.threads import ThreadPoolExecutor
 
 config = Config(
@@ -7,5 +8,5 @@
             label='local_threads_checkpoint_dfk_exit',
         )
     ],
-    checkpoint_mode='dfk_exit'
+    memoizer=BasicMemoizer(checkpoint_mode='dfk_exit')
 )

parsl/tests/configs/local_threads_checkpoint_task_exit.py

@@ -1,4 +1,5 @@
 from parsl.config import Config
+from parsl.dataflow.memoization import BasicMemoizer
 from parsl.executors.threads import ThreadPoolExecutor
 
 config = Config(
@@ -7,5 +8,5 @@
             label='local_threads_checkpoint_task_exit',
         )
     ],
-    checkpoint_mode='task_exit'
+    memoizer=BasicMemoizer(checkpoint_mode='task_exit')
 )

parsl/tests/test_checkpointing/test_periodic.py

@@ -3,15 +3,18 @@
 import parsl
 from parsl.app.app import python_app
 from parsl.config import Config
+from parsl.dataflow.memoization import BasicMemoizer
 from parsl.executors.threads import ThreadPoolExecutor
 
 
 def fresh_config():
     tpe = ThreadPoolExecutor(label='local_threads_checkpoint_periodic', max_threads=1)
     return Config(
         executors=[tpe],
-        checkpoint_mode='periodic',
-        checkpoint_period='00:00:02'
+        memoizer=BasicMemoizer(
+            checkpoint_mode='periodic',
+            checkpoint_period='00:00:02'
+        )
     )
 
 
@@ -32,7 +35,8 @@ def test_periodic():
     """Test checkpointing with task_periodic behavior
     """
     with parsl.load(fresh_config()):
-        h, m, s = map(int, parsl.dfk().config.checkpoint_period.split(":"))
+        memoizer = parsl.dfk().memoizer
+        h, m, s = map(int, memoizer.checkpoint_period.split(":"))
         assert h == 0, "Verify test setup"
         assert m == 0, "Verify test setup"
         assert s > 0, "Verify test setup"

parsl/tests/test_checkpointing/test_regression_233.py

@@ -1,12 +1,13 @@
 import pytest
 
 from parsl.app.app import python_app
+from parsl.config import Config
 from parsl.dataflow.dflow import DataFlowKernel
+from parsl.dataflow.memoization import BasicMemoizer
 
 
 def run_checkpointed(checkpoints):
-    from parsl.tests.configs.local_threads_checkpoint_task_exit import config
-    config.checkpoint_files = checkpoints
+    config = Config(memoizer=BasicMemoizer(checkpoint_files=checkpoints, checkpoint_mode='task_exit'))
     dfk = DataFlowKernel(config=config)
 
     @python_app(data_flow_kernel=dfk, cache=True)

parsl/tests/test_python_apps/test_memoize_2.py

@@ -5,6 +5,7 @@
 import parsl
 from parsl.app.app import python_app
 from parsl.config import Config
+from parsl.dataflow.memoization import BasicMemoizer
 from parsl.executors.threads import ThreadPoolExecutor
 
 
@@ -13,7 +14,7 @@ def local_config():
         executors=[
             ThreadPoolExecutor(max_threads=4),
         ],
-        app_cache=False
+        memoizer=BasicMemoizer(memoize=False)  # TODO: this should be a better do-nothing impl?
     )
 
 

parsl/tests/test_python_apps/test_memoize_plugin.py

@@ -0,0 +1,53 @@
+import argparse
+
+import pytest
+
+import parsl
+from parsl.app.app import python_app
+from parsl.config import Config
+from parsl.dataflow.memoization import BasicMemoizer
+from parsl.dataflow.taskrecord import TaskRecord
+
+
+class DontReuseSevenMemoizer(BasicMemoizer):
+    def check_memo(self, task_record: TaskRecord):
+        if task_record['args'][0] == 7:
+            return None  # we didn't find a suitable memo record...
+        else:
+            return super().check_memo(task_record)
+
+
+def local_config():
+    return Config(memoizer=DontReuseSevenMemoizer())
+
+
+@python_app(cache=True)
+def random_uuid(x, cache=True):
+    import uuid
+    return str(uuid.uuid4())
+
+
+@pytest.mark.local
+def test_python_memoization(n=10):
+    """Testing python memoization disable
+    """
+
+    # TODO: this .result() needs to be here, not in the loop
+    # because otherwise we race to complete... and then
+    # we might sometimes get a memoization before the loop
+    # and sometimes not...
+    x = random_uuid(0).result()
+
+    for i in range(0, n):
+        foo = random_uuid(0)
+        print(i)
+        print(foo.result())
+        assert foo.result() == x, "Memoized results were incorrectly not used"
+
+    y = random_uuid(7).result()
+
+    for i in range(0, n):
+        foo = random_uuid(7)
+        print(i)
+        print(foo.result())
+        assert foo.result() != y, "Memoized results were incorrectly used"