一个线程是否可以被重用来运行可变参数函数？

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