通过同步延长线程的生命周期(C++ 11)

Question

我有一个程序,其函数将指针作为arg和main.主要是创建n个线程,每个线程根据传递的内容在不同的内存区域运行arg.然后连接线程,主要在区域之间执行一些数据混合,并创建n个新线程,这些线程执行与旧线程相同的操作.

为了改进程序,我想保持线程活着,消除创建它们所需的长时间.线程应在主要工作时休眠,并在必须再次出现时通知.同样,主要应该在线程工作时等待,就像连接一样.

我无法最终实现这一点,总是陷入僵局.

简单的基线代码,任何有关如何修改它的提示将非常感激

#include <thread>
#include <climits>

...

void myfunc(void * p) {
  do_something(p);
}

int main(){
  void * myp[n_threads] {a_location, another_location,...};
  std::thread mythread[n_threads];
  for (unsigned long int j=0; j < ULONG_MAX; j++) {
    for (unsigned int i=0; i < n_threads; i++) {
      mythread[i] = std::thread(myfunc, myp[i]);
    }
    for (unsigned int i=0; i < n_threads; i++) {
      mythread[i].join();
    }
    mix_data(myp); 
  }
  return 0;
}

Answer 1

这是一种仅使用C++ 11标准库中的类的可能方法.基本上,您创建的每个线程都有一个关联的命令队列(封装在std::packaged_task<>对象中),它会持续检查.如果队列为空,则线程将只在条件变量(std::condition_variable)上等待.

虽然通过使用std::mutex和std::unique_lock<>RAII包装器避免了数据争用,但主线程可以通过存储与std::future<>每个提交的对象相关联的对象std::packaged_tast<>并wait()在其上调用来等待特定作业终止.

以下是遵循此设计的简单程序.评论应该足以解释它的作用:

#include <thread>
#include <iostream>
#include <sstream>
#include <future>
#include <queue>
#include <condition_variable>
#include <mutex>

// Convenience type definition
using job = std::packaged_task<void()>;

// Some data associated to each thread.
struct thread_data
{
    int id; // Could use thread::id, but this is filled before the thread is started
    std::thread t; // The thread object
    std::queue<job> jobs; // The job queue
    std::condition_variable cv; // The condition variable to wait for threads
    std::mutex m; // Mutex used for avoiding data races
    bool stop = false; // When set, this flag tells the thread that it should exit
};

// The thread function executed by each thread
void thread_func(thread_data* pData)
{
    std::unique_lock<std::mutex> l(pData->m, std::defer_lock);
    while (true)
    {
        l.lock();

        // Wait until the queue won't be empty or stop is signaled
        pData->cv.wait(l, [pData] () {
            return (pData->stop || !pData->jobs.empty()); 
            });

        // Stop was signaled, let's exit the thread
        if (pData->stop) { return; }

        // Pop one task from the queue...
        job j = std::move(pData->jobs.front());
        pData->jobs.pop();

        l.unlock();

        // Execute the task!
        j();
    }
}

// Function that creates a simple task
job create_task(int id, int jobNumber)
{
    job j([id, jobNumber] ()
    {
        std::stringstream s;
        s << "Hello " << id << "." << jobNumber << std::endl;
        std::cout << s.str();
    });

    return j;
}

int main()
{
    const int numThreads = 4;
    const int numJobsPerThread = 10;
    std::vector<std::future<void>> futures;

    // Create all the threads (will be waiting for jobs)
    thread_data threads[numThreads];
    int tdi = 0;
    for (auto& td : threads)
    {
        td.id = tdi++;
        td.t = std::thread(thread_func, &td);
    }

    //=================================================
    // Start assigning jobs to each thread...

    for (auto& td : threads)
    {
        for (int i = 0; i < numJobsPerThread; i++)
        {
            job j = create_task(td.id, i);
            futures.push_back(j.get_future());

            std::unique_lock<std::mutex> l(td.m);
            td.jobs.push(std::move(j));
        }

        // Notify the thread that there is work do to...
        td.cv.notify_one();
    }

    // Wait for all the tasks to be completed...
    for (auto& f : futures) { f.wait(); }
    futures.clear();


    //=================================================
    // Here the main thread does something...

    std::cin.get();

    // ...done!
    //=================================================


    //=================================================
    // Posts some new tasks...

    for (auto& td : threads)
    {
        for (int i = 0; i < numJobsPerThread; i++)
        {
            job j = create_task(td.id, i);
            futures.push_back(j.get_future());

            std::unique_lock<std::mutex> l(td.m);
            td.jobs.push(std::move(j));
        }

        // Notify the thread that there is work do to...
        td.cv.notify_one();
    }

    // Wait for all the tasks to be completed...
    for (auto& f : futures) { f.wait(); }
    futures.clear();

    // Send stop signal to all threads and join them...
    for (auto& td : threads)
    {
        std::unique_lock<std::mutex> l(td.m);
        td.stop = true;
        td.cv.notify_one();
    }

    // Join all the threads
    for (auto& td : threads) { td.t.join(); }
}

