http_server.py 27.8 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# Copyright 2023-2024 SGLang Team
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""
The entry point of inference server. (SRT = SGLang Runtime)

Wang Ran (汪然)'s avatar
Wang Ran (汪然) committed
17
This file implements HTTP APIs for the inference engine via fastapi.
18
19
20
21
"""

import asyncio
import dataclasses
22
import json
23
24
25
26
27
28
import logging
import multiprocessing as multiprocessing
import os
import threading
import time
from http import HTTPStatus
Lianmin Zheng's avatar
Lianmin Zheng committed
29
from typing import AsyncIterator, Callable, Dict, Optional
30
31
32
33

# Fix a bug of Python threading
setattr(threading, "_register_atexit", lambda *args, **kwargs: None)

34
35
36
from contextlib import asynccontextmanager

import numpy as np
37
38
39
40
41
42
43
44
45
import orjson
import requests
import uvicorn
import uvloop
from fastapi import FastAPI, File, Form, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import ORJSONResponse, Response, StreamingResponse

from sglang.srt.entrypoints.engine import _launch_subprocesses
YAMY's avatar
YAMY committed
46
from sglang.srt.function_call_parser import FunctionCallParser
47
48
49
50
51
52
53
54
from sglang.srt.managers.io_struct import (
    CloseSessionReqInput,
    ConfigureLoggingReq,
    EmbeddingReqInput,
    GenerateReqInput,
    GetWeightsByNameReqInput,
    InitWeightsUpdateGroupReqInput,
    OpenSessionReqInput,
55
    ParseFunctionCallReq,
56
    ProfileReqInput,
57
58
    ReleaseMemoryOccupationReqInput,
    ResumeMemoryOccupationReqInput,
Xihuai Wang's avatar
Xihuai Wang committed
59
    SeparateReasoningReqInput,
60
    SetInternalStateReq,
61
62
    UpdateWeightFromDiskReqInput,
    UpdateWeightsFromDistributedReqInput,
63
    UpdateWeightsFromTensorReqInput,
64
    VertexGenerateReqInput,
65
)
66
from sglang.srt.managers.tokenizer_manager import TokenizerManager
67
68
69
70
71
72
73
74
75
76
77
78
79
80
from sglang.srt.metrics.func_timer import enable_func_timer
from sglang.srt.openai_api.adapter import (
    v1_batches,
    v1_cancel_batch,
    v1_chat_completions,
    v1_completions,
    v1_delete_file,
    v1_embeddings,
    v1_files_create,
    v1_retrieve_batch,
    v1_retrieve_file,
    v1_retrieve_file_content,
)
from sglang.srt.openai_api.protocol import ModelCard, ModelList
Xihuai Wang's avatar
Xihuai Wang committed
81
from sglang.srt.reasoning_parser import ReasoningParser
82
83
84
85
86
from sglang.srt.server_args import ServerArgs
from sglang.srt.utils import (
    add_api_key_middleware,
    add_prometheus_middleware,
    delete_directory,
87
    get_bool_env_var,
88
89
90
    kill_process_tree,
    set_uvicorn_logging_configs,
)
91
from sglang.srt.warmup import execute_warmups
92
93
94
95
96
97
98
99
100
101
from sglang.utils import get_exception_traceback
from sglang.version import __version__

logger = logging.getLogger(__name__)
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())


# Store global states
@dataclasses.dataclass
class _GlobalState:
102
    tokenizer_manager: TokenizerManager
103
104
105
106
107
108
109
110
111
112
113
    scheduler_info: Dict


_global_state: Optional[_GlobalState] = None


def set_global_state(global_state: _GlobalState):
    global _global_state
    _global_state = global_state


114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@asynccontextmanager
async def lifespan(fast_api_app: FastAPI):
    server_args: ServerArgs = fast_api_app.server_args
    if server_args.warmups is not None:
        await execute_warmups(
            server_args.warmups.split(","), _global_state.tokenizer_manager
        )
        logger.info("Warmup ended")

    warmup_thread = getattr(fast_api_app, "warmup_thread", None)
    if warmup_thread is not None:
        warmup_thread.start()
    yield


# Fast API
130
131
132
133
app = FastAPI(
    lifespan=lifespan,
    openapi_url=None if get_bool_env_var("DISABLE_OPENAPI_DOC") else "/openapi.json",
)
134
135
136
137
138
139
140
141
142
143
144
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

HEALTH_CHECK_TIMEOUT = int(os.getenv("SGLANG_HEALTH_CHECK_TIMEOUT", 20))


145
146
147
148
149
150
151
152
153
154
155
156
157
##### Native API endpoints #####


@app.get("/health")
async def health() -> Response:
    """Check the health of the http server."""
    return Response(status_code=200)


@app.get("/health_generate")
async def health_generate(request: Request) -> Response:
    """Check the health of the inference server by generating one token."""

158
159
    sampling_params = {"max_new_tokens": 1, "temperature": 0.0}
    rid = f"HEALTH_CHECK_{time.time()}"
160

161
162
163
    if _global_state.tokenizer_manager.is_image_gen:
        raise NotImplementedError()
    elif _global_state.tokenizer_manager.is_generation:
164
        gri = GenerateReqInput(
165
166
167
168
            rid=rid,
            input_ids=[0],
            sampling_params=sampling_params,
            log_metrics=False,
169
170
171
        )
    else:
        gri = EmbeddingReqInput(
172
            rid=rid, input_ids=[0], sampling_params=sampling_params, log_metrics=False
173
174
        )

175
    async def gen():
176
        async for _ in _global_state.tokenizer_manager.generate_request(gri, request):
177
            break
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

    tic = time.time()
    task = asyncio.create_task(gen())
    while time.time() < tic + HEALTH_CHECK_TIMEOUT:
        await asyncio.sleep(1)
        if _global_state.tokenizer_manager.last_receive_tstamp > tic:
            task.cancel()
            _global_state.tokenizer_manager.rid_to_state.pop(rid, None)
            return Response(status_code=200)

    task.cancel()
    tic_time = time.strftime("%H:%M:%S", time.localtime(tic))
    last_receive_time = time.strftime(
        "%H:%M:%S", time.localtime(_global_state.tokenizer_manager.last_receive_tstamp)
    )
    logger.error(
        f"Health check failed. Server couldn't get a response from detokenizer for last "
        f"{HEALTH_CHECK_TIMEOUT} seconds. tic start time: {tic_time}. "
        f"last_heartbeat time: {last_receive_time}"
    )
    _global_state.tokenizer_manager.rid_to_state.pop(rid, None)
    return Response(status_code=503)
200
201
202
203
204
205


@app.get("/get_model_info")
async def get_model_info():
    """Get the model information."""
    result = {
206
207
208
        "model_path": _global_state.tokenizer_manager.model_path,
        "tokenizer_path": _global_state.tokenizer_manager.server_args.tokenizer_path,
        "is_generation": _global_state.tokenizer_manager.is_generation,
209
210
211
212
213
214
    }
    return result


@app.get("/get_server_info")
async def get_server_info():
215
    internal_states = await _global_state.tokenizer_manager.get_internal_state()
216
    return {
217
        **dataclasses.asdict(_global_state.tokenizer_manager.server_args),
218
        **_global_state.scheduler_info,
219
        **internal_states,
220
221
222
223
        "version": __version__,
    }


224
225
226
227
228
229
@app.api_route("/set_internal_state", methods=["POST", "PUT"])
async def set_internal_state(obj: SetInternalStateReq, request: Request):
    res = await _global_state.tokenizer_manager.set_internal_state(obj)
    return res


230
231
232
233
234
235
236
237
# fastapi implicitly converts json in the request to obj (dataclass)
@app.api_route("/generate", methods=["POST", "PUT"])
async def generate_request(obj: GenerateReqInput, request: Request):
    """Handle a generate request."""
    if obj.stream:

        async def stream_results() -> AsyncIterator[bytes]:
            try:
238
                async for out in _global_state.tokenizer_manager.generate_request(
239
240
241
242
243
244
245
                    obj, request
                ):
                    yield b"data: " + orjson.dumps(
                        out, option=orjson.OPT_NON_STR_KEYS
                    ) + b"\n\n"
            except ValueError as e:
                out = {"error": {"message": str(e)}}
246
                logger.error(f"Error: {e}")
247
248
249
250
251
252
253
254
                yield b"data: " + orjson.dumps(
                    out, option=orjson.OPT_NON_STR_KEYS
                ) + b"\n\n"
            yield b"data: [DONE]\n\n"

        return StreamingResponse(
            stream_results(),
            media_type="text/event-stream",
255
            background=_global_state.tokenizer_manager.create_abort_task(obj),
256
257
258
        )
    else:
        try:
259
            ret = await _global_state.tokenizer_manager.generate_request(
260
261
262
263
264
265
266
267
                obj, request
            ).__anext__()
            return ret
        except ValueError as e:
            logger.error(f"Error: {e}")
            return _create_error_response(e)


268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
@app.api_route("/generate_from_file", methods=["POST"])
async def generate_from_file_request(file: UploadFile, request: Request):
    """Handle a generate request, this is purely to work with input_embeds."""
    content = await file.read()
    input_embeds = json.loads(content.decode("utf-8"))

    obj = GenerateReqInput(
        input_embeds=input_embeds,
        sampling_params={
            "repetition_penalty": 1.2,
            "temperature": 0.2,
            "max_new_tokens": 512,
        },
    )

    try:
        ret = await _global_state.generate_request(obj, request).__anext__()
        return ret
    except ValueError as e:
        logger.error(f"Error: {e}")
        return _create_error_response(e)


291
292
293
294
@app.api_route("/encode", methods=["POST", "PUT"])
async def encode_request(obj: EmbeddingReqInput, request: Request):
    """Handle an embedding request."""
    try:
295
        ret = await _global_state.tokenizer_manager.generate_request(
296
297
298
299
300
301
302
303
304
305
306
            obj, request
        ).__anext__()
        return ret
    except ValueError as e:
        return _create_error_response(e)


@app.api_route("/classify", methods=["POST", "PUT"])
async def classify_request(obj: EmbeddingReqInput, request: Request):
    """Handle a reward model request. Now the arguments and return values are the same as embedding models."""
    try:
307
        ret = await _global_state.tokenizer_manager.generate_request(
308
309
310
311
312
313
314
            obj, request
        ).__anext__()
        return ret
    except ValueError as e:
        return _create_error_response(e)


315
@app.api_route("/flush_cache", methods=["GET", "POST"])
316
317
async def flush_cache():
    """Flush the radix cache."""
318
    ret = await _global_state.tokenizer_manager.flush_cache()
319
320
321
    return Response(
        content="Cache flushed.\nPlease check backend logs for more details. "
        "(When there are running or waiting requests, the operation will not be performed.)\n",
322
        status_code=200 if ret.success else HTTPStatus.BAD_REQUEST,
323
324
325
326
    )


@app.api_route("/start_profile", methods=["GET", "POST"])
327
async def start_profile_async(obj: Optional[ProfileReqInput] = None):
328
    """Start profiling."""
329
330
331
332
333
334
    if obj is None:
        obj = ProfileReqInput()

    await _global_state.tokenizer_manager.start_profile(
        obj.output_dir, obj.num_steps, obj.activities
    )
335
336
337
338
339
340
341
342
343
    return Response(
        content="Start profiling.\n",
        status_code=200,
    )


@app.api_route("/stop_profile", methods=["GET", "POST"])
async def stop_profile_async():
    """Stop profiling."""
344
    _global_state.tokenizer_manager.stop_profile()
345
346
347
348
349
350
    return Response(
        content="Stop profiling. This will take some time.\n",
        status_code=200,
    )


351
352
353
@app.api_route("/start_expert_distribution_record", methods=["GET", "POST"])
async def start_expert_distribution_record_async():
    """Start recording the expert distribution. Clear the previous record if any."""
354
    await _global_state.tokenizer_manager.start_expert_distribution_record()
355
356
357
358
359
360
361
362
363
    return Response(
        content="Start recording the expert distribution.\n",
        status_code=200,
    )


@app.api_route("/stop_expert_distribution_record", methods=["GET", "POST"])
async def stop_expert_distribution_record_async():
    """Stop recording the expert distribution."""
364
    await _global_state.tokenizer_manager.stop_expert_distribution_record()
365
366
367
368
369
370
371
372
373
    return Response(
        content="Stop recording the expert distribution.\n",
        status_code=200,
    )


@app.api_route("/dump_expert_distribution_record", methods=["GET", "POST"])
async def dump_expert_distribution_record_async():
    """Dump expert distribution record."""
374
    await _global_state.tokenizer_manager.dump_expert_distribution_record()
375
376
377
378
379
380
    return Response(
        content="Dump expert distribution record.\n",
        status_code=200,
    )


381
382
@app.post("/update_weights_from_disk")
async def update_weights_from_disk(obj: UpdateWeightFromDiskReqInput, request: Request):
383
384
385
    """Update the weights from disk inplace without re-launching the server."""
    success, message, num_paused_requests = (
        await _global_state.tokenizer_manager.update_weights_from_disk(obj, request)
386
    )
387
388
389
390
391
    content = {
        "success": success,
        "message": message,
        "num_paused_requests": num_paused_requests,
    }
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
    if success:
        return ORJSONResponse(
            content,
            status_code=HTTPStatus.OK,
        )
    else:
        return ORJSONResponse(
            content,
            status_code=HTTPStatus.BAD_REQUEST,
        )


@app.post("/init_weights_update_group")
async def init_weights_update_group(
    obj: InitWeightsUpdateGroupReqInput, request: Request
):
    """Initialize the parameter update group."""
409
    success, message = await _global_state.tokenizer_manager.init_weights_update_group(
410
411
412
413
414
415
416
417
418
        obj, request
    )
    content = {"success": success, "message": message}
    if success:
        return ORJSONResponse(content, status_code=200)
    else:
        return ORJSONResponse(content, status_code=HTTPStatus.BAD_REQUEST)


419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
@app.post("/update_weights_from_tensor")
async def update_weights_from_tensor(
    obj: UpdateWeightsFromTensorReqInput, request: Request
):
    """Update the weights from tensor inplace without re-launching the server.
    Notes:
    1. Ensure that the model is on the correct device (e.g., GPU) before calling this endpoint. If the model is moved to the CPU unexpectedly, it may cause performance issues or runtime errors.
    2. HTTP will transmit only the metadata of the tensor, while the tensor itself will be directly copied to the model.
    3. Any binary data in the named tensors should be base64 encoded.
    """

    success, message = await _global_state.tokenizer_manager.update_weights_from_tensor(
        obj, request
    )
    content = {"success": success, "message": message}
    return ORJSONResponse(
        content, status_code=200 if success else HTTPStatus.BAD_REQUEST
    )


439
440
441
442
443
@app.post("/update_weights_from_distributed")
async def update_weights_from_distributed(
    obj: UpdateWeightsFromDistributedReqInput, request: Request
):
    """Update model parameter from distributed online."""
444
445
446
447
    success, message = (
        await _global_state.tokenizer_manager.update_weights_from_distributed(
            obj, request
        )
448
449
450
451
452
453
454
455
456
457
458
459
    )
    content = {"success": success, "message": message}
    if success:
        return ORJSONResponse(content, status_code=200)
    else:
        return ORJSONResponse(content, status_code=HTTPStatus.BAD_REQUEST)


@app.api_route("/get_weights_by_name", methods=["GET", "POST"])
async def get_weights_by_name(obj: GetWeightsByNameReqInput, request: Request):
    """Get model parameter by name."""
    try:
460
        ret = await _global_state.tokenizer_manager.get_weights_by_name(obj, request)
461
462
463
464
465
466
467
468
469
470
471
472
        if ret is None:
            return _create_error_response("Get parameter by name failed")
        else:
            return ORJSONResponse(ret, status_code=200)
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/release_memory_occupation", methods=["GET", "POST"])
async def release_memory_occupation(
    obj: ReleaseMemoryOccupationReqInput, request: Request
):
473
    """Release GPU memory occupation temporarily."""
474
    try:
475
        await _global_state.tokenizer_manager.release_memory_occupation(obj, request)
476
477
478
479
480
481
482
483
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/resume_memory_occupation", methods=["GET", "POST"])
async def resume_memory_occupation(
    obj: ResumeMemoryOccupationReqInput, request: Request
):
484
    """Resume GPU memory occupation."""
485
    try:
486
        await _global_state.tokenizer_manager.resume_memory_occupation(obj, request)
487
488
489
490
491
492
493
494
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/open_session", methods=["GET", "POST"])
async def open_session(obj: OpenSessionReqInput, request: Request):
    """Open a session, and return its unique session id."""
    try:
495
        session_id = await _global_state.tokenizer_manager.open_session(obj, request)
496
497
498
499
500
501
502
503
504
505
506
        if session_id is None:
            raise Exception(
                "Failed to open the session. Check if a session with the same id is still open."
            )
        return session_id
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/close_session", methods=["GET", "POST"])
async def close_session(obj: CloseSessionReqInput, request: Request):
507
    """Close the session."""
508
    try:
509
        await _global_state.tokenizer_manager.close_session(obj, request)
510
511
512
513
514
515
516
        return Response(status_code=200)
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/configure_logging", methods=["GET", "POST"])
async def configure_logging(obj: ConfigureLoggingReq, request: Request):
517
    """Configure the request logging options."""
518
    _global_state.tokenizer_manager.configure_logging(obj)
519
520
521
    return Response(status_code=200)


522
523
@app.post("/parse_function_call")
async def parse_function_call_request(obj: ParseFunctionCallReq, request: Request):
YAMY's avatar
YAMY committed
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
    """
    A native API endpoint to parse function calls from a text.
    """
    # 1) Initialize the parser based on the request body
    parser = FunctionCallParser(tools=obj.tools, tool_call_parser=obj.tool_call_parser)

    # 2) Call the non-stream parsing method (non-stream)
    normal_text, calls = parser.parse_non_stream(obj.text)

    # 3) Organize the response content
    response_data = {
        "normal_text": normal_text,
        "calls": [
            call.model_dump() for call in calls
        ],  # Convert pydantic objects to dictionaries
    }

    return ORJSONResponse(content=response_data, status_code=200)


Xihuai Wang's avatar
Xihuai Wang committed
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@app.post("/separate_reasoning")
async def separate_reasoning_request(obj: SeparateReasoningReqInput, request: Request):
    """
    A native API endpoint to separate reasoning from a text.
    """
    # 1) Initialize the parser based on the request body
    parser = ReasoningParser(model_type=obj.reasoning_parser)

    # 2) Call the non-stream parsing method (non-stream)
    reasoning_text, normal_text = parser.parse_non_stream(obj.text)

    # 3) Organize the response content
    response_data = {
        "reasoning_text": reasoning_text,
        "text": normal_text,
    }

    return ORJSONResponse(content=response_data, status_code=200)


564
565
566
567
568
##### OpenAI-compatible API endpoints #####


@app.post("/v1/completions")
async def openai_v1_completions(raw_request: Request):
569
    return await v1_completions(_global_state.tokenizer_manager, raw_request)
570
571
572
573


@app.post("/v1/chat/completions")
async def openai_v1_chat_completions(raw_request: Request):
574
    return await v1_chat_completions(_global_state.tokenizer_manager, raw_request)
575
576
577
578


@app.post("/v1/embeddings", response_class=ORJSONResponse)
async def openai_v1_embeddings(raw_request: Request):
579
    response = await v1_embeddings(_global_state.tokenizer_manager, raw_request)
580
581
582
583
584
585
    return response


@app.get("/v1/models", response_class=ORJSONResponse)
def available_models():
    """Show available models."""
586
    served_model_names = [_global_state.tokenizer_manager.served_model_name]
587
588
    model_cards = []
    for served_model_name in served_model_names:
589
590
591
592
593
594
595
        model_cards.append(
            ModelCard(
                id=served_model_name,
                root=served_model_name,
                max_model_len=_global_state.tokenizer_manager.model_config.context_len,
            )
        )
596
597
598
599
600
601
    return ModelList(data=model_cards)


@app.post("/v1/files")
async def openai_v1_files(file: UploadFile = File(...), purpose: str = Form("batch")):
    return await v1_files_create(
602
        file, purpose, _global_state.tokenizer_manager.server_args.file_storage_path
603
604
605
606
607
608
609
610
611
612
613
    )


@app.delete("/v1/files/{file_id}")
async def delete_file(file_id: str):
    # https://platform.openai.com/docs/api-reference/files/delete
    return await v1_delete_file(file_id)


@app.post("/v1/batches")
async def openai_v1_batches(raw_request: Request):
614
    return await v1_batches(_global_state.tokenizer_manager, raw_request)
615
616
617
618
619


@app.post("/v1/batches/{batch_id}/cancel")
async def cancel_batches(batch_id: str):
    # https://platform.openai.com/docs/api-reference/batch/cancel
620
    return await v1_cancel_batch(_global_state.tokenizer_manager, batch_id)
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639


@app.get("/v1/batches/{batch_id}")
async def retrieve_batch(batch_id: str):
    return await v1_retrieve_batch(batch_id)


@app.get("/v1/files/{file_id}")
async def retrieve_file(file_id: str):
    # https://platform.openai.com/docs/api-reference/files/retrieve
    return await v1_retrieve_file(file_id)


@app.get("/v1/files/{file_id}/content")
async def retrieve_file_content(file_id: str):
    # https://platform.openai.com/docs/api-reference/files/retrieve-contents
    return await v1_retrieve_file_content(file_id)


640
641
642
643
644
645
646
647
648
649
650
651
## SageMaker API
@app.get("/ping")
async def sagemaker_health() -> Response:
    """Check the health of the http server."""
    return Response(status_code=200)


@app.post("/invocations")
async def sagemaker_chat_completions(raw_request: Request):
    return await v1_chat_completions(_global_state.tokenizer_manager, raw_request)


652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
## Vertex AI API
@app.post(os.environ.get("AIP_PREDICT_ROUTE", "/vertex_generate"))
async def vertex_generate(vertex_req: VertexGenerateReqInput, raw_request: Request):
    if not vertex_req.instances:
        return []
    inputs = {}
    for input_key in ("text", "input_ids", "input_embeds"):
        if vertex_req.instances[0].get(input_key):
            inputs[input_key] = [
                instance.get(input_key) for instance in vertex_req.instances
            ]
            break
    image_data = [
        instance.get("image_data")
        for instance in vertex_req.instances
        if instance.get("image_data") is not None
    ] or None
    req = GenerateReqInput(
        **inputs,
        image_data=image_data,
        **(vertex_req.parameters or {}),
    )
    ret = await generate_request(req, raw_request)
    return ORJSONResponse({"predictions": ret})


678
679
680
681
682
683
684
685
686
def _create_error_response(e):
    return ORJSONResponse(
        {"error": {"message": str(e)}}, status_code=HTTPStatus.BAD_REQUEST
    )


def launch_server(
    server_args: ServerArgs,
    pipe_finish_writer: Optional[multiprocessing.connection.Connection] = None,
687
    launch_callback: Optional[Callable[[], None]] = None,
688
689
690
691
692
693
694
695
):
    """
    Launch SRT (SGLang Runtime) Server.

    The SRT server consists of an HTTP server and an SRT engine.

    - HTTP server: A FastAPI server that routes requests to the engine.
    - The engine consists of three components:
