Skip to content

Commit 545bf3b

Browse files
authored
DPL: add support to retrieve smaps of running processes (#16023)
1 parent 1bd4969 commit 545bf3b

3 files changed

Lines changed: 142 additions & 0 deletions

File tree

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

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,14 @@
2727
{"cmd":"list_metrics","device":"<name>"}
2828
{"cmd":"subscribe","device":"<name>","metrics":["m1","m2"]}
2929
{"cmd":"unsubscribe","device":"<name>","metrics":["m1"]}
30+
{"cmd":"smaps","device":"<name>"}
3031
3132
Wire protocol (driver -> client)
3233
--------------------------------
3334
{"type":"snapshot","devices":[{"name","pid","active","streamingState","deviceState"},...]}
3435
{"type":"update","device":<idx>,"name":"<name>","metrics":{<name:value,...>}}
3536
{"type":"metrics_list","device":"<name>","metrics":["m1","m2",...]}
37+
{"type":"smaps","device":"<name>","pid":<pid>,"fields":[...],"mappings":{"<path>":[<kB>,...]}}
3638
"""
3739

3840
from __future__ import annotations
@@ -63,6 +65,7 @@ def __init__(self, *, url: str, name: str, extra_headers: dict[str, str] | None
6365
self.updates: list[dict] = []
6466
self.logs: list[dict] = []
6567
self.metrics_lists: dict[str, list[str]] = {}
68+
self.smaps: dict[str, dict] = {}
6669

6770
async def ensure_connected(self) -> None:
6871
"""Connect (or reconnect) to the driver's /status WebSocket."""
@@ -113,6 +116,8 @@ async def _reader(self) -> None:
113116
elif t == "metrics_list":
114117
device = msg.get("device", "")
115118
self.metrics_lists[device] = msg.get("metrics", [])
119+
elif t == "smaps":
120+
self.smaps[msg.get("device", "")] = msg
116121
except Exception:
117122
pass
118123
finally:
@@ -315,6 +320,71 @@ async def list_metrics(workflow: str, device: str) -> str:
315320
return f"No reply from driver for device '{device}' (timeout)."
316321

317322

323+
# Where a mapping's memory is attributed, first match wins (O2Physics before O2,
324+
# since both paths contain "/O2").
325+
_SMAPS_GROUPS = (
326+
("O2Physics libs", lambda p: "/O2Physics/" in p),
327+
("O2 libs", lambda p: "/O2/" in p),
328+
("ROOT libs", lambda p: "/ROOT/" in p),
329+
("arrow libs", lambda p: "/arrow/" in p),
330+
("other libs", lambda p: ".so" in p),
331+
("heap", lambda p: p == "[heap]"),
332+
("anon", lambda p: p == "[anon]"),
333+
("shm", lambda p: p.startswith("/dev/shm") or "SYSV" in p or "fmq_" in p),
334+
("other", lambda p: True),
335+
)
336+
337+
338+
@mcp.tool()
339+
async def smaps(workflow: str, device: str, top: int = 25) -> str:
340+
"""Memory of a DPL device per mapping, read from /proc/<pid>/smaps by the driver.
341+
342+
The driver runs as the same user as its devices, so this works where an
343+
outside reader cannot (e.g. Hyperloop tests running as another account).
344+
Reports PSS / private-dirty / RSS per group (O2Physics, O2, ROOT, arrow and
345+
other libs, heap, anon, shm) and the ``top`` paths by PSS. Private-dirty is
346+
the device's own cost, independent of how many processes share a library;
347+
PSS splits shared pages between them. Linux only.
348+
349+
Args:
350+
workflow: Workflow name as returned by connect.
351+
device: Device name exactly as shown by list_devices.
352+
top: Number of individual paths to list.
353+
"""
354+
conn = _get(workflow)
355+
conn.smaps.pop(device, None)
356+
await conn.send({"cmd": "smaps", "device": device})
357+
for _ in range(100): # up to 5 s
358+
await asyncio.sleep(0.05)
359+
if device in conn.smaps:
360+
break
361+
else:
362+
return f"No reply from driver for device '{device}' (timeout, or a driver without the smaps command)."
363+
msg = conn.smaps[device]
364+
if "error" in msg:
365+
return f"{device} (pid {msg.get('pid')}): {msg['error']}"
366+
idx = {f: i for i, f in enumerate(msg["fields"])}
367+
rows = sorted(((p, v[idx["pss"]], v[idx["private_dirty"]], v[idx["rss"]]) for p, v in msg["mappings"].items()),
368+
key=lambda r: -r[1])
369+
groups = {name: [0, 0, 0] for name, _ in _SMAPS_GROUPS}
370+
for path, pss, pd, rss in rows:
371+
g = groups[next(n for n, match in _SMAPS_GROUPS if match(path))]
372+
g[0] += pss
373+
g[1] += pd
374+
g[2] += rss
375+
mb = lambda kb: f"{kb / 1024:9.1f}"
376+
total = [sum(g[i] for g in groups.values()) for i in range(3)]
377+
lines = [f"{device} (pid {msg['pid']})", "", f"{'group':16} {'PSS MB':>9} {'privDirty':>9} {'RSS MB':>9}"]
378+
for name, (pss, pd, rss) in sorted(groups.items(), key=lambda kv: -kv[1][0]):
379+
if rss:
380+
lines.append(f"{name:16} {mb(pss)} {mb(pd)} {mb(rss)}")
381+
lines.append(f"{'TOTAL':16} {mb(total[0])} {mb(total[1])} {mb(total[2])}")
382+
lines += ["", f"top {top} paths by PSS:", f"{'PSS MB':>9} {'privDirty':>9} {'RSS MB':>9} path"]
383+
for path, pss, pd, rss in rows[:max(1, top)]:
384+
lines.append(f"{mb(pss)} {mb(pd)} {mb(rss)} {path}")
385+
return "\n".join(lines)
386+
387+
318388
@mcp.tool()
319389
async def subscribe(workflow: str, device: str, metrics: list[str]) -> str:
320390
"""Subscribe to one or more metrics for a DPL device.

‎Framework/Core/src/StatusWebSocketHandler.cxx‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,13 @@
2424
#include "Framework/LogParsingHelpers.h"
2525
#include "Framework/Signpost.h"
2626
#include <algorithm>
27+
#include <array>
2728
#include <cstdio>
29+
#include <fstream>
30+
#include <sstream>
2831
#include <string>
2932
#include <string_view>
33+
#include <unordered_map>
3034
#include <vector>
3135

3236
namespace o2::framework
@@ -262,6 +266,8 @@ void StatusWebSocketHandler::frame(char const* data, size_t s)
262266
handleEnableSignpost(deviceName, extractArrayField(msg, "streams"));
263267
} else if (cmd == "disable_signpost") {
264268
handleDisableSignpost(deviceName, extractArrayField(msg, "streams"));
269+
} else if (cmd == "smaps") {
270+
handleSmaps(deviceName);
265271
}
266272
}
267273

@@ -445,6 +451,64 @@ size_t StatusWebSocketHandler::findDeviceIndex(std::string_view name) const
445451
return SIZE_MAX;
446452
}
447453

454+
void StatusWebSocketHandler::handleSmaps(std::string_view deviceName)
455+
{
456+
size_t di = findDeviceIndex(deviceName);
457+
if (di == SIZE_MAX) {
458+
return;
459+
}
460+
auto pid = (*mContext.infos)[di].pid;
461+
std::string out = R"({"type":"smaps","device":")";
462+
out += jsonEscape(deviceName);
463+
out += R"(","pid":)";
464+
out += std::to_string(pid);
465+
466+
std::ifstream smaps("/proc/" + std::to_string(pid) + "/smaps");
467+
if (!smaps) {
468+
out += R"(,"error":"cannot read /proc/)" + std::to_string(pid) + R"(/smaps"})";
469+
sendText(out);
470+
return;
471+
}
472+
constexpr std::array<std::string_view, 7> fields{"Rss:", "Pss:", "Shared_Clean:", "Shared_Dirty:", "Private_Clean:", "Private_Dirty:", "Swap:"};
473+
std::unordered_map<std::string, std::array<uint64_t, fields.size()>> usage;
474+
std::array<uint64_t, fields.size()>* current = nullptr;
475+
std::string line;
476+
while (std::getline(smaps, line)) {
477+
std::istringstream tokens(line);
478+
std::string first;
479+
tokens >> first;
480+
if (first.empty()) {
481+
continue;
482+
}
483+
if (first.back() != ':') {
484+
std::string perms, offset, dev, inode, name;
485+
tokens >> perms >> offset >> dev >> inode;
486+
std::getline(tokens >> std::ws, name);
487+
current = &usage[name.empty() ? "[anon]" : name];
488+
continue;
489+
}
490+
auto field = std::find(fields.begin(), fields.end(), first);
491+
uint64_t kb = 0;
492+
if (current && field != fields.end() && tokens >> kb) {
493+
(*current)[field - fields.begin()] += kb;
494+
}
495+
}
496+
out += R"(,"fields":["rss","pss","shared_clean","shared_dirty","private_clean","private_dirty","swap"],"mappings":{)";
497+
bool first = true;
498+
for (auto const& [name, kbs] : usage) {
499+
out += first ? "\"" : ",\"";
500+
first = false;
501+
out += jsonEscape(name);
502+
out += "\":[";
503+
for (size_t fi = 0; fi < kbs.size(); ++fi) {
504+
out += (fi ? "," : "") + std::to_string(kbs[fi]);
505+
}
506+
out += ']';
507+
}
508+
out += "}}";
509+
sendText(out);
510+
}
511+
448512
void StatusWebSocketHandler::handleStartDevices()
449513
{
450514
for (auto const& info : *mContext.infos) {

‎Framework/Core/src/StatusWebSocketHandler.h‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,9 @@ struct WSDPLHandler;
4848
/// {"cmd":"disable_signpost","device":"<name>","streams":["device","completion",...]}
4949
/// → disable the named signpost log streams for a device
5050
///
51+
/// {"cmd":"smaps","device":"<name>"}
52+
/// → driver reads /proc/<pid>/smaps of the device and replies with its usage per mapping
53+
///
5154
/// {"cmd":"list_signposts"}
5255
/// → driver replies with {"type":"signposts_list","streams":["device","completion",...]}
5356
/// → lists the known stream names
@@ -62,6 +65,10 @@ struct WSDPLHandler;
6265
/// {"type":"metrics_list","device":"<name>","metrics":["m1","m2",...]}
6366
/// → reply to list_metrics command
6467
///
68+
/// {"type":"smaps","device":"<name>","pid":<pid>,"fields":["rss","pss",...],"mappings":{"<path>":[<kB>,...],...}}
69+
/// → reply to smaps: kB per field, summed over all mappings of a path ("[anon]" for unnamed ones),
70+
/// or {"type":"smaps","device":"<name>","pid":<pid>,"error":"<text>"} where /proc is unavailable
71+
///
6572
/// {"type":"log","device":"<name>","level":"<level>","line":"<text>"}
6673
/// → pushed for each new log line from a subscribed device
6774
struct StatusWebSocketHandler : public WebSocketHandler {
@@ -96,6 +103,7 @@ struct StatusWebSocketHandler : public WebSocketHandler {
96103
void handleSubscribeLogs(std::string_view deviceName);
97104
void handleUnsubscribeLogs(std::string_view deviceName);
98105
void handleStartDevices();
106+
void handleSmaps(std::string_view deviceName);
99107
void handleEnableSignpost(std::string_view deviceName, std::string_view streamsArr);
100108
void handleDisableSignpost(std::string_view deviceName, std::string_view streamsArr);
101109
size_t findDeviceIndex(std::string_view name) const;

0 commit comments

Comments
 (0)