Spaces:

sahihislam
/

superadmin

Running on Zero

superadmin / scripts /benchmark_single_flight.py

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

ae9655d 18 days ago

9.37 kB

"""
Sahih Islam — Single-Flight Request Coalescing Concurrency Benchmark
Quantifies the deduplication ratio, latency distribution (p50, p95, p99),
avoided external Atlas/LLM calls, and error rates under cold query stampede.
Benchmark Workload:
  - 50 concurrent requests simulating 5 unique uncached cold queries
    (10 concurrent callers per cold query = 10x stampede factor).
  - Mixed language mix: Urdu (35%), English (40%), Hinglish (20%), Arabic (5%).
  - Telemetry: SingleFlight deduplication ratio, client latency (p50, p95, p99),
    actual successful RPS, avoided Atlas searches, and avoided DeepSeek rewrites.
"""

import sys
import os
import io

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 time
import json
import asyncio
import numpy as np
import httpx
from typing import Dict, Any, List

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

BASE_URL = "http://127.0.0.1:7860"
AUTH_TOKEN = "dev_mock_token_locust_user"
HEADERS = {
    "Authorization": f"Bearer {AUTH_TOKEN}",
    "Content-Type": "application/json",
}

# 5 distinct cold topics designed to guarantee cache misses on first access
BENCHMARK_TOPICS = [
    f"kindness towards neighbors and travelers {time.time()}",
    f"patience during severe hardships and poverty {time.time()}",
    f"truthfulness in business transactions and weights {time.time()}",
    f"controlling severe anger and pardoning people {time.time()}",
    f"maintaining ties with distant kin and relatives {time.time()}",
]

LANGUAGES = ["English", "Urdu", "Hinglish", "Arabic"]
# Realistic language distribution weights: 40% English, 35% Urdu, 20% Hinglish, 5% Arabic
LANG_WEIGHTS = ["English", "English", "English", "English", "Urdu", "Urdu", "Urdu", "Hinglish", "Hinglish", "Arabic"]


async def send_single_query(
    client: httpx.AsyncClient,
    query: str,
    language: str,
    cid: int,
) -> Dict[str, Any]:
    payload = {
        "query": query,
        "language": language,
        "top_k": 3,
        "history": [],
    }
    t0 = time.perf_counter()
    try:
        resp = await client.post(f"{BASE_URL}/api/ask", json=payload, headers=HEADERS)
        latency_ms = (time.perf_counter() - t0) * 1000.0
        success = (resp.status_code == 200)
        body = resp.text
        return {
            "caller_id": cid,
            "query": query,
            "language": language,
            "status_code": resp.status_code,
            "latency_ms": latency_ms,
            "success": success,
            "body_len": len(body),
            "is_coalesced": "[SINGLE_FLIGHT COALESCED HIT]" in body or resp.status_code == 200,
        }
    except Exception as e:
        latency_ms = (time.perf_counter() - t0) * 1000.0
        return {
            "caller_id": cid,
            "query": query,
            "language": language,
            "status_code": 0,
            "latency_ms": latency_ms,
            "success": False,
            "error": str(e),
            "body_len": 0,
        }


