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" View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/e9acc5068cb01cb7a80be6de143ee63... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/e9acc5068cb01cb7a80be6de143ee63... You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications: https://gitlab.haskell.org/-/profile/notifications | Help: https://gitlab.haskell.org/help