[Fix] Fix uvloop get_event_loop() is not suitable for 0.22.x (#13612)

Signed-off-by: lzy <tomlzy213@gmail.com>
Co-authored-by: lzy <tomlzy213@gmail.com>
This commit is contained in:
Zhi Yiliu
2025-11-25 01:20:00 +08:00
committed by GitHub
co-authored by lzy
parent 98b38de3f2
commit a95a38078b
9 changed files with 32 additions and 20 deletions
+7 -7
View File
@@ -240,7 +240,7 @@ async def make_http_call(
api_name: Name of the API for error messages api_name: Name of the API for error messages
""" """
try: try:
start_time = asyncio.get_event_loop().time() start_time = asyncio.get_running_loop().time()
request_json = build_http_request_json(request_data) request_json = build_http_request_json(request_data)
headers = {"Content-Type": "application/json"} headers = {"Content-Type": "application/json"}
@@ -253,7 +253,7 @@ async def make_http_call(
f"[HTTP] {api_name} Request {request_id} failed with status " f"[HTTP] {api_name} Request {request_id} failed with status "
f"{resp.status}: {resp_text}" f"{resp.status}: {resp_text}"
) )
completion_time = asyncio.get_event_loop().time() completion_time = asyncio.get_running_loop().time()
await results_queue.put((request_id, 0, False, completion_time)) await results_queue.put((request_id, 0, False, completion_time))
return return
@@ -271,13 +271,13 @@ async def make_http_call(
) )
success = False success = False
completion_time = asyncio.get_event_loop().time() completion_time = asyncio.get_running_loop().time()
elapsed_time = (completion_time - start_time) * 1000 elapsed_time = (completion_time - start_time) * 1000
await results_queue.put((request_id, elapsed_time, success, completion_time)) await results_queue.put((request_id, elapsed_time, success, completion_time))
except Exception as e: except Exception as e:
print(f"[HTTP] {api_name} Error for request {request_id}: {e}") print(f"[HTTP] {api_name} Error for request {request_id}: {e}")
completion_time = asyncio.get_event_loop().time() completion_time = asyncio.get_running_loop().time()
await results_queue.put((request_id, 0, False, completion_time)) await results_queue.put((request_id, 0, False, completion_time))
@@ -738,7 +738,7 @@ async def run_generic_benchmark(
tasks = [] tasks = []
# Track timing for sending requests # Track timing for sending requests
send_start_time = asyncio.get_event_loop().time() send_start_time = asyncio.get_running_loop().time()
# HTTP implementation # HTTP implementation
async with aiohttp.ClientSession( async with aiohttp.ClientSession(
@@ -778,7 +778,7 @@ async def run_generic_benchmark(
if i < len(all_requests) - 1: if i < len(all_requests) - 1:
await sleep_with_distribution(config.distribution, rps) await sleep_with_distribution(config.distribution, rps)
send_end_time = asyncio.get_event_loop().time() send_end_time = asyncio.get_running_loop().time()
send_duration = send_end_time - send_start_time send_duration = send_end_time - send_start_time
# Wait for all requests to complete with progress tracking # Wait for all requests to complete with progress tracking
@@ -796,7 +796,7 @@ async def run_generic_benchmark(
if config.profile: if config.profile:
await send_profile_request("STOP_PROFILE", http_url, session=session) await send_profile_request("STOP_PROFILE", http_url, session=session)
completion_end_time = asyncio.get_event_loop().time() completion_end_time = asyncio.get_running_loop().time()
total_duration = completion_end_time - send_start_time total_duration = completion_end_time - send_start_time
return await process_results( return await process_results(
+1 -1
View File
@@ -69,7 +69,7 @@ dependencies = [
"tqdm", "tqdm",
"transformers==4.57.1", "transformers==4.57.1",
"uvicorn", "uvicorn",
"uvloop==0.21.0", "uvloop",
"xgrammar==0.1.27", "xgrammar==0.1.27",
"grpcio==1.75.1", # keep it align with compile_proto.py "grpcio==1.75.1", # keep it align with compile_proto.py
"grpcio-tools==1.75.1", # keep it align with compile_proto.py "grpcio-tools==1.75.1", # keep it align with compile_proto.py
+1 -2
View File
@@ -533,8 +533,7 @@ class Engine(EngineBase):
zmq_handles=zmq_handles, zmq_handles=zmq_handles,
flush_cache=flush_cache, flush_cache=flush_cache,
) )
loop = asyncio.get_event_loop() return self.loop.run_until_complete(
return loop.run_until_complete(
self.tokenizer_manager.update_weights_from_ipc(obj, None) self.tokenizer_manager.update_weights_from_ipc(obj, None)
) )
@@ -28,7 +28,7 @@ from sglang.srt.managers.io_struct import (
TokenizedGenerateReqInput, TokenizedGenerateReqInput,
) )
from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.server_args import PortArgs, ServerArgs
from sglang.srt.utils import get_zmq_socket, kill_process_tree from sglang.srt.utils import get_or_create_event_loop, get_zmq_socket, kill_process_tree
from sglang.utils import get_exception_traceback from sglang.utils import get_exception_traceback
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -876,7 +876,7 @@ class GrpcRequestManager:
return return
self.no_create_loop = True self.no_create_loop = True
loop = asyncio.get_event_loop() loop = get_or_create_event_loop()
self.asyncio_tasks.add( self.asyncio_tasks.add(
loop.create_task(print_exception_wrapper(self.handle_loop)) loop.create_task(print_exception_wrapper(self.handle_loop))
) )
@@ -98,6 +98,7 @@ from sglang.srt.utils import (
dataclass_to_string_truncated, dataclass_to_string_truncated,
freeze_gc, freeze_gc,
get_bool_env_var, get_bool_env_var,
get_or_create_event_loop,
get_zmq_socket, get_zmq_socket,
kill_process_tree, kill_process_tree,
) )
@@ -1354,12 +1355,13 @@ class TokenizerManager(TokenizerCommunicatorMixin):
def auto_create_handle_loop(self): def auto_create_handle_loop(self):
if self._chosen_loop is not None: if self._chosen_loop is not None:
current_loop = get_or_create_event_loop()
assert ( assert (
asyncio.get_event_loop() == self._chosen_loop current_loop == self._chosen_loop
), f"Please ensure only one event loop is ever used with SGLang. Previous loop: {self._chosen_loop}, current loop: {asyncio.get_event_loop()}" ), f"Please ensure only one event loop is ever used with SGLang. Previous loop: {self._chosen_loop}, current loop: {current_loop}"
return return
loop = asyncio.get_event_loop() loop = get_or_create_event_loop()
self._chosen_loop = loop self._chosen_loop = loop
self.asyncio_tasks.add( self.asyncio_tasks.add(
loop.create_task(print_exception_wrapper(self.handle_loop)) loop.create_task(print_exception_wrapper(self.handle_loop))
@@ -89,7 +89,7 @@ class LlavaImageProcessor(BaseMultimodalProcessor):
grid_pinpoints: str, grid_pinpoints: str,
): ):
if self.cpu_executor is not None: if self.cpu_executor is not None:
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
return await loop.run_in_executor( return await loop.run_in_executor(
self.cpu_executor, self.cpu_executor,
LlavaImageProcessor._process_single_image_task, LlavaImageProcessor._process_single_image_task,
+10
View File
@@ -104,6 +104,16 @@ show_time_cost = False
time_infos = {} time_infos = {}
def get_or_create_event_loop():
"""Gets the running event loop or creates a new one if it doesn't exist."""
try:
return asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop
HIP_FP8_E4M3_FNUZ_MAX = 224.0 HIP_FP8_E4M3_FNUZ_MAX = 224.0
+2 -1
View File
@@ -8,6 +8,7 @@ from typing import Optional
import numpy as np import numpy as np
import sglang as sgl import sglang as sgl
from sglang.srt.utils import get_or_create_event_loop
from sglang.utils import download_and_cache_file, read_jsonl from sglang.utils import download_and_cache_file, read_jsonl
INVALID = -9999999 INVALID = -9999999
@@ -89,7 +90,7 @@ def run_eval(args):
# Run requests # Run requests
tic = time.perf_counter() tic = time.perf_counter()
loop = asyncio.get_event_loop() loop = get_or_create_event_loop()
outputs = loop.run_until_complete( outputs = loop.run_until_complete(
concurrent_generate(engine, prompts, sampling_param) concurrent_generate(engine, prompts, sampling_param)
@@ -6,7 +6,6 @@ or
python -m unittest discover -s tests -p "test_*unit.py" -v python -m unittest discover -s tests -p "test_*unit.py" -v
""" """
import asyncio
import json import json
import unittest import unittest
import uuid import uuid
@@ -21,6 +20,7 @@ from sglang.srt.entrypoints.openai.protocol import (
) )
from sglang.srt.entrypoints.openai.serving_chat import OpenAIServingChat from sglang.srt.entrypoints.openai.serving_chat import OpenAIServingChat
from sglang.srt.managers.io_struct import GenerateReqInput from sglang.srt.managers.io_struct import GenerateReqInput
from sglang.srt.utils import get_or_create_event_loop
class _MockTokenizerManager: class _MockTokenizerManager:
@@ -389,7 +389,7 @@ class ServingChatTestCase(unittest.TestCase):
break break
return line return line
loop = asyncio.get_event_loop() loop = get_or_create_event_loop()
line = loop.run_until_complete(collect_first_tool_chunk()) line = loop.run_until_complete(collect_first_tool_chunk())
self.assertIsNotNone(line) self.assertIsNotNone(line)
self.assertTrue(line.startswith("data: ")) self.assertTrue(line.startswith("data: "))
@@ -564,7 +564,7 @@ class ServingChatTestCase(unittest.TestCase):
break break
return line return line
loop = asyncio.get_event_loop() loop = get_or_create_event_loop()
line = loop.run_until_complete(collect_first_tool_chunk()) line = loop.run_until_complete(collect_first_tool_chunk())
self.assertIsNotNone(line) self.assertIsNotNone(line)
self.assertTrue(line.startswith("data: ")) self.assertTrue(line.startswith("data: "))