c++ - 提升 this_thread interruptation_point not captured

Question

在 Debug模式下运行时，抛出中断错误时崩溃:

如何中断线程池，以及如何正确捕获中断(以及在哪里？在分派(dispatch)的作业中，还是在 CancelAll 中？)如果没有中断点，它不会崩溃，但线程不会被中断。当我放置中断点时，它会在抛出中断错误时崩溃。

class ThreadAsioPool
{
public:
  ThreadAsioPool(const std::size_t nb_threads);
  virtual ~ThreadAsioPool() { CancelAll(); }

  void CancelAll();
  void AddJob(Wrapcontainer &wrap);

protected:
  static void job(Wrapcontainer wrap);
  void AddThread() {m_threadgroup.create_thread(boost::bind(&boost::asio::io_service::run, &io)); }
private:
  std::size_t nbThreads;
  boost::asio::io_service io;
  boost::thread_group m_threadgroup;
  boost::asio::io_service::work *work;
};

ThreadAsioPool::ThreadAsioPool(const std::size_t nb_threads): nbThreads(nb_threads)
{
  work = new boost::asio::io_service::work(io);//will keep io busy so it won't stop
  for (std::size_t i = 0; i < nbThreads; ++i)
    AddThread();
}
void ThreadAsioPool::AddJob(Wrapcontainer &wrap)
{
  //after a CancelAll, the m_threadgroup would be empty, so add more thread if needed
  if (m_threadgroup.size() < nbThreads )
    AddThread();

  io.post(boost::bind(&ThreadAsioPool::job,wrap));
  io.reset();
  io.poll();
}
void ThreadAsioPool::CancelAll()
{
  try{
    io.stop();//not sure that's necessary to stop the queue from processing
    m_threadgroup.interrupt_all(); //this to empty the queue
    m_threadgroup.join_all();//I assume after this line that the m_threadgroup is empty
  }
  catch(boost::thread_interrupted const& e){
    int i=0;//so I can put a breakpoint to see if it gets there, and it doesn't
    i=1;
  }
  catch(std::exception const& e){
    int i=0;//doesn't get here either
    i=2;
  }
  catch(...){
    int i=0;//doesn't get here either
    i=3;
  }

}
void ThreadAsioPool::job(Wrapcontainer wrap)
{

  try{
    boost::this_thread::interruption_point();
    ...some work..
    boost::this_thread::interruption_point(); //inserting few interruption point between heavy task
    ...some work..
    boost::this_thread::sleep(boost::posix_time::milliseconds(50)); //different type of breakpoint see if it makes a difference
    ...some work..
    boost::this_thread::interruption_point();
  }//try
  catch(boost::thread_interrupted& e){
    int i=0;//so I can put a breakpoint to see if it gets there, and it doesn't
    i=1;
  }
  catch(std::exception const& e){
    int i=0;//doesn't get here either
    i=2;
  }
  catch(...){
    int i=0;//doesn't get here either
    i=3;
  }


}

void main()
{
  ThreadAsioPool threadpool(3);

  //add 50 jobs in the queue
  for (int i=0;i<50;i++){
    Wrapcontainer itemdata;//just a class to contain data for the job
    ... fill up 'itemdata' with required data for the job

    threadpool.AddJob(itemdata);//feed a job
  }
  threadpool.CancelAll();
}

Answer 1

在 AddJob 中，您已经完成了所有工作。你没有排队任何东西并且执行是顺序的: Simple Demo

Job Job 1 done main.cpp
2 done main.cpp
Job 3 done main.cpp
Job 4 done main.cpp
Job 5 done main.cpp
Job 6 done main.cpp
Job 7 done main.cpp
Job 8 done main.cpp
Job 9 done main.cpp
Job 10 done main.cpp
Canceling

通过不执行reset() 和poll() 来修复它(您已经让工作人员执行run()，对吧？)。

示例(一分钟内):

Live On Coliru

This sample leaves "resetting" the thread pool/queue after cancelling as an exercise to the reader. Consider KISS though: you can always make the ThreadAsioPool cancel during destruction, and just create a new instance for any more work.

This would have exactly the desired behaviour with no complexity in coding.

#include <boost/thread.hpp>
#include <boost/asio.hpp>

struct Wrapcontainer {
    int id = id_gen();
  private:
    static int id_gen() {
        static int seed = 0;
        return ++seed;
    }
};

class ThreadAsioPool {
  public:
    ThreadAsioPool(const std::size_t nb_threads);
    virtual ~ThreadAsioPool() { CancelAll(); }

    void CancelAll();
    void AddJob(Wrapcontainer &wrap);

  protected:
    static void job(Wrapcontainer wrap);
    void AddThread() { m_threadgroup.create_thread(boost::bind(&boost::asio::io_service::run, &io)); }

  private:
    std::size_t nbThreads;
    boost::asio::io_service io;
    boost::thread_group m_threadgroup;
    boost::optional<boost::asio::io_service::work> work;
};

ThreadAsioPool::ThreadAsioPool(const std::size_t nb_threads) 
    : nbThreads(nb_threads), io(), 
      work(boost::asio::io_service::work(io))
{
    for (std::size_t i = 0; i < nbThreads; ++i)
        AddThread();
}

void ThreadAsioPool::AddJob(Wrapcontainer &wrap) {
    // after a CancelAll, the m_threadgroup would be empty, so add more thread if needed
    if (m_threadgroup.size() < nbThreads)
        AddThread();

    io.post(boost::bind(&ThreadAsioPool::job, wrap));
}

void ThreadAsioPool::CancelAll() {
    try {
        work.reset();
        io.stop(); // not sure that's necessary to stop the queue from processing
        m_threadgroup.interrupt_all(); // this to empty the queue
        m_threadgroup.join_all(); // I assume after this line that the m_threadgroup is empty
    } catch (boost::thread_interrupted const &e) {
        std::cout << "CAUGHT " + std::to_string(__LINE__) << "\n";
    } catch (std::exception const &e) {
        std::cout << "CAUGHT" + std::to_string(__LINE__) << "\n";
    } catch (...) {
        std::cout << "CAUGHT " + std::to_string(__LINE__) << "\n";
    }
}
void ThreadAsioPool::job(Wrapcontainer wrap) {
    try {
        boost::this_thread::interruption_point();
        boost::this_thread::interruption_point(); // inserting few interruption point between heavy task
        boost::this_thread::sleep(boost::posix_time::milliseconds(50)); // different type of breakpoint see if it makes a difference
        boost::this_thread::interruption_point();
        std::cout << "Job " << wrap.id << " done " + std::to_string(__LINE__) << "\n";
    } // try
    catch (boost::thread_interrupted &e) {
        std::cout << "CAUGHT " + std::to_string(__LINE__) << "\n";
    } catch (std::exception const &e) {
        std::cout << "CAUGHT " + std::to_string(__LINE__) << "\n";
    } catch (...) {
        std::cout << "CAUGHT " + std::to_string(__LINE__) << "\n";
    }
}

int main() {
    ThreadAsioPool threadpool(3);

    // add 50 jobs in the queue
    for (int i = 0; i < 50; i++) {
        Wrapcontainer itemdata; // just a class to contain data for the job
        //... fill up 'itemdata' with required data for the job
        threadpool.AddJob(itemdata);//feed a job
    }
    std::cout << "Canceling\n";

    threadpool.CancelAll();
}

打印

Canceling
CAUGHT 71
CAUGHT 71
CAUGHT 71

注意 在 CancelAllJobs 中捕获中断的异常有点滑稽(除非您想在从池线程中调用时防止异常泄漏)。