分类： 程序设计

从我开始研究ECS算起, 到现在已经将近20天了。

第一版ECS库终于实现完成了。先不论性能如何，基本功能都实现了。

在我的理解中，ECS中最复杂的地方是EC部分的管理和查询。而S部分的复杂度主要是依赖关系的问题，这会取决于具体的项目。

因此，在这个ECS库中主要解决EC的问题，关于S的部分并没有提供。这也是我称它为库而不是框架的原因。

在整个实现过程中，由于我还没能完全克服性能强迫症，导致我的心路历程非常坎坷（每次实现到一半，总会因为这样或那样的原因，让我推倒重来）。

最开始，我认为守望先锋的ECS之所以那么复杂，是因为他们使用了C++这种强类型语言。为了解决动态组合（动态添加和删除C）的问题，不得不在API上做出一些让步。

如果拿Lua来实现，语言本身就支持动态组合，那添加/删除Component的行为，可以退化为添加/删除“标签”功能。

每个System只需要过滤出含有特定“标签”组的Entity, 然后加以处理就行了。

很快我放弃了这一想法，主要原因是我认为作为一个合格的框架或库，它应该提供一些限制。可以让我们写出符合ECS原则，更易读的代码。

在上面的设计中，客户程序员很容易就违反了ECS原则，他完全可以只过滤某一个ComponentA, 然后去修改这个Entity中的ComponentB, 甚至删掉ComponentB但是并不会删除ComponentB的标签。这会导致一些很奇怪的Bug。而且从代码的易读性上来讲也没有好处。

在后续的设计中，我又陆续纠结了，Eid的分配问题， Component的存储问题，同一个Entity中的Component的关联问题。

在经过陆陆续续几次推倒重来之后，直到今天才实现完第一个版本。

在这不断的推倒重来中，我总是在是否“需要暴露Eid给客户程序”之间摇摆不定。最终，我认为是需要的。

我们总是需要在程序的某处去New出一个个的Entity。同样我们也总会需要在程序的某处，去修改某个特定Entity的某个Component数据。

在我看来，整个ECS的运行机制很像一个巨大的“粉碎机”。 我们总是在某一个入口投入足量的Entity, 然后ECS库或框架将这些Entity粉碎成各种Component，供System查询并操作。

因此在这一版的ECS库的实现中，我把Component作为主角来实现的。Entity的作用在这里，将一组Component进行关联，以方便Component查询和生命周期的管理。

先简单介绍一下API：

--创建一个名为Admin的world对象。使用相同名字多次调用ECS.fetch_world, 返回的是同一个world对象
local world = ECS.fetch_world("Admin")

--注册Component类型。 其中world.register的第二个参数是为了方便建立Component缓存池和Debug阶段检查一些Component的合法性（暂时还没有实现）。
world:register("vector2", {x = 0, y = 0})
world:register("vector3", {x = 0, y = 0, z = 0})

--创建一个Entity, 这个Entity只含有一个"vector2"的Component
local eid = world:new { vector2 = {x = 2, y = 2}}

--向eid所代表的Entity上添加一个"vector3"的Component
world:add(eid, "vector3", {x = 3, y = 3, z = 3})

--向eid所代表的Entity上删除一个"vector3"的Component
world:remove(eid, "vector3")

--查询world中的所有类型为"vector2"的Component
for v2 in world:match("all", "vector2") do
    w:touch(v2) --将Component v2置为脏标记
end

--查询world中所有被w:touch过的类型为"vector2"的Component
for v2 in world:match("dirty", "vector2") do
end

--查询world中所有已经死亡的类型为"vector2"的Component
for v2 in world:match("dead", "vector2") do

end

--删除Entity
world:del(eid)

--执行清理操作，每调一次为一个逻辑帧
world:update()

整个设计大概是这样的：

每个Component类型都有一个数字id称为tid。每个Component实例都有一个数字id称为cid。我们总是可以根据tid和cid来找到某一个具体的Component实例。

在相同的Component类型中，新创建的Component的cid总是比旧的Component的cid要大。在world:update时所有Component的cid会进行重排，但是依然满足这一约束。这会提供一个便利，在我们使用for遍历world:match时，依然可以不受限制的添加任何Compoent实例。

