7. 并发模型、线程与内存基础设施

// 多线程架构的死锁场景
线程1: 玩家A.attack(玩家B)
  └── 需要获取 B 的锁
  └── 等待 B 的锁释放

线程2: 玩家B.attack(玩家A)
  └── 需要获取 A 的锁
  └── 等待 A 的锁释放

结果: 死锁！

// 单线程架构
主线程: A.attack(B) → B.attack(A)
  └── 串行执行，无死锁风险

// AOI (Area of Interest) 事件传播
玩家进入新区域:
1. 查询周围玩家（遍历 Entity 列表，需要全局锁）
2. 向每个玩家发送"新玩家进入"消息（跨线程通信）
3. 每个玩家创建对新玩家的引用（引用计数同步）

多线程实现的问题:
- 需要全局 AOI 锁 → 性能瓶颈
- 消息队列复杂 → 跨线程通信开销
- 引用计数同步 → 原子操作开销（每次 +10-50ns）

单线程实现:
- 直接遍历列表 → 2-3ns
- 直接发送消息 → 无同步开销
- 直接操作引用 → 无需原子操作

// 同一 tick 内的操作顺序影响游戏逻辑

Tick #1000 内的操作序列1:
1. 玩家A拾取宝剑（攻击力+10）
2. 玩家A攻击怪物（造成100伤害）

Tick #1000 内的操作序列2:
1. 玩家A攻击怪物（造成90伤害）
2. 玩家A拾取宝剑

多线程执行 → 顺序不可预测 → 游戏结果不一致 → 玩家投诉

单线程执行 → 顺序确定 → 结果一致 → 公平性保障

// 实测：10000 个实体，每秒 100000 次攻击操作

单线程执行:
  - 吞吐量: 95000 次/秒
  - CPU 使用: 65%
  - 延迟: P99 = 5ms
  - 缓存命中率: 95% (都在 L1 缓存)

多线程 + 细粒度锁:
  - 吞吐量: 70000 次/秒 (↓26%)
  - CPU 使用: 95% (锁竞争)
  - 延迟: P99 = 50ms (↑10倍)
  - 缓存命中率: 60% (缓存失效)

多线程 + 无锁 (CAS):
  - 吞吐量: 85000 次/秒 (↓11%)
  - CPU 使用: 85% (CAS 重试 + 缓存失效)
  - 延迟: P99 = 20ms (↑4倍)
  - 缓存命中率: 75% (缓存一致性开销)

结论: 单线程最快，多线程反而更慢

// 多核 CPU 的缓存一致性问题

线程0 (核0): entity->hp += damage
  └── 修改 L1 缓存中的 entity
  └── MESI 协议通知其他核缓存失效
  └── 延迟：10-100ns

线程1 (核1): entity->hp -= heal
  └── 等待缓存同步
  └── 性能损失严重

// 单线程实现
主线程: entity->hp += damage; entity->hp -= heal;
  └── 都在 L1 缓存
  └── 无缓存同步
  └── 延迟：2-3ns

// 无锁链表的 ABA 问题
初始状态: A → B → C

线程1: 读取头节点 A
线程2: 删除 A，插入新节点 D (复用了 A 的内存地址)
线程1: CAS(A, D) → 成功！但链表已损坏

// 游戏服务器中的 ABA 问题
玩家列表: [玩家A, 玩家B, 玩家C]
线程1: 读取玩家A的引用
线程2: 玩家A下线，新玩家E复用了A的内存地址
线程1: CAS操作成功，但实际操作了错误的玩家

解决方案: Hazard Pointers 或 RCU (Read-Copy-Update)
代价: 实现复杂，性能开销大

// 无锁链表的内存回收问题

void remove(Node* node) {
    // 不能立即 delete，其他线程可能还在访问
    // 需要 Hazard Pointers 或延迟删除
    defer_delete(node, 100ms);
}

// 问题：
// 1. 内存占用高 (延迟删除的节点)
// 2. 实现复杂 (Hazard Pointers 需要追踪每个线程的访问)
// 3. 性能不如单线程 (单线程直接 delete)

// 单线程实现
void remove(Node* node) {
    delete node;  // 立即释放
}

// 游戏服务器的复合操作示例

