Circular Queue

Circular Queue란

선입선출 특성을 지니는 Queue의 일종으로, 최대 크기가 고정된 대기열이 필요할 때 사용할 수 있는 자료구조이다.

배열의 처음과 끝을 원처럼 연결한 것처럼 바라보아 요소를 넣을 지점과 꺼낼 지점에 해당하는 커서들을 이동시키며 큐로 사용하도록 구현되어 있다.

아래 구현 코드는 덮어쓰기가 불가능하도록 작성되어 있다. 그러나 일부 상황에서는 이를 허용해야 하는 상황이 생긴다. (소리 출력 버퍼가 원형 큐로 구현되어 있다고 하자. 소비자의 속도가 느릴 경우 공급자는 데이터를 덮어쓴다. 소비자는 손실된 출력 데이터를 무시하고 항상 최신 데이터를 소비한다.)

구현

from abc import ABCMeta, abstractmethod
from typing import Generic, TypeVar, cast

T = TypeVar("T")


class AbstractQueue(Generic[T], metaclass=ABCMeta):
    @abstractmethod
    def enqueue(self, __object: T) -> None:
        ...

    @abstractmethod
    def dequeue(self) -> T:
        ...


class EmptyQueue(RuntimeError):
    ...


class FullQueue(RuntimeError):
    ...


class EmptyType:
    def __repr__(self) -> str:
        return f"Empty"


Empty = EmptyType()


class Queue(AbstractQueue[T]):
    def __init__(self, *args: T, max_size: int):
        self._max_size = max_size
        self._queue: list[T | EmptyType] = [Empty for _ in range(self._max_size)]
        self._start, self._end = 0, 0
        self._size = 0
        for data in args:
            self.enqueue(data)

    def enqueue(self, data: T) -> None:
        if not self._queue[self._end] is Empty:
            raise FullQueue("enqueue from fulled circular queue")
        self._queue[self._end] = data
        self._end = (self._end + 1) % self._max_size
        self._size += 1

    def dequeue(self) -> T:
        result = self._queue[self._start]
        if result is Empty:
            raise EmptyQueue("dequeue from empty circular queue")
        self._queue[self._start] = Empty
        result = cast(T, result)
        self._start = (self._start + 1) % self._max_size
        self._size -= 1
        return result

    def peek(self) -> T:
        result = self._queue[self._start]
        if result is Empty:
            raise EmptyQueue("peek from empty circular queue")
        result = cast(T, result)
        return result

    def __bool__(self) -> bool:  # or "empty"
        return self._start != self._end

    def __len__(self) -> int:  # or "size"
        return self._size


queue: Queue[int] = Queue(max_size=5)

try:
    queue.dequeue()
    raise Exception()
except EmptyQueue:
    ...

try:
    for i in range(5):
        queue.enqueue(i)
    queue.enqueue(6)
    raise Exception()
except FullQueue:
    ...

assert len(queue) == 5  # size
assert queue.dequeue() == 0  # dequeue
assert queue.peek() == 1  # peek

temp: list[int] = []
while queue:  # empty
    temp.append(queue.dequeue())
assert temp == [1, 2, 3, 4]