【问题标题】:How to improve performance of distribution to std::function listeners?如何提高分发到 std::function 侦听器的性能?
【发布时间】:2018-06-24 00:22:43
【问题描述】:

简而言之,有什么明显的方法可以让下面代码中的distributor.distribute() 调用运行得更快吗?

#include <iostream>
#include <memory>
#include <functional>
#include <vector>
#include <typeindex>
#include <unordered_map>
#include <chrono>


// ---------------------------------------------------------------------
// Things to get passed around
// ---------------------------------------------------------------------
class Base {
public:
  virtual ~Base() {};
};
class Derived : public Base {};

// ---------------------------------------------------------------------
// Base class for our Handler class so we can store them in a container
// ---------------------------------------------------------------------
class BaseHandler
{
public:
  virtual ~BaseHandler() {};
  virtual void handle(std::shared_ptr<const Base> ptr) = 0;
};

// ---------------------------------------------------------------------
// Handler class to wrap a std::function. This is helpful because it
// allows us to add metadata to the function call such as call priority
// (not implemented here for simplification)
// ---------------------------------------------------------------------
template <typename T>
class Handler : public BaseHandler
{
public:
  Handler(std::function<void(std::shared_ptr<const T>)> handlerFn)
  : handlerFn(handlerFn) {};
  void handle(std::shared_ptr<const Base> ptr) override {
    handlerFn(std::static_pointer_cast<const T>(ptr));
  }
private:
  std::function<void(std::shared_ptr<const T>)> handlerFn;
};

// ---------------------------------------------------------------------
// Distributor keeps a record of listeners by type and calls them when a
// corresponding object of that type needs to be distributed.
// ---------------------------------------------------------------------
class Distributor
{
public:
  template <typename T>
  void addHandler(std::shared_ptr<Handler<T>> handler)
  {
    handlerMap[std::type_index(typeid(T))].emplace_back(handler);
  }
  void distribute(std::shared_ptr<const Base> basePtr)
  {
    const Base& base = *basePtr;
    std::type_index typeIdx(typeid(base));

    for(auto& handler : handlerMap[typeIdx])
    {
      handler->handle(basePtr);
    }
  }
private:
  std::unordered_map<std::type_index, std::vector<std::shared_ptr<BaseHandler>>> handlerMap;
};

// ---------------------------------------------------------------------
// Benchmarking code
// ---------------------------------------------------------------------

// Test handler function
void handleDerived(std::shared_ptr<const Derived> derived) { }

int main ()
{
  size_t iters = 10000000;
  size_t numRuns = 10;

  Distributor distributor;

  // add our test handler
  distributor.addHandler(std::make_shared<Handler<Derived>>(&handleDerived));

  std::cout << "Raw Func Call\t|\tDistributor\t|\tRatio" << std::endl;
  std::cout << "-------------\t|\t-----------\t|\t-----" << std::endl;

  for(size_t i = 0; i < numRuns; i++)
  {
    auto evt = std::make_shared<Derived>();

    // time raw function calls
    auto start = std::chrono::steady_clock::now();
    for (size_t i = 0; i < iters; i++) {
      handleDerived(evt);
    }
    auto d = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start);

    // time calls through the distributor
    start = std::chrono::steady_clock::now();
    for (size_t i = 0; i < iters; i++) {
      distributor.distribute(evt);
    }
    auto d2 = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start);

    std::cout << d.count() << "\t\t|\t" << d2.count() << "\t\t|\t" << (d2*1.0/d) << std::endl;
  }


}

在运行 MinGW-W64 g++ 8.1.0 的 Windows 10 机器上使用 -O3 标志优化的结果:

Raw Func Call   |       Distributor     |       Ratio
-------------   |       -----------     |       -----
256             |       1256            |       4.90625
258             |       1224            |       4.74419
273             |       1222            |       4.47619
246             |       1261            |       5.12602
270             |       1257            |       4.65556
248             |       1276            |       5.14516
272             |       1274            |       4.68382
265             |       1208            |       4.55849
240             |       1224            |       5.1
239             |       1163            |       4.86611

如您所见,分发器调用开销会导致大约 4.5-5 倍的减速(与从指向非const 的指针到指向const 的指针的所需转换相比)。不过,是否有任何明确的方法可以在保持给定设计模式的同时改进这一点?

应该给处理程序shared_ptrs,因为如果他们愿意,我希望他们能够保留对传递对象的引用。但他们可能真的想要也可能不想保留对它的引用。

我想知道是否有某种方法可以通过避免 shared_ptr 复制构造来提高性能,但我不确定最好的方法。

编辑:这个设计有几个方面对我来说非常重要。它们如下:

  1. 我的实际用例要求原来的shared_ptr必须是指向非const的指针,处理程序收到的shared_ptr必须是指向const 的指针。因此,我本质上是在比较 distribute 调用的成本与调用将导致该转换作为参考点的函数的成本。
  2. Distributor 类的用户不需要需要担心强制转换。任何到 Base 并返回到 Derived 类的转换都应该对用户不可见。
  3. 我愿意支持几乎所有种类的处理函数(lambda、函子、成员函数、函数指针等),但如果限制性更强的性能优势显着,可能会改变我的想法。

代码的其他方面(如注册侦听器)的效率改进也是受欢迎的,但不是那么重要。最令人担忧的是让Distributor 尽可能高效地呼叫所有侦听器。

