Merge 144c9e9 into 1576295

daliuge-common/dlg/common/reproducibility/reprodata_compare.py

@@ -153,6 +153,8 @@ def write_outfile(data, outfilepath, outfilesuffix="summary", verbose=False):
         writer.writerow(fieldnames)
 
         for filepath, signature_data in data.items():
+            if signature_data == {}:
+                continue
             row = [filepath] + [signature_data[rmode.value] for rmode in ALL_RMODES]
             writer.writerow(row)
             if verbose:

daliuge-common/dlg/common/reproducibility/reproducibility.py

@@ -92,14 +92,13 @@ def accumulate_lg_drop_data(drop: dict, level: ReproducibilityFlags):
         raise NotImplementedError(
             f"Reproducibility level {level.name} not yet supported"
         )
-    category_type = drop.get(
-        "categoryType", ""
-    )  # Made conditional to support older graphs
     category = drop.get("category", "")
 
     # Cheeky way to get field list into dicts. map(dict, drop...) makes a copy
-    fields = {e.pop("name"): e["value"] for e in map(dict, drop["fields"])}
-    lg_fields = lg_block_fields(category_type, category, level)
+    fields = {e.pop("name"): e["value"] for e in map(dict, drop.get("fields", {}))}
+    app_fields = {e.pop("name"): e["value"] for e in map(dict, drop.get("applicationArgs", {}))}
+    fields.update(app_fields)
+    lg_fields = lg_block_fields(category, level, app_fields.keys())
     data = extract_fields(fields, lg_fields)
     return data
 

daliuge-common/dlg/common/reproducibility/reproducibility_fields.py

@@ -85,34 +85,29 @@ def lgt_block_fields(rmode: ReproducibilityFlags):
     return data
 
 
-def lg_block_fields(
-    category: Categories, category_type: str, rmode: ReproducibilityFlags
-):
+def lg_block_fields(category_type: str, rmode: ReproducibilityFlags, custom_fields=None):
     """
     Collects dict of fields and operations for all drop types at the lg layer for
     the supplied reproducibility standard.
     :param category: The broad type of drop
-    :param category_type: The specific type of drop
     :param rmode: The reproducibility level in question
+    :param custom_fields: Additional application args (used in custom components)
     :return: Dictionary of <str, FieldOp> pairs
     """
     data = {}
     if rmode in (
-        ReproducibilityFlags.NOTHING,
-        ReproducibilityFlags.RERUN,
-        ReproducibilityFlags.REPRODUCE,
-        ReproducibilityFlags.REPLICATE_SCI,
+            ReproducibilityFlags.NOTHING,
+            ReproducibilityFlags.RERUN,
+            ReproducibilityFlags.REPRODUCE,
+            ReproducibilityFlags.REPLICATE_SCI,
     ):
         return data
-    # Drop category considerations
-    if category == "Application":
-        data["execution_time"] = FieldOps.STORE
-        data["num_cpus"] = FieldOps.STORE
-    elif category == "Group":
-        data["inputApplicationName"] = FieldOps.STORE
-        data["inputApplicationType"] = FieldOps.STORE
-    elif category == Categories.DATA:  # An anomaly, I know
-        data["data_volume"] = FieldOps.STORE
+    # Drop category considerations - Just try to get everything we can, will be filtered later
+    data["execution_time"] = FieldOps.STORE
+    data["num_cpus"] = FieldOps.STORE
+    data["inputApplicationName"] = FieldOps.STORE
+    data["inputApplicationType"] = FieldOps.STORE
+    data["data_volume"] = FieldOps.STORE
 
     # Drop type considerations
     if category_type == Categories.START:
@@ -126,8 +121,8 @@ def lg_block_fields(
     elif category_type == Categories.FILE:
         data["check_filepath_exists"] = FieldOps.STORE
         if rmode in (
-            ReproducibilityFlags.RECOMPUTE,
-            ReproducibilityFlags.REPLICATE_COMP,
+                ReproducibilityFlags.RECOMPUTE,
+                ReproducibilityFlags.REPLICATE_COMP,
         ):
             data["filepath"] = FieldOps.STORE
             data["dirname"] = FieldOps.STORE
@@ -188,6 +183,10 @@ def lg_block_fields(
         data["libpath"] = FieldOps.STORE
     elif category_type == Categories.DYNLIB_PROC_APP:
         data["libpath"] = FieldOps.STORE
+    if custom_fields is not None and rmode in (
+            ReproducibilityFlags.RECOMPUTE, ReproducibilityFlags.REPLICATE_COMP):
+        for name in custom_fields:
+            data[name] = FieldOps.STORE
     return data
 
 

daliuge-engine/dlg/drop.py

@@ -1141,12 +1141,14 @@ def cancel(self):
         if self.status in [DROPStates.INITIALIZED, DROPStates.WRITING]:
             self._closeWriters()
             self.status = DROPStates.CANCELLED
+        self.completedrop()
 
     def skip(self):
         """Moves this drop to the SKIPPED state closing any writers we opened"""
         if self.status in [DROPStates.INITIALIZED, DROPStates.WRITING]:
             self._closeWriters()
             self.status = DROPStates.SKIPPED
+        self.completedrop()
 
     @property
     def node(self):
@@ -1867,7 +1869,11 @@ def dataURL(self) -> str:
     def generate_reproduce_data(self):
         from .droputils import allDropContents
 
-        data = allDropContents(self, self.size)
+        data = b""
+        try:
+            data = allDropContents(self, self.size)
+        except Exception:
+            logger.debug("Could not read drop reproduce data")
         return {"data_hash": common_hash(data)}