696
        1. TokenizerManager: Tokenizes the requests and sends them to the scheduler.
697
698
699
700
        2. Scheduler (subprocess): Receives requests from the Tokenizer Manager, schedules batches, forwards them, and sends the output tokens to the Detokenizer Manager.
        3. DetokenizerManager (subprocess): Detokenizes the output tokens and sends the result back to the Tokenizer Manager.

    Note:
701
    1. The HTTP server, Engine, and TokenizerManager both run in the main process.
702
    2. Inter-process communication is done through IPC (each process uses a different port) via the ZMQ library.
703
    """
704
    tokenizer_manager, scheduler_info = _launch_subprocesses(server_args=server_args)
705
706
    set_global_state(
        _GlobalState(
707
            tokenizer_manager=tokenizer_manager,
708
709
710
711
712
713
714
715
716
717
718
719
720
            scheduler_info=scheduler_info,
        )
    )

    # Add api key authorization
    if server_args.api_key:
        add_api_key_middleware(app, server_args.api_key)

    # Add prometheus middleware
    if server_args.enable_metrics:
        add_prometheus_middleware(app)
        enable_func_timer()

721
722
723
    # Send a warmup request - we will create the thread launch it
    # in the lifespan after all other warmups have fired.
    warmup_thread = threading.Thread(
724
725
726
727
        target=_wait_and_warmup,
        args=(
            server_args,
            pipe_finish_writer,
728
            _global_state.tokenizer_manager.image_token_id,
729
            launch_callback,
730
731
        ),
    )
732
    app.warmup_thread = warmup_thread
733
734
735
736

    try:
        # Update logging configs
        set_uvicorn_logging_configs()
737
        app.server_args = server_args
738
739
740
741
742
743
744
745
746
747
        # Listen for HTTP requests
        uvicorn.run(
            app,
            host=server_args.host,
            port=server_args.port,
            log_level=server_args.log_level_http or server_args.log_level,
            timeout_keep_alive=5,
            loop="uvloop",
        )
    finally:
748
        warmup_thread.join()
749
750


751
752
753
754
755
756
def _wait_and_warmup(
    server_args: ServerArgs,
    pipe_finish_writer: Optional[multiprocessing.connection.Connection],
    image_token_text: str,
    launch_callback: Optional[Callable[[], None]] = None,
):
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
    headers = {}
    url = server_args.url()
    if server_args.api_key:
        headers["Authorization"] = f"Bearer {server_args.api_key}"

    # Wait until the server is launched
    success = False
    for _ in range(120):
        time.sleep(1)
        try:
            res = requests.get(url + "/get_model_info", timeout=5, headers=headers)
            assert res.status_code == 200, f"{res=}, {res.text=}"
            success = True
            break
        except (AssertionError, requests.exceptions.RequestException):
            last_traceback = get_exception_traceback()
            pass

    if not success:
        if pipe_finish_writer is not None:
            pipe_finish_writer.send(last_traceback)
        logger.error(f"Initialization failed. warmup error: {last_traceback}")
        kill_process_tree(os.getpid())
        return

    model_info = res.json()

    # Send a warmup request
    request_name = "/generate" if model_info["is_generation"] else "/encode"
    max_new_tokens = 8 if model_info["is_generation"] else 1
    json_data = {
        "sampling_params": {
            "temperature": 0,
            "max_new_tokens": max_new_tokens,
        },
    }
    if server_args.skip_tokenizer_init:
fzyzcjy's avatar
fzyzcjy committed
794
        json_data["input_ids"] = [[10, 11, 12] for _ in range(server_args.dp_size)]
fzyzcjy's avatar
fzyzcjy committed
795
796
797
        # TODO Workaround the bug that embedding errors for list of size 1
        if server_args.dp_size == 1:
            json_data["input_ids"] = json_data["input_ids"][0]
798
    else:
fzyzcjy's avatar
fzyzcjy committed
799
        json_data["text"] = ["The capital city of France is"] * server_args.dp_size
fzyzcjy's avatar
fzyzcjy committed
800
801
802
        # TODO Workaround the bug that embedding errors for list of size 1
        if server_args.dp_size == 1:
            json_data["text"] = json_data["text"][0]
803

804
805
806
807
808
809
810
811
    # Debug dumping
    if server_args.debug_tensor_dump_input_file:
        json_data.pop("text", None)
        json_data["input_ids"] = np.load(
            server_args.debug_tensor_dump_input_file
        ).tolist()
        json_data["sampling_params"]["max_new_tokens"] = 0

812
    try:
813
814
815
816
817
818
819
820
821
822
823
        if server_args.disaggregation_mode == "null":
            res = requests.post(
                url + request_name,
                json=json_data,
                headers=headers,
                timeout=600,
            )
            assert res.status_code == 200, f"{res}"
        else:
            # Warmup request currently hangs in disaggregation mode, so we skip it.
            logger.info("Skipping warmup request in disaggregation mode")
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
    except Exception:
        last_traceback = get_exception_traceback()
        if pipe_finish_writer is not None:
            pipe_finish_writer.send(last_traceback)
        logger.error(f"Initialization failed. warmup error: {last_traceback}")
        kill_process_tree(os.getpid())
        return

    # Debug print
    # logger.info(f"{res.json()=}")

    logger.info("The server is fired up and ready to roll!")
    if pipe_finish_writer is not None:
        pipe_finish_writer.send("ready")

    if server_args.delete_ckpt_after_loading:
        delete_directory(server_args.model_path)
841
842
843
844
845
846

    if server_args.debug_tensor_dump_input_file:
        kill_process_tree(os.getpid())

    if launch_callback is not None:
        launch_callback()