10

When I run my code :

nb workers = 12
I'm i : 0
HELLO I'm func1
BYE I'm func2
terminate called after throwing an instance of 'std::system_error'
  what():  Invalid argument
Aborted (core dumped)

terminate called after throwing an instance of 'std::system_error'l

what(): Invalid argument

#ifndef CPP_PLAZZA_EXAMPLE_H
#define CPP_PLAZZA_EXAMPLE_H

#include <thread>
#include <vector>
#include <list>
#include <memory>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <iterator>
#include <tuple>

class ThreadPool
{
 public:
  ThreadPool(size_t numThreads);
  virtual ~ThreadPool();
  void executeJob(std::function<void()> job, std::function<void()> notificationJob);
  void wait_for_done();
 private:
  void loop();
  std::pair<std::function<void()>, std::function<void()> > getNextJob();
  std::vector<std::thread> m_workers;
  std::list<std::pair<std::function<void()>, std::function<void()> > > m_jobs;
  std::mutex m_lockJobsList;
  std::condition_variable m_notifyJob;
  std::atomic<bool> m_bTerminate;
  class Terminated: public std::runtime_error
  {
   public:
    Terminated(const std::string& what): std::runtime_error(what) {}
  };

};

#endif //CPP_PLAZZA_EXAMPLE_H

and here it is my .cpp

#include <iostream>
#include "example.h"


ThreadPool::ThreadPool(size_t numThreads):
 m_workers(numThreads), m_bTerminate(false) {
  m_workers.reserve(numThreads);
  for (size_t i = 0; i < numThreads; i++) {
    this->m_workers.emplace_back(&ThreadPool::loop, this);
  }
  /*for (std::vector<std::thread>::iterator it = this->m_workers.begin(); it != this->m_workers.end(); it++)
    assert(std::next(it, 1) ==);*/
}

ThreadPool::~ThreadPool() {
    {
        std::unique_lock<std::mutex> lockList(m_lockJobsList);
        m_bTerminate = true;
        m_notifyJob.notify_all();
    }

/*  for(std::vector<std::thread>::iterator it = m_workers.begin(); it != m_workers.end(); it++) {
    it->join();
  }*/
  std::this_thread::sleep_for(std::chrono::seconds(5));
}

void ThreadPool::executeJob(std::function<void()> job, std::function<void()> notificationJob) {
    std::unique_lock<std::mutex> lockList(m_lockJobsList);
  m_jobs.emplace_back(std::pair<std::function<void()>, std::function<void()> >(std::move(job), std::move(notificationJob)));
  std::cout << m_jobs.size() << std::endl;
    m_notifyJob.notify_one();
}

std::pair<std::function<void()>, std::function<void()> > ThreadPool::getNextJob() {
    std::unique_lock<std::mutex> lockList(m_lockJobsList);

    while(!m_bTerminate)
    {
        if(!m_jobs.empty())
        {
            std::pair<std::function<void()>, std::function<void()>> job = std::ref(m_jobs.front());
            m_jobs.pop_front();
            return job;
        }

        m_notifyJob.wait(lockList);
    }

    throw Terminated("Thread terminated");
}

void        func1() {
  std::cout << "HELLO I'm func1" << std::endl;

}

void ThreadPool::loop()
{
    try
    {
        for(;;)
        {
      std::pair<std::function<void()>, std::function<void()> > job = getNextJob();
      job.first();
      job.second();
        }
    }
    catch(Terminated& e)
    {
    }
}



void        func2() {
  std::cout << "BYE I'm func2" << std::endl;

}

void        ThreadPool::wait_for_done()
{
  std::cout << "nb workers = " << this->m_workers.size() << std::endl;
  int i = 0;
  for(std::vector<std::thread>::iterator it = m_workers.begin(); it != m_workers.end(); ++it) {
    std::cout << "je suis i :  " << i << std::endl;
    i++;

    (*it).join();
  }
}

int     main()
{
  ThreadPool        pool(6);

  pool.executeJob(func1, func2);
  pool.wait_for_done();
}

I think that my error is I join several time on one thread but how to fix it ?

Compilation line : 

g++ -Wall -Werror -W -Wextra example.cpp -pthread -std=c++11

I tried joinable before join like this (in wait for done) :

for(std::vector<std::thread>::iterator it = m_workers.begin(); it != m_workers.end(); ++it) {
    if ((*it).joinable())
        (*it).join();
  }

And I had an infinite loop

Itoun
  • 3,713
  • 5
  • 27
  • 47

2 Answers2

9

Your m_lockJobsList mutex (and m_notifyJob condvar) gets destroyed before m_workers threads that try to lock it when wake after condvar notification at ThreadPool destructor.

user7860670
  • 35,849
  • 4
  • 58
  • 84
  • I had a highly intermittent core-dump and head-scratching due to `detach` threads under `TEST` googletest case. It turned out exactly to be as you described here, thank you. I solved it by using `std::vector v0;` and `std::for_each(v0.begin(),v0.end(),[](std::thread& t){t.join();});` at the bottom before `TEST` exits. This ensures my local mutex to last longer when the threads needs it. – daparic Jul 30 '19 at 05:41
  • This helped me out as well. Was scratching my head pretty hard until I found this question. Thank you! – Crankycyclops Jan 16 '20 at 01:52
-1

Just in case somebody still needs this This might happen because one your threads was neither joined, nor detatched. Every thread that you use in the main thread must have one the following commands:

  1. thread_name.join()
  2. thread_.detatch()

to know what exactly thsese commands are doing, I recomend simply googling it. This is the way I solved the problem, hope it helped.

user17697202
  • 1