Performance Optimization

This guide covers techniques for optimizing MAID server performance, including profiling, memory management, and best practices.

Profiling Tools

MAID includes built-in profiling capabilities:

CLI Profiling

Known issue — the maid dev profile CLI does not currently emit a report. When a --duration elapses, the profiling session auto-stops and is removed from the manager's active set; the CLI's follow-up stop_session() call then returns None, so the command prints No results collected and exits with a non-zero status (the collected results are retained internally but not read back). Until this is fixed, use the programmatic profiling API below, which gives you the ProfileResults object directly. The commands are shown here for reference only:

# Profile memory and tick performance
maid dev profile --types=memory,tick --duration=60

# Save report to file
maid dev profile -t memory,tick -d 60 -o report.html

# JSON output for CI/CD
maid dev profile -t all --json

Memory Snapshots

# Take a memory snapshot
maid dev memory-snapshot

# Save to file with more detail
maid dev memory-snapshot -o snapshot.json --depth 15 --top 50

Programmatic Profiling

This is the reliable path today. Call stop_session() yourself before the configured duration elapses (or omit duration entirely for fully manual control); otherwise the session auto-stops and stop_session() returns None.

from maid_engine.profiling import ProfileManager, ProfileType

manager = ProfileManager(enabled=True)

# Omit `duration` so the session only stops when you ask it to.
session = await manager.start_session(
    ProfileType.TICK | ProfileType.MEMORY,
)

# ... run game ...

# Stop and get results (returns ProfileResults, or None if already stopped)
results = await manager.stop_session(session.session_id)
print(results)

Tick Performance

Understanding Tick Budget

At 4 ticks per second, each tick has ~250ms budget:

Tick Budget: 250ms
├── System Updates: ~150ms (target)
├── Event Processing: ~50ms (target)
├── Network I/O: ~30ms (target)
└── Buffer: ~20ms (safety margin)

Monitoring Tick Time

import time
from collections import deque
from maid_engine.core.world import World

class TickMonitor:
    def __init__(self, world: World, window: int = 1000):
        self._world = world
        # Bounded rolling window. A plain list grows without limit — one entry
        # per tick, forever — which leaks memory on a long-running server (the
        # exact problem this guide is about). deque(maxlen=...) keeps only the
        # most recent `window` samples.
        self._tick_times: deque[float] = deque(maxlen=window)

    async def tick_wrapper(self, delta: float) -> None:
        start = time.perf_counter()
        await self._world.tick(delta)
        elapsed = time.perf_counter() - start

        self._tick_times.append(elapsed)

        if elapsed > 0.25:  # Over budget
            logger.warning(f"Tick took {elapsed*1000:.1f}ms (budget: 250ms)")

    def get_stats(self) -> dict:
        if not self._tick_times:
            return {}

        return {
            "avg_tick_ms": sum(self._tick_times) / len(self._tick_times) * 1000,
            "max_tick_ms": max(self._tick_times) * 1000,
            "min_tick_ms": min(self._tick_times) * 1000,
            "over_budget_count": sum(1 for t in self._tick_times if t > 0.25),
        }

Optimizing System Updates

class OptimizedSystem(System):
    priority = 50

    def __init__(self, world: World):
        super().__init__(world)
        # Cache frequently used queries
        self._cached_entities: list[Entity] = []
        self._cache_dirty = True

    async def startup(self) -> None:
        # Subscribe to invalidate cache on entity changes
        self.events.subscribe(EntityCreatedEvent, self._invalidate_cache)
        self.events.subscribe(EntityDestroyedEvent, self._invalidate_cache)
        self.events.subscribe(ComponentAddedEvent, self._invalidate_cache)
        self.events.subscribe(ComponentRemovedEvent, self._invalidate_cache)

    async def _invalidate_cache(self, event) -> None:
        self._cache_dirty = True

    def _refresh_cache(self) -> None:
        if self._cache_dirty:
            self._cached_entities = list(
                self.entities.with_components(HealthComponent, CombatComponent)
            )
            self._cache_dirty = False

    async def update(self, delta: float) -> None:
        self._refresh_cache()

        # Use cached list instead of querying every tick
        for entity in self._cached_entities:
            await self._process_entity(entity, delta)

Entity Queries

Efficient Queries

# Good - single query, iterate once
async def update(self, delta: float) -> None:
    for entity in self.entities.with_components(HealthComponent):
        health = entity.get(HealthComponent)
        if health.current < health.maximum:
            health.current += delta

# Bad - multiple queries
async def update(self, delta: float) -> None:
    for entity in self.entities.all():
        if entity.has(HealthComponent):  # Second query
            health = entity.get(HealthComponent)
            if health.current < health.maximum:
                health.current += delta

Batch Processing

# Process entities in batches to avoid long-running loops
BATCH_SIZE = 100

async def update(self, delta: float) -> None:
    entities = list(self.entities.with_components(AIComponent))

    for i in range(0, len(entities), BATCH_SIZE):
        batch = entities[i:i + BATCH_SIZE]
        for entity in batch:
            await self._process_ai(entity, delta)

        # Yield to event loop between batches
        await asyncio.sleep(0)

