Merge remote-tracking branch 'upstream/develop' into develop

Author: kevincheng2

fastdeploy/engine/engine.py

@@ -734,10 +734,6 @@ def insert_tasks(self, tasks, current_id=-1, allocated=False):
         """
         Insert tasks to engine.
         """
-        for task in tasks:
-            start_span_request("DEQUEUE", task, trace.SpanKind.CONSUMER)
-            if task.sampling_params.bad_words is not None:
-                task.sampling_params.update_from_tokenizer(self.data_processor.tokenizer)
         # TODO 返回至 scheduler
         if allocated:
             current_tasks = []
@@ -764,6 +760,11 @@ def insert_tasks(self, tasks, current_id=-1, allocated=False):
             self.engine_worker_queue.put_tasks((current_tasks, self.resource_manager.real_bsz))
             return True
 
+        for task in tasks:
+            start_span_request("DEQUEUE", task, trace.SpanKind.CONSUMER)
+            if task.sampling_params.bad_words is not None:
+                task.sampling_params.update_from_tokenizer(self.data_processor.tokenizer)
+
         self.resource_manager.check_and_free_block_tables()
 
         if not isinstance(tasks, list):

fastdeploy/entrypoints/openai/api_server.py

@@ -176,10 +176,10 @@ async def connection_manager():
         await asyncio.wait_for(connection_semaphore.acquire(), timeout=0.001)
         yield
     except asyncio.TimeoutError:
-        api_server_logger.info(f"Reach max request release: {connection_semaphore.status()}")
-        if connection_semaphore.locked():
-            connection_semaphore.release()
-        raise HTTPException(status_code=429, detail="Too many requests")
+        api_server_logger.info(f"Reach max request concurrency, semaphore status: {connection_semaphore.status()}")
+        raise HTTPException(
+            status_code=429, detail=f"Too many requests,current max concurrency is {args.max_concurrency}"
+        )
 
 
 # TODO 传递真实引擎值 通过pid 获取状态
@@ -266,9 +266,11 @@ async def create_chat_completion(request: ChatCompletionRequest):
             inject_to_metadata(request)
             generator = await app.state.chat_handler.create_chat_completion(request)
             if isinstance(generator, ErrorResponse):
+                api_server_logger.debug(f"release: {connection_semaphore.status()}")
                 connection_semaphore.release()
                 return JSONResponse(content={"detail": generator.model_dump()}, status_code=generator.code)
             elif isinstance(generator, ChatCompletionResponse):
+                api_server_logger.debug(f"release: {connection_semaphore.status()}")
                 connection_semaphore.release()
                 return JSONResponse(content=generator.model_dump())
             else:

fastdeploy/entrypoints/openai/serving_chat.py

@@ -78,34 +78,48 @@ async def create_chat_completion(self, request: ChatCompletionRequest):
             err_msg = f"Only master node can accept completion request, please send request to master node: {self.pod_ips[0]}"
             api_server_logger.error(err_msg)
             return ErrorResponse(message=err_msg, code=400)
-
-        if request.user is not None:
-            request_id = f"chatcmpl-{request.user}-{uuid.uuid4()}"
-        else:
-            request_id = f"chatcmpl-{uuid.uuid4()}"
-        api_server_logger.info(f"create chat completion request: {request_id}")
-        text_after_process = None
         try:
-            current_req_dict = request.to_dict_for_infer(request_id)
-            if "chat_template" not in current_req_dict:
-                current_req_dict["chat_template"] = self.chat_template
-            current_req_dict["arrival_time"] = time.time()
-            prompt_token_ids = self.engine_client.format_and_add_data(current_req_dict)
-            text_after_process = current_req_dict.get("text_after_process")
-            if isinstance(prompt_token_ids, np.ndarray):
-                prompt_token_ids = prompt_token_ids.tolist()
-        except Exception as e:
-            error_msg = f"request[{request_id}] send to infer error: {str(e)}, {str(traceback.format_exc())}"
-            api_server_logger.error(error_msg)
-            return ErrorResponse(code=400, message=error_msg)
-
-        del current_req_dict
-        try:
-            api_server_logger.debug(f"{self.engine_client.semaphore.status()}")
             if self.max_waiting_time < 0:
                 await self.engine_client.semaphore.acquire()
             else:
                 await asyncio.wait_for(self.engine_client.semaphore.acquire(), timeout=self.max_waiting_time)
