(C/C++线程):创建将在需要时监听作业并同时执行它们的工作线程

Question

假设我们有两个工人.每个工人的id为0和1.还假设我们有工作一直到达,每个工作也有一个标识符0或1指定哪个工作人员必须完成这项工作.

我想创建最初锁定的2个线程,然后当两个作业到达时,解锁它们,每个都完成它们的工作,然后再次锁定它们直到其他作业到达.

我有以下代码:

  #include <iostream>
  #include <thread>
  #include <mutex>

  using namespace std;

  struct job{

      thread jobThread;
      mutex jobMutex;

  };

  job jobs[2];


  void executeJob(int worker){

      while(true){

          jobs[worker].jobMutex.lock();

          //do some job

      }

   }

  void initialize(){

      int i;
      for(i=0;i<2;i++){
                jobs[i].jobThread = thread(executeJob, i);
      }

   }

  int main(void){

      //initialization
      initialize();

      int buffer[2];
      int bufferSize = 0;

      while(true){
          //jobs arrive here constantly, 
            //once the buffer becomes full, 
            //we unlock the threads(workers) and they start working
          bufferSize = 2;
          if(bufferSize == 2){
              for(int i = 0; i<2; i++){
                  jobs[i].jobMutex.unlock();
              }
          }
           break;
     }

  }

我std::thread几天前开始使用,我不确定为什么但Visual Studio给我一个错误说abort() has been called.我相信有些东西缺失,但由于我的无知,我无法弄清楚是什么.

我希望这段代码真的可以实现

1. 初始化两个线程然后锁定它们

2. 在main函数内解锁两个线程,两个线程将完成它们的工作(在这种情况下没有),然后它们将再次被锁定.

但它给了我一个错误.我究竟做错了什么？

先感谢您!

Answer 1

为此,您可以使用boost的threadpool类.它效率高,经过充分测试.开源库而不是你新写的并稳定它.

http://threadpool.sourceforge.net/

main()
{
    pool tp(2);   //number of worker threads-currently its 2.

    // Add some tasks to the pool.
    tp.schedule(&first_task);
    tp.schedule(&second_task);
}

void first_task()
{
    ...
}

void second_task()
{
    ...
}

注意:

对您的示例的建议:您不需要为每个线程都有单独的互斥对象.单个互斥锁对象锁本身将执行所有线程之间的同步.您正在executejob函数中锁定一个线程的互斥锁,而没有解锁另一个线程正在使用不同的互斥锁对象调用锁定导致死锁或未定义的行为.

此外,因为你在whileloop中调用mutex.lock()而没有解锁,所以同一个线程试图用相同的互斥对象锁定自身,从而导致未定义的行为.

如果您不需要并行执行线程,则可以在executejob函数内部使用一个全局互斥对象来锁定和解锁.

mutex m;

void executeJob(int worker)
{
    m.lock();

    //do some job

    m.unlock();
}

如果你想执行作业并行使用boost threadpool,正如我之前建议的那样.

Answer 2

通常,您可以编写类似于以下的算法.它适用于pthreads.我相信它也适用于c ++线程.

1. 创建线程并使它们等待条件变量,例如work_exists.
2. 当工作到达时,您通知正在等待该条件变量的所有线程.然后在主线程中开始等待另一个条件变量work_done
3. 收到work_exists通知后,工作线程唤醒,从中获取分配的工作jobs[worker],执行它们,然后发送work_done变量通知,然后返回等待work_exists条件变量
4. 当主线程收到work_done通知时,它会检查是否所有线程都已完成.如果没有,它会一直等到最后完成线程的通知到达.