19 · 信号、进程组与文件锁：退出以后谁释放资源？

        with ExitStack() as cleanup:
            # The controller owns both the queue and its checkpoint branch.
            # Hold the save-directory lock until close() (or actor exit), and
            # publish the branch before any queue writes can become durable.
            if self.restore_plan.root is not None:
                plan = self.restore_plan
                root = Path(plan.root)
                (root / "rollout").mkdir(parents=True, exist_ok=True)
                self._checkpoint_lock = cleanup.enter_context((root / "rollout/session.lock").open("a"))
                try:
                    fcntl.flock(self._checkpoint_lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                except BlockingIOError as error:
                    raise RuntimeError(f"Another training job owns save directory {root}") from error

def test_checkpoint_lock_released_when_controller_process_is_killed(tmp_path):
    import select

    from slime.data.checkpoint import resolve_checkpoint

    root = tmp_path / "run"
    args = _checkpoint_args(None, root)
    args, plan_args = resolve_checkpoint(args)
    process = subprocess.Popen(
        [
            sys.executable,
            "-c",
            """
import pickle, sys
from slime.data.queue_data_source import RolloutQueueController

args, plan_args = pickle.load(sys.stdin.buffer)
controller = RolloutQueueController(args, restore_plan=plan_args)
print("ready", flush=True)
sys.stdin.buffer.read(1)
""",
        ],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        pickle.dump((args, plan_args), process.stdin)
        process.stdin.flush()
        assert select.select([process.stdout], [], [], 60)[0], "Controller did not start"
        assert process.stdout.readline() == b"ready\n"
        again = _checkpoint_args(None, root)
        again, plan_again = resolve_checkpoint(again)
        with pytest.raises(RuntimeError, match="Another training job"):
            RolloutQueueController(again, restore_plan=plan_again)
    finally:
        process.kill()
        process.wait(timeout=30)
        process.stdin.close()
        process.stdout.close()
        process.stderr.close()
    # No close() ran in the killed process; the kernel must release its lock.
    with closing(RolloutQueueController(again, restore_plan=plan_again)):
        pass

class FdLock:
    def __init__(self, path_pattern, gpu_id: int):
        self.gpu_id = gpu_id
        self.path = _get_lock_path(path_pattern, self.gpu_id)
        self.fd = None

    def open(self):
        assert self.fd is None
        self.fd = open(self.path, "a+")

    def lock(self):
        assert self.fd is not None
        fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)

    def close(self):
        assert self.fd is not None
        try:
            self.fd.close()

def _launch_server(server_args, parent_pid):
    from sglang.srt.entrypoints.http_server import launch_server
    from sglang.srt.utils import kill_itself_when_parent_died

    # A wedged actor cannot run shutdown(). Its server must still release GPUs
    # when the owner kills that actor before creating a replacement engine.
    kill_itself_when_parent_died()
    if os.getppid() == parent_pid:
        launch_server(server_args)

            Path(manager.args.rollout_data_dir).parent / "crash.json",
            {
                "manager_pid": os.getpid(),
                "samples": [
                    {"index": sample.index, "tokens": sample.tokens, "reward": sample.reward} for sample in samples
                ],
            },
        )
        if phase == "conversion_reply_lost":
            raise TimeoutError("Conversion committed but its RPC reply was lost")
        os.kill(os.getpid(), signal.SIGKILL)

def _run_command(args):
    cmd = args.cmd
    if cmd[0] == "--":
        cmd = cmd[1:]

    child = subprocess.Popen(cmd, start_new_session=True)

    def terminate_child(signum, _frame):
        if child.poll() is None:
            child.terminate()
            try:
                child.wait(timeout=5)
                sys.exit(128 + signum)
            except subprocess.TimeoutExpired:
                pass

            try:
                os.killpg(child.pid, signum)
            except ProcessLookupError:
                pass
            try:
                child.wait(timeout=5)
            except subprocess.TimeoutExpired:
                try:
                    os.killpg(child.pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass
                child.wait()
        sys.exit(128 + signum)

    previous_handlers = {}
    for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
        previous_handlers[signum] = signal.getsignal(signum)
        signal.signal(signum, terminate_child)

    try:
        while True:
            returncode = child.poll()
            if returncode is not None:
                return _normalize_returncode(returncode)
            time.sleep(1)
    finally:
        for signum, handler in previous_handlers.items():
            signal.signal(signum, handler)

class timeout:

    def __init__(self, seconds=1, error_message="Timeout"):
        self.seconds = seconds
        self.error_message = error_message

    def handle_timeout(self, signum, frame):
        raise TimeoutError(self.error_message)

    def __enter__(self):
        signal.signal(signal.SIGALRM, self.handle_timeout)
        signal.alarm(self.seconds)

    def __exit__(self, type, value, traceback):
        signal.alarm(0)

def main(fp8_path, bf16_path):
    torch.set_default_dtype(torch.bfloat16)
    os.makedirs(bf16_path, exist_ok=True)
    os.system("cp -rf " + fp8_path + "/config.json " + bf16_path)
    os.system("cp -rf " + fp8_path + "/*.py " + bf16_path)
    os.system("cp -rf " + fp8_path + "/tokenizer* " + bf16_path)
    os.system("cp -rf " + fp8_path + "/chat_template* " + bf16_path)