Merge pull request #117 from KhrulkovV/worktree-benchmark-polish

perf: Benchmark suite + ORM-style field projection (44% collector speedup)

Author: KhrulkovV

gigaevo/database/program_storage.py

@@ -14,29 +14,43 @@ class PopulationSnapshot:
     Shared across all collector instances via ``storage.snapshot``.
     Call :meth:`bump` at phase boundaries to invalidate; collectors that
     see a stale epoch will re-fetch exactly once (others piggyback).
+
+    Single-slot cache keyed on ``(epoch, exclude)``. Works correctly when all
+    callers within an epoch use the same exclude value (which is true in practice:
+    EvolutionaryStatisticsCollector always uses exclude={"stage_results"}, all
+    other callers use exclude=None). If a new caller with a different exclude
+    value is added, the cache will thrash with no benefits -- a latent hazard
+    that would require a dict-based multi-slot cache to fully address.
     """
 
-    __slots__ = ("_epoch", "_cached_epoch", "_cached", "_lock")
+    __slots__ = ("_epoch", "_cached_epoch", "_cached_exclude", "_cached", "_lock")
 
     def __init__(self) -> None:
         self._epoch: int = 0
         self._cached_epoch: int = -1
+        self._cached_exclude: frozenset[str] | None = None
         self._cached: list[Program] = []
         self._lock = asyncio.Lock()
 
     def bump(self) -> None:
         """Increment the epoch, invalidating the cached snapshot."""
         self._epoch += 1
 
-    async def get_all(self, storage: ProgramStorage) -> list[Program]:
-        """Return cached programs if epoch matches, else fetch + cache."""
-        if self._cached_epoch == self._epoch:
+    async def get_all(
+        self,
+        storage: ProgramStorage,
+        *,
+        exclude: frozenset[str] | None = None,
+    ) -> list[Program]:
+        """Return cached programs if epoch+exclude matches, else fetch + cache."""
+        if self._cached_epoch == self._epoch and self._cached_exclude == exclude:
             return self._cached
         async with self._lock:
-            if self._cached_epoch == self._epoch:
+            if self._cached_epoch == self._epoch and self._cached_exclude == exclude:
                 return self._cached
-            programs = await storage.get_all()
+            programs = await storage.get_all(exclude=exclude)
             self._cached = programs
+            self._cached_exclude = exclude
             self._cached_epoch = self._epoch
             return programs
 
@@ -79,10 +93,20 @@ async def publish_status_event(
     ) -> None: ...
 
     @abstractmethod
-    async def get_all(self) -> list[Program]: ...
+    async def get_all(self, *, exclude: frozenset[str] | None = None) -> list[Program]:
+        """Return all programs.
+
+        Args:
+            exclude: Optional set of field names to skip during deserialization.
+                Excluded fields get their Pydantic defaults. Implementations
+                that don't support projection may ignore this parameter.
+        """
+        ...
 
     @abstractmethod
-    async def get_all_by_status(self, status: str) -> list[Program]: ...
+    async def get_all_by_status(
+        self, status: str, *, exclude: frozenset[str] | None = None
+    ) -> list[Program]: ...
 
     @abstractmethod
     async def get_ids_by_status(self, status: str) -> list[str]:

gigaevo/database/redis_program_storage.py

@@ -98,22 +98,32 @@ def _chunks(items: Iterable[str], n: int) -> Iterable[list[str]]:
             yield batch
 
     @staticmethod
-    def _safe_deserialize(raw: str, ctx: str) -> Program | None:
+    def _safe_deserialize(
+        raw: str,
+        ctx: str,
+        *,
+        exclude: frozenset[str] | None = None,
+    ) -> Program | None:
         try:
-            return Program.from_dict(_loads(raw))
+            return Program.from_dict(_loads(raw), exclude=exclude)
         except Exception as e:
             logger.warning("[RedisProgramStorage] Corrupt data in {}: {}", ctx, e)
             return None
 
     async def _mget_by_keys(
-        self, r: aioredis.Redis, keys: list[str], ctx: str
+        self,
+        r: aioredis.Redis,
+        keys: list[str],
+        ctx: str,
+        *,
+        exclude: frozenset[str] | None = None,
     ) -> list[Program]:
         out: list[Program] = []
         for batch in self._chunks(keys, MGET_CHUNK_SIZE):
             blobs = await r.mget(*batch)
             for raw in blobs:
                 if raw:
-                    p = self._safe_deserialize(raw, ctx)
+                    p = self._safe_deserialize(raw, ctx, exclude=exclude)
                     if p is not None:
                         out.append(p)
         return out
@@ -270,8 +280,15 @@ async def _size(r: aioredis.Redis) -> int:
 
         return await self._conn.execute("size", _size)
 
-    async def get_all(self) -> list[Program]:
-        """Get all programs using SCAN + chunked MGET."""
+    async def get_all(self, *, exclude: frozenset[str] | None = None) -> list[Program]:
+        """Get all programs using SCAN + chunked MGET.
+
+        Args:
+            exclude: Optional set of field names to skip during deserialization
+                (passed through to :meth:`Program.from_dict`). Excluded fields
+                get their Pydantic defaults. Example: ``exclude=frozenset({"stage_results"})``
+                saves ~33% deserialization cost for analytics callers.
+        """
 
         async def _scan_then_mget(r: aioredis.Redis) -> list[Program]:
             keys: list[str] = []
@@ -281,7 +298,7 @@ async def _scan_then_mget(r: aioredis.Redis) -> list[Program]:
                 keys.append(key)
             if not keys:
                 return []
-            return await self._mget_by_keys(r, keys, "get_all")
+            return await self._mget_by_keys(r, keys, "get_all", exclude=exclude)
 
         return await self._conn.execute("get_all", _scan_then_mget)
 
@@ -340,14 +357,18 @@ async def _event(r: aioredis.Redis) -> None:
 
         await self._conn.execute("publish_status_event", _event)
 
-    async def get_all_by_status(self, status: str) -> list[Program]:
+    async def get_all_by_status(
+        self, status: str, *, exclude: frozenset[str] | None = None
+    ) -> list[Program]:
         ids = await self._ids_for_status(status)
         if not ids:
             return []
 
         async def _by_status(r: aioredis.Redis) -> list[Program]:
             keys = [self._keys.program(pid) for pid in ids]
-            programs = await self._mget_by_keys(r, keys, f"get_all_by_status:{status}")
+            programs = await self._mget_by_keys(
+                r, keys, f"get_all_by_status:{status}", exclude=exclude
+            )
             return [p for p in programs if p.state.value == status]
 
         return await self._conn.execute("get_all_by_status", _by_status)

gigaevo/programs/program.py

@@ -54,6 +54,13 @@ def _metadata_is_json_native(value: dict[str, Any]) -> bool:
     }
 )
 
+# Fields safe to exclude in Program.from_dict: have safe Pydantic defaults.
+# code is required (no default), id generates new UUIDs, state resets to
+# QUEUED — none of those are safe to silently default.
+DEFERRABLE_FIELDS: frozenset[str] = frozenset(
+    {"stage_results", "metadata", "metrics", "name"}
+)
+
 
 def _utcnow() -> datetime:
     return datetime.now(UTC)
@@ -162,8 +169,34 @@ def to_dict(self) -> dict[str, Any]:
         return self.model_dump(mode="json")
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> Program:
+    def from_dict(
+        cls,
+        data: dict[str, Any],
+        *,
+        exclude: frozenset[str] | None = None,
+    ) -> Program:
+        """Deserialize a Program from a dict (e.g. JSON-decoded Redis blob).
+
+        Args:
+            data: Dict as produced by :meth:`to_dict`.
+            exclude: Optional set of field names to skip during deserialization.
+                Excluded fields get their Pydantic defaults (e.g. ``{}`` for
+                ``stage_results``). Works like Django ORM's ``defer()`` —
+                callers that only need metrics/lineage can skip expensive fields
+                such as ``stage_results`` to avoid reconstruction cost.
+        """
+        if exclude:
+            bad = exclude - DEFERRABLE_FIELDS
+            if bad:
+                raise ValueError(
+                    f"Cannot exclude field(s): {bad}. "
+                    f"Only {DEFERRABLE_FIELDS} may be excluded."
+                )
+
         d = dict(data)
+        if exclude:
+            for field in exclude:
+                d.pop(field, None)
         if "metadata" in d and isinstance(d["metadata"], str):
             d["metadata"] = pickle_b64_deserialize(d["metadata"])
         if "stage_results" in d and isinstance(d["stage_results"], dict):

gigaevo/programs/stages/collector.py

@@ -360,8 +360,11 @@ def __init__(self, *, metrics_context: MetricsContext, **kwargs: Any):
         super().__init__(**kwargs)
         self.metrics_context = metrics_context
 
+    #: Fields the collector never accesses — skip during deserialization.
+    _EXCLUDE = frozenset({"stage_results"})
+
     async def _collect_programs(self, program: Program) -> list[Program]:
-        return await self.storage.snapshot.get_all(self.storage)
+        return await self.storage.snapshot.get_all(self.storage, exclude=self._EXCLUDE)
 
     async def _process(
         self, program: Program, programs: list[Program]

tests/benchmarks/conftest.py

@@ -336,3 +336,36 @@ def __enter__(self):
 
     def __exit__(self, *args):
         self.elapsed_ms = (time.perf_counter() - self._start) * 1000
+
+
+# ---------------------------------------------------------------------------
+# Realistic code generator
+# ---------------------------------------------------------------------------
+
+# Reusable code snippets that compose into valid Python of varying sizes.
+_CODE_BLOCKS = [
+    "def helper_{i}(data):\n    return [x * {i} + 1 for x in data if x > 0]\n\n",
+    "def transform_{i}(items):\n    result = {{}}\n    for idx, val in enumerate(items):\n        if val is not None:\n            result[f'k_{{idx}}'] = val ** 2\n    return result\n\n",
+    "class Handler_{i}:\n    def __init__(self, threshold={i}):\n        self.threshold = threshold\n        self._cache = {{}}\n\n    def process(self, values):\n        out = []\n        for v in values:\n            if v > self.threshold:\n                out.append(v - self.threshold)\n            else:\n                out.append(0)\n        self._cache[len(out)] = out\n        return out\n\n",
+    "def analyze_{i}(matrix):\n    rows = len(matrix)\n    if rows == 0:\n        return []\n    cols = len(matrix[0])\n    totals = [0.0] * cols\n    for r in range(rows):\n        for c in range(cols):\n            totals[c] += matrix[r][c]\n    return [t / rows for t in totals]\n\n",
+    "def search_{i}(haystack, needle):\n    lo, hi = 0, len(haystack) - 1\n    while lo <= hi:\n        mid = (lo + hi) // 2\n        if haystack[mid] == needle:\n            return mid\n        elif haystack[mid] < needle:\n            lo = mid + 1\n        else:\n            hi = mid - 1\n    return -1\n\n",
+]
+
+
+def make_realistic_code(target_bytes: int) -> str:
+    """Generate valid Python code of approximately *target_bytes* size.
+
+    Uses composable function/class blocks that look like real evolved code
+    rather than random strings. The result always parses as valid Python.
+    Stops adding blocks once *target_bytes* is reached (never truncates mid-block).
+    """
+    header = "def run_code():\n    return 42\n\n"
+    parts = [header]
+    current_size = len(header)
+    idx = 0
+    while current_size < target_bytes:
+        block = _CODE_BLOCKS[idx % len(_CODE_BLOCKS)].format(i=idx)
+        parts.append(block)
+        current_size += len(block)
+        idx += 1
+    return "".join(parts)