From 07355aede164926e2a8de5f85e1fb3a5ef8080c4 Mon Sep 17 00:00:00 2001 From: Rian354 Date: Sat, 5 Sep 2026 14:50:25 -0400 Subject: [PATCH] benchmark_perf: report PSS alongside summed RSS Every harness peaked on parent RSS plus the summed RSS of all children. That double-counts copy-on-write pages shared between forked workers, and the error is not small: a 64-worker pandarallel ingest measured this way summed to 1.89 TB on a 480 GB node. The same accounting inflated our own PyHealth 2.0 worker sweep, where PSS shows 32.1 GB at 32 workers against 39.1 GB for summed RSS. Adds PSS (proportional set size -- each shared page divided among the processes mapping it) across the process tree, with an RSS fallback where PSS is unavailable. Additive: peak_rss_bytes keeps its existing meaning so numbers already collected stay interpretable, and peak_pss_bytes lands beside it. Applies to all 16 harnesses: legacy, workers_n, meds_reader, and patient_exploration variants. --- .../benchmark_perf/benchmark_workers_n.py | 33 ++++++++++++++++++ ...benchmark_workers_n_drug_recommendation.py | 33 ++++++++++++++++++ .../benchmark_workers_n_length_of_stay.py | 33 ++++++++++++++++++ .../legacy_ver/benchmark_legacy_drug_rec.py | 33 ++++++++++++++++++ .../legacy_ver/benchmark_legacy_los.py | 33 ++++++++++++++++++ .../legacy_ver/benchmark_legacy_mortality.py | 33 ++++++++++++++++++ .../benchmark_meds_reader_drug_rec.py | 33 ++++++++++++++++++ ...hmark_meds_reader_drug_rec_pyhealth_etl.py | 34 +++++++++++++++++++ .../benchmark_meds_reader_los.py | 33 ++++++++++++++++++ .../benchmark_meds_reader_los_pyhealth_etl.py | 34 +++++++++++++++++++ .../benchmark_meds_reader_mortality.py | 33 ++++++++++++++++++ ...mark_meds_reader_mortality_pyhealth_etl.py | 34 +++++++++++++++++++ .../benchmark_patient_access_legacy.py | 34 +++++++++++++++++++ .../benchmark_patient_access_meds_reader.py | 34 +++++++++++++++++++ .../benchmark_patient_access_pandas.py | 34 +++++++++++++++++++ .../benchmark_patient_access_pyhealth2.py | 34 +++++++++++++++++++ 16 files changed, 535 insertions(+) diff --git a/examples/benchmark_perf/benchmark_workers_n.py b/examples/benchmark_perf/benchmark_workers_n.py index 6b53091cb..16d9f88b6 100644 --- a/examples/benchmark_perf/benchmark_workers_n.py +++ b/examples/benchmark_perf/benchmark_workers_n.py @@ -54,6 +54,7 @@ class RunResult: base_cache_bytes: int task_cache_bytes: int peak_rss_bytes: int + peak_pss_bytes: int def format_size(size_bytes: int) -> str: @@ -108,6 +109,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -116,14 +118,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -140,9 +169,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -329,6 +361,7 @@ def main() -> None: base_cache_bytes=base_cache_bytes, task_cache_bytes=task_cache_bytes, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), ) ) diff --git a/examples/benchmark_perf/benchmark_workers_n_drug_recommendation.py b/examples/benchmark_perf/benchmark_workers_n_drug_recommendation.py index 2d12f746e..583fe94b4 100644 --- a/examples/benchmark_perf/benchmark_workers_n_drug_recommendation.py +++ b/examples/benchmark_perf/benchmark_workers_n_drug_recommendation.py @@ -54,6 +54,7 @@ class RunResult: base_cache_bytes: int task_cache_bytes: int peak_rss_bytes: int + peak_pss_bytes: int def format_size(size_bytes: int) -> str: @@ -108,6 +109,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -116,14 +118,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -140,9 +169,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -331,6 +363,7 @@ def main() -> None: base_cache_bytes=base_cache_bytes, task_cache_bytes=task_cache_bytes, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), ) ) diff --git a/examples/benchmark_perf/benchmark_workers_n_length_of_stay.py b/examples/benchmark_perf/benchmark_workers_n_length_of_stay.py index 79c2a6f23..8bf832d03 100644 --- a/examples/benchmark_perf/benchmark_workers_n_length_of_stay.py +++ b/examples/benchmark_perf/benchmark_workers_n_length_of_stay.py @@ -54,6 +54,7 @@ class RunResult: base_cache_bytes: int task_cache_bytes: int peak_rss_bytes: int + peak_pss_bytes: int def format_size(size_bytes: int) -> str: @@ -108,6 +109,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -116,14 +118,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -140,9 +169,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -331,6 +363,7 @@ def main() -> None: base_cache_bytes=base_cache_bytes, task_cache_bytes=task_cache_bytes, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), ) ) diff --git a/examples/benchmark_perf/legacy_ver/benchmark_legacy_drug_rec.py b/examples/benchmark_perf/legacy_ver/benchmark_legacy_drug_rec.py index 780e762a5..a8009280c 100644 --- a/examples/benchmark_perf/legacy_ver/benchmark_legacy_drug_rec.py +++ b/examples/benchmark_perf/legacy_ver/benchmark_legacy_drug_rec.py @@ -96,6 +96,7 @@ class RunResult: total_s: float base_cache_bytes: int peak_rss_bytes: int + peak_pss_bytes: int num_samples: int @@ -186,6 +187,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -194,14 +196,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -218,9 +247,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -411,6 +443,7 @@ def main() -> None: total_s=total_s, base_cache_bytes=base_cache_bytes, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), num_samples=num_samples, ) ) diff --git a/examples/benchmark_perf/legacy_ver/benchmark_legacy_los.py b/examples/benchmark_perf/legacy_ver/benchmark_legacy_los.py index 7d036f40b..d68ae891f 100644 --- a/examples/benchmark_perf/legacy_ver/benchmark_legacy_los.py +++ b/examples/benchmark_perf/legacy_ver/benchmark_legacy_los.py @@ -96,6 +96,7 @@ class RunResult: total_s: float base_cache_bytes: int peak_rss_bytes: int + peak_pss_bytes: int num_samples: int @@ -186,6 +187,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -194,14 +196,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -218,9 +247,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -412,6 +444,7 @@ def main() -> None: total_s=total_s, base_cache_bytes=base_cache_bytes, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), num_samples=num_samples, ) ) diff --git a/examples/benchmark_perf/legacy_ver/benchmark_legacy_mortality.py b/examples/benchmark_perf/legacy_ver/benchmark_legacy_mortality.py index 21db0053d..17c513b64 100644 --- a/examples/benchmark_perf/legacy_ver/benchmark_legacy_mortality.py +++ b/examples/benchmark_perf/legacy_ver/benchmark_legacy_mortality.py @@ -247,6 +247,7 @@ class RunResult: total_s: float base_cache_bytes: int peak_rss_bytes: int + peak_pss_bytes: int num_samples: int @@ -337,6 +338,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -345,14 +347,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -369,9 +398,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -589,6 +621,7 @@ def main() -> None: total_s=total_s, base_cache_bytes=base_cache_bytes, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), num_samples=num_samples, ) ) diff --git a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_drug_rec.py b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_drug_rec.py index a9b719cb0..f445b32c9 100644 --- a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_drug_rec.py +++ b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_drug_rec.py @@ -487,6 +487,7 @@ class RunResult: task_process_s: float # Time to run the ML task total_s: float # Total time (conversion + task) peak_rss_bytes: int + peak_pss_bytes: int num_samples: int conversion_cached: bool # True if conversion was skipped @@ -518,6 +519,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -526,14 +528,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -550,9 +579,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -767,6 +799,7 @@ def main() -> None: task_process_s=task_process_s, total_s=total_s, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), num_samples=num_samples, conversion_cached=conversion.was_cached, ) diff --git a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_drug_rec_pyhealth_etl.py b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_drug_rec_pyhealth_etl.py index 1f98335ad..c18dc3077 100644 --- a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_drug_rec_pyhealth_etl.py +++ b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_drug_rec_pyhealth_etl.py @@ -509,6 +509,7 @@ class RunResult: task_process_s: float total_s: float peak_rss_bytes: int + peak_pss_bytes: int num_samples: int conversion_cached: bool @@ -528,6 +529,7 @@ def __init__(self, poll_interval_s: float = 0.1): self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self): @@ -536,11 +538,38 @@ def start(self): def reset(self): with self._lock: self._peak = 0 + self._peak_pss = 0 + + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -557,9 +586,12 @@ def _total_rss_bytes(self) -> int: def _run(self): while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -702,6 +734,7 @@ def main(): task_process_s = time.time() - task_start total_s = time.time() - run_start peak_rss = tracker.peak_bytes() + peak_pss = tracker.peak_pss_bytes() results.append(RunResult( num_threads=t, @@ -711,6 +744,7 @@ def main(): task_process_s=task_process_s, total_s=total_s, peak_rss_bytes=peak_rss, + peak_pss_bytes=peak_pss, num_samples=len(dataset), conversion_cached=conversion.was_cached, )) diff --git a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_los.py b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_los.py index 5acbd076e..e0891efa2 100644 --- a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_los.py +++ b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_los.py @@ -479,6 +479,7 @@ class RunResult: task_process_s: float # Time to run the ML task total_s: float # Total time (conversion + task) peak_rss_bytes: int + peak_pss_bytes: int num_samples: int conversion_cached: bool # True if conversion was skipped @@ -510,6 +511,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -518,14 +520,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -542,9 +571,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -759,6 +791,7 @@ def main() -> None: task_process_s=task_process_s, total_s=total_s, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), num_samples=num_samples, conversion_cached=conversion.was_cached, ) diff --git a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_los_pyhealth_etl.py b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_los_pyhealth_etl.py index 9dc787f87..b2536c7dd 100644 --- a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_los_pyhealth_etl.py +++ b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_los_pyhealth_etl.py @@ -538,6 +538,7 @@ class RunResult: task_process_s: float total_s: float peak_rss_bytes: int + peak_pss_bytes: int num_samples: int conversion_cached: bool @@ -557,6 +558,7 @@ def __init__(self, poll_interval_s: float = 0.1): self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self): @@ -565,11 +567,38 @@ def start(self): def reset(self): with self._lock: self._peak = 0 + self._peak_pss = 0 + + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -586,9 +615,12 @@ def _total_rss_bytes(self) -> int: def _run(self): while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -736,6 +768,7 @@ def main(): task_process_s = time.time() - task_start total_s = time.time() - run_start peak_rss = tracker.peak_bytes() + peak_pss = tracker.peak_pss_bytes() results.append(RunResult( num_threads=t, @@ -745,6 +778,7 @@ def main(): task_process_s=task_process_s, total_s=total_s, peak_rss_bytes=peak_rss, + peak_pss_bytes=peak_pss, num_samples=len(dataset), conversion_cached=conversion.was_cached, )) diff --git a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_mortality.py b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_mortality.py index cceda55c8..aeeebf56d 100644 --- a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_mortality.py +++ b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_mortality.py @@ -491,6 +491,7 @@ class RunResult: task_process_s: float # Time to run the ML task total_s: float # Total time (conversion + task) peak_rss_bytes: int + peak_pss_bytes: int num_samples: int conversion_cached: bool # True if conversion was skipped @@ -522,6 +523,7 @@ def __init__(self, poll_interval_s: float = 0.1) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -530,14 +532,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -554,9 +583,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -770,6 +802,7 @@ def main() -> None: task_process_s=task_process_s, total_s=total_s, peak_rss_bytes=peak_rss_bytes, + peak_pss_bytes=tracker.peak_pss_bytes(), num_samples=num_samples, conversion_cached=conversion.was_cached, ) diff --git a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_mortality_pyhealth_etl.py b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_mortality_pyhealth_etl.py index 754c482a0..ec5385170 100644 --- a/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_mortality_pyhealth_etl.py +++ b/examples/benchmark_perf/meds_reader_ver/benchmark_meds_reader_mortality_pyhealth_etl.py @@ -512,6 +512,7 @@ class RunResult: task_process_s: float total_s: float peak_rss_bytes: int + peak_pss_bytes: int num_samples: int conversion_cached: bool @@ -531,6 +532,7 @@ def __init__(self, poll_interval_s: float = 0.1): self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self): @@ -539,11 +541,38 @@ def start(self): def reset(self): with self._lock: self._peak = 0 + self._peak_pss = 0 + + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -560,9 +589,12 @@ def _total_rss_bytes(self) -> int: def _run(self): while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -704,6 +736,7 @@ def main(): task_process_s = time.time() - task_start total_s = time.time() - run_start peak_rss = tracker.peak_bytes() + peak_pss = tracker.peak_pss_bytes() results.append(RunResult( num_threads=t, @@ -713,6 +746,7 @@ def main(): task_process_s=task_process_s, total_s=total_s, peak_rss_bytes=peak_rss, + peak_pss_bytes=peak_pss, num_samples=len(dataset), conversion_cached=conversion.was_cached, )) diff --git a/examples/benchmark_perf/patient_exploration/benchmark_patient_access_legacy.py b/examples/benchmark_perf/patient_exploration/benchmark_patient_access_legacy.py index 50a280797..1ef7f90b3 100644 --- a/examples/benchmark_perf/patient_exploration/benchmark_patient_access_legacy.py +++ b/examples/benchmark_perf/patient_exploration/benchmark_patient_access_legacy.py @@ -63,6 +63,7 @@ class BenchmarkResult: patient_access_2nd_s: float # Second access (warm cache) total_s: float peak_rss_bytes: int + peak_pss_bytes: int patient_found: bool num_events: int num_visits: int @@ -85,6 +86,7 @@ def __init__(self, poll_interval_s: float = 0.05) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -93,14 +95,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -117,9 +146,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -270,6 +302,7 @@ def main() -> None: total_s = data_load_s + patient_access_1st_s + patient_access_2nd_s peak_rss = tracker.peak_bytes() + peak_pss = tracker.peak_pss_bytes() tracker.stop() @@ -280,6 +313,7 @@ def main() -> None: patient_access_2nd_s=patient_access_2nd_s, total_s=total_s, peak_rss_bytes=peak_rss, + peak_pss_bytes=peak_pss, patient_found=patient_found, num_events=num_events, num_visits=num_visits, diff --git a/examples/benchmark_perf/patient_exploration/benchmark_patient_access_meds_reader.py b/examples/benchmark_perf/patient_exploration/benchmark_patient_access_meds_reader.py index 961615ddf..922ae49e9 100644 --- a/examples/benchmark_perf/patient_exploration/benchmark_patient_access_meds_reader.py +++ b/examples/benchmark_perf/patient_exploration/benchmark_patient_access_meds_reader.py @@ -61,6 +61,7 @@ class BenchmarkResult: patient_access_2nd_s: float # Second access (warm cache) total_s: float peak_rss_bytes: int + peak_pss_bytes: int patient_found: bool num_events: int used_cached_db: bool @@ -83,6 +84,7 @@ def __init__(self, poll_interval_s: float = 0.05) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -91,14 +93,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -115,9 +144,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -354,6 +386,7 @@ def main() -> None: total_s = data_load_s + patient_access_1st_s + patient_access_2nd_s peak_rss = tracker.peak_bytes() + peak_pss = tracker.peak_pss_bytes() tracker.stop() @@ -366,6 +399,7 @@ def main() -> None: patient_access_2nd_s=patient_access_2nd_s, total_s=total_s, peak_rss_bytes=peak_rss, + peak_pss_bytes=peak_pss, patient_found=patient_found, num_events=num_events, used_cached_db=used_cached_db, diff --git a/examples/benchmark_perf/patient_exploration/benchmark_patient_access_pandas.py b/examples/benchmark_perf/patient_exploration/benchmark_patient_access_pandas.py index 47fd434bf..c999d45e6 100644 --- a/examples/benchmark_perf/patient_exploration/benchmark_patient_access_pandas.py +++ b/examples/benchmark_perf/patient_exploration/benchmark_patient_access_pandas.py @@ -51,6 +51,7 @@ class BenchmarkResult: patient_access_2nd_s: float # Second access (warm cache) total_s: float peak_rss_bytes: int + peak_pss_bytes: int patient_found: bool num_events: int num_visits: int @@ -75,6 +76,7 @@ def __init__(self, poll_interval_s: float = 0.05) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -83,14 +85,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -107,9 +136,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -523,6 +555,7 @@ def main() -> None: # Calculate total time total_s = csv_load_s + parquet_write_s + parquet_read_s + patient_access_1st_s + patient_access_2nd_s peak_rss = tracker.peak_bytes() + peak_pss = tracker.peak_pss_bytes() finally: tracker.stop() @@ -541,6 +574,7 @@ def main() -> None: patient_access_2nd_s=patient_access_2nd_s, total_s=total_s, peak_rss_bytes=peak_rss, + peak_pss_bytes=peak_pss, patient_found=patient_found, num_events=num_events, num_visits=num_visits, diff --git a/examples/benchmark_perf/patient_exploration/benchmark_patient_access_pyhealth2.py b/examples/benchmark_perf/patient_exploration/benchmark_patient_access_pyhealth2.py index 736ae3023..d4fcb56c4 100644 --- a/examples/benchmark_perf/patient_exploration/benchmark_patient_access_pyhealth2.py +++ b/examples/benchmark_perf/patient_exploration/benchmark_patient_access_pyhealth2.py @@ -43,6 +43,7 @@ class BenchmarkResult: patient_access_2nd_s: float # Second access (warm cache) total_s: float peak_rss_bytes: int + peak_pss_bytes: int patient_found: bool num_events: int @@ -64,6 +65,7 @@ def __init__(self, poll_interval_s: float = 0.05) -> None: self._stop = threading.Event() self._lock = threading.Lock() self._peak = 0 + self._peak_pss = 0 self._thread = threading.Thread(target=self._run, daemon=True) def start(self) -> None: @@ -72,14 +74,41 @@ def start(self) -> None: def reset(self) -> None: with self._lock: self._peak = 0 + self._peak_pss = 0 def stop(self) -> None: self._stop.set() + def peak_pss_bytes(self) -> int: + with self._lock: + return self._peak_pss + def peak_bytes(self) -> int: with self._lock: return self._peak + def _total_pss_bytes(self) -> int: + """Proportional set size across the process tree. + + Summing RSS over forked workers counts every shared page once per + process; PSS divides it among the mappers, so the total is additive. + Falls back to RSS where PSS is unavailable (non-Linux, or no perms). + """ + total = 0 + try: + procs = [self._proc] + self._proc.children(recursive=True) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0 + for p in procs: + try: + total += p.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError): + try: + total += p.memory_info().rss + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return total + def _total_rss_bytes(self) -> int: total = 0 try: @@ -96,9 +125,12 @@ def _total_rss_bytes(self) -> int: def _run(self) -> None: while not self._stop.is_set(): rss = self._total_rss_bytes() + pss = self._total_pss_bytes() with self._lock: if rss > self._peak: self._peak = rss + if pss > self._peak_pss: + self._peak_pss = pss time.sleep(self._poll_interval_s) @@ -219,6 +251,7 @@ def main() -> None: total_s = data_load_s + patient_access_1st_s + patient_access_2nd_s peak_rss = tracker.peak_bytes() + peak_pss = tracker.peak_pss_bytes() tracker.stop() @@ -234,6 +267,7 @@ def main() -> None: patient_access_2nd_s=patient_access_2nd_s, total_s=total_s, peak_rss_bytes=peak_rss, + peak_pss_bytes=peak_pss, patient_found=patient_found, num_events=num_events, )