from abc import (
ABC,
abstractmethod,
)
from random import Random
from collections import OrderedDict
from collections.abc import Iterator
from bluebase.core import (
dump,
assignment,
)
__all__ = (
'PfReplacer',
'PfRandomReplacer',
'PfFifoReplacer',
'PfClockReplacer',
'PfLruReplacer',
)
[docs]
class PfReplacer(ABC):
"""
An abstract base class for replacers that select eviction candidate keys.
"""
[docs]
@abstractmethod
def add(self, key: object) -> None: # pragma: no cover
"""
Add a key.
"""
pass
[docs]
@abstractmethod
def remove(self, key: object) -> None: # pragma: no cover
"""
Remove a key.
"""
pass
[docs]
@abstractmethod
def touch(self, key: object) -> None: # pragma: no cover
"""
Notify the replacer that a key has been accessed.
"""
pass
[docs]
@abstractmethod
def candidates(self) -> Iterator: # pragma: no cover
"""
Yield eviction candidate keys in priority order.
"""
pass
[docs]
class PfRandomReplacer(PfReplacer):
"""
A replacer that selects eviction candidate keys using a random policy.
Attributes:
seed: the random seed
rng: the random number generator
set: the container holding the currently managed keys
"""
seed: int
rng: Random
keys: set
def __repr__(self) -> str: # pragma: no cover
return f"PfRandomReplacer({dump(seed=self.seed)})"
def __init__(self, *, seed: int = 0) -> None:
self.seed = seed
self.rng = Random(seed)
self.keys: set = set()
[docs]
def add(self, key: object) -> None:
self.keys.add(key)
[docs]
def remove(self, key: object) -> None:
self.keys.discard(key)
[docs]
def touch(self, key: object) -> None:
"""
Notify the replacer that a key has been accessed.
Under the random policy, this should do nothing.
"""
pass
[docs]
def candidates(self) -> Iterator:
"""
Yield eviction candidate keys in random order.
"""
keys = list(self.keys)
self.rng.shuffle(keys)
for key in keys:
yield key
[docs]
class PfFifoReplacer(PfReplacer):
"""
A replacer that selects eviction candidate keys using a FIFO policy.
Attributes:
keys: the container holding the currently managed keys
"""
keys: OrderedDict[object, None]
def __repr__(self) -> str: # pragma: no cover
return "PfFifoReplacer()"
def __init__(self) -> None:
self.keys: OrderedDict[object, None] = OrderedDict()
[docs]
def add(self, key: object) -> None:
self.keys[key] = None
[docs]
def remove(self, key: object) -> None:
assert key in self.keys
self.keys.pop(key)
[docs]
def touch(self, key: object) -> None:
"""
Notify the replacer that a key has been accessed.
Under the FIFO policy implemented by `OrderedDict`, this should do nothing.
"""
pass
[docs]
def candidates(self) -> Iterator:
"""
Yield eviction candidate keys in FIFO order.
"""
for key in self.keys:
yield key
[docs]
class PfClockReplacer(PfReplacer):
"""
A replacer that selects eviction candidate keys using the Clock algorithm.
Attributes:
chance: the maximum grace count used by the Clock algorithm
slots: the container holding the currently managed keys
counter: the grace counts corresponding to the slots
find: a mapping from each key to its slot index
frees: the indices of free slots
hand: the slot index currently pointed to by the clock hand
"""
chance: int
slots: list[object | None]
counter: list[int]
find: dict[object, int]
frees: list[int]
hand: int
def __repr__(self) -> str: # pragma: no cover
return f"PfClockReplacer({dump(chance=self.chance)})"
def __init__(self, *, chance: int = 2) -> None:
self.chance = chance
self.slots = []
self.counter = []
self.find = {}
self.frees = []
self.hand = 0
[docs]
def add(self, key: object) -> None:
if len(self.frees) > 0:
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
[docs]
def remove(self, key: object) -> None:
assert key in self.find
index = self.find.pop(key)
self.slots[index] = None
self.counter[index] = 0
self.frees.append(index)
[docs]
def touch(self, key: object) -> None:
"""
Notify the replacer that a key has been accessed.
Under the Clock algorithm, this increases the remaining grace count of the key.
"""
assert key in self.find
index = self.find[key]
self.counter[index] = min(self.counter[index] + 1, self.chance - 1)
[docs]
def candidates(self) -> Iterator:
"""
Yield eviction candidate keys according to the Clock algorithm.
"""
size = len(self.slots)
for _ in range(size * self.chance):
index = self.hand
self.hand = (self.hand + 1) % size
key = self.slots[index]
if key is not None:
if not self.counter[index]:
yield key
else:
self.counter[index] -= 1
[docs]
class PfLruReplacer(PfReplacer):
"""
A replacer that selects eviction candidate keys using an LRU policy.
Attributes:
keys: the container holding the currently managed keys
"""
keys: OrderedDict[object, None]
def __repr__(self) -> str: # pragma: no cover
return "PfLruReplacer()"
def __init__(self) -> None:
self.keys: OrderedDict[object, None] = OrderedDict()
[docs]
@assignment
def add(self, key: object) -> None:
"""
Add a key.
Hint:
When using either `dict` or `OrderedDict`, the insertion order is preserved.
"""
raise NotImplementedError
[docs]
@assignment
def remove(self, key: object) -> None:
"""
Remove a key.
Hint:
It should not raise an exception even if the key is not found.
"""
raise NotImplementedError
[docs]
@assignment
def touch(self, key: object) -> None:
"""
Notify the replacer that a key has been accessed.
Under the LRU policy, this gives the key the lowest eviction priority.
Hint:
This can be implemented easily using `OrderedDict`.
This kind of data structure is abstractly known as a position list.
"""
raise NotImplementedError
[docs]
@assignment
def candidates(self) -> Iterator:
"""
Yield eviction candidate keys in LRU order.
Hint:
If `add` and `touch` are implemented correctly, this should be easy to implement.
"""
raise NotImplementedError