from typing import (
ClassVar,
TYPE_CHECKING,
)
if TYPE_CHECKING:
from .file import PfFile
from abc import ABC
from bluebase.core import (
LayoutField,
LayoutMixin,
assignment,
)
from .error import (
PfUnpinnedPageAccessError,
PfPageDataOutOfRangeError,
)
from .type import (
PfPageId,
PfPageData,
)
__all__ = (
'PfPageView',
'PfPageHeader',
'PfPage',
)
[docs]
class PfPageView(ABC):
"""
An abstract base class that provides a memory view of page data.
Attributes:
page: the target page
offset: the offset of the view
limit: the size of the view
view: a memory view of the page data
buffer: a buffer for the view, applied to the view by `commit`
"""
page: 'PfPage'
offset: int
limit: int
view: memoryview
buffer: bytearray
def __init__(self, page: 'PfPage', *, offset: int, limit: int) -> None:
assert offset >= 0 and offset + limit <= page.size
assert page.is_pinned()
page.touch()
self.page = page
self.offset = offset
self.limit = limit
self.view = memoryview(page.data)[offset: offset + limit]
self.buffer = bytearray(self.view)
[docs]
def commit(self) -> None:
"""
Commit the buffer to the view.
Raises:
PfUnpinnedPageAccessError: if the page is not pinned
"""
self.page.validate_pinned()
self.view[:] = PfPageData(self.buffer)
self.page.touch()
self.page.mark_dirty()
[docs]
class PfPage:
"""
Attributes:
pid: the ID of the page
file: the file that contains the page
data: the page data
dirty: whether the page is dirty
pin_count: the pin count of the page
"""
pid: PfPageId
file: 'PfFile'
data: PfPageData
dirty: bool
pin_count: int
def __repr__(self) -> str: # pragma: no cover
return f"PfPage({self.pid})"
def __init__(self, pid: PfPageId, file: 'PfFile', data: PfPageData) -> None:
self.pid = pid
self.data = data
self.file = file
self.dirty = False
self.pin_count = 0
@property
def size(self) -> int:
"""
The size of the page.
"""
return len(self.data)
@property
def header_offset(self) -> int:
"""
The offset of the page header.
"""
return 0
@property
def header_size(self) -> int:
"""
The size of the page header.
"""
return PfPageHeader.size
@property
def payload_offset(self) -> int:
"""
The offset of the page payload.
"""
return self.header_size
@property
def payload_size(self) -> int:
"""
The size of the page payload.
"""
return self.size - self.header_size
@property
def free(self) -> bool:
"""
Whether the page is free.
"""
header = self.get_header()
return bool(header.get('free'))
@free.setter
def free(self, free: bool) -> None:
header = self.get_header()
header.set('free', bool(free))
header.commit()
@property
def next_free_pid(self) -> PfPageId | None:
"""
The ID of the next free page.
"""
header = self.get_header()
pid_ = header.get('next_free_pid')
if pid_ == int(PfPageHeader.NO_NEXT_FREE_PAGE):
return None
return PfPageId(pid_)
@next_free_pid.setter
def next_free_pid(self, pid: PfPageId | None) -> None:
if pid is None:
pid = PfPageHeader.NO_NEXT_FREE_PAGE
header = self.get_header()
header.set('next_free_pid', int(pid))
header.commit()
[docs]
def validate_pinned(self) -> None:
"""
Validate whether the page is pinned.
Raises:
PfUnpinnedPageAccessError: if the page is not pinned
"""
if not self.is_pinned():
raise PfUnpinnedPageAccessError(page=self)
[docs]
def is_dirty(self) -> bool:
"""
Return whether the page is dirty.
"""
return self.dirty
[docs]
def mark_dirty(self) -> None:
"""
Mark the page as dirty.
"""
self.dirty = True
[docs]
def clear_dirty(self) -> None:
"""
Clear the dirty state of the page.
"""
self.dirty = False
[docs]
def is_pinned(self) -> bool:
"""
Return whether the page is pinned.
"""
return self.pin_count > 0
[docs]
def pin(self) -> None:
"""
Pin the page.
"""
self.pin_count += 1
[docs]
def touch(self) -> None:
"""
Notify to the page cache that the page has been accessed.
"""
self.file.cache.touch(self)
[docs]
def recycle(self) -> None:
"""
Reset the state and data of a free page for reuse.
"""
self.free = False
self.next_free_pid = None
self.set_data(self.payload_offset, PfPageData(self.payload_size))
[docs]
@assignment
def unpin(self) -> None:
"""
Unpin the page.
Raises:
PfUnpinnedPageAccessError: if the page is not pinned
Hint:
Always validate the page is pinned before unpinning.
Then, decrease the pin count.
"""
raise NotImplementedError
[docs]
@assignment
def get_data(self, offset: int, limit: int) -> PfPageData:
"""
Return page data.
Raises:
PfUnpinnedPageAccessError: if the page is not pinned
PfPageDataOutOfRangeError: if the data range is invalid
Hint:
Always validate the page is pinned before getting data.
Also, validate the data range.
Then, touch the page to notify the page cache.
Finally, return the sliced data.
"""
raise NotImplementedError
[docs]
@assignment
def set_data(self, offset: int, data: PfPageData) -> None:
"""
Set page data.
Raises:
PfUnpinnedPageAccessError: if the page is not pinned
PfPageDataOutOfRangeError: if the data range is invalid
Hint:
Always validate the page is pinned before setting data.
Also, validate the data range.
Then, set the sliced data and mark the page as dirty.
Finally, touch the page to notify the page cache.
"""
raise NotImplementedError