DPL: add support to retrieve smaps of running processes

Framework/Core/scripts/dpl-mcp-server/dpl_mcp_server.py

@@ -27,12 +27,14 @@
     {"cmd":"list_metrics","device":"<name>"}
     {"cmd":"subscribe","device":"<name>","metrics":["m1","m2"]}
     {"cmd":"unsubscribe","device":"<name>","metrics":["m1"]}
+    {"cmd":"smaps","device":"<name>"}
 
 Wire protocol (driver -> client)
 --------------------------------
     {"type":"snapshot","devices":[{"name","pid","active","streamingState","deviceState"},...]}
     {"type":"update","device":<idx>,"name":"<name>","metrics":{<name:value,...>}}
     {"type":"metrics_list","device":"<name>","metrics":["m1","m2",...]}
+    {"type":"smaps","device":"<name>","pid":<pid>,"fields":[...],"mappings":{"<path>":[<kB>,...]}}
 """
 
 from __future__ import annotations
@@ -63,6 +65,7 @@ def __init__(self, *, url: str, name: str, extra_headers: dict[str, str] | None
         self.updates: list[dict] = []
         self.logs: list[dict] = []
         self.metrics_lists: dict[str, list[str]] = {}
+        self.smaps: dict[str, dict] = {}
 
     async def ensure_connected(self) -> None:
         """Connect (or reconnect) to the driver's /status WebSocket."""
@@ -113,6 +116,8 @@ async def _reader(self) -> None:
                 elif t == "metrics_list":
                     device = msg.get("device", "")
                     self.metrics_lists[device] = msg.get("metrics", [])
+                elif t == "smaps":
+                    self.smaps[msg.get("device", "")] = msg
         except Exception:
             pass
         finally:
@@ -315,6 +320,71 @@ async def list_metrics(workflow: str, device: str) -> str:
     return f"No reply from driver for device '{device}' (timeout)."
 
 
