EngineX-Hygon/sglang
5
0
Fork 0
Code Issues Pull Requests Actions 7 Projects Releases Wiki Activity

Use monotonic clock for interval measurement (#6211)

Browse Source 
Signed-off-by: Lifu Huang <lifu.hlf@gmail.com>
This commit is contained in:
Lifu Huang
2025-05-17 16:49:18 -07:00
committed by GitHub
parent 2716830802
commit 3cf1473a09
21 changed files with 72 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
python/sglang/bench_one_batch.py
View File

@@ -373,10 +373,10 @@ def latency_test_run_once(
# Prefill
synchronize(device)
tic = time.time()
tic = time.perf_counter()
next_token_ids, _, batch = extend(reqs, model_runner)
synchronize(device)
prefill_latency = time.time() - tic
prefill_latency = time.perf_counter() - tic
tot_latency += prefill_latency
throughput = input_len * batch_size / prefill_latency
rank_print(
@@ -389,10 +389,10 @@ def latency_test_run_once(
decode_latencies = []
for i in range(output_len - 1):
synchronize(device)
tic = time.time()
tic = time.perf_counter()
next_token_ids, _ = decode(next_token_ids, batch, model_runner)
synchronize(device)
latency = time.time() - tic
latency = time.perf_counter() - tic
tot_latency += latency
throughput = batch_size / latency
decode_latencies.append(latency)

10
python/sglang/bench_one_batch_server.py
View File

@@ -92,8 +92,8 @@ def launch_server_process(server_args: ServerArgs):
base_url = f"http://{server_args.host}:{server_args.port}"
timeout = 600
start_time = time.time()
while time.time() - start_time < timeout:
start_time = time.perf_counter()
while time.perf_counter() - start_time < timeout:
try:
headers = {
"Content-Type": "application/json; charset=utf-8",
@@ -141,7 +141,7 @@ def run_one_case(
else:
json_schema = None
tic = time.time()
tic = time.perf_counter()
response = requests.post(
url + "/generate",
json={
@@ -175,9 +175,9 @@ def run_one_case(
or data["meta_info"]["finish_reason"]["type"] == "length"
)
if data["meta_info"]["completion_tokens"] == 1:
ttft = time.time() - tic
ttft = time.perf_counter() - tic
latency = time.time() - tic
latency = time.perf_counter() - tic
input_throughput = batch_size * input_len / ttft
output_throughput = batch_size * output_len / (latency - ttft)
overall_throughput = batch_size * (input_len + output_len) / latency

8
python/sglang/compile_deep_gemm.py
View File

@@ -82,8 +82,8 @@ def launch_server_process_and_send_one_request(
base_url = f"http://{server_args.host}:{server_args.port}"
timeout = compile_args.timeout
start_time = time.time()
while time.time() - start_time < timeout:
start_time = time.perf_counter()
while time.perf_counter() - start_time < timeout:
try:
headers = {
"Content-Type": "application/json; charset=utf-8",
@@ -112,9 +112,9 @@ def launch_server_process_and_send_one_request(
raise RuntimeError(f"Sync request failed: {error}")
# Other nodes should wait for the exit signal from Rank-0 node.
else:
start_time_waiting = time.time()
start_time_waiting = time.perf_counter()
while proc.is_alive():
if time.time() - start_time_waiting < timeout:
if time.perf_counter() - start_time_waiting < timeout:
time.sleep(10)
else:
raise TimeoutError("Waiting for main node timeout!")

6
python/sglang/srt/distributed/utils.py
View File

@@ -127,14 +127,14 @@ class StatelessProcessGroup:
key = f"send_to/{dst}/{self.send_dst_counter[dst]}"
self.store.set(key, pickle.dumps(obj))
self.send_dst_counter[dst] += 1
self.entries.append((key, time.time()))
self.entries.append((key, time.perf_counter()))
def expire_data(self):
"""Expire data that is older than `data_expiration_seconds` seconds."""
while self.entries:
# check the oldest entry
key, timestamp = self.entries[0]
if time.time() - timestamp > self.data_expiration_seconds:
if time.perf_counter() - timestamp > self.data_expiration_seconds:
self.store.delete_key(key)
self.entries.popleft()
else:
@@ -158,7 +158,7 @@ class StatelessProcessGroup:
key = f"broadcast_from/{src}/" f"{self.broadcast_send_counter}"
self.store.set(key, pickle.dumps(obj))
self.broadcast_send_counter += 1
self.entries.append((key, time.time()))
self.entries.append((key, time.perf_counter()))
return obj
else:
key = f"broadcast_from/{src}/" f"{self.broadcast_recv_src_counter[src]}"

4
python/sglang/srt/entrypoints/http_server.py
View File

@@ -182,9 +182,9 @@ async def health_generate(request: Request) -> Response:
async for _ in _global_state.tokenizer_manager.generate_request(gri, request):
break
tic = time.time()
tic = time.perf_counter()
task = asyncio.create_task(gen())
while time.time() < tic + HEALTH_CHECK_TIMEOUT:
while time.perf_counter() < tic + HEALTH_CHECK_TIMEOUT:
await asyncio.sleep(1)
if _global_state.tokenizer_manager.last_receive_tstamp > tic:
task.cancel()

4
python/sglang/srt/entrypoints/http_server_engine.py
View File

@@ -24,10 +24,10 @@ def launch_server_process(server_args: ServerArgs) -> multiprocessing.Process:
base_url = server_args.url()
timeout = 300.0 # Increased timeout to 5 minutes for downloading large models
start_time = time.time()
start_time = time.perf_counter()
with requests.Session() as session:
while time.time() - start_time < timeout:
while time.perf_counter() - start_time < timeout:
try:
headers = {
"Content-Type": "application/json; charset=utf-8",

26
python/sglang/srt/managers/scheduler.py
View File

@@ -348,8 +348,8 @@ class Scheduler(
self.forward_ct_decode = 0
self.num_generated_tokens = 0
self.num_prefill_tokens = 0
self.last_decode_stats_tic = time.time()
self.last_prefill_stats_tic = time.time()
self.last_decode_stats_tic = time.perf_counter()
self.last_prefill_stats_tic = time.perf_counter()
self.return_health_check_ct = 0
self.current_stream = torch.get_device_module(self.device).current_stream()
if self.device == "cpu":
@@ -1032,13 +1032,13 @@ class Scheduler(
add_to_grammar_queue = True
if add_to_grammar_queue:
req.queue_time_start = time.time()
req.queue_time_start = time.perf_counter()
self.grammar_queue.append(req)
else:
self._add_request_to_queue(req)
def _add_request_to_queue(self, req: Req):
req.queue_time_start = time.time()
req.queue_time_start = time.perf_counter()
if self.disaggregation_mode == DisaggregationMode.PREFILL:
self.disagg_prefill_bootstrap_queue.add(req)
elif self.disaggregation_mode == DisaggregationMode.DECODE:
@@ -1085,7 +1085,7 @@ class Scheduler(
req.finished_reason = FINISH_ABORT(
error_msg, HTTPStatus.BAD_REQUEST, "BadRequestError"
)
req.queue_time_start = time.time()
req.queue_time_start = time.perf_counter()
self.waiting_queue.append(req)
return
@@ -1109,8 +1109,8 @@ class Scheduler(
can_run_list: List[Req],
running_bs: int,
):
gap_latency = time.time() - self.last_prefill_stats_tic
self.last_prefill_stats_tic = time.time()
gap_latency = time.perf_counter() - self.last_prefill_stats_tic
self.last_prefill_stats_tic = time.perf_counter()
self.last_input_throughput = self.num_prefill_tokens / gap_latency
self.num_prefill_tokens = 0
@@ -1160,8 +1160,8 @@ class Scheduler(
):
batch = running_batch or self.running_batch
gap_latency = time.time() - self.last_decode_stats_tic
self.last_decode_stats_tic = time.time()
gap_latency = time.perf_counter() - self.last_decode_stats_tic
self.last_decode_stats_tic = time.perf_counter()
self.last_gen_throughput = self.num_generated_tokens / gap_latency
self.num_generated_tokens = 0
num_running_reqs = len(batch.reqs)
@@ -1245,7 +1245,7 @@ class Scheduler(
if (
self.enable_metrics
and self.attn_tp_rank == 0
and time.time() > self.metrics_collector.last_log_time + 30
and time.perf_counter() > self.metrics_collector.last_log_time + 30
):
# During idle time, also collect metrics every 30 seconds.
num_used = self.max_total_num_tokens - (
@@ -1410,7 +1410,7 @@ class Scheduler(
if self.enable_metrics:
# only record queue time when enable_metrics is True to avoid overhead
for req in can_run_list:
req.queue_time_end = time.time()
req.queue_time_end = time.perf_counter()
self.waiting_queue = [
x for x in self.waiting_queue if x not in set(can_run_list)
@@ -1783,10 +1783,10 @@ class Scheduler(
def watchdog_thread(self):
"""A watch dog thread that will try to kill the server itself if one forward batch takes too long."""
self.watchdog_last_forward_ct = 0
self.watchdog_last_time = time.time()
self.watchdog_last_time = time.perf_counter()
while True:
current = time.time()
current = time.perf_counter()
if self.cur_batch is not None:
if self.watchdog_last_forward_ct == self.forward_ct:
if current > self.watchdog_last_time + self.watchdog_timeout:

8
python/sglang/srt/mem_cache/hiradix_cache.py
View File

@@ -335,13 +335,13 @@ class HiRadixCache(RadixCache):
return value, last_node
def _match_prefix_helper(self, node: TreeNode, key: List):
node.last_access_time = time.time()
node.last_access_time = time.monotonic()
child_key = self.get_child_key_fn(key)
value = []
while len(key) > 0 and child_key in node.children.keys():
child = node.children[child_key]
child.last_access_time = time.time()
child.last_access_time = time.monotonic()
prefix_len = self.key_match_fn(child.key, key)
if prefix_len < len(child.key):
new_node = self._split_node(child.key, child, prefix_len)
@@ -386,7 +386,7 @@ class HiRadixCache(RadixCache):
return new_node
def _insert_helper(self, node: TreeNode, key: List, value):
node.last_access_time = time.time()
node.last_access_time = time.monotonic()
if len(key) == 0:
return 0
@@ -395,7 +395,7 @@ class HiRadixCache(RadixCache):
while len(key) > 0 and child_key in node.children.keys():
node = node.children[child_key]
node.last_access_time = time.time()
node.last_access_time = time.monotonic()
prefix_len = self.key_match_fn(node.key, key)
if prefix_len == len(node.key):

10
python/sglang/srt/mem_cache/radix_cache.py
View File

@@ -45,7 +45,7 @@ class TreeNode:
self.key = None
self.value = None
self.lock_ref = 0
self.last_access_time = time.time()
self.last_access_time = time.monotonic()
self.hit_count = 0
# indicating the node is loading KV cache from host
@@ -322,14 +322,14 @@ class RadixCache(BasePrefixCache):
##### Internal Helper Functions #####
def _match_prefix_helper(self, node: TreeNode, key: List):
node.last_access_time = time.time()
node.last_access_time = time.monotonic()
child_key = self.get_child_key_fn(key)
value = []
while len(key) > 0 and child_key in node.children.keys():
child = node.children[child_key]
child.last_access_time = time.time()
child.last_access_time = time.monotonic()
prefix_len = self.key_match_fn(child.key, key)
if prefix_len < len(child.key):
new_node = self._split_node(child.key, child, prefix_len)
@@ -361,7 +361,7 @@ class RadixCache(BasePrefixCache):
return new_node
def _insert_helper(self, node: TreeNode, key: List, value):
node.last_access_time = time.time()
node.last_access_time = time.monotonic()
if len(key) == 0:
return 0
@@ -370,7 +370,7 @@ class RadixCache(BasePrefixCache):
total_prefix_length = 0
while len(key) > 0 and child_key in node.children.keys():
node = node.children[child_key]
node.last_access_time = time.time()
node.last_access_time = time.monotonic()
prefix_len = self.key_match_fn(node.key, key)
total_prefix_length += prefix_len
key = key[prefix_len:]

4
python/sglang/srt/metrics/collector.py
View File

@@ -154,7 +154,7 @@ class SchedulerMetricsCollector:
from prometheus_client import Counter, Gauge
self.labels = labels
self.last_log_time = time.time()
self.last_log_time = time.perf_counter()
self.num_running_reqs = Gauge(
name="sglang:num_running_reqs",
@@ -294,7 +294,7 @@ class SchedulerMetricsCollector:
self.num_decode_transfer_queue_reqs, stats.num_decode_transfer_queue_reqs
)
self.last_log_time = time.time()
self.last_log_time = time.perf_counter()
class TokenizerMetricsCollector:

4
python/sglang/srt/model_executor/model_runner.py
View File

@@ -1019,7 +1019,7 @@ class ModelRunner:
if self.server_args.disable_cuda_graph:
return
tic = time.time()
tic = time.perf_counter()
before_mem = get_available_gpu_memory(self.device, self.gpu_id)
logger.info(
f"Capture cuda graph begin. This can take up to several minutes. avail mem={before_mem:.2f} GB"
@@ -1027,7 +1027,7 @@ class ModelRunner:
self.cuda_graph_runner = CudaGraphRunner(self)
after_mem = get_available_gpu_memory(self.device, self.gpu_id)
logger.info(
f"Capture cuda graph end. Time elapsed: {time.time() - tic:.2f} s. "
f"Capture cuda graph end. Time elapsed: {time.perf_counter() - tic:.2f} s. "
f"mem usage={(before_mem - after_mem):.2f} GB. avail mem={after_mem:.2f} GB."
)

4
python/sglang/srt/speculative/eagle_worker.py
View File

@@ -228,7 +228,7 @@ class EAGLEWorker(TpModelWorker):
return
# Capture draft
tic = time.time()
tic = time.perf_counter()
before_mem = get_available_gpu_memory(self.device, self.gpu_id)
logger.info(
f"Capture draft cuda graph begin. This can take up to several minutes. avail mem={before_mem:.2f} GB"
@@ -236,7 +236,7 @@ class EAGLEWorker(TpModelWorker):
self.cuda_graph_runner = EAGLEDraftCudaGraphRunner(self)
after_mem = get_available_gpu_memory(self.device, self.gpu_id)
logger.info(
f"Capture draft cuda graph end. Time elapsed: {time.time() - tic:.2f} s. avail mem={after_mem:.2f} GB. mem usage={(before_mem - after_mem):.2f} GB."
f"Capture draft cuda graph end. Time elapsed: {time.perf_counter() - tic:.2f} s. avail mem={after_mem:.2f} GB. mem usage={(before_mem - after_mem):.2f} GB."
)
# Capture extend

8
python/sglang/srt/utils.py
View File

@@ -246,7 +246,7 @@ def mark_start(name, interval=0.1, color=0, indent=0):
torch.cuda.synchronize()
if time_infos.get(name, None) is None:
time_infos[name] = TimeInfo(name, interval, color, indent)
time_infos[name].acc_time -= time.time()
time_infos[name].acc_time -= time.perf_counter()
def mark_end(name):
@@ -254,7 +254,7 @@ def mark_end(name):
if not show_time_cost:
return
torch.cuda.synchronize()
time_infos[name].acc_time += time.time()
time_infos[name].acc_time += time.perf_counter()
if time_infos[name].check():
time_infos[name].pretty_print()
@@ -264,11 +264,11 @@ def calculate_time(show=False, min_cost_ms=0.0):
def inner_func(*args, **kwargs):
torch.cuda.synchronize()
if show:
start_time = time.time()
start_time = time.perf_counter()
result = func(*args, **kwargs)
torch.cuda.synchronize()
if show:
cost_time = (time.time() - start_time) * 1000
cost_time = (time.perf_counter() - start_time) * 1000
if cost_time > min_cost_ms:
print(f"Function {func.__name__} took {cost_time} ms to run.")
return result

4
python/sglang/test/test_utils.py
View File

@@ -526,9 +526,9 @@ def popen_launch_pd_server(
else:
process = subprocess.Popen(command, stdout=None, stderr=None, env=env)
start_time = time.time()
start_time = time.perf_counter()
with requests.Session() as session:
while time.time() - start_time < timeout:
while time.perf_counter() - start_time < timeout:
try:
headers = {
"Content-Type": "application/json; charset=utf-8",

4
python/sglang/utils.py
View File

@@ -436,7 +436,7 @@ def wait_for_server(base_url: str, timeout: int = None) -> None:
base_url: The base URL of the server
timeout: Maximum time to wait in seconds. None means wait forever.
"""
start_time = time.time()
start_time = time.perf_counter()
while True:
try:
response = requests.get(
@@ -455,7 +455,7 @@ def wait_for_server(base_url: str, timeout: int = None) -> None:
)
break
if timeout and time.time() - start_time > timeout:
if timeout and time.perf_counter() - start_time > timeout:
raise TimeoutError("Server did not become ready within timeout period")
except requests.exceptions.RequestException:
time.sleep(1)