LIU-368: Update expireAfterUse default behaviour

Files: expireAfterUse=False
Memory: expireAfterUse=True

This is irrespective of 'persist'

Author: myxie

daliuge-engine/dlg/data/drops/file.py

@@ -75,18 +75,15 @@ def __init__(self, *args, **kwargs):
         """
         Initialise default drop behaviour when it is completed with the following rules:
 
-            - "persist": Replicate and store the data in the specified persistent store
-            Files should be persistent by default.
+        - "expireAfterUse": Remove the data from the workspace once it has been used
+        by all consumers. This is independent of the "persist" flag. This is false
+       by default for FileDrops.
 
-            - "expireAfterUse": Remove the data from the workspace once it has been used
-            by all consumers. This is independent of the "persist" flag.
-
-            Use the default behaviour for expireAfterUse in AbstractDROP,
-            unless specified otherwise.
         """
 
-        if "persist" not in kwargs:
-            kwargs["persist"] = True
+        # 'lifespan' and 'expireAfterUse' are mutually exclusive
+        if "lifespan" not in kwargs and "expireAfterUse" not in kwargs:
+            kwargs["expireAfterUse"] = False
         self.is_dir = False
         super().__init__(*args, **kwargs)
 

daliuge-engine/dlg/drop.py

@@ -373,11 +373,6 @@ def __init__(self, oid, uid, **kwargs):
         # No DROP should be persisted unless stated otherwise; used for replication
         self._persist: bool = self._popArg(kwargs, "persist", False)
 
-        # Non-persistent drops should expire after they are used
-        # If expirationDate is set, the drops will expire so we do not need to update
-        if not self._persist and self._expirationDate == -1:
-            self._expireAfterUse = True
-
         # Useful to have access to all EAGLE parameters without a prior knowledge
         self._parameters = dict(kwargs)
         self.autofill_environment_variables()

daliuge-engine/test/lifecycle/test_dlm.py

@@ -35,6 +35,7 @@
 from dlg.apps.app_base import BarrierAppDROP
 from dlg.data.drops.directorycontainer import DirectoryContainer
 from dlg.data.drops.file import FileDROP
+from dlg.data.drops.memory import InMemoryDROP
 from dlg.droputils import DROPWaiterCtx
 from dlg.lifecycle import dlm
 
@@ -61,7 +62,15 @@ def test_dropAddition(self):
 
     def test_dropCompleteTriggersReplication(self):
         with dlm.DataLifecycleManager(enable_drop_replication=True) as manager:
-            drop = FileDROP("oid:A", "uid:A1", expectedSize=1)
+
+            # By default a file is non-persistent
+            drop = FileDROP("oid:B", "uid:B1", expectedSize=1)
+            manager.addDrop(drop)
+            self._writeAndClose(drop)
+            self.assertEqual(DROPPhases.GAS, drop.phase)
+            self.assertEqual(1, len(manager.getDropUids(drop)))
+
+            drop = FileDROP("oid:A", "uid:A1", expectedSize=1, persist=True)
             manager.addDrop(drop)
             self._writeAndClose(drop)
 
@@ -70,12 +79,6 @@ def test_dropCompleteTriggersReplication(self):
             self.assertEqual(DROPPhases.SOLID, drop.phase)
             self.assertEqual(2, len(manager.getDropUids(drop)))
 
-            # Try the same with a non-persisted data object, it shouldn't be replicated
-            drop = FileDROP("oid:B", "uid:B1", expectedSize=1, persist=False)
-            manager.addDrop(drop)
-            self._writeAndClose(drop)
-            self.assertEqual(DROPPhases.GAS, drop.phase)
-            self.assertEqual(1, len(manager.getDropUids(drop)))
 
     def test_expiringNormalDrop(self):
         with dlm.DataLifecycleManager(check_period=0.5) as manager:
@@ -130,55 +133,113 @@ def test_cleanupExpiredDrops(self):
             self.assertEqual(DROPStates.DELETED, drop.status)
             self.assertFalse(drop.exists())
 