Index Optimization

# Use tags for fast filtering
class OptimizedWorld:
    def get_players_in_room(self, room_id: UUID) -> list[Entity]:
        """Get players in a room efficiently."""
        # Fast: tag query + room check
        return [
            e for e in self.entities.with_tag("player")
            if e.get(PositionComponent).room_id == room_id
        ]

        # Slow: check all entities
        # return [
        #     e for e in self.entities.all()
        #     if e.has_tag("player") and e.get(PositionComponent).room_id == room_id
        # ]

Memory Management

Component Memory

# Good - minimal component data
class HealthComponent(Component):
    current: int
    maximum: int

# Bad - unnecessary data
class HealthComponent(Component):
    current: int
    maximum: int
    history: list[int] = []  # Grows unbounded!
    last_damage_source: str = ""
    created_at: datetime = None

Entity Cleanup

class CleanupSystem(System):
    """Periodically clean up dead entities."""
    priority = 200  # Run late

    async def update(self, delta: float) -> None:
        to_destroy = []

        for entity in self.entities.with_tag("dead"):
            # Check if death animation/effects are done
            if entity.try_get(DeathTimerComponent) is None:
                to_destroy.append(entity.id)

        for entity_id in to_destroy:
            self.entities.destroy(entity_id)

Event Cleanup

# Avoid holding references in event handlers
async def bad_handler(event: DamageEvent) -> None:
    # Bad - holds reference to entity
    self._damaged_entities.append(
        world.get_entity(event.target_id)
    )

async def good_handler(event: DamageEvent) -> None:
    # Good - stores only ID
    self._damaged_entity_ids.append(event.target_id)

Memory Profiling

from maid_engine.profiling import MemoryCollector

collector = MemoryCollector(
    trace_depth=10,
    top_allocations_count=20,
)

collector.start()

# Run some game ticks
for _ in range(100):
    await world.tick(0.25)

snapshot = collector.take_snapshot()
collector.stop()

print(f"Total allocated: {snapshot.total_allocated / 1024 / 1024:.1f} MB")
print(f"By module:")
for module, size in sorted(snapshot.by_module.items(), key=lambda x: -x[1])[:10]:
    print(f"  {module}: {size / 1024:.1f} KB")

Network Optimization

Message Batching

class BatchedMessageSender:
    """Batch messages to reduce network calls."""

    def __init__(self, session: Session, flush_interval: float = 0.1):
        self._session = session
        self._buffer: list[str] = []
        self._flush_interval = flush_interval
        self._last_flush = time.time()

    def queue(self, message: str) -> None:
        self._buffer.append(message)

    async def flush_if_needed(self) -> None:
        now = time.time()
        if self._buffer and (now - self._last_flush) >= self._flush_interval:
            await self._session.send("\n".join(self._buffer))
            self._buffer.clear()
            self._last_flush = now

Compression

import zlib

class CompressedSession:
    """Session with compression for large messages."""

    COMPRESSION_THRESHOLD = 1024  # Bytes

    async def send(self, message: str) -> None:
        data = message.encode()

        if len(data) > self.COMPRESSION_THRESHOLD:
            # Compress large messages
            compressed = zlib.compress(data)
            await self._raw_send(compressed, compressed=True)
        else:
            await self._raw_send(data, compressed=False)

AI Performance

Rate Limiting

import asyncio
import time

from maid_engine.ai.providers.base import (
    CompletionResult,
    LLMProvider,
    Message,
)

class RateLimitedAI:
    """Rate-limited wrapper around an LLMProvider."""

    def __init__(self, provider: LLMProvider, calls_per_second: float = 2.0):
        self._provider = provider
        self._min_interval = 1.0 / calls_per_second
        self._last_call = 0.0
        self._lock = asyncio.Lock()

    async def complete(self, messages: list[Message]) -> CompletionResult:
        async with self._lock:
            # Enforce rate limit
            elapsed = time.time() - self._last_call
            if elapsed < self._min_interval:
                await asyncio.sleep(self._min_interval - elapsed)

            self._last_call = time.time()
            # LLMProvider.complete() returns a CompletionResult, not a str.
            return await self._provider.complete(messages)

Response Caching

import hashlib

from maid_engine.ai.providers.base import LLMProvider, Message

class CachedAI:
    """Cache AI responses for identical prompts."""

    def __init__(self, provider: LLMProvider, cache_size: int = 100):
        self._provider = provider
        self._cache: dict[str, str] = {}
        self._cache_size = cache_size

    def _cache_key(self, messages: list[Message]) -> str:
        content = str([(m.role, m.content) for m in messages])
        return hashlib.md5(content.encode()).hexdigest()

    async def complete(self, messages: list[Message]) -> str:
        key = self._cache_key(messages)

        if key in self._cache:
            return self._cache[key]

        # complete() returns a CompletionResult; pull the text off .content.
        result = await self._provider.complete(messages)

        # LRU eviction
        if len(self._cache) >= self._cache_size:
            oldest = next(iter(self._cache))
            del self._cache[oldest]

        self._cache[key] = result.content
        return result.content

