如何在 C++ 中并行化一个 for 循环，只创建一次线程池

Question

我有一个必须在 Windows 中运行的 C++ 程序。

我有一个游戏，在函数WinMain 中有一个主循环，它在每次迭代中调用Update 函数。我想将多线程应用到更新函数的循环中。

int __stdcall WinMain()
{
    // Windows initializations

    // Main loop
    {
        Game game = new Game();
        bool bExit = false;
        while (!bExit)
        {
            MSG msg;
            while (PeekMessage(&msg, NULL, 0, 0, PM_REMOVE))
            {
                if (msg.message == WM_QUIT) bExit = true;
                TranslateMessage(&msg);
                DispatchMessage(&msg);
            }

            game->Update();
        }
        delete game;
        game = nullptr;
    }

    // Windows destructions

    return 0;
}

// Game.cpp

void Game::Update() {
    // Loop that I want to be parallelized but I don't want to create the thread pool here because it's called on every frame
    for (size_t i = 0; i < entities.size(); i++) {
        entities[i]->Update();
    }
}

我尝试使用 OpenMP，但我无法将 #pragma omp parallel 放在 WinMain 函数中（应用程序会崩溃），如果我在循环之前将 #pragma omp parallel for 放在 Game::Update 中，它实际上会降低性能，因为它在每一帧中创建线程循环。

我正在寻找一种基于库的解决方案，或者最好是本机解决方案，它可以让我轻松地并行化这个循环。

唯一的要求是它可以在 Windows 中运行，我宁愿不使用库（比如 boost，虽然 OpenMP 很好）。

编辑：我让它与 PPL 的 concurrency::parallel_for 一起工作。它不会降低性能，但也不会提高性能...我对此的问题是：此函数是否在每次调用时创建和销毁其线程池？

Answer 1

根据我们在 cmets 中的讨论提供一个 Boost.Asio 示例：

#include <thread>

#include <boost/asio.hpp>

namespace asio = boost::asio;

int main( int argc, int* argv[] ) {
    asio::io_context context{};

    asio::post( context, []() { /* any arbitrary job */ } );

    // this donates the new_thread to the thread pool
    std::thread new_thread{ [&](){ context.run(); } };

    // this donates the current thread to the thread pool
    context.run();
}

这里需要注意的主要事项是：

1. asio::post 允许您向io_context 提交任意作业
2. 从线程调用 io_context::run 会将线程捐赠给运行上下文的线程池

如果你想要一个预建的线程池，你可以使用boost::asio::thread_pool，并调用asio::post以同样的方式将作业放到线程池中。

您应该只需要下载 Boost，不需要安装它。如果您将BOOST_ERROR_CODE_HEADER_ONLY 添加到您的编译中，Boost.Asio 完全是仅标题。

编辑：

您仍然需要运行 Boost 的设置脚本，但您不需要构建任何库。

Answer 2

这里有一个基本线程库的解决方案。我模拟了您的 Entity 和 Game 课程。
请注意，在此解决方案中，工作线程会在开始时创建并启动一次。他们将在每个Update 电话中被调用。当Update未被调用时，线程休眠...

我已尽力保留您程序的架构。 请注意，我们可以使用 std::thread、std::mutex 的另一种实现......但我只是想给你一个想法......

#define NB_ENTITIES 10

class CEntity
{
public:
    void Update(){};
    ~CEntity () {}
};

typedef struct ThreadData
{
    HANDLE  hMutex;
    HANDLE  hMutexDestructor;
    CEntity *pCEntity;  
    DWORD   *dwStat;

} ThreadData;

//------------------------------------
// This function will call "Upadate"
//------------------------------------
DWORD WINAPI MyThreadFunction( LPVOID lpParam )
{
    ThreadData *ThreadDatpa      = (ThreadData*) lpParam;
    CEntity    *pEntity          = ThreadDatpa->pCEntity;
    HANDLE     hMutex            = ThreadDatpa->hMutex;
    HANDLE     hMutexDestructor  = ThreadDatpa->hMutexDestructor;
    DWORD      *dwStat           = ThreadDatpa->dwStat;

    while (true)
    {
        // When no update, thread sleep ... 0% CPU ...
        WaitForSingleObject(hMutex, INFINITE);
        if ( 0 == *dwStat ) break; // here thread stat for stopping 

        if ( nullptr != pEntity ) 
            pEntity->Update(); // Call your unpdate function ...
    }

    // Each worker thread must release it semaphore.
    // Destructor must get ALL released semaphore before deleting memory
    ReleaseSemaphore(hMutexDestructor, 1, NULL );

    return 0;
}

class Game 
{
public :
    vector<ThreadData*>  entities; // Vector of entities pointers
    vector<HANDLE>   thread_group; // vector of threads handle


    //This function must called ONE time at the beginning (at init)
    void StartThreads ()
    {
        DWORD  dwRet = 0;
        HANDLE hTemp = NULL;

        for (size_t i = 0; i <NB_ENTITIES; i++) 
        {
            CEntity *pCEntity = new CEntity (); // just to simulate entity

            // This semaphore is used to release thread when update is called
            HANDLE ghMutex= CreateSemaphore( NULL,0, 1, NULL); 

            // This semaphore is used when destruction to check if all threads is terminated
            HANDLE ghMutexDestructor= CreateSemaphore( NULL,0, 1, NULL);

            // create a new CEntity data ...
            ThreadData *pThreadData = new ThreadData ();

            pThreadData->pCEntity = pCEntity;
            pThreadData->hMutex   = ghMutex;
            pThreadData->hMutexDestructor = ghMutexDestructor;
            pThreadData->dwStat   = new DWORD (1); // default status = 1 

            entities.push_back ( pThreadData );         
        }

        // Here we start ONE time Threads worker.
        // Threads are stopped untile update was called
        for (size_t i = 0; i < entities.size(); i++) 
        {
            // Each thread has it own entity
            hTemp = CreateThread( NULL,0, MyThreadFunction, entities.at(i), 0, &dwRet);  
            if ( NULL != hTemp )
                thread_group.push_back (hTemp);
        }

    }

    // Your function update juste wakeup threads
    void Game::Update() {
        for (size_t i = 0; i < entities.size(); i++) 
        {
            HANDLE hMutex = entities.at(i)->hMutex;
            if ( NULL != hMutex )
                ReleaseSemaphore(hMutex, 1, NULL );
        }
    }


    ~Game()
    {
        // Modifie stat before releasing threads
        for (size_t i = 0; i < entities.size(); i++) 
            *(entities.at(i)->dwStat) = 0;

        // Release threads (status =0 so break ...)
        Update();

        // This can be replaced by waitformultipleobjects ...
        for (size_t i = 0; i < entities.size(); i++)
            WaitForSingleObject ( entities.at(i)->hMutexDestructor, INFINITE);

        // Now i'm sur that all threads are terminated
        for (size_t i = 0; i < entities.size(); i++)
        {
            delete entities.at(i)->pCEntity;
            delete entities.at(i)->dwStat;
            CloseHandle (entities.at(i)->hMutex);
            CloseHandle (entities.at(i)->hMutexDestructor);
            delete entities.at(i);
        }

    }
};