如何在最小程度影响每秒更新的情况下存储和推送模拟状态？

Question

我的应用程序由两个线程组成：

1. GUI线程（使用Qt）
2. 模拟线程

我之所以使用两个线程，是为了保持GUI的响应性，同时让Sim线程尽可能地快速运行。

在我的GUI线程中，我以30-60帧的FPS渲染Sim中的实体；然而，我希望我的Sim能够“向前推进” - 所谓的 - 并排队等待游戏状态最终被绘制出来（类似于流媒体视频，你有一个缓冲区）。

现在对于我渲染的每一帧Sim，我需要相应的模拟“状态”。因此，我的Sim线程看起来像这样：

while(1) {
    simulation.update();
    SimState* s = new SimState;
    simulation.getAgents( s->agents ); // store agents
    // store other things to SimState here..
    stateStore.enqueue(s); // stateStore is a QQueue<SimState*>
    if( /* some threshold reached */ )
        // push stateStore
}

SimState看起来像这样：

struct SimState {
    std::vector<Agent> agents;
    //other stuff here
};

而Simulation::getAgents看起来像：

void Simulation::getAgents(std::vector<Agent> &a) const
{
    // mAgents is a std::vector<Agent>
    std::vector<Agent> a_tmp(mAgents);
    a.swap(a_tmp);
}

Agent类本身是一些相对复杂的类。成员包括一些int和float以及两个std::vector<float>。

在当前设置下，模拟无法比GUI线程绘制得更快。我已经验证了当前瓶颈是simulation.getAgents(s->agents)，因为即使我省略了推送更新，每秒的更新速度也很慢。如果我注释掉那行代码，我会看到更新/秒有几个数量级的改善。

那么，我应该使用哪些容器来存储模拟状态？我知道目前存在大量复制，但其中一些是不可避免的。我应该在向量中存储Agent*而不是Agent吗？

注意：实际上，模拟不在一个循环中，而是使用Qt的QMetaObject::invokeMethod(this, "doSimUpdate", Qt::QueuedConnection);，所以我可以使用信号/插槽在线程之间通信；然而，我已经验证了一个简单版本，使用while(1){}，问题仍然存在。

- Casey

---

我很想知道你是怎样解决这个问题的，尽管可能已经有一段时间了。我也遇到了类似的问题。 - woosah

Answer 1

尝试重复使用SimState对象（使用某种池机制），而不是每次分配它们。在几个模拟循环后，被重复使用的SimState对象将具有已增长到所需大小的向量，从而避免重新分配并节省时间。

实现池的简单方法是最初将一堆预先分配的SimState对象推入std :: stack 中。请注意，由于您想要获取更有可能在缓存内存中“热”的SimState对象（最近使用的SimState对象将位于堆栈顶部），因此堆栈优于队列。您的模拟队列从堆栈中弹出SimState对象，并用计算出的SimState填充它们。然后将这些计算出的SimState对象推入生产者/消费者队列以供GUI线程使用。在被GUI线程呈现后，它们将被推回到SimState堆栈（即“池”）。在执行所有这些操作时，请尽量避免不必要的SimState对象复制。在“管道”的每个阶段直接使用SimState对象。

当然，在SimState堆栈和队列中使用适当的同步机制以避免竞态条件。Qt可能已经有线程安全的堆栈/队列。如果存在大量争用，则无锁堆栈/队列可能会加速（Intel Thread Building Blocks提供此类无锁队列）。鉴于计算SimState需要约1/50秒的时间，我怀疑争用不会成为问题。

如果SimState池变得不足，则意味着您的模拟线程过于“超前”，可以等待一些SimState对象返回到池中。模拟线程应该阻塞（使用条件变量），直到池中再次可用SimState对象。SimState池的大小对应于可以缓冲多少SimState（例如，大约50个对象的池可以给您提供长达约1秒的紧凑时间）。

您还可以尝试运行并行模拟线程以利用多核处理器。Thread Pool模式在这里可能很有用。但是，必须注意计算的SimStates按正确顺序排队。按时间戳排序的线程安全优先级队列可能适用于此处。

这是我建议的管道架构的简单图示：

（右键单击并选择查看图像以获得更清晰的视图。）

（注意：池和队列通过指针持有SimState，而不是值！）

希望这可以帮到您。

---

如果您计划重复使用SimState对象，则您的Simulation :: getAgents 方法将是低效的。这是因为vector<Agent>& a 参数很可能已经具有足够的容量来保存代理列表。

您目前的做法会丢弃已分配的向量并从头开始创建一个新向量。

我个人认为，您的getAgents 应该是：

void Simulation::getAgents(std::vector<Agent> &a) const
{
    a = mAgents;
}

是的，您会失去异常安全性，但您可能会获得更好的性能（特别是使用可重复使用的SimState方法）。

---

另一个想法：您可以尝试使用C风格数组（或boost::array）和“计数”变量而不是std::vector来使您的Agent对象具有固定大小的浮点列表成员。只需将固定大小的数组足够大以适应模拟中的任何情况即可。是的，您会浪费空间，但您可能会获得很多速度。

然后，您可以使用固定大小的对象分配器（例如boost::pool）池化您的Agents，并通过指针（或shared_ptr）传递它们。这将消除很多堆分配和复制。

您可以单独或与上述想法结合使用此想法。这个想法似乎比上面的管道事情更容易实现，所以您可能首先要尝试它。

---

另一个想法：不使用线程池运行模拟循环，而是将模拟分解成几个阶段，并在各自的线程中执行每个阶段。生产者/消费者队列用于在阶段之间交换SimState对象。为了使其有效，不同的阶段需要具有大致相似的CPU工作负载（否则，一个阶段将成为瓶颈）。这是利用并行性的另一种方式。

- Emile Cormier

---

请查看我的回答中的后续内容。临时变量和交换惯用语是确保异常安全性的好方法，但可能会牺牲性能。既然您发现该方法成为了瓶颈，那么我认为牺牲异常安全性是合理的。只需再次进行分析以确保您实际上获得了性能提升。 - Emile Cormier

上一句话中有错别字。忽略“b/c”。 - Casey

我现在明白了，通过指针传递Agent可能没有什么帮助。 Simulation需要它自己独立于存储在SimState中的代理的一组代理。因此需要按值复制Agent。从Agent中消除std::vector加快了复制过程，可能是因为代理现在可以简单地进行memcpy，而不是成员逐个复制。 - Emile Cormier

哦，在我实施了您的建议并将渲染器与计算一起运行之前，我每秒获得约50个滴答声。因此，如果我事先分配足够的SimState *对象，我可以维持一段时间的良好滴答/秒，尽管由于模拟以如此高的速率产生SimState *，期望总是有足够的SimStates可能是不可行的。我需要做数学运算。 - Casey

不要分配比原始池中的更多的“SimState”！一旦您耗尽了池，模拟线程应该阻塞，直到池中再次有可用的SimState。使用与生产者/消费者队列相同的阻塞机制（通常使用条件变量完成）。如果您到达池被耗尽的地步，那么这意味着模拟线程在“预缓冲”SimState对象方面做得很好。 - Emile Cormier