c++ - 如何提高分发到 std::function 监听器的性能？

Question

简而言之，有什么明显的方法可以使 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 并使用 -O3 优化的 Windows 10 机器上的结果标志:

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_ptr因为如果他们愿意，我希望他们能够保留对传递的对象的引用。但他们可能真的想也可能不想保留对它的引用。

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

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

1. 我的实际用例要求原始 shared_ptr 必须 是指向非 const 的指针, 和 shared_ptr处理程序接收到的必须是指向const的指针.因此，我基本上是在比较 distribute 的成本。反对调用函数的成本，该函数作为引用点发生该转换。
2. Distributor 的用户类应该不需要担心转换。任何转换到Base然后回到 Derived类应该对用户不可见。
3. 我愿意支持几乎所有种类的处理程序函数(lambda、仿函数、成员函数、函数指针等)，但如果限制性更强的性能优势显着，我可能会改变主意。

代码的其他方面(如注册监听器)的效率改进也很受欢迎，但不是必需的。最令人担忧的是 Distributor尽可能高效地调用所有听众。

Answer 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调用，
• 哈希查找，
• 迭代一个 vector ，
• 虚拟电话，
• 通过函数指针调用。

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

#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。