rts/base: Fix #9423

Summary:
Fix #9423.

The problem in #9423 is caused when code invoked by `hs_exit()` waits
on all foreign calls to return, but some IO managers are in `safe` foreign
calls and do not return. The previous design signaled to the timer manager
(via its control pipe) that it should "die" and when the timer manager
returned to Haskell-land, the Haskell code in timer manager then signalled
to the IO manager threads that they should return from foreign calls and
`die`. Unfortunately, in the shutdown sequence the timer manager is unable
to return to Haskell-land fast enough and so the code that signals to the
IO manager threads (via their control pipes) is never executed and the IO
manager threads remain out in the foreign calls.

This patch solves this problem by having the RTS signal to all the IO
manager threads (via their control pipes; and in addition to signalling
to the timer manager thread) that they should shutdown (in `ioManagerDie()`
in `rts/Signals.c`. To do this, we arrange for each IO manager thread to
register its control pipe with the RTS (in `GHC.Thread.startIOManagerThread`).
In addition, `GHC.Thread.startTimerManagerThread` registers its control pipe.
These are registered via C functions `setTimerManagerControlFd` (in
`rts/Signals.c`) and `setIOManagerControlFd` (in `rts/Capability.c`). The IO
manager control pipe file descriptors are stored in a new field of the
`Capability_ struct`.

Test Plan: See the notes on #9423 to recreate the problem and to verify that it no longer occurs with the fix.

Auditors: simonmar

Reviewers: simonmar, edsko, ezyang, austin

Reviewed By: austin

Subscribers: phaskell, simonmar, ezyang, carter, relrod

Differential Revision: https://phabricator.haskell.org/D129

GHC Trac Issues: #9423, #9284

9 files changed, 92 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 d55d5b1193..80c05f7c0c 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 f581330e25..63a72ef80b 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..4543ec7d1a 100644
--- a/rts/Capability.c
+++ b/rts/Capability.c
@@ -255,6 +255,7 @@ initCapability( Capability *cap, nat i )
     cap->spark_stats.converted  = 0;
     cap->spark_stats.gcd        = 0;
     cap->spark_stats.fizzled    = 0;
+    cap->io_manager_control_wr_fd = -1;
 #endif
     cap->total_allocated        = 0;
 
@@ -1076,6 +1077,16 @@ rtsBool checkSparkCountInvariant (void)
 }
 #endif
 
+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
+}
+
 // Local Variables:
 // mode: C
 // fill-column: 80
diff --git a/rts/Capability.h b/rts/Capability.h
index c7dceefe9f..d5f36c54bc 100644
--- a/rts/Capability.h
+++ b/rts/Capability.h
@@ -126,6 +126,9 @@ struct Capability_ {
 
     // Stats on spark creation/conversion
     SparkCounters spark_stats;
+
+    // IO manager for this cap
+    int io_manager_control_wr_fd;
 #endif
     // Total words allocated by this cap since rts start
     W_ total_allocated;
diff --git a/rts/Linker.c b/rts/Linker.c
index 0a81b83d80..1f716e3a6d 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..