使用同步机制（C++11）延长线程的生命周期

Question

我有一个程序，其中有一个函数需要接收指针作为参数，还有一个主函数。主函数创建n个线程，每个线程在不同的内存区域上运行函数，具体取决于传递的arg。然后等待所有线程完成，主函数对这些区域进行一些数据混合，并创建n个新线程，这些线程执行与旧线程相同的操作。

为了改进程序，我想让线程保持活动状态，消除创建线程所需的长时间。当主函数工作时，线程应处于休眠状态，并在需要再次启动时得到通知。同样地，当线程正在工作时，主函数应等待它们完成，就像使用join一样。

我无法得到一个强大的实现，总是陷入死锁中。

这是一个简单的基准代码，希望能提供修改的提示。

#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;
}

- DarioP

---

不，这是我现在拥有的代码示例，它不会死锁。 我找不到更好的实现方式（不删除线程并加入它们），也不会死锁。 - DarioP

你可能在寻找线程池吗？ - us2012

@DarioP - 是的，你需要，诚实地说 :) - Martin James

@DarioP 换句话说 - 在SO上搜索“线程管理器”，看看有多少尝试失败了。很多，你的也可能会加入其中（抱歉）。像其他人建议的那样使用线程池。努力避免join()，也就是所谓的“死锁生成器”。 - Martin James

@slebetman，不是的。他所尝试做的与线程池没有任何关系。 - user1773602

Answer 1

这里有一种可能的方法，只使用 C++11 标准库中的类。基本上，你创建的每个线程都有一个相关联的命令队列（封装在 std::packaged_task<> 对象中），它会不断地检查。如果队列为空，线程将在条件变量 (std::condition_variable) 上等待。

通过使用 std::mutex 和 std::unique_lock<> RAII 包装器避免了数据竞争，主线程可以通过存储与每个提交的 std::packaged_tast<> 相关联的 std::future<> 对象并调用 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(); }
}

- Andy Prowl

---

是的！非常感谢。虽然这个解决方案并不像屏障那么简单，但理解、实现和控制起来并不太困难。我在发布时就有类似的想法，现在我需要弄清楚这是否真正是我想要的。 - DarioP

Answer 2

你所需要的概念是线程池。这个SO question讨论了现有的实现方式。

其思路是创建一个包含多个线程实例的容器。每个实例都与一个函数相关联，该函数轮询任务队列，并在任务可用时拉取并运行它。一旦任务完成（如果终止，则另一个问题），线程就简单地循环到任务队列。

因此，您需要一个同步队列、一个实现队列循环的线程类、一个任务对象的接口，以及可能需要一个驱动整个过程的类（池类）。

或者，您可以为要执行的任务制作一个非常专业化的线程类（例如，只有内存区域作为参数）。这需要一种通知机制，以便线程指示它们已完成当前迭代。

线程主函数将是特定任务的循环，一次迭代结束后，线程会发出结束信号，并等待条件变量启动下一个循环。本质上，您将在线程中嵌入任务代码，完全消除了队列的需求。

 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，我只能支持这个想法：如果可能，请确保将我的信号量类替换为该类，原因是依赖于经过良好测试和维护的外部代码比自己实现相同特性更好。

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

更新：代码已修复并测试完成。用信号量替换了简单的条件变量。

- didierc

---

这就是我在评论中建议的，但显然 OP 认为它们不符合他的需求。你能否解决他对线程池可能过度的担忧？ - us2012

我想我差不多做到了。我没有提供一个示例实现，因为我正在使用我的手机，而且我对c++ std::threads和condition vars的知识不够充分，无法自信地编写它。欢迎编辑。 - didierc

这是寻找池的第二个建议。我会考虑这个建议，但仍然必须有基于互斥和条件变量的解决方案。我也想从这个角度来解决问题。 - DarioP

@DarioP 线程池基于互斥锁和条件变量，只是通过一些漂亮干净的接口隐藏了底层细节。 - Stephan Dollberg

好的 - 请采取池的方法。过度管理线程几乎总是失败的，通常情况很糟糕。 - Martin James

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

- user1773602

---

这是为科学软件而设计的，原则上应该能够在许多不同的机器上编译和运行，并且很容易适应集群运行。出于可移植性考虑，我想使用标准库或至少POSIX，以避免使用boost。然而，这个解决方案真的非常好和简单！我会认真考虑这个建议，谢谢！ - DarioP

@DarioP，barrier实现起来非常简单，只需要几十行代码。你可以从这里复制粘贴http://www.boost.org/doc/libs/1_53_0/boost/thread/barrier.hpp。只需将`boost::mutex`和`conditional_variable`替换为posix等效即可。 - user1773602