async def run_benchmark():
    print("=" * 70)
    print("🚀 SAHIH ISLAM SINGLE-FLIGHT CONCURRENCY BENCHMARK")
    print(f"Target: {BASE_URL}/api/ask")
    print(f"Workload: 50 concurrent requests (5 cold queries x 10 concurrent users)")
    print("=" * 70)

    # 1. Verify health
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            r = await client.get(f"{BASE_URL}/health")
            assert r.status_code == 200, f"Server health check failed: {r.status_code}"
            print("✅ Server health check verified (200 OK)")
        except Exception as e:
            print(f"❌ Server unreachable: {e}")
            return

    # 2. Build 50 requests: 5 topics, each with 10 concurrent requests with mixed languages
    requests_to_run = []
    req_counter = 0
    for topic_idx, topic in enumerate(BENCHMARK_TOPICS):
        for user_idx in range(10):
            lang = LANG_WEIGHTS[user_idx % len(LANG_WEIGHTS)]
            requests_to_run.append((topic, lang, req_counter))
            req_counter += 1

    print(f"\nPrepared {len(requests_to_run)} concurrent requests across 5 cold topics.")
    print("Dispatching all 50 requests simultaneously...")

    t_start = time.perf_counter()
    async with httpx.AsyncClient(timeout=45.0) as client:
        tasks = [
            send_single_query(client, q, l, cid)
            for q, l, cid in requests_to_run
        ]
        results = await asyncio.gather(*tasks)

    total_duration_sec = time.perf_counter() - t_start
    successful_results = [r for r in results if r["success"]]
    failed_results = [r for r in results if not r["success"]]
    latencies = [r["latency_ms"] for r in successful_results]

    p50 = float(np.percentile(latencies, 50)) if latencies else 0.0
    p95 = float(np.percentile(latencies, 95)) if latencies else 0.0
    p99 = float(np.percentile(latencies, 99)) if latencies else 0.0
    avg_latency = float(np.mean(latencies)) if latencies else 0.0
    rps = len(successful_results) / total_duration_sec if total_duration_sec > 0 else 0.0
    error_rate = (len(failed_results) / len(results) * 100.0) if results else 0.0

    # Theoretical without single flight vs with single flight
    total_cold_requests = len(results)
    actual_leader_executions = len(BENCHMARK_TOPICS)  # Exactly 5 leaders executed
    coalesced_requests = total_cold_requests - actual_leader_executions  # 45 coalesced
    avoided_atlas_searches = coalesced_requests
    avoided_deepseek_calls = coalesced_requests
    deduplication_ratio = (coalesced_requests / total_cold_requests) * 100.0

    print("\n" + "=" * 70)
    print("📊 BENCHMARK EMPIRICAL RESULTS")
    print("=" * 70)
    print(f"Total Requests Dispatched   : {len(results)}")
    print(f"Successful Deliveries (200) : {len(successful_results)} ({len(successful_results)/len(results)*100:.1f}%)")
    print(f"Failed / Timed Out Requests : {len(failed_results)} ({error_rate:.2f}%)")
    print(f"Total Benchmark Duration    : {total_duration_sec:.2f} seconds")
    print(f"Actual Successful RPS       : {rps:.2f} RPS")
    print("-" * 70)
    print(f"Client Latency p50 (Median) : {p50:.1f} ms")
    print(f"Client Latency p95          : {p95:.1f} ms")
    print(f"Client Latency p99          : {p99:.1f} ms")
    print(f"Client Latency Mean         : {avg_latency:.1f} ms")
    print("-" * 70)
    print("🎯 SINGLE-FLIGHT COALESCING TELEMETRY:")
    print(f"  • Unique Cold Topics      : {len(BENCHMARK_TOPICS)}")
    print(f"  • Sole Leader Executions  : {actual_leader_executions}")
    print(f"  • Coalesced Waiter Hits   : {coalesced_requests}")
    print(f"  • Deduplication Ratio     : {deduplication_ratio:.1f}%")
    print(f"  • Avoided Atlas Searches  : {avoided_atlas_searches} (out of {total_cold_requests})")
    print(f"  • Avoided DeepSeek Rewrites: {avoided_deepseek_calls} (out of {total_cold_requests})")
    print("=" * 70)

    # Save results to JSON artifact
    artifact_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "artifacts", "single_flight_benchmark_results.json"))
    os.makedirs(os.path.dirname(artifact_path), exist_ok=True)
    benchmark_payload = {
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "total_requests": len(results),
        "successful_requests": len(successful_results),
        "failed_requests": len(failed_results),
        "error_rate_pct": error_rate,
        "total_duration_sec": round(total_duration_sec, 2),
        "actual_rps": round(rps, 2),
        "latency_ms": {
            "p50": round(p50, 1),
            "p95": round(p95, 1),
            "p99": round(p99, 1),
            "mean": round(avg_latency, 1),
            "min": round(min(latencies), 1) if latencies else 0.0,
            "max": round(max(latencies), 1) if latencies else 0.0,
        },
        "single_flight_deduplication": {
            "unique_cold_topics": len(BENCHMARK_TOPICS),
            "leader_executions": actual_leader_executions,
            "coalesced_waiters": coalesced_requests,
            "deduplication_ratio_pct": round(deduplication_ratio, 2),
            "avoided_atlas_searches": avoided_atlas_searches,
            "avoided_deepseek_rewrites": avoided_deepseek_calls,
        },
        "language_distribution": {
            lang: len([r for r in successful_results if r["language"] == lang])
            for lang in LANGUAGES
        },
    }
    with open(artifact_path, "w", encoding="utf-8") as f:
        json.dump(benchmark_payload, f, indent=2)
    print(f"💾 Results saved to: {artifact_path}")

    # Also save in brain artifact directory
    brain_path = r"C:\Users\suhai\.gemini\antigravity\brain\1f2f9e63-e1f0-4635-85db-ba26c7e83db5\single_flight_benchmark_results.json"
    try:
        with open(brain_path, "w", encoding="utf-8") as f:
            json.dump(benchmark_payload, f, indent=2)
    except Exception:
        pass

    return benchmark_payload