当某个Component实例被删除时，仅将其挂在“dead”链表上，并不做其他操作。如果已经在“dead”链表上，则不做任何处理。这会产生一个限制，刚对某个Entity删除了一个Component之后,不可以立马添加一个同类型的Component。

当某个Component实例被touch时，仅将其挂在“dirty”链表上。

当某个Entity被删除时，将此Entity下的所有Component标记为"dead", 然后将Entity挂在"dead"链表，不做任何处理。

在执行world:update时会产生以下行为：

1. 释放所有的Entity及其eid（以备后面复用)
2. 释放所有标记为“dead"的Component, 并整理存活的Component的cid
3. 清除"dead"链表
4. 清除"dirty"链表

总的来讲，所有的添加都是立即生效，所有的释放都会延迟到world:update中执行。

ps. 在这次纠结的过程中，在一定程度上治愈了我的性能强迫症。

大约在1个月前，游戏终于上线了，在这一个月以来，服务器竟然crash了5+次，还有几次严重Bug.

除了觉得测试力度不够之外，我也在想到底有那些环节是我能做而没做好的。

仔细思考下来，出Bug的原因大概有两种典型情况。

写代码时，逻辑思维不严谨，并且函数之间耦合性过强，导致随便一个很平常的改动，都会产生新的Bug, 甚至是crash. 比如我经常写出类似下面这种函数之间耦合性过强代码。

struct foo {
    bool isfinish;
    time_t rsttime;
    //...
};
static viod try_reset(struct foo *f) {
    if (f->rsttime < time(NULL)) {
        //reset foo fields
    }
}
static int bar(struct foo *f) {
    if (f->isfinish)
        return -1;
    try_reset(f);
    //....do more things
}

这上面这个代码里，bar和try_reset耦合性实在是太强了，强到甚至我调换一下try_reset的调用顺序，都会导致整个逻辑出错。这次出的Bug中有好几个都是这种原因，原始逻辑是1年前写的，上线前调整了一下，结果没测到，然后就出问题了。

我仔细回忆了一下，这种强耦合的实现方式，是我最近两年才开始出现的情况。

之所以会出现这种情况，一方面是因为代码估算做多了之后，我对一条两条指令的开销同样很敏感，以至于很多时候把控不住优化的尺度，把设计弄糟。另一方面，我一直坚持设计正确，不需额外检查这一原则。对于上面的代码，本来设计try_reset就是需要调用方保证，一定在foo.finish为false时才调用，所以try_reset没有理由去检查foo.finish字段。

事实上，这并不是我第一次发现这个问题，以前也发生过几次这种问题。每次发生之后，总是自嘲一句“过早的优化是万恶之源”，然后改正了事。

但是连续几次的Bug, 让我头脑清醒了很多，我觉得我有必要重新审视一下这个问题，这应该不仅仅是过早优化的问题。

这段代码相关的需求大概是这样：如果foo在开始之后，需要完成若干操作，如果在开始之后的规定时间段内没有完成，则重置完成进度，重新完成。

为了“削峰填谷”，我并没有为每一个foo设置一个定时器（即使是固定时间轮），只是在每次获取foo结构时，尝试重置foo的进度。

经过重新审视了整个需求发现，其实“try_reset"的前置条件有两个：一个是没有完成，一个是过期时间。

但是在实现过程中， 把条件1放在了调用方，条件2放在了被调用方（也就是try_reset)。这样try_reset执行结果是否正确，竟然需要调用方来保证。以致于所有调用try_reset的函数，都会与try_reset有强耦合。这其实很像是“契约式编程”，但是除非有语言层面支持，不然只是靠人脑来保证契约前置条件是不靠谱的，所以我个人其实也不是很相信“契约式编程”。

将try_reset函数更改如下，其实可以避免很多Bug, 尤其是多年之后需要修改更时如此：

static viod try_reset(struct foo *f) {
    if (f->isfinish == false && f->rsttime < time(NULL)) {
        //reset foo fields
    }
}

那么上面问题的答案来了，是“过早优化么”？不是。“设计正确不需要检查”这个原则正确么，我依然认为是正确的。因为这种类型的Bug，其根本原因是没有设计好。

