C ++ ThreadPool不是并行运行

问题描述

这是我已经有的了。

  //包括... 
 
 void call（）
 {
 std :: cout< ;& Hi i'm thread no< std :: this_thread :: get_id（）<< std :: endl; 
 std :: this_thread :: sleep_for（std :: chrono :: seconds（2））; 
 std :: cout<< ready<< std :: this_thread :: get_id（）<< std :: endl; 
}; 
 
 //这里不显示实现以减少代码
 class WorkQueue {
 public：
 bool push（std :: function< void（）> const&值）; 
 void pop（）; 
 bool empty（）; 
}; 
 
 std :: condition_variable g_queuecheck; 
 std :: mutex g_lockqueue; 
 std :: atomic< bool> g_notified; 
 
 class ThreadPool 
 {
 public：
 ThreadPool（int iNoThread）：
 m_noThread（iNoThread）
 {
 g_notified。 store（false）; 
 m_threads.resize（iNoThread）; 
 bIsReady.store（false）; 
 for（int i = 0; i  m_threads [i] = std :: thread（& ThreadPool :: run，this）; 
} 
 
 void run（）
 {
 while（！bIsReady ||！m_workQueue.empty（））
 {
 std： ：unique_lock< std :: mutex> locker（g_lockqueue）; 
 if（m_workQueue.empty（））
 {
 while（！g_notified）//用于避免虚假唤醒
 {
 g_queuecheck.wait（locker）; 
} 
 if（！bIsReady）
 g_notified.store（false）; 
} 
 
 m_workQueue.pop（）; 
} 
}; 
 
 void addWork（std :: function< void（）> func）
 {
 m_workQueue.push（func）; 
 g_notified.store（true）; 
 g_queuecheck.notify_one（）; 
} 
 
 void join（）
 {
 bIsReady.store（true）; 
 g_notified.store（true）; 
 g_queuecheck.notify_all（）; 
 
 for（int i = 0; i  m_threads [i] .join（）; 
} 
 
〜ThreadPool（）
 {} 
 
 
 WorkQueue m_workQueue; 
 int m_noThread; 
 std :: vector< std :: thread> m_threads; 
 std :: atomic< bool> bIsReady; 
}; 
 
 int _tmain（int argc，_TCHAR * argv []）
 {
 {
 ThreadPool pool（4） 
 
 for（int i = 0; i  pool.addWork（call）; //这项工作按顺序完成
 
 pool.join（）; 
} 
 
 std :: cin.ignore（）; 
 return 0; 
} 
  

我的问题是工作是按顺序完成的。

$ b

我的ThreadPool有其他问题吗？

li>等待最佳实践？

解决方案

我使用boost :: asio一个线程池。希望这可以帮助。此实现是从中收集的。我的例子工作的关键是范围的asio :: io_service ::工作，并有在范围之外的join_all。

  #include< boost / chrono.hpp> 
 #include< boost / thread.hpp> 
 #include< boost / asio.hpp> 
 
 
 boost :: mutex output_mutex; 
 void call（）
 {
 {
 boost :: mutex :: scoped_lock print_lock（output_mutex）; 
 std :: cout<< Hi i'm thread no< boost :: this_thread :: get_id（）<< std :: endl; 
} 
 
 boost :: this_thread :: sleep_for（boost :: chrono :: seconds（2））; 
 {
 boost :: mutex :: scoped_lock print_lock（output_mutex）; 
 std :: cout<< ready<< boost :: this_thread :: get_id（）<< std :: endl; 
} 
}; 
 
 int main（int argc，char ** argv）
 {
 size_t number_of_threads = boost :: thread :: hardware_concurrency（）; 
 
 boost :: asio :: io_service io_service; 
 boost :: thread_group threads; 
 {
 boost :: scoped_ptr< boost :: asio :: io_service :: work> work（new boost :: asio :: io_service :: work（io_service））; 
 for（size_t t = 0; t  {
 threads.create_thread（boost :: bind（& boost :: asio :: io_service :: run，& ; io_service））; 
} 
 
 for（size_t t = 0; t  {
 io_service.post（boost :: bind（call））; 
} 
} 
 threads.join_all（）; 
 
 return 0; 
} 
  

