Remove quasi-monitoring checkpoint file dfk.pkl (#3760)

This file is saved alongside checkpoints, but stores an extremely light
summary of DFK state rather than of checkpoints: the DFK's task count
and run directory path.

That is information that falls into the space of monitoring, not of
checkpointing - so this PR removes that file, in favour of the much
richer monitoring database.


# Changed Behaviour

Checkpoints will no longer contain the light monitoring data that was
previously in `dfk.pkl`. Users will have to get that information some
other way (for example, by looking at the location of the checkpoint
directory to discover the run directory, and by using Parsl's monitoring
facility for task counts)

## Type of change

- Code maintenance/cleanup

parsl/dataflow/dflow.py

@@ -1263,7 +1263,7 @@ def checkpoint(self, tasks: Optional[Sequence[TaskRecord]] = None) -> str:
         Returns:
             Checkpoint dir if checkpoints were written successfully.
             By default the checkpoints are written to the RUNDIR of the current
-            run under RUNDIR/checkpoints/{tasks.pkl, dfk.pkl}
+            run under RUNDIR/checkpoints/tasks.pkl
         """
         with self.checkpoint_lock:
             if tasks:
@@ -1273,18 +1273,11 @@ def checkpoint(self, tasks: Optional[Sequence[TaskRecord]] = None) -> str:
                 self.checkpointable_tasks = []
 
             checkpoint_dir = '{0}/checkpoint'.format(self.run_dir)
-            checkpoint_dfk = checkpoint_dir + '/dfk.pkl'
             checkpoint_tasks = checkpoint_dir + '/tasks.pkl'
 
             if not os.path.exists(checkpoint_dir):
                 os.makedirs(checkpoint_dir, exist_ok=True)
 
-            with open(checkpoint_dfk, 'wb') as f:
-                state = {'rundir': self.run_dir,
-                         'task_count': self.task_count
-                         }
-                pickle.dump(state, f)
-
             count = 0
 
             with open(checkpoint_tasks, 'ab') as f:

parsl/tests/test_checkpointing/test_python_checkpoint_1.py

@@ -27,8 +27,5 @@ def test_initial_checkpoint_write():
 
     cpt_dir = parsl.dfk().checkpoint()
 
-    cptpath = cpt_dir + '/dfk.pkl'
-    assert os.path.exists(cptpath), f"DFK checkpoint missing: {cptpath}"
-
     cptpath = cpt_dir + '/tasks.pkl'
     assert os.path.exists(cptpath), f"Tasks checkpoint missing: {cptpath}"