daniel156161
39a7a53873
Build and Push Docker Container / build-and-push (push) Successful in 3m52s
- Add compact storage package: codec, records, hydrate, replay blobs, cache. - Running games write game_turns and snake_states keyed by integer ids. - Encode moves as 1-byte codes and bodies as head plus step codes. - Store food and hazards only when they changed since the previous turn. - Strip reasoning fields that duplicate stored values and restore on read. - Pack finished games into one zstd blob per game in game_replays. - Merge late turns into packed blobs during the stale-game sweep. - Extract SQLite and PostgreSQL schemas into dedicated modules. - Add read/import game bundle and list_game_ids to the backend template. - Add legacy readers and converter for the old turns/snake_turns layout. - Add migrate_gameplay_to_compact.py with in-place or cross-DB migration. - Verify every migrated game and keep legacy rows on failure. - Support --since, --minimum-quality and guarded --drop-legacy. - Add just recipe migrate-gameplay-postgres. - Rewrite merge script to copy game by game via backend bundles. - Move benchmark state loading into benchmark_states for both backends. - Update PostgreSQL cleanup script to the new replay tables. - Drop sqlite-zstd build stages from the Dockerfile. - Remove old migration scripts, normalized_turn and their tests. - Add tests for compact storage, schema, legacy migration and PostgreSQL. - Document the compact layout and migration workflow in the README.
34 lines
963 B
Python
34 lines
963 B
Python
"""Per-process cache of running games to avoid repeated lookups per turn."""
|
|
|
|
from collections import OrderedDict
|
|
from dataclasses import dataclass, field
|
|
|
|
from .records import PointsState
|
|
|
|
@dataclass
|
|
class RunningGame:
|
|
gid:int
|
|
snake_ids:dict[str, int] = field(default_factory=dict)
|
|
points:PointsState|None = None
|
|
|
|
class RunningGameCache:
|
|
def __init__(self, max_games:int=512):
|
|
self._games:OrderedDict[str, RunningGame] = OrderedDict()
|
|
self._max_games = max(1, max_games)
|
|
|
|
def get(self, game_id:str) -> RunningGame|None:
|
|
game = self._games.get(game_id)
|
|
if game is not None:
|
|
self._games.move_to_end(game_id)
|
|
return game
|
|
|
|
def put(self, game_id:str, game:RunningGame) -> RunningGame:
|
|
self._games[game_id] = game
|
|
self._games.move_to_end(game_id)
|
|
while len(self._games) > self._max_games:
|
|
self._games.popitem(last=False)
|
|
return game
|
|
|
|
def drop(self, game_id:str) -> None:
|
|
self._games.pop(game_id, None)
|