git.iko.soy/distributed-process
on master help sign in
Don't hold the localState MVar while calling potentially blocking operations.

The node controller could be blocked if the MVar is not promptly refilled.

commit04ba68213e6e1cc314e19bbecbf48f8daf4f48a9 authorFacundo Domínguez <facundo.dominguez@tweag.io> dateWed 15 Jul 2015, 10:28 (11 years ago) parent5336c01
3 files changed+45−16
compare with master
unifiedsplit
src/Control/Distributed/Process/Internal/Messaging.hs+17−9 src/Control/Distributed/Process/Internal/Types.hs+12 src/Control/Distributed/Process/Node.hs+16−7
src/Control/Distributed/Process/Internal/Messaging.hs+17−9
@@ -15,7 +15,9 @@ import qualified Data.ByteString.Lazy as BSL (toChunks)
1515import qualified Data.ByteString as BSS (ByteString)
1616import Control.Distributed.Process.Serializable ()
1717
18+import Control.Concurrent (forkIO)
1819import Control.Concurrent.Chan (writeChan)
20+import Control.Exception (mask_)
1921import Control.Monad (unless)
2022import Control.Monad.IO.Class (liftIO)
2123import Control.Monad.Reader (ask)
@@ -30,6 +32,7 @@ import qualified Network.Transport as NT
3032import Control.Distributed.Process.Internal.Types
3133 ( LocalNode(localState, localEndPoint, localCtrlChan)
3234 , withValidLocalState
35+ , modifyValidLocalState
3336 , modifyValidLocalState_
3437 , Identifier
3538 , localConnections
@@ -51,6 +54,7 @@ import Control.Distributed.Process.Internal.Types
5154 , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)
5255 )
5356import Control.Distributed.Process.Serializable (Serializable)
57+import Data.Foldable (forM_)
5458
5559--------------------------------------------------------------------------------
5660-- Message sending --
@@ -135,24 +139,28 @@ connBetween node from to implicitReconnect = do
135139 setupConnBetween node from to implicitReconnect
136140
137141disconnect :: LocalNode -> Identifier -> Identifier -> IO ()
138−disconnect node from to =
139− modifyValidLocalState_ node $ \vst ->
142+disconnect node from to = mask_ $ do
143+ mio <- modifyValidLocalState node $ \vst ->
140144 case vst ^. localConnectionBetween from to of
141145 Nothing ->
142− return vst
146+ return (vst, return ())
143147 Just (conn, _) -> do
144− NT.close conn
145− return $ localConnectionBetween from to ^= Nothing $ vst
148+ return ( localConnectionBetween from to ^= Nothing $ vst
149+ , NT.close conn
150+ )
151+ forM_ mio forkIO
146152
147153closeImplicitReconnections :: LocalNode -> Identifier -> IO ()
148−closeImplicitReconnections node to =
149− modifyValidLocalState_ node $ \vst -> do
154+closeImplicitReconnections node to = mask_ $ do
155+ mconns <- modifyValidLocalState node $ \vst -> do
150156 let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to'
151157 shouldClose _ _ = False
152158 let (affected, unaffected) =
153159 Map.partitionWithKey shouldClose (vst ^. localConnections)
154− mapM_ (NT.close . fst) (Map.elems affected)
155− return $ localConnections ^= unaffected $ vst
160+ return ( localConnections ^= unaffected $ vst
161+ , map fst $ Map.elems affected
162+ )
163+ forM_ mconns $ forkIO . mapM_ NT.close
156164
157165-- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node)
158166-- implies the death of @b@ (for instance, a process on that node)
src/Control/Distributed/Process/Internal/Types.hs+12
@@ -24,6 +24,7 @@ module Control.Distributed.Process.Internal.Types
2424 , LocalNodeState(..)
2525 , ValidLocalNodeState(..)
2626 , withValidLocalState
27+ , modifyValidLocalState
2728 , modifyValidLocalState_
2829 , Tracer(..)
2930 , MxEventBus(..)
@@ -128,6 +129,7 @@ import Control.Distributed.Process.Internal.CQueue (CQueue)
128129import Control.Distributed.Process.Internal.StrictMVar
129130 ( StrictMVar
130131 , withMVar
132+ , modifyMVar
131133 , modifyMVar_
132134 )
133135import Control.Distributed.Process.Internal.WeakTQueue (TQueue)
@@ -291,6 +293,16 @@ withValidLocalState node f = withMVar (localState node) $ \st -> case st of
291293 LocalNodeValid vst -> f vst
292294 LocalNodeClosed -> throwIO $ userError $ "Node closed " ++ show (localNodeId node)
293295
296+-- | Wrapper around 'modifyMVar' that checks that the local node is still in
297+-- a valid state.
298+modifyValidLocalState :: LocalNode
299+ -> (ValidLocalNodeState -> IO (ValidLocalNodeState, a))
300+ -> IO (Maybe a)
301+modifyValidLocalState node f = modifyMVar (localState node) $ \st -> case st of
302+ LocalNodeValid vst -> do (vst', a) <- f vst
303+ return (LocalNodeValid vst', Just a)
304+ LocalNodeClosed -> return (LocalNodeClosed, Nothing)
305+
294306-- | Wrapper around 'modifyMVar_' that checks that the local node is still in
295307-- a valid state.
296308modifyValidLocalState_ :: LocalNode
src/Control/Distributed/Process/Node.hs+16−7
@@ -99,6 +99,7 @@ import qualified Network.Transport as NT
9999 , TransportError(..)
100100 , address
101101 , closeEndPoint
102+ , Connection
102103 , ConnectionId
103104 , close
104105 , EndPointAddress
@@ -120,7 +121,7 @@ import Control.Distributed.Process.Internal.Types
120121 , LocalNodeState(..)
121122 , ValidLocalNodeState(..)
122123 , withValidLocalState
123− , modifyValidLocalState_
124+ , modifyValidLocalState
124125 , LocalProcess(..)
125126 , LocalProcessState(..)
126127 , Process(..)
@@ -389,7 +390,12 @@ forkProcess node proc =
389390 (return . DiedException . (show :: SomeException -> String)))]
390391
391392 -- [Unified: Table 4, rules termination and exiting]
392− modifyValidLocalState_ node (cleanupProcess pid)
393+ mconns <- modifyValidLocalState node (cleanupProcess pid)
394+ -- XXX: Revisit after agreeing on the bigger picture for the semantics
395+ -- of transport operations.
396+ -- https://github.com/haskell-distributed/distributed-process/issues/204
397+ forM_ mconns $ forkIO . mapM_ NT.close
398+
393399 writeChan (localCtrlChan node) NCMsg
394400 { ctrlMsgSender = ProcessIdentifier pid
395401 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
@@ -420,14 +426,17 @@ forkProcess node proc =
420426 )
421427 startProcess LocalNodeClosed = throwIO $ userError $ "Node closed " ++ show (localNodeId node)
422428
423− cleanupProcess :: ProcessId -> ValidLocalNodeState -> IO ValidLocalNodeState
429+ cleanupProcess :: ProcessId
430+ -> ValidLocalNodeState
431+ -> IO (ValidLocalNodeState, [NT.Connection])
424432 cleanupProcess pid vst = do
425433 let pid' = ProcessIdentifier pid
426434 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
427− mapM_ (NT.close . fst) (Map.elems affected)
428− return $ (localProcessWithId (processLocalId pid) ^= Nothing)
429− . (localConnections ^= unaffected)
430− $ vst
435+ return ( (localProcessWithId (processLocalId pid) ^= Nothing)
436+ . (localConnections ^= unaffected)
437+ $ vst
438+ , map fst $ Map.elems affected
439+ )
431440
432441-- note [tracer/forkProcess races]
433442--