根据历史的设计经验，我一般都会有意识的将class/module设计为自成闭环。但是在一个class/module内部函数级实现时，几乎没有仔细思考抽象过，都是实现调用方函数的过程中做的，顺手提取出一个子函数供调用方调用，因此大部分情况下，内部函数和调用方之间都有很强的耦合性。当一个模块过大时，这种耦合性会呈几何倍数增加。

现在回过头来看，其实每一个函数在提取时，都值得仔细抽象来和调用方解偶，即使现在它只有一个调用方。据我过去的经验，你最开始抽象出来的函数，往往会随着业务逻辑的衍变，产生多个调用方。

上面的问题，大部分情况下只是会产生逻辑bug, 一般不太会诱发crash.

据最近几次crash的经验来看，主要原因就是在对于个unordered_map在for循环时，删除了其中的一些元素。比如下面代码：

struct st {
    int id;
    int progress;
    //other fields
};
std::unordered_map<int, struct st> DB;

bool bar(struct st &x)
{
    ++x.progress;
    if (x.progress >= MAX_PROCESS) {
        database_delete_by_id(x.id);
        DB.erase(x.id);
        return false;
    }
    //process x
    return true;
}

void foo(std::vector<int> &l)
{
    l.reserve(DB.size())
    for (auto &iter:DB) {
        auto &x = iter.second;
        if (bar(x)) {
            l.emplace_back(x.f1);
        }
    }
}

这种Bug最为讨厌，只要bar函数删除一个元素，就会导致整个迭代器失效，并且不是必崩的。所以这种Bug即使在测试足够的情况下，也很容易逃逸到线上。一般我都会将代码改成如下：

struct st {
    int id;
    int progress;
    //other fields
};
std::unordered_map<int, struct st> DB;

static std::unordered_map<int, struct st>::iterator
check(std::unordered_map<int, struct st>::iterator iter, bool &clear)
{
    auto &x = iter->second;
    if (x.progress >= MAX_PROCESS) {
        database_delete_by_id(x.id);
        clear = true;
        return DB.erase(iter);
    } 
    clear = false;
    return ++iter;
}

static std::unordered_map<int, struct st>::iterator
bar(std::unordered_map<int, struct st>::iterator iter, bool &clear)
{
    bool clear;
    auto &x = iter->second;
    ++x.progress;
    iter = check(iter, clear);
    if (clear == false) {
        //process x
    }
    return iter;
}

void foo(std::vector<int> &l)
{
    bool clear;
    l.reserve(DB.size())
    for (auto iter = DB.begin(); iter != DB.end(); ) {
        bar(iter, clear);
        if (clear == false) {
            l.emplace_back(x.f1);
        }
    }
}

事实上，我个人很不喜欢这种改动，一股bad taste扑面而来。我也很不喜欢unordered_map的这个限制，他会让for循环方和被调用方强耦合在一起。

for循环方必须保证被调用方不会删除unordered_map中的元素才可以调用。即使是第二种写法，也有一种浓浓的耦合味道。

最近重新审视这种代码时，我发现也许我们可以使用“变换式编程（《程序员修炼之道(第二版)》P149)来解决这种问题。

struct st {
    int id;
    int progress;
    //other fields
};

std::unordered_map<int, struct st> DB;

static std::unordered_map<int, struct st>::iterator
check(std::unordered_map<int, struct st>::iterator iter)
{
    auto &x = iter->second;
    if (x.progress >= MAX_PROCESS) {
        database_delete_by_id(x.id);
        return DB.erase(iter);
    } else {
        return ++iter;
    }
}

static void
batch_check()
{
    auto iter = DB.begin();
    while (iter != DB.end()) {
        iter = check(iter);
    }
}

static void
batch_process()
{
    for (auto &iter:DB)
        ++iter->second.progress;
}

void foo(std::vector<int> &l)
{
    bool clear;
    l.reserve(DB.size())
    batch_process();
    batch_check();
    for (auto &iter:DB) {
        l.emplace_back(iter.second.f1);
    }
}

当然这样的改动会增加一点点性能开销，但是我认为这点开销不会成为热点的致因。

在研究上述问题的同时，我也在想，除了良好的抽象之外，我还能为代码的可靠性做些什么。

是的，答案只有一个，为代码写测试代码。

