Boost Thread+C++20编译崩溃求助：根因排查咨询

背景与疑问 #

• 是否支持使用C++20编译选项构建Boost Thread与atomic库？在Boost官方文档中未找到相关说明。
• 应用程序在GCC 7.1、C17、Boost 1.75环境下运行正常，但升级到GCC 11.1、C20后出现Boost Thread崩溃问题。
• Sanitizer未报告任何异常，程序核心使用了boost::condition_variable。

复现代码 #

#include "boost/thread/thread.hpp"
#include "boost/shared_ptr.hpp"
#include "boost/date_time/posix_time/posix_time.hpp"
#include "boost/utility.hpp"
#include "boost/thread/condition_variable.hpp"
#include <boost/thread/thread.hpp>
#include <thread>

#include <algorithm>
#include <cassert>
#include <atomic>


#include <vector>

class Dispatcher;

class Task
{
public:
   virtual void run() = 0;

   virtual ~Task() {};
};


class TaskPool : boost::noncopyable
{
public:
   typedef boost::shared_ptr<Task>            task_ptr_type;
   typedef boost::shared_ptr<Dispatcher>  dispatcher_ptr_type;
   typedef std::vector<dispatcher_ptr_type>         thread_pool_type;
   typedef boost::posix_time::time_duration         time_duration_type;
   typedef std::size_t                              size_type;

   TaskPool(const size_type Size);

   ~TaskPool();

   size_type maxSize() const { return max_size_; }

   size_type watermark() const { return watermark_; }

   void setInactivTm(const time_duration_type& Inactivity_Time_Out);

   const time_duration_type& getInactivTm() const { return inactivity_time_out_; }

   void execute(const task_ptr_type& Task);

private:
   typedef boost::mutex            mutex_type;
   typedef mutex_type::scoped_lock lock_type;

   mutable mutex_type mutex_;

   size_type          max_size_;
   thread_pool_type * thread_pool_;

   time_duration_type inactivity_time_out_;

   size_type          watermark_;
   size_type          invocations_;
   size_type          executions_;
   size_type          loops_;
};


class Dispatcher : boost::noncopyable
{
public:
   typedef TaskPool::task_ptr_type      task_ptr_type;
   typedef TaskPool::time_duration_type time_duration_type;
   typedef TaskPool::size_type          size_type;

   Dispatcher();

   ~Dispatcher();

   void setInactivTm(const time_duration_type& Inactivity_Time_Out);

   bool waitReady(const time_duration_type& Time_Out);

   void execute(const task_ptr_type& Task);

   void terminate();

   static time_duration_type defaultActivTm()
   {
      return boost::posix_time::milliseconds( 1 );
   }

   static time_duration_type minActivTm()
   {
      return boost::posix_time::milliseconds( 1 );
   }

private:
   typedef boost::mutex              mutex_type;
   typedef boost::condition_variable condition_variable_type;
   typedef mutex_type::scoped_lock   lock_type;

   friend class Runner;

   bool Queued_() const volatile;
   bool NotQueued_() const volatile;

   void execute_();

   boost::thread thread_;
   task_ptr_type task_;

   mutable mutex_type      mutex_;
   condition_variable_type task_busy_cond_;
   condition_variable_type task_available_cond_;

   volatile bool is_terminated_;

   time_duration_type inactivity_time_out_;

   size_type invocations_;
   size_type executions_;
   size_type thread_created_;
   size_type thread_terminated_;
};


class Runner
{
public:
   explicit Runner(Dispatcher * Disp)
    : disp_( Disp )
   { }

   void operator()()
   {
      disp_->execute_();
   }

private:
   Dispatcher * const disp_;
};


Dispatcher::Dispatcher()
 : is_terminated_( false ),
   inactivity_time_out_( defaultActivTm() ),
   invocations_( 0 ),
   executions_( 0 ),
   thread_created_( 0 ),
   thread_terminated_( 0 )
{ }


Dispatcher::~Dispatcher()
{
   terminate();
}


void Dispatcher::setInactivTm(const time_duration_type& Inactivity_Time_Out)
{
   lock_type lock( mutex_ );

   inactivity_time_out_ = Inactivity_Time_Out;
   assert( inactivity_time_out_ >= minActivTm() );
}


bool Dispatcher::waitReady(const time_duration_type& Time_Out)
{
   lock_type lock( mutex_ );

   if ( !is_terminated_ &&
        (thread_.get_id() == boost::thread::id()) )
   {
      return true;
   }
   while ( Queued_() )
   {
      if ( !task_busy_cond_.timed_wait(lock,
                                       Time_Out) )
      {
         return false;
      }
   }
   return !is_terminated_;
}


void Dispatcher::execute(const task_ptr_type& Task)
{
   lock_type lock( mutex_ );

   if ( thread_.get_id() == boost::thread::id() )
   {
      thread_created_ += 1;
      thread_ = boost::thread( Runner(this) );
   }
   while ( Queued_() )
   {
      task_busy_cond_.wait(lock);
   }
   if ( !is_terminated_ )
   {
      task_ = Task;
      task_available_cond_.notify_one();
   }
   invocations_ += 1;
}


