Python调用/usr/bin/time测进程到SIGINT的最大内存失败,求解决方案
需求与问题
我有一个无法修改的服务器程序my-server,需要测量它从启动到某个外部事件发生时的最大内存占用。手动在Shell中可通过以下步骤实现:
- 执行命令:
/usr/bin/time -f "%M" my-server - 触发外部事件
- 按
CTRL+C发送INT信号终止进程 - 读取标准错误输出获取结果
我尝试用Python编程实现该流程,但通过subprocess启动命令并发送SIGINT信号后,proc.communicate()返回的out和err均为空。脚本对正常终止的进程(如echo hello)可正常工作,但对需信号终止的进程无法获取标准错误输出。
测试用等效代码
terminate-me.py
import signal import sys def signal_handler(sig, frame): print('Terminated', file=sys.stderr) # 需要读取这个输出 sys.exit(0) signal.signal(signal.SIGINT, signal_handler) print('Send SIGINT to terminate') signal.pause()
main.py
import signal import subprocess proc = subprocess.Popen( ["python", "terminate-me.py"], stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True ) # 此处无需等待外部事件 proc.send_signal(signal.SIGINT) out, err = proc.communicate() print(out) print(err)
解决方案
1. 修复测试代码的信号时序问题
测试代码中main.py在进程启动后立即发送SIGINT,此时terminate-me.py可能还未完成信号处理函数的注册,导致默认终止逻辑触发,无法输出内容。添加短暂等待确保进程初始化完成即可:
import signal import subprocess import time proc = subprocess.Popen( ["python", "terminate-me.py"], stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True ) # 等待进程完成信号注册 time.sleep(0.1) proc.send_signal(signal.SIGINT) out, err = proc.communicate() print(out) print(err)
2. 针对/usr/bin/time的生产环境实现
实际场景中,需等待my-server启动完成(通过读取启动日志判断),再触发外部事件,最后发送信号终止进程,捕获/usr/bin/time的标准错误输出:
import signal import subprocess # 启动/usr/bin/time包裹的my-server proc = subprocess.Popen( ["/usr/bin/time", "-f", "%M", "my-server"], stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True ) # 等待服务器启动完成(替换为实际的启动标志字符串) while True: line = proc.stdout.readline() if "服务启动完成" in line: break # 触发外部事件(根据实际业务逻辑实现,比如调用API、发送请求等) # trigger_external_event() # 发送SIGINT终止进程 proc.send_signal(signal.SIGINT) # 获取输出,/usr/bin/time的内存结果在stderr中 _, err = proc.communicate() max_memory_kb = err.strip() print(f"最大内存占用: {max_memory_kb} KB")
3. 备选方案:用psutil直接监控内存
如果/usr/bin/time的方式不符合需求,可使用psutil库实时监控进程内存:
import psutil import subprocess import time import signal # 启动服务器 proc = subprocess.Popen( ["my-server"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # 等待启动完成 while True: line = proc.stdout.readline() if "服务启动完成" in line: break max_memory = 0 process = psutil.Process(proc.pid) try: # 模拟等待外部事件完成(替换为实际触发逻辑) time.sleep(5) # 持续监控内存直到外部事件结束 while proc.poll() is None: # 获取物理内存占用(字节) current_mem = process.memory_info().rss if current_mem > max_memory: max_memory = current_mem time.sleep(0.1) finally: # 终止进程 proc.send_signal(signal.SIGINT) proc.communicate() print(f"最大内存占用: {max_memory / 1024:.2f} KB")
内容的提问来源于stack exchange,提问作者L. Gig
相关产品推荐
相关产品推荐

