Merge pull request #221 from Gabriel439/gabriel/restore_2

Correctly unmask exceptions when running subprocesses

src/Turtle/Bytes.hs

@@ -349,7 +349,9 @@ system p s = liftIO (do
                 feedIn restore =
                     restore (ignoreSIGPIPE (outhandle hIn s))
                     `Exception.finally` close hIn
-            Exception.mask_ (Async.withAsyncWithUnmask feedIn (\a -> Process.waitForProcess ph <* halt a) )
+            Exception.mask (\restore ->
+                Async.withAsync (feedIn restore) (\a ->
+                    restore (Process.waitForProcess ph) <* halt a ) )
         handle (Nothing , ph) = do
             Process.waitForProcess ph
 
@@ -390,7 +392,9 @@ systemStrict p s = liftIO (do
                 `Exception.finally` close hIn
 
         Async.concurrently
-            (Exception.mask_ (Async.withAsyncWithUnmask feedIn (\a -> liftIO (Process.waitForProcess ph) <* halt a)))
+            (Exception.mask (\restore ->
+                Async.withAsync (feedIn restore) (\a ->
+                    restore (Process.waitForProcess ph) <* halt a ) ))
             (Data.ByteString.hGetContents hOut) ) )
 
 systemStrictWithErr
@@ -428,7 +432,9 @@ systemStrictWithErr p s = liftIO (do
                 `Exception.finally` close hIn
 
         runConcurrently $ (,,)
-            <$> Concurrently (Exception.mask_ (Async.withAsyncWithUnmask feedIn (\a -> liftIO (Process.waitForProcess ph) <* halt a)))
+            <$> Concurrently (Exception.mask (\restore ->
+                    Async.withAsync (feedIn restore) (\a ->
+                        restore (Process.waitForProcess ph) <* halt a ) ))
             <*> Concurrently (Data.ByteString.hGetContents hOut)
             <*> Concurrently (Data.ByteString.hGetContents hErr) ) )
 
@@ -499,7 +505,10 @@ stream p s = do
                 liftIO (Data.ByteString.hPut hIn bytes) ) )
             `Exception.finally` close hIn
 
-    a <- using (Managed.managed (Exception.mask_ . Async.withAsyncWithUnmask feedIn))
+    a <- using
+        (Managed.managed (\k ->
+            Exception.mask (\restore ->
+                Async.withAsync (feedIn restore) k ) ))
     inhandle hOut <|> (liftIO (Process.waitForProcess ph *> halt a) *> empty)
 
 streamWithErr
@@ -563,9 +572,18 @@ streamWithErr p s = do
             x1 <- loop x0 (0 :: Int)
             done x1 )
 
-    a <- using (Managed.managed (Exception.mask_ . Async.withAsyncWithUnmask feedIn    ))
-    b <- using (Managed.managed (Exception.mask_ . Async.withAsyncWithUnmask forwardOut))
-    c <- using (Managed.managed (Exception.mask_ . Async.withAsyncWithUnmask forwardErr))
+    a <- using
+        (Managed.managed (\k ->
+            Exception.mask (\restore ->
+                Async.withAsync (feedIn restore) k ) ))
+    b <- using
+        (Managed.managed (\k ->
+            Exception.mask (\restore ->
+                Async.withAsync (forwardOut restore) k ) ))
+    c <- using
+        (Managed.managed (\k ->
+            Exception.mask (\restore ->
+                Async.withAsync (forwardErr restore) k ) ))
     let l `also` r = do
             _ <- l <|> (r *> STM.retry)
             _ <- r

src/Turtle/Prelude.hs

