Load Test

[linomp]

Current findings with Locust (testing API only):

• server tops up around 200 simultaneous users (at 20 new users per sec.) - only advancing turns until game over; no db access, no RUL inference nor other operations.

TODO:

• find tool & do an initial load test
• do full-game emulation in api test (Improve game emulation in load test #74)
• identify bottlenecks to address in V2

[linomp]

If sessions dict becomes bottleneck, consider redis for storing sessions:

see code

```python # session_store.py

import json
from typing import Dict, Optional
from pydantic import BaseModel
import os

try:
from redis import Redis
redis_client = Redis(host='localhost', port=6379, db=0, decode_responses=True)
REDIS_AVAILABLE = True
except ImportError:
redis_client = None
REDIS_AVAILABLE = False

Configuration to decide storage backend

USE_REDIS = os.getenv('USE_REDIS', 'false').lower() == 'true' and REDIS_AVAILABLE

In-memory fallback storage

_sessions: Dict[str, dict] = {}

class GameSession(BaseModel):
session_id: str
player_data: dict
game_state: dict

def to_dict(self):
    return self.dict()

@classmethod
def from_dict(cls, data: dict):
    return cls(**data)

def save_session(session: GameSession, ttl: Optional[int] = None):
"""
Save a session to Redis or in-memory dictionary.
"""
if USE_REDIS:
# Serialize the GameSession to JSON and save to Redis
if ttl:
redis_client.set(session.session_id, json.dumps(session.to_dict()), ex=ttl)
else:
redis_client.set(session.session_id, json.dumps(session.to_dict()))
else:
# Save the GameSession to the in-memory dictionary
_sessions[session.session_id] = session.to_dict()

def load_session(session_id: str) -> Optional[GameSession]:
"""
Load a session from Redis or in-memory dictionary.
"""
if USE_REDIS:
# Load the GameSession from Redis and deserialize from JSON
session_data = redis_client.get(session_id)
if session_data:
return GameSession.from_dict(json.loads(session_data))
return None
else:
# Load the GameSession from the in-memory dictionary
session_data = _sessions.get(session_id)
if session_data:
return GameSession.from_dict(session_data)
return None

def delete_session(session_id: str):
"""
Delete a session from Redis or in-memory dictionary.
"""
if USE_REDIS:
# Delete the GameSession from Redis
redis_client.delete(session_id)
else:
# Delete the GameSession from the in-memory dictionary
_sessions.pop(session_id, None)

Example usage

if name == "main":
# Create a new game session
session = GameSession(session_id="session1", player_data={"score": 100}, game_state={"level": 1})

# Save the session
save_session(session)

# Load the session
loaded_session = load_session("session1")
print("Loaded session:", loaded_session)

# Delete the session
delete_session("session1")

# Try to load again to confirm deletion
loaded_session_after_delete = load_session("session1")
print("Loaded session after delete:", loaded_session_after_delete)

</details>

[linomp]

Found this error with 1000 concurrent users: sqlite sqlalchemy.exc.TimeoutError: QueuePool limit of size 5 overflow 10 reached, connection timed out, timeout 30.00