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, )