from pathlib import (
Path,
)
from .error import (
PfInvalidPageSizeError,
PfInvalidBufferCapacityError,
PfFileIsClosedError,
PfFileIsOpenedError,
)
from .cache import (
PfCachePolicy,
)
from .buffer import (
PfBufferManager,
)
from .page import (
PfPageHeader,
)
from .file import (
PfFileId,
PfFileHeader,
PfFile,
)
__all__ = (
'PfManager',
)
[문서]
class PfManager:
"""
PF component의 메인 manager 클래스입니다.
manager당 buffer manager가 하나씩 있고,
이 buffer manager는 여러개의 file들을 처리합니다.
Attributes:
page_size (int):
page의 크기.
_buffer_manager (PfBufferManager):
buffer manager.
_next_fid_number (int):
다음으로 부여할 file의 ID.
_opened_file_table (dict[Path, PfFile]):
열린 file들의 테이블.
"""
[문서]
def __init__(self,
page_size: int = 4096,
buffer_capacity: int = 40,
cache_policy: PfCachePolicy | None = None,
) -> None:
"""
manager를 초기화합니다.
Args:
page_size (int):
page의 크기 (기본값: `4096`).
buffer_capacity (int):
buffer의 최대 크기 (기본값: `40`).
cache_policy (PfCachePolicy | None):
사용할 cache policy (기본값: `None`).
`None` 이면 `PfRandomPolicy` 를 사용합니다.
Raises:
PfInvalidPageSizeError:
`page_size` 가 유효하지 않은 경우.
PfInvalidBufferCapacityError:
`buffer_capacity` 가 유효하지 않은 경우.
"""
# 파라미터 확인
if page_size < PfPageHeader.SIZE + 1:
raise PfInvalidPageSizeError(
f"`page_size`는 {PfPageHeader.SIZE + 1} 이상만 가능합니다 "
f"(현재 입력: {page_size}).",
)
if buffer_capacity < 1:
raise PfInvalidBufferCapacityError(
"`buffer_capacity`는 1 이상만 가능합니다 "
f"(현재 입력: {buffer_capacity}).",
)
# 속성 초기화
self.page_size = page_size
self._buffer_manager = PfBufferManager(
page_size=page_size,
capacity=buffer_capacity,
cache_policy=cache_policy,
)
self._next_fid_number = 0
self._opened_file_table: dict[Path, PfFile] = {}
def _normalize_path(self, path: Path) -> Path:
return path.resolve()
def _validate_file_opened(self, path: Path) -> None:
if path not in self._opened_file_table:
raise PfFileIsClosedError(
"file이 닫혀있습니다 "
f"(파일 경로: {path}).",
)
def _validate_file_closed(self, path: Path) -> None:
if path in self._opened_file_table:
raise PfFileIsOpenedError(
"file이 열려있습니다 "
f"(파일 경로: {path}).",
)
[문서]
def create_file(self, path: Path) -> PfFile:
"""
file을 생성하고, file header를 초기화합니다.
file header는 0번째 페이지에 저장됩니다.
Args:
path (Path):
대상 file의 경로.
Returns:
PfFile:
생성된 file.
Raises:
FileExistsError:
`path` 에 file이 이미 있는 경우.
"""
# file 생성
path.touch(exist_ok=False)
# file 열기
file = self.open_file(path)
# header 초기화
header_pid = PfFileHeader.PID
header_page = self._buffer_manager.allocate_page(file, header_pid)
with header_page.pinned():
header = PfFileHeader(
page=header_page,
offset=PfPageHeader.SIZE,
limit=PfFileHeader.SIZE,
)
header.page_count = 1
header.first_free_pid = PfFileHeader.NO_FREE_PAGE
header_page.persist()
return file
[문서]
def destroy_file(self, path: Path) -> None:
"""
file을 삭제합니다.
Args:
path (Path):
대상 file의 경로.
Raises:
PfFileIsOpenedError:
`path` 에 file이 열려있는 경우.
FileNotFoundError:
`path` 에 file이 없는 경우.
"""
# path 확인
path = self._normalize_path(path)
self._validate_file_closed(path)
# file 삭제
path.unlink()
[문서]
def open_file(self, path: Path) -> PfFile:
"""
file을 열고, `fid` 를 부여합니다.
`_opened_file_table` 에서 해당 `path` 로 열린 file을 추적합니다.
Args:
path (Path):
대상 file의 경로.
Returns:
PfFile:
열린 file.
Raises:
PfFileIsOpenedError:
`path` 에 file이 이미 열려있는 경우.
FileNotFoundError:
`path` 에 file이 없는 경우.
"""
# path 확인
path = self._normalize_path(path)
self._validate_file_closed(path)
# fid 부여
fid = PfFileId(self._next_fid_number)
self._next_fid_number += 1
# file 열기
stream = path.open('rb+')
file = PfFile(
fid=fid,
path=path,
stream=stream,
buffer_manager=self._buffer_manager,
)
self._opened_file_table[path] = file
return file
[문서]
def close_file(self, path: Path) -> None:
"""
file을 flush 하고, stream을 닫습니다.
`_opened_file_table` 에서 해당 `path` 로 열린 file을 제거합니다.
Args:
path (Path):
대상 file의 경로.
Raises:
PfFileIsClosedError:
`path` 에 file이 이미 닫혀있는 경우.
"""
# path 확인
path = self._normalize_path(path)
self._validate_file_opened(path)
# file 정리
file = self._opened_file_table[path]
file.flush()
file.stream.close()
# file 해제
del self._opened_file_table[path]