I've tried to implement a ThreadPool, but unfortunately I'm running into some problems.

This is what I have already.

//includes ...

void call()
{
    std::cout << "Hi i'm thread no " << std::this_thread::get_id() << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "ready " << std::this_thread::get_id() << std::endl;
};

//Implementation is not shown here to reduce code
class WorkQueue {
    public:
        bool push(std::function<void()> const& value);
        void pop();
        bool empty();
};

std::condition_variable g_queuecheck;
std::mutex              g_lockqueue;
std::atomic<bool>       g_notified;

class ThreadPool
{
    public:
        ThreadPool(int iNoThread) :
            m_noThread(iNoThread)
        {
            g_notified.store(false);
            m_threads.resize(iNoThread);
            bIsReady.store(false);
            for (int i = 0; i < iNoThread; ++i)
                m_threads[i] = std::thread(&ThreadPool::run, this);
        }

        void run()
        {
            while (!bIsReady || !m_workQueue.empty())
            {
                std::unique_lock<std::mutex> locker(g_lockqueue);
                if (m_workQueue.empty())
                {
                    while (!g_notified) // Used to avoid spurious wakeups
                    {
                        g_queuecheck.wait(locker);
                    }
                    if(!bIsReady)
                        g_notified.store(false);
                }

                m_workQueue.pop();
            }
        };

        void addWork(std::function<void()> func)
        {
            m_workQueue.push(func);
            g_notified.store(true);
            g_queuecheck.notify_one();
        }

        void join()
        {
            bIsReady.store(true);
            g_notified.store(true);
            g_queuecheck.notify_all();

            for (int i = 0; i < m_noThread; ++i)
                m_threads[i].join();
        }

        ~ThreadPool()
        {}


        WorkQueue m_workQueue;
        int m_noThread;
        std::vector<std::thread> m_threads;
        std::atomic<bool> bIsReady;
};

int _tmain(int argc, _TCHAR* argv[])
{
    {
        ThreadPool pool(4);

        for (int i = 0; i < 8; ++i)
            pool.addWork(call); //This work is done sequentially

        pool.join();
    }

    std::cin.ignore();
    return 0;
}

My problem is that the work is done sequentially.

1. How can I fix this?
2. Is something else wrong with my ThreadPool?
3. Is the waiting best-practice?

解决方案

I use boost::asio to implement a thread pool. Hope this helps. This implementation was gleaned from the Asio Thread Pool. The key for me to get the example to work is scoping the asio::io_service::work and have the join_all outside that scope.

#include <boost/chrono.hpp>
#include <boost/thread.hpp>
#include <boost/asio.hpp>


boost::mutex output_mutex;
void call()
{
    {
       boost::mutex::scoped_lock print_lock(output_mutex);
       std::cout << "Hi i'm thread no " << boost::this_thread::get_id() << std::endl;
    }

    boost::this_thread::sleep_for(boost::chrono::seconds(2));
    {
       boost::mutex::scoped_lock print_lock(output_mutex);
       std::cout << "ready " << boost::this_thread::get_id() << std::endl;
    }
};

int main(int argc, char **argv)
{
    size_t number_of_threads = boost::thread::hardware_concurrency();

    boost::asio::io_service io_service;
    boost::thread_group threads;
    {
        boost::scoped_ptr< boost::asio::io_service::work > work( new boost::asio::io_service::work(io_service) );
        for(size_t t = 0; t < number_of_threads; t++)
        {
            threads.create_thread(boost::bind(&boost::asio::io_service::run, &io_service));
        }

        for( size_t t = 0; t < number_of_threads; t++ )
        {
            io_service.post(boost::bind(call) );
        }
    }
    threads.join_all();

    return 0;
}

这篇关于C ++ ThreadPool不是并行运行的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！