pf.cache의 소스 코드

from typing import (
    Iterator,
)

from abc import (
    ABC,
    abstractmethod,
)
from random import (
    Random,
)
from collections import (
    OrderedDict,
)


__all__ = (
    'PfCachePolicy',
    'PfRandomPolicy',
    'PfFifoPolicy',
    'PfClockPolicy',
    'PfCustomPolicy',
)




class PfCachePolicy[Key](ABC):
    """
    buffer에서 page frame을 관리하는 cache policy의 기반 추상 클래스입니다.
    """



    @abstractmethod
    def access(self, key: Key) -> None:  # pragma: no cover
        """
        key에 해당하는 page frame를 가장 최근에 접근한 것으로 표시합니다.

        Args:
            key (Key):
                대상 page frame의 key.
        """
        pass




    @abstractmethod
    def insert(self, key: Key) -> None:  # pragma: no cover
        """
        key에 해당하는 page frame를 cache에 추가합니다.

        Args:
            key (Key):
                대상 page frame의 key.
        """
        pass




    @abstractmethod
    def remove(self, key: Key) -> None:  # pragma: no cover
        """
        key에 해당하는 page frame를 cache에서 제거합니다.

        Args:
            key (Key):
                대상 page frame의 key.
        """
        pass




    @abstractmethod
    def victims(self) -> Iterator[Key]:  # pragma: no cover
        """
        cache에서 evict 후보인 page frame들을 yield 합니다.

        Yields:
            Key:
                evict 후보 page frame들의 key.
        """
        pass






class PfRandomPolicy[Key](PfCachePolicy[Key]):
    """
    evict 후보 page frame을 무작위로 선택하는 cache policy 클래스입니다.

    Attributes:
        _seed (int):
            랜덤 시드.
        _rng (Random):
            랜덤 생성기.
        _set (set[Key]):
            cache에 있는 page frame들의 key를 관리하는 컨테이너.
    """



    def __init__(self, seed: int = 0) -> None:
        """
        cache policy를 초기화합니다.

        Args:
            seed (int):
                랜덤 시드 (기본값: `0`).
        """
        self._seed = seed
        self._rng = Random(seed)
        self._set: set[Key] = set()




    def access(self, key: Key) -> None:
        pass




    def insert(self, key: Key) -> None:
        self._set.add(key)




    def remove(self, key: Key) -> None:
        self._set.discard(key)




    def victims(self) -> Iterator[Key]:
        keys = list(self._set)
        self._rng.shuffle(keys)
        for key in keys:
            yield key






class PfFifoPolicy[Key](PfCachePolicy[Key]):
    """
    evict 후보 page frame을 FIFO 순서로 선택하는 cache policy 클래스입니다.

    Attributes:
        _table (OrderedDict[Key, None]):
            cache에 있는 page frame들의 key를 관리하는 컨테이너.
    """



    def __init__(self) -> None:
        """
        cache policy를 초기화합니다.
        """
        self._table: OrderedDict[Key, None] = OrderedDict()




    def access(self, key: Key) -> None:
        pass




    def insert(self, key: Key) -> None:
        self._table[key] = None




    def remove(self, key: Key) -> None:
        self._table.pop(key)




    def victims(self) -> Iterator[Key]:
        for key in self._table:
            yield key






class PfClockPolicy[Key](PfCachePolicy[Key]):
    """
    evict 후보 page frame을 Clock 알고리즘으로 선택하는 cache policy 클래스입니다.

    Attributes:
        _chance (int):
            Clock 알고리즘에서 잔여 유예 횟수.
        _slots (list[Key | None]):
            cache에 있는 page frame들의 key를 관리하는 컨테이너.
        _counter (list[int]):
            slot에 해당하는 잔여 유예 횟수를 관리하는 컨테이너.
        _find (dict[Key, int]):
            key에 해당하는 slot의 인덱스를 관리하는 컨테이너.
        _frees (list[int]):
            free slot의 인덱스를 관리하는 컨테이너.
        _hand (int):
            Clock 알고리즘에서 현재 가리키고 있는 인덱스.
    """



    def __init__(self, chance: int = 2) -> None:
        """
        cache policy를 초기화합니다.

        Args:
            chance (int):
                Clock 알고리즘에서 잔여 유예 횟수 (기본값: `2`).
        """
        self._chance = chance
        self._slots: list[Key | None] = []
        self._counter: list[int] = []
        self._find: dict[Key, int] = {}
        self._frees: list[int] = []
        self._hand = 0




    def access(self, key: Key) -> None:
        index = self._find[key]
        self._counter[index] = min(self._counter[index] + 1, self._chance - 1)




    def insert(self, key: Key) -> None:
        if self._frees:
            index = self._frees.pop()
            self._slots[index] = key
            self._counter[index] = 0
        else:
            index = len(self._slots)
            self._slots.append(key)
            self._counter.append(0)
        self._find[key] = index




    def remove(self, key: Key) -> None:
        index = self._find.pop(key)
        self._slots[index] = None
        self._counter[index] = 0
        self._frees.append(index)




    def victims(self) -> Iterator[Key]:
        size = len(self._slots)
        for _ in range(size * self._chance):
            key = self._slots[self._hand]
            assert key is not None
            if not self._counter[self._hand]:
                yield key
            else:
                self._counter[self._hand] -= 1
            self._hand = (self._hand + 1) % size






class PfCustomPolicy[Key](PfCachePolicy[Key]):
    """
    evict 후보 page frame을 사용자가 구현한 방식으로 선택하는 cache policy 클래스입니다.

    이 클래스의 attribute는 자유롭게 정의할 수 있습니다.
    """



    def __init__(self) -> None:
        """
        cache policy를 초기화합니다.

        이 클래스의 초기화 argument는 자유롭게 정의할 수 있습니다.
        다만, 기본값을 정의해야 테스트에서 사용할 수 있습니다.
        """
        raise NotImplementedError




    def access(self, key: Key) -> None:
        raise NotImplementedError




    def insert(self, key: Key) -> None:
        raise NotImplementedError




    def remove(self, key: Key) -> None:
        raise NotImplementedError




    def victims(self) -> Iterator[Key]:
        raise NotImplementedError