提升 asio io_service 多线程性能不佳答案

【问题标题】：Boost asio io_service mutithreading poor performance提升 asio io_service 多线程性能不佳
【发布时间】：2018-06-13 02:06:23
【问题描述】：

我正在使用元胞自动机并尝试通过使用多线程来提高我的应用程序性能。但我有一些有趣的结果。我真的不知道为什么会发生这种情况以及我错过了什么......

所以我的目标是尽可能快地处理大量数据。在我的示例中，我有一个大 (20000 x 20000) 布尔数组并将其转换为图像（一个布尔值到一个像素）。这个过程可以并行进行；像素之间没有任何依赖关系。我将 bool 数组分成threadCount 个块，为每个块启动一个新线程，让它们运行，等待它们完成。

我假设使用更多线程我会获得更好的运行时。（我没有使用不切实际的线程数，只是在 1 和逻辑核心数之间。）

所以我写了这个：

typedef std::size_t Size;
typedef std::vector<bool> Data;
typedef std::vector<Data> History;

class RenderTask
{
  public:
    typedef void result_type;

  public:
    RenderTask(Ppm& ppm,
               const Ppm::Pixel& fColor)
      : mPpm(ppm),
        mForegroundColor(fColor)
    {
    }

    void operator()(const History& history,
                    const Size minIdxX,
                    const Size countX,
                    const Size minIdxY,
                    const Size countY)
    {
      const Size maxIdxX(minIdxX + countX);
      const Size maxIdxY(minIdxY + countY);
      for(Size y(minIdxY); y < maxIdxY; ++y)
      {
        for(Size x(minIdxX); x < maxIdxX; ++x)
        {
          if(history[y][x])
          {
            mPpm.setPixel(x, y, mForegroundColor);
          }
        }
      }
    }

  private:
    Ppm& mPpm;
    const Ppm::Pixel mForegroundColor;
};

void render(const History& history,
            Ppm& ppm,
            const Ppm::Pixel& fColor,
            const Size threadCount)
{
  boost::asio::io_service io_service;
  boost::thread_group threads;
  for(Size i(0); i < threadCount; ++i)
  {
    threads.create_thread(boost::bind(&boost::asio::io_service::run,
                                      &io_service));
  }

  RenderTask task(ppm, fColor);
  io_service.reset();

  const Size count(history.size() / threadCount);
  const Size rem(history.size() % threadCount);
  Size minIdxY(0);
  for(Size i(0); i < threadCount; ++i)
  {
    const bool addRemainders(rem && i == threadCount - 1);
    io_service.post(boost::bind(task,
                                boost::cref(history),
                                0,
                                history.front().size(),
                                minIdxY,
                                addRemainders ? count + rem : count));
    minIdxY += count;
  }

  threads.join_all();
}

int main(int argc, char* argv[])
{
  const Size rule(parseNumber<Size>(argv[1]));
  const Size size(parseNumber<Size>(argv[2]));
  const Size iteration(parseNumber<Size>(argv[3]));
  const Size threadCount(clamp(1,
                               static_cast<Size>(boost::thread::physical_concurrency())
                               parseNumber<Size>(argv[4])));
  ...
  History history(iteration, Data(size, false));
  history.front()[size / 2] = true;
  ...
  process(history, rule, threadCount);
  ...
  Ppm ppm(history.front().size(), history.size(), Ppm::Pixel(30, 30, 30));
  std::cout << "rendering... ";
  t.start();
  render(history, ppm, Ppm::Pixel(200, 200, 200),  threadCount);
  t.stop();
  std::cout << t.ms() << " ms" << std::endl;
}

但是当我使用不同数量的线程运行程序时，我得到了以下结果：

我不知道为什么更多的内核不能带来更好的性能。两核会好一些，但有趣的是，三核几乎与单核相同......这些值是平均值：

ThreadingTest.exe 110 20000 20000 1 test.ppm
rendering... 554.95 ms

ThreadingTest.exe 110 20000 20000 2 test.ppm
rendering... 289.75 ms

ThreadingTest.exe 110 20000 20000 3 test.ppm
rendering... 555.37 ms

ThreadingTest.exe 110 20000 20000 4 test.ppm
rendering... 554.23 ms

ThreadingTest.exe 110 20000 20000 5 test.ppm
rendering... 564.23 ms

ThreadingTest.exe 110 20000 20000 6 test.ppm
rendering... 551.82 ms

ThreadingTest.exe 110 20000 20000 7 test.ppm
rendering... 555.22 ms

ThreadingTest.exe 110 20000 20000 8 test.ppm
rendering... 510.12 ms

什么会导致这种情况？我是否以错误的方式使用 io_service？不涉及 I/O 操作，只是纯内存。

我的机器中有 8 个内核和 16 GB 的 RAM。

更多细节，这里是 Ppm 类的概要：

class Ppm
{
  public:
    struct Pixel
    {
      typedef unsigned char ChannelType;
      ChannelType  mRed, mGreen, mBlue;
      ...
    };

    typedef std::vector<Pixel> ImageData;

