Race condition and crash in cancellations when using concurrent_channel to cancel a coro

[dragon-dreamer]

I have a use case when I need to gracefully shutdown multiple co_spawned execution threads which. I do not see any better way to do this (apart from stopping the thread pool) than using a concurrent_channel (MSVC2022, Win10, boost 1.87):

#include <boost/asio/io_context.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/asio/as_tuple.hpp>
#include <boost/asio/experimental/concurrent_channel.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/asio/thread_pool.hpp>

#include <iostream>

boost::asio::awaitable<int> expect_that_this_function_is_cancelled(
	boost::asio::steady_timer& timer_to_be_cancelled)
{
	co_await boost::asio::this_coro::throw_if_cancelled(false);

	int result = 0;
	for (int i = 0; i < 1000000; ++i)
	{
		result += std::pow(i, 0.1);
		result += std::pow(i, 0.1);
	}
	co_await boost::asio::post(co_await boost::asio::this_coro::executor,
		boost::asio::use_awaitable);

	co_return result;
}

boost::asio::awaitable<void> wait_for_cancellation(
	boost::asio::experimental::concurrent_channel<void()>& cancellation_channel)
{
	co_await cancellation_channel.async_send(boost::asio::as_tuple(boost::asio::use_awaitable));
	std::cerr << "Channel cancelled...\n";
}

int main()
{
	boost::asio::thread_pool ctx{ 32 };

	boost::asio::steady_timer timer_to_be_cancelled(ctx,
		std::chrono::steady_clock::time_point::max());
	boost::asio::experimental::concurrent_channel<void()> cancellation_channel(ctx);

	using namespace boost::asio::experimental::awaitable_operators;
	for (int i = 0; i != 100; ++i)
	{
		boost::asio::co_spawn(ctx,
			(expect_that_this_function_is_cancelled(timer_to_be_cancelled)
				|| wait_for_cancellation(cancellation_channel)),
			boost::asio::detached);

		if (i == 50)
		{
			cancellation_channel.close();
			cancellation_channel.cancel();
		}
	}

	ctx.join();
}

In some runs, this program crashes on the following line, because the handler_ pointer becomes invalid or null while being used:

asio/asio/include/asio/cancellation_signal.hpp

Lines 97 to 98 in 62481a2

	if (handler_)
	handler_->call(type);

Here is the call stack:

