naively move checkpointable tasks list into memoizer, with accesses still happening from the DFK

Author: benclifford

parsl/dataflow/dflow.py

@@ -175,7 +175,6 @@ def __init__(self, config: Config) -> None:
         self._checkpoint_timer = None
         self.checkpoint_mode = config.checkpoint_mode
         self._modify_checkpointable_tasks_lock = threading.Lock()
-        self.checkpointable_tasks: List[TaskRecord] = []
 
         # this must be set before executors are added since add_executors calls
         # job_status_poller.add_executors.
@@ -568,7 +567,7 @@ def handle_app_update(self, task_record: TaskRecord, future: AppFuture) -> None:
             self.memoizer.checkpoint(tasks=[task_record])
         elif self.checkpoint_mode in ('manual', 'periodic', 'dfk_exit'):
             with self._modify_checkpointable_tasks_lock:
-                self.checkpointable_tasks.append(task_record)
+                self.memoizer.checkpointable_tasks.append(task_record)
         elif self.checkpoint_mode is None:
             pass
         else:
@@ -1206,7 +1205,7 @@ def cleanup(self) -> None:
 
             # TODO: accesses to self.checkpointable_tasks should happen
             # under a lock?
-            self.memoizer.checkpoint(self.checkpointable_tasks)
+            self.memoizer.checkpoint(self.memoizer.checkpointable_tasks)
 
             if self._checkpoint_timer:
                 logger.info("Stopping checkpoint timer")
@@ -1270,8 +1269,8 @@ def cleanup(self) -> None:
 
     def checkpoint(self) -> None:
         with self._modify_checkpointable_tasks_lock:
-            self.memoizer.checkpoint(self.checkpointable_tasks)
-            self.checkpointable_tasks = []
+            self.memoizer.checkpoint(self.memoizer.checkpointable_tasks)
+            self.memoizer.checkpointable_tasks = []
 
     @staticmethod
     def _log_std_streams(task_record: TaskRecord) -> None:

parsl/dataflow/memoization.py

@@ -169,6 +169,8 @@ def __init__(self, *,
         self.checkpoint_files = checkpoint_files
         self.checkpoint_mode = checkpoint_mode
 
+        self.checkpointable_tasks: List[TaskRecord] = []
+
     def start(self) -> None:
         if self.checkpoint_files is not None:
             checkpoint_files = self.checkpoint_files