void Dispatcher::terminate()
{
   is_terminated_ = true;

   thread_.interrupt();
   thread_.join();
}


bool Dispatcher::Queued_() const volatile
{
   return const_cast<const task_ptr_type&>(task_) &&
          !is_terminated_;
}


bool Dispatcher::NotQueued_() const volatile
{
   return !const_cast<const task_ptr_type&>(task_) &&
          !is_terminated_;
}


void Dispatcher::execute_()
{
   {
      lock_type lock( mutex_ );
      is_terminated_ = false;
   }

   while ( 1 )
   {
      task_ptr_type tmp_task;

      {
         lock_type lock( mutex_ );

         while ( NotQueued_() )
         {
            if ( !task_available_cond_.timed_wait(lock,
                                                  inactivity_time_out_) )
            {
               thread_terminated_ += 1;
               thread_.detach();
               return;
            }
         }
         if ( is_terminated_ )
         {
            thread_terminated_ += 1;
            return;
         }
         tmp_task.swap( task_ );
         task_busy_cond_.notify_one();
      }
      executions_ += 1;

      try
      {
         tmp_task->run();
      }
      catch (const boost::thread_interrupted&)
      {
         thread_terminated_ += 1;
         thread_.detach();
         return;
      }
      catch (...)
      {
      }
   }
}


TaskPool::TaskPool(const size_type Size)
 : max_size_( Size ),
   thread_pool_( 0 ),
   inactivity_time_out_( Dispatcher::defaultActivTm() ),
   watermark_( 0 ),
   invocations_( 0 ),
   executions_( 0 ),
   loops_( 0 )
{
   assert( max_size_ > 0 );
   thread_pool_ = new thread_pool_type( max_size_ );
}


TaskPool::~TaskPool()
{
   delete thread_pool_;
}


void TaskPool::setInactivTm(const time_duration_type& Inactivity_Time_Out)
{
   lock_type lock( mutex_ );

   inactivity_time_out_ = Inactivity_Time_Out;
   assert( inactivity_time_out_ >= Dispatcher::minActivTm() );

   for (thread_pool_type::iterator iter = thread_pool_->begin();
        thread_pool_->end() != iter;
        ++iter)
   {
      dispatcher_ptr_type& p( *iter );

      if ( p )
      {
         p->setInactivTm( inactivity_time_out_ );
      }
   }
}


void TaskPool::execute(const task_ptr_type& Task)
{
   lock_type lock( mutex_ );

   invocations_ += 1;

   const time_duration_type min_iteration_timeout( boost::posix_time::microsec( 100 ) );
   const time_duration_type max_iteration_timeout( boost::posix_time::microsec( 100000 ) );

   time_duration_type timeout( 1 == max_size_ ? time_duration_type( boost::posix_time::pos_infin )
                                              : time_duration_type( boost::posix_time::microsec(0) ) );

   while ( 1 )
   {
      for (thread_pool_type::iterator iter = thread_pool_->begin();
           thread_pool_->end() != iter;
           ++iter)
      {
         dispatcher_ptr_type& p( *iter );

         loops_ += 1;

         if ( !p )
         {
            p.reset( new Dispatcher );
            p->setInactivTm( inactivity_time_out_ );

            watermark_ = iter - thread_pool_->begin();
         }
         if ( p->waitReady( timeout ) )
         {
            p->execute( Task );
            executions_ += 1;
            return;
         }
      }
      if ( timeout != boost::posix_time::pos_infin )
      {
         timeout *= 2;

         timeout = std::max(timeout,
                            min_iteration_timeout);
         timeout = std::min(timeout,
                            max_iteration_timeout);
      }
   }
}


static TaskPool threadPool = 10;

class Wrapper : public Task
{
public:
   Wrapper()
   {
      listener = new Listener;
   }

   virtual void run()
   {
      boost::this_thread::sleep( boost::posix_time::seconds(10) );
      listener->run();
   }

   struct Listener
   {
      std::string s;
      void run()
      {
         s = "Hello";
      }
   };

   Listener* listener;
};

struct Executer
{
   std::vector<std::thread> threads;

   void dispatch()
   {
      for (auto i=0; i<2; ++i)
      {
         threads.push_back(std::move(std::thread([&]()
         {
            int index = 0;
            while (true)
            {
               {
                  boost::shared_ptr<Wrapper> task( new Wrapper );
                  threadPool.execute( task );
               }

               if (index % 1000 == 0) boost::this_thread::sleep( boost::posix_time::seconds(5) );
            }
         })));
      }
   }

   ~Executer()
   {
      for (auto i=0; i<2; ++i) threads[i].join();
   }
};

int main()
{
  std::thread t1([](){Executer a; a.dispatch();});
  std::thread t2([](){Executer a; a.dispatch();});
  t1.join();
  t2.join();
}

排查发现 #

崩溃时观察到boost::shared_ptr的use_count_=1、weak_count_=1，无法理解为何weak_count不为0，恳请协助排查根因。

内容的提问来源于stack exchange，提问作者Peter