    Ppm( const SizeType width
         , const SizeType height
         , const Pixel& color = Pixel() )
      : mWidth( width )
      , mHeight( height )
      , mImageData( mWidth * mHeight, color )
    { }

    void setPixel( SizeType x, SizeType y, const Pixel& p )
    {
      mImageData[x + y * mWidth] = p;
    }
    ...
  private:
    const SizeType mWidth;
    const SizeType mHeight;
    ImageData mImageData;
};

更新

在您宝贵的 cmets 之后，我改变了很多方法并写了这个：现在我正在使用纯 c++'11 的东西，不再涉及任何提升..

class ThreadPool
{
  public:
    ThreadPool(const Size threadCount);
    ~ThreadPool();

  public:
    template<class T>
    void addTask(T task);
    void wait();

  private:
    bool mStopped;
    Size mRunningCount;
    std::vector<std::thread> mWorkers;
    std::deque<std::function<void()>> mTasks;
    std::mutex mMutex;
    std::condition_variable mCondition;
    std::condition_variable mFinishCondition;
};

ThreadPool::ThreadPool(const Size threadCount)
  : mStopped(false),
    mRunningCount(0)
{
  for (Size i(0); i < threadCount; ++i)
  {
    mWorkers.push_back(std::thread([this]()
      {
        std::function<void()> task;
        while(true)
        {
          {
            std::unique_lock<std::mutex> lock(this->mMutex);

            this->mCondition.wait(lock, [this] { return this->mStopped ||
                                                        !this->mTasks.empty();
                                               });
            if(this->mStopped)
            {
              return;
            }

            ++this->mRunningCount;
            task = this->mTasks.front();
            this->mTasks.pop_front();
          }

          task();

          {
            std::unique_lock<std::mutex> lock(this->mMutex);
            --this->mRunningCount;
          }

          this->mFinishCondition.notify_all();
      }
    }));
  }
}

ThreadPool::~ThreadPool()
{
  {
    std::unique_lock<std::mutex> lock(mMutex);
    mStopped = true;
    mCondition.notify_all();
  }

  for(auto& worker : mWorkers)
  {
    worker.join();
  }
}

template<class T>
void ThreadPool::addTask(T task)
{
  {
    std::unique_lock<std::mutex> lock(mMutex);
    mTasks.push_back(std::function<void()>(task));
  }
  mCondition.notify_one();
}

void ThreadPool::wait()
{
  std::unique_lock<std::mutex> lock(mMutex);
  mFinishCondition.wait(lock, [this]()
    {
      return mTasks.empty() && mRunningCount == 0;
    });
}

现在性能还可以；使用更多线程运行时变得更快...... 但是等待方法有一些不好的地方。我是这样使用的：

ThreadPool pool(threadCount);
for(Size i(1); i < iteration; ++i)
{
  Size count(history.front().size() / threadCount);
  Size rem(history.front().size() % threadCount);
  Size minIdx(0);
  for(Size n(0); n < threadCount; ++n)
  {
    pool.addTask(std::bind(ECATask(rule),
                           std::cref(history[i-1]),
                           std::ref(history[i]),
                           minIdx,
                           (rem && n == threadCount - 1) ?
                             count + rem :
                             count));
    minIdx += count;
  }
  pool.wait();
}

这个问题还不清楚，但似乎pool.wait() 有时不会等待所有当前任务完成并且代码开始新的迭代......你能帮我做一个代码审查吗？ :)

【问题讨论】：

代码中使用了哪个clamp -- std::clamp？如果是这样，则参数的顺序似乎不正确。另外，boost::thread::physical_concurrency() 返回什么值？
什么是Ppm？确定在调用mPpm.setPixel(x, y, mForegroundColor);的时候可以同时从不同线程修改吗？
也尝试对线程创建循环进行计时。创建线程并不是世界上最快的事情，而且由于创建线程池的成本很高，大多数线程池的寿命往往很长。
我同意@UKMonkey ...您应该只对并发操作计时，即使 thread.wait 会导致延迟，对相同的配置使用多次运行的 exe 以获得平均结果。除了提升的线程 CPU 亲和力之外，您可能还需要通过 native_handle 进行调整才能看到更好的结果（？）
我怀疑您的瓶颈在于处理图形子系统。图形库有异步接口吗？

标签： c++ multithreading boost boost-asio

【解决方案1】：

有很多迹象表明您对 io_service 感到困惑：

  boost::asio::io_service io_service;
  boost::thread_group threads;
  for(Size i(0); i < threadCount; ++i)
  {
    threads.create_thread(boost::bind(&boost::asio::io_service::run
                                      , &io_service));
  }

  RenderTask task(ppm, fColor);
  io_service.reset();

问题：

你不应该在那里调用 reset()
线程在 io_service 开始工作之前启动。这意味着run() 立即完成，退出线程。

所以整个事情是一个巨大的竞争条件：如果你幸运的话，一些线程（可能没有线程）不会在第一个任务入队之前启动io_service::run()，所以这一切似乎都可以工作。

在不使用 Boost Asio 和使用 Boost Asio 的情况下查看线程池的良好并排答案：c++ work queues with blocking

注意使用io_service::work 以避免过早退出工作线程。

【讨论】：