0820

Author: SangChengC

lightllm/server/api_http.py

@@ -100,6 +100,21 @@ def set_args(self, args):
                 visual_port=args.visual_port,
                 # metric_port=args.metric_port,
             )
+        elif args.run_mode == "llm_only":
+            init_tokenizer(args)  # for openai api
+            SamplingParams.load_generation_cfg(args.model_dir)
+            self.metric_client = MetricClient(args.metric_port)
+            self.httpserver_manager = HttpServerManager(
+                args,
+                router_port=args.router_port,
+                cache_port=None,
+                detokenization_pub_port=args.detokenization_pub_port,
+                visual_port=None,
+                enable_multimodal=args.enable_multimodal,
+                metric_port=args.metric_port,
+            )
+            dp_size_in_node = max(1, args.dp // args.nnodes)  # 兼容多机纯tp的运行模式，这时候 1 // 2 == 0, 需要兼容
+            self.shared_token_load = TokenLoad(f"{get_unique_server_name()}_shared_token_load", dp_size_in_node)
         else:
             init_tokenizer(args)  # for openai api
             SamplingParams.load_generation_cfg(args.model_dir)

lightllm/server/api_server.py

@@ -5,13 +5,15 @@
     torch.multiprocessing.set_start_method("spawn")  # this code will not be ok for settings to fork to subprocess
     parser = make_argument_parser()
     args = parser.parse_args()
-    from .api_start import pd_master_start, normal_or_p_d_start, visual_only_start, config_server_start
+    from .api_start import pd_master_start, normal_or_p_d_start, visual_only_start, config_server_start, llm_only_start
 
     if args.run_mode == "pd_master":
         pd_master_start(args)
     elif args.run_mode == "config_server":
         config_server_start(args)
     elif args.run_mode == "visual_only":
         visual_only_start(args)
+    elif args.run_mode == "llm_only":
+        llm_only_start(args)
     else:
         normal_or_p_d_start(args)

lightllm/server/api_start.py

@@ -362,6 +362,105 @@ def normal_or_p_d_start(args):
     return
 
 
+def llm_only_start(args):
+
+    check_and_set_args(args)
+    already_uesd_ports = [args.nccl_port, args.port]
+
+    # 提前锁定端口，防止在单个机器上启动多个实列的时候，要到模型启动的时候才能
+    # 捕获到端口设置冲突的问题
+    ports_locker = PortLocker(already_uesd_ports)
+    ports_locker.lock_port()
+
+    node_world_size = args.tp // args.nnodes
+    can_use_ports = alloc_can_use_network_port(num=4 + node_world_size, used_nccl_ports=already_uesd_ports)
+    logger.info(f"alloced ports: {can_use_ports}")
+    (
+        router_port,
+        detokenization_port,
+        detokenization_pub_port,
+        metric_port,
+    ) = can_use_ports[0:4]
+    can_use_ports = can_use_ports[4:]
+
+    # 将申请好的端口放入args参数中
+    args.router_port = router_port
+    args.detokenization_port = detokenization_port
+    args.detokenization_pub_port = detokenization_pub_port
+    args.metric_port = metric_port
+
+    # 申请在 p d 分离模式下，会用的端口
+    args.pd_node_infer_rpyc_ports = can_use_ports[0:node_world_size]
+    # p d 分离模式下用于标识节点的id
+    args.pd_node_id = uuid.uuid4().int
+    # p 节点用来建立torch kv 传输分布组的可用端口范围
+    args.pd_p_allowed_port_min = 20000
+    args.pd_p_allowed_port_max = 30000
+
+    # p d 分离模式下，decode节点的调度间隙是0
+    if args.run_mode == "decode":
+        args.router_max_wait_tokens = 0
+
+    send_and_receive_node_ip(args)  # 多机用于收发node ip
+    set_env_start_args(args)
+    logger.info(f"all start args:{args}")
+
+    ports_locker.release_port()
+
+    process_manager.start_submodule_processes(
+        start_funcs=[
+            start_metric_manager,
+        ],
+        start_args=[(metric_port, args)],
+    )
+
+    process_manager.start_submodule_processes(
+        start_funcs=[start_router_process, start_detokenization_process],
+        start_args=[
+            (args, router_port, detokenization_port, metric_port),
+            (args, detokenization_port, detokenization_pub_port),
+        ],
+    )
+
+    # 启动 gunicorn
+    command = [
+        "gunicorn",
+        "--workers",
+        f"{args.httpserver_workers}",
+        "--worker-class",
+        "uvicorn.workers.UvicornWorker",
+        "--bind",
+        f"{args.host}:{args.port}",
+        "--log-level",
+        "info",
+        "--access-logfile",
+        "-",
+        "--error-logfile",
+        "-",
+        "lightllm.server.api_http:app",
+        "--timeout",
+        f"{get_lightllm_gunicorn_time_out_seconds()}",
+        "--keep-alive",
+        f"{get_lightllm_gunicorn_keep_alive()}",
+    ]
+
+    # 启动子进程
+    http_server_process = subprocess.Popen(command)
+
+    if "s3://" in args.model_dir:
+        from lightllm.utils.petrel_helper import s3_model_clear
+
+        s3_model_clear(args.model_dir)
+
+    if args.health_monitor:
+        from lightllm.server.health_monitor.manager import start_health_check_process
+
+        process_manager.start_submodule_processes(start_funcs=[start_health_check_process], start_args=[(args,)])
+    setup_signal_handlers(http_server_process, process_manager)
+    http_server_process.wait()
+    return
+
+
 def pd_master_start(args):
     set_unique_server_name(args)
     if args.run_mode != "pd_master":

lightllm/server/embed_cache/impl/naive_memory_cache.py

@@ -65,7 +65,7 @@ def _check_and_set_new_id_range(self, alloced_token_num):
                     except BaseException as e:
                         logger.exception(str(e))
                         time.sleep(3)
-        return self.token_id_range_start
+        return
 
     def _clear(self, free_max_count: int):
         deleted = 0

lightllm/server/httpserver/manager.py

@@ -81,11 +81,13 @@ def __init__(
                 )
 
         self.enable_multimodal = enable_multimodal
-        if self.enable_multimodal:
+        if self.enable_multimodal and self.args.run_mode != "llm_only":
             self.cache_client = rpyc.connect("localhost", cache_port, config={"allow_pickle": True})
-            if self.args.run_mode != "llm_only":
-                self.send_to_visual = context.socket(zmq.PUSH)
-                self.send_to_visual.connect(f"{args.zmq_mode}127.0.0.1:{visual_port}")
+            self.send_to_visual = context.socket(zmq.PUSH)
+            self.send_to_visual.connect(f"{args.zmq_mode}127.0.0.1:{visual_port}")
+
+        self.token_id_range_start = 100000000
+        self.token_id_range_end = 2 ** 63 - 1
 
         self.shm_req_manager = ShmReqManager()
 
@@ -115,6 +117,10 @@ def __init__(
         self.latest_success_infer_time_mark.set_value(int(time.time()))
         return
 
+    async def _check_and_set_new_id_range(self, token_num):
+        assert self.token_id_range_start + token_num < self.token_id_range_end
+        self.token_id_range_start += token_num
+
     async def _alloc_resource(self, items, md5sums, token_nums, datas):
 
         while True:
@@ -199,7 +205,7 @@ async def _get_image_embedding_from_afs(self, multimodal_params: MultimodalParam
                 await self._wait_for_afs_embed(md5sum)
             img.uuid = uid_int
             img.afs_embed = True
-            token_id_range_start = self.cache_client.root._check_and_set_new_id_range(token_num)
+            token_id_range_start = self.token_id_range_start
             img.token_id = token_id_range_start
             img.token_num = token_num
 
@@ -216,7 +222,7 @@ async def _get_image_embedding_from_afs(self, multimodal_params: MultimodalParam
             uid_int = int(md5sum, 16)
             audio.uuid = uid_int
             audio.afs_embed = True
-            token_id_range_start = self.cache_client.root._check_and_set_new_id_range(token_num)
+            token_id_range_start = self.token_id_range_start
             audio.token_id = token_id_range_start
             audio.token_num = token_num
         return

lightllm/server/httpserver_for_visual_only/manager.py

@@ -238,7 +238,7 @@ async def generate(
                 req_obj.init(
                     group_request_id + i,
                     #  随便写的，后面改掉
-                    [24, 67],
+                    [21456],
                     sampling_params,
                     self.tokenizer,
                     chunked_prefill_size=self.args.chunked_prefill_size,