You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何限制Python单进程的CPU和内存占用 适配exec/eval用户代码运行场景

解决方案

核心认知纠正

你对resource包的作用范围存在误解:resource.setrlimit是进程级限制,仅作用于调用该方法的当前进程,不会影响父进程/其他无关子进程。你当前用ProcessPoolExecutor将任务放到独立子进程执行的架构是完全可行的,只需要调整限制逻辑即可。

具体实现方案

1. 内存限制调整

RLIMIT_AS参数控制进程的虚拟地址空间上限,对应内存占用限制,你POC中设置的1024字节远小于正常Python进程运行所需的基础内存,会直接触发OOM终止。需要先读取系统总内存,再取1%作为软限制:

import os
import resource
import psutil
from concurrent.futures import ProcessPoolExecutor

def get_1_percent_mem():
    total_mem = psutil.virtual_memory().total
    return int(total_mem * 0.01)

def task(user_code: str):
    # 子进程内设置内存限制,仅对当前子进程生效
    mem_limit = get_1_percent_mem()
    soft, hard = resource.getrlimit(resource.RLIMIT_AS)
    resource.setrlimit(resource.RLIMIT_AS, (mem_limit, hard))
    # 设置CPU运行时间硬限制(可选兜底,避免进程死循环占满资源)
    cpu_soft, cpu_hard = resource.getrlimit(resource.RLIMIT_CPU)
    resource.setrlimit(resource.RLIMIT_CPU, (30, cpu_hard)) # 单任务最多累计运行30秒
    
    # 执行用户传入的代码
    exec(user_code, {})
    print(f"进程 {os.getpid()} 执行完成")

def main():
    executor = ProcessPoolExecutor(max_workers=3)
    # 示例传入用户代码
    user_code = """
s = "helloworld"
for j in range(1, 10):
    s = s + s
print(len(s))
    """
    executor.submit(task, user_code)
    executor.submit(task, user_code)

if __name__ == '__main__':
    main()

2. 1% CPU使用率限制

resource包无法直接限制CPU使用率,仅能限制累计运行时长,要实现使用率控制推荐两种方案:

  • Linux环境首选cgroups:为每个执行用户代码的子进程单独创建cgroup,设置cpu.cfs_quota_us=1000、cpu.cfs_period_us=100000,即可固定限制CPU使用率为1%,限制精度高且无额外监控开销
  • 跨平台场景用psutil轮询:启动子进程后定期读取其CPU使用率,连续超过1%阈值3秒直接终止进程,实现简单但存在一定延迟

安全补充提示

直接执行用户传入的eval/exec代码存在极高安全风险,即使做了资源限制也可能出现代码逃逸、敏感文件读取等问题,生产环境建议额外增加沙箱隔离,比如用Docker容器运行每个用户的代码任务,进一步缩小攻击面。

内容的提问来源于stack exchange,提问作者dinesh707

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.20 15:45:04