Method handle_loop

lightllm/server/httpserver/manager.py:658–736

(self)

        return

    async def handle_loop(self):
        self.recycle_event = asyncio.Event()
        asyncio.create_task(self.recycle_resource_loop())

        # 多节点tp模式下的slave节点，需要开启一个协程task用来接收
        # master 转发过来的请求对象。
        if self.is_multinode_tp_slave:
            asyncio.create_task(self.loop_for_request())

        if self.pd_mode.is_P_or_D():
            from lightllm.server.httpserver.pd_loop import pd_handle_loop

            asyncio.create_task(pd_handle_loop(self))

        while True:
            try:
                await asyncio.wait_for(self.recv_from_detokenization.recv_pyobj(), timeout=0.05)
            except asyncio.TimeoutError:
                pass

            try:
                for group_req_id_ in list(self.req_id_to_out_inf.keys()):
                    req_status = self.req_id_to_out_inf.get(group_req_id_, None)
                    if req_status is None:
                        continue

                    token_list = []
                    for req in req_status.group_req_objs.shm_req_objs:
                        req_id = req.request_id
                        read_token_count = 1
                        if req.out_tokens_queue.is_full():
                            read_token_count = LIGHTLLM_OUT_TOKEN_QUEUE_SIZE

                        for _ in range(read_token_count):
                            if not req.out_tokens_queue.is_empty():

                                text, src_index, special, count_output_tokens = req.out_tokens_queue.peek()
                                req.cumlogprob += float(req.shm_logprobs.arr[src_index])
                                metadata = {
                                    "id": int(req.shm_prompt_ids.arr[src_index]),
                                    "logprob": float(req.shm_logprobs.arr[src_index]),
                                    "cumlogprob": float(req.cumlogprob) / count_output_tokens,
                                    "special": special,
                                    "count_output_tokens": count_output_tokens,
                                    "prompt_cache_len": req.prompt_cache_len,
                                    "mtp_accepted_token_num": req.mtp_accepted_token_num,
                                }
                                if self.args.return_all_prompt_logprobs:
                                    metadata.update(req.get_all_prompt_metadata())
                                if self.args.use_reward_model:
                                    metadata["score"] = float(req.reward_score)

                                req.out_tokens_queue.pop_no_ret()

                                finished_token_index = (
                                    req.stop_str_matched_token_index if req.stop_str_matched else req.finish_token_index
                                )