diff --git a/hosted/opd/backend.py b/hosted/opd/backend.py index 8321854..de5999e 100644 --- a/hosted/opd/backend.py +++ b/hosted/opd/backend.py @@ -239,6 +239,76 @@ def close(self): self.service.close() +class SelfHosted: + """A self-hosted Tinker-compatible server at $TINKER_BASE_URL (see hosted/selfhosted/README.md). + + The servers do not authenticate, but the SDK requires a key that starts with `tml-`. A teacher path of the form + `base:` samples a frozen model that the server serves by name (verl-tinker teachers); any other + path is a sampler checkpoint on the same server. Subclasses set what each server accepts. + """ + name = 'self-hosted' + ACCEPTS_SEED = True # create_lora_training_client(seed=...) + ACCEPTS_METADATA = True # user_metadata on the session and on each training client + BASE_PREFIX = 'base:' + + def __init__(self, experiment): + self.url = os.environ['TINKER_BASE_URL'] + os.environ.setdefault('TINKER_API_KEY', 'tml-self-hosted') + server = json.loads(Path(os.environ['TINKER_SERVER_INFO']).read_text()) if os.environ.get('TINKER_SERVER_INFO') else {} + self.account = {'endpoint': self.url, 'server': self.name, 'server_info': server} + metadata = {'user_metadata': {'experiment': experiment}} if self.ACCEPTS_METADATA else {} + self.service = tinker.ServiceClient(base_url=self.url, api_key=os.environ['TINKER_API_KEY'], **metadata) + + def capabilities(self, model): + caps = self.service.get_server_capabilities() + assert model in [m.model_name for m in caps.supported_models], f'{model} is not served at {self.url}' + return caps.model_dump() + + def new_student(self, model, rank, seed, metadata): + return self.service.create_lora_training_client( + model, rank=rank, seed=seed if self.ACCEPTS_SEED else None, + **({'user_metadata': metadata} if self.ACCEPTS_METADATA else {})) + + def resume_student(self, state_path): + return self.service.create_training_client_from_state_with_optimizer(state_path) + + def sampler(self, model=None, path=None): + if path and path.startswith(self.BASE_PREFIX): + return self.service.create_sampling_client(base_model=path[len(self.BASE_PREFIX):]) + return self.service.create_sampling_client(**({'model_path': path} if path else {'base_model': model})) + + def sync(self, train, name): + path = train.save_weights_for_sampler(name=name).result().path + return self.sampler(path=path), path + + def use_tokenizers(self, student, teacher): + pass + + def close(self): + pass + + +class SkyRL(SelfHosted): + """SkyRL's Tinker API server (`python -m skyrl.tinker.api --backend megatron`). LoRA alpha is a server setting, and + the seed is ignored. A sampler checkpoint stays servable only while its model is loaded, so the server runs with + --session-timeout-sec -1 and teachers are trained on the same server instance that serves them.""" + name = 'skyrl' + + +class Miles(SelfHosted): + """Miles' Tinker gateway (`serve_tinker.py`). It rejects a LoRA seed and user metadata, and caps the rank at the + server's --lora-rank.""" + name = 'miles' + ACCEPTS_SEED = False + ACCEPTS_METADATA = False + + +class Verl(SelfHosted): + """verl-tinker (`python -m verl_tinker.start`). One trainable model per server: teachers are frozen models that the + server config serves by name (`base:` paths), and every training run needs its own server.""" + name = 'verl' + + def service_config(resources, experiment=None): """from_firetitan_config arguments for the lease's trainer and deployment; closing never deletes them.""" return dict( @@ -255,4 +325,5 @@ def service_config(resources, experiment=None): **({'user_metadata': {'experiment': experiment}} if experiment else {})) -BACKENDS = {'tinker': Tinker, 'fireworks': Fireworks, 'fireworks-serverless': FireworksServerless} +BACKENDS = {'tinker': Tinker, 'fireworks': Fireworks, 'fireworks-serverless': FireworksServerless, + 'skyrl': SkyRL, 'miles': Miles, 'verl': Verl} diff --git a/hosted/opd/rl.py b/hosted/opd/rl.py new file mode 100644 index 0000000..bcdd7b0 --- /dev/null +++ b/hosted/opd/rl.py @@ -0,0 +1,167 @@ +"""Reinforcement learning (GRPO) on a Tinker-compatible API: a fresh Qwen3.6-35B-A3B LoRA learns the reasoning_gym +tasks from the verifier reward alone. Thinking is on. + +Every update the student samples `--group-size` responses to each of `--prompts` training problems (split equally +between `--domains`). The reward is the shared scorer's 0/1 score; each response's advantage is its reward normalized +within its prompt's group, applied to every response token. The update is one clipped PPO step (ratio 0.8-1.2). +Groups with no reward variance are kept with zero advantage. The settings are those of the released GRPO teachers +(hosted/tinker/README.md): 32 prompts x 8 responses, LR 1e-5 for LoRA, Adam betas 0.9/0.98, gradient clip 1. + + python hosted/opd/rl.py --domains caesar_cipher --output /abs/rl-caesar --backend skyrl + python hosted/opd/rl.py --output /abs/rl-both --updates 20 + +Records, evaluation, checkpoints and --policy-lag behave as in opd.py and mopd.py (see common.py). +""" +import argparse +import json +import statistics +import time + +import tinker + +from common import Experiment, add_protocol_arguments, check_protocol_arguments, policy_datum, run +from shared import recipe, scoring + +GRPO = {'prompts': 32, 'group_size': 8, 'clip_low_threshold': 0.8, 'clip_high_threshold': 1.2} +ADAM = {'beta1': 0.9, 'beta2': 0.98, 'eps': 1e-8, 'weight_decay': 0.0, 'grad_clip_norm': 1.0} + + +def group_advantages(rewards): + """Rewards normalized within one prompt's group; a group with one response or no variance gets zeros.""" + mean = statistics.mean(rewards) + std = statistics.stdev(rewards) if len(rewards) > 1 else 0.0 + return [(r - mean) / (std + 1e-6) for r in rewards] + + +class RL(Experiment): + """An Experiment whose rollouts are groups scored by the verifier instead of by a teacher.""" + + def start(self): + a = self.args + self.teachers = {} + if a.resume: + train = self.backend.resume_student(a.resume) + else: + train = self.backend.new_student(self.student_model[0], a.rank, a.seed, {'run': self.out.name, 'kind': self.kind}) + try: + (self.out/'model-info.json').write_text(train.get_info().model_dump_json(indent=2)) + except Exception as error: + (self.out/'model-info.json').write_text(json.dumps({'unavailable': str(error)[:300]})) + return train, self.backend.sync(train, f'update-{a.start_update:03d}')[0] + + def rollout(self, student, domain, row, seed, t0): + """One prompt's group. A multi-response request leaves the seed unset, as in the cookbook.""" + a = self.args + prompt = self.prompt(row) + begin = time.monotonic() + result = student.sample(tinker.ModelInput.from_ints(prompt), num_samples=a.group_size, + sampling_params=self.params(1.0, seed if a.group_size == 1 else None)).result() + sampled = time.monotonic() + assert len(result.sequences) == a.group_size + samples = [] + for j, sequence in enumerate(result.sequences): + tokens, logprobs = list(sequence.tokens), list(sequence.logprobs) + assert len(tokens) == len(logprobs) + response = self.tokenizer.decode(tokens, skip_special_tokens=True) + samples.append({'sample': j, 'tokens': tokens, 'logprobs': logprobs, 'response': response, + 'score': scoring.score(response, row['label']), 'stop_reason': sequence.stop_reason}) + for s, advantage in zip(samples, group_advantages([s['score'] for s in samples])): + s['advantage'] = advantage + return {'domain': domain, 'index': row['_index'], 'seed': seed, 'prompt_tokens': prompt, 'samples': samples, + 'submitted': begin - t0, 'sampled': sampled - t0, 'scored': sampled - t0} + + def update(self, train, batch, prefetch): + a = self.args + update = batch['update'] + t0 = time.monotonic() + self.log('update_start', update=update + 1, prompts=len(batch['futures'])) + groups = [f.result() for f in batch['futures']] + rollout_seconds = time.monotonic() - t0 + lag = update - batch['version'] + assert 0 <= lag <= a.policy_lag, 'Refuse a stale or future-policy batch' + upcoming = prefetch() + self.save_rollouts(f'update-{update+1:03d}', [{'update': update + 1, 'policy_lag': lag, **g} for g in groups]) + + train_start = time.monotonic() + flat = [(g, s) for g in groups for s in g['samples']] + datums = [policy_datum(g['prompt_tokens'], s['tokens'], s['logprobs'], [s['advantage']] * len(s['tokens'])) + for g, s in flat] + fb = train.forward_backward(datums, loss_fn='ppo', loss_fn_config={ + 'clip_low_threshold': a.clip_low, 'clip_high_threshold': a.clip_high}) + opt = train.optim_step(tinker.AdamParams(learning_rate=a.learning_rate, **ADAM)) + fb, opt = fb.result(), opt.result() + train_seconds = time.monotonic() - train_start + mismatch = [float(x) - y for (g, s), o in zip(flat, fb.loss_fn_outputs, strict=True) if 'logprobs' in o + for x, y in zip(o['logprobs'].data[len(g['prompt_tokens']) - 1:], s['logprobs'], strict=True)] + + sync_start = time.monotonic() + student, sampler_path = self.backend.sync(train, f'update-{update+1:03d}') + sync_seconds = time.monotonic() - sync_start + self.append('metrics.jsonl', self.rl_metrics(update, groups, t0, rollout_seconds, train_seconds, sync_seconds, + mismatch, sampler_path, fb, opt, lag)) + return student, upcoming + + def rl_metrics(self, update, groups, t0, rollout_seconds, train_seconds, sync_seconds, mismatch, sampler_path, + fb, opt, lag): + samples = [s for g in groups for s in g['samples']] + sampled = [g['sampled'] for g in groups] + response_tokens = sum(len(s['tokens']) for s in samples) + prompt_tokens = sum(len(g['prompt_tokens']) * len(g['samples']) for g in groups) + by_domain = {} + for d in self.domains: + gs = [g for g in groups if g['domain'] == d] + ss = [s for g in gs for s in g['samples']] + by_domain[d] = {'groups': len(gs), 'responses': len(ss), 'reward': statistics.mean(s['score'] for s in ss), + 'groups_with_signal': sum(len({s['score'] for s in g['samples']}) > 1 for g in gs), + 'mean_response_tokens': statistics.mean(len(s['tokens']) for s in ss)} + update_seconds = time.monotonic() - t0 + metric = { + 'update': update + 1, 'policy_lag': lag, 'groups': len(groups), 'responses': len(samples), + 'learning_rate': self.args.learning_rate, 'prompt_tokens': prompt_tokens, 'response_tokens': response_tokens, + 'max_response_tokens': max(len(s['tokens']) for s in samples), + 'truncated': sum(s['stop_reason'] == 'length' for s in samples), 'by_domain': by_domain, + 'reward': statistics.mean(s['score'] for s in samples), + 'nonzero_advantage_responses': sum(s['advantage'] != 0 for s in samples), + 'mean_unique_responses_per_group': statistics.mean(len({s['response'] for s in g['samples']}) for g in groups), + 'timing': {'rollout': rollout_seconds, 'train': train_seconds, 'sync': sync_seconds, 'update': update_seconds, + 'first_group': min(sampled), 'median_group': statistics.median(sampled), + 'last_group': max(sampled)}, + 'throughput': {'sampled_tokens_per_second': response_tokens / max(sampled), + 'trained_tokens_per_second': (prompt_tokens + response_tokens) / train_seconds, + 'update_response_tokens_per_second': response_tokens / update_seconds}, + 'train_sampler_logprob_delta_mean': statistics.mean(mismatch) if mismatch else None, + 'train_sampler_logprob_delta_abs_mean': statistics.mean(abs(x) for x in mismatch) if mismatch else None, + 'sampler_path': sampler_path, 'forward_backward_metrics': fb.metrics, 'optimizer_metrics': opt.metrics} + self.log('update_complete', **{k: v for k, v in metric.items() if k != 'forward_backward_metrics'}) + return metric + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument('--domains', choices=recipe.DOMAINS, nargs='+', default=list(recipe.DOMAINS)) + parser.add_argument('--group-size', type=int, default=GRPO['group_size']) + parser.add_argument('--clip-low', type=float, default=GRPO['clip_low_threshold']) + parser.add_argument('--clip-high', type=float, default=GRPO['clip_high_threshold']) + add_protocol_arguments(parser) + parser.set_defaults(prompts=GRPO['prompts']) + args = parser.parse_args() + check_protocol_arguments(parser, args) + if args.group_size < 2: parser.error('--group-size must be at least 2 for group-normalized advantages') + if args.prompts % len(args.domains): parser.error('--prompts must split evenly over the domains') + domains = [d for d in recipe.DOMAINS if d in args.domains] + share = args.prompts // len(domains) + + class Run(RL): + def record_inputs(self): + super().record_inputs() + config = json.loads((self.out/'config.json').read_text()) + config.update(loss='ppo', advantage='group-normalized verifier reward', adam=ADAM, + clip=[args.clip_low, args.clip_high], teachers=None) + (self.out/'config.json').write_text(json.dumps(config, indent=2)) + + run(lambda: Run(args, 'rl', {d: None for d in domains}), + lambda experiment, update: [p for d in domains for p in experiment.train_prompts(d, update, share)]) + + +if __name__ == '__main__': + main() diff --git a/hosted/selfhosted/campaign.sh b/hosted/selfhosted/campaign.sh new file mode 100644 index 0000000..f1b6152 --- /dev/null +++ b/hosted/selfhosted/campaign.sh @@ -0,0 +1,58 @@ +#!/bin/bash +# Run the rest of one framework's campaign on a running server, one stage at a time, then summarize every run. +# +# CAMPAIGN=/shared/... CLIENT_VENV=... hosted/selfhosted/campaign.sh FRAMEWORK SERVER_DIR [STAGE ...] +# +# Stages (default: sft rl mopd), each run through client.sh so it gets its own telemetry window: +# sft both SFT teachers at once (sft.py train, the released-teacher defaults); skipped if they already exist +# rl GRPO on both domains (rl.py defaults) +# mopd MOPD from the two SFT teachers (mopd.py defaults); teachers.json is built from the teachers' teacher.json +# A stage waits for the previous one and stops the campaign if it fails. Outputs: $CAMPAIGN/FRAMEWORK/out/, +# $CAMPAIGN/FRAMEWORK/runs/.{log,window.json,summary.json}. +set -euo pipefail +: "${CAMPAIGN:?CAMPAIGN must be set}" +FRAMEWORK=$1; SERVER=$(cd "$2" && pwd); shift 2 +if [[ $# -eq 0 ]]; then STAGES=(sft rl mopd); else STAGES=("$@"); fi +OUT=$CAMPAIGN/$FRAMEWORK/out; RUNS=$CAMPAIGN/$FRAMEWORK/runs +SELFHOSTED=$CAMPAIGN/opd/hosted/selfhosted +mkdir -p "$OUT" "$RUNS" +log() { echo "$(date -u +%FT%TZ) $*"; } + +wait_run() { # Wait for a client run's window file; fail if it exited non-zero. + local name=$1 + until [[ -f $RUNS/$name.window.json ]]; do sleep 30; done + local code; code=$(python3 -c "import json,sys; print(json.load(open(sys.argv[1]))['exit_code'])" "$RUNS/$name.window.json") + "$CAMPAIGN/client-venv/bin/python" "$SELFHOSTED/summarize.py" "$RUNS/$name.window.json" "$OUT/$name" \ + --telemetry-code "$CAMPAIGN/telemetry-scripts" || log "summary failed for $name" + [[ $code == 0 ]] || { log "$name exited $code; stopping"; exit 1; } + log "$name complete" +} + +start() { # start NAME -- COMMAND... + local name=$1; shift 2 + [[ -f $RUNS/$name.window.json ]] && { log "$name already ran"; return; } + [[ -f $RUNS/$name.log ]] || bash "$SELFHOSTED/client.sh" "$SERVER" "$name" "$FRAMEWORK" -- "$@" +} + +for stage in "${STAGES[@]}"; do + case $stage in + sft) + for d in caesar_cipher simple_geometry; do + start "sft-$d" -- python hosted/opd/sft.py train --backend "$FRAMEWORK" --domain "$d" --output "$OUT/sft-$d" + done + wait_run sft-caesar_cipher; wait_run sft-simple_geometry ;; + rl) + start rl -- python hosted/opd/rl.py --backend "$FRAMEWORK" --output "$OUT/rl" + wait_run rl ;; + mopd) + python3 - "$OUT" > "$OUT/teachers.json" <<'PY' +import json, sys +out = sys.argv[1] +print(json.dumps({d: json.load(open(f'{out}/sft-{d}/teacher.json'))['sampler_path'] for d in ('caesar_cipher', 'simple_geometry')}, indent=2)) +PY + start mopd -- python hosted/opd/mopd.py --backend "$FRAMEWORK" --teachers "$OUT/teachers.json" --output "$OUT/mopd" + wait_run mopd ;; + *) log "unknown stage $stage"; exit 2 ;; + esac +done +log "campaign stages done: ${STAGES[*]}" diff --git a/hosted/selfhosted/client.sh b/hosted/selfhosted/client.sh new file mode 100644 index 0000000..9bc1646 --- /dev/null +++ b/hosted/selfhosted/client.sh @@ -0,0 +1,33 @@ +#!/bin/bash +# Run one client (smoke.py, sft.py, rl.py, opd.py or mopd.py) against a running self-hosted server, as a step of the +# server's Slurm job on its head node, with the server's endpoint and description in the environment. +# +# CAMPAIGN=/shared/... hosted/selfhosted/client.sh SERVER_DIR NAME BACKEND -- python hosted/opd/sft.py train ... +# +# SERVER_DIR is the server's output directory (it holds server.json). The client's console goes to +# SERVER_DIR/../runs/NAME.log, and NAME.window.json records its UTC start and end, the server job and the exit +# code, so the server's telemetry can be cut to the run. The step runs in the background and survives the caller. +# CLIENT_VENV (default $CAMPAIGN/client-venv) selects the client environment: SkyRL pins tinker 0.25.0, whose +# responses the 0.30 SDK rejects (it requires sample_sequence_ids), so SkyRL clients use a 0.25.0 environment. +set -euo pipefail +: "${CAMPAIGN:?CAMPAIGN must be set}" +SERVER=$(cd "$1" && pwd); NAME=$2; BACKEND=$3; shift 3 +[[ ${1:-} == -- ]] && shift +RUNS=$(dirname "$SERVER")/runs +VENV=${CLIENT_VENV:-$CAMPAIGN/client-venv} +mkdir -p "$RUNS" +read -r JOB HEAD ENDPOINT < <(python3 -c "import json,sys; s=json.load(open(sys.argv[1])); print(s['slurm_job_id'], s['head'], s['endpoint'])" "$SERVER/server.json") +[[ -e $RUNS/$NAME.log ]] && { echo "$RUNS/$NAME.log exists; pick a new name" >&2; exit 1; } +printf -v COMMAND '%q ' "$@" +nohup srun --overlap --jobid="$JOB" -N1 -w "$HEAD" bash -c " + export HF_HOME=$CAMPAIGN/hf HF_HUB_OFFLINE=1 PYTHONDONTWRITEBYTECODE=1 PATH=$VENV/bin:\$PATH + export CAMPAIGN_PYDEPS=$VENV/lib/python3.12/site-packages + export TINKER_BASE_URL=$ENDPOINT TINKER_SERVER_INFO=$SERVER/server.json + cd $CAMPAIGN/opd + start=\$(date -u +%Y-%m-%dT%H:%M:%S.%3NZ) + $COMMAND; code=\$? + python3 -c 'import json,sys; json.dump(dict(name=sys.argv[1], backend=sys.argv[2], server=sys.argv[3], slurm_job_id=sys.argv[4], + start_utc=sys.argv[5], end_utc=sys.argv[6], exit_code=int(sys.argv[7]), command=sys.argv[8]), open(sys.argv[9], \"w\"), indent=2)' \ + $NAME $BACKEND $SERVER $JOB \$start \$(date -u +%Y-%m-%dT%H:%M:%S.%3NZ) \$code '$COMMAND' $RUNS/$NAME.window.json + exit \$code" > "$RUNS/$NAME.log" 2>&1 & +echo "started $NAME on $HEAD (job $JOB): $RUNS/$NAME.log" diff --git a/hosted/selfhosted/common.sh b/hosted/selfhosted/common.sh new file mode 100644 index 0000000..18afca9 --- /dev/null +++ b/hosted/selfhosted/common.sh @@ -0,0 +1,75 @@ +# Cluster steps that every self-hosted server launcher shares; sourced by /serve.sbatch inside a Slurm job. +# +# selfhosted_setup FRAMEWORK IMAGE CONTAINER_NAME +# Sets RUN ($CAMPAIGN//server-), NODES, HEAD, HEAD_IP and CONTAINER (the pyxis flags); removes +# leaked /dev/shm segments; writes node-ips.json; starts the GPU, fabric and InfiniBand samplers and the +# before-snapshot on every node; exports the NCCL settings. +# selfhosted_ray PYTHON_BIN_DIR +# Starts a Ray head on HEAD and a worker on every other node, inside the container, and waits until every GPU +# has joined. +# selfhosted_finish +# Takes the after-snapshot and removes leaked /dev/shm segments. +: "${CAMPAIGN:?CAMPAIGN must be set}" +SELFHOSTED=$CAMPAIGN/opd/hosted/selfhosted + +selfhosted_setup() { + local framework=$1 image=$2 name=$3 + RUN=$CAMPAIGN/$framework/server-$SLURM_JOB_ID + mkdir -p "$RUN/logs" + mapfile -t NODES < <(scontrol show hostnames "$SLURM_JOB_NODELIST") + HEAD=${NODES[0]} + HEAD_IP=$(srun --overlap -N1 -w "$HEAD" hostname -I | awk '{print $1}') + # node -> IP, so summarize.py can map engine and trainer addresses in server logs to nodes. + srun --overlap -N"${#NODES[@]}" --ntasks-per-node=1 bash -c 'echo "$(hostname -s) $(hostname -I | cut -d" " -f1)"' \ + | python3 -c 'import json, sys; print(json.dumps(dict(l.split() for l in sys.stdin if l.strip()), indent=2))' \ + > "$RUN/node-ips.json" + # The pods' /dev/shm is 16 GiB, and killed vLLM, SGLang and PyTorch workers leave their segments in it; NCCL + # then fails to create its own. Remove every segment no process holds. + srun --overlap -N"${#NODES[@]}" --ntasks-per-node=1 python3 "$SELFHOSTED/shm_clean.py" + + export MILES_GPU_TELEMETRY_DIR=$RUN/telemetry/gpu MILES_FABRIC_TELEMETRY_DIR=$RUN/telemetry/fabric + export MILES_SYSTEM_TELEMETRY_DIR=$RUN/telemetry/system + srun --overlap -N"${#NODES[@]}" --ntasks-per-node=1 bash "$CAMPAIGN/telemetry-scripts/system_snapshot.sh" before + srun --overlap -N"${#NODES[@]}" --ntasks-per-node=1 bash "$CAMPAIGN/telemetry-scripts/gpu_monitor.sh" & + srun --overlap -N"${#NODES[@]}" --ntasks-per-node=1 bash "$CAMPAIGN/telemetry-scripts/fabric_monitor.sh" & + # The pods' sysfs has no InfiniBand port counters, so fabric_monitor.sh records empty IB rows; perfquery reads them. + srun --overlap -N"${#NODES[@]}" --ntasks-per-node=1 bash "$SELFHOSTED/ib_monitor.sh" & + + export NVIDIA_VISIBLE_DEVICES=all HF_HOME=$CAMPAIGN/hf HF_HUB_OFFLINE=1 PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True + export NCCL_DEBUG=INFO NCCL_DEBUG_SUBSYS=INIT,NET NCCL_DEBUG_FILE=$RUN/logs/nccl-%h-%p.out + # Images built for AWS load the aws-ofi-nccl plugin, which finds no EFA device and falls back to Libfabric TCP + # over the pod network. Use NCCL's own InfiniBand transport on the eight 400 Gb/s NDR rails (NCCL_NET=IB fails + # instead of falling back); override NCCL_IB_HCA on other clusters. + export NCCL_NET_PLUGIN=none NCCL_NET=IB + export NCCL_IB_HCA=${NCCL_IB_HCA:-=mlx5_0,mlx5_1,mlx5_2,mlx5_3,mlx5_4,mlx5_9,mlx5_12,mlx5_13} + CONTAINER=(--container-image="$image" --container-name="$name" --container-mounts=/shared:/shared + --container-writable --export=ALL) +} + +selfhosted_ray() { + local bin=$1 node gpus want=$((${#NODES[@]} * 8)) + srun --overlap -N1 -w "$HEAD" "${CONTAINER[@]}" bash -c \ + "$bin/ray start --head --node-ip-address=$HEAD_IP --port=6379 --disable-usage-stats --block" \ + > "$RUN/logs/ray-$HEAD.log" 2>&1 & + sleep 20 + for node in "${NODES[@]:1}"; do + srun --overlap -N1 -w "$node" "${CONTAINER[@]}" bash -c \ + "$bin/ray start --address=$HEAD_IP:6379 --node-ip-address=\$(hostname -I | awk '{print \$1}') --disable-usage-stats --block" \ + > "$RUN/logs/ray-$node.log" 2>&1 & + done + # Serve only once every GPU has joined; a node whose container failed to start would otherwise leave a + # placement group pending until the first request times out. + for _ in $(seq 60); do + gpus=$(srun --overlap -N1 -w "$HEAD" "${CONTAINER[@]}" "$bin/python3" -c \ + "import ray; ray.init(address='$HEAD_IP:6379', logging_level='ERROR'); print(int(ray.cluster_resources().get('GPU', 0)))" 2>/dev/null | tail -1) + [[ $gpus == "$want" ]] && return 0 + sleep 10 + done + echo "Ray has ${gpus:-0} of $want GPUs; see $RUN/logs/ray-*.log" >&2 + exit 1 +} + +selfhosted_finish() { + srun --overlap -N"${#NODES[@]}" --ntasks-per-node=1 bash "$CAMPAIGN/telemetry-scripts/system_snapshot.sh" after + srun --overlap -N"${#NODES[@]}" --ntasks-per-node=1 python3 "$SELFHOSTED/shm_clean.py" +} diff --git a/hosted/selfhosted/ib_monitor.sh b/hosted/selfhosted/ib_monitor.sh new file mode 100644 index 0000000..eb60561 --- /dev/null +++ b/hosted/selfhosted/ib_monitor.sh @@ -0,0 +1,27 @@ +#!/bin/bash +# Sample InfiniBand port counters with perfquery, for pods whose sysfs has no ports/*/counters (the ClusterMAX +# fabric_monitor.sh then records empty IB rows). The CSV has fabric_monitor.sh's IB columns, and PortXmitData and +# PortRcvData are in 4-byte words like the sysfs counters, so ClusterMAX's summarize_ib reads it unchanged; +# summarize.py uses telemetry/fabric/ib-perfquery in place of telemetry/fabric/ib when it exists. +set -u +: "${MILES_FABRIC_TELEMETRY_DIR:?MILES_FABRIC_TELEMETRY_DIR must be set}" +: "${IB_SAMPLE_SECONDS:=5}" +node="$(hostname -s 2>/dev/null || hostname)" +out="$MILES_FABRIC_TELEMETRY_DIR/ib-perfquery/${node}.csv" +mkdir -p "$(dirname "$out")" +[[ -s $out ]] || printf 'collector_time_utc,node,device,port,state,rate,port_xmit_data,port_rcv_data,port_xmit_packets,port_rcv_packets,symbol_error,link_downed,port_rcv_errors,port_xmit_discards\n' > "$out" # A restart appends. +devices=() +for d in /sys/class/infiniband/*; do + [[ $(cat "$d/ports/1/link_layer" 2>/dev/null) == InfiniBand ]] && devices+=("$(basename "$d")") +done +while true; do + stamp="$(date -u +%Y-%m-%dT%H:%M:%S.%NZ)" + for dev in "${devices[@]}"; do + state=$(cat "/sys/class/infiniband/$dev/ports/1/state" 2>/dev/null | tr -d ' ') + rate=$(cat "/sys/class/infiniband/$dev/ports/1/rate" 2>/dev/null | tr -d ' ') + perfquery -x -C "$dev" -P 1 2>/dev/null | awk -v t="$stamp" -v n="$node" -v d="$dev" -v s="$state" -v r="$rate" -F'[:.]+' ' + /^PortXmitData/ {x=$NF} /^PortRcvData/ {rx=$NF} /^PortXmitPkts/ {xp=$NF} /^PortRcvPkts/ {rp=$NF} + END { if (x != "") printf "%s,%s,%s,1,\"%s\",\"%s\",%s,%s,%s,%s,,,,\n", t, n, d, s, r, x, rx, xp, rp }' >> "$out" + done + sleep "$IB_SAMPLE_SECONDS" +done diff --git a/hosted/selfhosted/miles/serve.sbatch b/hosted/selfhosted/miles/serve.sbatch new file mode 100644 index 0000000..c8e7ee9 --- /dev/null +++ b/hosted/selfhosted/miles/serve.sbatch @@ -0,0 +1,69 @@ +#!/bin/bash +# The Miles Tinker gateway (serve_tinker.py: Megatron LoRA trainer, SGLang samplers) on four 8-GPU Slurm nodes. +# +# CAMPAIGN=/shared/.../tinker-endpoints sbatch -p PARTITION hosted/selfhosted/miles/serve.sbatch +# +# Layout, matching skyrl/serve.sbatch: the trainer on two nodes, four 4-GPU SGLang engines on the other two, not +# colocated (the gateway forbids --colocate). The Qwen3.6-35B-A3B parallelism is that of Miles' +# scripts/run_qwen3_5_35b_a3b_lora.py: TP2 (two query groups cap it), EP8, full recompute, bshd micro-batches of +# one sequence (Megatron's GatedDeltaNet rejects packed sequences). The gateway rejects a datum longer than +# --max-tokens-per-gpu, so it is 32,768 for 30,720-token responses. Clients see the model as Qwen/Qwen3.6-35B-A3B +# (--tinker-model-name), must not send a LoRA seed or user metadata (the Miles backend drops both), and get at most +# rank 64. Every adapter trains with alpha 32, as on SkyRL. +# +# Inputs under $CAMPAIGN: images/miles-dev-202609290439.sqsh (radixark/miles, Miles 9e4260de with the gateway), hf/. +# Outputs go to $CAMPAIGN/miles/server-: server.json, telemetry/, logs/, checkpoints/. Serves until cancelled. +#SBATCH -J miles-tinker +#SBATCH -N 4 +#SBATCH --exclusive +#SBATCH --gres=gpu:8 +#SBATCH --time=UNLIMITED +#SBATCH -o %x-%j.out +set -euo pipefail +: "${CAMPAIGN:?CAMPAIGN must be set}" +PORT=${PORT:-8000} +IMAGE=${IMAGE:-$CAMPAIGN/images/miles-dev-202609290439.sqsh} +BASE_MODEL=Qwen/Qwen3.6-35B-A3B +MODEL_DIR=$CAMPAIGN/hf/hub/models--Qwen--Qwen3.6-35B-A3B/snapshots/995ad96eacd98c81ed38be0c5b274b04031597b0 +LORA_RANK=${LORA_RANK:-64} +LORA_ALPHA=${LORA_ALPHA:-32} +ADAPTERS=${ADAPTERS:-6} +source "$CAMPAIGN/opd/hosted/selfhosted/common.sh" +selfhosted_setup miles "$IMAGE" miles-tinker +mkdir -p "$RUN/checkpoints" +selfhosted_ray /opt/sglang/bin + +TRAIN_ARGS="--hf-checkpoint $MODEL_DIR --megatron-to-hf-mode bridge \ + --lora-rank $LORA_RANK --lora-alpha $LORA_ALPHA --lora-dropout 0.0 --no-gradient-accumulation-fusion \ + --multi-lora-n-adapters $ADAPTERS --target-modules attn,mlp,unembed \ + --tinker-server-port $PORT --tinker-checkpoint-root $RUN/checkpoints --tinker-model-name $BASE_MODEL \ + --optimizer adam --lr 1e-5 --weight-decay 0.0 --adam-beta1 0.9 --adam-beta2 0.999 \ + --tensor-model-parallel-size 2 --sequence-parallel --pipeline-model-parallel-size 1 --context-parallel-size 1 \ + --expert-model-parallel-size 8 --expert-tensor-parallel-size 1 \ + --recompute-granularity full --recompute-method uniform --recompute-num-layers 1 \ + --qkv-format bshd --micro-batch-size 1 --max-tokens-per-gpu 32768 \ + --rollout-num-gpus-per-engine 4 --sglang-mem-fraction-static 0.8 --sglang-dtype bfloat16 \ + --sglang-max-lora-rank $LORA_RANK --sglang-lora-backend triton --sglang-context-length 32768 \ + --actor-num-nodes 2 --actor-num-gpus-per-node 8 --rollout-num-gpus 16 \ + --attention-dropout 0.0 --hidden-dropout 0.0 --accumulate-allreduce-grads-in-fp32 --attention-softmax-in-fp32 \ + --attention-backend flash --update-weight-buffer-size 536870912" + +python3 - "$RUN/server.json" < "$RUN/logs/server.log" 2>&1 & +SERVER=$! +wait $SERVER || true +selfhosted_finish diff --git a/hosted/selfhosted/report.py b/hosted/selfhosted/report.py new file mode 100644 index 0000000..be3daaf --- /dev/null +++ b/hosted/selfhosted/report.py @@ -0,0 +1,91 @@ +"""Compare the self-hosted Tinker-compatible frameworks from their run summaries (summarize.py). + + python hosted/selfhosted/report.py $CAMPAIGN --output $CAMPAIGN/report + +Reads $CAMPAIGN//runs/*.summary.json for every framework directory and writes comparison.json (one row +per framework and run) and REPORT.md (the tables below). Throughput is the summary's window: tokens trained in steps +2..N over the time from step 1 to step N, per physical GPU of the 32-GPU allocation. The quality columns are the +teacher benchmark for SFT (avg@3 on the first 100 dev problems of the teacher's domain) and the evaluations at +updates 0 and last for RL and MOPD (avg@3 on the first 100 dev problems of each domain, temperature 1). +""" +import argparse +import json +from pathlib import Path + +DOMAINS = ('caesar_cipher', 'simple_geometry') + + +def row(summary): + t = summary.get('throughput_window') or {} + gpu = summary['telemetry']['gpu'] + ib = summary['telemetry']['ib'].get('by_role', {}) + nvlink = summary['telemetry']['nvlink'].get('by_role', {}) + quality = {} + if 'benchmark' in summary['quality']: + bench = summary['quality']['benchmark'] + quality = {d: {'final': bench[d]['accuracy']} for d in DOMAINS if d in bench} + else: + evals = summary['quality']['evaluations'] + if evals: + quality = {d: {'initial': evals[0][d]['accuracy'], 'final': evals[-1][d]['accuracy']} for d in DOMAINS if d in evals[0]} + roles = summary['topology']['node_roles'] + trained_role = 'trainer' if 'trainer' in gpu else 'shared' + sampling_role = 'inference' if 'inference' in gpu else 'shared' + return { + 'framework': summary['framework'], 'run': summary['name'], 'workload': summary['workload'], + 'complete': summary['complete'], 'steps': len(summary['steps']), 'wall_seconds': summary['window']['seconds'], + 'median_step_seconds': (summary.get('step_seconds') or {}).get('median'), + 'tokens_per_second': t.get('tokens_per_second'), + 'tokens_per_second_per_physical_gpu': t.get('tokens_per_second_per_physical_gpu'), + 'tokens_per_second_per_trainer_gpu': t.get('tokens_per_second_per_trainer_gpu'), + 'trainer_gpus': summary['topology']['trainer_gpus'], 'inference_gpus': summary['topology']['inference_gpus'], + 'node_roles': roles, + 'trainer_gpu_utilization_pct': gpu.get(trained_role, {}).get('utilization_mean_pct'), + 'inference_gpu_utilization_pct': gpu.get(sampling_role, {}).get('utilization_mean_pct'), + 'all_gpu_power_watts': gpu.get('all', {}).get('power_mean_watts'), + 'nvlink_tx_gigabytes_per_second': nvlink.get('all', {}).get('tx_mean_gigabytes_per_second'), + 'ib_xmit_gigabytes_per_second': ib.get('all', {}).get('xmit_mean_gigabytes_per_second'), + 'quality': quality, + } + + +def fmt(value, digits=0, pct=False): + if value is None: + return 'n/a' + return f'{100 * value:.1f}%' if pct else f'{value:,.{digits}f}' + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument('campaign', type=Path) + p.add_argument('--output', type=Path, required=True) + args = p.parse_args() + rows = [row(json.loads(path.read_text())) for path in sorted(args.campaign.glob('*/runs/*.summary.json'))] + args.output.mkdir(parents=True, exist_ok=True) + (args.output/'comparison.json').write_text(json.dumps(rows, indent=2)) + lines = ['# Self-hosted Tinker-compatible endpoints: SFT, RL and MOPD', '', + 'Every run used the same client code (hosted/opd) and 32 B200 GPUs. Throughput is steps 2..N; see ' + 'hosted/selfhosted/summarize.py for the definitions.', '', + '| Framework | Run | Complete | Steps | Wall (min) | Median step (s) | Tokens/s | Tokens/s/GPU | Trainer GPU util | ' + 'Sampler GPU util | NVLink TX GB/s | IB TX GB/s |', + '|---|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|'] + for r in rows: + lines.append(f"| {r['framework']} | {r['run']} | {'yes' if r['complete'] else 'no'} | {r['steps']} | " + f"{fmt(r['wall_seconds'] / 60, 1)} | {fmt(r['median_step_seconds'], 1)} | {fmt(r['tokens_per_second'])} | " + f"{fmt(r['tokens_per_second_per_physical_gpu'])} | {fmt(r['trainer_gpu_utilization_pct'], 1)} | " + f"{fmt(r['inference_gpu_utilization_pct'], 1)} | {fmt(r['nvlink_tx_gigabytes_per_second'], 1)} | " + f"{fmt(r['ib_xmit_gigabytes_per_second'], 2)} |") + lines += ['', '## Quality (avg@3)', '', '| Framework | Run | caesar_cipher | simple_geometry |', '|---|---|---|---|'] + for r in rows: + cells = [] + for d in DOMAINS: + q = r['quality'].get(d) + cells.append('' if not q else (f"{fmt(q.get('initial'), pct=True)} -> {fmt(q['final'], pct=True)}" + if 'initial' in q else fmt(q['final'], pct=True))) + lines.append(f"| {r['framework']} | {r['run']} | {cells[0]} | {cells[1]} |") + (args.output/'REPORT.md').write_text('\n'.join(lines) + '\n') + print(args.output/'REPORT.md') + + +if __name__ == '__main__': + main() diff --git a/hosted/selfhosted/shm_clean.py b/hosted/selfhosted/shm_clean.py new file mode 100644 index 0000000..7e317be --- /dev/null +++ b/hosted/selfhosted/shm_clean.py @@ -0,0 +1,25 @@ +"""Delete /dev/shm files that no process holds open or maps: segments leaked by killed vLLM, PyTorch and NCCL workers.""" +import os +from pathlib import Path + +held = set() +for proc in Path('/proc').iterdir(): + if not proc.name.isdigit(): + continue + try: + for fd in (proc/'fd').iterdir(): + target = os.readlink(fd) + if target.startswith('/dev/shm/'): + held.add(target.split(' (deleted)')[0]) + for line in (proc/'maps').read_text().splitlines(): + if '/dev/shm/' in line: + held.add(line[line.index('/dev/shm/'):].split(' (deleted)')[0]) + except (PermissionError, FileNotFoundError, ProcessLookupError): + continue +freed = removed = 0 +for path in Path('/dev/shm').iterdir(): + if path.is_file() and str(path) not in held: + freed += path.stat().st_size + path.unlink() + removed += 1 +print(f'{os.uname().nodename}: removed {removed} unheld /dev/shm files ({freed / 2**30:.1f} GiB), kept {len(held)} held') diff --git a/hosted/selfhosted/skyrl/none-metric.patch b/hosted/selfhosted/skyrl/none-metric.patch new file mode 100644 index 0000000..8f19a1d --- /dev/null +++ b/hosted/selfhosted/skyrl/none-metric.patch @@ -0,0 +1,22 @@ +diff --git a/skyrl/tinker/proto_serialization.py b/skyrl/tinker/proto_serialization.py +index 7a85fbf0..9120b5cb 100644 +--- a/skyrl/tinker/proto_serialization.py ++++ b/skyrl/tinker/proto_serialization.py +@@ -17,6 +17,7 @@ these via server-side client_config flags and fall back to JSON): + the same endpoint via the ``forward_only`` flag instead of ``/api/v1/forward``. + The parser here is the inverse of the SDK's ``tinker/proto/request_conv.py``. + """ ++import logging + + import base64 + from collections.abc import Iterable, Sequence +@@ -190,6 +191,9 @@ def _serialize_forward_backward_output(result_data: dict) -> bytes: + proto = pb.ForwardBackwardOutput() + proto.loss_fn_output_type = output.loss_fn_output_type + for name, value in output.metrics.items(): ++ if value is None: # ClusterMAX patch: a None metric crashed serialization; skip it and name it. ++ logging.getLogger(__name__).warning("forward_backward metric %s is None; not serialized", name) ++ continue + proto.metrics[name] = float(value) + + if not output.loss_fn_outputs: diff --git a/hosted/selfhosted/skyrl/serve.sbatch b/hosted/selfhosted/skyrl/serve.sbatch new file mode 100644 index 0000000..2f49a08 --- /dev/null +++ b/hosted/selfhosted/skyrl/serve.sbatch @@ -0,0 +1,99 @@ +#!/bin/bash +# The SkyRL Tinker API server (Megatron trainer, vLLM samplers) on four 8-GPU Slurm nodes, with telemetry. +# +# CAMPAIGN=/shared/.../tinker-endpoints sbatch -p PARTITION hosted/selfhosted/skyrl/serve.sbatch +# +# Layout: the Megatron LoRA trainer on two nodes (TP4, EP8), four TP4 vLLM engines on the other two, not colocated, +# so sampling overlaps training. Micro-batches are bin-packed up to 32,768 tokens; one sequence per micro-batch made +# SFT on the short simple_geometry traces 3x slower per token than on caesar_cipher. Every adapter (student and +# teachers) is served by the same engines. The server never expires a session (--session-timeout-sec -1), because +# SkyRL unloads a stale session's adapter and a teacher must stay servable after its SFT client exits. The cluster +# steps (shm cleanup, telemetry, NCCL on InfiniBand, the Ray cluster) are in ../common.sh. +# +# Inputs under $CAMPAIGN: images/skyrl-megatron.sqsh, src/skyrl (the SkyRL checkout, with .venv from `uv sync +# --extra tinker --extra megatron`), hf/ (a Hugging Face cache holding the base model), telemetry-scripts/ (the +# ClusterMAX gpu_monitor.sh, fabric_monitor.sh and system_snapshot.sh). Outputs go to $CAMPAIGN/skyrl/server-: +# server.json (endpoint, revisions, backend config), node-roles.json, telemetry/, logs/. The job serves until cancelled. +#SBATCH -J skyrl-tinker +#SBATCH -N 4 +#SBATCH --exclusive +#SBATCH --gres=gpu:8 +#SBATCH --time=UNLIMITED +#SBATCH -o %x-%j.out +set -euo pipefail +: "${CAMPAIGN:?CAMPAIGN must be set}" +PORT=${PORT:-8000} +BASE_MODEL=${BASE_MODEL:-Qwen/Qwen3.6-35B-A3B} +LORA_ALPHA=${LORA_ALPHA:-32} +SRC=$CAMPAIGN/src/skyrl +source "$CAMPAIGN/opd/hosted/selfhosted/common.sh" +# SkyRL at 7d94ccf crashes serializing a forward_backward metric that is None (importance_ratio:mean with +# loss_fn=importance_sampling); none-metric.patch skips such a metric and logs its name. +PATCH=$SELFHOSTED/skyrl/none-metric.patch +git -C "$SRC" apply --reverse --check "$PATCH" 2>/dev/null || git -C "$SRC" apply "$PATCH" +selfhosted_setup skyrl "$CAMPAIGN/images/skyrl-megatron.sqsh" skyrl-megatron +mkdir -p "$RUN/lora-sync" "$RUN/checkpoints" +export VLLM_USE_FLASHINFER_MOE_FP16=0 # Blackwell: the flashinfer bf16 MoE kernel needs a multiple-of-128 expert size. +export FLA_TILELANG=0 # Blackwell: the TileLang GatedDeltaNet backward aborts. +export VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS=1800 +selfhosted_ray "$SRC/.venv/bin" + +BACKEND_CONFIG=$(cat < -m skyrl.tinker.api`: it rebuilds that command to spawn its engine. +# --frozen --no-sync keeps both on the installed .venv. +cd "$SRC" +srun --overlap -N1 -w "$HEAD" "${CONTAINER[@]}" bash -c "cd $SRC && export RAY_ADDRESS=$HEAD_IP:6379 \ + UV_PROJECT_ENVIRONMENT=$SRC/.venv UV_CACHE_DIR=$CAMPAIGN/uv-cache && \ + uv run --frozen --no-sync --extra tinker --extra megatron -m skyrl.tinker.api \ + --host 0.0.0.0 --port $PORT --base-model $BASE_MODEL --backend megatron \ + --checkpoints-base $RUN/checkpoints --database-url sqlite:///$RUN/tinker.db --session-timeout-sec -1 \ + --backend-config '$BACKEND_CONFIG'" > "$RUN/logs/server.log" 2>&1 & +SERVER=$! +wait $SERVER || true +selfhosted_finish diff --git a/hosted/selfhosted/smoke.py b/hosted/selfhosted/smoke.py new file mode 100644 index 0000000..ebbd752 --- /dev/null +++ b/hosted/selfhosted/smoke.py @@ -0,0 +1,150 @@ +"""Check that a Tinker-compatible server supports every call sft.py, rl.py, opd.py and mopd.py make. + + TINKER_BASE_URL=http://HOST:PORT python hosted/selfhosted/smoke.py --backend skyrl --output /abs/smoke.json + +Steps, each timed and recorded: capabilities; a fresh LoRA; sampling with logprobs, a stop sequence and a seed; +prefill logprobs (compute_logprobs); cross_entropy, importance_sampling and ppo forward_backward on a +--long-tokens datum; optim_step; save_state; save_weights_for_sampler and sampling from that path; restoring +from the state with its optimizer. `--teacher PATH` also samples a sampler path saved by an earlier process, +which is how opd.py and mopd.py use SFT teachers. The first failing step stops the check and is recorded. +""" +import argparse +import json +from pathlib import Path +import sys +import time +import traceback + +sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'opd')) +import tinker # noqa: E402 + +from backend import BACKENDS # noqa: E402 +from common import DEFAULT_MODEL, MODELS, load_renderer # noqa: E402 + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument('--backend', choices=sorted(BACKENDS), required=True) + p.add_argument('--output', type=Path, required=True) + p.add_argument('--rank', type=int, default=64) + p.add_argument('--long-tokens', type=int, default=30_000, help='Length of the training datum') + p.add_argument('--teacher', help='A sampler path saved by another process') + args = p.parse_args() + model = MODELS[DEFAULT_MODEL] + tokenizer, renderer = load_renderer(model) + record = {'backend': args.backend, 'steps': []} + + def step(name, fn): + start = time.monotonic() + try: + detail = fn() + except Exception as error: + record['steps'].append({'step': name, 'ok': False, 'seconds': time.monotonic() - start, + 'error': f'{type(error).__name__}: {error}'[:2000], 'traceback': traceback.format_exc()}) + args.output.write_text(json.dumps(record, indent=2)) + raise SystemExit(f'FAILED {name}: {error}') + record['steps'].append({'step': name, 'ok': True, 'seconds': time.monotonic() - start, **(detail or {})}) + print(json.dumps(record['steps'][-1]), flush=True) + args.output.write_text(json.dumps(record, indent=2)) + + state = {} + messages = [{'role': 'user', 'content': 'What is 17 * 23? Answer with N.'}] + prompt = tinker.ModelInput.from_ints(renderer.build_generation_prompt(messages).to_ints()) + params = tinker.SamplingParams(max_tokens=256, temperature=1.0, top_p=1.0, top_k=-1, + stop=renderer.get_stop_sequences(), seed=7) + + def connect(): + state['backend'] = BACKENDS[args.backend](experiment='smoke') + record['account'] = state['backend'].account + return {'models': [m['model_name'] for m in state['backend'].capabilities(model[0])['supported_models']]} + + def new_lora(): + state['train'] = state['backend'].new_student(model[0], args.rank, 20260921, {'purpose': 'smoke'}) + try: + return {'info': json.loads(state['train'].get_info().model_dump_json())} + except Exception as error: + return {'info_unavailable': str(error)[:300]} + + def sample(client_key, n=2): + def fn(): + client = state[client_key] + result = client.sample(prompt, num_samples=n, sampling_params=params).result() + seqs = result.sequences + assert len(seqs) == n and all(len(s.tokens) == len(s.logprobs) for s in seqs) + return {'lengths': [len(s.tokens) for s in seqs], 'stop_reasons': [s.stop_reason for s in seqs], + 'text': tokenizer.decode(seqs[0].tokens)[-200:]} + return fn + + def base_sampler(): + state['sampler'] = state['backend'].sync(state['train'], 'smoke-000')[0] + + def prefill(): + ids = prompt.to_ints() + tokenizer.encode(' The answer is 391.') + lp = state['sampler'].compute_logprobs(tinker.ModelInput.from_ints(ids)).result() + assert len(lp) == len(ids) and lp[0] is None and all(x is not None for x in lp[1:]) + return {'tokens': len(ids), 'mean_logprob': sum(lp[1:]) / (len(lp) - 1)} + + def long_datum(loss): + n = args.long_tokens + ids = (tokenizer.encode('The quick brown fox jumps over the lazy dog. ') * (n // 10 + 1))[:n] + shape = [n - 1] + if loss == 'cross_entropy': + inputs = {'target_tokens': tinker.TensorData(data=ids[1:], dtype='int64', shape=shape), + 'weights': tinker.TensorData(data=[1.0] * (n - 1), dtype='float32', shape=shape)} + else: + inputs = {'target_tokens': tinker.TensorData(data=ids[1:], dtype='int64', shape=shape), + 'logprobs': tinker.TensorData(data=[-1.0] * (n - 1), dtype='float32', shape=shape), + 'advantages': tinker.TensorData(data=[0.1] * (n - 1), dtype='float32', shape=shape)} + return tinker.Datum(model_input=tinker.ModelInput.from_ints(ids[:-1]), loss_fn_inputs=inputs) + + def forward_backward(loss, config=None): + def fn(): + out = state['train'].forward_backward([long_datum(loss)], loss_fn=loss, loss_fn_config=config).result() + return {'metrics': out.metrics, 'outputs': sorted(out.loss_fn_outputs[0])} + return fn + + def optim(): + out = state['train'].optim_step(tinker.AdamParams(learning_rate=1e-5, beta1=0.9, beta2=0.999, eps=1e-8, + weight_decay=0.0, grad_clip_norm=1.0)).result() + return {'metrics': out.metrics} + + def save_state(): + state['state_path'] = state['train'].save_state(name='smoke-state').result().path + return {'path': state['state_path']} + + def save_sampler(): + state['sampler'], path = state['backend'].sync(state['train'], 'smoke-001') + return {'path': path} + + def restore(): + state['restored'] = state['backend'].resume_student(state['state_path']) + state['restored_sampler'], path = state['backend'].sync(state['restored'], 'smoke-restored') + return {'path': path} + + def teacher(): + state['teacher'] = state['backend'].sampler(path=args.teacher) + return sample('teacher')() + + step('capabilities', connect) + step('create_lora_training_client', new_lora) + step('save_weights_for_sampler (initial)', base_sampler) + step('sample (logprobs, stop, seed)', sample('sampler')) + step('compute_logprobs', prefill) + step(f'forward_backward cross_entropy ({args.long_tokens} tokens)', forward_backward('cross_entropy')) + step(f'forward_backward importance_sampling ({args.long_tokens} tokens)', forward_backward('importance_sampling')) + step(f'forward_backward ppo ({args.long_tokens} tokens)', + forward_backward('ppo', {'clip_low_threshold': 0.8, 'clip_high_threshold': 1.2})) + step('optim_step', optim) + step('save_state', save_state) + step('save_weights_for_sampler and sample', lambda: (save_sampler(), sample('sampler')())[1]) + step('create_training_client_from_state_with_optimizer', restore) + step('sample restored', sample('restored_sampler')) + if args.teacher: + step('sample a sampler path from an earlier process', teacher) + record['ok'] = True + args.output.write_text(json.dumps(record, indent=2)) + print('SMOKE OK', flush=True) + + +if __name__ == '__main__': + main() diff --git a/hosted/selfhosted/summarize.py b/hosted/selfhosted/summarize.py new file mode 100644 index 0000000..a8821bd --- /dev/null +++ b/hosted/selfhosted/summarize.py @@ -0,0 +1,196 @@ +"""Summarize one self-hosted client run (sft.py, rl.py, opd.py or mopd.py) for PostTrainingX. + + python hosted/selfhosted/summarize.py RUNS_DIR/NAME.window.json OUTPUT_DIR --telemetry-code DIR + +NAME.window.json (written by client.sh) names the server directory and the run's UTC window. OUTPUT_DIR is the run +directory the client wrote. The summary goes to RUNS_DIR/NAME.summary.json: + +- framework, workload, server description (server.json), exit code, window; +- topology: nodes, physical GPUs, and trainer and inference nodes (server.json `node_roles`, else inferred from GPU + memory: an inference node's engines reserve at least 80% of every GPU before the run); +- the protocol (config.json) and one record per optimizer step with its UTC completion time, tokens and timings; +- the throughput window: tokens trained in steps 2..N over the time from step 1's completion to step N's, per + second, per physical GPU (the whole allocation) and per trainer GPU. Step 1 is excluded because it carries + compilation and engine start. RL and OPD tokens are prompt plus response tokens of every trained sequence; SFT + tokens are prompt plus completion tokens; +- quality: the evaluations (opd, mopd, rl) or the teacher benchmark (sft); +- telemetry cut to the window and summarized by the ClusterMAX collector's own functions (GPU utilization, power, + memory and clocks by role; NVLink, InfiniBand and Ethernet bytes by role). +""" +import argparse +import csv +from datetime import datetime +import importlib.util +import json +from pathlib import Path +import statistics +import tempfile + +GPUS_PER_NODE = 8 + + +def utc(text): + return datetime.fromisoformat(text.replace('Z', '+00:00')) + + +def load_collector(directory): + """The ClusterMAX miles-terminal-lego collect_results.py, for its telemetry summarizers.""" + spec = importlib.util.spec_from_file_location('clustermax_collect', Path(directory)/'collect_results.py') + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def cut_csv(source, target, start, end): + """Copy the rows of a telemetry CSV whose collector_time_utc is inside [start, end].""" + target.parent.mkdir(parents=True, exist_ok=True) + with source.open(newline='') as f, target.open('w', newline='') as g: + reader = csv.DictReader(f) + writer = csv.DictWriter(g, fieldnames=reader.fieldnames) + writer.writeheader() + for row in reader: + try: + stamp = utc(row['collector_time_utc']) + except (KeyError, ValueError): + continue + if start <= stamp <= end: + writer.writerow(row) + + +def server_roles(server_dir): + """Trainer and inference nodes from the server's own logs: the SkyRL server log names its vLLM servers' URLs.""" + ips = json.loads((server_dir/'node-ips.json').read_text()) if (server_dir/'node-ips.json').is_file() else {} + log = server_dir/'logs'/'server.log' + if not ips or not log.is_file(): + return None + import re + text = log.read_text(errors='replace').replace('\n', '') + match = re.search(r"server_urls=\[([^\]]*)\]", re.sub(r'\s+', '', text)) + if not match: + return None + engine_ips = set(re.findall(r'//([0-9.]+):', match.group(1))) + inference = sorted(n for n, ip in ips.items() if ip in engine_ips) + return {'trainer': sorted(n for n in ips if n not in inference), 'inference': inference} + + +def infer_roles(gpu_dir, start): + """Inference nodes are those whose every GPU already holds at least 80% of its memory at the run's start.""" + roles = {'trainer': [], 'inference': []} + for path in sorted(gpu_dir.glob('*.csv')): + with path.open(newline='') as f: + rows = [r for r in csv.DictReader(f) if r.get('collector_time_utc') and utc(r['collector_time_utc']) >= start] + first = rows[:GPUS_PER_NODE] + if not first: + continue + full = all(float(r['memory.used']) >= 0.8 * float(r['memory.total']) for r in first) + roles['inference' if full else 'trainer'].append(path.stem) + return roles + + +def read_jsonl(path): + return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] if path.is_file() else [] + + +def step_records(output, workload): + """(step, completion time, tokens, record) per optimizer step.""" + rows = [] + if workload == 'sft': + for m in read_jsonl(output/'metrics.jsonl'): + rows.append({'step': m['step'], 'utc': m['utc'], 'tokens': m['prompt_tokens'] + m['completion_tokens'], + 'prompt_tokens': m['prompt_tokens'], 'completion_tokens': m['completion_tokens'], + 'sequences': m['sequences'], 'train_nll': m['train_nll'], 'learning_rate': m['learning_rate'], + 'interval_seconds': m['interval_seconds'], 'step_seconds': m['step_seconds']}) + return rows + completed = {e['update']: e['utc'] for e in read_jsonl(output/'events.jsonl') if e['event'] == 'update_complete'} + for m in read_jsonl(output/'metrics.jsonl'): + row = {'step': m['update'], 'utc': completed.get(m['update']), 'tokens': m['prompt_tokens'] + m['response_tokens'], + 'prompt_tokens': m['prompt_tokens'], 'response_tokens': m['response_tokens'], 'responses': m['responses'], + 'policy_lag': m['policy_lag'], 'truncated': m['truncated'], 'timing': m['timing'], + 'throughput': m['throughput'], 'by_domain': m['by_domain'], + 'train_sampler_logprob_delta_abs_mean': m.get('train_sampler_logprob_delta_abs_mean')} + if 'reward' in m: + row['reward'] = m['reward'] + rows.append(row) + return rows + + +def throughput(steps, physical_gpus, trainer_gpus): + timed = [s for s in steps if s.get('utc')] + if len(timed) < 2: + return None + first, last = timed[0], timed[-1] + seconds = (utc(last['utc']) - utc(first['utc'])).total_seconds() + tokens = sum(s['tokens'] for s in timed[1:]) + rate = tokens / seconds + return {'first_step': first['step'], 'last_step': last['step'], 'seconds': seconds, 'tokens': tokens, + 'tokens_per_second': rate, 'tokens_per_second_per_physical_gpu': rate / physical_gpus, + 'tokens_per_second_per_trainer_gpu': rate / trainer_gpus if trainer_gpus else None, + 'steps_per_hour': 3600 * (len(timed) - 1) / seconds} + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument('window', type=Path) + p.add_argument('output', type=Path) + p.add_argument('--telemetry-code', type=Path, required=True, help='Directory holding ClusterMAX collect_results.py') + args = p.parse_args() + window = json.loads(args.window.read_text()) + server_dir = Path(window['server']) + server = json.loads((server_dir/'server.json').read_text()) + config = json.loads((args.output/'config.json').read_text()) if (args.output/'config.json').is_file() else {} + workload = 'sft' if config.get('mode') == 'train' else config.get('kind') + start, end = utc(window['start_utc']), utc(window['end_utc']) + + collector = load_collector(args.telemetry_code) + with tempfile.TemporaryDirectory() as scratch: + scratch = Path(scratch) + for source in (server_dir/'telemetry').rglob('*.csv'): + relative = source.relative_to(server_dir) + if relative.parts[2:3] == ('ib-perfquery',): # perfquery samples replace fabric_monitor.sh's empty IB rows. + relative = Path('telemetry', 'fabric', 'ib', source.name) + elif relative.parts[2:3] == ('ib',) and (server_dir/'telemetry'/'fabric'/'ib-perfquery').is_dir(): + continue + cut_csv(source, scratch/relative, start, end) + roles, role_source = server.get('node_roles'), 'server.json' + if not roles: + roles, role_source = server_roles(server_dir), 'server log' + if not roles: + roles, role_source = infer_roles(scratch/'telemetry'/'gpu', start), 'gpu-memory' + (scratch/'provenance').mkdir() + (scratch/'provenance'/'node-roles.json').write_text(json.dumps(roles)) + role_sets = collector.node_roles(scratch) + telemetry = {'gpu': collector.summarize_gpu(scratch, role_sets), + 'nvlink': collector.summarize_nvlink(scratch, role_sets), + 'ib': collector.summarize_ib(scratch, role_sets), + 'net': collector.summarize_net(scratch, role_sets)} + + nodes = server['nodes'] + # A colocated server (verl-tinker's hybrid engine) trains and samples on the same 'shared' nodes. + trainer_gpus = GPUS_PER_NODE * len(roles.get('trainer', []) + roles.get('shared', [])) + steps = step_records(args.output, workload) + intervals = [(utc(b['utc']) - utc(a['utc'])).total_seconds() for a, b in zip(steps, steps[1:]) if a.get('utc') and b.get('utc')] + quality = ({'benchmark': json.loads((args.output/'benchmark.json').read_text())} if (args.output/'benchmark.json').is_file() + else {'evaluations': read_jsonl(args.output/'evaluations.jsonl')}) + summary = { + 'schema_version': 1, 'framework': server['framework'], 'workload': workload, 'name': window['name'], + 'exit_code': window['exit_code'], 'complete': window['exit_code'] == 0 and not (args.output/'error.txt').exists(), + 'window': {'start_utc': window['start_utc'], 'end_utc': window['end_utc'], 'seconds': (end - start).total_seconds()}, + 'server': server, 'command': window['command'], 'output_dir': str(args.output), + 'topology': {'nodes': nodes, 'gpus_per_node': GPUS_PER_NODE, 'physical_gpus': GPUS_PER_NODE * len(nodes), + 'node_roles': roles, 'role_source': role_source, + 'trainer_gpus': trainer_gpus, + 'inference_gpus': GPUS_PER_NODE * len(roles.get('inference', []) + roles.get('shared', []))}, + 'protocol': config, 'steps': steps, + 'step_seconds': ({'median': statistics.median(intervals), 'mean': statistics.mean(intervals), + 'min': min(intervals), 'max': max(intervals), 'basis': 'steps 2..N completion intervals'} + if intervals else None), + 'throughput_window': throughput(steps, GPUS_PER_NODE * len(nodes), trainer_gpus), + 'quality': quality, 'telemetry': telemetry, + } + target = args.window.with_name(args.window.name.replace('.window.json', '.summary.json')) + target.write_text(json.dumps(summary, indent=2, default=str)) + print(target) + + +if __name__ == '__main__': + main() diff --git a/hosted/selfhosted/verl/actor_rollout.yaml b/hosted/selfhosted/verl/actor_rollout.yaml new file mode 100644 index 0000000..901b029 --- /dev/null +++ b/hosted/selfhosted/verl/actor_rollout.yaml @@ -0,0 +1,77 @@ +# verl-tinker server for SFT and RL: a Qwen3.6-35B-A3B FSDP actor with server-side LoRA and a colocated vLLM rollout +# on all four nodes. Adapted from verl-recipe verl_tinker/configs/advance/qwen3_8b_actor_rollout_lora.yaml +# (242c73f). verl-tinker serves one trainable model, so every SFT teacher and the RL run needs its own server. +# The rank and alpha are the Tinker clients' (64, and alpha 32 as on SkyRL and Miles); verl-tinker ignores the +# client's lora_config. Lengths follow shared/recipe.py: 32,768-token context, up to 30,720 response tokens. +server: + model_name: ${oc.env:TINKER_SERVER_MODEL_NAME,Qwen/Qwen3.6-35B-A3B} + host: ${oc.env:TINKER_SERVER_HOST,0.0.0.0} + port: ${oc.env:TINKER_SERVER_PORT,8000} + ray_address: ${oc.env:RAY_ADDRESS,local} + checkpoint_dir: ${oc.env:TINKER_CHECKPOINT_DIR,/tmp/tinker-checkpoints} + server_max_runtime: null + max_concurrent_samples: ${oc.env:TINKER_SERVER_MAX_CONCURRENT_SAMPLES,512} + enable_offload: true + auto_merge_verl_default_config: true + +actor_rollout_ref: + actor: + strategy: fsdp + use_kl_loss: false # Tinker folds any KL into client-side advantages. + entropy_coeff: 0 + ppo_mini_batch_size: 32 + use_dynamic_bsz: true + ppo_max_token_len_per_gpu: 32768 + ppo_micro_batch_size_per_gpu: 1 + fsdp_config: + model_dtype: bfloat16 + param_offload: false + optimizer_offload: false + use_orig_params: true # PEFT adapter export needs the original parameter views (see the template). + optim: + lr_scheduler_type: constant + lr: 1.0e-05 + weight_decay: 0.0 + rollout: + name: vllm + mode: async + tensor_model_parallel_size: 4 + pipeline_model_parallel_size: 1 + gpu_memory_utilization: 0.7 + prompt_length: 2048 + response_length: 30720 + max_num_seqs: 256 + max_num_batched_tokens: 32768 + enable_chunked_prefill: true + calculate_log_probs: true + log_prob_use_dynamic_bsz: true + log_prob_max_token_len_per_gpu: 32768 + log_prob_micro_batch_size_per_gpu: 1 + free_cache_engine: true + enforce_eager: false + load_format: safetensors + engine_kwargs: + vllm: + disable_custom_all_reduce: true + layered_summon: false + checkpoint_engine: + update_weights_bucket_megabytes: 4096 + ref: + enable: false + model: + path: ${oc.env:TINKER_SERVER_MODEL_PATH,Qwen/Qwen3.6-35B-A3B} + use_remove_padding: true + enable_gradient_checkpointing: true + trust_remote_code: true + lora_rank: 64 + lora_alpha: 32 + target_modules: all-linear + lora_adapter_path: null + +trainer: + logger: + - console + project_name: verl_tinker_server + experiment_name: qwen3_6_35b_a3b_lora + nnodes: ${oc.env:TINKER_SERVER_NNODES,4} + n_gpus_per_node: 8 diff --git a/hosted/selfhosted/verl/mopd_teachers.yaml b/hosted/selfhosted/verl/mopd_teachers.yaml new file mode 100644 index 0000000..69fb367 --- /dev/null +++ b/hosted/selfhosted/verl/mopd_teachers.yaml @@ -0,0 +1,126 @@ +# verl-tinker server for MOPD: the actor_rollout.yaml actor and colocated rollout on two nodes, and one dedicated +# frozen vLLM teacher pool per domain on the other two (two TP4 replicas each). verl-tinker teachers load full +# models, so each is an SFT teacher merged into the base model by merge_teacher.py, under $TINKER_TEACHERS_DIR. +# Clients reach a teacher by its model_name: teachers.json holds base:caesar_cipher-teacher and +# base:simple_geometry-teacher, which the Verl backend turns into create_sampling_client(base_model=...). +server: + model_name: ${oc.env:TINKER_SERVER_MODEL_NAME,Qwen/Qwen3.6-35B-A3B} + host: ${oc.env:TINKER_SERVER_HOST,0.0.0.0} + port: ${oc.env:TINKER_SERVER_PORT,8000} + ray_address: ${oc.env:RAY_ADDRESS,local} + checkpoint_dir: ${oc.env:TINKER_CHECKPOINT_DIR,/tmp/tinker-checkpoints} + server_max_runtime: null + max_concurrent_samples: ${oc.env:TINKER_SERVER_MAX_CONCURRENT_SAMPLES,512} + enable_offload: true + auto_merge_verl_default_config: true + +actor_rollout_ref: + actor: + strategy: fsdp + use_kl_loss: false # Tinker folds any KL into client-side advantages. + entropy_coeff: 0 + ppo_mini_batch_size: 32 + use_dynamic_bsz: true + ppo_max_token_len_per_gpu: 32768 + ppo_micro_batch_size_per_gpu: 1 + fsdp_config: + model_dtype: bfloat16 + param_offload: false + optimizer_offload: false + use_orig_params: true # PEFT adapter export needs the original parameter views (see the template). + optim: + lr_scheduler_type: constant + lr: 1.0e-05 + weight_decay: 0.0 + rollout: + name: vllm + mode: async + tensor_model_parallel_size: 4 + pipeline_model_parallel_size: 1 + gpu_memory_utilization: 0.7 + prompt_length: 2048 + response_length: 30720 + max_num_seqs: 256 + max_num_batched_tokens: 32768 + enable_chunked_prefill: true + calculate_log_probs: true + log_prob_use_dynamic_bsz: true + log_prob_max_token_len_per_gpu: 32768 + log_prob_micro_batch_size_per_gpu: 1 + free_cache_engine: true + enforce_eager: false + load_format: safetensors + engine_kwargs: + vllm: + disable_custom_all_reduce: true + layered_summon: false + checkpoint_engine: + update_weights_bucket_megabytes: 4096 + ref: + enable: false + model: + path: ${oc.env:TINKER_SERVER_MODEL_PATH,Qwen/Qwen3.6-35B-A3B} + use_remove_padding: true + enable_gradient_checkpointing: true + trust_remote_code: true + lora_rank: 64 + lora_alpha: 32 + target_modules: all-linear + lora_adapter_path: null + +distillation: + enabled: true + nnodes: 2 + n_gpus_per_node: 8 + dedicated_resource_pools: true + teacher_models: + caesar_cipher_teacher: + key: caesar_cipher + model_name: caesar_cipher-teacher + model_path: ${oc.env:TINKER_TEACHERS_DIR}/caesar_cipher + num_replicas: 2 + inference: + name: vllm + tensor_model_parallel_size: 4 + data_parallel_size: 1 + pipeline_model_parallel_size: 1 + gpu_memory_utilization: 0.8 + prompt_length: 2048 + response_length: 30720 + max_model_len: 32768 + max_num_seqs: 256 + max_num_batched_tokens: 32768 + enable_chunked_prefill: true + enable_prefix_caching: true + engine_kwargs: + vllm: + disable_custom_all_reduce: true + simple_geometry_teacher: + key: simple_geometry + model_name: simple_geometry-teacher + model_path: ${oc.env:TINKER_TEACHERS_DIR}/simple_geometry + num_replicas: 2 + inference: + name: vllm + tensor_model_parallel_size: 4 + data_parallel_size: 1 + pipeline_model_parallel_size: 1 + gpu_memory_utilization: 0.8 + prompt_length: 2048 + response_length: 30720 + max_model_len: 32768 + max_num_seqs: 256 + max_num_batched_tokens: 32768 + enable_chunked_prefill: true + enable_prefix_caching: true + engine_kwargs: + vllm: + disable_custom_all_reduce: true + +trainer: + logger: + - console + project_name: verl_tinker_server + experiment_name: qwen3_6_35b_a3b_lora + nnodes: 2 + n_gpus_per_node: 8 diff --git a/hosted/selfhosted/verl/serve.sbatch b/hosted/selfhosted/verl/serve.sbatch new file mode 100644 index 0000000..c570570 --- /dev/null +++ b/hosted/selfhosted/verl/serve.sbatch @@ -0,0 +1,53 @@ +#!/bin/bash +# The verl-tinker server (verl-recipe verl_tinker: FSDP LoRA actor, colocated vLLM rollout, optional frozen vLLM +# teachers) on four 8-GPU Slurm nodes, with telemetry. +# +# CAMPAIGN=/shared/... CONFIG=actor_rollout.yaml sbatch -p PARTITION hosted/selfhosted/verl/serve.sbatch +# +# verl-tinker holds one trainable model, so each training client (every SFT teacher, RL, MOPD) gets its own server +# job. CONFIG is a file in this directory: actor_rollout.yaml (SFT, RL) or mopd_teachers.yaml (MOPD, with the +# merged SFT teachers from merge_teacher.py). The runtime is verl-recipe's own `uv sync --project verl_tinker` +# environment in $CAMPAIGN/src/verl-recipe/verl_tinker/.venv, run inside the CUDA 13 SkyRL image. The server +# speaks JSON only, so clients use tinker 0.19 (the version it pins); 0.23 and later send protobuf. +# Outputs go to $CAMPAIGN/verl/server-: server.json, telemetry/, logs/, checkpoints/. Serves until cancelled. +#SBATCH -J verl-tinker +#SBATCH -N 4 +#SBATCH --exclusive +#SBATCH --gres=gpu:8 +#SBATCH --time=UNLIMITED +#SBATCH -o %x-%j.out +set -euo pipefail +: "${CAMPAIGN:?CAMPAIGN must be set}" +: "${CONFIG:=actor_rollout.yaml}" +PORT=${PORT:-8000} +SRC=$CAMPAIGN/src/verl-recipe +VENV=$SRC/verl_tinker/.venv +source "$CAMPAIGN/opd/hosted/selfhosted/common.sh" +selfhosted_setup verl "$CAMPAIGN/images/skyrl-megatron.sqsh" skyrl-megatron +mkdir -p "$RUN/checkpoints" +cp "$SELFHOSTED/verl/$CONFIG" "$RUN/config.yaml" +export TINKER_SERVER_MODEL_NAME=Qwen/Qwen3.6-35B-A3B TINKER_SERVER_PORT=$PORT TINKER_CHECKPOINT_DIR=$RUN/checkpoints +export TINKER_SERVER_MODEL_PATH=$CAMPAIGN/hf/hub/models--Qwen--Qwen3.6-35B-A3B/snapshots/995ad96eacd98c81ed38be0c5b274b04031597b0 +export TINKER_TEACHERS_DIR=$CAMPAIGN/verl/teachers VLLM_USE_FLASHINFER_MOE_FP16=0 FLA_TILELANG=0 +selfhosted_ray "$VENV/bin" + +python3 - "$RUN/server.json" < "$RUN/logs/server.log" 2>&1 & +SERVER=$! +for _ in $(seq 180); do # Readiness is healthz reporting ready, not an open port (actor and rollout load after it). + curl -sf "http://$HEAD_IP:$PORT/api/v1/healthz" | grep -q ready && { echo "ready $(date -u +%FT%TZ)" > "$RUN/ready"; break; } + kill -0 $SERVER 2>/dev/null || break + sleep 20 +done +wait $SERVER || true +selfhosted_finish