Interruptible‑инстансы: чекпоинты и устойчивость

				
					version: "3.9"

x-env: &env
  CHUNK_MAX_SEC: "120"
  CHECKPOINT_LOCAL: "/nvme/ckpt"
  CHECKPOINT_REMOTE: "/obj/ckpt"
  WAL_DIR: "/nvme/wal"
  LEASE_DIR: "/obj/leases"
  JOBS_MANIFEST: "/obj/jobs/chunks.json"
  AUTOSAVE_SEC: "45"
  GRACEFUL_TIMEOUT_SEC: "20"
  PRECISION: "fp16"  # bf16|fp16|fp32

services:
  intr-worker:
    image: cloudcompute/interruptible-worker:latest
    environment: *env
    deploy:
      resources:
        reservations:
          devices: [{capabilities: ["gpu"]}]
    volumes:
      - /nvme/cache:/nvme
      - /mnt/object:/obj
    command: ["python","worker.py","--manifest","/obj/jobs/chunks.json"]


				
			

				
					import os, json, time, signal, uuid, shutil, tempfile, pathlib
LEASE_TTL = 180
GRACE = int(os.getenv("GRACEFUL_TIMEOUT_SEC", "20"))

_should_exit = False
def _graceful(*_):  # ловим SIGTERM/SIGINT
    global _should_exit; _should_exit = True

for s in (signal.SIGTERM, signal.SIGINT): signal.signal(s, _graceful)

def lease_take(lease_dir, chunk_id, holder):
    p = pathlib.Path(lease_dir, f"{chunk_id}.json")
    now = time.time()
    try:
        if p.exists():
            lease = json.loads(p.read_text())
            if now - lease["ts"] < LEASE_TTL: return False
        p.write_text(json.dumps({"holder": holder, "ts": now}))
        return True
    except Exception:
        return False

def lease_heartbeat(lease_dir, chunk_id, holder):
    p = pathlib.Path(lease_dir, f"{chunk_id}.json")
    p.write_text(json.dumps({"holder": holder, "ts": time.time()}))

def atomic_commit(tmp_dir, final_dir):
    os.makedirs(final_dir, exist_ok=True)
    staging = tempfile.mkdtemp(prefix="commit_", dir=final_dir)
    for name in os.listdir(tmp_dir):
        shutil.move(os.path.join(tmp_dir, name), staging)
    os.replace(staging, os.path.join(final_dir, pathlib.Path(staging).name))
    pathlib.Path(final_dir, "DONE").touch()

def process_chunk(chunk, ckpt_dir) -> bool:
    # 1) compute → периодический autosave в ckpt_dir
    # 2) собрать результаты в tmp_out, затем atomic_commit
    return True

def main(manifest, lease_dir, ckpt_local, out_remote):
    holder = f"{uuid.uuid4()}"
    chunks = json.loads(open(manifest).read())["chunks"]
    for ch in chunks:
        if not lease_take(lease_dir, ch["id"], holder): continue
        start = time.time()
        last_hb = start
        while not _should_exit:
            ok = process_chunk(ch, ckpt_local)
            if not ok: break
            if time.time() - last_hb > 10:
                lease_heartbeat(lease_dir, ch["id"], holder); last_hb = time.time()
            break
        if _should_exit:
            # быстрый чекпоинт перед выходом
            pathlib.Path(ckpt_local, "quick.ckpt").touch()
            break
        # COMMIT
        atomic_commit(tmp_dir=os.path.join(ckpt_local, "out", ch["id"]),
                      final_dir=os.path.join(out_remote, ch["id"]))