>	boost_asio_cancel_race.exe!boost::asio::cancellation_signal::emit(boost::asio::cancellation_type type) Line 99	C++
 	boost_asio_cancel_race.exe!boost::asio::cancellation_state::impl<boost::asio::cancellation_filter<1>,boost::asio::cancellation_filter<1>>::operator()(boost::asio::cancellation_type in) Line 222	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::cancellation_handler<boost::asio::cancellation_state::impl<boost::asio::cancellation_filter<1>,boost::asio::cancellation_filter<1>>>::call(boost::asio::cancellation_type type) Line 56	C++
 	boost_asio_cancel_race.exe!boost::asio::cancellation_signal::emit(boost::asio::cancellation_type type) Line 99	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::co_spawn_cancellation_handler<boost::asio::experimental::detail::parallel_group_op_handler<1,boost::asio::experimental::wait_for_one_success,boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>,boost::asio::any_io_executor,void>::operator()(boost::asio::cancellation_type type) Line 296	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::cancellation_handler<boost::asio::detail::co_spawn_cancellation_handler<boost::asio::experimental::detail::parallel_group_op_handler<1,boost::asio::experimental::wait_for_one_success,boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>,boost::asio::any_io_executor,void>>::call(boost::asio::cancellation_type type) Line 56	C++
 	boost_asio_cancel_race.exe!boost::asio::cancellation_signal::emit(boost::asio::cancellation_type type) Line 99	C++
 	boost_asio_cancel_race.exe!boost::asio::experimental::detail::parallel_group_launch<boost::asio::experimental::wait_for_one_success,boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>,0,1>(boost::asio::experimental::wait_for_one_success cancellation_condition, boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void> handler, std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>> & ops, std::integer_sequence<unsigned __int64,0,1> __formal) Line 388	C++
 	boost_asio_cancel_race.exe!boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait::operator()<boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>,boost::asio::experimental::wait_for_one_success>(boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void> && h, boost::asio::experimental::wait_for_one_success && c, std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>> && ops) Line 151	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::completion_handler_async_result<boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>,void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr)>::initiate<boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait,boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>,boost::asio::experimental::wait_for_one_success,std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>>(boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait && initiation, boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void> && token, boost::asio::experimental::wait_for_one_success && <args_0>, std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>> && <args_1>) Line 329	C++
 	boost_asio_cancel_race.exe!boost::asio::async_initiate<boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>,void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait,boost::asio::experimental::wait_for_one_success,std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>>(boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait && initiation, boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void> & token, boost::asio::experimental::wait_for_one_success && <args_0>, std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>> && <args_1>) Line 629	C++
 	boost_asio_cancel_race.exe!boost::asio::deferred_async_operation<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait,boost::asio::experimental::wait_for_one_success,std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>>::invoke_helper<boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>,0,1>(boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void> && token, std::integer_sequence<unsigned __int64,0,1> __formal) Line 319	C++
 	boost_asio_cancel_race.exe!boost::asio::deferred_async_operation<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait,boost::asio::experimental::wait_for_one_success,std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>>::operator()<boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void>>(boost::asio::detail::awaitable_async_op_handler<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::any_io_executor,void> && token) Line 355	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::awaitable_async_op<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::deferred_async_operation<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait,boost::asio::experimental::wait_for_one_success,std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>>,boost::asio::any_io_executor>::await_suspend::__l2::<lambda_1>::operator()(void * arg) Line 1040	C++
 	boost_asio_cancel_race.exe!`boost::asio::detail::awaitable_async_op<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::deferred_async_operation<void __cdecl(std::array<unsigned __int64,2>,std::exception_ptr,int,std::exception_ptr),boost::asio::experimental::parallel_group<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>::initiate_async_wait,boost::asio::experimental::wait_for_one_success,std::tuple<boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr,int),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<int,boost::asio::any_io_executor>>,boost::asio::deferred_async_operation<void __cdecl(std::exception_ptr),boost::asio::detail::initiate_co_spawn<boost::asio::any_io_executor>,boost::asio::detail::awaitable_as_function<void,boost::asio::any_io_executor>>>>,boost::asio::any_io_executor>::await_suspend'::`2'::<lambda_1>::<lambda_invoker_cdecl>(void * arg) Line 1042	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::awaitable_frame_base<boost::asio::any_io_executor>::resume() Line 501	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::awaitable_thread<boost::asio::any_io_executor>::pump() Line 769	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::awaitable_async_op_handler<void __cdecl(void),boost::asio::any_io_executor,void>::operator()() Line 803	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::binder0<boost::asio::detail::awaitable_async_op_handler<void __cdecl(void),boost::asio::any_io_executor,void>>::operator()() Line 56	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::executor_function::complete<boost::asio::detail::binder0<boost::asio::detail::awaitable_async_op_handler<void __cdecl(void),boost::asio::any_io_executor,void>>,std::allocator<void>>(boost::asio::detail::executor_function::impl_base * base, bool call) Line 113	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::executor_function::operator()() Line 61	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::executor_op<boost::asio::detail::executor_function,std::allocator<void>,boost::asio::detail::scheduler_operation>::do_complete(void * owner, boost::asio::detail::scheduler_operation * base, const boost::system::error_code & __formal, unsigned __int64 __formal) Line 70	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::scheduler_operation::complete(void * owner, const boost::system::error_code & ec, unsigned __int64 bytes_transferred) Line 40	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::scheduler::do_run_one(boost::asio::detail::conditionally_enabled_mutex::scoped_lock & lock, boost::asio::detail::scheduler_thread_info & this_thread, const boost::system::error_code & ec) Line 492	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::scheduler::run(boost::system::error_code & ec) Line 208	C++
 	boost_asio_cancel_race.exe!boost::asio::thread_pool::thread_function::operator()() Line 39	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::win_thread::func<boost::asio::thread_pool::thread_function>::run() Line 122	C++
 	boost_asio_cancel_race.exe!boost::asio::detail::win_thread_function(void * arg) Line 119	C++

I guess, this happens because cancellation slots are not thread safe, and the usage is not valid... Then the question is how do I cancel all all those co_spawned threads gracefully without stopping the thread_pool?