[0826]modify visual server

Author: SangChengC

lightllm/models/vit/model.py

@@ -178,7 +178,7 @@ def encode(self, images: List[ImageItem]):
         for i, img in enumerate(images):
             if isinstance(img, ImageItem):
                 uuids.append(img.uuid)
-                image_data = read_shm(get_shm_name_data(img.uuid))
+                image_data = img._preload_data
                 image_data = Image.open(BytesIO(image_data))
                 t = self.load_image_func(image_data, max_num=img.extra_params["image_patch_max_num"])
                 img_tensors.append(t)

lightllm/server/api_http.py

@@ -42,7 +42,6 @@
 from .httpserver.manager import HttpServerManager
 from .visualserver.manager import VisualManager
 from .httpserver_for_pd_master.manager import HttpServerManagerForPDMaster
-from .httpserver_for_visual_only.manager import HttpServerManagerForVisualOnly
 from .api_lightllm import lightllm_get_score, lightllm_get_image_embedding
 from lightllm.utils.envs_utils import get_env_start_args, get_lightllm_websocket_max_message_size
 from lightllm.utils.log_utils import init_logger
@@ -70,7 +69,7 @@ class G_Objs:
     args: object = None
     g_generate_func: Callable = None
     g_generate_stream_func: Callable = None
-    httpserver_manager: Union[HttpServerManager, HttpServerManagerForPDMaster, HttpServerManagerForVisualOnly] = None
+    httpserver_manager: Union[HttpServerManager, HttpServerManagerForPDMaster, VisualManager] = None
     visual_manager: VisualManager = None
     shared_token_load: TokenLoad = None
 
@@ -94,11 +93,12 @@ def set_args(self, args):
             )
         elif args.run_mode == "visual_only":
             self.metric_client = MetricClient(args.metric_port)
-            self.httpserver_manager = HttpServerManagerForVisualOnly(
+            self.httpserver_manager = VisualManager(
                 args,
-                cache_port=args.cache_port,
+                next_module_port=None,
                 visual_port=args.visual_port,
-                metric_port=args.metric_port,
+                cache_port=None,
+                visual_model_rpc_ports=args.visual_model_rpc_ports,
             )
         elif args.run_mode == "llm_only":
             init_tokenizer(args)  # for openai api
@@ -372,6 +372,10 @@ async def startup_event():
     logger.info("server start up")
     loop = asyncio.get_event_loop()
     g_objs.set_args(get_env_start_args())
-    loop.create_task(g_objs.httpserver_manager.handle_loop())
+    if g_objs.args.run_mode == "visual_only":
+        await g_objs.httpserver_manager.wait_to_model_ready()
+        loop.create_task(g_objs.httpserver_manager.loop_for_fwd_visual_only())
+    else:
+        loop.create_task(g_objs.httpserver_manager.handle_loop())
     logger.info(f"server start up ok, loop use is {asyncio.get_event_loop()}")
     return

lightllm/server/api_lightllm.py

@@ -5,7 +5,7 @@
 from lightllm.server.core.objs.sampling_params import SamplingParams
 from .multimodal_params import MultimodalParams
 from .httpserver.manager import HttpServerManager
-from .httpserver_for_visual_only.manager import HttpServerManagerForVisualOnly
+from .visualserver.manager import VisualManager
 from fastapi.responses import JSONResponse
 import ujson as json
 
@@ -140,9 +140,7 @@ async def stream_results() -> AsyncGenerator[bytes, None]:
     return StreamingResponse(stream_results(), media_type="text/event-stream", background=background_tasks)
 
 
-async def lightllm_get_image_embedding(
-    request: Request, httpserver_manager: HttpServerManagerForVisualOnly
-) -> Response:
+async def lightllm_get_image_embedding(request: Request, httpserver_manager: VisualManager) -> Response:
     request_dict = await request.json()
     # request_dict: {'parameters': {'max_new_tokens': 128},
     # 'multimodal_params': {'images': [{'type': 'base64', 'data': 'base64'}]}}
@@ -154,6 +152,5 @@ async def lightllm_get_image_embedding(
     multimodal_params = MultimodalParams(**multimodal_params_dict)
 
     await httpserver_manager.generate(sampling_params, multimodal_params, request=request)
-    # 5. Return JSON result
-    print("embedding OK")
+
     return JSONResponse({"message": "OK"}, status_code=200)

lightllm/server/api_start.py

@@ -548,17 +548,16 @@ def visual_only_start(args):
         return
     already_uesd_ports = args.visual_nccl_ports + [args.nccl_port, args.port]
     can_use_ports = alloc_can_use_network_port(
-        num=5 + args.visual_dp * args.visual_tp, used_nccl_ports=already_uesd_ports
+        num=4 + args.visual_dp * args.visual_tp, used_nccl_ports=already_uesd_ports
     )
     logger.info(f"alloced ports: {can_use_ports}")
     (
         router_port,
         visual_port,
         audio_port,
-        cache_port,
         metric_port,
-    ) = can_use_ports[0:5]
-    can_use_ports = can_use_ports[5:]
+    ) = can_use_ports[0:4]
+    can_use_ports = can_use_ports[4:]
 
     visual_model_tp_ports = []
     for _ in range(args.visual_dp):
@@ -570,7 +569,6 @@ def visual_only_start(args):
     args.router_port = router_port
     args.visual_port = visual_port
     args.audio_port = audio_port
-    args.cache_port = cache_port
     args.metric_port = metric_port
     args.visual_model_rpc_ports = visual_model_tp_ports
 
@@ -585,33 +583,17 @@ def visual_only_start(args):
         start_args=[(metric_port, args)],
     )
 