【问题讨论】:

  • std::make_shared&lt;Derived&gt;() vs std::make_unique&lt;Derived&gt;() -- 几乎你所有的时间实际上都花在了分配内存上,而慢的分配了两次内存,所以慢了大约 2 倍。 unique_ptr 可以转换为shared_ptr,但这需要为 RC 块分配内存。直接调用make_shared 会导致RC 块与对象连续分配。与大多数微基准测试一样,您并不是在衡量自己关心的内容。
  • @Yakk-AdamNevraumont 感谢您指出这一点!那是我的代码中的一个错误。我已经更新它以在测试中保持一致。这大大改善了结果。
  • 我想到的一件事是更改参数以通过 const 引用获取shared_ptr,以避免增加/减少计数器。
  • 如果我没记错的话,您当前的解决方案仍然分配了很多,因此您测量的是分配速度,而不是函数调用速度。从内循环中去掉make_shared调用,你会发现raw/std::function调用之间的区别要大得多。
  • @geza 这是我的另一个错误。不应该衡量分配。我应该通过 Yakk 的评论意识到这一点……但无论如何,谢谢!此问题已得到纠正。

标签: c++ c++14


【解决方案1】:

附注:

当一个函数按值获取std::shared_ptr 时,涉及追逐指针(潜在的缓存未命中)和原子增量(相对昂贵的操作)。避免按价值接受std::shared_ptr

首先,改变:

void distribute(std::shared_ptr<const Base> basePtr)

到:

void distribute(std::shared_ptr<const Base> const& basePtr)

然后在其他地方。


不过,您可以将直接致电handleDerived 的费用与以下情况进行比较:

  • 拨打typeid
  • 哈希查找,
  • 对向量进行迭代,
  • 虚拟通话,
  • 通过函数指针调用。

这是很多开销。你可以通过避免那些虚拟调用来减少它:

#include <iostream>
#include <memory>
#include <functional>
#include <vector>
#include <typeindex>
#include <unordered_map>
#include <chrono>

struct Base {
    virtual ~Base() {};
};
struct Derived :  Base {};

class Distributor
{
public:
    template <class T, typename F>
    void addHandler(F&& handler) {
        handlerMap[std::type_index(typeid(T))].emplace_back(std::forward<F>(handler));
    }

    void distribute(std::shared_ptr<const Base> const& basePtr) {
        std::type_index typeIdx(typeid(*basePtr));
        for(auto& handler : handlerMap[typeIdx])
            handler(basePtr);
    }

private:
    std::unordered_map<std::type_index, std::vector<std::function<void(std::shared_ptr<const Base> const&)>>> handlerMap;
};

void handleDerived(std::shared_ptr<const Derived> const&) { }

int main ()
{
    size_t iters = 10000000;
    size_t numRuns = 10;

    Distributor distributor;

    // add our test handler
    distributor.addHandler<Derived>([](std::shared_ptr<const Base> const& p) { 
        handleDerived(std::static_pointer_cast<const Derived>(p)); 
    });

    std::cout << "Raw Func Call\t|\tDistributor\t|\tRatio" << std::endl;
    std::cout << "-------------\t|\t-----------\t|\t-----" << std::endl;

    for(size_t i = 0; i < numRuns; i++)
    {
        auto evt = std::make_shared<Derived>();

        // time raw function calls
        auto start = std::chrono::steady_clock::now();
        for (size_t i = 0; i < iters; i++) {
            handleDerived(evt);
        }
        auto d = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start);

        // time calls through the distributor
        start = std::chrono::steady_clock::now();
        for (size_t i = 0; i < iters; i++) {
            distributor.distribute(evt);
        }
        auto d2 = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start);

        std::cout << d.count() << "\t\t|\t" << d2.count() << "\t\t|\t" << (d2*1.0/d) << std::endl;
    }
}

输出:

Raw Func Call   |       Distributor     |       Ratio
-------------   |       -----------     |       -----
72              |       238             |       3.30556
72              |       238             |       3.30556
72              |       238             |       3.30556
72              |       238             |       3.30556
72              |       238             |       3.30556
72              |       238             |       3.30556
72              |       238             |       3.30556
72              |       238             |       3.30556
72              |       238             |       3.30556
72              |       238             |       3.30556

在我的机器上,初始比率是 4.5。

【讨论】:

  • 这确实提高了性能。我按照您的建议更改了 distribute 函数和 handle 函数,现在我看到大约 4 倍的减速。谢谢!
  • 确实,我预计会产生一些开销。我要寻求帮助最小化该开销。虽然,正如我认为在 cmets 中提到的那样,对 handleDerived 的调用并没有完全内联为 0,因为它必须将 shared_ptr&lt;Derived&gt; 转换为 shared_ptr&lt;const Derived&gt;,这就是编译器无法完全消除它的原因。
  • @NanoWizard 你说得对。为你更新了答案。
  • 这是一个非常有用的比较。但是,我不想放弃设计的某些方面,您已将其删除。例如,对我来说重要的是,BaseDerived 的任何转换对于 Distributor 类的用户都是不可见的。尽管如此,还是很高兴了解这些虚拟呼叫在性能方面的成本。谢谢!
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2019-07-12
  • 2018-06-20
  • 1970-01-01
  • 2020-06-06
  • 1970-01-01
  • 1970-01-01
  • 2020-11-14
相关资源
最近更新 更多