message: Handle control events in runner
diff: @@ -492,8 +492,10 @@ where
  }
  /// Processes a single event on the event queue.
  + ///
  + /// Returns `false` if processing should stop.
  #[inline]
  - pub async fn crank(&mut self, rng: &mut NodeRng) {
  + pub async fn crank(&mut self, rng: &mut NodeRng) -> bool {
  // Create another span for tracing the processing of one event.
  let crank_span = debug_span!("crank", ev = self.event_count);
  let _inner_enter = crank_span.enter();
  @@ -561,7 +563,22 @@ where
  // Dispatch the event, then execute the resulting effect.
  let start = self.clock.start();
  - let effects = self.reactor.dispatch_event(effect_builder, rng, event);
  +
  + let (effects, keep_going) = if let Some(ctrl_ann) = event.as_control() {
  + // We've received a control event, which will _not_ be handled by the reactor.
  + match ctrl_ann {
  + ControlAnnouncement::FatalError { message } => {
  + error!(%message, "fatal error via control announcement");
  + (Default::default(), false)
  + }
  + }
  + } else {
  + (
  + self.reactor.dispatch_event(effect_builder, rng, event),
  + true,
  + )
  + };
  +
  let end = self.clock.end();
  // Warn if processing took a long time, record to histogram.
  @@ -587,6 +604,8 @@ where
  .await;
  self.event_count += 1;
  +
  + keep_going
  }
  /// Gets both the allocated and total memory from sys-info + jemalloc
  @@ -672,12 +691,11 @@ where
  /// Processes a single event if there is one, returns `None` otherwise.
  #[inline]
  #[cfg(test)]
  - pub async fn try_crank(&mut self, rng: &mut NodeRng) -> Option<()> {
  + pub async fn try_crank(&mut self, rng: &mut NodeRng) -> Option<bool> {
  if self.scheduler.item_count() == 0 {
  None
  } else {
  - self.crank(rng).await;
  - Some(())
  + Some(self.crank(rng).await)
  }
  }
  @@ -689,13 +707,15 @@ where
  match TERMINATION_REQUESTED.load(Ordering::SeqCst) as i32 {
  0 => {
  if let Some(reactor_exit) = self.reactor.maybe_exit() {
  - return reactor_exit;
  + break reactor_exit;
  + }
  + if !self.crank(rng).await {
  + break ReactorExit::ProcessShouldExit(ExitCode::Abort);
  }
  - self.crank(rng).await;
  }
  - SIGINT => return ReactorExit::ProcessShouldExit(ExitCode::SigInt),
  - SIGQUIT => return ReactorExit::ProcessShouldExit(ExitCode::SigQuit),
  - SIGTERM => return ReactorExit::ProcessShouldExit(ExitCode::SigTerm),
  + SIGINT => break ReactorExit::ProcessShouldExit(ExitCode::SigInt),
  + SIGQUIT => break ReactorExit::ProcessShouldExit(ExitCode::SigQuit),
  + SIGTERM => break ReactorExit::ProcessShouldExit(ExitCode::SigTerm),
  _ => error!("should be unreachable - bug in signal handler"),
  }
  }
  