@@ -802,20 +802,7 @@ namespace hive { namespace chain {
802802 FC_THROW (" unknown purpose" );
803803 }
804804
805- std::shared_ptr< std::thread > queue_filler_thread;
806-
807- BOOST_SCOPE_EXIT ( &queue_filler_thread, &stop_requested, &block_queue_condition ) {
808- ilog (" Queue filler thread is joining." );
809- if ( queue_filler_thread )
810- {
811- stop_requested = true ;
812- block_queue_condition.notify_one ();
813- queue_filler_thread->join ();
814- }
815- ilog (" Queue filler thread was joined." );
816- } BOOST_SCOPE_EXIT_END
817-
818- queue_filler_thread = std::make_shared<std::thread>([&]() {
805+ std::thread queue_filler_thread ([&]() {
819806 fc::set_thread_name (" for_each_io" ); // tells the OS the thread's name
820807 fc::thread::current ().set_name (" for_each_io" ); // tells fc the thread's name for logging
821808 for (uint32_t block_number = starting_block_number; block_number <= ending_block_number; ++block_number)
@@ -826,36 +813,67 @@ namespace hive { namespace chain {
826813 while (block_queue.size () >= max_blocks_to_prefetch && !stop_requested)
827814 block_queue_condition.wait (lock);
828815 if (stop_requested)
816+ {
817+ ilog (" Leaving the queue thread" );
829818 return ;
819+ }
830820 block_queue.push (full_block);
821+ block_queue_condition.notify_one ();
831822 }
832- block_queue_condition.notify_one ();
833823 hive::chain::blockchain_worker_thread_pool::get_instance ().enqueue_work (full_block, worker_thread_processing);
834824 }
825+
826+ ilog (" Exiting the queue thread" );
835827 });
836828
837829 for (uint32_t block_number = starting_block_number; block_number <= ending_block_number; ++block_number)
838830 {
839831 std::shared_ptr<full_block_type> full_block;
840832 {
841833 std::unique_lock<std::mutex> lock (block_queue_mutex);
842- while (block_queue.empty ())
834+ while (block_queue.empty () && !stop_requested )
843835 block_queue_condition.wait (lock);
844- full_block = block_queue.front ();
845- block_queue.pop ();
836+
837+ if (!stop_requested)
838+ {
839+ full_block = block_queue.front ();
840+ block_queue.pop ();
841+ }
842+
843+ block_queue_condition.notify_one ();
846844 }
847- block_queue_condition.notify_one ();
848845
849- if (! processor (full_block))
846+ try
850847 {
848+ if (!stop_requested)
849+ stop_requested = !processor (full_block);
850+
851+ if (stop_requested)
851852 {
853+ ilog (" Attempting to break a block processing loop and request block queue stop." );
852854 std::unique_lock<std::mutex> lock (block_queue_mutex);
853- stop_requested = true ;
855+ block_queue_condition.notify_one ();
856+ ilog (" Block queue stop requested." );
857+ break ;
854858 }
855- block_queue_condition.notify_one ();
856- break ;
857859 }
860+ FC_CAPTURE_CALL_LOG_AND_RETHROW ([&]()
861+ {
862+ {
863+ std::unique_lock<std::mutex> lock (block_queue_mutex);
864+ stop_requested = true ;
865+ block_queue_condition.notify_one ();
866+ }
867+
868+ ilog (" Attempting to join queue_filler_thread..." );
869+ queue_filler_thread.join ();
870+ ilog (" queue_filler_thread joined." );
871+ }, ());
858872 }
873+
874+ ilog (" Attempting to join queue_filler_thread..." );
875+ queue_filler_thread.join ();
876+ ilog (" queue_filler_thread joined." );
859877 }
860878
861879 void block_log::truncate (uint32_t new_head_block_num)
0 commit comments