diff --git a/cvs/input/config_file/inference/sglang/mi30x_sglang_distributed.json b/cvs/input/config_file/inference/sglang/mi30x_sglang_distributed.json new file mode 100644 index 00000000..d2f8e108 --- /dev/null +++ b/cvs/input/config_file/inference/sglang/mi30x_sglang_distributed.json @@ -0,0 +1,188 @@ +{ + "config": + { + "container_image": "rocm/sgl-dev:v0.5.12.post1-rocm720-mi30x-20260603", + "_container_image": "rocm/sgl-dev:v0.5.12.post1-rocm720-mi30x-20260601", + "container_name": "sglang_container", + "_example_nnodes": "4", + "nnodes": "2", + "hf_token_file": "/home/{user-id}/.hf_token", + "shm_size": "128G", + "_log_dir_comments": "Provide some common file system that is accessible from any node", + "log_dir": "/home/{user-id}/LOGS/sglang", + "log_level": "info", + "nic_type": "thor2", + "_example_nccl_ib_hca_list": "rdma0,rdma1,rdma2,rdma3,rdma4,rdma5,rdma6,rdma7", + "nccl_ib_hca_list": "", + "_example_nccl_ib_hca": "rdma0,rdma1,rdma2,rdma3,rdma4,rdma5,rdma6,rdma7", + "nccl_ib_hca": "", + "_example_nccl_socket_ifname": "eno0", + "nccl_socket_ifname": "", + "_example_gloo_socket_ifname": "eno0", + "gloo_socket_ifname": "", + "_example_gloo_tcp_ifname": "eno0", + "gloo_tcp_ifname": "", + "nccl_ib_gid_index": "3", + "nccl_debug": "ERROR", + "prefill_node_list": ["", ""], + "decode_node_list": ["", ""], + "proxy_router_node": "", + "benchmark_serv_node": "", + "prefill_serv_port": "30001", + "decode_serv_port": "30002", + "proxy_router_port": "8000", + "_prefill_coordinator_addr": "This is the master address for co-ordination among all prefill nodes", + "prefill_coordinator_addr": "", + "_decode_coordinator_addr": "This is the master address for co-ordination among all decode nodes", + "decode_coordinator_addr": "", + "prefill_coordinator_port": "40001", + "decode_coordinator_port": "40002", + "proxy_router_serv_port": "8000", + "container_config": + { + "device_list": [ "/dev/dri", "/dev/kfd", "/dev/infiniband/rdma_cm" ], + "volume_dict": + { + "/home/{user-id}": "/home/{user-id}", + "/mnt/dtni/models": "/root/models", + "/dev/infiniband": "/dev/infiniband", + "/usr/local/lib/libbnxt_re-rdmav34.so": "/usr/lib/x86_64-linux-gnu/libibverbs/libbnxt_re-rdmav34.so.host", + "/lib/libibverbs.d": "/lib/libibverbs.d" + }, + "env_dict": + { + } + } + + }, + "benchmark_params": + { + "llama-70b": + { + "backend": "sglang", + "max_concurrency": "25", + "model": "meta-llama/Llama-3.1-70B-Instruct", + "prefill_policy": "cache_aware", + "decode_policy": "cache_aware", + "tensor_parallelism": "8", + "memory_fraction": "0.85", + "tokenizer_mode": "auto", + "inference_poll_iterations": "16", + "inference_tests": + { + "gsm8k": + { + "backend": "sglang", + "num_questions": "1000", + "max_concurrency": "25", + "expected_results": + { + "auto": + { + "tokens_per_sec": "350" + } + } + }, + "bench_serv_random": + { + "backend": "sglang", + "data_set_name": "random", + "num_prompts": "100", + "input_length": "1024", + "output_length": "1024", + "random_range_ratio": "0.5", + "expected_results": + { + "auto": + { + "output_throughput_per_sec": "1000", + "mean_ttft_ms": "60000", + "mean_tpot_ms": "150" + } + } + }, + "bench_serv_generated_shared_prefix": + { + "backend": "sglang", + "gsp_num_groups": "1", + "gsp_prompts_per_group": "16", + "gsp_system_prompt_len": "0", + "gsp_question_len": "1024", + "gsp_output_len": "1024", + "expected_results": + { + "auto": + { + } + } + } + + } + }, + "deepseek-r1": + { + "backend": "sglang", + "max_concurrency": "64", + "_comments_model": "If the model is local, specify the full path of the model", + "model": "/root/models/DeepSeek-R1-0528", + "prefill_policy": "cache_aware", + "decode_policy": "cache_aware", + "tensor_parallelism": "16", + "memory_fraction": "0.85", + "tokenizer_mode": "auto", + "inference_poll_iterations": "16", + "inference_tests": + { + "gsm8k": + { + "backend": "sglang", + "num_questions": "1000", + "max_concurrency": "100", + "expected_results": + { + "auto": + { + "tokens_per_sec": "700" + } + } + }, + "bench_serv_random": + { + "backend": "sglang", + "data_set_name": "random", + "num_prompts": "100", + "input_length": "1024", + "output_length": "1024", + "random_range_ratio": "0.5", + "expected_results": + { + "auto": + { + "output_throughput_per_sec": "1400", + "mean_ttft_ms": "60000", + "mean_tpot_ms": "110" + } + } + }, + "bench_serv_generated_shared_prefix": + { + "backend": "sglang", + "gsp_num_groups": "1", + "gsp_prompts_per_group": "16", + "gsp_system_prompt_len": "0", + "gsp_question_len": "1024", + "gsp_output_len": "1024", + "expected_results": + { + "auto": + { + } + } + } + + } + } + + } + +} diff --git a/cvs/lib/sglang_disagg_lib.py b/cvs/lib/sglang_disagg_lib.py index bf41e239..5748a787 100644 --- a/cvs/lib/sglang_disagg_lib.py +++ b/cvs/lib/sglang_disagg_lib.py @@ -267,15 +267,16 @@ def exec_nic_setup_scripts( self.nccl_ib_gid_index = 3 cmd = f'docker exec {self.container_name} /bin/bash -c "sudo \ cp /usr/lib/x86_64-linux-gnu/libibverbs/libbnxt_re-rdmav34.so.host \ - /usr/lib/x86_64-linux-gnu/libibverbs/libbnxt_re-rdmav34.so;" ' + /usr/lib/x86_64-linux-gnu/libibverbs/libbnxt_re-rdmav34.so; \ + sleep 2; ibv_devinfo; sleep 2;" ' pout_dict = self.p_phdl.exec(cmd) dout_dict = self.d_phdl.exec(cmd) for node in pout_dict.keys(): - if not re.search('hca_id:\s+bnxt_', pout_dict[node], re.I): + if not re.search('hca_id:\s+(bnxt_|rocep)', pout_dict[node], re.I): log.info("%s", pout_dict[node]) fail_test(f'Broadcom libbnxt rdma driver is not properly copied on node {node}') for node in dout_dict.keys(): - if not re.search('hca_id:\s+bnxt_', dout_dict[node], re.I): + if not re.search('hca_id:\s+(bnxt_|rocep)', dout_dict[node], re.I): log.info("%s", dout_dict[node]) fail_test(f'Broadcom libbnxt rdma driver is not properly copied on node {node}') @@ -500,6 +501,7 @@ def launch_prefill_servers(self, dtype='auto', kv_cache_dtype='auto'): cmd_list = [] prefill_node_list = self.inf_dict['prefill_node_list'] log.info('%%%% self.prefill_nnodes {}'.format(self.prefill_nnodes)) + dist_init_addr = f"{self.inf_dict['prefill_coordinator_addr']}:{self.inf_dict['prefill_coordinator_port']}" for i in range(0, int(self.prefill_nnodes)): cmd = f'''docker exec {self.container_name} /bin/bash -c "echo ' export NNODES={self.prefill_nnodes} @@ -514,6 +516,9 @@ def launch_prefill_servers(self, dtype='auto', kv_cache_dtype='auto'): --kv-cache-dtype {kv_cache_dtype} \ --trust-remote-code \ --tp {self.bp_dict['tensor_parallelism']} \ + --nnodes {self.prefill_nnodes} \ + --node-rank {i} \ + --dist-init-addr {dist_init_addr} \ --disable-radix-cache --disable-cuda-graph \ --mem-fraction-static {self.bp_dict['memory_fraction']} \ --attention-backend aiter \ @@ -568,25 +573,8 @@ def launch_decode_servers(self, dtype='auto', kv_cache_dtype='auto'): cmd_list = [] decode_node_list = self.inf_dict['decode_node_list'] log.info('%%%% self.decode_nnodes {}'.format(self.decode_nnodes)) + dist_init_addr = f"{self.inf_dict['decode_coordinator_addr']}:{self.inf_dict['decode_coordinator_port']}" for i in range(0, int(self.decode_nnodes)): - # ------------------------------------------------------------------ - # Construct a command that writes a Decode server launch script - # into /tmp/decode_launch_script.sh inside the container - # - # Key configuration details: - # - NNODES / NODE_RANK: Distributed topology for SGLang - # - disaggregation-mode decode: Run in Decode-only mode - # - disaggregation-ib-device: RDMA device used for KV transfers - # - host / port: Network endpoint for this Decode server - # - dtype / kv-cache-dtype: Compute and KV precision - # - tensor parallelism: Model sharding across GPUs - # - aiter backend: Optimized attention backend for AMD GPUs - # - memory fraction: Static GPU memory reservation - # - # NOTE: - # The script is written (echo > file), not executed here. - # Execution is handled by a separate orchestration step. - # ------------------------------------------------------------------ cmd = f'''docker exec {self.container_name} /bin/bash -c "echo ' export NNODES={self.decode_nnodes} export NODE_RANK={i} @@ -600,6 +588,9 @@ def launch_decode_servers(self, dtype='auto', kv_cache_dtype='auto'): --dtype {dtype} \ --kv-cache-dtype {kv_cache_dtype} \ --tp {self.bp_dict['tensor_parallelism']} \ + --nnodes {self.decode_nnodes} \ + --node-rank {i} \ + --dist-init-addr {dist_init_addr} \ --disable-radix-cache --disable-cuda-graph \ --mem-fraction-static {self.bp_dict['memory_fraction']} \ --attention-backend aiter \ @@ -647,10 +638,12 @@ def poll_and_check_server_ready( """ log.info('Waiting 120 secs after launching decode script') time.sleep(120) - for node_no in range(0, self.prefill_nnodes): - self.poll_for_server_ready(node_no, 'prefill') - for node_no in range(0, self.decode_nnodes): - self.poll_for_server_ready(node_no, 'decode') + # for node_no in range(0, self.prefill_nnodes): + # self.poll_for_server_ready(node_no, 'prefill') + # for node_no in range(0, self.decode_nnodes): + # self.poll_for_server_ready(node_no, 'decode') + self.poll_for_server_ready(0, 'prefill') + self.poll_for_server_ready(0, 'decode') def launch_proxy_router( self, @@ -682,18 +675,18 @@ def launch_proxy_router( # Each Prefill server is specified as: # --prefill http://: # ------------------------------------------------------------------ - prefill_str = '' - for prefill_node in self.prefill_node_list: - prefill_str = prefill_str + f"--prefill http://{prefill_node}:{self.inf_dict['prefill_serv_port']} " + + prefill_str = ( + f"--prefill http://{self.inf_dict['prefill_coordinator_addr']}:{self.inf_dict['prefill_serv_port']} " + ) # ------------------------------------------------------------------ # Build Decode endpoint arguments for the router # # Each Decode server is specified as: # --decode http://: # ------------------------------------------------------------------ - decode_str = '' - for decode_node in self.decode_node_list: - decode_str = decode_str + f"--decode http://{decode_node}:{self.inf_dict['decode_serv_port']} " + + decode_str = f"--decode http://{self.inf_dict['decode_coordinator_addr']}:{self.inf_dict['decode_serv_port']} " log.info('#================ * * * =========================#') log.info('Create Proxy Router launch script on Proxy Router nodes') log.info('#================ * * * =========================#') @@ -844,7 +837,7 @@ def benchserv_test_random(self, d_type='auto'): --random-output {i_dict['output_length']} \ --random-range-ratio {i_dict['random_range_ratio']} \ --host 0.0.0.0 --port {self.inf_dict['proxy_router_serv_port']} \ - > {self.log_dir}/benchmark_node/benchmark_results.log" ''' + > {self.log_dir}/benchmark_node/benchmark_results.log 2>&1" ''' formatted_cmd = textwrap_for_yml(cmd) self.b_phdl.exec(formatted_cmd, timeout=500) time.sleep(5) @@ -881,38 +874,38 @@ def poll_for_server_ready(self, node_no, sglang_function, no_of_iterations=16): # Prefill server readiness check # ------------------------------------------------------------------ if re.search('prefill', sglang_function): - head_node = self.prefill_node_list[0] for j in range(1, no_of_iterations): log.info(f'Starting poll iteration {j}') out_dict = self.p_phdl.exec( f'grep -B 20 -A 20 "200 OK" {self.log_dir}/prefill_node{node_no}/prefill_server.log' ) - if re.search('GET|POST', out_dict[head_node], re.I): + target_pnode = self.prefill_node_list[node_no] + if re.search('GET|POST', out_dict[target_pnode], re.I): log.info('Wait 60 secs to start serving traffic') time.sleep(60) - # if re.search('fired up and ready to roll', out_dict[head_node], re.I ): + # if re.search('fired up and ready to roll', out_dict[target_pnode], re.I ): # print('Prefill server {node_no} ready to serve') return else: log.info('Wait for 120 secs and continue polling') time.sleep(120) - head_node = self.prefill_node_list[0] + log.warning(f'Prefill node {node_no} did not get to ready to serve 200 OK state in {j} iterations') fail_test(f'Prefill node {node_no} did not get to ready to serve 200 OK state in {j} iterations') # ------------------------------------------------------------------ # Decode server readiness check # ------------------------------------------------------------------ elif re.search('decode', sglang_function): - head_node = self.decode_node_list[0] for j in range(1, no_of_iterations): log.info(f'Starting poll iteration {j}') out_dict = self.d_phdl.exec( f'grep -B 20 -A 20 "200 OK" {self.log_dir}/decode_node{node_no}/decode_server.log' ) - if re.search('GET|POST', out_dict[head_node]): + target_dnode = self.decode_node_list[node_no] + if re.search('GET|POST', out_dict[target_dnode]): log.info('Wait 60 secs to start serving traffic') time.sleep(60) - # if re.search('fired up and ready to roll', out_dict[head_node], re.I ): + # if re.search('fired up and ready to roll', out_dict[target_dnode], re.I ): # print('Decode server {node_no} ready to serve') return else: @@ -1034,7 +1027,7 @@ def scan_for_inference_errors( # Scan all prefill nodes for j in range(0, int(self.prefill_nnodes)): - cmd = f"sudo cat {self.log_dir}/prefill_node{j}/prefill_server.log" + cmd = f"sudo tail -500 {self.log_dir}/prefill_node{j}/prefill_server.log" cmd_list.append(cmd) out_dict = self.p_phdl.exec_cmd_list(cmd_list) @@ -1047,8 +1040,9 @@ def scan_for_inference_errors( inference_pass = False # Scan all decode nodes + cmd_list = [] for j in range(0, int(self.decode_nnodes)): - cmd = f"sudo cat {self.log_dir}/decode_node{j}/decode_server.log" + cmd = f"sudo tail -500 {self.log_dir}/decode_node{j}/decode_server.log" cmd_list.append(cmd) out_dict = self.d_phdl.exec_cmd_list(cmd_list) @@ -1259,3 +1253,66 @@ def verify_inference_results(self, test_name, expected_result_dict): verify_dmesg_for_errors(self.r_phdl, self.inference_start_time, self.inference_end_time) verify_dmesg_for_errors(self.b_phdl, self.inference_start_time, self.inference_end_time) log.info("%s", self.inference_results_dict) + + def sglang_disagg_gpu_counts(self, mem_threshold_mb=5000): + """ + After model load, count occupied GPUs per prefill/decode node via amd-smi. + """ + tp = int(self.bp_dict["tensor_parallelism"]) + + def _count_per_node(phdl): + per_node = {} + for node, payload in phdl.exec("sudo amd-smi metric --json").items(): + count = 0 + try: + entries = json.loads(payload.strip()) + except (json.JSONDecodeError, AttributeError): + log.warning("Failed to parse amd-smi JSON on node %s", node) + per_node[node] = 0 + continue + if isinstance(entries, dict) and "gpu_data" in entries: + entries = entries["gpu_data"] + if not isinstance(entries, list): + per_node[node] = 0 + continue + for g in entries: + used_mb = g.get("mem_usage", {}).get("used_vram", {}).get("value", 0) + if used_mb > mem_threshold_mb: + count += 1 + per_node[node] = count + return per_node + + prefill_per_node = _count_per_node(self.p_phdl) + decode_per_node = _count_per_node(self.d_phdl) + occupied_prefill = sum(prefill_per_node.values()) + occupied_decode = sum(decode_per_node.values()) + + result = { + "configured_tp": tp, + "prefill_per_node": prefill_per_node, + "decode_per_node": decode_per_node, + "prefill_occupied_gpus": occupied_prefill, + "decode_occupied_gpus": occupied_decode, + "total_occupied_gpus": occupied_prefill + occupied_decode, + } + + lines = [ + "", + f"Configured TP: {tp}", + "", + "Prefill:", + ] + for node, count in prefill_per_node.items(): + lines.append(f" {node}: {count} occupied GPUs") + lines.append(f" Total: {occupied_prefill} occupied GPUs") + lines.append("") + lines.append("Decode:") + for node, count in decode_per_node.items(): + lines.append(f" {node}: {count} occupied GPUs") + lines.append(f" Total: {occupied_decode} occupied GPUs") + lines.append("") + lines.append("Total hardware GPUs consumed:") + lines.append(f" {occupied_prefill + occupied_decode}") + + log.info("\n".join(lines)) + return result diff --git a/cvs/tests/inference/sglang/sglang_deepseek_r1_671b_distributed.py b/cvs/tests/inference/sglang/sglang_deepseek_r1_671b_distributed.py index c7134ea3..e8692c44 100644 --- a/cvs/tests/inference/sglang/sglang_deepseek_r1_671b_distributed.py +++ b/cvs/tests/inference/sglang/sglang_deepseek_r1_671b_distributed.py @@ -223,7 +223,7 @@ def test_cleanup_stale_containers(p_phdl, d_phdl, r_phdl, b_phdl, inference_dict def test_launch_inference_containers(p_phdl, d_phdl, r_phdl, b_phdl, inference_dict): - log.info('Testcase launch InferenceMax containers') + log.info('Testcase launch SGLang containers') globals.error_list = [] container_name = inference_dict['container_name'] # Launch the containers .. @@ -427,3 +427,13 @@ def test_run_benchmark_test(im_obj): im_obj.setup_benchmark_serv_container_env() im_obj.benchserv_test_random(d_type='auto') update_test_result() + + +# Test to validate the prefill/decode GPU layout +def test_disagg_gpu_topology(im_obj): + """ + Report occupied GPUs on prefill/decode nodes after model load. + """ + globals.error_list = [] + im_obj.sglang_disagg_gpu_counts() + update_test_result() diff --git a/cvs/tests/inference/sglang/sglang_llama_70b_distributed.py b/cvs/tests/inference/sglang/sglang_llama_70b_distributed.py index f70b7af7..5ed25ce1 100644 --- a/cvs/tests/inference/sglang/sglang_llama_70b_distributed.py +++ b/cvs/tests/inference/sglang/sglang_llama_70b_distributed.py @@ -223,7 +223,7 @@ def test_cleanup_stale_containers(p_phdl, d_phdl, r_phdl, b_phdl, inference_dict def test_launch_inference_containers(p_phdl, d_phdl, r_phdl, b_phdl, inference_dict): - log.info('Testcase launch InferenceMax containers') + log.info('Testcase launch SGLang containers') globals.error_list = [] container_name = inference_dict['container_name'] # Launch the containers .. @@ -427,3 +427,13 @@ def test_run_benchmark_test(im_obj): im_obj.setup_benchmark_serv_container_env() im_obj.benchserv_test_random(d_type='auto') update_test_result() + + +# Test to validate the prefill/decode GPU layout +def test_disagg_gpu_topology(im_obj): + """ + Report occupied GPUs on prefill/decode nodes after model load. + """ + globals.error_list = [] + im_obj.sglang_disagg_gpu_counts() + update_test_result()