GitLab

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

Commits:

8b83a7e2

by Duncan Coutts at 2026-07-16T09:02:50+01:00

Add interruptIOManager support for poll I/O manager

Uses the FdWakup mechanism.

A quirk we have to cope with is that we now need to poll one more fd --
the wakeup_fd_r -- but this fd has no corresponding entry in the
aiop_table. This is awkward since we have set up our aiop_poll_table to
be an auxilliary table with matching indicies.

The solution this patch uses (and described in the comments) is to have
two tables: struct pollfd *aiop_poll_table, *full_poll_table;
and to have the aiop_poll_table alias the tail of the full_poll_table.
The head entry in the full_poll_table is the extra fd. So we poll the
full_poll_table, while the aiop_poll_table still has matching indicies
with the aiop_table.

Hurrah for C aliasing rules.

26a76dbb

by Duncan Coutts at 2026-07-16T09:02:50+01:00

Add interruptIOManager support for win32 legacy I/O manager

And remove unused related helper resetAbandonRequestWait. It is not
called because the event is created in auto-reset mode, so never needs
to be reset manually.

fe8915f1

by Duncan Coutts at 2026-07-16T09:02:50+01:00

Note lack of interruptIOManager support for WinIO I/O manager

Though there's a plausible design, we can't sanely test it at the moment
due to related WinIO bugs. Filed as issue #27403.

9 changed files:

rts/IOManager.c
rts/IOManagerInternals.h
rts/posix/FdWakeup.h
rts/posix/Poll.c
rts/posix/Poll.h
rts/win32/AsyncMIO.c
rts/win32/AsyncMIO.h
rts/win32/AwaitEvent.c
rts/win32/AwaitEvent.h

Changes:

rts/IOManager.c

@@ -745,7 +745,7 @@ bool awaitCompletedTimeoutsOrIO(CapIOManager *iomgr)
  #if defined(IOMGR_ENABLED_POLL)
          case IO_MANAGER_POLL:
 -          awaitCompletedTimeoutsOrIOPoll(iomgr);
 +          completed = awaitCompletedTimeoutsOrIOPoll(iomgr);
            break;
  #endif
@@ -757,7 +757,7 @@ bool awaitCompletedTimeoutsOrIO(CapIOManager *iomgr)
  #if defined(IOMGR_ENABLED_WINIO)
          case IO_MANAGER_WINIO:
  #endif
 -          awaitCompletedTimeoutsOrIOWin32(iomgr->cap, true);
 +          completed = awaitCompletedTimeoutsOrIOWin32(iomgr->cap, true);
            break;
  #endif
          default:
@@ -784,6 +784,26 @@ void interruptIOManager(CapIOManager *iomgr)
              break;
  #endif
 +#if defined(IOMGR_ENABLED_POLL)
 +        case IO_MANAGER_POLL:
 +            interruptIOManagerPoll(iomgr);
 +            break;
 +#endif
++
 +#if defined(IOMGR_ENABLED_WIN32_LEGACY)
 +        case IO_MANAGER_WIN32_LEGACY:
 +            abandonRequestWait();
 +            break;
 +#endif
++
 +#if defined(IOMGR_ENABLED_WINIO)
 +        case IO_MANAGER_WINIO:
 +            /* FIXME: no support yet for interrupting in WinIO I/O manager
 +             * See issue #27403
 +             */
 +            break;
 +#endif
++
          default:
              break;
+     }

rts/IOManagerInternals.h

