GitLab

Duncan Coutts pushed to branch wip/dcoutts/io-manager-tidy at Glasgow Haskell Compiler / GHC

Commits:

b7568cd6

by Duncan Coutts at 2026-07-08T12:26:05+01:00

Add a new interruptIOManager API for the I/O managers

It will be used to interrupt awaitCompletedTimeoutsOrIO. Also update the
return type and docs for awaitCompletedTimeoutsOrIO to have it return
false when it gets interrupted, and have no useful post condition in
that case.

c2368cb2

by Duncan Coutts at 2026-07-08T12:26:05+01:00

Add interruptIOManager support for select I/O manager

Uses the FdWakup mechanism.

5 changed files:

rts/IOManager.c
rts/IOManager.h
rts/IOManagerInternals.h
rts/posix/Select.c
rts/posix/Select.h

Changes:

rts/IOManager.c

@@ -343,9 +343,7 @@ void initCapabilityIOManager(CapIOManager *iomgr)
      switch (iomgr_type) {
  #if defined(IOMGR_ENABLED_SELECT)
          case IO_MANAGER_SELECT:
 -            iomgr->blocked_queue_hd = END_TSO_QUEUE;
 -            iomgr->blocked_queue_tl = END_TSO_QUEUE;
 -            iomgr->sleeping_queue   = END_TSO_QUEUE;
 +            initCapabilityIOManagerSelect(iomgr);
              break;
  #endif
@@ -376,6 +374,12 @@ void initCapabilityIOManager(CapIOManager *iomgr)
  void freeCapabilityIOManager(CapIOManager *iomgr)
+ {
      switch (iomgr_type) {
 +#if defined(IOMGR_ENABLED_SELECT)
 +        case IO_MANAGER_SELECT:
 +            freeCapabilityIOManagerSelect(iomgr);
 +            break;
 +#endif
++
  #if defined(IOMGR_ENABLED_POLL)
          case IO_MANAGER_POLL:
              freeCapabilityIOManagerPoll(iomgr);
@@ -728,13 +732,14 @@ void pollCompletedTimeoutsOrIO(CapIOManager *iomgr)
+ }
 -void awaitCompletedTimeoutsOrIO(CapIOManager *iomgr)
 +bool awaitCompletedTimeoutsOrIO(CapIOManager *iomgr)
+ {
      debugTrace(DEBUG_iomanager, "waiting for completed IO or timeouts");
 +    bool completed = true; // wait completed or interrupted?
      switch (iomgr_type) {
  #if defined(IOMGR_ENABLED_SELECT)
          case IO_MANAGER_SELECT:
 -          awaitCompletedTimeoutsOrIOSelect(iomgr, true);
 +          completed = awaitCompletedTimeoutsOrIOSelect(iomgr, true);
            break;
  #endif
@@ -756,9 +761,32 @@ void awaitCompletedTimeoutsOrIO(CapIOManager *iomgr)
            break;
  #endif
          default:
 -            barf("pollCompletedTimeoutsOrIO not implemented");
 +            barf("awaitCompletedTimeoutsOrIO not implemented");
 +    }
 +    ASSERT(!emptyRunQueue(iomgr->cap) ||
 +           getSchedState() != SCHED_RUNNING ||
 +           !completed);
 +    return completed;
 +}
++
++
 +/* Interrupt the I/O manager if it is blocked in awaitCompletedTimeoutsOrIO,
 + * causing it to return early and return false.
 + */
 +void interruptIOManager(CapIOManager *iomgr)
 +{
 +    debugTrace(DEBUG_iomanager, "Interrupting the I/O manager...");
 +    switch (iomgr_type) {
++
 +#if defined(IOMGR_ENABLED_SELECT)
 +        case IO_MANAGER_SELECT:
 +            interruptIOManagerSelect(iomgr);
 +            break;
 +#endif
++
 +        default:
 +            break;
+     }
 -    ASSERT(!emptyRunQueue(iomgr->cap) || getSchedState() != SCHED_RUNNING);
+ }

rts/IOManager.h

@@ -365,20 +365,32 @@ bool anyPendingTimeoutsOrIO(CapIOManager *iomgr);
   */
  void pollCompletedTimeoutsOrIO(CapIOManager *iomgr);
 - /* If there are any completed I/O operations or expired timers, process the
 +/* If there are any completed I/O operations or expired timers, process the
   * completions as appropriate. If there are none, wait until I/O or a timer
   * does complete (or we get a signal with a handler) and process the
   * completions as appropriate.
+  *
 - * Upon return this guarantees that the scheduler run queue is non-empty or
 - * that the scheduler is no longer in the running state. Succinctly, the
 - * post-condition is (!emptyRunQueue(cap) || getSchedState() != SCHED_RUNNING).
 + * Upon returning true this guarantees that the scheduler run queue is
 + * non-empty or that the scheduler is no longer in the running state.
 + * Succinctly, the post-condition in the return true case is
 + * (!emptyRunQueue(cap) || getSchedState() != SCHED_RUNNING).
 + * A false result means the wait was interrupted by interruptIOManager, and
 + * there is no post-condition in this case.
+  *
   * This is only expected to be called if anyPendingTimeoutsOrIO() returns true,
   * i.e. there actually is something to wait for.
+  *
   * Called from schedule() both *before* and *after* scheduleDetectDeadlock().
   */
 -void awaitCompletedTimeoutsOrIO(CapIOManager *iomgr);
 +bool awaitCompletedTimeoutsOrIO(CapIOManager *iomgr);
++
 +/* Interrupt the I/O manager if it is blocked in awaitCompletedTimeoutsOrIO,
 + * causing it to return early.
 + *
 + * Its use is inherently concurrent and racy: the interrupt races against any
 + * I/O or timer completion. This does not matter for the intended use case of
 + * returning control to the scheduler.
 + */
 +void interruptIOManager(CapIOManager *iomgr);
  #include "EndPrivate.h"

rts/IOManagerInternals.h

@@ -46,6 +46,11 @@ struct _CapIOManager {
      StgTSO *sleeping_queue;
  #endif
 +#if defined(IOMGR_ENABLED_SELECT)
 +    /* FDs for interrupting up the I/O manager when it is blocked waiting */
 +    int interrupt_fd_r, interrupt_fd_w;
 +#endif
++
  #if defined(IOMGR_ENABLED_POLL)
      /* AIOP and timeout collections shared by several I/O manager impls */
      ClosureTable     aiop_table;

rts/posix/Select.c

@@ -22,6 +22,7 @@
  #include "IOManagerInternals.h"
  #include "Stats.h"
  #include "GetTime.h"
 +#include "FdWakeup.h"
  # if defined(HAVE_SYS_SELECT_H)
  #  include <sys/select.h>
@@ -54,6 +55,31 @@
  #define TimeToLowResTimeRoundUp(t)   (t)
  #endif
 +void initCapabilityIOManagerSelect(CapIOManager *iomgr)
 +{
 +    iomgr->blocked_queue_hd = END_TSO_QUEUE;
 +    iomgr->blocked_queue_tl = END_TSO_QUEUE;
 +    iomgr->sleeping_queue   = END_TSO_QUEUE;
++
 +#if defined(HAVE_PREEMPTION)
 +    newFdWakeup(&iomgr->interrupt_fd_r, &iomgr->interrupt_fd_w);
 +#endif
 +}
++
 +void freeCapabilityIOManagerSelect(CapIOManager *iomgr)
 +{
 +#if defined(HAVE_PREEMPTION)
 +    closeFdWakeup(iomgr->interrupt_fd_r, iomgr->interrupt_fd_w);
 +#endif
 +}
++
 +void interruptIOManagerSelect(CapIOManager *iomgr)
 +{
 +#if defined(HAVE_PREEMPTION)
 +    sendFdWakeup(iomgr->interrupt_fd_w);
 +#endif
 +}
++
  /*
   * Return the time since the program started, in LowResTime,
   * rounded down.
@@ -215,7 +241,7 @@ static enum FdState fdPollWriteState (int fd)
   * not write handles.
+  *
   */
 -void
 +bool
  awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait)
+ {
      StgTSO *tso, *prev, *next;
@@ -225,6 +251,7 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait)
      bool seen_bad_fd = false;
      struct timeval tv, *ptv;
      LowResTime now;
 +    bool interrupt = false; /* got interrupted up via interruptIOManager */
      IF_DEBUG(scheduler,
               debugBelch("scheduler: checking for threads blocked on I/O");
@@ -243,7 +270,7 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait)
        now = getLowResTimeOfDay();
        if (wakeUpSleepingThreads(iomgr, now)) {
 -          return;
 +          return true;
+       }
        /*
@@ -252,6 +279,15 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait)
        FD_ZERO(&rfd);
        FD_ZERO(&wfd);
 +#if defined(HAVE_PREEMPTION)
 +      /* We're always interested in our interrupt fd */
 +      {
 +          int fd = iomgr->interrupt_fd_r;
 +          maxfd = (fd > maxfd) ? fd : maxfd;
 +          FD_SET(fd, &rfd);
 +      }
 +#endif
++
        for(tso = iomgr->blocked_queue_hd;
            tso != END_TSO_QUEUE;
            tso = next) {
@@ -354,14 +390,14 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait)
  #if defined(RTS_USER_SIGNALS)
            if (RtsFlags.MiscFlags.install_signal_handlers && signals_pending()) {
                startSignalHandlers(iomgr->cap);
 -              return; /* still hold the lock */
 +              return true; /* still hold the lock */
+           }
  #endif
            /* we were interrupted, return to the scheduler immediately.
             */
            if (getSchedState() >= SCHED_INTERRUPTING) {
 -              return; /* still hold the lock */
 +              return true; /* still hold the lock */
+           }
            /* check for threads that need waking up
@@ -372,10 +408,19 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait)
             * I/O and run them.
             */
            if (!emptyRunQueue(iomgr->cap)) {
 -              return; /* still hold the lock */
 +              return true; /* still hold the lock */
+           }
+       }
 +#if defined(HAVE_PREEMPTION)
 +      /* If the interrupt_fd_r is ready, collect it */
 +      if (FD_ISSET(iomgr->interrupt_fd_r, &rfd)) {
 +          collectFdWakeup(iomgr->interrupt_fd_r);
 +          interrupt = true;
 +          debugTrace(DEBUG_iomanager, "Received interrupt in select I/O manager");
 +      }
 +#endif
++
        /* Step through the waiting queue, unblocking every thread that now has
         * a file descriptor in a ready state.
         */
@@ -458,7 +503,9 @@ awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait)
+       }
      } while (wait && getSchedState() == SCHED_RUNNING
 -                  && emptyRunQueue(iomgr->cap));
 +                  && emptyRunQueue(iomgr->cap)
 +                  && !interrupt);
 +    return !interrupt;
+ }
  #endif /* IOMGR_ENABLED_SELECT */

rts/posix/Select.h

@@ -15,7 +15,12 @@ typedef StgWord LowResTime;
  LowResTime getDelayTarget (HsInt us);
 -void awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait);
 +void initCapabilityIOManagerSelect(CapIOManager *iomgr);
 +void freeCapabilityIOManagerSelect(CapIOManager *iomgr);
 +void wakeupIOManagerSelect(CapIOManager *iomgr);
++
 +bool awaitCompletedTimeoutsOrIOSelect(CapIOManager *iomgr, bool wait);
 +void interruptIOManagerSelect(CapIOManager *iomgr);
  #include "EndPrivate.h"