Don't create checkpoint in multiprocessing case if the job failed (#692)

* don't create checkpoint in multiprocessing case if the job failed

* update changelog

Author: philippotto

cluster_tools/Changelog.md

@@ -16,7 +16,7 @@ For upgrade instructions, please check the respective *Breaking Changes* section
 ### Changed
 
 ### Fixed
-
+- Fixed that the ProcessPoolExecutor by the cluster tools would also create a checkpoint if the job failed. This was a regression introduced by #686. [#692](https://github.com/scalableminds/webknossos-libs/pull/692)
 
 ## [0.9.18](https://github.com/scalableminds/webknossos-libs/releases/tag/v0.9.18) - 2022-04-06
 [Commits](https://github.com/scalableminds/webknossos-libs/compare/v0.9.17...v0.9.18)

cluster_tools/cluster_tools/__init__.py

@@ -168,10 +168,16 @@ def _execute_and_persist_function(output_pickle_path, *args, **kwargs):
             result = False, exc
             logging.warning(f"Job computation failed with:\n{exc.__repr__()}")
 
-        with open(output_pickle_path, "wb") as file:
-            pickling.dump(result, file)
-
         if result[0]:
+            # Only pickle the result in the success case, since the output
+            # is used as a checkpoint.
+            # Note that this behavior differs a bit from the cluster executor
+            # which will always serialize the output (even exceptions) to
+            # disk. However, the output will have a .preliminary prefix at first
+            # which is only removed in the success case so that a checkpoint at
+            # the desired target only exists if the job was successful.
+            with open(output_pickle_path, "wb") as file:
+                pickling.dump(result, file)
             return result[1]
         else:
             raise result[1]

cluster_tools/cluster_tools/schedulers/cluster_executor.py

@@ -224,13 +224,14 @@ def _completion(self, jobid, failed_early):
 
         if success:
             # Remove the .preliminary postfix since the job was finished
-            # successfully. # Therefore, the result can be used as a checkpoint
+            # successfully. Therefore, the result can be used as a checkpoint
             # by users of the clustertools.
             os.rename(preliminary_outfile_name, outfile_name)
             logging.debug("Pickle file renamed to {}.".format(outfile_name))
 
             fut.set_result(result)
         else:
+            # Don't remove the .preliminary postfix since the job failed.
             fut.set_exception(RemoteException(result, jobid))
 
         # Clean up communication files.