PY代码

import os
from collections import defaultdict
from datetime import datetime
from socket import gethostname

SHM_PATH = "/dev/shm"
SEM_PREFIX = "sem.mp-"
MAPS_MARKER = "/dev/shm/sem.mp-"


def _holders_of(names):
    holders = defaultdict(set)
    try:
        pids = os.listdir("/proc")
    except OSError:
        return holders

    for pid in pids:
        if not pid.isdigit():
            continue
        try:
            with open(f"/proc/{pid}/maps", encoding="utf-8", errors="ignore") as f:
                for line in f:
                    if MAPS_MARKER not in line:
                        continue
                    path = line[line.index(MAPS_MARKER):].strip()
                    name = os.path.basename(path).replace(" (deleted)", "").strip()
                    if name in names:
                        holders[name].add(int(pid))
        except (FileNotFoundError, PermissionError, ProcessLookupError, OSError):
            continue
    return holders


param = Pipeline.params or {}
dry_run = bool(param.get("dry_run", True))
min_age_days = float(param.get("min_age_days", 1))
min_age_seconds = min_age_days * 24 * 3600
now = datetime.now().timestamp()
cutoff = now - min_age_seconds

if os.name != "posix" or not os.path.isdir(SHM_PATH):
    return {
        "ok": False,
        "error": "需要 Linux /dev/shm，请在 pipeline 所在 Pod 上跑",
    }

items = []
for entry in os.scandir(SHM_PATH):
    if not entry.name.startswith(SEM_PREFIX):
        continue
    try:
        stat = entry.stat(follow_symlinks=False)
    except FileNotFoundError:
        continue
    items.append({
        "name": entry.name,
        "mtime": stat.st_mtime,
        "time": datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M:%S"),
        "old_enough": stat.st_mtime < cutoff,
    })

names = {item["name"] for item in items}
holders = _holders_of(names)

orphans = []
kept_mapped = []
kept_young = []
for item in items:
    pids = sorted(holders.get(item["name"], []))
    if pids:
        kept_mapped.append(item["name"])
        continue
    if not item["old_enough"]:
        kept_young.append(item["name"])
        continue
    orphans.append(item["name"])

deleted = []
failed = []
if not dry_run:
    for name in orphans:
        path = os.path.join(SHM_PATH, name)
        try:
            os.unlink(path)
            deleted.append(name)
        except FileNotFoundError:
            continue
        except OSError as exc:
            failed.append({"name": name, "error": str(exc)})

return {
    "hostname": gethostname(),
    "pid": os.getpid(),
    "dry_run": dry_run,
    "min_age_days": min_age_days,
    "sem_mp_count": len(items),
    "orphan_old_count": len(orphans),
    "deleted_count": len(deleted),
    "not_found_in_proc_maps": [
        item["name"] for item in items if not holders.get(item["name"])
    ],
    "would_delete": orphans,
    "deleted": deleted,
    "failed": failed,
    "kept_mapped_count": len(kept_mapped),
    "kept_young_count": len(kept_young),
}