""" Sahih Islam — Multilingual Concurrency & Telemetry Load Test Tests 50, 100, and 200 concurrent users with realistic mixed-language requests: - Languages: English (40%), Urdu (35%), Hinglish (20%), Arabic (5%) - Tiers: 50 users, 100 users, 200 users - Telemetry: • Cache hit/miss p50, p95, p99 latency • Canonical cache hit rate & full response cache hit rate • Per-request Atlas queries & LLM calls • Redis & L1 memory consumption / process RSS """ import sys import os import io import time import json import random import asyncio from typing import Dict, List, Any, Tuple if sys.platform == "win32": try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stderr.reconfigure(encoding="utf-8", errors="replace") except Exception: sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace") import httpx import psutil # Core Islamic topic queries (simulating high-frequency questions) CORE_QUERIES = [ "virtues of prayer salah on time", "fasting rules in ramadan", "rights of neighbors in islam", "patience during trials and tribulations sabr", "manners of speaking and honesty in speech", "seeking knowledge is obligatory for every muslim", "supplication dua acceptance conditions", "charity sadaqah and removing harm from path", "night of decree laylatul qadr virtues", "kindness to parents mother and father", ] # Tail Islamic queries (simulating occasional long-tail searches) TAIL_QUERIES = [ "rules of inheritance distribution for daughters", "prostration of forgetfulness sajdah sahw in prayer", "etiquette of slaughtering animals dhabihah", "borrowing money with collateral rahn", "virtue of reciting surah al-kahf on friday", ] # Language distribution weights: English 40%, Urdu 35%, Hinglish 20%, Arabic 5% LANGUAGES_POOL = ["English"] * 40 + ["Urdu"] * 35 + ["Hinglish"] * 20 + ["Arabic"] * 5 def get_backend_pid() -> int: """Finds the Uvicorn backend process PID listening on 7860.""" for proc in psutil.process_iter(['pid', 'name', 'cmdline']): try: cmd = " ".join(proc.info['cmdline'] or []) if "run.py" in cmd or "uvicorn" in cmd: return proc.info['pid'] except (psutil.NoSuchProcess, psutil.AccessDenied): pass return os.getpid() async def get_redis_stats() -> Dict[str, Any]: """Retrieves live Redis memory and key count.""" try: from app.infrastructure.redis import redis_manager await redis_manager.init() if redis_manager._client: info = await redis_manager._client.info('memory') keys = await redis_manager._client.dbsize() return { "used_memory_human": info.get("used_memory_human", "N/A"), "peak_memory_human": info.get("used_memory_peak_human", "N/A"), "total_keys": keys, } else: return { "used_memory_human": "in-memory", "total_keys": len(redis_manager._memory_cache), } except Exception as e: return {"error": str(e), "total_keys": 0, "used_memory_human": "N/A"} def calc_percentiles(latencies: List[float]) -> Dict[str, float]: """Calculates p50, p90, p95, p99 min and max.""" if not latencies: return {"p50": 0.0, "p90": 0.0, "p95": 0.0, "p99": 0.0, "min": 0.0, "max": 0.0} s = sorted(latencies) n = len(s) return { "p50": round(s[int(0.50 * n)], 1), "p90": round(s[min(n - 1, int(0.90 * n))], 1), "p95": round(s[min(n - 1, int(0.95 * n))], 1), "p99": round(s[min(n - 1, int(0.99 * n))], 1), "min": round(s[0], 1), "max": round(s[-1], 1), } async def simulate_user( user_id: int, client: httpx.AsyncClient, base_url: str, requests_per_user: int, results: List[Dict[str, Any]], lock: asyncio.Lock ): """Simulates a concurrent user submitting mixed-language Hadith queries.""" for req_idx in range(requests_per_user): # 80% chance popular core query (higher cache reuse), 20% long-tail if random.random() < 0.80: query = random.choice(CORE_QUERIES) else: query = random.choice(TAIL_QUERIES) lang = random.choice(LANGUAGES_POOL) t0 = time.perf_counter() status_code = 0 is_success = False text_received = "" is_hit = False try: resp = await client.post( f"{base_url}/api/ask", json={ "query": query, "language": lang, "top_k": 3, "strict_mode": False, "history": [], }, timeout=35.0 ) lat_ms = (time.perf_counter() - t0) * 1000.0 status_code = resp.status_code if status_code == 200: is_success = True text_received = resp.text # If latency is under 200ms, it's served from cache (Tier 1 or Tier 1b) if lat_ms < 250.0: is_hit = True except Exception as e: lat_ms = (time.perf_counter() - t0) * 1000.0 status_code = 500 record = { "user_id": user_id, "req_idx": req_idx, "query": query, "language": lang, "latency_ms": lat_ms, "status_code": status_code, "success": is_success, "is_hit": is_hit, "content_len": len(text_received), } async with lock: results.append(record) # Realistic user think-time between queries (10ms - 40ms) await asyncio.sleep(random.uniform(0.01, 0.04)) async def run_concurrency_tier( num_users: int, requests_per_user: int, base_url: str, backend_pid: int ) -> Dict[str, Any]: """Executes a load test tier with num_users concurrent clients.""" print(f"\n================================================================================") print(f"🚀 EXECUTING TIER: {num_users} CONCURRENT USERS ({num_users * requests_per_user} TOTAL REQUESTS)") print(f"================================================================================") # Pre-tier memory check try: proc = psutil.Process(backend_pid) pre_rss_mb = proc.memory_info().rss / 1024 / 1024 except Exception: pre_rss_mb = 0.0 pre_redis = await get_redis_stats() limits = httpx.Limits(max_connections=max(250, num_users + 50), max_keepalive_connections=max(200, num_users)) async with httpx.AsyncClient(limits=limits, timeout=40.0) as client: results: List[Dict[str, Any]] = [] lock = asyncio.Lock() t_start = time.perf_counter() tasks = [ simulate_user(uid, client, base_url, requests_per_user, results, lock) for uid in range(num_users) ] await asyncio.gather(*tasks) total_time_s = time.perf_counter() - t_start # Post-tier memory check try: post_rss_mb = proc.memory_info().rss / 1024 / 1024 except Exception: post_rss_mb = 0.0 post_redis = await get_redis_stats() # Process metrics total_reqs = len(results) success_reqs = sum(1 for r in results if r["success"]) fail_reqs = total_reqs - success_reqs rps = total_reqs / total_time_s if total_time_s > 0 else 0.0 hit_results = [r for r in results if r["is_hit"] and r["success"]] miss_results = [r for r in results if not r["is_hit"] and r["success"]] all_lats = [r["latency_ms"] for r in results] hit_lats = [r["latency_ms"] for r in hit_results] miss_lats = [r["latency_ms"] for r in miss_results] all_p = calc_percentiles(all_lats) hit_p = calc_percentiles(hit_lats) miss_p = calc_percentiles(miss_lats) # Estimate Canonical vs Full Response Hit Breakdown # Latencies < 25ms are Full Response RAM hits, 25ms - 250ms are Canonical Multilingual hits full_resp_hits = sum(1 for r in hit_results if r["latency_ms"] < 25.0) canonical_hits = sum(1 for r in hit_results if r["latency_ms"] >= 25.0) total_hits = len(hit_results) hit_rate_pct = (total_hits / max(1, total_reqs)) * 100.0 canonical_pct = (canonical_hits / max(1, total_reqs)) * 100.0 full_resp_pct = (full_resp_hits / max(1, total_reqs)) * 100.0 # Avoided calls: # Each canonical hit avoided 1 Atlas vector search + 3 translation calls # Each full response hit avoided 1 Atlas search + 1 query rewrite + 3 translation calls atlas_avoided = total_hits llm_translations_avoided = total_hits * 3 llm_rewrites_avoided = full_resp_hits # Language breakdown lang_counts = {} for r in results: lang_counts[r["language"]] = lang_counts.get(r["language"], 0) + 1 print(f"⏱️ Duration: {total_time_s:.2f}s | Throughput: {rps:.2f} RPS | Success Rate: {(success_reqs/max(1,total_reqs))*100:.1f}%") print(f"📊 Latency Percentiles (Overall): p50={all_p['p50']}ms | p90={all_p['p90']}ms | p95={all_p['p95']}ms | p99={all_p['p99']}ms") print(f"⚡ Cache Hit Latencies (p50/p95): p50={hit_p['p50']}ms | p95={hit_p['p95']}ms | min={hit_p['min']}ms | max={hit_p['max']}ms") print(f"❄️ Cache Miss Latencies (p50/p95): p50={miss_p['p50']}ms | p95={miss_p['p95']}ms | min={miss_p['min']}ms | max={miss_p['max']}ms") print(f"🎯 Total Hit Rate: {hit_rate_pct:.1f}% (Canonical Hit Rate: {canonical_pct:.1f}%, Full Response Hit Rate: {full_resp_pct:.1f}%)") print(f"🛡️ Calls Avoided: Atlas Queries={atlas_avoided} | LLM Translations={llm_translations_avoided} | LLM Rewrites={llm_rewrites_avoided}") print(f"💾 Memory: Process RSS={post_rss_mb:.1f}MB (Δ {post_rss_mb - pre_rss_mb:+.1f}MB) | Redis Keys={post_redis.get('total_keys', 0)} ({post_redis.get('used_memory_human', 'N/A')})") print(f"🌐 Languages: {lang_counts}") return { "num_users": num_users, "total_requests": total_reqs, "success_requests": success_reqs, "failed_requests": fail_reqs, "duration_seconds": round(total_time_s, 2), "rps": round(rps, 2), "hit_rate_pct": round(hit_rate_pct, 1), "canonical_hit_rate_pct": round(canonical_pct, 1), "full_response_hit_rate_pct": round(full_resp_pct, 1), "atlas_avoided": atlas_avoided, "llm_translations_avoided": llm_translations_avoided, "llm_rewrites_avoided": llm_rewrites_avoided, "percentiles_overall": all_p, "percentiles_cache_hit": hit_p, "percentiles_cache_miss": miss_p, "process_rss_pre_mb": round(pre_rss_mb, 1), "process_rss_post_mb": round(post_rss_mb, 1), "redis_keys": post_redis.get("total_keys", 0), "redis_memory": post_redis.get("used_memory_human", "N/A"), "languages": lang_counts, } async def main(): base_url = "http://127.0.0.1:7860" backend_pid = get_backend_pid() print("================================================================================") print("🌙 SAHIH ISLAM: SUSTAINED MULTILINGUAL CONCURRENCY LOAD TEST") print(" Testing 50, 100, and 200 Concurrent Users with Realistic Mixed-Language Traffic") print(" Tracking: Latency Percentiles (p50/p95), Cache Hit Rates, External Calls, Memory") print("================================================================================") print(f"Backend Server: {base_url} (Target PID: {backend_pid})\n") # Warm-up phase: Pre-seed core queries in English so canonical translations are available print("🔥 PRE-WARMING: Priming canonical translation records for core Islamic topics...") limits = httpx.Limits(max_connections=20, max_keepalive_connections=10) async with httpx.AsyncClient(limits=limits, timeout=30.0) as client: for q in CORE_QUERIES[:6]: try: await client.post(f"{base_url}/api/ask", json={ "query": q, "language": "English", "top_k": 3, "strict_mode": False, }) except Exception: pass await asyncio.sleep(0.8) print("✓ Warm-up complete. Canonical translations ready for mixed-language queries.\n") tier_results = [] # Tier 1: 50 Concurrent Users (3 requests each = 150 requests) t1 = await run_concurrency_tier(50, 3, base_url, backend_pid) tier_results.append(t1) await asyncio.sleep(1.0) # Tier 2: 100 Concurrent Users (3 requests each = 300 requests) t2 = await run_concurrency_tier(100, 3, base_url, backend_pid) tier_results.append(t2) await asyncio.sleep(1.0) # Tier 3: 200 Concurrent Users (3 requests each = 600 requests) t3 = await run_concurrency_tier(200, 3, base_url, backend_pid) tier_results.append(t3) # Save final results JSON out_path = os.path.join(os.path.dirname(__file__), "..", "multilingual_concurrency_test_results.json") with open(out_path, "w", encoding="utf-8") as f: json.dump(tier_results, f, indent=2) print("\n================================================================================") print("🏆 FINAL CONCURRENCY BENCHMARK SUMMARY TABLE") print("================================================================================") print(f"{'Users':<6} | {'RPS':<7} | {'Success':<8} | {'Hit Rate':<9} | {'Canon Hit':<10} | {'Hit p50':<9} | {'Hit p95':<9} | {'Miss p50':<10} | {'RAM (MB)':<8}") print("-" * 92) for t in tier_results: u = t["num_users"] r = t["rps"] s = f"{(t['success_requests']/max(1, t['total_requests']))*100:.1f}%" h = f"{t['hit_rate_pct']:.1f}%" c = f"{t['canonical_hit_rate_pct']:.1f}%" hp50 = f"{t['percentiles_cache_hit']['p50']}ms" hp95 = f"{t['percentiles_cache_hit']['p95']}ms" mp50 = f"{t['percentiles_cache_miss']['p50']}ms" ram = f"{t['process_rss_post_mb']}MB" print(f"{u:<6} | {r:<7.1f} | {s:<8} | {h:<9} | {c:<10} | {hp50:<9} | {hp95:<9} | {mp50:<10} | {ram:<8}") print("-" * 92) print(f"Results persisted to: {out_path}") if __name__ == "__main__": asyncio.run(main())