Smooth transient social-agent aggregate dropouts

Author: thiennamdinh

compose/social-agent-network/aggregate_results.py

@@ -20,19 +20,34 @@
 DEFAULT_HOST = "0.0.0.0"
 DEFAULT_PORT = 8090
 DEFAULT_THROUGHPUT_WINDOW_SECONDS = 8.0
+DEFAULT_SNAPSHOT_STALE_SECONDS = 5.0
 
 
-def load_agent_snapshots():
+def load_agent_snapshots(cached=None, now_epoch=None, stale_after_seconds=DEFAULT_SNAPSHOT_STALE_SECONDS):
+    cached = {} if cached is None else cached
+    now_epoch = time.time() if now_epoch is None else now_epoch
     snapshots = []
+    next_cached = {}
     for path in sorted(RESULTS_DIR.glob(INPUT_PATTERN)):
         try:
             with path.open(encoding="utf-8") as fd:
-                snapshots.append(json.load(fd))
+                snapshot = json.load(fd)
+                snapshots.append(snapshot)
+                next_cached[str(path)] = {
+                    "snapshot": snapshot,
+                    "loaded_at": now_epoch,
+                }
         except (FileNotFoundError, json.JSONDecodeError):
             # Writers replace these files atomically, but readers can still
             # catch a transient missing/partial state between polls.
-            continue
-    return snapshots
+            cached_entry = cached.get(str(path))
+            if (
+                cached_entry is not None
+                and (now_epoch - cached_entry["loaded_at"]) <= stale_after_seconds
+            ):
+                snapshots.append(cached_entry["snapshot"])
+                next_cached[str(path)] = cached_entry
+    return snapshots, next_cached
 
 
 def aggregate_snapshots(snapshots, now_epoch, previous=None):
@@ -407,13 +422,19 @@ def log_message(self, format, *args):
 def run(interval_seconds, once, host, port, history_limit):
     state = AggregationState(history_limit=history_limit)
     previous = None
+    cached_snapshots = {}
 
     def aggregate_forever():
-        nonlocal previous
+        nonlocal previous, cached_snapshots
         while True:
             now_epoch = time.time()
+            snapshots, cached_snapshots = load_agent_snapshots(
+                cached=cached_snapshots,
+                now_epoch=now_epoch,
+                stale_after_seconds=max(DEFAULT_SNAPSHOT_STALE_SECONDS, interval_seconds * 2),
+            )
             snapshot = aggregate_snapshots(
-                load_agent_snapshots(), now_epoch, previous=previous
+                snapshots, now_epoch, previous=previous
             )
             state.update(snapshot)
             previous = snapshot