void Entity::attack(Entity* target) {
    // 1. 检查距离 (访问 position 属性)
    if (!inRange(target)) return;

    // 2. 计算伤害 (访问多个属性：attack, defense, level...)
    int damage = calculateDamage(target);

    // 3. 扣血 (修改 target->hp)
    target->hp -= damage;

    // 4. 检查死亡 (可能触发其他逻辑)
    if (target->hp <= 0) {
        target->die();
        onEntityKilled(target);  // 掉落、任务、成就...
    }
}

// 无锁实现的问题：
// - 需要多次 CAS 操作
// - 中间状态可能被其他线程看到 (hp 扣了但还没死)
// - 回滚逻辑复杂
// - 最终性能不如单线程

// 单线程实现 (100 行，1 天开发)
void Entity::attack(Entity* target) {
    target->hp -= damage;
    broadcastDamage(target, damage);
    if (target->hp <= 0) {
        target->die();
    }
}

// 无锁实现 (1000+ 行，2 周开发)
class LockFreeEntity {
    std::atomic<int> hp_;
    std::atomic<int> defense_;
    std::atomic<bool> is_dead_;

    void attack(Entity* target) {
        // CAS 循环
        int old_hp, new_hp;
        do {
            old_hp = target->hp_.load(std::memory_order_relaxed);
            if (old_hp <= 0) return;  // 可能已经死亡

            int damage = calculateDamage();
            new_hp = old_hp - damage;
        } while (!target->hp_.compare_exchange_weak(
            old_hp, new_hp,
            std::memory_order_release,
            std::memory_order_relaxed));

        // 检查是否死亡 (又是 CAS)
        if (new_hp <= 0) {
            bool expected = false;
            if (target->is_dead_.compare_exchange_strong(expected, true)) {
                onEntityKilled(target);
            }
        }
    }
};

// 维护成本对比：
// 单线程: 每次修改 1-2 小时
// 无锁: 每次修改 1-2 天，需要重新验证并发正确性
// 并发 bug 复现困难，调试成本高

// 有状态游戏（长连接架构）
class GameServer {
    // 维护所有在线玩家的状态在内存中
    std::map<int, Entity*> onlinePlayers_;

    void onAttack(Entity* attacker, Entity* target) {
        // 直接访问内存中的实体状态
        target->hp -= attacker->attack;

        // 实时广播给周围玩家
        broadcastDamage(target, attacker->attack);
    }
};

// 特点：
// - 维护玩家长连接
// - 实体状态在内存
// - 实时交互（毫秒级）
// - 必须单线程处理游戏逻辑


// 无状态游戏（HTTP API 架构）
@RestController
@RequestMapping("/api/battle")
public class BattleController {

    @PostMapping("/attack")
    public ResponseEntity<Result> attack(@RequestBody AttackRequest req) {
        // 1. 验证 token（无状态，JWT）
        User user = authService.verify(req.getToken());

        // 2. 从 DB 加载状态（每次请求都从 DB 读）
        BattleState state = dbService.loadBattleState(req.getBattleId());

        // 3. 执行战斗逻辑（纯计算，无共享状态）
        Result result = battleSystem.calculate(state, req.getAttack());

        // 4. 保存到 DB
        dbService.saveBattleState(req.getBattleId(), state);

        // 5. 返回结果
        return ResponseEntity.ok(result);
    }
};

// 特点：
// - 每个 HTTP 请求独立
// - 不维护玩家连接状态
// - 可以任意水平扩展
// - 非常适合多线程！

// 文件：kbe/src/lib/thread/threadpool.h（简化）
class ThreadPool
{
    bool addTask(TPTask* tptask);           // 投递任务
    virtual void onMainThreadTick();        // 主线程取回结果

    // 内部管理
    std::queue<TPTask*> bufferedTaskList_;  // 待处理队列
    std::list<TPTask*> finiTaskList_;       // 已完成列表
    std::list<TPThread*> busyThreadList_;   // 繁忙线程
    std::list<TPThread*> freeThreadList_;   // 空闲线程
    uint32 maxThreadCount_;                 // 上限
};

// 文件：kbe/src/lib/thread/threadtask.h
class TPTask : public Task
{
    enum TPTaskState
    {
        TPTASK_STATE_COMPLETED = 0,               // 任务完成，删除
        TPTASK_STATE_CONTINUE_MAINTHREAD = 1,     // 结果回主线程继续执行
        TPTASK_STATE_CONTINUE_CHILDTHREAD = 2,    // 继续在子线程执行
    };