+            api_server_logger.info(f"current {self.engine_client.semaphore.status()}")
+
+            if request.user is not None:
+                request_id = f"chatcmpl-{request.user}-{uuid.uuid4()}"
+            else:
+                request_id = f"chatcmpl-{uuid.uuid4()}"
+            api_server_logger.info(f"create chat completion request: {request_id}")
+            text_after_process = None
+            try:
+                current_req_dict = request.to_dict_for_infer(request_id)
+                if "chat_template" not in current_req_dict:
+                    current_req_dict["chat_template"] = self.chat_template
+                current_req_dict["arrival_time"] = time.time()
+                prompt_token_ids = self.engine_client.format_and_add_data(current_req_dict)
+                text_after_process = current_req_dict.get("text_after_process")
+                if isinstance(prompt_token_ids, np.ndarray):
+                    prompt_token_ids = prompt_token_ids.tolist()
+            except Exception as e:
+                error_msg = f"request[{request_id}] generator error: {str(e)}, {str(traceback.format_exc())}"
+                api_server_logger.error(error_msg)
+                return ErrorResponse(code=400, message=error_msg)
+
+            del current_req_dict
+
+            if request.stream:
+                return self.chat_completion_stream_generator(
+                    request, request_id, request.model, prompt_token_ids, text_after_process
+                )
+            else:
+                try:
+                    return await self.chat_completion_full_generator(
+                        request, request_id, request.model, prompt_token_ids, text_after_process
+                    )
+                except Exception as e:
+                    error_msg = f"request[{request_id}]full generator error: {str(e)}, {str(traceback.format_exc())}"
+                    api_server_logger.error(error_msg)
+                    return ErrorResponse(code=408, message=error_msg)
         except Exception as e:
             error_msg = (
                 f"request[{request_id}] waiting error: {str(e)}, {str(traceback.format_exc())}, "
@@ -114,20 +128,6 @@ async def create_chat_completion(self, request: ChatCompletionRequest):
             api_server_logger.error(error_msg)
             return ErrorResponse(code=408, message=error_msg)
 
-        if request.stream:
-            return self.chat_completion_stream_generator(
-                request, request_id, request.model, prompt_token_ids, text_after_process
-            )
-        else:
-            try:
-                return await self.chat_completion_full_generator(
-                    request, request_id, request.model, prompt_token_ids, text_after_process
-                )
-            except Exception as e:
-                error_msg = f"request[{request_id}] generator error: {str(e)}, {str(traceback.format_exc())}"
-                api_server_logger.error(error_msg)
-                return ErrorResponse(code=400, message=error_msg)
-
     def _create_streaming_error_response(self, message: str) -> str:
         api_server_logger.error(message)
         error_response = ErrorResponse(
@@ -264,6 +264,7 @@ async def chat_completion_stream_generator(
                         logprobs_res = self._create_chat_logprobs(
                             output_top_logprobs, request.logprobs, request.top_logprobs
                         )
+
                     if self.engine_client.data_processor.tool_parser_obj and not res["finished"]:
                         tool_delta_message = output["tool_delta_message"]
                         if tool_delta_message is None:
@@ -287,7 +288,6 @@ async def chat_completion_stream_generator(
                         logprobs=logprobs_res,
                         arrival_time=arrival_time,
                     )
-
                     if res["finished"]:
                         num_choices -= 1
                         work_process_metrics.e2e_request_latency.observe(
@@ -319,7 +319,6 @@ async def chat_completion_stream_generator(
                     if len(choices) == max_streaming_response_tokens or res["finished"]:
                         chunk.choices = choices
                         yield f"data: {chunk.model_dump_json(exclude_unset=True)}\n\n"
-                        # 打印尾包
                         if res["finished"]:
                             api_server_logger.info(f"Chat Streaming response last send: {chunk.model_dump_json()}")
                         choices = []
@@ -429,8 +428,9 @@ async def chat_completion_full_generator(
                 if task_is_finished:
                     break
         finally:
-            self.engine_client.semaphore.release()
             dealer.close()
+            self.engine_client.semaphore.release()
+            api_server_logger.info(f"release {self.engine_client.semaphore.status()}")
 
         choices = []
         output = final_res["outputs"]

fastdeploy/entrypoints/openai/serving_completion.py

@@ -104,6 +104,19 @@ async def create_completion(self, request: CompletionRequest):
         api_server_logger.info(f"start inference for request {num_choices}")
         prompt_batched_token_ids = []
         text_after_process_list = []
+        try:
+            if self.max_waiting_time < 0:
+                await self.engine_client.semaphore.acquire()
+            else:
+                await asyncio.wait_for(self.engine_client.semaphore.acquire(), timeout=self.max_waiting_time)
+        except Exception as e:
+            error_msg = (
+                f"OpenAIServingCompletion waiting error: {e}, {str(traceback.format_exc())}, "
+                f"max waiting time: {self.max_waiting_time}"
+            )
+            api_server_logger.error(error_msg)
+            return ErrorResponse(code=408, message=error_msg)
+
         try:
             for idx, prompt in enumerate(request_prompts):
                 request_id_idx = f"{request_id}-{idx}"
@@ -122,19 +135,6 @@ async def create_completion(self, request: CompletionRequest):
 
                 del current_req_dict
 
-            try:
-                if self.max_waiting_time < 0:
-                    await self.engine_client.semaphore.acquire()
-                else:
-                    await asyncio.wait_for(self.engine_client.semaphore.acquire(), timeout=self.max_waiting_time)
-            except Exception as e:
-                error_msg = (
-                    f"OpenAIServingCompletion waiting error: {e}, {str(traceback.format_exc())}, "
-                    f"max waiting time: {self.max_waiting_time}"
-                )
-                api_server_logger.error(error_msg)
-                return ErrorResponse(code=408, message=error_msg)
-
             if request.stream:
                 return self.completion_stream_generator(
                     request=request,

fastdeploy/inter_communicator/zmq_client.py

@@ -32,7 +32,7 @@ class ZmqClient:
     """
 
     def __init__(self, name, mode):
-        self.context = zmq.Context()
+        self.context = zmq.Context(4)
         self.socket = self.context.socket(mode)
         self.file_name = f"/dev/shm/{name}.socket"
         self.router_path = f"/dev/shm/router_{name}.ipc"
@@ -68,6 +68,7 @@ def create_router(self):
         """
         self.router = self.context.socket(zmq.ROUTER)
         self.router.setsockopt(zmq.SNDHWM, self.ZMQ_SNDHWM)
+        self.router.setsockopt(zmq.ROUTER_MANDATORY, 1)
         self.router.setsockopt(zmq.SNDTIMEO, -1)
         self.router.bind(f"ipc://{self.router_path}")
 
@@ -126,6 +127,11 @@ def send_multipart(self, req_id, data):
                 else:
                     break
 
+        if self.req_dict[req_id] == -1:
+            if data[-1].finished:
+                with self.mutex:
+                    self.req_dict.pop(req_id, None)
+            return
         try:
             start_send = time.time()
             if self.aggregate_send:
@@ -134,7 +140,9 @@ def send_multipart(self, req_id, data):
                 result = msgpack.packb([response.to_dict() for response in data])
             self.router.send_multipart([self.req_dict[req_id], b"", result])
             llm_logger.debug(f"send_multipart result: {req_id} len {len(data)} elapse: {time.time()-start_send}")
-
+        except zmq.ZMQError as e:
+            llm_logger.error(f"[{req_id}] zmq error: {e}")
+            self.req_dict[req_id] = -1
         except Exception as e:
             llm_logger.error(f"Send result to zmq client failed: {e}, {str(traceback.format_exc())}")
 

fastdeploy/model_executor/layers/sample/early_stopper.py

@@ -67,16 +67,17 @@ def process(self, probs: paddle.Tensor, next_tokens: paddle.Tensor, stop_flags:
     def process_normal(self, probs: paddle.Tensor, next_tokens: paddle.Tensor, stop_flags: paddle.Tensor):
         # Get the probability score corresponding to next_tokens in this step
         next_scores = paddle.index_sample(probs, next_tokens)
+        real_bsz = probs.shape[0]
 
         # Sliding window: Move left one grid and insert new score
-        self.trunc_scores[:, :-1] = self.trunc_scores[:, 1:]
-        self.trunc_scores[:, -1:] = next_scores
+        self.trunc_scores[:real_bsz, :-1] = self.trunc_scores[:real_bsz, 1:]
+        self.trunc_scores[:real_bsz, -1:] = next_scores
 
         # Determine which samples need to be terminated: all trunc_scores are greater than threshold
         need_trunc_all = paddle.all(self.trunc_scores > self.threshold, axis=-1).unsqueeze(-1)
 
         # Add the stop flags
-        stop_flags[need_trunc_all] = True
+        stop_flags[need_trunc_all[:real_bsz]] = True
 
         # Reset trunc_scores of truncated samples to 0 to avoid false triggering in the next step
         reset_mask = need_trunc_all.tile([1, self.window_size])

scripts/coverage_run.sh

@@ -26,7 +26,6 @@ done
 failed_tests_file="failed_tests.log"
 > "$failed_tests_file"
 disabled_tests=(
-  layers/test_sampler.py
   layers/test_append_attention.py
   layers/test_attention.py
   operators/test_rejection_top_p_sampling.py
@@ -36,7 +35,6 @@ disabled_tests=(
   operators/test_stop_generation.py
   operators/test_air_topp_sampling.py
   operators/test_fused_moe.py
-  layers/test_repetition_early_stopper.py
   operators/test_stop_generation_multi_ends.py
   graph_optimization/test_cuda_graph.py
 )

test/layers/test_repetition_early_stopper.py

@@ -170,7 +170,69 @@ def test_consistency():
         actual = triggered_step_triton[i]
         assert expected == actual, f"Sample {i} triggered at different steps: {expected} vs {actual}"
 
-    print("Triton vs Normal: All tokens, states, and trigger timings match.")
+    print("[consistency]Triton vs Normal: All tokens, states, and trigger timings match.")
+
+
+def test_consistency_with_real_batch_size():
+    batch_size = 20
+    real_batch_size = 15
+    vocab_size = 103424
+    window_size = 3000
+    threshold = 0.9
+    eos_token_id = vocab_size
+    max_steps = 10
+
+    fixed_token_id = np.random.randint(0, vocab_size)
+    early_stop_batch_id = np.random.randint(0, real_batch_size)
+
+    trigger_step_flags = [[i, np.random.randint(0, max_steps + 1)] for i in range(batch_size)]
+    trigger_step_flags = dict(trigger_step_flags)
+    cfg = EarlyStopConfig({"enable_early_stop": True, "window_size": window_size, "threshold": threshold})
+    stopper_normal = RepetitionEarlyStopper()
+    stopper_normal.initialize(batch_size, cfg)
+    stopper_triton = RepetitionEarlyStopper()
+    stopper_triton.initialize(batch_size, cfg)
+
+    next_tokens_normal = paddle.randint(0, vocab_size, shape=[real_batch_size, 1], dtype="int64")
+    next_tokens_triton = next_tokens_normal.clone()
+
+    next_tokens_normal[early_stop_batch_id, 0] = fixed_token_id
+    next_tokens_triton[early_stop_batch_id, 0] = fixed_token_id
+
+    stop_flags_normal = paddle.zeros_like(next_tokens_normal)
+    stop_flags_triton = stop_flags_normal.clone()
+
+    triggered_step_normal = [None] * batch_size
+    triggered_step_triton = [None] * batch_size
+
+    for step in range(max_steps):
+
+        flags = [trigger_step_flags[i] for i in range(real_batch_size)]
+        probs_np = simulate_step_probs(real_batch_size, early_stop_batch_id, fixed_token_id, vocab_size, step, flags)
+        probs = paddle.to_tensor(probs_np)
+
+        stopper_normal.process_normal(probs, next_tokens_normal, stop_flags_normal)
+        stopper_triton.process_triton(probs, next_tokens_triton, stop_flags_triton)
+
+        assert np.allclose(stop_flags_normal.numpy(), stop_flags_triton.numpy()), f"stop flags mismatch at step {step}"
+
+        trunc_scores_diff = paddle.abs(stopper_normal.trunc_scores - stopper_triton.trunc_scores)
+        assert paddle.all(trunc_scores_diff < 1e-5), f"trunc_scores mismatch at step {step}"
+
+        out_normal = stop_flags_normal.numpy()
+        out_triton = stop_flags_triton.numpy()
+        for i in range(real_batch_size):
+            if out_normal[i, 0] == eos_token_id and triggered_step_normal[i] is None:
+                triggered_step_normal[i] = step
+            if out_triton[i, 0] == eos_token_id and triggered_step_triton[i] is None:
+                triggered_step_triton[i] = step
+
+    for i in range(batch_size):
+        expected = triggered_step_normal[i]
+        actual = triggered_step_triton[i]
+        assert expected == actual, f"Sample {i} triggered at different steps: {expected} vs {actual}"
+
+    print("[consistency_with_real_batch_size]Triton vs Normal: All tokens, states, and trigger timings match.")
 
 
 def test_performance():
@@ -232,4 +294,5 @@ def test_performance():
 if __name__ == "__main__":
     test_repetition_early_stopper()
     test_consistency()
+    test_consistency_with_real_batch_size()
     test_performance()

test/layers/test_sampler.py

@@ -57,6 +57,7 @@ def _create_default_sampling_metadata(
         bad_words_token_ids=paddle.full(shape=[batch_size], fill_value=-1, dtype="int64"),
         eos_token_ids=paddle.full(shape=[batch_size], fill_value=-2, dtype="int64"),
         min_p=paddle.randn([batch_size]),
+        seed=paddle.to_tensor([[2025]]),
     )
     return fake_sampling_metadata