#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
获取密集库位处理器并发问题修复演示
展示优化前后的区别
"""

import asyncio
import time
import random
from typing import Dict, Any, List
from datetime import datetime


class MockDatabase:
    """模拟数据库"""
    def __init__(self):
        # 模拟5个空闲库位
        self.storage_sites = {
            f"A00{i}": {
                "is_locked": False,
                "is_occupied": False,
                "last_access": datetime.now()
            }
            for i in range(1, 6)
        }
        self.operation_log = []
    
    async def query_available_sites(self) -> List[str]:
        """查询可用库位"""
        await asyncio.sleep(0.01)  # 模拟数据库查询延迟
        available = [
            site_id for site_id, info in self.storage_sites.items()
            if not info["is_locked"] and not info["is_occupied"]
        ]
        return sorted(available)
    
    async def lock_site_old_way(self, site_id: str, task_id: str) -> bool:
        """旧方式：分别查询后锁定（有竞争条件）"""
        # 先查询
        available_sites = await self.query_available_sites()
        if site_id not in available_sites:
            return False
        
        # 模拟网络延迟和处理时间
        await asyncio.sleep(random.uniform(0.01, 0.05))
        
        # 尝试锁定（可能已被其他请求抢占）
        if self.storage_sites[site_id]["is_locked"]:
            return False
        
        self.storage_sites[site_id]["is_locked"] = True
        self.operation_log.append(f"[OLD] Task {task_id} locked {site_id}")
        return True
    
    async def lock_site_new_way(self, site_id: str, task_id: str) -> bool:
        """新方式：原子性锁定（无竞争条件）"""
        # 原子操作：检查并锁定
        if not self.storage_sites[site_id]["is_locked"] and not self.storage_sites[site_id]["is_occupied"]:
            self.storage_sites[site_id]["is_locked"] = True
            self.operation_log.append(f"[NEW] Task {task_id} locked {site_id}")
            return True
        return False
    
    def reset(self):
        """重置数据库状态"""
        for site_info in self.storage_sites.values():
            site_info["is_locked"] = False
        self.operation_log.clear()


async def simulate_old_approach(db: MockDatabase, num_tasks: int) -> Dict[str, Any]:
    """模拟旧的实现方式"""
    print(f"\n=== 模拟旧方式：查询后锁定（存在竞争条件）===")
    
    async def old_task(task_id: int):
        # 查询可用库位
        available_sites = await db.query_available_sites()
        if not available_sites:
            return {"task_id": task_id, "success": False, "reason": "no_sites"}
        
        # 选择第一个可用库位
        target_site = available_sites[0]
        
        # 尝试锁定
        success = await db.lock_site_old_way(target_site, f"task_{task_id}")
        return {
            "task_id": task_id, 
            "success": success, 
            "site_id": target_site if success else None,
            "reason": "success" if success else "race_condition"
        }
    
    # 重置数据库
    db.reset()
    start_time = time.time()
    
    # 并发执行任务
    tasks = [old_task(i) for i in range(num_tasks)]
    results = await asyncio.gather(*tasks)
    
    end_time = time.time()
    
    # 分析结果
    successful = [r for r in results if r["success"]]
    failed = [r for r in results if not r["success"]]
    
    # 检查是否有重复分配
    allocated_sites = [r["site_id"] for r in successful if r["site_id"]]
    duplicates = len(allocated_sites) - len(set(allocated_sites))
    
    print(f"执行时间: {end_time - start_time:.3f}s")
    print(f"成功任务: {len(successful)}")
    print(f"失败任务: {len(failed)}")
    print(f"重复分配: {duplicates} 个库位被重复分配")
    
    if duplicates > 0:
        print("❌ 发现并发竞争问题：有库位被重复分配！")
    else:
        print("✓ 未发现重复分配")
    
    print("操作日志:")
    for log in db.operation_log[:10]:  # 只显示前10条
        print(f"  {log}")
    
    return {
        "total_tasks": num_tasks,
        "successful": len(successful),
        "failed": len(failed),
        "duplicates": duplicates,
        "execution_time": end_time - start_time
    }


async def simulate_new_approach(db: MockDatabase, num_tasks: int) -> Dict[str, Any]:
    """模拟新的实现方式"""
    print(f"\n=== 模拟新方式：原子性获取（无竞争条件）===")
    
    async def new_task(task_id: int):
        # 查询可用库位
        available_sites = await db.query_available_sites()
        if not available_sites:
            return {"task_id": task_id, "success": False, "reason": "no_sites"}
        
        # 按优先级尝试原子性锁定每个库位
        for site_id in available_sites:
            success = await db.lock_site_new_way(site_id, f"task_{task_id}")
            if success:
                return {
                    "task_id": task_id, 
                    "success": True, 
                    "site_id": site_id,
                    "reason": "success"
                }
        
        return {"task_id": task_id, "success": False, "reason": "all_occupied"}
    
    # 重置数据库
    db.reset()
    start_time = time.time()
    
    # 并发执行任务
    tasks = [new_task(i) for i in range(num_tasks)]
    results = await asyncio.gather(*tasks)
    
    end_time = time.time()
    
    # 分析结果
    successful = [r for r in results if r["success"]]
    failed = [r for r in results if not r["success"]]
    
    # 检查是否有重复分配
    allocated_sites = [r["site_id"] for r in successful if r["site_id"]]
    duplicates = len(allocated_sites) - len(set(allocated_sites))
    
    print(f"执行时间: {end_time - start_time:.3f}s")
    print(f"成功任务: {len(successful)}")
    print(f"失败任务: {len(failed)}")
    print(f"重复分配: {duplicates} 个库位被重复分配")
    
    if duplicates > 0:
        print("❌ 仍然存在并发竞争问题！")
    else:
        print("✓ 成功解决并发竞争问题")
    
    print("操作日志:")
    for log in db.operation_log[:10]:  # 只显示前10条
        print(f"  {log}")
    
    return {
        "total_tasks": num_tasks,
        "successful": len(successful),
        "failed": len(failed),
        "duplicates": duplicates,
        "execution_time": end_time - start_time
    }


async def run_comparison_demo():
    """运行对比演示"""
    print("🔧 获取密集库位处理器并发问题修复演示")
    print("=" * 60)
    
    db = MockDatabase()
    num_concurrent_tasks = 10  # 10个并发任务竞争5个库位
    
    print(f"测试场景: {num_concurrent_tasks} 个并发任务竞争 {len(db.storage_sites)} 个库位")
    
    # 测试旧方式
    old_results = await simulate_old_approach(db, num_concurrent_tasks)
    
    # 测试新方式
    new_results = await simulate_new_approach(db, num_concurrent_tasks)
    
    # 对比结果
    print(f"\n=== 对比结果 ===")
    print(f"旧方式 - 成功: {old_results['successful']}, 失败: {old_results['failed']}, 重复分配: {old_results['duplicates']}")
    print(f"新方式 - 成功: {new_results['successful']}, 失败: {new_results['failed']}, 重复分配: {new_results['duplicates']}")
    
    if old_results['duplicates'] > 0 and new_results['duplicates'] == 0:
        print("🎉 优化成功！新方式完全解决了并发竞争问题")
    elif old_results['duplicates'] == new_results['duplicates'] == 0:
        print("ℹ️  在当前测试条件下，两种方式都没有出现竞争问题")
        print("   可能需要增加并发压力或调整延迟来复现问题")
    else:
        print("⚠️  优化可能不够完善，需要进一步检查")
    
    # 性能对比
    performance_diff = ((new_results['execution_time'] - old_results['execution_time']) / old_results['execution_time']) * 100
    if performance_diff > 0:
        print(f"📊 性能: 新方式比旧方式慢 {performance_diff:.1f}%")
    else:
        print(f"📊 性能: 新方式比旧方式快 {abs(performance_diff):.1f}%")


async def stress_test():
    """压力测试"""
    print(f"\n=== 压力测试 ===")
    print("高并发场景下测试并发安全性...")
    
    db = MockDatabase()
    num_tasks = 50  # 50个并发任务
    
    print(f"压力测试: {num_tasks} 个高并发任务")
    
    # 只测试新方式的压力承受能力
    results = await simulate_new_approach(db, num_tasks)
    
    success_rate = (results['successful'] / results['total_tasks']) * 100
    print(f"成功率: {success_rate:.1f}%")
    print(f"平均每任务耗时: {results['execution_time']/results['total_tasks']*1000:.2f}ms")
    
    if results['duplicates'] == 0:
        print("✅ 压力测试通过：无并发竞争问题")
    else:
        print("❌ 压力测试失败：仍存在并发竞争问题")


if __name__ == "__main__":
    asyncio.run(run_comparison_demo())
    asyncio.run(stress_test())