c++ - boost asio io_service 多线程性能不佳

Question

我正在玩元胞自动机，并尝试通过使用多线程来 boost 我的应用程序性能。但我有一些有趣的结果。我真的不知道为什么会这样，也不知道我错过了什么......

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

我假设使用更多线程我会获得更好的运行时间。 (我没有使用不切实际的线程数，只是在 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 个内核，RAM 为 16 GB。

有关更多详细信息，这是 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;
};

更新

在您提出宝贵意见后，我改变了很多方法并写下了这个:现在我正在使用纯 c++'11 的东西，不再涉及任何 boost ..

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;
    });
}

现在性能还可以；使用更多线程运行时变得更快......但是 wait 方法有些不好。我以这种方式使用它:

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() 有时不等待所有当前任务完成并且代码开始新的迭代......可以请你为我做一次代码审查？ :)

Answer 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

Note the use of io_service::work to avoid prematurely exiting the worker threads.