18 · 进程生命周期：从队列工作到超时回收

    multiprocessing.set_start_method("spawn", force=True)
    server_args.host = server_args.host.strip("[]")
    p = multiprocessing.Process(target=_launch_server, args=(server_args, os.getpid()))
    p.start()

    if getattr(server_args, "node_rank", 0) != 0:
        return p

    _wait_server_healthy(
        base_url=server_args.url(),
        api_key=server_args.api_key,
        is_process_alive=lambda: p.is_alive(),
    )

    return p

        existing_children = {process.pid for process in multiprocessing.active_children()}
        try:
            # Constructor failure must release partial topology too. A detached
            # actor can fail before any driver obtains a usable deployment.
            self.placements = create_placement_groups(args)
            try:
                self.servers, handles = (
                    ({}, []) if args.debug_train_only else start_rollout_servers(args, self.placements["rollout"])
                )
            finally:
                self.router_processes = [
                    process for process in multiprocessing.active_children() if process.pid not in existing_children
                ]
            ray.get(handles)
            self.lock = Lock.options(num_cpus=0, num_gpus=0).remote()
        except BaseException:
            try:
                self.dispose()
            except Exception:
                logging.getLogger(__name__).exception("Failed to dispose partial serving deployment")
            raise

def worker_process(task_queue, done_queue, rollout_func, reward_func, client, sampling_params):

    for line in iter(task_queue.get, "STOP"):
        if isinstance(line, str):
            item = json.loads(line)
        else:
            item = line

        # try:
        messages, finish_reason = rollout_func(client, item["prompt"], sampling_params)

        item["uid"] = str(uuid.uuid4())
        item["messages"] = messages
        reward = reward_func(item)
        item["rollout_index"] = 1
        item["reward"] = reward
        item["extra_info"] = {}
        item.update(sampling_params)
        item["timestamp"] = str(time.time())
        item["round_number"] = len([_ for _ in item["messages"] if _["role"] == "assistant"])
        item["finish_reason"] = finish_reason

        output_item = {
            "uid": item.pop("uid"),
            "messages": messages,
            "reward": reward,
            "instance_id": item.pop("instance_id"),
            "extra_info": item,
        }

        done_queue.put(output_item)

    done_queue.put("COMPLETE")

    def run(self, input_file, rollout_func, reward_func):
        task_queue, done_queue = Queue(maxsize=self.queue_size), Queue(maxsize=self.queue_size)

        def read_data_into_queue():
            cnt = 0
            items = []
            skipped_count = 0
            with open(input_file) as f:
                for i, line in enumerate(f):
                    item = json.loads(line)
                    if "instance_id" not in item:
                        item["instance_id"] = i
                    items.append(item)
            random.shuffle(items)

            for _ in range(self.num_repeats):

                for item in items:
                    for _ in range(self.num_repeat_per_sample):
                        item_repeat = copy.deepcopy(item)

                        if "uid" not in item_repeat:
                            item_repeat["uid"] = str(uuid.uuid4())

                        # Check if instance_id should be skipped
                        if self.skip_instance_ids is not None and item_repeat["instance_id"] in self.skip_instance_ids:
                            print(f"Skipping instance_id: {item_repeat['instance_id']}")
                            # Remove from skip list to handle potential duplicates in multiple epochs
                            self.skip_instance_ids.remove(item_repeat["instance_id"])
                            skipped_count += 1
                            continue

                        task_queue.put(item_repeat)
                    cnt += 1
                time.sleep(300)

            if skipped_count > 0:
                remaining_skip_count = len(self.skip_instance_ids) if self.skip_instance_ids is not None else 0
                print(
                    f"Rollout summary: skipped {skipped_count} instance_ids, {remaining_skip_count} still in skip list"
                )

            for _ in range(self.num_process):
                task_queue.put("STOP")

        processes = []
        SAMPLING_PARAMS["max_tokens"] = self.max_tokens

        for _ in range(self.num_process):
            process = Process(
                target=partial(worker_process, client=self.client, sampling_params=SAMPLING_PARAMS),
                args=(task_queue, done_queue, rollout_func, reward_func),
            )
            process.start()
            processes.append(process)

        process = Process(target=read_data_into_queue)
        process.start()

        progress_bar = tqdm()
        num_finished = 0
        while num_finished < self.num_process:
            item = done_queue.get()
            if item == "COMPLETE":
                num_finished += 1
            else:
                assert "reward" in item, f"reward not in item: {item}"
                assert "instance_id" in item, f"instance_id not in item: {item}"
                self.send_data_to_buffer(item)
                progress_bar.update(1)

        progress_bar.close()

        return "finished"

            for router in self.router_processes:
                if router.is_alive():
                    cleanup.run("terminate router", router.terminate)
                    cleanup.run("join router", router.join, timeout=min(10, cleanup.remaining / 2))
                    if router.is_alive():
                        cleanup.run("kill router", router.kill)
                        cleanup.run("reap router", router.join, timeout=cleanup.remaining)

    ctx = multiprocessing.get_context("spawn")
    with ctx.Pool(num_workers) as pool:
        results = pool.starmap(
            conversion_worker,
            [
                (
                    i,
                    key_chunks[i],
                    args,
                    megatron_args,
                    args.model_name,
                    state_dict_metadata_dict,
                    load_max_workers,
                    save_max_workers,
                )
                for i in range(num_workers)
            ],
        )

        try:
            proc = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                env=env,
                shell=False,  # Never use shell=True for security
            )
            # Add timeout to prevent hanging processes
            stdout, stderr = proc.communicate(timeout=self.execution_timeout)
            return_code = proc.returncode
        except subprocess.TimeoutExpired:
            proc.kill()
            stdout, stderr = proc.communicate()
            return_code = proc.returncode
            timeout_msg = f"Process timed out after {self.execution_timeout} seconds and was terminated."
            stderr = f"{stderr}\n{timeout_msg}"

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()

    proc = subprocess.Popen(cmd, env=env, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    lines = []
    for line in proc.stdout:
        lines.append(line)
        sys.stdout.write(line)
        sys.stdout.flush()
    proc.wait()
    return proc.returncode, "".join(lines)

    # Successful training cleans up its own resources; failures retain them.
    if not external_ray:
        # A manual trainer restart must leave the Ray head and its detached
        # rollout session alive. A stale address file alone is not proof of a
        # running cluster, so query the head before deciding to start one.
        ray_running = (
            subprocess.run(
                ["ray", "status"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30
            ).returncode
            == 0
        )

def exec_command(cmd: str, capture_output: bool = False) -> str | None:
    print(f"EXEC: {cmd}", flush=True)

    try:
        result = subprocess.run(
            ["bash", "-c", cmd],
            shell=False,
            check=True,
            capture_output=capture_output,
            **(dict(text=True) if capture_output else {}),
        )
    except subprocess.CalledProcessError as e:
        if capture_output:
            print(f"{e.stdout=} {e.stderr=}")
        raise

    if capture_output:
        print(f"Captured stdout={result.stdout} stderr={result.stderr}")
        return result.stdout

        else:  # Unix-like systems
            # On Unix-like systems, use which to find the command
            try:
                with open(os.devnull, "w") as devnull:
                    subprocess.check_call(
                        ["which", command],
                        stdout=devnull,
                        stderr=devnull,
                        shell=False,
                    )
                return True
            except subprocess.CalledProcessError:
                return False

    if not net_root.exists():
        return None

    bond_ifaces = [
        path.name for path in net_root.iterdir() if path.name.startswith("bond") and path.name[4:].isdigit()
    ]
    bond_ifaces.sort(key=lambda name: int(name[4:]))
    for iface in bond_ifaces:
        try:
            output = subprocess.check_output(["ip", "-o", "-4", "addr", "show", "dev", iface], text=True)
        except (OSError, subprocess.CalledProcessError):
            continue