(
self,
*,
path: Path,
tokens: Tuple[int, ...],
has_prompt_logits: bool,
last_accessed: Optional[float] = None,
)
| 12481 | continue |
| 12482 | |
| 12483 | def _add_entry( |
| 12484 | self, |
| 12485 | *, |
| 12486 | path: Path, |
| 12487 | tokens: Tuple[int, ...], |
| 12488 | has_prompt_logits: bool, |
| 12489 | last_accessed: Optional[float] = None, |
| 12490 | ) -> None: |
| 12491 | existing = self._entries_by_tokens.get(tokens) |
| 12492 | if existing is not None: |
| 12493 | self._remove_entry(existing) |
| 12494 | entry_id = self._next_entry_id |
| 12495 | self._next_entry_id += 1 |
| 12496 | size_bytes = path.stat().st_size |
| 12497 | entry = SequenceDiskCache.Entry( |
| 12498 | entry_id=entry_id, |
| 12499 | path=path, |
| 12500 | tokens=tokens, |
| 12501 | size_bytes=size_bytes, |
| 12502 | has_prompt_logits=has_prompt_logits, |
| 12503 | last_accessed=last_accessed if last_accessed is not None else time.time(), |
| 12504 | ) |
| 12505 | self._entries_by_id[entry_id] = entry |
| 12506 | self._entries_by_tokens[tokens] = entry |
| 12507 | self._trie.extend(entry_id, tokens) |
| 12508 | self._size_bytes += size_bytes |
| 12509 | |
| 12510 | def _remove_entry(self, entry: "SequenceDiskCache.Entry") -> None: |
| 12511 | if entry.entry_id in self._trie.sequence_lengths: |
no test coverage detected