    virtual TPTaskState presentMainThread() {
        return TPTASK_STATE_COMPLETED;
    }
};

// 文件：lib/cstdmf/bgtask_manager.hpp（简化）
class BackgroundTask : public SafeReferenceCount
{
    virtual void doBackgroundTask(TaskManager& mgr,
            BackgroundTaskThread* pThread) = 0;  // 子线程执行
    virtual void doMainThreadTask(TaskManager& mgr) {} // 主线程回调
};

class TaskManager
{
    void addBackgroundTask(BackgroundTaskPtr pTask, int priority);
    void tick();  // 主线程 tick，执行 fgTaskList_

    // 带信号量的优先级任务队列
    class BackgroundTaskList {
        BW::list<std::pair<int, BackgroundTaskPtr>> list_;
        SimpleSemaphore semaphore_;
    };
};

// 1. 数据库操作
class DBQueryTask : public TPTask {
    void process() override {
        // 在工作线程执行 DB 查询
        result = database->query("SELECT * FROM users WHERE id=" + userId);
    }

    TPTaskState presentMainThread() override {
        // 回到主线程处理结果
        entity->onDBLoadComplete(result);
        return TPTASK_STATE_COMPLETED;
    }
};

// 2. 文件 I/O
class MapLoadTask : public TPTask {
    void process() override {
        // 在工作线程加载文件
        data = readFile("maps/map_001.dat");
    }

    TPTaskState presentMainThread() override {
        // 主线程使用加载的数据
        gameWorld->initializeMap(data);
        return TPTASK_STATE_COMPLETED;
    }
};

// 3. HTTP 请求
class PaymentTask : public TPTask {
    void process() override {
        // 在工作线程调用第三方 API
        response = httpClient->post(url, paymentData);
    }

    TPTaskState presentMainThread() override {
        // 主线程处理支付结果
        player->grantItems(response.items);
        return TPTASK_STATE_COMPLETED;
    }
};

// 4. 日志写入
class LogWriteTask : public TPTask {
    void process() override {
        // 在工作线程写日志
        logFile->write(message);
    }
};

// 1. 实体交互（涉及多个实体的状态修改）
void Entity::attack(Entity* target) {
    // 不能在工作线程执行，原因：
    // - 需要访问 target 的状态（可能被其他线程修改）
    // - 可能触发 AOI 广播（需要访问全局实体列表）
    // - 需要保证同一 tick 内的操作顺序
}

// 2. AOI 管理（需要访问全局状态）
void Entity::onMove(Position newPos) {
    // 不能在工作线程执行，原因：
    // - 需要查询周围玩家列表（全局数据结构）
    // - 需要向多个玩家发送消息（跨线程通信复杂）
    // - 需要创建/销毁实体引用（引用计数同步）
}

// 3. 游戏循环 tick（确定性要求）
void EntityApp::handleGameTick() {
    // 不能在工作线程执行，原因：
    // - 必须按固定顺序执行所有实体的 tick
    // - 玩家期望的确定性：先输入先执行
    // - 并行执行会导致顺序不可预测
}

// 4. 任务系统（涉及多个系统）
void Entity::checkQuests() {
    // 不能在工作线程执行，原因：
    // - 需要检查背包、等级、成就等多个系统
    // - 可能触发连锁反应（完成任务 → 获得道具 → 触发新任务）
    // - 需要保证状态的一致性
}

// MMORPG 服务器的实际任务分布

工作线程池处理（约 20-30% 的任务）:
├── 数据库操作 (60%)  - 玩家登录、保存、背包查询
├── 日志写入 (20%)    - 战斗日志、交易日志
├── HTTP 请求 (15%)   - 支付、第三方接口
└── 文件加载 (5%)     - 配置文件、地图数据

主线程处理（约 70-80% 的任务）:
├── 移动/寻路 (25%)   - 玩家移动、NPC 路径
├── 战斗计算 (30%)    - 攻击、技能、伤害
├── AI 决策 (20%)     - 怪物 AI、NPC 行为
├── AOI 管理 (15%)    - 视野管理、消息广播
└── 其他 (10%)        - 任务、成就、副本

