Class CircularQueue

lightllm/server/core/objs/out_token_circlequeue.py:46–107

class CircularQueue(ctypes.Structure):
    _pack_ = 4
    _fields_ = [
        ("items", QueueItem * LIGHTLLM_OUT_TOKEN_QUEUE_SIZE),  # 循环队列的元素
        ("head", ctypes.c_int),  # 指向队列头部
        ("tail", ctypes.c_int),  # 指向队列尾部
    ]

    def __init__(self):
        # 初始化头和尾
        self.head = 0
        self.tail = 0

    def is_empty(self):
        return self.head == self.tail

    def is_full(self):
        return (self.tail + 1) % LIGHTLLM_OUT_TOKEN_QUEUE_SIZE == self.head

    def push(self, token_str: str, src_index: int, special: bool, count_output_tokens: int):
        if self.is_full():
            raise Exception("Queue is full")

        # 添加元素
        item: QueueItem = self.items[self.tail]
        item.set(token_str, src_index, special, count_output_tokens)

        # 更新尾部
        self.tail = (self.tail + 1) % LIGHTLLM_OUT_TOKEN_QUEUE_SIZE

    def pop(self) -> Tuple[str, int, bool, int]:
        if self.is_empty():
            raise Exception("Queue is empty")

        # 移除元素
        item: QueueItem = self.items[self.head]
        result = item.get()

        # 更新头部
        self.head = (self.head + 1) % LIGHTLLM_OUT_TOKEN_QUEUE_SIZE
        return result

    def peek(self) -> Tuple[str, int, bool, int]:
        if self.is_empty():
            raise Exception("Queue is empty")

        item: QueueItem = self.items[self.head]
        result = item.get()
        return result

    def pop_no_ret(self) -> None:
        """
        移除一个对象，但是不返回， 与peek配合使用
        """
        if self.is_empty():
            raise Exception("Queue is empty")
        self.head = (self.head + 1) % LIGHTLLM_OUT_TOKEN_QUEUE_SIZE
        return