@@ -46,10 +46,12 @@ struct _CapIOManager {
      StgTSO *sleeping_queue;
  #endif
 -#if defined(IOMGR_ENABLED_SELECT)
 -    /* FDs for interrupting up the I/O manager when it is blocked waiting */
 +#if defined(IOMGR_ENABLED_SELECT) || defined(IOMGR_ENABLED_POLL)
 +#if defined(HAVE_PREEMPTION)
 +    /* FDs for waking up the I/O manager when it is blocked waiting */
      int interrupt_fd_r, interrupt_fd_w;
  #endif
 +#endif
  #if defined(IOMGR_ENABLED_POLL)
      /* AIOP and timeout collections shared by several I/O manager impls */
@@ -58,8 +60,11 @@ struct _CapIOManager {
  #endif
  #if defined(IOMGR_ENABLED_POLL)
 -    /* Auxiliary table with size and indexes matching the aiop_table */
 -    struct pollfd *aiop_poll_table;
 +    /* Auxiliary table with size and indexes matching the aiop_table. This is
 +     * aliased to the tail of the full poll table, which has a head entry for
 +     * the wakeup_fd_r above, so we can also poll that fd.
 +     */
 +    struct pollfd *aiop_poll_table, *full_poll_table;
  #endif
  #if defined(IOMGR_ENABLED_WIN32_LEGACY)

rts/posix/FdWakeup.h

@@ -29,12 +29,14 @@
  #include "BeginPrivate.h"
 +#if defined(HAVE_PREEMPTION)
  void newFdWakeup(int *fd_r, int *fd_w);
  void closeFdWakeup(int fd_r, int fd_w);
  /* This is safe to use from a signal handler */
  void sendFdWakeup(int fd_w);
  void collectFdWakeup(int fd_r);
 +#endif
  #include "EndPrivate.h"

rts/posix/Poll.c

@@ -41,6 +41,7 @@
  #include "IOManagerInternals.h"
  #include "Timeout.h"
 +#include "FdWakeup.h"
  /******************************************************************************
@@ -107,8 +108,9 @@ timeout (if any) as the poll() timeout parameter.
  The CapIOManager structure for this I/O manager contains:
      ClosureTable     aiop_table;
 -    struct pollfd   *aiop_poll_table;
 +    struct pollfd   *aiop_poll_table, *full_poll_table;
      StgTimeoutQueue *timeout_queue;
 +    int interrupt_fd_r, interrupt_fd_w;
  We also support the Linux-specific ppoll API which supports higher resolution
  time delays -- nanoseconds rather than milliseconds as in classic poll(). It
@@ -117,6 +119,15 @@ also allows the signal mask to be adjusted, but we do not make use of this.
     int ppoll(struct pollfd *fds, nfds_t nfds,
             const struct timespec *tmo_p, const sigset_t *sigmask);
 +We have both aiop_poll_table and full_poll_table. This is to cope with needing
 +to wait on the special extra file descriptor interrupt_fd_r. This fd is used to
 +support waking the I/O manager when we are blocked in a poll call. This
 +requires waiting on an extra fd that has no corresponding entry in the
 +aiop_table. To manage this quirk, we alias the aiop_poll_table to be the tail
 +of the full_poll_table and have the first entry of the full_poll_table be the
 +interrupt_fd_r. This means the aiop_poll_table indicies match up exactly with
 +the aiop_table, but still allows the full_poll_table to have an extra entry.
++
  ******************************************************************************/
  /* Forward declarations */
@@ -129,16 +140,34 @@ static void reportPollError(int res, nfds_t nfds) STG_NORETURN;
  void initCapabilityIOManagerPoll(CapIOManager *iomgr)
+ {
      initClosureTable(&iomgr->aiop_table, ClosureTableCompact);
 -    iomgr->aiop_poll_table = NULL;
      iomgr->timeout_queue = emptyTimeoutQueue();
++
 +#if defined(HAVE_PREEMPTION)
 +    newFdWakeup(&iomgr->interrupt_fd_r, &iomgr->interrupt_fd_w);
 +#endif
++
 +    iomgr->full_poll_table = stgMallocBytes(sizeof(struct pollfd) /* size 1 */,
 +                                            "initCapabilityIOManagerPoll");
 +    iomgr->full_poll_table[0] = (struct pollfd) {
 +#if defined(HAVE_PREEMPTION)
 +                                  .fd      = iomgr->interrupt_fd_r,
 +                                  .events  = POLLIN,
 +#else
 +                                  .fd      = -1, // unused
 +                                  .events  = 0,  // unused
 +#endif
 +                                  .revents = 0
 +                                };
 +    iomgr->aiop_poll_table = iomgr->full_poll_table+1; /* hence empty */
+ }
  void freeCapabilityIOManagerPoll(CapIOManager *iomgr)
+ {
 -    if (iomgr->aiop_poll_table) {
 -        stgFree(iomgr->aiop_poll_table);
 -    }
 +    stgFree(iomgr->full_poll_table);
 +#if defined(HAVE_PREEMPTION)
 +    closeFdWakeup(iomgr->interrupt_fd_r, iomgr->interrupt_fd_w);
 +#endif
+ }
@@ -295,7 +324,7 @@ static void notifyIOCompletion(CapIOManager *iomgr, StgAsyncIOOp *aiop)
+ }
 -static void processIOCompletions(CapIOManager *iomgr, int ncompletions)
 +static bool processIOCompletions(CapIOManager *iomgr, int ncompletions)
+ {
      /* The scheme we use with poll is that we have a dense poll table, and a
       * corresponding table that maps to the closure table index. The poll
@@ -305,6 +334,19 @@ static void processIOCompletions(CapIOManager *iomgr, int ncompletions)
       */
      debugTrace(DEBUG_iomanager, "processIOCompletions(ncompletions = %d)",
                                  ncompletions);
++
 +    bool interrupt = false;
 +#if defined(HAVE_PREEMPTION)
 +    /* If the interrupt_fd_r is ready, collect it */
 +    if (iomgr->full_poll_table[0].revents) {
 +        ASSERT(iomgr->full_poll_table[0].fd == iomgr->interrupt_fd_r);
 +        collectFdWakeup(iomgr->interrupt_fd_r);
 +        ncompletions--;
 +        interrupt = true;
 +        debugTrace(DEBUG_iomanager, "Received interrupt in poll I/O manager");
 +    }
 +#endif
++
      struct pollfd *aiop_poll_table = iomgr->aiop_poll_table;
      int n = ncompletions;
      int i = 0;
@@ -357,11 +399,14 @@ static void processIOCompletions(CapIOManager *iomgr, int ncompletions)
              i++;
+         }
+     }
 +    return interrupt;