我查了很多关于单元测试理论性的书籍，他们都告诉我，单元测试就是要把被测试的类通过mock类隔离掉，转而只测试这一个类。

但是游戏服务器的业务太复杂了，以至于会有少量循环引用的情况。这样我实现的mock类到最后，除了可以设置假数据之外，几乎和真实的类的功能一模一样了。

而且我看到书上的很多例子，为了可测试性，会对被测试类的接口加以修改。这种侵入式的测试，我很难适应。

而且游戏服务器还有三大不可控因素: 时间,配表,数据库。

时间 – 在业务逻辑中，有大量时间相关的逻辑，比如多久回复多少资源, 执行一次操作后有多久的CD。与此相对的是，测试代码需要尽可能快的跑完。

配表 – 在业务逻辑中，有大量逻辑是根据配置来，因此测试代码并不能写死，需要根据配置自适应。而且升级配表之后，很难保证测试代码的正确性。

数据库 – 在业务逻辑中，玩家所有的操作造成的影响都会被存入数据库，因此测试不具有可重复性。

其实我很早以前就想为我写过的业务逻辑编写测试代码，但是基于以上种种原因，尝试了几次均以失败而告终。

由于最近出Bug的概率太高了，我下定决心想去解决游戏服务器的单元测式。

即然书上的理论在我这里行不通，那么我去看一些开源项目对单元测试是如何落地的。

我分别考察了Lua和Redis，发现他们都没有做大量侵入式的单元测试。而是通过他们“特有的方式”来触发相关的代码执行。

比如Lua是通过执行Lua代码来测试Lua虚拟机的各种角落，而Redis则是通过网络协议来测试。

我研究了一下这种方式，发现对游戏服务器同样可行。

其实我们根本不需要mock类。理论上我们可以通过网络协议来操作模块处理指定状态。然后再发协议操作某个特定模块，以验证这个模块的正确性。只要我们为服务器的每一个模块编写一个对应的测试操作模块，就可以大幅度减少测试代码, 大大提高了测试的可行性。

比如account操作模块提供一个account.eusure_create_role_whith_money(name, passwd, money)来保证创建一个名字为name, 密码为passwd, 初始金币为money的账号。

解决了mock类的问题之后，再来看看时间问题。

为了让测试代码尽可能快的运行，我们只有一种选择，那就是调时间。但是频繁去调开发机的时间，会造成一些很麻烦的后果，而且每个单元测试之后，都要把时间调回来。

好在，当时为了减少time()函数造成的系统调用，我封装了一个time.cpp静态类，所有的时间都是通过time::get()函数来获取的。我设计了一条GM指令，可以调整时间偏移量。为了避免测试相关的GM指令逃逸到线上去。我在本地Makefile加入了一个TEST宏，以确保只有test版本才会有这些指令。如此，时间的问题解决了。事实上不仅仅是调时间，我还加了少部分用于控制特殊逻辑的GM开关，以方便测试代码执行（就像Lua同样在源码里留了一些桩点用于测试使用)，这些控制逻辑同样是被TEST控制。

配置表的问题，咋一看很麻烦，但是静下心来想想，其实我们的代码依赖的并不是表中的内容。而是表结构。所以在写完测试代码之后，除非功能有变化，不然根本没有理由去更改表的内容。惟一麻烦的是，有时候我们需要通过改表来隔绝其他模块造成的干扰。这种情况下，我们就需要为这引单元测试代码定制一份配置表出来。这也意味着我们的单元测试框架需要有，能为不同单元测试代码指定不同配置的能力。

数据库的问题，其实在我们编写测试操作模块时就已经解决了。我们只需要像Redis一样，在执行每个单元测试之前，把所有测试进程杀死，并且清空数据库即可。每个单元测试自己负责构造自己所需要的数据。由于我们的测试操作模块提供的都是很上层的操作。因此创建数据部分，并不会花费太多代码。

基于以上事实，我实现了一个简易的单元测试框架，并配合valgrindt和GCC的ASan，可以同时进行逻辑测试和内存问题测试。

这个框架有部分项目相关性，而且代码并不算太多，因此并没有开源。

