0909

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 = img._preload_data
+                image_data = read_shm(get_shm_name_data(img.uuid))
                 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/embed_cache/impl/memory_cache_with_redis.py

@@ -47,6 +47,17 @@ def set_items_embed(self, ids: list[int]) -> None:
                 self._records[id].ref -= 1
 
     def get_items_embed(self, ids: list[int]) -> list[Optional[bool]]:
+        ret = []
+        for id in ids:
+            print(f"id is {id}")
+            print(f"self.redis_cache.query(str(id)) is {self.redis_cache.query(str(id))}")
+            exist = self.redis_cache.query(str(id))
+            ret.append(exist)
+            if exist:
+                self._records[id].embed = True
+        return ret
+
+    def get_items_embed_and_incre(self, ids: list[int]) -> list[Optional[bool]]:
         ret = []
         for id in ids:
             # if self.redis_cache.query(str(id)):

lightllm/server/embed_cache/utils.py

@@ -326,17 +326,19 @@ def _delete_afs_files(self, victims: List[str]) -> None:
   return {-1, 0}  -- Not found
 end
 
+--ref 递减到 0 时保留键，只更新计数与 LRU
 local rc = tonumber(val) - 1
-if rc <= 0 then
-  redis.call('DEL', ref_key)
-  redis.call('ZREM', zset, md5)
-  return {0, 1}  -- Deleted
-else
-  redis.call('SET', ref_key, rc)
-  local now = redis.call('TIME')[1] * 1000
-  redis.call('ZADD', zset, now, md5)
-  return {rc, 0}  -- Updated
+if rc < 0 then
+  rc = 0
 end
+
+redis.call('SET', ref_key, rc)
+
+-- 更新 LRU 时间戳（最近释放的条目更不容易被立即逐出）
+local now = redis.call('TIME')[1] * 1000
+redis.call('ZADD', zset, now, md5)
+
+return {rc, 0}  -- 未删除
 """
 
     _EVICT_AND_INSERT_LUA = r"""

lightllm/server/httpserver/manager.py

@@ -687,7 +687,6 @@ async def recycle_resource_loop(self):
                 for req in req_status.group_req_objs.shm_req_objs:
                     await self.shm_req_manager.async_put_back_req_obj(req)
                     await self.shm_req_manager.async_release_req_index(req.index_in_shm_mem)
-                print("begin release")
                 await self._release_multimodal_resources(req_status.group_req_objs.multimodal_params)
 
             # 先保留这个关键得日志，用于方便定位重构中的问题。

lightllm/server/visualserver/manager.py

@@ -160,13 +160,14 @@ def _recv_reqs(self):
             #     img for img in recv_req.multimodal_params.images
             #     if not self.cache_client.root.get_item_embed(img.uuid)  # embed已存在的被丢弃 , ref +1
             # ]
-            uuids = []
+            uuids = [img.uuid for img in recv_req.multimodal_params.images]
+            already_embed = self.cache_client.root.get_items_embed(uuids)
             token_nums = []
-            for img in recv_req.multimodal_params.images:
-                uuids.append(img.uuid)
-                token_nums.append(img.token_num)
-                record = self.cache_client.root.alloc(uuids, token_nums)
-            print(f"record is {record}")
+            for img, embed in zip(recv_req.multimodal_params.images, already_embed):
+                if not embed:
+                    uuids.append(img.uuid)
+                    token_nums.append(img.token_num)
+            self.cache_client.root.alloc(uuids, token_nums)
             return recv_req
         else:
             return self.vit_receiver.recv_pyobj(zmq.NOBLOCK)
@@ -182,6 +183,7 @@ async def loop_for_netio_req(self):
                     if isinstance(recv_req, GroupReqIndexes):
                         # print(recv_req, flush=True)
                         self.waiting_reqs.append(recv_req)
+                        print(f"recv_req.multimodal_params is {recv_req.multimodal_params}")
                     else:
                         assert False, f"Error Req Inf {recv_req}"
                 self.visual_recv_max_count = min(self.visual_recv_max_count * 1.3, 256)

lightllm/server/visualserver/vit_connect.py

@@ -5,6 +5,8 @@
 import pickle
 from typing import Dict, List, Optional, Any
 from lightllm.utils.log_utils import init_logger
+from lightllm.server.core.objs.io_objs import GroupReqObjs, GroupReqIndexes
+from lightllm.server.multimodal_params import MultimodalParams
 import httpx
 import base64
 from dataclasses import dataclass
@@ -48,8 +50,10 @@ def _setup_vit_connections(self):
         """
         if self.remote_vit:
             # 远程VIT实例模式
+            print("remote")
             self._setup_remote_vit_connections()
         else:
+            print("not remote")
             self._setup_local_vit_connection()
 
     def _setup_local_vit_connection(self):
@@ -156,31 +160,33 @@ def _get_vit_instance(self):
         self.current_vit_index = index
         return list(self.remote_vit_instances.values())[index]
 
-    async def send_to_vit(self, data, protocol=pickle.HIGHEST_PROTOCOL):
+    async def send_to_vit(self, req: GroupReqIndexes, protocol=pickle.HIGHEST_PROTOCOL):
         """
         发送数据到VIT实例，支持本地和远程模式
         """
         instance = self._get_vit_instance()
         # 本地模式下，提前释放图片资源，降低传输开销
         if not self.remote_vit:
-            data.multimodal_params.free()
+            req.multimodal_params.free()
 
         try:
             print(instance, flush=True)
-            instance.send_pyobj(data, protocol=protocol)
+            instance.send_pyobj(req, protocol=protocol)
         except Exception as e:
             logger.error(f"Failed to send to VIT instance: {e}")
             raise Exception(f"Failed to send to VIT instance: {e}")
 
         # 远程模式下，发送完以后，在释放图片资源
-        await self._wait_visual_embed_ready(data)
+        await self._wait_visual_embed_ready(req)
         if self.remote_vit:
-            data.multimodal_params.free()
+            req.multimodal_params.free()
 
     async def vit_handle_loop(self):
         """
         异步VIT连接管理循环，由外部启动
         """
+        if not self.remote_vit:
+            return
         logger.info("Starting VIT connection management loop")
         while True:
             try:
@@ -211,12 +217,12 @@ async def _async_get_vit_objs(self) -> Optional[Dict[int, VIT_Obj]]:
             logger.exception(f"Error getting VIT instances: {e}")
             return None
 
-    async def _wait_visual_embed_ready(self, data, timeout_seconds: int = 100):
+    async def _wait_visual_embed_ready(self, req: GroupReqIndexes, timeout_seconds: int = 100):
         # 本地模式不需要等待
         if not self.remote_vit:
             return
 
-        uuids = data.multimodal_params.get_all_uuids()
+        uuids = req.multimodal_params.get_all_uuids()
         print(f"uuids is {uuids}")
 
         async def wait_for_embeds():
@@ -227,5 +233,5 @@ async def wait_for_embeds():
             await asyncio.wait_for(wait_for_embeds(), timeout=timeout_seconds)
         except asyncio.TimeoutError:
             logger.error(
-                f"Req {data.group_req_id}: timeout waiting for visual embed ready after {timeout_seconds} seconds"
+                f"Req {req.group_req_id}: timeout waiting for visual embed ready after {timeout_seconds} seconds"
             )