ircd::fs::aio: Synchronize interruption and shutdown.

ircd/aio.cc

@@ -29,10 +29,44 @@ ircd::fs::aio::aio()
 ircd::fs::aio::~aio()
 noexcept
 {
-	resfd.cancel();
+	interrupt();
+	wait_interrupt();
+
+	boost::system::error_code ec;
+	resfd.close(ec);
+
 	syscall<SYS_io_destroy>(idp);
 }
 
+bool
+ircd::fs::aio::interrupt()
+{
+	if(!resfd.is_open())
+		return false;
+
+	resfd.cancel();
+	return true;
+}
+
+bool
+ircd::fs::aio::wait_interrupt()
+{
+	if(!resfd.is_open())
+		return false;
+
+	log::debug
+	{
+		"Waiting for AIO context %p", this
+	};
+
+	dock.wait([this]
+	{
+		return semval == uint64_t(-1);
+	});
+
+	return true;
+}
+
 void
 ircd::fs::aio::set_handle()
 {
@@ -46,7 +80,7 @@ ircd::fs::aio::set_handle()
 void
 ircd::fs::aio::handle(const boost::system::error_code &ec,
                       const size_t bytes)
-noexcept
+noexcept try
 {
 	assert((bytes == 8 && !ec && semval >= 1) || (bytes == 0 && ec));
 	assert(!ec || ec.category() == asio::error::get_system_category());
@@ -58,7 +92,7 @@ noexcept
 			break;
 
 		case boost::system::errc::operation_canceled:
-			return;
+			throw ctx::interrupted();
 
 		default:
 			throw boost::system::system_error(ec);
@@ -66,6 +100,16 @@ noexcept
 
 	set_handle();
 }
+catch(const ctx::interrupted &)
+{
+	log::debug
+	{
+		"AIO context %p interrupted", this
+	};
+
+	semval = -1;
+	dock.notify_all();
+}
 
 void
 ircd::fs::aio::handle_events()

ircd/aio.h

@@ -23,6 +23,9 @@ struct ircd::fs::aio
 	/// Maximum number of events we can submit to kernel
 	static constexpr const size_t &MAX_EVENTS {64};
 
+	/// Internal semaphore for synchronization of this object
+	ctx::dock dock;
+
 	/// The semaphore value for the eventfd which we keep here.
 	uint64_t semval {0};
 
@@ -43,6 +46,9 @@ struct ircd::fs::aio
 	void handle(const boost::system::error_code &, const size_t) noexcept;
 	void set_handle();
 
+	bool wait_interrupt();
+	bool interrupt();
+
 	aio();
 	~aio() noexcept;
 };