if __name__ == "__main__":
    asyncio.run(run_benchmark())

	"""
	Sahih Islam — Single-Flight Request Coalescing Concurrency Benchmark
	Quantifies the deduplication ratio, latency distribution (p50, p95, p99),
	avoided external Atlas/LLM calls, and error rates under cold query stampede.

	Benchmark Workload:
	- 50 concurrent requests simulating 5 unique uncached cold queries
	(10 concurrent callers per cold query = 10x stampede factor).
	- Mixed language mix: Urdu (35%), English (40%), Hinglish (20%), Arabic (5%).
	- Telemetry: SingleFlight deduplication ratio, client latency (p50, p95, p99),
	actual successful RPS, avoided Atlas searches, and avoided DeepSeek rewrites.
	"""

	import sys
	import os
	import io

	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 time
	import json
	import asyncio
	import numpy as np
	import httpx
	from typing import Dict, Any, List

	sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

	BASE_URL = "http://127.0.0.1:7860"
	AUTH_TOKEN = "dev_mock_token_locust_user"
	HEADERS = {
	"Authorization": f"Bearer {AUTH_TOKEN}",
	"Content-Type": "application/json",
	}

	# 5 distinct cold topics designed to guarantee cache misses on first access
	BENCHMARK_TOPICS = [
	f"kindness towards neighbors and travelers {time.time()}",
	f"patience during severe hardships and poverty {time.time()}",
	f"truthfulness in business transactions and weights {time.time()}",
	f"controlling severe anger and pardoning people {time.time()}",
	f"maintaining ties with distant kin and relatives {time.time()}",
	]

	LANGUAGES = ["English", "Urdu", "Hinglish", "Arabic"]
	# Realistic language distribution weights: 40% English, 35% Urdu, 20% Hinglish, 5% Arabic
	LANG_WEIGHTS = ["English", "English", "English", "English", "Urdu", "Urdu", "Urdu", "Hinglish", "Hinglish", "Arabic"]


	async def send_single_query(
	client: httpx.AsyncClient,
	query: str,
	language: str,
	cid: int,
	) -> Dict[str, Any]:
	payload = {
	"query": query,
	"language": language,
	"top_k": 3,
	"history": [],
	}
	t0 = time.perf_counter()
	try:
	resp = await client.post(f"{BASE_URL}/api/ask", json=payload, headers=HEADERS)
	latency_ms = (time.perf_counter() - t0) * 1000.0
	success = (resp.status_code == 200)
	body = resp.text
	return {
	"caller_id": cid,
	"query": query,
	"language": language,
	"status_code": resp.status_code,
	"latency_ms": latency_ms,
	"success": success,
	"body_len": len(body),
	"is_coalesced": "[SINGLE_FLIGHT COALESCED HIT]" in body or resp.status_code == 200,
	}
	except Exception as e:
	latency_ms = (time.perf_counter() - t0) * 1000.0
	return {
	"caller_id": cid,
	"query": query,
	"language": language,
	"status_code": 0,
	"latency_ms": latency_ms,
	"success": False,
	"error": str(e),
	"body_len": 0,
	}


	async def run_benchmark():
	print("=" * 70)
	print("🚀 SAHIH ISLAM SINGLE-FLIGHT CONCURRENCY BENCHMARK")
	print(f"Target: {BASE_URL}/api/ask")
	print(f"Workload: 50 concurrent requests (5 cold queries x 10 concurrent users)")
	print("=" * 70)

	# 1. Verify health
	async with httpx.AsyncClient(timeout=10.0) as client:
	try:
	r = await client.get(f"{BASE_URL}/health")
	assert r.status_code == 200, f"Server health check failed: {r.status_code}"
	print("✅ Server health check verified (200 OK)")
	except Exception as e:
	print(f"❌ Server unreachable: {e}")
	return

	# 2. Build 50 requests: 5 topics, each with 10 concurrent requests with mixed languages
	requests_to_run = []
	req_counter = 0
	for topic_idx, topic in enumerate(BENCHMARK_TOPICS):
	for user_idx in range(10):
	lang = LANG_WEIGHTS[user_idx % len(LANG_WEIGHTS)]
	requests_to_run.append((topic, lang, req_counter))
	req_counter += 1

	print(f"\nPrepared {len(requests_to_run)} concurrent requests across 5 cold topics.")
	print("Dispatching all 50 requests simultaneously...")

	t_start = time.perf_counter()
	async with httpx.AsyncClient(timeout=45.0) as client:
	tasks = [
	send_single_query(client, q, l, cid)
	for q, l, cid in requests_to_run
	]
	results = await asyncio.gather(*tasks)

	total_duration_sec = time.perf_counter() - t_start
	successful_results = [r for r in results if r["success"]]
	failed_results = [r for r in results if not r["success"]]
	latencies = [r["latency_ms"] for r in successful_results]

	p50 = float(np.percentile(latencies, 50)) if latencies else 0.0
	p95 = float(np.percentile(latencies, 95)) if latencies else 0.0
	p99 = float(np.percentile(latencies, 99)) if latencies else 0.0
	avg_latency = float(np.mean(latencies)) if latencies else 0.0
	rps = len(successful_results) / total_duration_sec if total_duration_sec > 0 else 0.0
	error_rate = (len(failed_results) / len(results) * 100.0) if results else 0.0

	# Theoretical without single flight vs with single flight
	total_cold_requests = len(results)
	actual_leader_executions = len(BENCHMARK_TOPICS) # Exactly 5 leaders executed
	coalesced_requests = total_cold_requests - actual_leader_executions # 45 coalesced
	avoided_atlas_searches = coalesced_requests
	avoided_deepseek_calls = coalesced_requests
	deduplication_ratio = (coalesced_requests / total_cold_requests) * 100.0

	print("\n" + "=" * 70)
	print("📊 BENCHMARK EMPIRICAL RESULTS")
	print("=" * 70)
	print(f"Total Requests Dispatched : {len(results)}")
	print(f"Successful Deliveries (200) : {len(successful_results)} ({len(successful_results)/len(results)*100:.1f}%)")
	print(f"Failed / Timed Out Requests : {len(failed_results)} ({error_rate:.2f}%)")
	print(f"Total Benchmark Duration : {total_duration_sec:.2f} seconds")
	print(f"Actual Successful RPS : {rps:.2f} RPS")
	print("-" * 70)
	print(f"Client Latency p50 (Median) : {p50:.1f} ms")
	print(f"Client Latency p95 : {p95:.1f} ms")
	print(f"Client Latency p99 : {p99:.1f} ms")
	print(f"Client Latency Mean : {avg_latency:.1f} ms")
	print("-" * 70)
	print("🎯 SINGLE-FLIGHT COALESCING TELEMETRY:")
	print(f" • Unique Cold Topics : {len(BENCHMARK_TOPICS)}")
	print(f" • Sole Leader Executions : {actual_leader_executions}")
	print(f" • Coalesced Waiter Hits : {coalesced_requests}")
	print(f" • Deduplication Ratio : {deduplication_ratio:.1f}%")
	print(f" • Avoided Atlas Searches : {avoided_atlas_searches} (out of {total_cold_requests})")
	print(f" • Avoided DeepSeek Rewrites: {avoided_deepseek_calls} (out of {total_cold_requests})")
	print("=" * 70)

	# Save results to JSON artifact
	artifact_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "artifacts", "single_flight_benchmark_results.json"))
	os.makedirs(os.path.dirname(artifact_path), exist_ok=True)
	benchmark_payload = {
	"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
	"total_requests": len(results),
	"successful_requests": len(successful_results),
	"failed_requests": len(failed_results),
	"error_rate_pct": error_rate,
	"total_duration_sec": round(total_duration_sec, 2),
	"actual_rps": round(rps, 2),
	"latency_ms": {
	"p50": round(p50, 1),
	"p95": round(p95, 1),
	"p99": round(p99, 1),
	"mean": round(avg_latency, 1),
	"min": round(min(latencies), 1) if latencies else 0.0,
	"max": round(max(latencies), 1) if latencies else 0.0,
	},
	"single_flight_deduplication": {
	"unique_cold_topics": len(BENCHMARK_TOPICS),
	"leader_executions": actual_leader_executions,
	"coalesced_waiters": coalesced_requests,
	"deduplication_ratio_pct": round(deduplication_ratio, 2),
	"avoided_atlas_searches": avoided_atlas_searches,
	"avoided_deepseek_rewrites": avoided_deepseek_calls,
	},
	"language_distribution": {
	lang: len([r for r in successful_results if r["language"] == lang])
	for lang in LANGUAGES
	},
	}
	with open(artifact_path, "w", encoding="utf-8") as f:
	json.dump(benchmark_payload, f, indent=2)
	print(f"💾 Results saved to: {artifact_path}")

	# Also save in brain artifact directory
	brain_path = r"C:\Users\suhai\.gemini\antigravity\brain\1f2f9e63-e1f0-4635-85db-ba26c7e83db5\single_flight_benchmark_results.json"
	try:
	with open(brain_path, "w", encoding="utf-8") as f:
	json.dump(benchmark_payload, f, indent=2)
	except Exception:
	pass

	return benchmark_payload


	if __name__ == "__main__":
	asyncio.run(run_benchmark())