+ }
  void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
+ {
 +    ASSERT(iomgr->aiop_poll_table == iomgr->full_poll_table+1);
++
      if (!isEmptyTimeoutQueue(iomgr->timeout_queue)) {
          Time now = getProcessElapsedTime();
          processTimeoutCompletions(iomgr, now);
@@ -369,20 +414,28 @@ void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
      if (!isEmptyClosureTable(&iomgr->aiop_table)) {
 -        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table);
 +#if defined(HAVE_PREEMPTION)
 +        /* the full_poll_table includes interrupt_fd_r */
 +        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table) + 1;
 +        struct pollfd *poll_table = iomgr->full_poll_table;
 +#else
 +        /* the aiop_poll_table does not include interrupt_fd_r */
 +        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table) + 0;
 +        struct pollfd *poll_table = iomgr->aiop_poll_table;
 +#endif
          /* Poll for I/O readiness, without waiting. */
  #if defined(HAVE_DECL_PPOLL) && HAVE_DECL_PPOLL == 1
          /* We could use poll here, since we use no timeout, but for
             consistency we use the same syscall as at the other call site. */
          struct timespec tv = (struct timespec) { .tv_sec = 0, .tv_nsec = 0 };
 -        int res = ppoll(iomgr->aiop_poll_table, nfds, &tv, NULL);
 +        int res = ppoll(poll_table, nfds, &tv, NULL);
          debugTrace(DEBUG_iomanager,
                     "ppoll(nfds = %d, timeout.sec = 0, timeout.nsec = 0) = %d",
                     nfds, res);
  #else
 -        int res = poll(iomgr->aiop_poll_table, nfds, 0);
 +        int res = poll(poll_table, nfds, 0);
          debugTrace(DEBUG_iomanager,
                     "poll(nfds = %d, timeout_ms = 0) = %d",
@@ -408,8 +461,12 @@ void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
+ }
 -void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
 +bool awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
+ {
 +    bool interrupt = false; /* got woken up via interruptIOManager */
++
 +    ASSERT(iomgr->aiop_poll_table == iomgr->full_poll_table+1);
++
      /* Loop until we've woken up some threads. This loop is needed because the
       * poll() timing isn't accurate, we sometimes sleep for a while but not
       * long enough to wake up a thread in a threadDelay. Or we may need to
@@ -431,6 +488,16 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
           */
          bool wait = emptyRunQueue(iomgr->cap);
 +#if defined(HAVE_PREEMPTION)
 +        /* the full_poll_table includes interrupt_fd_r */
 +        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table) + 1;
 +        struct pollfd *poll_table = iomgr->full_poll_table;
 +#else
 +        /* the aiop_poll_table does not include interrupt_fd_r */
 +        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table) + 0;
 +        struct pollfd *poll_table = iomgr->aiop_poll_table;
 +#endif
++
          /* Decide if we are going to wait if no I/O is ready, either:
           * poll only, wait indefinitely, or wait until a timeout.
           */