// 代码量：~100 行
// 开发时间：1 天
// 维护成本：低

void Entity::attack(Entity* target) {
    // 1. 检查距离
    if (!inRange(target)) return;

    // 2. 计算伤害
    int damage = calculateDamage(target);

    // 3. 扣血
    target->hp -= damage;

    // 4. 广播伤害
    broadcastDamage(target, damage);

    // 5. 检查死亡
    if (target->hp <= 0) {
        target->die();
        onEntityKilled(target);
    }
}

// 代码量：~500 行
// 开发时间：1 周
// 维护成本：高（每月 2-3 个并发 bug）

void Entity::attack(Entity* target) {
    // 1. 获取目标锁（可能死锁）
    std::unique_lock<std::mutex> lock(target->mutex);

    // 2. 检查距离（需要锁保护坐标读取）
    if (!inRange(target)) return;

    // 3. 计算伤害（目标属性可能被其他线程修改）
    int damage = calculateDamage(target);

    // 4. 扣血（需要原子操作）
    target->hp.fetch_sub(damage);

    // 5. 广播（需要线程安全的消息队列）
    threadSafeBroadcast(target, damage);

    // 6. 检查死亡（需要避免重复触发）
    bool expected = false;
    if (target->is_dead.compare_exchange_strong(expected, true)) {
        target->die();
        onEntityKilled(target);
    }
}

// 问题：
// - 锁开销：每次操作多 50-100ns
// - 死锁风险：A 攻击 B，B 攻击 A
// - 竞态条件：目标属性可能被修改
// - 调试困难：并发 bug 难以复现

// 代码量：~1000+ 行
// 开发时间：2 周
// 维护成本：极高（需要并发专家）

class LockFreeEntity {
    std::atomic<int> hp_;
    std::atomic<int> defense_;
    std::atomic<bool> is_dead_;

    void attack(Entity* target) {
        // 1. 检查距离（需要 CAS 循环）
        Position old_pos, new_pos;
        do {
            old_pos = target->position_.load();
            if (!inRange(old_pos)) return;
            new_pos = old_pos;  // 假设位置没变
        } while (!target->position_.compare_exchange_weak(old_pos, new_pos));

        // 2. 计算伤害（需要读取多个原子变量）
        int defense = target->defense_.load();
        int damage = calculateDamage(defense);

        // 3. 扣血（CAS 循环）
        int old_hp, new_hp;
        do {
            old_hp = target->hp_.load();
            if (old_hp <= 0) return;  // 已经死亡
            new_hp = old_hp - damage;
        } while (!target->hp_.compare_exchange_weak(old_hp, new_hp));

        // 4. 广播（无锁队列）
        lockFreeQueue.push(DamageMessage{target, damage});

        // 5. 检查死亡（又是 CAS）
        if (new_hp <= 0) {
            bool expected = false;
            if (target->is_dead_.compare_exchange_strong(expected, true)) {
                // 触发死亡逻辑（又是一个复杂的 CAS 操作）
                onEntityKilled(target);
            }
        }
    }
};

// 问题：
// - CAS 失败重试：性能不如单线程
// - ABA 问题：需要 Hazard Pointers
// - 内存回收：需要延迟删除
// - 复合操作：难以保证原子性
// - 开发成本：需要并发专家
// - 维护成本：每次修改需要重新验证

// 工作线程任务的三种返回状态

enum TPTaskState {
    TPTASK_STATE_COMPLETED = 0,               // 任务完成，删除
    TPTASK_STATE_CONTINUE_MAINTHREAD = 1,     // 结果回主线程继续执行
    TPTASK_STATE_CONTINUE_CHILDTHREAD = 2,    // 继续在子线程执行
};

// 类似 Promise 的状态：
// - COMPLETED → Promise.resolve()
// - 继续执行 → Promise.then()
// - 失败 → Promise.reject()

