"""
Ch11 配套代码 1 / 4 —— 分布式锁演进史 V1 / V2 / V3

本脚本演示三个版本锁实现的优缺点：
  V1: SETNX + DEL          —— 客户端崩溃 → 永久死锁
  V2: SETNX + EXPIRE 两步  —— 两步之间崩溃 → 仍然死锁
  V3: SET key val NX EX    —— 一条原子命令，正确实现 ✅

每个版本都会在 Redis 上真实模拟「持锁线程崩溃」的场景，
并打印其他线程能否在合理时间内拿到锁。
"""

import time
import threading
import uuid

try:
    import redis
except ImportError:
    print("请先 pip install redis")
    raise SystemExit(1)

POOL = redis.ConnectionPool(host="127.0.0.1", port=6379, decode_responses=True)


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


def reset(key: str) -> None:
    redis.Redis(connection_pool=POOL).delete(key)


# -------------------------------------------------------------
# V1: SETNX + DEL  —— 致命缺陷：客户端崩了锁永远不释放
# -------------------------------------------------------------
def v1_acquire(r, key: str) -> bool:
    return r.setnx(key, "1") == 1


def v1_release(r, key: str) -> None:
    r.delete(key)


def demo_v1() -> None:
    section("V1: SETNX + DEL  →  崩溃后死锁")
    key = "lock:v1"
    reset(key)
    r = redis.Redis(connection_pool=POOL)

    print("[A] 加锁")
    assert v1_acquire(r, key), "应当加锁成功"
    print("[A] 业务执行中... 模拟进程崩溃（不调 release）")

    # 模拟 A 崩溃，不释放锁

    print("[B] 尝试加锁（最多 3 秒）")
    start = time.time()
    while time.time() - start < 3:
        if v1_acquire(r, key):
            print("[B] 拿到锁")
            return
        time.sleep(0.2)
    print(f"[B] ❌ 等了 {time.time()-start:.1f}s 仍拿不到 —— 死锁演示成功")
    print("    （需要人工 DEL，否则永久占用）")
    reset(key)


# -------------------------------------------------------------
# V2: SETNX + EXPIRE 分两步 —— 两步之间崩仍死锁
# -------------------------------------------------------------
def v2_acquire(r, key: str, ttl: int, *, simulate_crash_between: bool = False) -> bool:
    if r.setnx(key, "1") != 1:
        return False
    if simulate_crash_between:
        print("[A] 💥 SETNX 之后、EXPIRE 之前崩溃")
        raise SystemExit  # 模拟进程死亡
    r.expire(key, ttl)
    return True


def v2_release(r, key: str) -> None:
    r.delete(key)


def demo_v2() -> None:
    section("V2: SETNX + EXPIRE 两步  →  两步之间崩溃仍死锁")
    key = "lock:v2"
    reset(key)
    r = redis.Redis(connection_pool=POOL)

    def crashed_acquire():
        try:
            v2_acquire(r, key, ttl=5, simulate_crash_between=True)
        except SystemExit:
            pass  # 子线程"崩溃"

    t = threading.Thread(target=crashed_acquire)
    t.start(); t.join()

    ttl = r.ttl(key)
    print(f"[A] 崩溃后，锁的 TTL = {ttl}  （-1 = 永不过期，死锁！）")

    print("[B] 尝试加锁（最多 3 秒）")
    start = time.time()
    while time.time() - start < 3:
        if v2_acquire(r, key, ttl=5):
            print("[B] 拿到锁")
            return
        time.sleep(0.2)
    print(f"[B] ❌ 等了 {time.time()-start:.1f}s 仍拿不到 —— 仍然死锁")
    reset(key)


# -------------------------------------------------------------
# V3: SET key val NX EX  —— 原子，推荐 ✅
# -------------------------------------------------------------
def v3_acquire(r, key: str, token: str, ttl: int) -> bool:
    return r.set(key, token, nx=True, ex=ttl) is True


def v3_release(r, key: str, token: str) -> int:
    """ 释放前校验 token，避免误删别人的锁；用 Lua 保证原子。"""
    lua = """
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
    else
        return 0
    end
    """
    return r.eval(lua, 1, key, token)


def demo_v3() -> None:
    section("V3: SET key val NX EX  →  正确实现 ✅")
    key = "lock:v3"
    reset(key)
    r = redis.Redis(connection_pool=POOL)
    token = uuid.uuid4().hex

    print(f"[A] 加锁 (token={token[:8]}, TTL=2s)")
    assert v3_acquire(r, key, token, ttl=2)
    print("[A] 模拟业务后崩溃（不释放）")
    # 不调 release —— 模拟进程死亡

    print("[B] 等待锁过期...")
    start = time.time()
    while time.time() - start < 5:
        b_token = uuid.uuid4().hex
        if v3_acquire(r, key, b_token, ttl=2):
            print(f"[B] ✅ {time.time()-start:.1f}s 后拿到锁，自动恢复！")
            v3_release(r, key, b_token)
            return
        time.sleep(0.3)
    print("[B] 异常：超时未拿到")


if __name__ == "__main__":
    try:
        redis.Redis(connection_pool=POOL).ping()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败 ({e})，请确保 127.0.0.1:6379 可用")
        raise SystemExit(1)

    demo_v1()
    demo_v2()
    demo_v3()
    print("\n✅ 三版对比完成。结论：永远使用 V3（SET ... NX EX ...）")