@@ -442,9 +509,8 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
  #endif
          /* Check for I/O readiness, possibly waiting. */
 -        nfds_t nfds = sizeClosureTable(&iomgr->aiop_table);
  #if defined(HAVE_DECL_PPOLL) && HAVE_DECL_PPOLL == 1
 -        int res = ppoll(iomgr->aiop_poll_table, nfds, timeout_ns, NULL);
 +        int res = ppoll(poll_table, nfds, timeout_ns, NULL);
          debugTrace(DEBUG_iomanager,
                     "ppoll(nfds = %d, timeout.sec = %d, timeout.nsec = %d) = %d",
@@ -452,7 +518,7 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
                           timeout_ns == NULL ?  0 : timeout_ns->tv_nsec,
                     res);
  #else
 -        int res = poll(iomgr->aiop_poll_table, nfds, timeout_ms);
 +        int res = poll(poll_table, nfds, timeout_ms);
          debugTrace(DEBUG_iomanager,
                     "poll(nfds = %d, timeout_ms = %d) = %d",
@@ -474,7 +540,7 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
          } else if (res > 0) {
              int ncompletions = res;
              ASSERT(ncompletions <= (int)nfds);
 -            processIOCompletions(iomgr, ncompletions);
 +            interrupt = processIOCompletions(iomgr, ncompletions);
              // FIXME: do we also need to check for timeout completions now?
              // we have a non-empty queue, but if !wait then we have also moved
              // on and so we sould check for timeouts.
@@ -502,7 +568,9 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr)
+         }
      } while (emptyRunQueue(iomgr->cap)
 +         && !interrupt
           && (getSchedState() == SCHED_RUNNING));
 +    return !interrupt;
+ }
  static void reportPollError(int res, nfds_t nfds)
@@ -521,6 +589,14 @@ static void reportPollError(int res, nfds_t nfds)
+ }
 +void interruptIOManagerPoll(CapIOManager *iomgr)
 +{
 +#if defined(HAVE_PREEMPTION)
 +    sendFdWakeup(iomgr->interrupt_fd_w);
 +#endif
 +}
++
++
  /* Helper function to double the size of the aiop_table and aiop_poll_table.
   */
  static bool enlargeTables(CapIOManager *iomgr)
@@ -531,13 +607,17 @@ static bool enlargeTables(CapIOManager *iomgr)
      bool ok = enlargeClosureTable(iomgr->cap, &iomgr->aiop_table, newcapacity);
      if (RTS_UNLIKELY(!ok)) return false;
 -    /* Update the auxiliary aiop_poll_table to match */
 -    struct pollfd *aiop_poll_table;
 -    aiop_poll_table = stgReallocBytes(iomgr->aiop_poll_table,
 -                                      sizeof(struct pollfd) * newcapacity,
 -                                      "Poll.c: enlargeTables");
 -    iomgr->aiop_poll_table = aiop_poll_table;
 +    /* Update the auxiliary aiop_poll_table to match. The full_poll_table is
 +     * one bigger than the aiop_poll_table, since it has an extra entry at the
 +     * front for interrupt_fd_r, with no corresponding aiop. */
 +    iomgr->full_poll_table =
 +        stgReallocBytes(iomgr->full_poll_table,
 +                        sizeof(struct pollfd) * (newcapacity+1),
 +                        "Poll.c: enlargeTables");
 +    iomgr->aiop_poll_table = iomgr->full_poll_table+1;
++
      /* Initialise the new part of the aiop_poll_table */
 +    struct pollfd *aiop_poll_table = iomgr->aiop_poll_table;
      for (int i = oldcapacity; i < newcapacity; i++) {
          aiop_poll_table[i] = (struct pollfd) {
                                 .fd      = -1,

rts/posix/Poll.h

@@ -32,7 +32,8 @@ void asyncIOCancelPoll(CapIOManager *iomgr, StgAsyncIOOp *aiop);
  /* Scheduler operations */
  bool anyPendingTimeoutsOrIOPoll(CapIOManager *iomgr);
  void pollCompletedTimeoutsOrIOPoll(CapIOManager *iomgr);
 -void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr);
 +bool awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr);
 +void interruptIOManagerPoll(CapIOManager *iomgr);
  #endif /* IOMGR_ENABLED_POLL */

rts/win32/AsyncMIO.c

@@ -221,8 +221,12 @@ shutdownAsyncIO(bool wait_threads)
   * requests to make further progress. In the latter scenario,
   * awaitRequests() will simply block waiting for worker threads
   * to complete if the 'completedTable' is empty.
 + *
 + * The result reports if the wait completed successfully (typically with some
 + * work available), or was interrupted by abandonRequestWait(), with true
 + * meaning completed, and false meaning interrupted.
   */
 -int
 +bool
  awaitRequests(bool wait)
