Spaces:

sahihislam
/

superadmin

Running on Zero

superadmin / scripts /benchmark_multilingual_concurrency.py

feat: resolve nearby masjids 500 error, add popular sanctuaries, and optimize search architecture

ae9655d 26 days ago

14.5 kB

"""
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())

	"""
	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())