869d2c41fb
16 -> 6 (daemon default, daocloud, 1ms.run, 1panel, rat.dev, ustc): the long tail never delivered anyway. All-6 failure now prints an explicit self-load recipe (save/load via an egress machine, or pull when the network recovers) instead of a terse RuntimeError. Co-Authored-By: Claude <noreply@anthropic.com>
237 lines
9.7 KiB
Python
237 lines
9.7 KiB
Python
"""Docker sandbox: one implementation serving both consumers.
|
||
|
||
exec(): untrusted model-generated code — hard isolation
|
||
(--network none, cpu/mem/pids caps, read-only rootfs, tmpfs /tmp;
|
||
writes escape only through explicit bind mounts)
|
||
serve(): trusted engine containers (vllm/sglang) — network ON (weights pull),
|
||
GPU passthrough; consumed by the model Deployer via acquire()
|
||
"""
|
||
|
||
import os
|
||
import shlex
|
||
import subprocess
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from .base import EnvHandle, ExecResult, Sandbox, acquire, register_sandbox
|
||
|
||
|
||
def _run(cmd: List[str], **kw) -> subprocess.CompletedProcess:
|
||
return subprocess.run(cmd, capture_output=True, text=True, **kw)
|
||
|
||
|
||
def docker_available() -> bool:
|
||
return _run(['docker', 'info']).returncode == 0
|
||
|
||
|
||
def ensure_image(img: str) -> None:
|
||
"""Fail-fast sandbox image preflight: present locally, else pull ONCE.
|
||
|
||
Without this, every sample's `docker run` tries its own pull at scoring
|
||
time -- a missing image burned 1140 x 3 retries x ~70s on bigcodebench
|
||
before anyone saw a 0.0%."""
|
||
if not img:
|
||
return
|
||
if _run(['docker', 'image', 'inspect', img]).returncode == 0:
|
||
return
|
||
# CN-mirror fallback chain (daocloud -> 1ms.run -> baidubce -> sjtu ->
|
||
# rat.dev), mirroring the SWE prefetch path; a hit is retagged to the
|
||
# canonical name so the recipe never knows which mirror answered
|
||
from .prefetch import _pull_one
|
||
try:
|
||
_pull_one(img)
|
||
return
|
||
except RuntimeError as e:
|
||
raise RuntimeError(
|
||
f'沙箱镜像 {img!r} 不可用：本地没有，全部镜像源尝试失败。'
|
||
f'{str(e)[:150]}\n'
|
||
'需要自己加载镜像（任选其一），然后重跑加 --rescore：\n'
|
||
f' A. 有外网的机器: docker pull {img} && docker save {img} | gzip > img.tgz\n'
|
||
f' 拷回本机后: docker load < img.tgz\n'
|
||
f' B. 本机网络恢复后: docker pull {img}\n'
|
||
'预测都在 checkpoint 里，镜像就位后纯判分即可。') from e
|
||
|
||
|
||
@register_sandbox('docker')
|
||
class DockerSandbox(Sandbox):
|
||
name = 'docker'
|
||
|
||
DEFAULT_EXEC_IMAGE = 'python:3.11-slim'
|
||
|
||
def exec(
|
||
self,
|
||
files: Dict[str, str],
|
||
entry: str = 'main.py',
|
||
mounts: Optional[Dict[str, str]] = None,
|
||
timeout_s: int = 60,
|
||
image: str = '',
|
||
entrypoint: str = '',
|
||
) -> ExecResult:
|
||
img = image or self.DEFAULT_EXEC_IMAGE
|
||
with tempfile.TemporaryDirectory(prefix='eh-sbx-') as host_dir:
|
||
workdir = Path(host_dir) / 'work'
|
||
workdir.mkdir()
|
||
for fname, content in (files or {}).items():
|
||
dest = workdir / fname
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
dest.write_text(content, encoding='utf-8')
|
||
cmd = [
|
||
'docker', 'run', '--rm',
|
||
'--network', 'none', # untrusted code: no egress
|
||
'--cpus', '2', '--memory', '2g', '--pids-limit', '256',
|
||
'--read-only', '--tmpfs', '/tmp:rw,size=64m',
|
||
# /work must be writable: BigCodeBench tasks write output
|
||
# files (task_func_data/, matplotlib caches, etc.) to cwd;
|
||
# the official Evaluate.Dockerfile runs with a writable fs
|
||
'-v', f'{workdir}:/work:rw',
|
||
]
|
||
out_host = None
|
||
if mounts:
|
||
for cpath, hpath in mounts.items():
|
||
out_host = Path(hpath).expanduser()
|
||
out_host.mkdir(parents=True, exist_ok=True)
|
||
cmd += ['-v', f'{out_host}:{cpath}:rw']
|
||
if entrypoint:
|
||
# images with an official-evaluator ENTRYPOINT (bigcodebench:
|
||
# python3 -m bigcodebench.evaluate) swallow our runner as
|
||
# CLI args -> override it. The entrypoint IS the interpreter
|
||
# now: pass only the script path, else 'python3 python
|
||
# /work/main.py' tries to open a file named 'python'
|
||
cmd += ['--entrypoint', entrypoint]
|
||
runner = [f'/work/{entry}']
|
||
else:
|
||
runner = ['python', f'/work/{entry}'] if entry.endswith('.py') \
|
||
else ['sh', f'/work/{entry}']
|
||
# Named + retried runs. A timed-out/killed `docker run` only kills
|
||
# the CLI client -- the container lives on (--rm fires on EXIT)
|
||
# and leaked containers slowly drown the daemon; the name gives
|
||
# us something to rm -f. Exit 125 is a DAEMON-side failure (shim
|
||
# error etc.), not the model's code failing -- scoring it pass=0
|
||
# would corrupt results, so retry it.
|
||
import uuid
|
||
|
||
t0 = time.time()
|
||
proc = None
|
||
for attempt in range(3):
|
||
cname = f'eh-exec-{uuid.uuid4().hex[:10]}'
|
||
full = cmd + ['--name', cname, img, *runner]
|
||
try:
|
||
proc = _run(full, timeout=timeout_s + 30)
|
||
except subprocess.TimeoutExpired:
|
||
_run(['docker', 'rm', '-f', cname], timeout=60) # CLI died, container didn't
|
||
return ExecResult(exit_code=-1, timed_out=True, duration_s=timeout_s,
|
||
error=f'sandbox timeout after {timeout_s}s')
|
||
except BaseException: # Ctrl+C / kill: reap, then propagate
|
||
_run(['docker', 'rm', '-f', cname], timeout=60)
|
||
raise
|
||
if proc.returncode != 125 or attempt == 2:
|
||
break
|
||
# image-not-found is PERMANENT: retrying it 3x per sample
|
||
# burned 1140 x ~200s on a nonexistent bigcodebench image
|
||
_nf = 'Unable to find image' in (proc.stderr or '') \
|
||
or 'failed to resolve' in (proc.stderr or '') \
|
||
or 'manifest unknown' in (proc.stderr or '') \
|
||
or 'pull access denied' in (proc.stderr or '')
|
||
if _nf:
|
||
break
|
||
# clear any husk; fresh name next try. Bounded: an rm against
|
||
# a bloated daemon can hang for minutes and silently eat the
|
||
# whole worker pool (7 of 8 workers were observed stuck here)
|
||
_run(['docker', 'rm', '-f', cname], timeout=60)
|
||
time.sleep(2 * (attempt + 1)) # give the daemon a beat
|
||
return ExecResult(
|
||
exit_code=proc.returncode,
|
||
stdout=proc.stdout,
|
||
stderr=proc.stderr,
|
||
duration_s=round(time.time() - t0, 2),
|
||
)
|
||
|
||
|
||
# ---------------- serve-side (model deployment environments) ----------------
|
||
|
||
|
||
def docker_serve(
|
||
engine: str,
|
||
model: str,
|
||
cfg: Dict[str, Any],
|
||
default_image: str,
|
||
engine_args: List[str],
|
||
) -> Dict[str, str]:
|
||
"""Start (or reuse) an OpenAI-protocol serving container.
|
||
|
||
Used through sandbox.acquire() by the model Deployer — the deployment
|
||
container is just another environment this layer provides. Bind-mounts
|
||
the HF cache (weights stay on the host; containers come and go).
|
||
"""
|
||
import socket
|
||
|
||
image = cfg.get('image', default_image)
|
||
port = int(cfg.get('port', 0))
|
||
if not port:
|
||
with socket.socket() as s:
|
||
s.bind(('', 0))
|
||
port = s.getsockname()[1]
|
||
hf = cfg.get('hf_home', os.environ.get('HF_HOME', '~/.cache/huggingface'))
|
||
gpus = str(cfg.get('gpus', 'all'))
|
||
container = f'evalharness-{engine}-{model}'.replace('/', '-')[:120]
|
||
|
||
_run(['docker', 'rm', '-f', container]) # stale instance from a crashed run
|
||
cmd = [
|
||
'docker', 'run', '-d',
|
||
'--name', container,
|
||
'--gpus', f'device={gpus}' if gpus.isdigit() else gpus,
|
||
'--network', 'host',
|
||
'-v', f'{Path(hf).expanduser()}:/root/.cache/huggingface',
|
||
'-e', f'HF_ENDPOINT={os.environ.get("HF_ENDPOINT", "https://hf-mirror.com")}',
|
||
image,
|
||
'--model', cfg.get('model_id', cfg.get('model', model)),
|
||
'--served-model-name', model,
|
||
'--port', str(port),
|
||
*engine_args,
|
||
*shlex.split(cfg.get('extra_args', '')),
|
||
]
|
||
for key, flag in (('gpu_mem_util', '--gpu-memory-utilization'),
|
||
('max_model_len', '--max-model-len'),
|
||
('tp_size', '--tensor-parallel-size'),
|
||
('dtype', '--dtype')):
|
||
if cfg.get(key):
|
||
cmd += [flag, str(cfg[key])]
|
||
r = _run(cmd)
|
||
if r.returncode != 0:
|
||
raise RuntimeError(f'docker run failed: {r.stderr[:500]}')
|
||
api_base = f'http://localhost:{port}/v1'
|
||
_wait_healthy(api_base, int(cfg.get('timeout_s', 1800)))
|
||
return {'api_base': api_base, 'model': model, 'container': container, 'port': str(port)}
|
||
|
||
|
||
def _wait_healthy(api_base: str, timeout_s: int) -> None:
|
||
import urllib.request
|
||
|
||
deadline = time.time() + timeout_s
|
||
while time.time() < deadline:
|
||
try:
|
||
with urllib.request.urlopen(f'{api_base}/models', timeout=5) as resp:
|
||
if resp.status == 200:
|
||
return
|
||
except Exception:
|
||
time.sleep(5)
|
||
raise TimeoutError(f'serving engine not healthy after {timeout_s}s at {api_base}')
|
||
|
||
|
||
def docker_stop(handle: EnvHandle) -> None:
|
||
if handle.container:
|
||
_run(['docker', 'rm', '-f', handle.container])
|
||
|
||
|
||
def serve_env(engine: str, model: str, cfg: Dict[str, Any],
|
||
default_image: str, engine_args: Optional[List[str]] = None) -> EnvHandle:
|
||
"""acquire() wrapper: shared, refcounted serve environment."""
|
||
return acquire(
|
||
kind=engine,
|
||
name=model,
|
||
start_fn=lambda _m: docker_serve(engine, _m, cfg, default_image, engine_args or []),
|
||
stop_fn=docker_stop,
|
||
)
|