+# Where a mapping's memory is attributed, first match wins (O2Physics before O2,
+# since both paths contain "/O2").
+_SMAPS_GROUPS = (
+    ("O2Physics libs", lambda p: "/O2Physics/" in p),
+    ("O2 libs", lambda p: "/O2/" in p),
+    ("ROOT libs", lambda p: "/ROOT/" in p),
+    ("arrow libs", lambda p: "/arrow/" in p),
+    ("other libs", lambda p: ".so" in p),
+    ("heap", lambda p: p == "[heap]"),
+    ("anon", lambda p: p == "[anon]"),
+    ("shm", lambda p: p.startswith("/dev/shm") or "SYSV" in p or "fmq_" in p),
+    ("other", lambda p: True),
+)
+
+
+@mcp.tool()
+async def smaps(workflow: str, device: str, top: int = 25) -> str:
+    """Memory of a DPL device per mapping, read from /proc/<pid>/smaps by the driver.
+
+    The driver runs as the same user as its devices, so this works where an
+    outside reader cannot (e.g. Hyperloop tests running as another account).
+    Reports PSS / private-dirty / RSS per group (O2Physics, O2, ROOT, arrow and
+    other libs, heap, anon, shm) and the ``top`` paths by PSS. Private-dirty is
+    the device's own cost, independent of how many processes share a library;
+    PSS splits shared pages between them. Linux only.
+
+    Args:
+        workflow: Workflow name as returned by connect.
+        device: Device name exactly as shown by list_devices.
+        top: Number of individual paths to list.
+    """
+    conn = _get(workflow)
+    conn.smaps.pop(device, None)
+    await conn.send({"cmd": "smaps", "device": device})
+    for _ in range(100):         # up to 5 s
+        await asyncio.sleep(0.05)
+        if device in conn.smaps:
+            break
+    else:
+        return f"No reply from driver for device '{device}' (timeout, or a driver without the smaps command)."
+    msg = conn.smaps[device]
+    if "error" in msg:
+        return f"{device} (pid {msg.get('pid')}): {msg['error']}"
+    idx = {f: i for i, f in enumerate(msg["fields"])}
+    rows = sorted(((p, v[idx["pss"]], v[idx["private_dirty"]], v[idx["rss"]]) for p, v in msg["mappings"].items()),
+                  key=lambda r: -r[1])
+    groups = {name: [0, 0, 0] for name, _ in _SMAPS_GROUPS}
+    for path, pss, pd, rss in rows:
+        g = groups[next(n for n, match in _SMAPS_GROUPS if match(path))]
+        g[0] += pss
+        g[1] += pd
+        g[2] += rss
+    mb = lambda kb: f"{kb / 1024:9.1f}"
+    total = [sum(g[i] for g in groups.values()) for i in range(3)]
+    lines = [f"{device} (pid {msg['pid']})", "", f"{'group':16} {'PSS MB':>9} {'privDirty':>9} {'RSS MB':>9}"]
+    for name, (pss, pd, rss) in sorted(groups.items(), key=lambda kv: -kv[1][0]):
+        if rss:
+            lines.append(f"{name:16} {mb(pss)} {mb(pd)} {mb(rss)}")
+    lines.append(f"{'TOTAL':16} {mb(total[0])} {mb(total[1])} {mb(total[2])}")
+    lines += ["", f"top {top} paths by PSS:", f"{'PSS MB':>9} {'privDirty':>9} {'RSS MB':>9}  path"]
+    for path, pss, pd, rss in rows[:max(1, top)]:
+        lines.append(f"{mb(pss)} {mb(pd)} {mb(rss)}  {path}")
+    return "\n".join(lines)
+
+
 @mcp.tool()
 async def subscribe(workflow: str, device: str, metrics: list[str]) -> str:
     """Subscribe to one or more metrics for a DPL device.

Framework/Core/src/StatusWebSocketHandler.cxx

@@ -24,9 +24,13 @@
 #include "Framework/LogParsingHelpers.h"
 #include "Framework/Signpost.h"
 #include <algorithm>
+#include <array>
 #include <cstdio>
+#include <fstream>
+#include <sstream>
 #include <string>
 #include <string_view>
+#include <unordered_map>
 #include <vector>
 
 namespace o2::framework
@@ -262,6 +266,8 @@ void StatusWebSocketHandler::frame(char const* data, size_t s)
     handleEnableSignpost(deviceName, extractArrayField(msg, "streams"));
   } else if (cmd == "disable_signpost") {
     handleDisableSignpost(deviceName, extractArrayField(msg, "streams"));
+  } else if (cmd == "smaps") {
+    handleSmaps(deviceName);
   }
 }
 
@@ -445,6 +451,64 @@ size_t StatusWebSocketHandler::findDeviceIndex(std::string_view name) const
   return SIZE_MAX;
 }
 
+void StatusWebSocketHandler::handleSmaps(std::string_view deviceName)
+{
+  size_t di = findDeviceIndex(deviceName);
+  if (di == SIZE_MAX) {
+    return;
+  }
+  auto pid = (*mContext.infos)[di].pid;
+  std::string out = R"({"type":"smaps","device":")";
+  out += jsonEscape(deviceName);
+  out += R"(","pid":)";
+  out += std::to_string(pid);
+
+  std::ifstream smaps("/proc/" + std::to_string(pid) + "/smaps");
+  if (!smaps) {
+    out += R"(,"error":"cannot read /proc/)" + std::to_string(pid) + R"(/smaps"})";
+    sendText(out);
+    return;
+  }
+  constexpr std::array<std::string_view, 7> fields{"Rss:", "Pss:", "Shared_Clean:", "Shared_Dirty:", "Private_Clean:", "Private_Dirty:", "Swap:"};
+  std::unordered_map<std::string, std::array<uint64_t, fields.size()>> usage;
+  std::array<uint64_t, fields.size()>* current = nullptr;
+  std::string line;
+  while (std::getline(smaps, line)) {
+    std::istringstream tokens(line);
+    std::string first;
+    tokens >> first;
+    if (first.empty()) {
+      continue;
+    }
+    if (first.back() != ':') {
+      std::string perms, offset, dev, inode, name;
+      tokens >> perms >> offset >> dev >> inode;
+      std::getline(tokens >> std::ws, name);
+      current = &usage[name.empty() ? "[anon]" : name];
+      continue;
+    }
+    auto field = std::find(fields.begin(), fields.end(), first);
+    uint64_t kb = 0;
+    if (current && field != fields.end() && tokens >> kb) {
+      (*current)[field - fields.begin()] += kb;
+    }
+  }
+  out += R"(,"fields":["rss","pss","shared_clean","shared_dirty","private_clean","private_dirty","swap"],"mappings":{)";
+  bool first = true;
+  for (auto const& [name, kbs] : usage) {
+    out += first ? "\"" : ",\"";
+    first = false;
+    out += jsonEscape(name);
+    out += "\":[";
+    for (size_t fi = 0; fi < kbs.size(); ++fi) {
+      out += (fi ? "," : "") + std::to_string(kbs[fi]);
+    }
+    out += ']';
+  }
+  out += "}}";
+  sendText(out);
+}
+
 void StatusWebSocketHandler::handleStartDevices()
 {
   for (auto const& info : *mContext.infos) {

Framework/Core/src/StatusWebSocketHandler.h

@@ -48,6 +48,9 @@ struct WSDPLHandler;
 ///   {"cmd":"disable_signpost","device":"<name>","streams":["device","completion",...]}
 ///     → disable the named signpost log streams for a device
 ///
+///   {"cmd":"smaps","device":"<name>"}
+///     → driver reads /proc/<pid>/smaps of the device and replies with its usage per mapping
+///
 ///   {"cmd":"list_signposts"}
 ///     → driver replies with {"type":"signposts_list","streams":["device","completion",...]}
 ///     → lists the known stream names
@@ -62,6 +65,10 @@ struct WSDPLHandler;
 ///   {"type":"metrics_list","device":"<name>","metrics":["m1","m2",...]}
 ///     → reply to list_metrics command
 ///
+///   {"type":"smaps","device":"<name>","pid":<pid>,"fields":["rss","pss",...],"mappings":{"<path>":[<kB>,...],...}}
+///     → reply to smaps: kB per field, summed over all mappings of a path ("[anon]" for unnamed ones),
+///       or {"type":"smaps","device":"<name>","pid":<pid>,"error":"<text>"} where /proc is unavailable
+///
 ///   {"type":"log","device":"<name>","level":"<level>","line":"<text>"}
 ///     → pushed for each new log line from a subscribed device
 struct StatusWebSocketHandler : public WebSocketHandler {
@@ -96,6 +103,7 @@ struct StatusWebSocketHandler : public WebSocketHandler {
   void handleSubscribeLogs(std::string_view deviceName);
   void handleUnsubscribeLogs(std::string_view deviceName);
   void handleStartDevices();
+  void handleSmaps(std::string_view deviceName);
   void handleEnableSignpost(std::string_view deviceName, std::string_view streamsArr);
   void handleDisableSignpost(std::string_view deviceName, std::string_view streamsArr);
   size_t findDeviceIndex(std::string_view name) const;