fix data-store memory problem

[dwsutherland]

closes #7199

before

after

Check List

• I have read CONTRIBUTING.md and added my name as a Code Contributor.
• Contains logically grouped changes (else tidy your branch by rebase).
• Does not contain off-topic changes (use other PRs for other changes).
• Applied any dependency changes to both setup.cfg (and conda-environment.yml if present).
• Code covered by existing tests.
• Changelog entry included if this is a change that can affect users
• Cylc-Doc pull request opened if required at cylc/cylc-doc/pull/XXXX.
• If this is a bug fix, PR should be raised against the relevant ?.?.x branch.

[oliver-sanders]

Cheers, will have a go at running with the original example.

Did you have to apply some diff to the profiler plugin to reveal the issue?

[dwsutherland]

Did you have to apply some diff to the profiler plugin to reveal the issue?

Not really, because they were very specific so I didn't think of a general use case.

Sorry, yes I had to change up the profiler plugin:

sutherlander@cortex-hyper:flow$ git diff main_loop/log_data_store.py
diff --git a/cylc/flow/main_loop/log_data_store.py b/cylc/flow/main_loop/log_data_store.py
index ec9fc8624..3d731d2fc 100644
--- a/cylc/flow/main_loop/log_data_store.py
+++ b/cylc/flow/main_loop/log_data_store.py
@@ -38,8 +38,21 @@ try:
 except ModuleNotFoundError:
     PLT = False
 
-from pympler.asizeof import asized
-
+from pympler.asizeof import asized, asizeof, Asizer
+
+#    'publish_deltas',
+#    'n_window_nodes',
+#    'n_window_edges',
+#    'n_window_node_walks',
+#    'n_window_completed_walks',
+#    'n_window_depths',
+#    'all_n_window_nodes',
+STORE_OTHER = {
+    'family_pruned_ids',
+    'prune_trigger_nodes',
+    'prune_flagged_nodes',
+    'pruned_task_proxies',
+}
 
 @startup
 async def init(scheduler, state):
@@ -51,19 +64,54 @@ async def init(scheduler, state):
         state['objects'][key] = []
         state['size'][key] = []
 
+    for attr_name in STORE_OTHER:
+        state['objects'][attr_name] = []
+        state['size'][attr_name] = []
+
+    #state['objects']['schd.config'] = []
+    #state['size']['schd.config'] = []
+
+    state['objects']['schd'] = []
+    state['size']['schd'] = []
+
+    state['objects']['dsmgr'] = []
+    state['size']['dsmgr'] = []
+
 
 @periodic
 async def log_data_store(scheduler, state):
     """Count the number of objects and the data store size."""
     state['times'].append(time())
-    for key, value in _iter_data_store(scheduler.data_store_mgr.data):
+    ds = scheduler.data_store_mgr
+    for key, value in _iter_data_store(ds.data):
         state['objects'][key].append(
             len(value)
         )
         state['size'][key].append(
-            asized(value).size
+            asizeof(value)
         )
 
+    for attr_name in STORE_OTHER:
+        attr_value = getattr(ds, attr_name)
+        state['objects'][attr_name].append(
+            len(attr_value)
+        )
+        state['size'][attr_name].append(
+            asizeof(attr_value)
+        )
+
+    #state['objects']['schd.config'].append(1)
+    #state['size']['schd.config'].append(asizeof(scheduler.config))
+    asizer = Asizer()
+    asizer.exclude_refs(scheduler.data_store_mgr)
+    state['objects']['schd'].append(1)
+    state['size']['schd'].append(asizer.asizeof(scheduler))
+
+    asizer = Asizer()
+    asizer.exclude_refs(scheduler)
+    state['objects']['dsmgr'].append(1)
+    state['size']['dsmgr'].append(asizer.asizeof(scheduler.data_store_mgr))
+
 
 @shutdown
 async def report(scheduler, state):
@@ -75,7 +123,9 @@ async def report(scheduler, state):
 def _iter_data_store(data_store):
     for item in data_store.values():
         for key, value in item.items():
-            if key != 'workflow':
+            if key == 'workflow':
+                yield (key, [value])
+            else:
                 yield (key, value)
         # there should only be one workflow in the data store
         break

But think it's probably too specific to make permanent..
Although I could make a cut-down version of it as a commit here.

[dwsutherland]

Have added a version of the plugin modifications used in leak detection:

No idea why some tests are failing (functionality hasn't changed).. Will look

Ah, it's a test of the memory profiling..

[dpmatthews]

I've tested this against my example workflow and against a less cut down version of the same workflow and can confirm this fixes the leak - thanks.

[oliver-sanders]

Profiling results look much better!

before:

after:

There's still quite an upward slope to prune_trigger_nodes and n_window_node_walks, but I'm not sure if that's a smaller leak. or just a natural increase due to the character of the workflow's graph.

---

Thanks for working on the plugin, it's a big help to get these diffs back into the project so we can reproduce the results. I've opened a PR to work on this a bit further (above graphs generated with this diff): dwsutherland#27

---

The code seems reasonable, but I'm having a bit of difficulty working out the purpose of the different attributes to understand when they should be housekept:

[dwsutherland]

There's still quite an upward slope

It tappers off if run longer:

[dwsutherland]

I've updated the plotting, to report on sets and also declutter by reporting on, in addition to the data, only attributes that are larger (any of their values) than 2% (by default) of the max size recorded:

Because it would get cluttered very quickly at a fixed 2kb.

The code seems reasonable, but I'm having a bit of difficulty working out the purpose of the different attributes to understand when they should be housekept

Well the module doc says:

Pruning of data-store elements is done using the collection/set of nodes
generated at the boundary of an active node's graph walk and registering active
node's parents against them. Once active, these boundary nodes act as the prune
triggers for the associated parent nodes. Set operations are used to do a diff
between the nodes of active paths (paths whose node is in n=0)
and the nodes of flagged paths (whose boundary node(s) have become active).

I've added a little more

This method is used to avoid "blinking", where a task becomes non-active then
is removed (along with it's window/walk) before a descendant is added, causing
it to disapear then reappear in the store (and, hence, UIs).

[dwsutherland]

to understand when they should be housekept

Also, I believe, the reason for prune_trigger_nodes building up in the first place is because graphs with paths not taken may have boundary nodes that are never reached (hence why the memory problem only presented in some workflows)... I'll comment to this effect above the fix.

[dwsutherland]

Here's the explanation for why the below fix was needed.

[oliver-sanders]

Makes sense!