Answer 2

你想要的概念是线程池.这个SO问题涉及现有的实现.

我们的想法是为多个线程实例创建一个容器.每个实例都与一个轮询任务队列的函数相关联,当一个任务可用时,将其拉出并运行它.一旦任务结束(如果它终止,但那是另一个问题),线程就会循环到任务队列.

因此,您需要一个同步队列,一个实现队列循环的线程类,一个任务对象的接口,以及一个驱动整个事物的类(池类).

或者,您可以为它必须执行的任务创建一个非常专业的线程类(例如,仅将内存区域作为参数).这需要线程的通知机制来指示它们是使用当前迭代完成的.

线程主函数将是该特定任务的循环,并且在一次迭代结束时,线程发出信号结束,并等待条件变量以启动下一个循环.实质上,您将在线程中内联任务代码,完全不需要队列.

 using namespace std;

 // semaphore class based on C++11 features
 class semaphore {
     private:
         mutex mMutex;
         condition_variable v;
         int mV;
     public:
         semaphore(int v): mV(v){}
         void signal(int count=1){
             unique_lock lock(mMutex);
             mV+=count;
             if (mV > 0) mCond.notify_all();
         }
         void wait(int count = 1){
             unique_lock lock(mMutex);
             mV-= count;
             while (mV < 0)
                 mCond.wait(lock);
         }
 };

template <typename Task>
class TaskThread {
     thread mThread;
     Task *mTask;
     semaphore *mSemStarting, *mSemFinished;
     volatile bool mRunning;
    public:
    TaskThread(Task *task, semaphore *start, semaphore *finish): 
         mTask(task), mRunning(true), 
         mSemStart(start), mSemFinished(finish),
        mThread(&TaskThread<Task>::psrun){}
    ~TaskThread(){ mThread.join(); }

    void run(){
        do {
             (*mTask)();
             mSemFinished->signal();
             mSemStart->wait();
        } while (mRunning);
    }

   void finish() { // end the thread after the current loop
         mRunning = false;
   }
private:
    static void psrun(TaskThread<Task> *self){ self->run();}
 };

 classcMyTask {
     public:
     MyTask(){}
    void operator()(){
        // some code here
     }
 };

int main(){
    MyTask task1;
    MyTask task2;
    semaphore start(2), finished(0);
    TaskThread<MyTask> t1(&task1, &start, &finished);
    TaskThread<MyTask> t2(&task2, &start, &finished);
    for (int i = 0; i < 10; i++){
         finished.wait(2);
         start.signal(2);
    }
    t1.finish();
    t2.finish();
}

上面提出的(粗略)实现依赖于Task必须提供的类型operator()(即类似仿函数).我说你可以先把任务代码直接合并到线程函数体中,但由于我不知道,所以我尽可能保持抽象.有一个条件变量用于线程的开始,一个用于它们的结束,两者都封装在信号量实例中.

看到提出使用的另一个答案 boost::barrier,我只能支持这个想法:如果可能的话,确保用该类替换我的信号量类,原因是依靠经过良好测试和维护的外部代码而不是自己实现的更好相同功能集的解决方案.

总而言之,这两种方法都是有效的,但前者放弃了一点点性能,有利于灵活性.如果要执行的任务花费足够长的时间,则管理和队列同步成本可以忽略不计.

更新:代码已修复并经过测试.用信号量替换了一个简单的条件变量.

Answer 3

使用屏障(只是条件变量和计数器上的便利包装)可以很容易地实现.它基本上阻塞,直到所有N个线程都到达"障碍".然后它再次"回收".Boost提供了一种实现方式.

void myfunc(void * p, boost::barrier& start_barrier, boost::barrier& end_barrier) {
  while (!stop_condition) // You'll need to tell them to stop somehow
  {
      start_barrier.wait ();
      do_something(p);
      end_barrier.wait ();
  }
}

int main(){
  void * myp[n_threads] {a_location, another_location,...};

  boost::barrier start_barrier (n_threads + 1); // child threads + main thread
  boost::barrier end_barrier (n_threads + 1); // child threads + main thread

  std::thread mythread[n_threads];

    for (unsigned int i=0; i < n_threads; i++) {
      mythread[i] = std::thread(myfunc, myp[i], start_barrier, end_barrier);
    }

  start_barrier.wait (); // first unblock the threads

  for (unsigned long int j=0; j < ULONG_MAX; j++) {
    end_barrier.wait (); // mix_data must not execute before the threads are done
    mix_data(myp); 
    start_barrier.wait (); // threads must not start new iteration before mix_data is done
  }
  return 0;
}