ps. 在编辑测试框架过程中，发现一个有意思的问题。当可执行程序在运行过程中，替换其所依赖的so文件会造成进程崩溃。研究了一下发现，这是因为so文件的代码部分是通过mmap到进程内存空间直接执行的。而mmap的特性就是，对映射内存的修改，直接反映到文件，对文件的修改直接反应到mmap的进程内存地址。而cp命的执行步骤一般是，将文件大小截断为0，然后再写入新文件内容。当文件大小截断为0这个步骤，一下子就把so的代码部分破块了，崩溃也就成了必然。

在日常开发中，伪随机函数几乎是必不可少的一个函数。

大部分我们在使用这个函数时，就自然而然拿来用了，很少去思考用的对不对，反正他是随机的，并且也很难去验证（需要各种大量数据统计)。

所以即使概率看起来不太对，也可以安慰自己说，其实是统计的数据量不够。但有时候真的是因为我们误用了随机函数。

在《计算机程序设计艺术》卷2中，详细介绍了线性同余序列的生成算法。
下面就以线性同余算法为例，来分析一下，为什么随机函数还有可能被误用，他原本不就是随机的么？

在游戏开发中，一般都会设计有开宝箱环节，假设每个宝箱每次开出A的概率是30%，开出B的概率是70%，宝箱可以重复开。

我们的代码可能会这么写

    int open_box(box *b)
    {
        int n = rand() % 1000;

        return  n < 300 ? b->a : b->b;
    }

是的,这段代码就是开宝箱存在“垫刀”的根本原因。

我们来看一下线性同余(LCG)伪随机算法的定义：

Nj+1 = (A*Nj + B) (mod M)(j, j+1为下标)

其中A,B,M为线性同余序列生成常数。

LCG周期为M，A,B,M的关系限定如下：

1. B,M互质
2. M的所有质因子都能整除A-1
3. 若M是4的倍数，则A-1 也是
4. A, B, N0都比M小
5. A,B是正整数

通俗点来讲就是，线性同余生成的[0,M)个数在统计学意义上，是等概率出现的。也就是说在足够多次随机以后，他们出现的次数是相同的。

咋一看，感觉上面的代码好像没啥问题。因为[0,M)是等概率出现的，因此rand()%1000之后的值，也是等概率出现的。

但是！我们忽略了一个事实，这段代码意味着。所有人的所有宝箱（甚至还有其他系统）共用了一个伪随机序列。

假设rand()%1000的伪随机序列是这样的：

900，1，300， 500， 299， 785， 556 …

我们来模拟一下多个宝箱交替打开的行为：

开宝箱1，rand()%1000返回的是900， 因此开出来的是B

开宝箱2，rand()%1000返回的是1， 因此开出来的是A

开宝箱1，rand()%1000返回的是300， 因此开出来的是B

开宝箱1，rand()%1000返回的是500, 因此开出来的是B

开宝箱2， rand()%1000返回的是299， 因此开出来的是A

如果宝箱1和宝箱2一直在以类似的顺序交替打开。即使开再多次，你也很难拍着胸脯说，宝箱1和宝箱2开出来的A，B概率分布是符合预期的。

毕竟你亲口告诉玩家，每个宝箱都有30%的概率开出来的是A，但是宝箱1却从来开不出A。

事情之所以会演变成这样。根本原因是，除了有一个伪随机序列之外，还有一个真随机事件，即玩家开宝箱的时机选择。

用软件工程的话来说，宝箱1和宝箱2通过一个全局变量（同一个线性同余序列）耦合在一起了，他们不是正交的。因此，开一个宝箱势必会影响另一个，所以它必然是错的。

还有很多类似的情况，比如一个技能的触发概率。我们本来告诉玩家的是每个技能以某种特定的概率触发，但是我们很可能做成了，以某种概率释放了某个技能。

在我们用随机函数之前，一定要先问问自己，所有使用rand()函数的地方其实是共用了同一个伪随机序列，这样真的没问题么？

2021/11/16补充:

经过这一年多来的观察, 除了"垫刀"的问题之外, 我们还习惯使用如1000, 10000之类的数字来代替100%的概率. 这就会产生一个现象, 就是1000或10000很可能和LCG公式中的M是有公约数的, 这会导致LCG的周期缩短, 其影响远大于"垫刀"产生的不良影响。在我们将接近1000或10000的质数作为100%概率的代表时，随机数的分布有明显改善。