@@ -260,13 +260,13 @@ module Turtle.Prelude (
 import Control.Applicative
 import Control.Concurrent (threadDelay)
 import Control.Concurrent.Async
-    (Async, withAsync, withAsyncWithUnmask, waitSTM, concurrently,
+    (Async, withAsync, waitSTM, concurrently,
      Concurrently(..))
 import qualified Control.Concurrent.Async
 import Control.Concurrent.MVar (newMVar, modifyMVar_)
 import qualified Control.Concurrent.STM as STM
 import qualified Control.Concurrent.STM.TQueue as TQueue
-import Control.Exception (Exception, bracket, bracket_, finally, mask_, throwIO)
+import Control.Exception (Exception, bracket, bracket_, finally, mask, throwIO)
 import Control.Foldl (Fold, FoldM(..), genericLength, handles, list, premap)
 import qualified Control.Foldl
 import qualified Control.Foldl.Text
@@ -536,7 +536,9 @@ system p s = liftIO (do
             let feedIn :: (forall a. IO a -> IO a) -> IO ()
                 feedIn restore =
                     restore (ignoreSIGPIPE (outhandle hIn s)) `finally` close hIn
-            mask_ (withAsyncWithUnmask feedIn (\a -> Process.waitForProcess ph <* halt a) )
+            mask (\restore ->
+                withAsync (feedIn restore) (\a ->
+                    restore (Process.waitForProcess ph) `finally` halt a) )
         handle (Nothing , ph) = do
             Process.waitForProcess ph
 
@@ -575,7 +577,9 @@ systemStrict p s = liftIO (do
                 restore (ignoreSIGPIPE (outhandle hIn s)) `finally` close hIn
 
         concurrently
-            (mask_ (withAsyncWithUnmask feedIn (\a -> liftIO (Process.waitForProcess ph) <* halt a)))
+            (mask (\restore ->
+                withAsync (feedIn restore) (\a ->
+                    restore (liftIO (Process.waitForProcess ph)) `finally` halt a ) ))
             (Text.hGetContents hOut) ) )
 
 systemStrictWithErr
@@ -611,7 +615,9 @@ systemStrictWithErr p s = liftIO (do
                 restore (ignoreSIGPIPE (outhandle hIn s)) `finally` close hIn
 
         runConcurrently $ (,,)
-            <$> Concurrently (mask_ (withAsyncWithUnmask feedIn (\a -> liftIO (Process.waitForProcess ph) <* halt a)))
+            <$> Concurrently (mask (\restore ->
+                    withAsync (feedIn restore) (\a ->
+                        restore (liftIO (Process.waitForProcess ph)) `finally` halt a ) ))
             <*> Concurrently (Text.hGetContents hOut)
             <*> Concurrently (Text.hGetContents hErr) ) )
 
@@ -676,7 +682,9 @@ stream p s = do
     let feedIn :: (forall a. IO a -> IO a) -> IO ()
         feedIn restore = restore (outhandle hIn s) `finally` close hIn
 
-    a <- using (managed (mask_ . withAsyncWithUnmask feedIn))
+    a <- using
+        (managed (\k ->
+            mask (\restore -> withAsync (feedIn restore) (restore . k))))
     inhandle hOut <|> (liftIO (Process.waitForProcess ph *> halt a) *> empty)
 
 streamWithErr
@@ -736,9 +744,15 @@ streamWithErr p s = do
             x1 <- loop x0 (0 :: Int)
             done x1 )
 
-    a <- using (managed (mask_ . withAsyncWithUnmask feedIn    ))
-    b <- using (managed (mask_ . withAsyncWithUnmask forwardOut))
-    c <- using (managed (mask_ . withAsyncWithUnmask forwardErr))
+    a <- using
+        (managed (\k ->
+            mask (\restore -> withAsync (feedIn restore) (restore . k)) ))
+    b <- using
+        (managed (\k ->
+            mask (\restore -> withAsync (forwardOut restore) (restore . k)) ))
+    c <- using
+        (managed (\k ->
+            mask (\restore -> withAsync (forwardErr restore) (restore . k)) ))
     let l `also` r = do
             _ <- l <|> (r *> STM.retry)
             _ <- r

test/RegressionMaskingException.hs

@@ -0,0 +1,12 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+import Turtle
+
+import qualified System.Timeout
+
+-- This test fails by hanging
+main :: IO ()
+main = runManaged (do
+    _ <- fork (shells "while true; do sleep 1; done" empty)
+    sleep 1
+    return () )

turtle.cabal

@@ -105,7 +105,17 @@ test-suite regression-broken-pipe
     GHC-Options: -Wall -threaded
     Default-Language: Haskell2010
     Build-Depends:
-        base    >= 4   && < 5   ,
+        base   >= 4 && < 5,
+        turtle
+
+test-suite regression-masking-exception
+    Type: exitcode-stdio-1.0
+    HS-Source-Dirs: test
+    Main-Is: RegressionMaskingException.hs
+    GHC-Options: -Wall -threaded
+    Default-Language: Haskell2010
+    Build-Depends:
+        base   >= 4 && < 5,
         turtle
 
 benchmark bench