From 697ee2b4cbbc79be8fde27bcfe197deb704e0167 Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:26:26 +0100 Subject: [PATCH] DPL: add support to retrieve smaps of running processes --- .../scripts/dpl-mcp-server/dpl_mcp_server.py | 70 +++++++++++++++++++ Framework/Core/src/StatusWebSocketHandler.cxx | 64 +++++++++++++++++ Framework/Core/src/StatusWebSocketHandler.h | 8 +++ 3 files changed, 142 insertions(+) diff --git a/Framework/Core/scripts/dpl-mcp-server/dpl_mcp_server.py b/Framework/Core/scripts/dpl-mcp-server/dpl_mcp_server.py index ed457b8a57d9d..95a7866dd2962 100644 --- a/Framework/Core/scripts/dpl-mcp-server/dpl_mcp_server.py +++ b/Framework/Core/scripts/dpl-mcp-server/dpl_mcp_server.py @@ -27,12 +27,14 @@ {"cmd":"list_metrics","device":""} {"cmd":"subscribe","device":"","metrics":["m1","m2"]} {"cmd":"unsubscribe","device":"","metrics":["m1"]} + {"cmd":"smaps","device":""} Wire protocol (driver -> client) -------------------------------- {"type":"snapshot","devices":[{"name","pid","active","streamingState","deviceState"},...]} {"type":"update","device":,"name":"","metrics":{}} {"type":"metrics_list","device":"","metrics":["m1","m2",...]} + {"type":"smaps","device":"","pid":,"fields":[...],"mappings":{"":[,...]}} """ 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//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. diff --git a/Framework/Core/src/StatusWebSocketHandler.cxx b/Framework/Core/src/StatusWebSocketHandler.cxx index 065a6f4b05b4a..ef5f18652e994 100644 --- a/Framework/Core/src/StatusWebSocketHandler.cxx +++ b/Framework/Core/src/StatusWebSocketHandler.cxx @@ -24,9 +24,13 @@ #include "Framework/LogParsingHelpers.h" #include "Framework/Signpost.h" #include +#include #include +#include +#include #include #include +#include #include 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 fields{"Rss:", "Pss:", "Shared_Clean:", "Shared_Dirty:", "Private_Clean:", "Private_Dirty:", "Swap:"}; + std::unordered_map> usage; + std::array* 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) { diff --git a/Framework/Core/src/StatusWebSocketHandler.h b/Framework/Core/src/StatusWebSocketHandler.h index fb2f0beebbec2..cd78b6402d053 100644 --- a/Framework/Core/src/StatusWebSocketHandler.h +++ b/Framework/Core/src/StatusWebSocketHandler.h @@ -48,6 +48,9 @@ struct WSDPLHandler; /// {"cmd":"disable_signpost","device":"","streams":["device","completion",...]} /// → disable the named signpost log streams for a device /// +/// {"cmd":"smaps","device":""} +/// → driver reads /proc//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":"","metrics":["m1","m2",...]} /// → reply to list_metrics command /// +/// {"type":"smaps","device":"","pid":,"fields":["rss","pss",...],"mappings":{"":[,...],...}} +/// → reply to smaps: kB per field, summed over all mappings of a path ("[anon]" for unnamed ones), +/// or {"type":"smaps","device":"","pid":,"error":""} where /proc is unavailable +/// /// {"type":"log","device":"","level":"","line":""} /// → 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;