-    from .visualserver.manager import start_visual_process
-
-    process_manager.start_submodule_processes(
-        start_funcs=[
-            start_cache_manager,
-        ],
-        start_args=[(cache_port, args)],
-    )
-    process_manager.start_submodule_processes(
-        start_funcs=[
-            start_visual_process,
-        ],
-        start_args=[
-            (args, audio_port, visual_port, cache_port, visual_model_tp_ports),
-        ],
-    )
-    if args.enable_multimodal_audio:
-        from .audioserver.manager import start_audio_process
-
-        process_manager.start_submodule_processes(
-            start_funcs=[
-                start_audio_process,
-            ],
-            start_args=[
-                (args, router_port, audio_port, cache_port),
-            ],
-        )
+    # if args.enable_multimodal_audio:
+    #     from .audioserver.manager import start_audio_process
+
+    #     process_manager.start_submodule_processes(
+    #         start_funcs=[
+    #             start_audio_process,
+    #         ],
+    #         start_args=[
+    #             (args, router_port, audio_port, cache_port),
+    #         ],
+    #     )
 
     # 启动 gunicorn
     command = [

lightllm/server/core/objs/io_objs/group_req.py

@@ -4,6 +4,12 @@
 from ..req import Req
 
 
+@dataclass
+class VisualOnlyReqIndexes:
+    group_req_id: int
+    multimodal_params: MultimodalParams
+
+
 @dataclass
 class GroupReqIndexes:
     group_req_id: int

lightllm/server/core/objs/req.py

@@ -153,6 +153,30 @@ def init(
 
         self.post_init()
 
+    def init_visual_only(
+        self,
+        request_id: int,
+    ):
+        # 只是为了有更好的编码辅助类型提示
+        self.index_in_shm_mem: int = self.index_in_shm_mem
+        self.ref_count: int = self.ref_count
+
+        self.request_id = request_id
+        self.group_req_id = convert_sub_id_to_group_id(request_id)
+        self.is_paused = False
+        self.finish_status = FinishStatus()
+        self.is_aborted = False
+        self.router_aborted = False
+        self.shm_infer_released = False
+        self.shm_cur_kv_len = 0
+        self.shm_cur_output_len = 0
+        self.candetoken_out_len = 0
+        self.prompt_cache_len = 0
+        self.finish_token_index = -1
+        self.can_released_mark = False
+
+        self.post_init()
+
     def post_init(self):
         # 子类继承进行一些额外的初始化操作
         pass
@@ -206,7 +230,9 @@ def can_release(self):
         # 只有管理节点有一个引用
         ref_count_ok = self.ref_count == 1
         can_released_mark = self.can_released_mark
-
+        print(f"self.is_aborted is {self.is_aborted}")
+        print(f"self.finish_status.is_finished() is {self.finish_status.is_finished()}")
+        print(f"self.ref_count is {self.ref_count}")
         if self.is_aborted and can_released_mark and ref_count_ok:
             return True
 

lightllm/server/embed_cache/utils.py

@@ -44,7 +44,7 @@ def create_shm(name, data):
 def create_afs(name, data):
     try:
         data_size = len(data)
-        path = os.path.join(get_env_start_args().visual_embed_path, name)
+        path = os.path.join("/mtc/sangchengmeng/afs", name)
         with open(path, "xb") as f:
             mem_view = memoryview(data)
             f.write(mem_view[:data_size])
@@ -79,7 +79,6 @@ def get_shm_name_data(uid):
 def get_shm_name_embed(uid):
     return str(uid) + "-embed"
 
-
 """
 Importable Redis-backed MD5 refcount with LRU eviction.
 
@@ -377,4 +376,4 @@ def _delete_afs_files(self, victims: List[str]) -> None:
 else
     return {0}  -- 逐出失败，没有足够的候选
 end
-"""
+"""