reset checkpoint task list after checkpointing all tasks, in memoizer

this was done in one occurence in the DFK.

in the other occurence, this happened at cleanup when no further
checkpointing would happen in the relevant modes - so this change
should not change user-facing behaviour

Author: benclifford

parsl/dataflow/dflow.py

@@ -1270,7 +1270,6 @@ def cleanup(self) -> None:
     def checkpoint(self) -> None:
         with self._modify_checkpointable_tasks_lock:
             self.memoizer.checkpoint()
-            self.memoizer.checkpointable_tasks = []
 
     @staticmethod
     def _log_std_streams(task_record: TaskRecord) -> None:

parsl/dataflow/memoization.py

@@ -411,3 +411,10 @@ def checkpoint(self, *, task: Optional[TaskRecord] = None) -> None:
                     logger.debug("No tasks checkpointed in this pass.")
             else:
                 logger.info("Done checkpointing {} tasks".format(count))
+
+            # TODO: test that tasks are not repeatedly checkpointed.
+            # TODO: race condition here, for tasks that were added while the list was being updated
+            # That's maybe a race condition in checkpointing in master at time of writing, and I could write a test for that?
+            # Or maybe a failure of iteration if the list is appended to while checkpointing is happening?
+            if not task:
+                self.checkpointable_tasks = []