-    def test_expireAfterUse(self):
+    def test_expireAfterUseForFile(self):
+        """
+        Test default and non-default behaviour for the expireAfterUse flag
+        for file drops.
+
+        Default: expireAfterUse=False, so the drop should still exist after it
+        has been consumed.
+        Non-default: expiredAfterUse=True, so the drop will be expired and
+        deleted after it is consumed.
+        """
+
+        class MyApp(BarrierAppDROP):
+            def run(self):
+                pass
+
+        with dlm.DataLifecycleManager(check_period=0.5, cleanup_period=2) as manager:
+
+            # Check default
+            default_fp, default_name = tempfile.mkstemp()
+            default = FileDROP(
+                "a",
+                "a",
+                filepath=default_name
+            )
+
+            expired_fp, expired_name = tempfile.mkstemp()
+            expired = FileDROP(
+                "b",
+                "b",
+                filepath=expired_name,
+                expireAfterUse=True  # Remove the file after use
+            )
+            c = MyApp("c", "c")
+            d = MyApp("d", "d")
+            default.addConsumer(c)
+            default.addConsumer(d)
+            expired.addConsumer(c)
+            expired.addConsumer(d)
+
+            manager.addDrop(default)
+            manager.addDrop(expired)
+            manager.addDrop(expired)
+            manager.addDrop(c)
+
+            # Make sure all consumers are done
+            with DROPWaiterCtx(self, [c, d], 1):
+                default.setCompleted()
+                expired.setCompleted()
+
+            # Both directories should be there, but after cleanup B's shouldn't
+            # be there anymore
+            self.assertTrue(default.exists())
+            self.assertTrue(expired.exists())
+            time.sleep(2.5)
+            self.assertTrue(default.exists())
+            self.assertFalse(expired.exists())
+            default.delete()
+
+    def test_expireAfterUseForMemory(self):
         """
-        Simple test for the expireAfterUse flag. Two DROPs are created with
-        different values, and after they are used we check whether their data
-        is still there or not
+        Default: expireAfterUse=True, so the drop should not exist after it
+        has been consumed.
+        Non-default: expiredAfterUse=False, so the drop will be not be expired
+        after it is consumed.
         """
 
         class MyApp(BarrierAppDROP):
             def run(self):
                 pass
 
         with dlm.DataLifecycleManager(check_period=0.5, cleanup_period=2) as manager:
-            a = DirectoryContainer(
+
+            # Check default behaviour - deleted for memory drops
+            default = InMemoryDROP(
                 "a",
                 "a",
-                persist=False,
-                expireAfterUse=False, # Marking persist as False should lead to expiry regardless of this flag
-                dirname=tempfile.mkdtemp(),
             )
-            b_dirname = tempfile.mkdtemp()
-            b = DirectoryContainer(
+
+            # Non-default behaviour - memory is not deleted
+            non_expired = InMemoryDROP(
                 "b",
                 "b",
-                persist=True,  # Persist should have no effect on expireAfterUse
-                expireAfterUse=False,
-                dirname=b_dirname,
+                expireAfterUse=False
             )
             c = MyApp("c", "c")
             d = MyApp("d", "d")
-            a.addConsumer(c)
-            a.addConsumer(d)
-            b.addConsumer(c)
-            b.addConsumer(d)
-
-            manager.addDrop(a)
-            manager.addDrop(b)
-            manager.addDrop(b)
+            default.addConsumer(c)
+            default.addConsumer(d)
+            non_expired.addConsumer(c)
+            non_expired.addConsumer(d)
+
+            manager.addDrop(default)
+            manager.addDrop(non_expired)
+            manager.addDrop(non_expired)
             manager.addDrop(c)
 
             # Make sure all consumers are done
             with DROPWaiterCtx(self, [c, d], 1):
-                a.setCompleted()
-                b.setCompleted()
+                default.setCompleted()
+                non_expired.setCompleted()
 
-            # Both directories should be there, but after cleanup A's shouldn't
+            # Both directories should be there, but after cleanup B's shouldn't
             # be there anymore
-            self.assertTrue(a.exists())
-            self.assertTrue(b.exists())
+            self.assertTrue(default.exists())
+            self.assertTrue(non_expired.exists())
             time.sleep(2.5)
-            self.assertFalse(a.exists())
-            self.assertTrue(b.exists())
-            b.delete()
+            self.assertFalse(default.exists())
+            self.assertTrue(non_expired.exists())
+            non_expired.delete()
+