diff options
| -rw-r--r-- | includes/rts/IOManager.h | 3 | ||||
| -rw-r--r-- | libraries/base/GHC/Event/Control.hs | 8 | ||||
| -rw-r--r-- | libraries/base/GHC/Event/Manager.hs | 1 | ||||
| -rw-r--r-- | libraries/base/GHC/Event/Thread.hs | 35 | ||||
| -rw-r--r-- | libraries/base/GHC/Event/TimerManager.hs | 1 | ||||
| -rw-r--r-- | rts/Capability.c | 19 | ||||
| -rw-r--r-- | rts/Capability.h | 4 | ||||
| -rw-r--r-- | rts/Linker.c | 1 | ||||
| -rw-r--r-- | rts/posix/Signals.c | 80 |
9 files changed, 101 insertions, 51 deletions
diff --git a/includes/rts/IOManager.h b/includes/rts/IOManager.h index 1c331b97af..7bf2cdf43c 100644 --- a/includes/rts/IOManager.h +++ b/includes/rts/IOManager.h @@ -26,7 +26,8 @@ void sendIOManagerEvent (HsWord32 event); #else -void setIOManagerControlFd (int fd); +void setIOManagerControlFd (nat cap_no, int fd); +void setTimerManagerControlFd(int fd); void setIOManagerWakeupFd (int fd); #endif diff --git a/libraries/base/GHC/Event/Control.hs b/libraries/base/GHC/Event/Control.hs index 2951a6a681..53a9bc86d8 100644 --- a/libraries/base/GHC/Event/Control.hs +++ b/libraries/base/GHC/Event/Control.hs @@ -17,6 +17,7 @@ module GHC.Event.Control , readControlMessage -- *** File descriptors , controlReadFd + , controlWriteFd , wakeupReadFd -- ** Control message sending , sendWakeup @@ -91,7 +92,6 @@ newControl shouldRegister = allocaArray 2 $ \fds -> do setCloseOnExec wr return (rd, wr) (ctrl_rd, ctrl_wr) <- createPipe - when shouldRegister $ c_setIOManagerControlFd ctrl_wr #if defined(HAVE_EVENTFD) ev <- throwErrnoIfMinus1 "eventfd" $ c_eventfd 0 0 setNonBlockingFD ev True @@ -200,9 +200,5 @@ foreign import ccall unsafe "sys/eventfd.h eventfd_write" c_eventfd_write :: CInt -> CULLong -> IO CInt #endif --- Used to tell the RTS how it can send messages to the I/O manager. -foreign import ccall "setIOManagerControlFd" - c_setIOManagerControlFd :: CInt -> IO () - -foreign import ccall "setIOManagerWakeupFd" +foreign import ccall unsafe "setIOManagerWakeupFd" c_setIOManagerWakeupFd :: CInt -> IO () diff --git a/libraries/base/GHC/Event/Manager.hs b/libraries/base/GHC/Event/Manager.hs index 1dbe036e0e..f07efba311 100644 --- a/libraries/base/GHC/Event/Manager.hs +++ b/libraries/base/GHC/Event/Manager.hs @@ -27,6 +27,7 @@ module GHC.Event.Manager -- * State , callbackTableVar + , emControl -- * Registering interest in I/O events , Event diff --git a/libraries/base/GHC/Event/Thread.hs b/libraries/base/GHC/Event/Thread.hs index dcfa32aa28..0a82a548da 100644 --- a/libraries/base/GHC/Event/Thread.hs +++ b/libraries/base/GHC/Event/Thread.hs @@ -22,6 +22,7 @@ import Data.List (zipWith3) import Data.Maybe (Maybe(..)) import Data.Tuple (snd) import Foreign.C.Error (eBADF, errnoToIOError) +import Foreign.C.Types (CInt(..), CUInt(..)) import Foreign.Ptr (Ptr) import GHC.Base import GHC.Conc.Sync (TVar, ThreadId, ThreadStatus(..), atomically, forkIO, @@ -33,12 +34,14 @@ import GHC.IO.Exception (ioError) import GHC.IOArray (IOArray, newIOArray, readIOArray, writeIOArray, boundsIOArray) import GHC.MVar (MVar, newEmptyMVar, newMVar, putMVar, takeMVar) +import GHC.Event.Control (controlWriteFd) import GHC.Event.Internal (eventIs, evtClose) import GHC.Event.Manager (Event, EventManager, evtRead, evtWrite, loop, new, registerFd, unregisterFd_) import qualified GHC.Event.Manager as M import qualified GHC.Event.TimerManager as TM import GHC.Num ((-), (+)) +import GHC.Real (fromIntegral) import GHC.Show (showSignedInt) import System.IO.Unsafe (unsafePerformIO) import System.Posix.Types (Fd) @@ -261,7 +264,11 @@ startIOManagerThread :: IOArray Int (Maybe (ThreadId, EventManager)) startIOManagerThread eventManagerArray i = do let create = do !mgr <- new True - !t <- forkOn i $ loop mgr + !t <- forkOn i $ do + c_setIOManagerControlFd + (fromIntegral i) + (fromIntegral $ controlWriteFd $ M.emControl mgr) + loop mgr labelThread t ("IOManager on cap " ++ show_int i) writeIOArray eventManagerArray i (Just (t,mgr)) old <- readIOArray eventManagerArray i @@ -277,6 +284,7 @@ startIOManagerThread eventManagerArray i = do -- the fork, for example. In this case we should clean up -- open pipes and everything else related to the event manager. -- See #4449 + c_setIOManagerControlFd (fromIntegral i) (-1) M.cleanup em create _other -> return () @@ -285,8 +293,10 @@ startTimerManagerThread :: IO () startTimerManagerThread = modifyMVar_ timerManagerThreadVar $ \old -> do let create = do !mgr <- TM.new + c_setTimerManagerControlFd + (fromIntegral $ controlWriteFd $ TM.emControl mgr) writeIORef timerManager $ Just mgr - !t <- forkIO $ TM.loop mgr `finally` shutdownManagers + !t <- forkIO $ TM.loop mgr labelThread t "TimerManager" return $ Just t case old of @@ -304,21 +314,11 @@ startTimerManagerThread = modifyMVar_ timerManagerThreadVar $ \old -> do mem <- readIORef timerManager _ <- case mem of Nothing -> return () - Just em -> TM.cleanup em + Just em -> do c_setTimerManagerControlFd (-1) + TM.cleanup em create _other -> return st -shutdownManagers :: IO () -shutdownManagers = - withMVar ioManagerLock $ \_ -> do - eventManagerArray <- readIORef eventManager - let (_, high) = boundsIOArray eventManagerArray - forM_ [0..high] $ \i -> do - mmgr <- readIOArray eventManagerArray i - case mmgr of - Nothing -> return () - Just (_,mgr) -> M.shutdown mgr - foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool ioManagerCapabilitiesChanged :: IO () @@ -352,3 +352,10 @@ ioManagerCapabilitiesChanged = do Just (_,mgr) <- readIOArray eventManagerArray i tid <- restartPollLoop mgr i writeIOArray eventManagerArray i (Just (tid,mgr)) + +-- Used to tell the RTS how it can send messages to the I/O manager. +foreign import ccall unsafe "setIOManagerControlFd" + c_setIOManagerControlFd :: CUInt -> CInt -> IO () + +foreign import ccall unsafe "setTimerManagerControlFd" + c_setTimerManagerControlFd :: CInt -> IO () diff --git a/libraries/base/GHC/Event/TimerManager.hs b/libraries/base/GHC/Event/TimerManager.hs index 7ba2aea8ff..cbfce59434 100644 --- a/libraries/base/GHC/Event/TimerManager.hs +++ b/libraries/base/GHC/Event/TimerManager.hs @@ -15,6 +15,7 @@ module GHC.Event.TimerManager , new , newWith , newDefaultBackend + , emControl -- * Running , finished diff --git a/rts/Capability.c b/rts/Capability.c index 29c5270416..a954006257 100644 --- a/rts/Capability.c +++ b/rts/Capability.c @@ -27,6 +27,10 @@ #include "STM.h" #include "RtsUtils.h" +#if !defined(mingw32_HOST_OS) +#include "rts/IOManager.h" // for setIOManagerControlFd() +#endif + #include <string.h> // one global capability, this is the Capability for non-threaded @@ -255,6 +259,9 @@ initCapability( Capability *cap, nat i ) cap->spark_stats.converted = 0; cap->spark_stats.gcd = 0; cap->spark_stats.fizzled = 0; +#if !defined(mingw32_HOST_OS) + cap->io_manager_control_wr_fd = -1; +#endif #endif cap->total_allocated = 0; @@ -1076,6 +1083,18 @@ rtsBool checkSparkCountInvariant (void) } #endif +#if !defined(mingw32_HOST_OS) +void setIOManagerControlFd(nat cap_no USED_IF_THREADS, int fd USED_IF_THREADS) { +#if defined(THREADED_RTS) + if (cap_no < n_capabilities) { + capabilities[cap_no]->io_manager_control_wr_fd = fd; + } else { + errorBelch("warning: setIOManagerControlFd called with illegal capability number."); + } +#endif +} +#endif + // Local Variables: // mode: C // fill-column: 80 diff --git a/rts/Capability.h b/rts/Capability.h index c7dceefe9f..9a6651e0e6 100644 --- a/rts/Capability.h +++ b/rts/Capability.h @@ -126,6 +126,10 @@ struct Capability_ { // Stats on spark creation/conversion SparkCounters spark_stats; +#if !defined(mingw32_HOST_OS) + // IO manager for this cap + int io_manager_control_wr_fd; +#endif #endif // Total words allocated by this cap since rts start W_ total_allocated; diff --git a/rts/Linker.c b/rts/Linker.c index b24be580b4..dba346eb86 100644 --- a/rts/Linker.c +++ b/rts/Linker.c @@ -860,6 +860,7 @@ typedef struct _RtsSymbolVal { #if !defined(mingw32_HOST_OS) #define RTS_USER_SIGNALS_SYMBOLS \ SymI_HasProto(setIOManagerControlFd) \ + SymI_HasProto(setTimerManagerControlFd) \ SymI_HasProto(setIOManagerWakeupFd) \ SymI_HasProto(ioManagerWakeup) \ SymI_HasProto(blockUserSignals) \ diff --git a/rts/posix/Signals.c b/rts/posix/Signals.c index d5129f0996..ba4a8b75ea 100644 --- a/rts/posix/Signals.c +++ b/rts/posix/Signals.c @@ -127,28 +127,27 @@ more_handlers(int sig) // Here's the pipe into which we will send our signals static int io_manager_wakeup_fd = -1; -static int io_manager_control_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 -setIOManagerWakeupFd (int fd) -{ - // only called when THREADED_RTS, but unconditionally - // compiled here because GHC.Event.Control depends on it. - io_manager_wakeup_fd = fd; +void setTimerManagerControlFd(int fd) { + timer_manager_control_wr_fd = fd; } void -setIOManagerControlFd (int fd) +setIOManagerWakeupFd (int fd) { // only called when THREADED_RTS, but unconditionally // compiled here because GHC.Event.Control depends on it. - io_manager_control_fd = fd; + io_manager_wakeup_fd = fd; } +/* ----------------------------------------------------------------------------- + * Wake up at least one IO or timer manager HS thread. + * -------------------------------------------------------------------------- */ void ioManagerWakeup (void) { @@ -170,14 +169,24 @@ ioManagerWakeup (void) void ioManagerDie (void) { + StgWord8 byte = (StgWord8)IO_MANAGER_DIE; + nat i; + int fd; int r; - // Ask the IO Manager thread to exit - if (io_manager_control_fd >= 0) { - StgWord8 byte = (StgWord8)IO_MANAGER_DIE; - r = write(io_manager_control_fd, &byte, 1); + + if (0 <= timer_manager_control_wr_fd) { + r = write(timer_manager_control_wr_fd, &byte, 1); if (r == -1) { sysErrorBelch("ioManagerDie: write"); } - io_manager_control_fd = -1; - io_manager_wakeup_fd = -1; + timer_manager_control_wr_fd = -1; + } + + for (i=0; i < n_capabilities; i++) { + fd = capabilities[i]->io_manager_control_wr_fd; + if (0 <= fd) { + r = write(fd, &byte, 1); + if (r == -1) { sysErrorBelch("ioManagerDie: write"); } + capabilities[i]->io_manager_control_wr_fd = -1; + } } } @@ -192,7 +201,7 @@ ioManagerStart (void) { // Make sure the IO manager thread is running Capability *cap; - if (io_manager_control_fd < 0 || io_manager_wakeup_fd < 0) { + if (timer_manager_control_wr_fd < 0 || io_manager_wakeup_fd < 0) { cap = rts_lock(); ioManagerStartCap(&cap); rts_unlock(cap); @@ -223,26 +232,37 @@ generic_handler(int sig USED_IF_THREADS, { #if defined(THREADED_RTS) - if (io_manager_control_fd != -1) - { - StgWord8 buf[sizeof(siginfo_t) + 1]; - int r; + StgWord8 buf[sizeof(siginfo_t) + 1]; + int r; - buf[0] = sig; + 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)); + } - 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)); + if (0 <= timer_manager_control_wr_fd) + { + r = write(timer_manager_control_wr_fd, buf, sizeof(siginfo_t)+1); + if (r == -1 && errno == EAGAIN) { + errorBelch("lost signal due to full pipe: %d\n", sig); } + } - r = write(io_manager_control_fd, buf, sizeof(siginfo_t)+1); - if (r == -1 && errno == EAGAIN) - { - errorBelch("lost signal due to full pipe: %d\n", sig); + nat i; + int fd; + for (i=0; i < n_capabilities; i++) { + fd = capabilities[i]->io_manager_control_wr_fd; + if (0 <= fd) { + r = write(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.. |