是的，实现栅栏类应该很容易。我也稍微改进了一下你的代码。你觉得可以吗？ - DarioP

@DarioP，不行。按照你的更正，代码会导致死锁。 - user1773602

事实上，这正是我在几个周期内获得的结果。现在我必须弄清楚为什么。 - DarioP

Answer 4

以下是一个简单的编译和工作代码，执行一些随机的事情。它实现了aleguna的屏障概念。每个线程的任务长度不同，因此有一个强大的同步机制是非常必要的。我将尝试在相同的任务上进行池化，并对结果进行基准测试，然后可能会使用Andy Prowl指出的futures。

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <complex>
#include <random>

const unsigned int n_threads=4; //varying this will not (almost) change the total amount of work
const unsigned int task_length=30000/n_threads;
const float task_length_variation=task_length/n_threads;
unsigned int rep=1000; //repetitions of tasks

class t_chronometer{
 private: 
  std::chrono::steady_clock::time_point _t;

 public:
  t_chronometer(): _t(std::chrono::steady_clock::now()) {;}
  void reset() {_t = std::chrono::steady_clock::now();}
  double get_now() {return std::chrono::duration_cast<std::chrono::duration<double>>(std::chrono::steady_clock::now() - _t).count();}
  double get_now_ms() {return 
      std::chrono::duration_cast<std::chrono::duration<double,std::milli>>(std::chrono::steady_clock::now() - _t).count();}
};

class t_barrier {
 private:
   std::mutex m_mutex;
   std::condition_variable m_cond;
   unsigned int m_threshold;
   unsigned int m_count;
   unsigned int m_generation;

 public:
   t_barrier(unsigned int count):
    m_threshold(count),
    m_count(count),
    m_generation(0) {
   }

   bool wait() {
      std::unique_lock<std::mutex> lock(m_mutex);
      unsigned int gen = m_generation;

      if (--m_count == 0)
      {
          m_generation++;
          m_count = m_threshold;
          m_cond.notify_all();
          return true;
      }

      while (gen == m_generation)
          m_cond.wait(lock);
      return false;
   }
};


using namespace std;

void do_something(complex<double> * c, unsigned int max) {
  complex<double> a(1.,0.);
  complex<double> b(1.,0.);
  for (unsigned int i = 0; i<max; i++) {
    a *= polar(1.,2.*M_PI*i/max);
    b *= polar(1.,4.*M_PI*i/max);
    *(c)+=a+b;
  }
}

bool done=false;
void task(complex<double> * c, unsigned int max, t_barrier* start_barrier, t_barrier* end_barrier) {
  while (!done) {
    start_barrier->wait ();
    do_something(c,max);
    end_barrier->wait ();
  }
  cout << "task finished" << endl;
}

int main() {
  t_chronometer t;

  std::default_random_engine gen;
  std::normal_distribution<double> dis(.0,1000.0);

  complex<double> cpx[n_threads];
  for (unsigned int i=0; i < n_threads; i++) {
    cpx[i] = complex<double>(dis(gen), dis(gen));
  }

  t_barrier start_barrier (n_threads + 1); // child threads + main thread
  t_barrier end_barrier (n_threads + 1); // child threads + main thread

  std::thread mythread[n_threads];
  unsigned long int sum=0;
  for (unsigned int i=0; i < n_threads; i++) {
    unsigned int max = task_length +  i * task_length_variation;
    cout << i+1 << "th task length: " << max << endl;
    mythread[i] = std::thread(task, &cpx[i], max, &start_barrier, &end_barrier);
    sum+=max;
  }
  cout << "total task length " << sum << endl;

  complex<double> c(0,0);
  for (unsigned long int j=1; j < rep+1; j++) {
    start_barrier.wait (); //give to the threads the missing call to start
    if (j==rep) done=true;
    end_barrier.wait (); //wait for the call from each tread
    if (j%100==0) cout << "cycle: " << j << endl;
    for (unsigned int i=0; i<n_threads; i++) {
      c+=cpx[i];
    }
  }
  for (unsigned int i=0; i < n_threads; i++) {
    mythread[i].join();
  }
  cout << "result: " << c << " it took: " << t.get_now() << " s." << endl;
  return 0;
}

- DarioP

---

我想说的是，屏障解决方案并没有提供很大的性能提升。对于当前的实现，加入线程和创建新线程所用的时间几乎与使用条件变量通知它们的时间相同。 - DarioP