// 数据库写入回调 - 类似 Promise.then(success, error)
void Entity::onWriteToDBCallback(ENTITY_ID eid,
                                  DBID entityDBID,
                                  uint16 dbInterfaceIndex,
                                  CALLBACK_ID callbackID,
                                  int8 shouldAutoLoad,
                                  bool success)  // ← 成功/失败标志
{
    isArchiveing_ = false;

    PyObjectPtr pyCallback;
    if(callbackID > 0)
        pyCallback = callbackMgr().take(callbackID);  // ← 根据 ID 取回回调

    if(callbackID > 0)
    {
        // 构造回调参数（类似 Promise 的 resolve/reject）
        PyObject* pyargs = PyTuple_New(2);

        Py_INCREF(this);
        PyTuple_SET_ITEM(pyargs, 0, PyBool_FromLong((long)success));  // ← 第一个参数：成功/失败
        PyTuple_SET_ITEM(pyargs, 1, this);  // ← 第二个参数：实体对象

        if(pyCallback != NULL)
        {
            // 执行回调（类似 Promise.then(callback)）
            PyObject* pyRet = PyObject_CallObject(pyCallback.get(), pyargs);
        }
    }
}

// 定义事件类型（类似 Promise 的状态）
#define CLIENT_EVENT_LOGIN_SUCCESS 6      // ← Promise.resolve()
#define CLIENT_EVENT_LOGIN_FAILED 7       // ← Promise.reject()

// 事件数据结构（类似 Promise 的 value/reason）
struct EventData_LoginSuccess : public EventData {
    EventData_LoginSuccess():
        EventData(CLIENT_EVENT_LOGIN_SUCCESS) {}
};

struct EventData_LoginFailed : public EventData {
    EventData_LoginFailed():
        EventData(CLIENT_EVENT_LOGIN_FAILED),
        failedcode(0) {}  // ← 失败原因
};

// 登录成功 - 类似 Promise.resolve()
void ClientObjectBase::onLoginSuccessfully(Network::Channel * pChannel, MemoryStream& s)
{
    // 解析服务器响应
    s >> accountName;
    s >> ip_;
    s >> tcp_port_;
    s.readBlob(serverDatas_);

    // 触发成功事件（类似 Promise.then()）
    EventData_LoginSuccess eventdata;
    eventHandler_.fire(&eventdata);  // ← 触发所有监听器
}

// 登录失败 - 类似 Promise.reject()
void ClientObjectBase::onLoginFailed(Network::Channel * pChannel, MemoryStream& s)
{
    SERVER_ERROR_CODE failedcode;
    s >> failedcode;  // ← 错误码
    s.readBlob(serverDatas_);

    // 触发失败事件（类似 Promise.catch()）
    EventData_LoginFailed eventdata;
    eventdata.failedcode = failedcode;  // ← 失败原因
    eventHandler_.fire(&eventdata);  // ← 触发所有监听器
}

// 文件：kbe/src/lib/common/objectpool.h（简化）
template<typename T, typename THREADMUTEX = KBEngine::thread::ThreadMutexNull>
class ObjectPool { /* ... */ };

// 主线程使用（默认空锁）
ObjectPool<MemoryStream> streamPool;              // ThreadMutexNull

// 多线程使用（传入真实锁）
ObjectPool<MemoryStream, ThreadMutex> sharedPool;  // ThreadMutex

// 文件：kbe/src/lib/thread/threadmutex.h
class ThreadMutexNull
{
public:
    virtual void lockMutex(void) {}
    virtual void unlockMutex(void) {}
};

class ThreadMutex : public ThreadMutexNull
{
    virtual void lockMutex(void) { THREAD_MUTEX_LOCK(mutex_); }
    virtual void unlockMutex(void) { THREAD_MUTEX_UNLOCK(mutex_); }
};

// 文件：lib/cstdmf/concurrency.hpp
class DummyMutex
{
public:
    void grab() {}
    bool grabTry() { return true; }
    void give() {}
};

class SimpleMutex
{
    void grab() { pthread_mutex_lock(&mutex_); }
    void give() { pthread_mutex_unlock(&mutex_); }
};

// 文件：lib/cstdmf/pool_allocator.hpp
template <class MUTEX = DummyMutex>
class PoolAllocator { /* ... */ MUTEX mutex_; };

// 文件：kbe/src/lib/common/objectpool.h（简化）
template<typename T, typename THREADMUTEX = ThreadMutexNull>
class ObjectPool
{
    T* createObject(const std::string& logPoint)
    {
        pMutex_->lockMutex();
        while (true) {
            if (obj_count_ > 0) {
                T* t = static_cast<T*>(*objects_.begin());
                objects_.pop_front();
                --obj_count_;
                t->isEnabledPoolObject(true);
                pMutex_->unlockMutex();
                return t;
            }
            assignObjs();  // 池空时批量预分配 16 个
        }
    }

