04ba682 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Don't hold the localState MVar while calling potentially blocking operations.

04ba682Facundo Domínguez11 years ago

src/Control/Distributed/Process/Internal/Messaging.hs+17−9

compare with master

unifiedsplit

@@ -15,7 +15,9 @@ import qualified Data.ByteString.Lazy as BSL (toChunks)

15import qualified Data.ByteString as BSS (ByteString)

16import Control.Distributed.Process.Serializable ()

18+import Control.Concurrent (forkIO)

18import Control.Concurrent.Chan (writeChan)

19import Control.Concurrent.Chan (writeChan)

20+import Control.Exception (mask_)

19import Control.Monad (unless)

21import Control.Monad (unless)

20import Control.Monad.IO.Class (liftIO)

22import Control.Monad.IO.Class (liftIO)

21import Control.Monad.Reader (ask)

23import Control.Monad.Reader (ask)

@@ -30,6 +32,7 @@ import qualified Network.Transport as NT

30import Control.Distributed.Process.Internal.Types

32import Control.Distributed.Process.Internal.Types

31 ( LocalNode(localState, localEndPoint, localCtrlChan)

33 ( LocalNode(localState, localEndPoint, localCtrlChan)

32 , withValidLocalState

34 , withValidLocalState

35+ , modifyValidLocalState

33 , modifyValidLocalState_

36 , modifyValidLocalState_

34 , Identifier

37 , Identifier

35 , localConnections

38 , localConnections

@@ -51,6 +54,7 @@ import Control.Distributed.Process.Internal.Types

51 , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)

54 , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)

52 )

55 )

53import Control.Distributed.Process.Serializable (Serializable)

56import Control.Distributed.Process.Serializable (Serializable)

57+import Data.Foldable (forM_)

55--------------------------------------------------------------------------------

59--------------------------------------------------------------------------------

56-- Message sending --

60-- Message sending --

@@ -135,24 +139,28 @@ connBetween node from to implicitReconnect = do

135 setupConnBetween node from to implicitReconnect

139 setupConnBetween node from to implicitReconnect

136

140

137disconnect :: LocalNode -> Identifier -> Identifier -> IO ()

141disconnect :: LocalNode -> Identifier -> Identifier -> IO ()

138−disconnect node from to =

142+disconnect node from to = mask_ $ do

139− modifyValidLocalState_ node $ \vst ->

143+ mio <- modifyValidLocalState node $ \vst ->

140 case vst ^. localConnectionBetween from to of

144 case vst ^. localConnectionBetween from to of

141 Nothing ->

145 Nothing ->

142− return vst

146+ return (vst, return ())

143 Just (conn, _) -> do

147 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

146

152

147closeImplicitReconnections :: LocalNode -> Identifier -> IO ()

153closeImplicitReconnections :: LocalNode -> Identifier -> IO ()

148−closeImplicitReconnections node to =

154+closeImplicitReconnections node to = mask_ $ do

149− modifyValidLocalState_ node $ \vst -> do

155+ mconns <- modifyValidLocalState node $ \vst -> do

150 let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to'

156 let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to'

151 shouldClose _ _ = False

157 shouldClose _ _ = False

152 let (affected, unaffected) =

158 let (affected, unaffected) =

153 Map.partitionWithKey shouldClose (vst ^. localConnections)

159 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

156

164

157-- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node)

165-- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node)

158-- implies the death of @b@ (for instance, a process on that node)

166-- 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

24 , LocalNodeState(..)

25 , ValidLocalNodeState(..)

26 , withValidLocalState

27+ , modifyValidLocalState

27 , modifyValidLocalState_

28 , modifyValidLocalState_

28 , Tracer(..)

29 , Tracer(..)

29 , MxEventBus(..)

30 , MxEventBus(..)

@@ -128,6 +129,7 @@ import Control.Distributed.Process.Internal.CQueue (CQueue)

128import Control.Distributed.Process.Internal.StrictMVar

129import Control.Distributed.Process.Internal.StrictMVar

129 ( StrictMVar

130 ( StrictMVar

130 , withMVar

131 , withMVar

132+ , modifyMVar

131 , modifyMVar_

133 , modifyMVar_

132 )

134 )

133import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

135import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

@@ -291,6 +293,16 @@ withValidLocalState node f = withMVar (localState node) $ \st -> case st of

291 LocalNodeValid vst -> f vst

293 LocalNodeValid vst -> f vst

292 LocalNodeClosed -> throwIO $ userError $ "Node closed " ++ show (localNodeId node)

294 LocalNodeClosed -> throwIO $ userError $ "Node closed " ++ show (localNodeId node)

293

295

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+

294-- | Wrapper around 'modifyMVar_' that checks that the local node is still in

306-- | Wrapper around 'modifyMVar_' that checks that the local node is still in

295-- a valid state.

307-- a valid state.

296modifyValidLocalState_ :: LocalNode

308modifyValidLocalState_ :: LocalNode

src/Control/Distributed/Process/Node.hs+16−7

@@ -99,6 +99,7 @@ import qualified Network.Transport as NT

99 , TransportError(..)

100 , address

101 , closeEndPoint

102+ , Connection

102 , ConnectionId

103 , ConnectionId

103 , close

104 , close

104 , EndPointAddress

105 , EndPointAddress

@@ -120,7 +121,7 @@ import Control.Distributed.Process.Internal.Types

120 , LocalNodeState(..)

121 , LocalNodeState(..)

121 , ValidLocalNodeState(..)

122 , ValidLocalNodeState(..)

122 , withValidLocalState

123 , withValidLocalState

123− , modifyValidLocalState_

124+ , modifyValidLocalState

124 , LocalProcess(..)

125 , LocalProcess(..)

125 , LocalProcessState(..)

126 , LocalProcessState(..)

126 , Process(..)

127 , Process(..)

@@ -389,7 +390,12 @@ forkProcess node proc =

389 (return . DiedException . (show :: SomeException -> String)))]

390 (return . DiedException . (show :: SomeException -> String)))]

390

391

391 -- [Unified: Table 4, rules termination and exiting]

392 -- [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+

393 writeChan (localCtrlChan node) NCMsg

399 writeChan (localCtrlChan node) NCMsg

394 { ctrlMsgSender = ProcessIdentifier pid

400 { ctrlMsgSender = ProcessIdentifier pid

395 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason

401 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason

@@ -420,14 +426,17 @@ forkProcess node proc =

420 )

426 )

421 startProcess LocalNodeClosed = throwIO $ userError $ "Node closed " ++ show (localNodeId node)

427 startProcess LocalNodeClosed = throwIO $ userError $ "Node closed " ++ show (localNodeId node)

422

428

423− cleanupProcess :: ProcessId -> ValidLocalNodeState -> IO ValidLocalNodeState

429+ cleanupProcess :: ProcessId

430+ -> ValidLocalNodeState

431+ -> IO (ValidLocalNodeState, [NT.Connection])

424 cleanupProcess pid vst = do

432 cleanupProcess pid vst = do

425 let pid' = ProcessIdentifier pid

433 let pid' = ProcessIdentifier pid

426 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)

434 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)

435+ return ( (localProcessWithId (processLocalId pid) ^= Nothing)

429− . (localConnections ^= unaffected)

436+ . (localConnections ^= unaffected)

430− $ vst

437+ $ vst

438+ , map fst $ Map.elems affected

439+ )

431

440

432-- note [tracer/forkProcess races]

441-- note [tracer/forkProcess races]

433--

442--