Merge branch 'develop' into metax-dev-v1-loader

Author: plusNew001

fastdeploy/engine/sched/resource_manager_v1.py

@@ -728,14 +728,7 @@ def _allocate_decode_and_extend():
             if scheduled_reqs:
                 llm_logger.debug(f"schedued_reqs: {scheduled_reqs}")
 
-            # Update metrics
-            num_tasks = sum([1 if task else 0 for task in self.tasks_list])
-            num_blocks_used_by_tasks = sum([len(task.block_tables) if task else 0 for task in self.tasks_list])
-            main_process_metrics.available_gpu_block_num.set(self.total_block_number() - num_blocks_used_by_tasks)
-            main_process_metrics.batch_size.set(self.max_num_seqs - self.available_batch())
-            main_process_metrics.gpu_cache_usage_perc.set(self.get_gpu_cache_usage_perc())
-            main_process_metrics.num_requests_running.set(len(self.running))
-            main_process_metrics.num_requests_waiting.set(num_tasks - len(self.running))
+            self.update_metrics()
 
             return scheduled_reqs
 
@@ -962,7 +955,10 @@ def finish_requests(self, request_ids: Union[str, Iterable[str]]):
                     if request in self.running:  # normally run and finished
                         self.running.remove(request)
                         request.status = RequestStatus.FINISHED
-                        self._free_blocks(request)
+                        try:
+                            self._free_blocks(request)
+                        except Exception as e:
+                            llm_logger.warning(f"release block failed {req_id}: {e}")
                     if (
                         request.request_id in self.to_be_rescheduled_request_id_set
                     ):  # finished after preempted, blocks have been recycled.
@@ -981,7 +977,19 @@ def finish_requests(self, request_ids: Union[str, Iterable[str]]):
                         del self.req_dict[req_id]
         except Exception as e:
             llm_logger.error(f"finish_request err: {e}, {str(traceback.format_exc())}")
+        finally:
+            self.update_metrics()
 
     def clear_data(self):
         self.waiting: deque[Request] = deque()
         self.to_be_rescheduled_request_id_set = set()
+
+    def update_metrics(self):
+        # Update metrics
+        num_tasks = sum([1 if task else 0 for task in self.tasks_list])
+        num_blocks_used_by_tasks = sum([len(task.block_tables) if task else 0 for task in self.tasks_list])
+        main_process_metrics.available_gpu_block_num.set(self.total_block_number() - num_blocks_used_by_tasks)
+        main_process_metrics.batch_size.set(self.max_num_seqs - self.available_batch())
+        main_process_metrics.gpu_cache_usage_perc.set(self.get_gpu_cache_usage_perc())
+        main_process_metrics.num_requests_running.set(len(self.running))
+        main_process_metrics.num_requests_waiting.set(num_tasks - len(self.running))

fastdeploy/model_executor/layers/mtp_linear.py

@@ -76,7 +76,7 @@ def __init__(
                 self.linear.weight,
                 {
                     "weight_loader": default_weight_loader(self.fd_config),
-                    "model_format": self.fd_config.model_config.model_format,
+                    "weight_need_transpose": self.fd_config.model_config.model_format == "torch",
                 },
             )
             if self.bias_key is not None:
@@ -100,7 +100,7 @@ def __init__(
                 self.linear.weight,
                 {
                     "weight_loader": default_weight_loader(self.fd_config),
-                    "model_format": self.fd_config.model_config.model_format,
+                    "weight_need_transpose": self.fd_config.model_config.model_format == "torch",
                 },
             )
             if self.nranks > 1:

fastdeploy/output/token_processor.py

@@ -847,7 +847,7 @@ def _record_speculative_decoding_mertics(self, accept_num):
     def clear_data(self):
         if envs.ENABLE_V1_KVCACHE_SCHEDULER:
             self.resource_manager.clear_data()
-        for i in range(self.cfg.max_num_seqs):
+        for i in range(self.resource_manager.max_num_seqs):
             if self.resource_manager.stop_flags[i]:
                 continue
             task = self.resource_manager.tasks_list[i]

fastdeploy/worker/worker_process.py

@@ -340,11 +340,14 @@ def event_loop_normal(self) -> None:
                 mmap_infos = create_mmap(
                     [MODEL_MAIN_NAME], self.local_rank, self.ranks, shm_uuid=os.getenv("SHM_UUID", ""), logger=logger
                 )
+
+        tp_size = self.parallel_config.tensor_parallel_size
         # Currently, only support single node
-        self.nnode = int((self.parallel_config.tensor_parallel_size + 7) // 8)
+        self.nnode = int((tp_size + 7) // 8)
         req_ids = []
         num_running_requests = 0
-        local_rank = self.local_rank % self.parallel_config.tensor_parallel_size
+        tp_rank = self.local_rank % tp_size
+
         self.model_weights_signal = np.zeros([1], dtype=np.int32)
         while True:
             if self.eplb_config.enable_redundant_experts:
@@ -385,35 +388,34 @@ def event_loop_normal(self) -> None:
                     if self.local_rank == 0:
                         rearrange_experts_status_array[0] = RearrangeExpertState.done.value
                     logger.info("redundant_expert: done")
-            if self.local_rank % self.parallel_config.tensor_parallel_size == 0:
+            if tp_rank == 0:
                 if self.model_weights_status.value[0] != ModelWeightsStatus.NORMAL:
                     self.model_weights_signal[0] = int(self.model_weights_status.value[0])
                 if self.fd_config.load_config.dynamic_load_weight and self.parallel_config.enable_expert_parallel:
                     self.model_weights_signal[0] = self._broadcast_model_weights_signal(
                         src=0, group=self.parallel_config.ep_group
                     )
-            if self.fd_config.load_config.dynamic_load_weight and self.parallel_config.tensor_parallel_size > 1:
+            if self.fd_config.load_config.dynamic_load_weight and tp_size > 1:
                 self.model_weights_signal[0] = self._broadcast_model_weights_signal(
                     src=0, group=self.parallel_config.tp_group
                 )
 
             self.insert_step = False
             req_dicts = None
-            local_rank = self.local_rank % self.parallel_config.tensor_parallel_size
-            self.worker_healthy_live_signal.value[local_rank % self.max_chips_per_node] = int(time.time())
+            self.worker_healthy_live_signal.value[tp_rank % self.max_chips_per_node] = int(time.time())
 
             # The first worker detects whether there are tasks in the task queue
-            if local_rank == 0:
+            if tp_rank == 0:
                 if self.task_queue.num_tasks() > 0:
                     if envs.ENABLE_V1_KVCACHE_SCHEDULER or not (
                         self.fd_config.model_config.enable_mm and self.worker.exist_prefill()
                     ):
-                        if self.nnode > 1 and self.parallel_config.tensor_parallel_size > self.max_chips_per_node:
+                        if self.nnode > 1 and tp_size > self.max_chips_per_node:
                             self.task_queue.read_finish_flag.set(1)
                         else:
                             self.exist_task_signal.value[0] = ExistTaskStatus.EXIST
 
-            if self.parallel_config.tensor_parallel_size > 1:
+            if tp_size > 1:
                 # Synchronize the signal for other workers
                 self._tp_barrier_wait()