from typing import (
NamedTuple,
)
from .error import (
PfInvalidPageSizeError,
PfInvalidBufferCapacityError,
PfInvalidPageDataSizeError,
PfNoEvictableFrameError,
PfFlushingPinnedPageError,
)
from .cache import (
PfCachePolicy,
PfRandomPolicy,
)
from .page import (
PfPageId,
PfPageHeader,
PfPageData,
PfPage,
)
from .file import (
PfFileId,
PfFile,
)
__all__ = (
'PfFrameKey',
'PfBufferManager',
)
[문서]
class PfFrameKey(NamedTuple):
"""
buffer에 있는 page frame의 key를 나타내는 클래스입니다.
Attributes:
fid (PfFileId):
file의 ID.
pid (PfPageId):
page의 ID.
"""
fid: PfFileId
"""
file의 ID.
"""
pid: PfPageId
"""
page의 ID.
"""
[문서]
class PfBufferManager:
"""
buffer manager를 나타내는 클래스입니다.
Attributes:
_page_size (int):
page의 크기.
_capacity (int):
buffer의 크기.
_cache_policy (PfCachePolicy):
사용할 cache policy 정책.
_cache_hit (int):
cache hit 횟수.
_cache_miss (int):
cache miss 횟수.
_frames (dict[PfFrameKey, PfPage]):
buffer에 있는 page frame들.
"""
[문서]
def __init__(self,
page_size: int,
capacity: int,
cache_policy: PfCachePolicy | None = None,
) -> None:
"""
buffer manager를 초기화합니다.
Args:
page_size (int):
page의 크기.
capacity (int):
buffer의 크기.
cache_policy (PfCachePolicy | None):
사용할 cache policy.
`None` 이면 `PfRandomPolicy` 를 사용합니다.
Raises:
PfInvalidPageSizeError:
`page_size` 가 유효하지 않은 경우.
PfInvalidBufferCapacityError:
`capacity` 가 유효하지 않은 경우.
"""
# 파라미터 확인
if page_size < PfPageHeader.SIZE + 1:
raise PfInvalidPageSizeError(
f"`page_size`는 {PfPageHeader.SIZE + 1} 이상만 가능합니다 "
f"(현재 입력: {page_size}).",
)
if capacity < 1:
raise PfInvalidBufferCapacityError(
"`capacity`는 1 이상만 가능합니다 "
f"(현재 입력: {capacity}).",
)
# 속성 초기화
self._page_size = page_size
self._capacity = capacity
self._cache_policy = cache_policy or PfRandomPolicy()
self._cache_hit = 0
self._cache_miss = 0
self._frames: dict[PfFrameKey, PfPage] = {}
def _debug_clear_buffer(self) -> None: # pragma: no cover
victims = []
for key, page in self._frames.items():
if page.is_pinned():
continue
victims.append(key)
for key in victims:
self._frames.pop(key)
self._cache_policy.remove(key)
def _debug_print_buffer(self) -> None: # pragma: no cover
print(
"(DEBUG) "
f"buffer: {len(self._frames)}/{self._capacity} "
f"(hit: {self._cache_hit}/{self._cache_hit + self._cache_miss})"
)
for index, page in enumerate(self._frames.values(), start=1):
with page.pinned():
print(f" [frame {index}]")
print(f" file.fid: {page.file.fid}")
print(f" file.path: {page.file.path}")
print(f" page.pid: {page.pid}")
print(f" page.dirty: {page._dirty}")
print(f" page.pin_count: {page._pin_count}")
print(f" page.free: {page.is_free()}")
def _debug_calc_hit_ratio(self) -> float: # pragma: no cover
total = self._cache_hit + self._cache_miss
try:
return self._cache_hit / total
except ZeroDivisionError:
return 0.0
[문서]
def read_page(self, file: PfFile, pid: PfPageId) -> PfPage:
"""
(internal)
file에서 page를 read합니다.
Args:
file (PfFile):
대상 file.
pid (PfPageId):
대상 page의 ID.
Returns:
PfPage:
read한 page.
Raises:
PfInvalidPageDataSizeError:
page data의 크기가 유효하지 않은 경우.
"""
raise NotImplementedError
[문서]
def write_page(self,
file: PfFile,
page: PfPage,
fsync: bool = False,
) -> bool:
"""
(internal)
page를 file에 write하고, dirty 상태를 해제합니다.
Args:
file (PfFile):
대상 file.
page (PfPage):
대상 page.
fsync (bool):
fsync 여부. `True` 라면 file stream을 flush합니다.
Returns:
bool:
write가 발생했는지 여부.
Raises:
PfInvalidPageDataSizeError:
page data의 크기가 유효하지 않은 경우.
"""
raise NotImplementedError
[문서]
def fetch_page(self, file: PfFile, pid: PfPageId) -> PfPage:
"""
file에서 page를 가져옵니다.
page가 buffer에 있으면, 그것을 반환합니다.
이때 `_cache_hit` 를 하나 증가시킵니다.
page가 buffer에 없는 경우,
만약 buffer가 가득 차있으면, 하나를 evict하면서 file에 write해서 빈 공간을 만듭니다.
buffer에 빈 공간이 있으면, file에서 page를 read하고, buffer에 추가한 뒤, 반환합니다.
buffer에 추가할때는 frame key를 cache policy에도 insert합니다.
이때 `_cache_miss` 를 하나 증가시킵니다.
Args:
file (PfFile):
대상 file.
pid (PfPageId):
대상 page의 ID.
Returns:
PfPage:
가져온 page.
Raises:
PfNoEvictableFrameError:
buffer에 있는 모든 page가 pin 되어있어서 evict할 수 없는 경우.
"""
raise NotImplementedError
[문서]
def allocate_page(self, file: PfFile, pid: PfPageId) -> PfPage:
"""
file에 새로운 page를 allocate하고, buffer에 추가합니다.
buffer가 가득 차있으면, 하나를 evict하면서 file에 write해서 빈 공간을 만듭니다.
buffer에 빈 공간이 있으면, 새로운 page를 생성하고, page header와 data를 초기화한 뒤, buffer에 추가합니다.
buffer에 추가할때는 frame key를 cache policy에도 insert합니다.
Args:
file (PfFile):
대상 file.
pid (PfPageId):
새로운 page의 ID.
Returns:
PfPage:
새로 allocate된 page.
Raises:
PfNoEvictableFrameError:
buffer에 있는 모든 page가 pin 되어있어서 evict할 수 없는 경우.
"""
raise NotImplementedError
[문서]
def touch_page(self, file: PfFile, pid: PfPageId) -> None:
"""
buffer에서 page를 가장 최근에 접근한 것으로 표시합니다.
cache policy에서 access를 해줍니다.
Args:
file (PfFile):
대상 file.
pid (PfPageId):
대상 page의 ID.
"""
raise NotImplementedError
[문서]
def persist_page(self, file: PfFile, pid: PfPageId) -> bool:
"""
page를 file에 즉시 write합니다.
즉, `fsync` 옵션을 `True` 로 설정한 `write_page` 메서드를 사용합니다.
buffer에 있는 page만 대상으로 한다고 가정합니다.
Args:
file (PfFile):
대상 file.
pid (PfPageId):
대상 page의 ID.
Returns:
bool:
write가 발생했는지 여부.
"""
raise NotImplementedError
[문서]
def flush(self, file: PfFile) -> int:
"""
file을 flush 합니다.
buffer에 있는, 해당하는 file의 모든 page를 file에 write하고,
buffer에서 제거합니다.
cache policy에서도 remove 해줍니다.
Returns:
int:
write가 발생한 page의 수.
Raises:
PfFlushingPinnedPageError:
page가 pin 되어있어서 flush할 수 없는 경우.
"""
raise NotImplementedError