Multi-World Optimization

World Tick Scheduling

async def optimized_tick_all(manager: WorldManager, delta: float) -> None:
    """Tick worlds with priority scheduling."""
    worlds = manager.list_worlds(include_private=True)

    # Sort by player count (busy worlds first)
    worlds.sort(
        key=lambda w: count_players(w.world),
        reverse=True,
    )

    # Tick high-priority worlds
    high_priority = [w for w in worlds if count_players(w.world) > 0]
    for world in high_priority:
        await world.world.tick(delta)

    # Tick empty worlds less frequently
    low_priority = [w for w in worlds if count_players(w.world) == 0]
    if should_tick_empty_worlds():
        for world in low_priority:
            await world.world.tick(delta)

Instance Pooling

class DungeonPool:
    """Pool dungeon instances for reuse."""

    def __init__(self, manager: WorldManager, pool_size: int = 5):
        self._manager = manager
        self._pool_size = pool_size
        self._available: deque[str] = deque()
        self._in_use: set[str] = set()

    async def initialize(self) -> None:
        """Pre-create dungeon instances."""
        for i in range(self._pool_size):
            world_id = f"dungeon-pool-{i}"
            await self._create_instance(world_id)
            self._available.append(world_id)

    async def acquire(self) -> str:
        """Get a dungeon instance from the pool."""
        if self._available:
            world_id = self._available.popleft()
        else:
            # Pool exhausted, create new instance
            world_id = f"dungeon-dynamic-{uuid4()}"
            await self._create_instance(world_id)

        self._in_use.add(world_id)
        return world_id

    async def release(self, world_id: str) -> None:
        """Return a dungeon instance to the pool."""
        self._in_use.discard(world_id)

        # Reset and return to pool
        await self._reset_instance(world_id)
        self._available.append(world_id)

Best Practices Summary

Do

• Cache entity queries when possible
• Use tags for fast entity categorization
• Batch network messages
• Rate limit AI calls
• Clean up dead entities promptly
• Profile before optimizing
• Use async operations for I/O

Don't

• Query entities multiple times per tick
• Store growing lists in components
• Make synchronous I/O calls in update()
• Create new objects every tick
• Hold references to destroyed entities
• Ignore memory growth
• Optimize prematurely

Monitoring

Health Checks

class HealthMonitor:
    """Monitor server health metrics."""

    def __init__(self, world: World):
        self._world = world
        self._start_time = time.time()

    def get_health(self) -> dict:
        uptime = time.time() - self._start_time
        # Use the world's own tick counter. A local counter would have to be
        # incremented somewhere; if you forget (as is easy to do), it stays 0
        # and ticks_per_second is always 0, so the "low tick rate" alert fires
        # forever. World.tick_count is incremented by World.tick() itself.
        tick_count = self._world.tick_count
        return {
            "uptime_seconds": uptime,
            "tick_count": tick_count,
            "ticks_per_second": tick_count / uptime if uptime > 0 else 0,
            "entity_count": self._world.entities.count(),
            "player_count": len(list(self._world.entities.with_tag("player"))),
            "memory_mb": get_process_memory_mb(),
        }

Alerts

async def check_performance_alerts(monitor: HealthMonitor) -> list[str]:
    """Check for performance issues."""
    alerts = []
    health = monitor.get_health()

    if health["ticks_per_second"] < 3.5:  # Below 4 TPS target
        alerts.append(f"Low tick rate: {health['ticks_per_second']:.1f} TPS")

    if health["memory_mb"] > 1024:  # Over 1GB
        alerts.append(f"High memory usage: {health['memory_mb']:.0f} MB")

    if health["entity_count"] > 10000:  # Many entities
        alerts.append(f"High entity count: {health['entity_count']}")

    return alerts

Performance Benchmarks

MAID includes performance benchmark tests in packages/maid-engine/tests/benchmarks/ that validate documented performance claims:

Operation	Target
Session authentication lookup	< 0.1ms
Command execution latency	< 2ms
Event dispatch latency	< 1ms
Room lookup latency	< 0.5ms
Entity creation	< 1ms
Entity query (1000 entities)	< 5ms

Running Benchmarks

# Run all benchmarks
uv run pytest packages/maid-engine/tests/benchmarks/ -v -m benchmark

# Skip benchmarks in regular test runs
uv run pytest packages/maid-engine/tests/ -m "not benchmark"

Future Enhancement: CI Benchmark Enforcement

Currently, performance benchmarks are available for manual execution but are not automatically enforced in CI pipelines. This is tracked as technical debt. Future enhancements should include:

• Automated benchmark runs on PRs that modify performance-critical code paths
• Regression detection with configurable thresholds
• Historical performance tracking and trend analysis
• Alerting when benchmarks exceed target thresholds

See GitHub Issue Tracking for updates on CI benchmark integration.

Next Steps

• Multi-World Support - Optimize multi-world setups
• AI Integration - Efficient AI usage