    void reclaimObject(T* obj)
    {
        obj->onReclaimObject();  // 重置状态
        if (size() >= max_ || isDestroyed_)
            delete obj;          // 池满则真正释放
        else
            objects_.push_back(obj);  // 回收到池
    }

    // 瘦身：每 300 秒检查，如果池大小 > OBJECT_POOL_INIT_SIZE 则缩减
};

class PoolObject {
    virtual void onReclaimObject() = 0;  // 回收时重置
    bool isEnabledPoolObject_;
    std::string poolObjectCreatePoint_;  // 创建位置追踪（防泄漏）
};

// RAII 包装
template<typename T>
class SmartPoolObject {
    ~SmartPoolObject() { objectPool_.reclaimObject(pPoolObject_); }
};

// 文件：lib/cstdmf/object_pool.hpp（简化）
// Handle：索引 + 世代号，防止悬垂引用
template<size_t indexBits, size_t generationBits>
struct Handle {
    baseType index_ : indexBits;
    baseType generation_ : generationBits;
};

// PackedObjectPool：紧凑排列 + swap-and-pop 删除
template<typename ObjectType, typename HandleType = DefaultHandle>
class PackedObjectPool
{
    HandleType create();           // 尾部追加
    void release(HandleType h);   // swap-and-pop（O(1)，保持紧凑）
    ObjectType& lookup(HandleType h);  // handle → 内部索引 → 对象

    // 双向映射
    HandleTable<HandleType> handles_;
    BW::LookUpTable<size_t> outerLookup_;  // handle → 内部索引
    BW::vector<size_t> innerToOuter_;      // 内部索引 → handle
    ObjectContainer innerLookup_;          // 紧凑存储的对象数组
};

// 文件：lib/cstdmf/allocator.hpp（简化）
namespace BW::Allocator {
    void* allocate(size_t size);
    void  deallocate(void* ptr);
    void* reallocate(void* ptr, size_t size);
    // 调试：setReportOnExit, setCrashOnLeak, debugReport
    // 填充模式：CleanLandFill(0xCD), DeadLandFill(0xDD), NoMansLandFill(0xFE)
}

// 文件：lib/cstdmf/fixed_sized_allocator.hpp（简化）
class FixedSizedAllocator
{
    static const int MaxPools = 16;
    size_t allocSizes_[MaxPools];     // 每个池的分配大小
    PoolHeader* topLevelPools_[MaxPools]; // 池链表头

    void* allocate(size_t size) {
        // 找到最匹配的池，从空闲链表取
    }
};

// 文件：lib/cstdmf/pool_allocator.hpp（简化）
template <class MUTEX = DummyMutex>
class PoolAllocator
{
    void* allocate(size_t size) {
        if (pHead_) { ret = pHead_; pHead_ = pHead_->next; }  // 从空闲链表取
        else { ret = new char[size]; }                          // 或新分配
    }
    void deallocate(void* p) {
        // 不真正释放，放回空闲链表
    }
};

// 文件：lib/cstdmf/bw_vector.hpp
namespace BW {
template<class T, class Allocator = StlAllocator<T>>
class vector : public std::vector<T, Allocator> { /* ... */ };
}

// 文件：kbe/src/lib/common/refcountable.h
// 非线程安全
class RefCountable {
    volatile mutable long refCount_;
    void incRef() const { ++refCount_; }
    void decRef() const { if (--refCount_ <= 0) onRefOver(); /* delete this */ }
};

// 线程安全（原子操作）
class SafeRefCountable {
    // Windows: InterlockedIncrement / InterlockedDecrement
    // Linux: __sync_add_and_fetch / __sync_sub_and_fetch
};

// 文件：lib/cstdmf/smartpointer.hpp
class ReferenceCount { /* 非线程安全 ++count_ */ };
class SafeReferenceCount {
    // BW_ATOMIC32_INC_AND_FETCH / BW_ATOMIC32_DEC_AND_FETCH
    void decRef() const {
        int32 count = BW_ATOMIC32_DEC_AND_FETCH(&count_);
        if (count == 0) this->destroy();
    }
};