"""
Ch8 配套代码 2 / 3 —— 测量「主写入 → 从读到」的延迟

⚠️ 真实主从环境用法：
    把 MASTER_HOST/PORT 指向主，REPLICA_HOST/PORT 指向从，
    脚本会真实测量主写入到从读到的延迟（毫秒级）。

⚠️ 单机环境用法（教程默认）：
    脚本检测到 MASTER == REPLICA 时，会自动用「镜像 key + 后台同步线程」
    模拟主从异步复制：
        主键   :  demo:lag:master   ← 模拟「写主」
        镜像键 :  demo:lag:replica  ← 模拟「读从」
        线程    :  每 SYNC_LAG_MS 毫秒把主键复制到镜像键
    这样能在单机上重现「写完立即读读到旧值 / 延迟分布」的现象。

测量方式：
    主：SET demo:lag:master "<纳秒时间戳>"
    从：循环 GET demo:lag:replica，直到值等于刚才写入的时间戳
    延迟 = 当前时间 - 时间戳
"""

import threading
import time
import statistics
import redis

MASTER_HOST,  MASTER_PORT  = "127.0.0.1", 6379
REPLICA_HOST, REPLICA_PORT = "127.0.0.1", 6379

KEY_MASTER  = "demo:lag:master"
KEY_REPLICA = "demo:lag:replica"

ROUNDS         = 30
WRITE_INTERVAL = 0.1            # 每 100 ms 写一次
POLL_INTERVAL  = 0.001          # 从节点轮询读，1 ms 间隔
POLL_TIMEOUT   = 2.0            # 2 秒还读不到就判超时
SYNC_LAG_MS    = 30             # 单机 mock 模式下模拟的同步延迟


master = redis.Redis(host=MASTER_HOST, port=MASTER_PORT, decode_responses=True)
replica = redis.Redis(host=REPLICA_HOST, port=REPLICA_PORT, decode_responses=True)

stop_event = threading.Event()


def section(title: str) -> None:
    print("\n" + "=" * 70)
    print(title)
    print("=" * 70)


def is_single_instance() -> bool:
    return (MASTER_HOST, MASTER_PORT) == (REPLICA_HOST, REPLICA_PORT)


def mock_replica_worker():
    """单机环境：模拟一个有 SYNC_LAG_MS 延迟的「主→从复制管道」。"""
    while not stop_event.is_set():
        v = master.get(KEY_MASTER)
        if v is not None:
            replica.set(KEY_REPLICA, v)
        time.sleep(SYNC_LAG_MS / 1000.0)


def measure_one_round() -> float:
    """一轮：写主 → 轮询读从 → 返回延迟（秒），超时返回 -1。"""
    ts = str(time.perf_counter_ns())
    write_at = time.perf_counter()
    master.set(KEY_MASTER, ts)

    deadline = write_at + POLL_TIMEOUT
    while time.perf_counter() < deadline:
        if replica.get(KEY_REPLICA) == ts:
            return time.perf_counter() - write_at
        time.sleep(POLL_INTERVAL)
    return -1.0


def run_demo():
    section("Demo: 测量主从复制延迟")

    if is_single_instance():
        print("  ⚠️  检测到单机环境（MASTER == REPLICA），")
        print(f"      启动后台线程模拟一个 ~{SYNC_LAG_MS}ms 同步延迟的「主→从管道」")
        t = threading.Thread(target=mock_replica_worker, daemon=True)
        t.start()
        # 等同步线程对齐一次
        master.set(KEY_MASTER, "init")
        time.sleep(SYNC_LAG_MS / 1000.0 * 2)
    else:
        print(f"  ✅ 真实主从模式：master={MASTER_HOST}:{MASTER_PORT}, "
              f"replica={REPLICA_HOST}:{REPLICA_PORT}")

    print(f"  共 {ROUNDS} 轮，每轮间隔 {int(WRITE_INTERVAL*1000)}ms")
    print(f"\n  {'轮':>4} | {'延迟 (ms)':>10} | 状态")
    print(f"  {'-'*4}-+-{'-'*10}-+-{'-'*20}")

    samples = []
    timeouts = 0
    for i in range(1, ROUNDS + 1):
        lag = measure_one_round()
        if lag < 0:
            print(f"  {i:>4} | {'TIMEOUT':>10} | ❌ 超过 {POLL_TIMEOUT}s 还没同步过来")
            timeouts += 1
        else:
            samples.append(lag * 1000)
            tag = "✅" if lag * 1000 < 100 else ("⚠️" if lag * 1000 < 500 else "🔥")
            print(f"  {i:>4} | {lag*1000:>10.2f} | {tag}")
        time.sleep(WRITE_INTERVAL)

    stop_event.set()

    section("结果统计")
    print(f"  总轮数: {ROUNDS}, 超时: {timeouts}, 有效样本: {len(samples)}")
    if samples:
        print(f"  min    = {min(samples):.2f} ms")
        print(f"  max    = {max(samples):.2f} ms")
        print(f"  mean   = {statistics.mean(samples):.2f} ms")
        print(f"  median = {statistics.median(samples):.2f} ms")
        if len(samples) >= 2:
            print(f"  stdev  = {statistics.stdev(samples):.2f} ms")
        # 简易 p95
        sorted_s = sorted(samples)
        p95 = sorted_s[max(0, int(len(sorted_s) * 0.95) - 1)]
        print(f"  p95    = {p95:.2f} ms")

    print("\n  💡 关键结论：")
    print("     - 真实同机房主从延迟通常 < 1ms；跨机房可能十几 ms")
    print("     - 写主后立即读从 → 落在「延迟窗口」内就读到旧值")
    print("     - 业务对一致性敏感 → 写后短时间路由到主 / 用 WAIT / 关闭读写分离")

    master.delete(KEY_MASTER, KEY_REPLICA)


if __name__ == "__main__":
    try:
        run_demo()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败: {e}")