一個線程是否可以被重用來運行可變參數函數？

Question

我試圖創建在C升壓線程++，可重複使用，運行多種，可以有不同的數量和args的類型的功能。

可以這樣用C++ 11X variadics做了什麼？我不需要一個隊列（如果線程繁忙，那麼方法就會失敗），但如果需要實現這個「統一」功能，我會不情願地這樣做。

我不明白如何處理與統一綁定或lambda讓一個線程可以調用不同的函數，每個人都有自己不同數量和類型參數的個數。

我已經大致記住以下幾點：

class WorkThread 
{ 
public: 
    WorkThread() 
     { 
     // create thread and bind runner to thread 
     } 

    ~WorkThread() 
     { 
     // tell runner to exit and wait and reap the thread 
     } 

    template<typename F,typename ... Arguments> 
    void doWork(F func, Arguments... args) 
     { 
     if already busy 
      return false; 
     // set indication to runner that there is new work 
     // here: how to pass f and args to runner? 
     } 

private: 
    void runner() 
     { 
     while (! time to quit) 
      { 
      wait for work 
      // here: how to get f and args from doWork? do I really need a queue? could wait on a variadic signal maybe? 
      f(args); 
      } 
     } 

    boost::thread* m_thread; 
}; 

class ThreadPool 
{ 
public: 
    template<typename F, typename ... Arguments> 
    bool doWork(F func,Arguments... args) 
      { 
      const int i = findAvailableWorkThread(); 
      m_thread[i].doWork(f,args); 
      } 
private: 
    // a pool of work threads m_thread; 
}; 

Answer 1

應該有很多現有的問題顯示如何做到這一點。

的規範C++ 11來表示一個任意的功能對象的方法是std::function<void()>，所以要該類型的共享對象，稱爲m_job在下面的代碼，這應該由互斥保護，你分配新作業它當它尚未設置：

template<typename F,typename ... Arguments> 
bool doWork(F func, Arguments&&... args) 
{ 
    std::lock_guard<std::mutex> l(m_job_mutex); 
    if (m_job) 
     return false; 
    m_job = std::bind(func, std::forward<Arguments>(args)...); 
    m_job_cond.notify_one(); 
    return true; 
} 

這使用std::bind把一個函數對象及其參數輸入不帶參數的函數對象。當調用它通過std::bind商店func副本，每個參數和返回的調用對象調用func(args...)

然後工人少了點：

void runner() 
{ 
    while (! time to quit) 
    { 
     std::function<void()> job; 
     { 
      std::unique_lock<std::mutex> l(m_job_mutex); 
      while (!m_job) 
       m_job_cond.wait(l); 
      swap(job, m_job); 
     } 
     job(); 
    } 
} 

此代碼是不是線程安全的：

template<typename F, typename ... Arguments> 
    bool doWork(F func,Arguments... args) 
    { 
     const int i = findAvailableWorkThread(); 
     m_thread[i].doWork(f,args); 
    } 

後findAvailableWorkThread回報，該線程可能會成爲忙，所以下一行將會失敗。您應該檢查可用性，並在一次操作中通過新作業，例如

template<typename F, typename ... Arguments> 
    bool doWork(F func,Arguments... args) 
    { 
     for (auto& t : m_thread) 
      if (t.doWork(f,args)) 
       return true; 
     return false; 
    } 

Answer 2

您可以通過使用boost::bind或C++ 11 lambda無論是統一所有的功能，一個簽名。所以你將有一個循環函數的線程，並且在這個循環中你將提取你想要執行的函數。 爲了達到這個目的，你可以使用boost::lockfree::queue<boost::function<void()>>，例如。

正如想法的例子，你可以使用以下命令：

class TaskLoop 
{ 
public: 
    typedef std::function<void()> Task_t; 
public: 
    TaskLoop(): 
     m_IsDone(false) 
    { 
     m_spThread.reset(new std::thread(&TaskLoop::_run, this)); 
    } 
    ~TaskLoop() 
    { 
     Task_t task = [this](){m_IsDone = true;}; 
     postTask(task); 
     m_spThread->join(); 
    } 
    void postTask(const Task_t& Msg) 
    { 
     std::lock_guard<std::mutex> lock(m_Mutex); 
     m_Tasks.push(Msg); 
    } 
    void wait() 
    { 
     while(!m_Tasks.empty()); 
    } 
private: 
    bool m_IsDone; 
    std::unique_ptr<std::thread> m_spThread; 
    std::mutex m_Mutex; 
    std::queue<Task_t> m_Tasks; 
private: 
    void _run() 
    { 
     while(!m_IsDone) 
     { 
      Task_t task; 
      m_Mutex.lock(); 
      if(!m_Tasks.empty()) 
      { 
       task = m_Tasks.front(); 
       m_Tasks.pop(); 
      } 
      m_Mutex.unlock(); 
      if(task) 
       task(); 
     } 
    } 
}; 

void foo(const std::string& first, int second) 
{ 
    std::cout << first << second << "\n"; 
} 

int main(int argc, char **argv) 
{ 
    TaskLoop loop; 
    loop.postTask([]{foo("task", 0);}); 
    loop.wait(); 
    return 0; 
} 

示例不使用併發隊列和相當簡單，所以你需要更換隊列和適應它您的要求

Answer 3

正如你已經在使用升壓，請參閱http://think-async.com/Asio/Recipes線程池類使用Boost.Asio的排隊工作職能沒有鎖定。

Answer 4

//這裏是我到目前爲止，似乎工作....

#include <boost/thread.hpp> 
#include <boost/thread/mutex.hpp> 
#include <boost/thread/condition.hpp> 

#include <queue> 
#include <atomic> 

class ThreadWorker 
    { 
    public: 
     typedef boost::function<void()> Task_t; 
     typedef std::unique_ptr<boost::thread> UniqueThreadPtr_t; 

     ThreadWorker() 
      : m_timeToQuit(false) 
      { 
      m_spThread = UniqueThreadPtr_t(new boost::thread(std::bind(&ThreadWorker::runner, this))); 
      } 

     virtual ~ThreadWorker() 
      { 
      quit(); 
      } 

     void quit() 
      { 
      Task_t task = [this]() { m_timeToQuit = true; }; 
      enqueue(task); 
      m_spThread->join(); 
      } 

     template<typename F,typename ...Args> 
     void enqueue(F&& f, Args&&... args) 
      { 
      boost::mutex::scoped_lock lock(m_jobMutex); 
      m_Tasks.push(std::bind(std::forward<F>(f),std::forward<Args>(args)...)); 
      m_newJob.notify_one(); 
      } 

    private: 
     std::atomic<bool>  m_timeToQuit; 
     UniqueThreadPtr_t  m_spThread; 
     mutable boost::mutex m_jobMutex; 
     std::queue<Task_t>  m_Tasks; 
     boost::condition  m_newJob; 

    private: 
     void runner() 
      { 
      while (! m_timeToQuit) 
       { 
       Task_t task; 
        { 
        boost::mutex::scoped_lock lock(m_jobMutex); 
        while (! task) 
         { 
         if (m_timeToQuit) 
          { 
          break; 
          } 
         if (m_Tasks.empty()) 
          { 
          m_newJob.wait(lock); 
          } 
         task = m_Tasks.front(); 
         m_Tasks.pop(); 
         } 
        } 
       if (task) 
        { 
        task(); 
        } 
       } 
      } 
    };