Duncan Coutts pushed to branch wip/dcoutts/io-manager-tidy at Glasgow Haskell Compiler / GHC Commits: d37b0453 by Duncan Coutts at 2026-07-08T12:20:28+01:00 Split posix/MIO.c out of posix/Signals.c The MIO I/O manager was secretly living inside the Signals file. Now it gets its own file, like any other self-respecting I/O manager. - - - - - e15e8577 by Duncan Coutts at 2026-07-08T12:20:28+01:00 Rationalise some scheduler run queue utilities Move them all to the same place in the file. Make some static that were used only internally. Also remove a redundant assignment after calling truncateRunQueue that is already done within truncateRunQueue. - - - - - fbaa3157 by Duncan Coutts at 2026-07-08T12:20:28+01:00 Rename initIOManager{AfterFork} to {re}startIOManager These are more accurate names, since these actions happen after initialisation and are really about starting (or restarting) background threads. - - - - - 6c27b769 by Duncan Coutts at 2026-07-08T12:20:28+01:00 Free per-cap I/O managers during shutdown and forkProcess Historically this was not strictly necessary. The select and win32 legacy I/O managers did not maintain any dynamically allocated resources. The new poll one does (an auxillary table), and so this should be freed. After forkProcess, all threads get deleted. This includes threads waiting on I/O or timers. So as of this patch, resetting the I/O manager is just about tidying things up. For example, for the poll I/O manager this will reset the size of the AIOP table (which otherwise grows but never shrinks). In future however the re-initialising will become neeecessary for functionality, since some I/O managers will need to re-initialise wakeup fds that are set CLOEXEC. - - - - - 2daa8f5e by Duncan Coutts at 2026-07-08T12:26:04+01:00 Add a TODO to the MIO I/O manager The direction of travel is to make I/O managers per-capability and have all their state live in the struct CapIOManager. The MIO I/O manager however still has a number of global variables. It's not obvious how handle these globals however. - - - - - 79d1cf40 by Duncan Coutts at 2026-07-08T12:26:04+01:00 Add a FIXME note in the Poll I/O manager - - - - - fd1eca2a by Duncan Coutts at 2026-07-08T12:26:04+01:00 Add missing updateRemembSetPushClosure in poll I/O manager For the non-moving GC. - - - - - 4ecf19cf by Duncan Coutts at 2026-07-08T12:26:04+01:00 Minor doc improvement to struct StgAsyncIOOp member outcome Mention the enumeration names, as well as their numeric values. The rest of the code uses the enum names. - - - - - 7b59929c by Duncan Coutts at 2026-07-08T12:26:05+01:00 Minor doc improvements for StgTSOBlockInfo Clarify that certain union members are used only by certain legacy I/O managers. Hopefully we will be able to remove these at some point. - - - - - 03cd3ca9 by Duncan Coutts at 2026-07-08T12:26:05+01:00 Avoid exporting various win32-specific rts symbols The BeginPrivate.h / EndPrivate.h scheme works perfectly well on Windows, but all of the rts/win32/*.h files were not using it. - - - - - d9414b2c by Duncan Coutts at 2026-07-08T12:26:05+01:00 Remove wakeupIOManager, ioManagerWakeup and setIOManagerWakeupFd We no longer need wakeupIOManager for the threaded RTS case, so we can remove it and the bits only needed to support it. This includes the pipe/eventfd fd shared between the RTS and the in-library I/O manager used for waking up the I/O manager thread. The pipe/eventfd still exists, but it no longer has to be communicated to the RTS, since the RTS no longer needs to use it. So we remove the RTS API export setIOManagerWakeupFd, and remove uses of it within the I/O managers in ghc-internal. - - - - - 28 changed files: - libraries/ghc-internal/src/GHC/Internal/Event/Control.hs - libraries/ghc-internal/src/GHC/Internal/Event/Manager.hs - libraries/ghc-internal/src/GHC/Internal/Event/TimerManager.hs - rts/Capability.c - rts/IOManager.c - rts/IOManager.h - rts/RtsStartup.c - rts/RtsSymbols.c - rts/Schedule.c - rts/Schedule.h - rts/include/rts/IOInterface.h - rts/include/rts/storage/Closures.h - rts/include/rts/storage/TSO.h - + rts/posix/MIO.c - + rts/posix/MIO.h - rts/posix/Poll.c - rts/posix/Poll.h - rts/posix/Signals.c - rts/posix/Signals.h - rts/rts.cabal - rts/win32/AsyncMIO.h - rts/win32/AsyncWinIO.h - rts/win32/AwaitEvent.h - rts/win32/ConsoleHandler.h - rts/win32/MIOManager.h - rts/win32/ThrIOManager.h - rts/win32/WorkQueue.h - rts/win32/veh_excn.h Changes: ===================================== libraries/ghc-internal/src/GHC/Internal/Event/Control.hs ===================================== @@ -39,7 +39,7 @@ import GHC.Internal.Show (Show) import GHC.Internal.Types (Bool(..), Int, IO) import GHC.Internal.Word (Word8) import GHC.Internal.Foreign.C.Error (throwErrnoIfMinus1_, throwErrno, getErrno) -import GHC.Internal.Foreign.C.Types (CInt(..), CSize(..)) +import GHC.Internal.Foreign.C.Types (CSize(..)) import GHC.Internal.Foreign.ForeignPtr (ForeignPtr, mallocForeignPtrBytes, withForeignPtr) import GHC.Internal.Foreign.Marshal.Alloc (alloca, allocaBytes) import GHC.Internal.Foreign.Marshal.Array (allocaArray) @@ -51,7 +51,7 @@ import GHC.Internal.System.Posix.Types (Fd) #if defined(HAVE_EVENTFD) import GHC.Internal.Foreign.C.Error (throwErrnoIfMinus1, eBADF) -import GHC.Internal.Foreign.C.Types (CULLong(..)) +import GHC.Internal.Foreign.C.Types (CInt(..), CULLong(..)) #else import GHC.Internal.Foreign.C.Error (eAGAIN, eWOULDBLOCK, eBADF) #endif @@ -78,7 +78,10 @@ data Control = W { , wakeupReadFd :: {-# UNPACK #-} !Fd , wakeupWriteFd :: {-# UNPACK #-} !Fd #endif - , didRegisterWakeupFd :: !Bool + , didRegisterWakeupFd :: !Bool -- ^ Now redundant. Always False. + --TODO: remove ^^ this redundant field. + -- Technically, removing this is an API change to base. Sigh. + -- | Have this Control's fds been cleaned up? , controlIsDead :: !(IORef Bool) } @@ -91,8 +94,8 @@ wakeupReadFd = controlEventFd -- | Create the structure (usually a pipe) used for waking up the IO -- manager thread from another thread. -newControl :: Bool -> IO Control -newControl shouldRegister = allocaArray 2 $ \fds -> do +newControl :: IO Control +newControl = allocaArray 2 $ \fds -> do let createPipe = do throwErrnoIfMinus1_ "pipe" $ c_pipe fds rd <- peekElemOff fds 0 @@ -108,10 +111,8 @@ newControl shouldRegister = allocaArray 2 $ \fds -> do ev <- throwErrnoIfMinus1 "eventfd" $ c_eventfd 0 0 setNonBlockingFD ev True setCloseOnExec ev - when shouldRegister $ c_setIOManagerWakeupFd ev #else (wake_rd, wake_wr) <- createPipe - when shouldRegister $ c_setIOManagerWakeupFd wake_wr #endif isDead <- newIORef False return W { controlReadFd = fromIntegral ctrl_rd @@ -122,25 +123,16 @@ newControl shouldRegister = allocaArray 2 $ \fds -> do , wakeupReadFd = fromIntegral wake_rd , wakeupWriteFd = fromIntegral wake_wr #endif - , didRegisterWakeupFd = shouldRegister + , didRegisterWakeupFd = False , controlIsDead = isDead } -- | Close the control structure used by the IO manager thread. --- N.B. If this Control is the Control whose wakeup file was registered with --- the RTS, then *BEFORE* the wakeup file is closed, we must call --- c_setIOManagerWakeupFd (-1), so that the RTS does not try to use the wakeup --- file after it has been closed. --- --- Note, however, that even if we do the above, this function is still racy --- since we do not synchronize between here and ioManagerWakeup. --- ioManagerWakeup ignores failures that arise from this case. closeControl :: Control -> IO () closeControl w = do _ <- atomicSwapIORef (controlIsDead w) True _ <- c_close . fromIntegral . controlReadFd $ w _ <- c_close . fromIntegral . controlWriteFd $ w - when (didRegisterWakeupFd w) $ c_setIOManagerWakeupFd (-1) #if defined(HAVE_EVENTFD) _ <- c_close . fromIntegral . controlEventFd $ w #else @@ -248,11 +240,3 @@ foreign import ccall unsafe "sys/eventfd.h eventfd" foreign import ccall unsafe "sys/eventfd.h eventfd_write" c_eventfd_write :: CInt -> CULLong -> IO CInt #endif - -#if defined(wasm32_HOST_ARCH) -c_setIOManagerWakeupFd :: CInt -> IO () -c_setIOManagerWakeupFd _ = return () -#else -foreign import ccall unsafe "setIOManagerWakeupFd" - c_setIOManagerWakeupFd :: CInt -> IO () -#endif ===================================== libraries/ghc-internal/src/GHC/Internal/Event/Manager.hs ===================================== @@ -194,7 +194,7 @@ newWith :: Backend -> IO EventManager newWith be = do iofds <- fmap (listArray (0, callbackArraySize-1)) $ replicateM callbackArraySize (newMVar =<< IT.new 8) - ctrl <- newControl False + ctrl <- newControl state <- newIORef Created us <- newSource _ <- mkWeakIORef state $ do ===================================== libraries/ghc-internal/src/GHC/Internal/Event/TimerManager.hs ===================================== @@ -126,7 +126,7 @@ new = newWith =<< newDefaultBackend newWith :: Backend -> IO TimerManager newWith be = do timeouts <- newIORef Q.empty - ctrl <- newControl True + ctrl <- newControl state <- newIORef Created us <- newSource _ <- mkWeakIORef state $ do ===================================== rts/Capability.c ===================================== @@ -1293,6 +1293,8 @@ shutdownCapabilities(Task *task, bool safe) static void freeCapability (Capability *cap) { + freeCapabilityIOManager(cap->iomgr); + stgFree(cap->iomgr); stgFree(cap->mut_lists); stgFree(cap->saved_mut_lists); if (cap->current_segments) { ===================================== rts/IOManager.c ===================================== @@ -39,7 +39,7 @@ #endif #if defined(IOMGR_ENABLED_MIO_POSIX) -#include "posix/Signals.h" +#include "posix/MIO.h" #include "Prelude.h" #endif @@ -373,11 +373,25 @@ void initCapabilityIOManager(CapIOManager *iomgr) } +void freeCapabilityIOManager(CapIOManager *iomgr) +{ + switch (iomgr_type) { +#if defined(IOMGR_ENABLED_POLL) + case IO_MANAGER_POLL: + freeCapabilityIOManagerPoll(iomgr); + break; +#endif + default: + break; + } +} + + /* Called late in the RTS initialisation */ -void initIOManager(void) +void startIOManager(void) { - debugTrace(DEBUG_iomanager, "initialising %s I/O manager", showIOManager()); + debugTrace(DEBUG_iomanager, "starting %s I/O manager", showIOManager()); switch (iomgr_type) { @@ -441,7 +455,7 @@ void initIOManager(void) /* Called from forkProcess in the child process on the surviving capability. */ void -initIOManagerAfterFork(CapIOManager *iomgr, Capability **pcap) +restartIOManager(CapIOManager *iomgr, Capability **pcap) { switch (iomgr_type) { @@ -541,42 +555,6 @@ exitIOManager(bool wait_threads) } } -/* Wakeup hook: called from the scheduler's wakeUpRts (currently only in - * threaded mode). - */ -void wakeupIOManager(void) -{ - switch (iomgr_type) { - -#if defined(IOMGR_ENABLED_MIO_POSIX) - case IO_MANAGER_MIO_POSIX: - /* MIO Posix implementation in posix/Signals.c */ - ioManagerWakeup(); - break; -#endif -#if defined(IOMGR_ENABLED_MIO_WIN32) - case IO_MANAGER_MIO_WIN32: - /* MIO Windows implementation in win32/ThrIOManager.c - * Yes, this is shared with the WinIO (threaded) impl. - */ - ioManagerWakeup(); - break; -#endif -#if defined(IOMGR_ENABLED_WINIO) - case IO_MANAGER_WINIO: -#if defined(THREADED_RTS) - /* WinIO threaded implementation in win32/ThrIOManager.c - * Yes, this is shared with the MIO win32 impl. - */ - ioManagerWakeup(); -#endif - break; -#endif - default: - break; - } -} - void markCapabilityIOManager(evac_fn evac, void *user, CapIOManager *iomgr) { switch (iomgr_type) { ===================================== rts/IOManager.h ===================================== @@ -243,11 +243,33 @@ CapIOManager *allocCapabilityIOManager(Capability *cap); */ void initCapabilityIOManager(CapIOManager *iomgr); +/* When shutting down a capability, or after forkProcess, free the resources + * held by a CapIOManager to put it back into a state in which either it can be + * re-initialised using initCapabilityIOManager, or the whole structure freed. + * + * Note that this does not free the CapIOManager structure itself, just the + * contents. + * + * This is used during capability shutdown, during RTS shutdown. It is not used + * when reducing the number of capabilities. Capabilities are disabled rather + * than freed entirely: the I/O manager keeps running but threads that become + * runnable are migrated away. + * + * It is also used after forkProcess. + */ +void freeCapabilityIOManager(CapIOManager *iomgr); + +/* CapIOManager life cycle: + * + * alloc -> init -> free -> free struct + * ^ | + * +--------+ + */ /* Init hook: called from hs_init_ghc, very late in the startup after almost * everything else is done. */ -void initIOManager(void); +void startIOManager(void); /* Init hook: called from forkProcess in the child process on the surviving @@ -256,8 +278,8 @@ void initIOManager(void); * This is synchronous and can run Haskell code, so can change the given cap. * TODO: it would make for a cleaner API here if this were made asynchronous. */ -void initIOManagerAfterFork(CapIOManager *iomgr, - /* inout */ Capability **pcap); +void restartIOManager(CapIOManager *iomgr, + /* inout */ Capability **pcap); /* TODO: rationalise initIOManager and initIOManagerAfterFork into a single per-capability init function. @@ -284,23 +306,6 @@ void stopIOManager(void); void exitIOManager(bool wait_threads); -/* Wakeup hook: called from the scheduler's wakeUpRts (currently only in - * threaded mode). - * - * The I/O manager can be blocked waiting on I/O or timers. Sometimes there are - * other external events where we need to wake up the I/O manager and return - * to the schedulr. - * - * At the moment, all the non-threaded I/O managers will do this automagically - * since a signal will interrupt any waiting system calls, so at the moment - * the implementation for the non-threaded I/O managers does nothing. - * - * For the I/O managers in threaded mode, this arranges to unblock the I/O - * manager if it waa blocked waiting. - */ -void wakeupIOManager(void); - - /* GC hook: mark any per-capability GC roots the I/O manager uses. */ void markCapabilityIOManager(evac_fn evac, void *user, CapIOManager *iomgr); ===================================== rts/RtsStartup.c ===================================== @@ -427,7 +427,7 @@ hs_init_ghc(int *argc, char **argv[], RtsConfig rts_config) } #endif - initIOManager(); + startIOManager(); x86_init_fpu(); ===================================== rts/RtsSymbols.c ===================================== @@ -265,7 +265,6 @@ extern char **environ; #define RTS_USER_SIGNALS_SYMBOLS \ SymI_HasProto(setIOManagerControlFd) \ SymI_HasProto(setTimerManagerControlFd) \ - SymI_HasProto(setIOManagerWakeupFd) \ SymI_HasProto(blockUserSignals) \ SymI_HasProto(unblockUserSignals) #else ===================================== rts/Schedule.c ===================================== @@ -175,6 +175,11 @@ static void deleteAllThreads (void); static void deleteThread_(StgTSO *tso); #endif +#if defined(FORKPROCESS_PRIMOP_SUPPORTED) +static void truncateRunQueue(Capability *cap); +#endif +static StgTSO *popRunQueue (Capability *cap); + /* --------------------------------------------------------------------------- Main scheduling loop. @@ -592,38 +597,6 @@ run_thread: } /* end of while() */ } -/* ----------------------------------------------------------------------------- - * Run queue operations - * -------------------------------------------------------------------------- */ - -static void -removeFromRunQueue (Capability *cap, StgTSO *tso) -{ - if (tso->block_info.prev == END_TSO_QUEUE) { - ASSERT(cap->run_queue_hd == tso); - cap->run_queue_hd = tso->_link; - } else { - setTSOLink(cap, tso->block_info.prev, tso->_link); - } - if (tso->_link == END_TSO_QUEUE) { - ASSERT(cap->run_queue_tl == tso); - cap->run_queue_tl = tso->block_info.prev; - } else { - setTSOPrev(cap, tso->_link, tso->block_info.prev); - } - tso->_link = tso->block_info.prev = END_TSO_QUEUE; - cap->n_run_queue--; - - IF_DEBUG(sanity, checkRunQueue(cap)); -} - -void -promoteInRunQueue (Capability *cap, StgTSO *tso) -{ - removeFromRunQueue(cap, tso); - pushOnRunQueue(cap, tso); -} - /* ----------------------------------------------------------------------------- * scheduleFindWork() * @@ -2199,7 +2172,15 @@ forkProcess(HsStablePtr *entry // bound threads for which the corresponding Task does not // exist. truncateRunQueue(cap); - cap->n_run_queue = 0; + + // Reset and re-initialise the capability's I/O manager, + // to get the I/O manager ready again. + // + // Any threads waiting on I/O or timers should have been + // removed from I/O manager queues by deleteThread_ above. + // TODO: but we could assert that here. + freeCapabilityIOManager(cap->iomgr); + initCapabilityIOManager(cap->iomgr); // Any suspended C-calling Tasks are no more, their OS threads // don't exist now: @@ -2216,7 +2197,7 @@ forkProcess(HsStablePtr *entry cap->n_returning_tasks = 0; #endif - // Release all caps except 0, we'll use that for starting + // Release all caps except 0, we'll use that for restarting // the IO manager and running the client action below. if (cap->no != 0) { task->cap = cap; @@ -2240,7 +2221,7 @@ forkProcess(HsStablePtr *entry // like startup event, capabilities, process info etc traceTaskCreate(task, cap); - initIOManagerAfterFork(cap->iomgr, &cap); + restartIOManager(cap->iomgr, &cap); // start timer after the IOManager is initialized // (the idle GC may wake up the IOManager) @@ -2339,6 +2320,10 @@ setNumCapabilities (uint32_t new_n_capabilities USED_IF_THREADS) // the capability; we don't have to worry about GC data // structures, the nursery, etc. // + // This approach also handles threads blocked on I/O. Such threads + // remain blocked, and when I/O completes and threads become runnable + // then they are migrated away. + // for (n = new_n_capabilities; n < enabled_capabilities; n++) { getCapability(n)->disabled = true; traceCapDisable(getCapability(n)); @@ -3016,7 +3001,7 @@ pushOnRunQueue (Capability *cap, StgTSO *tso) cap->n_run_queue++; } -StgTSO *popRunQueue (Capability *cap) +static StgTSO *popRunQueue (Capability *cap) { ASSERT(cap->n_run_queue > 0); StgTSO *t = cap->run_queue_hd; @@ -3036,6 +3021,45 @@ StgTSO *popRunQueue (Capability *cap) return t; } +#if defined(FORKPROCESS_PRIMOP_SUPPORTED) +static void truncateRunQueue(Capability *cap) +{ + // Can only be called by the task owning the capability. + TSAN_ANNOTATE_BENIGN_RACE(&cap->run_queue_hd, "truncateRunQueue"); + TSAN_ANNOTATE_BENIGN_RACE(&cap->run_queue_tl, "truncateRunQueue"); + TSAN_ANNOTATE_BENIGN_RACE(&cap->n_run_queue, "truncateRunQueue"); + cap->run_queue_hd = END_TSO_QUEUE; + cap->run_queue_tl = END_TSO_QUEUE; + cap->n_run_queue = 0; +} +#endif + +static void removeFromRunQueue (Capability *cap, StgTSO *tso) +{ + if (tso->block_info.prev == END_TSO_QUEUE) { + ASSERT(cap->run_queue_hd == tso); + cap->run_queue_hd = tso->_link; + } else { + setTSOLink(cap, tso->block_info.prev, tso->_link); + } + if (tso->_link == END_TSO_QUEUE) { + ASSERT(cap->run_queue_tl == tso); + cap->run_queue_tl = tso->block_info.prev; + } else { + setTSOPrev(cap, tso->_link, tso->block_info.prev); + } + tso->_link = tso->block_info.prev = END_TSO_QUEUE; + cap->n_run_queue--; + + IF_DEBUG(sanity, checkRunQueue(cap)); +} + +void promoteInRunQueue (Capability *cap, StgTSO *tso) +{ + removeFromRunQueue(cap, tso); + pushOnRunQueue(cap, tso); +} + /* ----------------------------------------------------------------------------- raiseExceptionHelper ===================================== rts/Schedule.h ===================================== @@ -164,10 +164,6 @@ void appendToRunQueue (Capability *cap, StgTSO *tso); */ void pushOnRunQueue (Capability *cap, StgTSO *tso); -/* Pop the first thread off the runnable queue. - */ -StgTSO *popRunQueue (Capability *cap); - INLINE_HEADER StgTSO * peekRunQueue (Capability *cap) { @@ -184,18 +180,6 @@ emptyRunQueue(Capability *cap) return cap->n_run_queue == 0; } -INLINE_HEADER void -truncateRunQueue(Capability *cap) -{ - // Can only be called by the task owning the capability. - TSAN_ANNOTATE_BENIGN_RACE(&cap->run_queue_hd, "truncateRunQueue"); - TSAN_ANNOTATE_BENIGN_RACE(&cap->run_queue_tl, "truncateRunQueue"); - TSAN_ANNOTATE_BENIGN_RACE(&cap->n_run_queue, "truncateRunQueue"); - cap->run_queue_hd = END_TSO_QUEUE; - cap->run_queue_tl = END_TSO_QUEUE; - cap->n_run_queue = 0; -} - #endif /* !IN_STG_CODE */ #include "EndPrivate.h" ===================================== rts/include/rts/IOInterface.h ===================================== @@ -33,7 +33,6 @@ void ioManagerFinished (void); void setIOManagerControlFd (uint32_t cap_no, int fd); void setTimerManagerControlFd(int fd); -void setIOManagerWakeupFd (int fd); #endif ===================================== rts/include/rts/storage/Closures.h ===================================== @@ -829,11 +829,11 @@ typedef struct { // here due to portability concerns for this C bitfield. uint16_t notify_type: 2; - // The outcome: - // 0: still in-progress, no detail - // 1: success, result field contains the result code - // 2: failed, error field contains the error code - // 3: cancelled, no detail + // The outcome (see enum IOOpOutcome): + // 0: IOOpOutcomeInFlight: still in-progress, no further detail. + // 1: IOOpOutcomeSuccess: result field contains the result code. + // 2: IOOpOutcomeFailed: error field contains the error code. + // 3: IOOpOutcomeCancelled: cancelled, no further detail. uint16_t outcome: 2; // 12 bits going spare! ===================================== rts/include/rts/storage/TSO.h ===================================== @@ -48,14 +48,16 @@ typedef union { StgAsyncIOOp *aiop; StgTimeoutQueue *timeout; #if defined(mingw32_HOST_OS) + // Only used by the Legacy Win32 I/O manager: the async request id for the + // operation. StgWord async_reqID; #endif #if !defined(THREADED_RTS) StgWord target; - // Only for the non-threaded RTS: the target time for a thread - // blocked in threadDelay, in units of 1ms. This is a - // compromise: we don't want to take up much space in the TSO. If - // you want better resolution for threadDelay, use -threaded. + // Only for the legacy select I/O manager: the target time for a thread + // blocked in threadDelay, in units of 1ms. This is a compromise: we don't + // want to take up much space in the TSO. If you want better resolution + // for threadDelay, use *literally any* other I/O manager. #endif } StgTSOBlockInfo; ===================================== rts/posix/MIO.c ===================================== @@ -0,0 +1,122 @@ +/* ----------------------------------------------------------------------------- + * + * (c) The GHC Team, 1998-2005 + * + * The MIO I/O manager RTS stub. + * + * This I/O manager lives primarily in Haskell in the ghc-internal package. + * This file implementes the required subset of the RTS's internal I/O manager + * API. + * + * ---------------------------------------------------------------------------*/ + +#include "rts/PosixSource.h" +#include "Rts.h" + +#include "Schedule.h" +#include "RtsUtils.h" +#include "Prelude.h" +#include "ThreadLabels.h" + +#include "MIO.h" +#include "IOManager.h" +#include "IOManagerInternals.h" + +#if defined(HAVE_ERRNO_H) +# include +#endif + +#include +#include + +// Here's the pipe into which we will send our signals +static int timer_manager_control_wr_fd = -1; +// TODO: Eliminate these globals. Put then into the CapIOManager, but the +// problem is these are shared across all caps, not per cap. + +#define IO_MANAGER_DIE 0xfe + +void setTimerManagerControlFd(int fd) { + RELAXED_STORE(&timer_manager_control_wr_fd, fd); +} + +#if defined(THREADED_RTS) +void timerManagerNotifySignal(int sig, siginfo_t *info) +{ + StgWord8 buf[sizeof(siginfo_t) + 1]; + int r; + + buf[0] = sig; + if (info == NULL) { + // info may be NULL on Solaris (see #3790) + memset(buf+1, 0, sizeof(siginfo_t)); + } else { + memcpy(buf+1, info, sizeof(siginfo_t)); + } + + int timer_control_fd = RELAXED_LOAD(&timer_manager_control_wr_fd); + if (0 <= timer_control_fd) + { + r = write(timer_control_fd, buf, sizeof(siginfo_t)+1); + if (r == -1 && errno == EAGAIN) { + errorBelch("lost signal due to full pipe: %d\n", sig); + } + } + + // If the IO manager hasn't told us what the FD of the write end + // of its pipe is, there's not much we can do here, so just ignore + // the signal.. +} +#endif + + +#if defined(THREADED_RTS) +void +ioManagerDie (void) +{ + StgWord8 byte = (StgWord8)IO_MANAGER_DIE; + uint32_t i; + int r; + + { + // Shut down timer manager + const int fd = RELAXED_LOAD(&timer_manager_control_wr_fd); + if (0 <= fd) { + r = write(fd, &byte, 1); + if (r == -1) { sysErrorBelch("ioManagerDie: write"); } + RELAXED_STORE(&timer_manager_control_wr_fd, -1); + } + } + + { + // Shut down IO managers + for (i=0; i < getNumCapabilities(); i++) { + const int fd = RELAXED_LOAD(&getCapability(i)->iomgr->control_fd); + if (0 <= fd) { + r = write(fd, &byte, 1); + if (r == -1) { sysErrorBelch("ioManagerDie: write"); } + RELAXED_STORE(&getCapability(i)->iomgr->control_fd, -1); + } + } + } +} + +void +ioManagerStartCap (Capability **cap) +{ + rts_evalIO(cap,ensureIOManagerIsRunning_closure,NULL); +} + +void +ioManagerStart (void) +{ + // Make sure the IO manager thread is running + Capability *cap; + if (SEQ_CST_LOAD(&timer_manager_control_wr_fd) < 0) { + cap = rts_lock(); + ioManagerStartCap(&cap); + rts_unlock(cap); + } +} +#endif + ===================================== rts/posix/MIO.h ===================================== @@ -0,0 +1,29 @@ +/* ----------------------------------------------------------------------------- + * + * (c) The GHC Team, 1998-2005 + * + * Signal processing / handling. + * + * ---------------------------------------------------------------------------*/ + +#pragma once + +#include "IOManager.h" + +#if defined(HAVE_SIGNAL_H) +# include +#endif + +#include "BeginPrivate.h" + +/* Communicating with the IO manager thread (see GHC.Conc). + */ +#if defined(THREADED_RTS) +void ioManagerDie (void); +void ioManagerStart (void); +void ioManagerStartCap (/* inout */ Capability **cap); + +void timerManagerNotifySignal(int sig, siginfo_t *info); +#endif + +#include "EndPrivate.h" ===================================== rts/posix/Poll.c ===================================== @@ -134,6 +134,14 @@ void initCapabilityIOManagerPoll(CapIOManager *iomgr) } +void freeCapabilityIOManagerPoll(CapIOManager *iomgr) +{ + if (iomgr->aiop_poll_table) { + stgFree(iomgr->aiop_poll_table); + } +} + + /* Used to implement syncIOWaitReady. * Result is true on success, or false on allocation failure. */ bool syncIOWaitReadyPoll(CapIOManager *iomgr, StgTSO *tso, @@ -194,6 +202,13 @@ void syncIOCancelPoll(CapIOManager *iomgr, StgTSO *tso) * status, as that is done by removeFromQueues (in the throwTo* functions). */ tso->block_info.closure = (StgClosure *)END_TSO_QUEUE; + + /* We are in the TSO case, where the aiop was only reachable from the TSO + * itself, and thus it is now no longer be reachable at all. + */ + IF_NONMOVING_WRITE_BARRIER_ENABLED { + updateRemembSetPushClosure(iomgr->cap, (StgClosure *)aiop); + } } @@ -251,7 +266,6 @@ static void notifyIOCompletion(CapIOManager *iomgr, StgAsyncIOOp *aiop) raiseAsync(iomgr->cap, tso, (StgClosure *)blockedOnBadFD_closure, false, NULL); - break; } else { /* We should be guaranteed that the tso is still on the same * cap because the tso was not on the run queue of any cap and @@ -262,6 +276,12 @@ static void notifyIOCompletion(CapIOManager *iomgr, StgAsyncIOOp *aiop) tso->_link = END_TSO_QUEUE; pushOnRunQueue(iomgr->cap, tso); } + /* For the TSO case, the aiop was only reachable from the TSO + * itself, and thus it is now no longer be reachable at all. + */ + IF_NONMOVING_WRITE_BARRIER_ENABLED { + updateRemembSetPushClosure(iomgr->cap, (StgClosure *)aiop); + } break; } case NotifyMVar: @@ -455,6 +475,9 @@ void awaitCompletedTimeoutsOrIOPoll(CapIOManager *iomgr) int ncompletions = res; ASSERT(ncompletions <= (int)nfds); 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. } else if (errno == EINTR) { /* We got interrupted by a signal. In the non-threaded RTS, if the ===================================== rts/posix/Poll.h ===================================== @@ -17,6 +17,7 @@ #if defined(IOMGR_ENABLED_POLL) void initCapabilityIOManagerPoll(CapIOManager *iomgr); +void freeCapabilityIOManagerPoll(CapIOManager *iomgr); /* Synchronous I/O and timer operations */ bool syncIOWaitReadyPoll(CapIOManager *iomgr, StgTSO *tso, ===================================== rts/posix/Signals.c ===================================== @@ -9,21 +9,13 @@ #include "rts/PosixSource.h" #include "Rts.h" -#include "Schedule.h" #include "RtsSignals.h" -#include "Signals.h" -#include "IOManager.h" #include "RtsUtils.h" +#include "Schedule.h" #include "Prelude.h" -#include "Ticker.h" #include "ThreadLabels.h" #include "Libdw.h" - -/* TODO: eliminate this include. This file should be about signals, not be - * part of an I/O manager implementation. The code here that are really part - * of an I/O manager should be moved into an appropriate I/O manager impl. - */ -#include "IOManagerInternals.h" +#include "MIO.h" #if defined(alpha_HOST_ARCH) # if defined(linux_HOST_OS) @@ -41,19 +33,10 @@ # include #endif -#if defined(HAVE_ERRNO_H) -# include -#endif - -#if defined(HAVE_EVENTFD_H) -# include -#endif - #if defined(HAVE_TERMIOS_H) #include #endif -#include #include /* This curious flag is provided for the benefit of the Haskell binding @@ -134,110 +117,6 @@ more_handlers(int sig) nHandlers = sig + 1; } -// Here's the pipe into which we will send our signals -static int io_manager_wakeup_fd = -1; -static int timer_manager_control_wr_fd = -1; - -#define IO_MANAGER_WAKEUP 0xff -#define IO_MANAGER_DIE 0xfe -#define IO_MANAGER_SYNC 0xfd - -void setTimerManagerControlFd(int fd) { - RELAXED_STORE(&timer_manager_control_wr_fd, fd); -} - -void -setIOManagerWakeupFd (int fd) -{ - // only called when THREADED_RTS, but unconditionally - // compiled here because GHC.Event.Control depends on it. - SEQ_CST_STORE(&io_manager_wakeup_fd, fd); -} - -/* ----------------------------------------------------------------------------- - * Wake up at least one IO or timer manager HS thread. - * -------------------------------------------------------------------------- */ -void -ioManagerWakeup (void) -{ - int r; - const int wakeup_fd = SEQ_CST_LOAD(&io_manager_wakeup_fd); - // Wake up the IO Manager thread by sending a byte down its pipe - if (wakeup_fd >= 0) { -#if defined(HAVE_EVENTFD) - StgWord64 n = (StgWord64)IO_MANAGER_WAKEUP; - r = write(wakeup_fd, (char *) &n, 8); -#else - StgWord8 byte = (StgWord8)IO_MANAGER_WAKEUP; - r = write(wakeup_fd, &byte, 1); -#endif - /* N.B. If the TimerManager is shutting down as we run this - * then there is a possibility that our first read of - * io_manager_wakeup_fd is non-negative, but before we get to the - * write the file is closed. If this occurs, io_manager_wakeup_fd - * will be written into with -1 (GHC.Event.Control does this prior - * to closing), so checking this allows us to distinguish this case. - * To ensure we observe the correct ordering, we declare the - * io_manager_wakeup_fd as volatile. - * Since this is not an error condition, we do not print the error - * message in this case. - */ - if (r == -1 && SEQ_CST_LOAD(&io_manager_wakeup_fd) >= 0) { - sysErrorBelch("ioManagerWakeup: write"); - } - } -} - -#if defined(THREADED_RTS) -void -ioManagerDie (void) -{ - StgWord8 byte = (StgWord8)IO_MANAGER_DIE; - uint32_t i; - int r; - - { - // Shut down timer manager - const int fd = RELAXED_LOAD(&timer_manager_control_wr_fd); - if (0 <= fd) { - r = write(fd, &byte, 1); - if (r == -1) { sysErrorBelch("ioManagerDie: write"); } - RELAXED_STORE(&timer_manager_control_wr_fd, -1); - } - } - - { - // Shut down IO managers - for (i=0; i < getNumCapabilities(); i++) { - const int fd = RELAXED_LOAD(&getCapability(i)->iomgr->control_fd); - if (0 <= fd) { - r = write(fd, &byte, 1); - if (r == -1) { sysErrorBelch("ioManagerDie: write"); } - RELAXED_STORE(&getCapability(i)->iomgr->control_fd, -1); - } - } - } -} - -void -ioManagerStartCap (Capability **cap) -{ - rts_evalIO(cap,ensureIOManagerIsRunning_closure,NULL); -} - -void -ioManagerStart (void) -{ - // Make sure the IO manager thread is running - Capability *cap; - if (SEQ_CST_LOAD(&timer_manager_control_wr_fd) < 0 || SEQ_CST_LOAD(&io_manager_wakeup_fd) < 0) { - cap = rts_lock(); - ioManagerStartCap(&cap); - rts_unlock(cap); - } -} -#endif - #if !defined(THREADED_RTS) #define N_PENDING_HANDLERS 16 @@ -260,31 +139,9 @@ generic_handler(int sig USED_IF_THREADS, void *p STG_UNUSED) { #if defined(THREADED_RTS) - - StgWord8 buf[sizeof(siginfo_t) + 1]; - int r; - - buf[0] = sig; - if (info == NULL) { - // info may be NULL on Solaris (see #3790) - memset(buf+1, 0, sizeof(siginfo_t)); - } else { - memcpy(buf+1, info, sizeof(siginfo_t)); - } - - int timer_control_fd = RELAXED_LOAD(&timer_manager_control_wr_fd); - if (0 <= timer_control_fd) - { - r = write(timer_control_fd, buf, sizeof(siginfo_t)+1); - if (r == -1 && errno == EAGAIN) { - errorBelch("lost signal due to full pipe: %d\n", sig); - } - } - - // If the IO manager hasn't told us what the FD of the write end - // of its pipe is, there's not much we can do here, so just ignore - // the signal.. - + //TODO: This calls MIO directly. We should go via IOManager API. + // The IOManager API should be extended to cover signals. + timerManagerNotifySignal(sig, info); #else /* not THREADED_RTS */ /* Can't call allocate from here. Probably can't call malloc ===================================== rts/posix/Signals.h ===================================== @@ -25,18 +25,6 @@ extern siginfo_t *next_pending_handler; void startSignalHandlers(Capability *cap); #endif -/* Communicating with the IO manager thread (see GHC.Conc). - * - * TODO: these I/O manager things are not related to signals and ought to live - * elsewhere, e.g. in a module specifically for the I/O manager. - */ -void ioManagerWakeup (void); -#if defined(THREADED_RTS) -void ioManagerDie (void); -void ioManagerStart (void); -void ioManagerStartCap (/* inout */ Capability **cap); -#endif - extern StgInt *signal_handlers; #include "EndPrivate.h" ===================================== rts/rts.cabal ===================================== @@ -583,6 +583,7 @@ library posix/OSMem.c posix/OSThreads.c posix/FdWakeup.c + posix/MIO.c posix/Poll.c posix/Select.c posix/Signals.c ===================================== rts/win32/AsyncMIO.h ===================================== @@ -12,6 +12,8 @@ #include "Rts.h" +#include "BeginPrivate.h" + extern unsigned int addIORequest(int fd, bool forWriting, @@ -28,3 +30,5 @@ extern int awaitRequests(bool wait); extern void abandonRequestWait(void); extern void resetAbandonRequestWait(void); extern HsInt rts_EINTR(void); + +#include "EndPrivate.h" ===================================== rts/win32/AsyncWinIO.h ===================================== @@ -14,12 +14,16 @@ #include #include -extern bool startupAsyncWinIO(void); -extern void shutdownAsyncWinIO(bool wait_threads); -extern void awaitAsyncRequests(bool wait); extern void registerIOCPHandle (HANDLE port); extern void registerAlertableWait (bool has_timeout, DWORD mssec); - extern OVERLAPPED_ENTRY* getOverlappedEntries (uint32_t *num); extern void completeSynchronousRequest (void); + +#include "BeginPrivate.h" + +extern bool startupAsyncWinIO(void); +extern void shutdownAsyncWinIO(bool wait_threads); +extern void awaitAsyncRequests(bool wait); extern bool queueIOThread(void); + +#include "EndPrivate.h" ===================================== rts/win32/AwaitEvent.h ===================================== @@ -1,4 +1,7 @@ #pragma once +#include "BeginPrivate.h" + void awaitCompletedTimeoutsOrIOWin32(Capability *cap, bool wait); +#include "EndPrivate.h" ===================================== rts/win32/ConsoleHandler.h ===================================== @@ -7,6 +7,8 @@ #pragma once +#include "BeginPrivate.h" + /* * Console control handlers lets an application handle Ctrl+C, Ctrl+Break etc. * in Haskell under Win32. Akin to the Unix signal SIGINT. @@ -69,3 +71,5 @@ extern int rts_waitConsoleHandlerCompletion(void); * Tear down and shut down user signal processing. */ extern void finiUserSignals(void); + +#include "EndPrivate.h" ===================================== rts/win32/MIOManager.h ===================================== @@ -12,6 +12,8 @@ #include +#include "BeginPrivate.h" + /* The IOManager subsystem provides a non-blocking view of I/O operations. It lets one (or more) OS thread(s) @@ -107,3 +109,5 @@ extern int AddProcRequest ( void* proc, extern void abandonWorkRequest ( int reqID ); extern void interruptIOManagerEvent ( void ); + +#include "EndPrivate.h" ===================================== rts/win32/ThrIOManager.h ===================================== @@ -12,7 +12,12 @@ /* Communicating with the IO manager thread (see GHC.Conc). */ +void ioManagerFinished (void); + +#include "BeginPrivate.h" + void ioManagerWakeup (void); void ioManagerDie (void); void ioManagerStart (void); -void ioManagerFinished (void); + +#include "EndPrivate.h" ===================================== rts/win32/WorkQueue.h ===================================== @@ -10,6 +10,8 @@ #include +#include "BeginPrivate.h" + /* This is a fixed-size queue. */ #define WORKQUEUE_SIZE 16 @@ -34,3 +36,5 @@ extern HANDLE GetWorkQueueHandle ( WorkQueue* pq ); extern BOOL GetWork ( WorkQueue* pq, void** ppw ); extern BOOL FetchWork ( WorkQueue* pq, void** ppw ); extern int SubmitWork ( WorkQueue* pq, void* pw ); + +#include "EndPrivate.h" ===================================== rts/win32/veh_excn.h ===================================== @@ -59,6 +59,8 @@ // simplified quite a bit. // Using VEH also means we don't have to worry about the dynamic code generated by GHCi. +#include "BeginPrivate.h" + // Prototype of the VEH callback function. // See https://msdn.microsoft.com/en-us/library/windows/desktop/ms681419(v=vs.85).a... // @@ -72,3 +74,5 @@ void __unregister_hs_exception_handler( void ); // prototypes for dump methods. void generateDump(EXCEPTION_POINTERS* pExceptionPointers); void generateStack (EXCEPTION_POINTERS* pExceptionPointers); + +#include "EndPrivate.h" View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/268152b0a44214325ee6b243d302d6e... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/268152b0a44214325ee6b243d302d6e... You're receiving this email because of your account on gitlab.haskell.org.