io: add option to reap all cqes when notified about io completions

emper/Emper.hpp

@@ -161,6 +161,14 @@ const bool IO_NOTIFICATION =
 #endif
 		;
 
+const bool IO_STEAL_ALL_ON_NOTIFICATION =
+#ifdef EMPER_IO_STEAL_ALL_ON_NOTIFICATION
+		true
+#else
+		false
+#endif
+		;
+
 const bool IO_PUNT_WAITFD_READ =
 #ifdef EMPER_IO_SKIP_WAITFD_READ_PUNT
 		false

emper/strategies/AbstractWorkStealingScheduler.cpp

@@ -152,7 +152,21 @@ auto AbstractWorkStealingScheduler::nextFiberViaAnywhereQueue() -> std::optional
 auto AbstractWorkStealingScheduler::tryStealIoCompletionFrom(workerid_t victim)
 		-> std::optional<NextFiberResult> {
 	auto* victimIo = runtime.ioContexts[victim];
-	Fiber* fiber = victimIo->reapSingleCompletion<CallerEnvironment::EMPER>();
+	Fiber* fiber;
+	if constexpr (emper::IO_STEAL_ALL_ON_NOTIFICATION) {
+		IoContext::ContinuationBuffer completions;
+		unsigned ncompletions =
+				IoContext::getWorkerIo()->reapCompletions<CallerEnvironment::EMPER>(completions);
+		if (ncompletions > 0) {
+			// Keep the first and schedule the rest
+			fiber = completions[0];
+			schedule(&completions[1], ncompletions - 1);
+			emper::statsAdd(awss::stats.nextIoFiberStolen, ncompletions - 1);
+		}
+	} else {
+		fiber = victimIo->reapSingleCompletion<CallerEnvironment::EMPER>();
+	}
+
 	if (fiber) {
 		emper::statsIncr(awss::stats.nextIoFiberStolen);
 		return NextFiberResult{fiber, emper::FiberSource::ioStolen};

meson.build

@@ -193,6 +193,8 @@ io_bool_options = [
 	 'dependencies': {'io_stealing': true, 'io_lockless_cq': true}},
 	{'option': 'notification',
 	 'dependencies': {'worker_sleep_strategy': 'waitfd'}},
+	{'option': 'steal_all_on_notification',
+	 'dependencies': {'io_notification': true}},
 	{'option': 'synchronous'},
 	{'option': 'skip_waitfd_read_punt',
 	 'dependencies': {'worker_sleep_strategy': 'waitfd'}},

meson_options.txt

@@ -333,6 +333,12 @@ option(
   description: 'Let the kernel notify suspended EMPER workers about new IO completions',
   value: false,
 )
+option(
+  'io_steal_all_on_notification',
+  type: 'boolean',
+  description: 'Reap all cqes when notified about available completions',
+  value: false,
+)
 option(
   'continuation_stealing_mode',
   type: 'combo',