+ {
  #if !defined(THREADED_RTS)
@@ -246,7 +250,7 @@ start:
  #endif
          ) {
          OS_RELEASE_LOCK(&queue_lock);
 -        return 0;
 +        return true;
+     }
      if (completed_hw == 0) {
          // empty table, drop lock and wait
@@ -259,22 +263,24 @@ start:
                  // a request was completed
                  break;
              case WAIT_OBJECT_0 + 1:
 +                // abandon_req_wait signaled, by abandonRequestWait()
 +                return false;
              case WAIT_TIMEOUT:
                  // timeout (unlikely) or told to abandon waiting
 -                return 0;
 +                return true;
              case WAIT_FAILED: {
                  DWORD dw = GetLastError();
                  fprintf(stderr, "awaitRequests: wait failed -- "
                                  "error code: %lu\n", dw); fflush(stderr);
 -                return 0;
 +                return true;
+             }
              default:
                  fprintf(stderr, "awaitRequests: unexpected wait return "
                                  "code %lu\n", dwRes); fflush(stderr);
 -                return 0;
 +                return true;
+             }
          } else {
 -            return 0;
 +            return true;
+         }
          goto start;
      } else {
@@ -352,7 +358,7 @@ start:
          completed_hw = 0;
          ResetEvent(completed_req_event);
          OS_RELEASE_LOCK(&queue_lock);
 -        return 1;
 +        return true;
+     }
  #endif /* !THREADED_RTS */
+ }
@@ -383,12 +389,6 @@ abandonRequestWait( void )
      interruptIOManagerEvent ();
+ }
 -void
 -resetAbandonRequestWait( void )
 -{
 -    ResetEvent(abandon_req_wait);
 -}
+-
  #endif /* !defined(THREADED_RTS) */
  HsInt rts_EINTR(void)

rts/win32/AsyncMIO.h

@@ -25,7 +25,7 @@ extern unsigned int addDoProcRequest(void* proc, void* param);
  extern int  startupAsyncIO(void);
  extern void shutdownAsyncIO(bool wait_threads);
 -extern int awaitRequests(bool wait);
 +extern bool awaitRequests(bool wait);
  extern void abandonRequestWait(void);
  extern void resetAbandonRequestWait(void);

rts/win32/AwaitEvent.c

@@ -28,17 +28,21 @@
  // Protected by sched_mutex.
  static bool workerWaitingForRequests = false;
 -void
 +bool
  awaitCompletedTimeoutsOrIOWin32(Capability *cap, bool wait)
+ {
 +  bool interrupt = false;
    do {
      /* Try to de-queue completed IO requests
       */
      workerWaitingForRequests = true;
      if (is_io_mng_native_p())
        awaitAsyncRequests(wait);
 +      /* FIXME: no support yet for interrupting in WinIO I/O manager
 +       * See issue #27403
 +       */
      else
 -      awaitRequests(wait);
 +      interrupt = !awaitRequests(wait);
      workerWaitingForRequests = false;
      // If a signal was raised, we need to service it
@@ -47,11 +51,12 @@ awaitCompletedTimeoutsOrIOWin32(Capability *cap, bool wait)
      // does it and I'm feeling too paranoid to refactor it today --SDM
      if (stg_pending_events != 0) {
          startSignalHandlers(cap);
 -        return;
 +        // This will normally cause emptyRunQueue to become false and
 +        // thus we will drop out of the loop.
+     }
 -    // The return value from awaitRequests() is a red herring: ignore
 -    // it.  Return to the scheduler if !wait, or
 +    // The return value from awaitRequests() reports if it was interrupted by
 +    // abandonRequestWait(). Return to the scheduler if !wait, or
      //
      //  - we were interrupted
      //  - the run-queue is now non- empty
@@ -59,6 +64,8 @@ awaitCompletedTimeoutsOrIOWin32(Capability *cap, bool wait)
    } while (wait
             && getSchedState() == SCHED_RUNNING
             && emptyRunQueue(cap)
 +           && !interrupt
        );
 +  return !interrupt;
+ }
  #endif

rts/win32/AwaitEvent.h

@@ -2,6 +2,6 @@
  #include "BeginPrivate.h"
 -void awaitCompletedTimeoutsOrIOWin32(Capability *cap, bool wait);
 +bool awaitCompletedTimeoutsOrIOWin32(Capability *cap, bool wait);
  #include "EndPrivate.h"