git.iko.soy/distributed-process
on master help sign in
Have spawnLocal inherit the masking state of the caller.
f7a8d77Facundo Domínguez
src/Control/Distributed/Process/Node.hs+29−6
compare with master
unifiedsplit
@@ -5,6 +5,7 @@
5{-# LANGUAGE BangPatterns #-}
5{-# LANGUAGE BangPatterns #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving #-}
7{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE RecordWildCards #-}
8+{-# LANGUAGE MagicHash #-}
8
9
9-- | Local nodes
10-- | Local nodes
10--
11--
@@ -60,6 +61,8 @@ import Control.Exception
60 , Exception
61 , Exception
61 , throwTo
62 , throwTo
62 , uninterruptibleMask_
63 , uninterruptibleMask_
64+ , getMaskingState
65+ , MaskingState(..)
63 )
66 )
64import qualified Control.Exception as Exception
67import qualified Control.Exception as Exception
65 ( Handler(..)
68 ( Handler(..)
@@ -206,9 +209,20 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a
206 , mapDefault
209 , mapDefault
207 )
210 )
208import Control.Monad.Catch (try)
211import Control.Monad.Catch (try)
212+import GHC.IO (IO(..), unsafeUnmask)
213+import GHC.Base ( maskAsyncExceptions# )
214+
209import Unsafe.Coerce
215import Unsafe.Coerce
210import Prelude
216import Prelude
211
217
218+-- Remove these definitions when the fix for
219+-- https://ghc.haskell.org/trac/ghc/ticket/10149
220+-- is included in all supported compilers:
221+block :: IO a -> IO a
222+block (IO io) = IO $ maskAsyncExceptions# io
223+unblock :: IO a -> IO a
224+unblock = unsafeUnmask
225+
212--------------------------------------------------------------------------------
226--------------------------------------------------------------------------------
213-- Initialization --
227-- Initialization --
214--------------------------------------------------------------------------------
228--------------------------------------------------------------------------------
@@ -346,11 +360,14 @@ runProcess node proc = do
346
360
347-- | Spawn a new process on a local node
361-- | Spawn a new process on a local node
348forkProcess :: LocalNode -> Process () -> IO ProcessId
362forkProcess :: LocalNode -> Process () -> IO ProcessId
349−forkProcess node proc =
363+forkProcess node proc = do
364+ ms <- getMaskingState
350− modifyMVarMasked (localState node) startProcess
365+ modifyMVarMasked (localState node) (startProcess ms)
351 where
366 where
367+ startProcess :: MaskingState
368+ -> LocalNodeState
352− startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
369+ -> IO (LocalNodeState, ProcessId)
353− startProcess (LocalNodeValid vst) = do
370+ startProcess ms (LocalNodeValid vst) = do
354 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
371 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
355 , lpidUnique = vst ^. localPidUnique
372 , lpidUnique = vst ^. localPidUnique
356 }
373 }
@@ -372,7 +389,13 @@ forkProcess node proc =
372 , processThread = tid
389 , processThread = tid
373 , processNode = node
390 , processNode = node
374 }
391 }
392+ -- Rewrite this code when this is fixed:
393+ -- https://ghc.haskell.org/trac/ghc/ticket/10149
394+ let unmask = case ms of
395+ Unmasked -> unblock
396+ MaskedInterruptible -> block
397+ MaskedUninterruptible -> id
375− tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do
398+ tid' <- uninterruptibleMask_ $ forkIO $ do
376 reason <- Exception.catches
399 reason <- Exception.catches
377 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
400 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
378 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
401 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
@@ -418,7 +441,7 @@ forkProcess node proc =
418 $ vst
441 $ vst
419 , pid
442 , pid
420 )
443 )
421− startProcess LocalNodeClosed =
444+ startProcess _ LocalNodeClosed =
422 throwIO $ NodeClosedException $ localNodeId node
445 throwIO $ NodeClosedException $ localNodeId node
423
446
424 cleanupProcess :: ProcessId
447 cleanupProcess :: ProcessId