if __debug__: # pragma: no cover
from bluebase.core import dump
from bluebase.core import (
optional,
assignment,
)
from .error import (
PfInvalidPageDataSizeError,
PfNoEvictableEntryError,
PfPinnedPageFlushError,
)
from .type import (
PfPageId,
PfPageData,
PfEntryId,
)
from .replacer import PfReplacer
from .page import (
PfPageHeader,
PfPage,
)
from .file import PfFile
__all__ = (
'PfPageCache',
)
[docs]
class PfPageCache:
"""
page cache를 관리하는 클래스입니다.
Attributes:
page_size: page의 크기.
capacity: cache의 용량.
replacer: cache가 사용하는 교체 정책.
entries: cache가 관리하는 entry들.
"""
page_size: int
capacity: int
replacer: PfReplacer
entries: dict[PfEntryId, PfPage]
if __debug__:
_debug_hit: int
_debug_miss: int
def __repr__(self) -> str: # pragma: no cover
return f"PfPageCache({len(self.entries)}/{self.capacity})"
def __init__(self,
page_size: int,
capacity: int,
replacer: PfReplacer,
) -> None:
assert page_size > PfPageHeader.size
assert capacity > 0
self.page_size = page_size
self.capacity = capacity
self.replacer = replacer
self.entries = {}
if __debug__:
self._debug_hit = 0
self._debug_miss = 0
[docs]
@optional
def _evict(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
def is_full(self) -> bool:
"""
"""
return len(self.entries) == self.capacity
[docs]
def get_page_offset(self, pid: PfPageId) -> int:
"""
"""
return int(pid) * self.page_size
[docs]
@assignment
def read(self, file: PfFile, pid: PfPageId) -> PfPage:
"""
Raises:
PfInvalidPageDataSizeError:
"""
raise NotImplementedError
[docs]
@assignment
def write(self, page: PfPage, *, fsync: bool = False) -> None:
"""
Raises:
PfInvalidPageDataSizeError:
"""
raise NotImplementedError
[docs]
@assignment
def fetch(self, file: PfFile, pid: PfPageId) -> PfPage:
"""
Raises:
PfNoEvictableEntryError:
"""
raise NotImplementedError
[docs]
@assignment
def acquire(self, file: PfFile, pid: PfPageId) -> PfPage:
"""
Raises:
PfNoEvictableEntryError:
"""
raise NotImplementedError
[docs]
@assignment
def touch(self, page: PfPage) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def persist(self, file: PfFile) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def flush(self, file: PfFile) -> None:
"""
Raises:
PfPinnedPageFlushError:
"""
raise NotImplementedError
if __debug__: # pragma: no cover
def _debug_dump(self) -> None:
print(f"[cache dump] {dump(
entries=len(self.entries),
capacity=self.capacity,
hit=self._debug_hit,
miss=self._debug_miss,
)}")
for i, page in enumerate(self.entries.values(), start=1):
page.pin()
print(f" [entry {i}]")
print(f" file: {page.file}")
print(f" page: {page}")
print(f" free: {page.free}")
print(f" pin: {page.pin_count - 1}")
print(f" dirty: {page.is_dirty()}")
page.unpin()
def _debug_calc_hr(self) -> float:
try:
return self._debug_hit / (self._debug_hit + self._debug_miss)
except ZeroDivisionError:
return 0.0