customized_stress_concurrency:int=128

customized_stress_request_rate:int=20

enable_accuracy_test:bool=False# Enable accuracy testing with GSM8K

accuracy_test_timeout:int=1200# 20 minutes timeout for accuracy tests

accuracy_test_concurrency:int=512# Concurrency for accuracy tests

accuracy_test_max_retries:int=3# Max retries for accuracy tests

accuracy_test_max_gen_toks:int=256# Max generation tokens for accuracy tests

accuracy_test_max_length:int=4096# Max input length for accuracy tests

defrequest_count_stress_test(self)->int:

returnFalse,f"Unexpected error during health check:{str(e)}"

eliftest_mode=="stress-stage-alone":

eliftest_mode=="stress-test-with-accuracy":

else:

pytest.skip(f"Skipping test for unsupported mode:{test_mode}. "

f"Supported modes: stress-test, stress-stage-alone")

pytest.skip(

)

pp_size=test_server_config.pp_size,

ep_size=8,# DeepSeek-V3 or DeepSeek-R1 specific ep_size

161,# DeepSeek-V3 or DeepSeek-R1 specific max_batch_size

2048,# DeepSeek-V3 or DeepSeek-R1 specific max_batch_size

1160,# DeepSeek-V3 or DeepSeek-R1 specific max_num_tokens

2048,# DeepSeek-V3 or DeepSeek-R1 specific max_num_tokens

0.7,# DeepSeek-V3 or DeepSeek-R1 specific kv_cache fraction

capacity_scheduler_policy=test_server_config.

ifrun_stress:

enable_accuracy= (test_mode=="stress-test-with-accuracy")

stress_config=StressTestConfig(model_config=config,

server_config=test_server_config)

server_config=test_server_config,

enable_accuracy_test=enable_accuracy)

ifstress_timeisnotNone:

server_config=test_server_config,

stress_time=stress_time,

ifstress_timeoutisnotNoneelsestress_time*2)

ifstress_timeoutisnotNoneelsestress_time*2,

enable_accuracy_test=enable_accuracy)

else:

print_info(

f"Server is running with model{model_name}. Starting tests...")

ifstress_configandstress_config.enable_accuracy_test:

baseline_accuracy_success,baseline_accuracy_value=run_accuracy_test(

model_path,test_server_config,stress_config,"baseline")

ifrun_performance:

stress_config,

None,

request_counter=request_counter)

ifstress_configandstress_config.enable_accuracy_test:

post_stress_accuracy_success,post_stress_accuracy_value=run_accuracy_test(

model_path,test_server_config,stress_config,

"post_stress")

ifbaseline_accuracy_successandpost_stress_accuracy_success:

print_info("=== ACCURACY TEST SUMMARY ===")

print_info("✓ Baseline accuracy test: PASSED")

print_info("✓ Post-stress accuracy test: PASSED")

ifbaseline_accuracy_valueisnotNoneandpost_stress_accuracy_valueisnotNone:

accuracy_drop_percentage= (

accuracy_drop/baseline_accuracy_value)*100

print_info(

f"Baseline accuracy:{baseline_accuracy_value:.4f}")

print_info(

)

print_info(

)

assertaccuracy_drop_percentage<= (

),f"Accuracy drop{accuracy_drop_percentage:.2f}% exceeds threshold{accuracy_drop_threshold*100}%"

print_info(

)

else:

print_warning("=== ACCURACY TEST SUMMARY ===")

ifnotbaseline_accuracy_success:

print_warning("✗ Baseline accuracy test: FAILED")

ifnotpost_stress_accuracy_success:

print_warning("✗ Post-stress accuracy test: FAILED")

print_warning(

"Model accuracy may be affected by stress conditions")

finally:

ifos.path.exists(extra_llm_options_path):

defparse_accuracy_from_lm_eval_output(output_text:str)->float:

patterns= [

r'flexible-extract\|\s+\d+\|exact_match\|\↑\s+\|(\d+\.\d+)',

]

forpatterninpatterns:

match=re.search(pattern,output_text)

ifmatch:

accuracy_value=float(match.group(1))

print_info(f"Extracted accuracy value:{accuracy_value}")

print_warning("Could not find accuracy value in lm_eval output")

print_warning(f"Output text:{output_text}")

defrun_accuracy_test(model_path:str,

server_config:ServerConfig,

stress_config:StressTestConfig,

test_phase:str="baseline")->tuple[bool,float]:

ifnotstress_config.enable_accuracy_test:

print_info(f"Skipping accuracy test for{test_phase} phase (disabled)")

returnTrue,None

print_info(f"=== Running{test_phase.upper()} ACCURACY TEST (GSM8K) ===")

lm_eval_cmd= [

"lm_eval","--model","local-completions","--tasks","gsm8k",

"--model_args",

f"max_length={stress_config.accuracy_test_max_length}",

]

test_start_time=time.time()

try:

print_info(f"Running lm_eval command:{' '.join(lm_eval_cmd)}")

result=subprocess.run(lm_eval_cmd,

capture_output=True,

text=True,

timeout=stress_config.accuracy_test_timeout)

ifresult.returncode==0:

test_end_time=time.time()

duration=int(test_end_time-test_start_time)

print_info(

)

output_text=result.stdout

accuracy_value=parse_accuracy_from_lm_eval_output(output_text)

returnTrue,accuracy_value

else:

print_warning(

f"lm_eval exited with non-zero code:{result.returncode}")

print_warning(f"stderr:{result.stderr}")

returnFalse,None

exceptsubprocess.TimeoutExpired:

print_warning(

)

returnFalse,None

exceptExceptionase:

print_warning(f"Error during{test_phase} accuracy test:{str(e)}")

returnFalse,None

defextract_stress_test_metrics(artifacts_dir="./artifacts",

current_model=None):

stress_test/stress_test.py::test_run_stress_test[DeepSeek-V3_tp8-stress_time_3600s_timeout_5400s-GUARANTEED_NO_EVICT-pytorch-stress-test-with-accuracy]

stress_test/stress_test.py::test_run_stress_test[DeepSeek-V3_tp8-stress_time_3600s_timeout_5400s-MAX_UTILIZATION-pytorch-stress-test-with-accuracy]

stress_test/stress_test.py::test_run_stress_test[DeepSeek-R1_tp8-stress_time_3600s_timeout_5400s-MAX_UTILIZATION-pytorch-stress-test-with-accuracy]