git.iko.soy/distributed-process
on master help sign in
Allow LocalNode's to be in closed state.

Introduce new LocalNodeClosed state. This is a final state for local nodes, which they reach when 'closeLocalNode' closes the node.

commit8f5b6238e3689d094b647f77857a4d924a9d3782 authorMathieu Boespflug <m@tweag.io> dateSat 7 Mar 2015, 19:55 (12 years ago) parentd87528c
3 files changed+86−53
compare with master
unifiedsplit
src/Control/Distributed/Process/Internal/Messaging.hs+15−17 src/Control/Distributed/Process/Internal/Types.hs+39−10 src/Control/Distributed/Process/Node.hs+32−26
src/Control/Distributed/Process/Internal/Messaging.hs+15−17
@@ -13,7 +13,6 @@ import Data.Binary (Binary, encode)
1313import qualified Data.Map as Map (partitionWithKey, elems)
1414import qualified Data.ByteString.Lazy as BSL (toChunks)
1515import qualified Data.ByteString as BSS (ByteString)
16−import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)
1716import Control.Distributed.Process.Serializable ()
1817
1918import Control.Concurrent.Chan (writeChan)
@@ -30,6 +29,8 @@ import qualified Network.Transport as NT
3029 )
3130import Control.Distributed.Process.Internal.Types
3231 ( LocalNode(localState, localEndPoint, localCtrlChan)
32+ , withValidLocalState
33+ , modifyValidLocalState_
3334 , Identifier
3435 , localConnections
3536 , localConnectionBetween
@@ -102,7 +103,7 @@ setupConnBetween :: LocalNode
102103 -> ImplicitReconnect
103104 -> IO (Maybe NT.Connection)
104105setupConnBetween node from to implicitReconnect = do
105− mConn <- NT.connect endPoint
106+ mConn <- NT.connect (localEndPoint node)
106107 (nodeAddress . nodeOf $ to)
107108 NT.ReliableOrdered
108109 NT.defaultConnectHints
@@ -113,14 +114,11 @@ setupConnBetween node from to implicitReconnect = do
113114 Left _ ->
114115 return Nothing
115116 Right () -> do
116− modifyMVar_ nodeState $ return .
117− (localConnectionBetween from to ^= Just (conn, implicitReconnect))
117+ modifyValidLocalState_ (localState node) $
118+ return . (localConnectionBetween from to ^= Just (conn, implicitReconnect))
118119 return $ Just conn
119120 Left _ ->
120121 return Nothing
121− where
122− endPoint = localEndPoint node
123− nodeState = localState node
124122
125123connBetween :: LocalNode
126124 -> Identifier
@@ -128,33 +126,33 @@ connBetween :: LocalNode
128126 -> ImplicitReconnect
129127 -> IO (Maybe NT.Connection)
130128connBetween node from to implicitReconnect = do
131− mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)
129+ mConn <- withValidLocalState (localState node) $
130+ return . (^. localConnectionBetween from to)
132131 case mConn of
133132 Just (conn, _) ->
134133 return $ Just conn
135134 Nothing ->
136135 setupConnBetween node from to implicitReconnect
137− where
138− nodeState = localState node
139136
140137disconnect :: LocalNode -> Identifier -> Identifier -> IO ()
141138disconnect node from to =
142− modifyMVar_ (localState node) $ \st ->
143− case st ^. localConnectionBetween from to of
139+ modifyValidLocalState_ (localState node) $ \vst ->
140+ case vst ^. localConnectionBetween from to of
144141 Nothing ->
145− return st
142+ return vst
146143 Just (conn, _) -> do
147144 NT.close conn
148− return (localConnectionBetween from to ^= Nothing $ st)
145+ return $ localConnectionBetween from to ^= Nothing $ vst
149146
150147closeImplicitReconnections :: LocalNode -> Identifier -> IO ()
151148closeImplicitReconnections node to =
152− modifyMVar_ (localState node) $ \st -> do
149+ modifyValidLocalState_ (localState node) $ \vst -> do
153150 let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to'
154151 shouldClose _ _ = False
155− let (affected, unaffected) = Map.partitionWithKey shouldClose (st ^. localConnections)
152+ let (affected, unaffected) =
153+ Map.partitionWithKey shouldClose (vst ^. localConnections)
156154 mapM_ (NT.close . fst) (Map.elems affected)
157− return (localConnections ^= unaffected $ st)
155+ return $ localConnections ^= unaffected $ vst
158156
159157-- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node)
160158-- implies the death of @b@ (for instance, a process on that node)
src/Control/Distributed/Process/Internal/Types.hs+39−10
@@ -20,9 +20,12 @@ module Control.Distributed.Process.Internal.Types
2020 , nullProcessId
2121 -- * Local nodes and processes
2222 , LocalNode(..)
23+ , LocalNodeState(..)
24+ , ValidLocalNodeState(..)
25+ , withValidLocalState
26+ , modifyValidLocalState_
2327 , Tracer(..)
2428 , MxEventBus(..)
25− , LocalNodeState(..)
2629 , LocalProcess(..)
2730 , LocalProcessState(..)
2831 , Process(..)
@@ -101,7 +104,7 @@ import qualified Data.ByteString.Lazy.Internal as BSL (ByteString(..))
101104import Data.Accessor (Accessor, accessor)
102105import Control.Category ((>>>))
103106import Control.DeepSeq (NFData(..))
104−import Control.Exception (Exception)
107+import Control.Exception (Exception, throwIO)
105108import Control.Concurrent (ThreadId)
106109import Control.Concurrent.Chan (Chan)
107110import Control.Concurrent.STM (STM)
@@ -121,7 +124,11 @@ import Control.Distributed.Process.Serializable
121124 , showFingerprint
122125 )
123126import Control.Distributed.Process.Internal.CQueue (CQueue)
124−import Control.Distributed.Process.Internal.StrictMVar (StrictMVar)
127+import Control.Distributed.Process.Internal.StrictMVar
128+ ( StrictMVar
129+ , withMVar
130+ , modifyMVar_
131+ )
125132import Control.Distributed.Process.Internal.WeakTQueue (TQueue)
126133import Control.Distributed.Static (RemoteTable, Closure)
127134import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)
@@ -253,7 +260,11 @@ data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect
253260 deriving (Eq, Show)
254261
255262-- | Local node state
256−data LocalNodeState = LocalNodeState
263+data LocalNodeState =
264+ LocalNodeValid {-# UNPACK #-} !ValidLocalNodeState
265+ | LocalNodeClosed
266+
267+data ValidLocalNodeState = ValidLocalNodeState
257268 { -- | Processes running on this node
258269 _localProcesses :: !(Map LocalProcessId LocalProcess)
259270 -- | Counter to assign PIDs
@@ -266,6 +277,24 @@ data LocalNodeState = LocalNodeState
266277 (NT.Connection, ImplicitReconnect))
267278 }
268279
280+-- | Wrapper around 'withMVar' that checks that the local node is still in
281+-- a valid state.
282+withValidLocalState :: StrictMVar LocalNodeState
283+ -> (ValidLocalNodeState -> IO r)
284+ -> IO r
285+withValidLocalState mv f = withMVar mv $ \st -> case st of
286+ LocalNodeValid vst -> f vst
287+ LocalNodeClosed -> throwIO $ userError "LocalNode closed"
288+
289+-- | Wrapper around 'modifyMVar_' that checks that the local node is still in
290+-- a valid state.
291+modifyValidLocalState_ :: StrictMVar LocalNodeState
292+ -> (ValidLocalNodeState -> IO ValidLocalNodeState)
293+ -> IO ()
294+modifyValidLocalState_ mv f = modifyMVar_ mv $ \st -> case st of
295+ LocalNodeValid vst -> LocalNodeValid <$> f vst
296+ LocalNodeClosed -> throwIO $ userError "LocalNode closed"
297+
269298-- | Processes running on our local node
270299data LocalProcess = LocalProcess
271300 { processQueue :: !(CQueue Message)
@@ -725,22 +754,22 @@ instance Binary ProcessInfoNone where
725754-- Accessors --
726755--------------------------------------------------------------------------------
727756
728−localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)
757+localProcesses :: Accessor ValidLocalNodeState (Map LocalProcessId LocalProcess)
729758localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
730759
731−localPidCounter :: Accessor LocalNodeState Int32
760+localPidCounter :: Accessor ValidLocalNodeState Int32
732761localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr })
733762
734−localPidUnique :: Accessor LocalNodeState Int32
763+localPidUnique :: Accessor ValidLocalNodeState Int32
735764localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
736765
737−localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect))
766+localConnections :: Accessor ValidLocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect))
738767localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })
739768
740−localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
769+localProcessWithId :: LocalProcessId -> Accessor ValidLocalNodeState (Maybe LocalProcess)
741770localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
742771
743−localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (NT.Connection, ImplicitReconnect))
772+localConnectionBetween :: Identifier -> Identifier -> Accessor ValidLocalNodeState (Maybe (NT.Connection, ImplicitReconnect))
744773localConnectionBetween from' to' = localConnections >>> DAC.mapMaybe (from', to')
745774
746775monitorCounter :: Accessor LocalProcessState Int32
src/Control/Distributed/Process/Node.hs+32−26
@@ -76,7 +76,6 @@ import Control.Distributed.Process.Internal.StrictMVar
7676 , newEmptyMVar
7777 , putMVar
7878 , takeMVar
79− , readMVar
8079 )
8180import Control.Concurrent.Chan (newChan, writeChan, readChan)
8281import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
@@ -119,6 +118,9 @@ import Control.Distributed.Process.Internal.Types
119118 , LocalNode(..)
120119 , MxEventBus(..)
121120 , LocalNodeState(..)
121+ , ValidLocalNodeState(..)
122+ , withValidLocalState
123+ , modifyValidLocalState_
122124 , LocalProcess(..)
123125 , LocalProcessState(..)
124126 , Process(..)
@@ -231,7 +233,7 @@ newLocalNode transport rtable = do
231233createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
232234createBareLocalNode endPoint rtable = do
233235 unq <- randomIO
234− state <- newMVar LocalNodeState
236+ state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
235237 { _localProcesses = Map.empty
236238 , _localPidCounter = firstNonReservedProcessId
237239 , _localPidUnique = unq
@@ -322,7 +324,8 @@ startServiceProcesses node = do
322324--
323325-- TODO: for now we just close the associated endpoint
324326closeLocalNode :: LocalNode -> IO ()
325−closeLocalNode node =
327+closeLocalNode node = do
328+ modifyMVar_ (localState node) $ const $ return LocalNodeClosed
326329 -- TODO: close all our processes, surely!?
327330 NT.closeEndPoint (localEndPoint node)
328331
@@ -342,9 +345,9 @@ forkProcess node proc =
342345 modifyMVarMasked (localState node) startProcess
343346 where
344347 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
345− startProcess st = do
346− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
347− , lpidUnique = st ^. localPidUnique
348+ startProcess (LocalNodeValid vst) = do
349+ let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
350+ , lpidUnique = vst ^. localPidUnique
348351 }
349352 let pid = ProcessId { processNodeId = localNodeId node
350353 , processLocalId = lpid
@@ -376,7 +379,7 @@ forkProcess node proc =
376379 (return . DiedException . (show :: SomeException -> String)))]
377380
378381 -- [Unified: Table 4, rules termination and exiting]
379− modifyMVar_ (localState node) (cleanupProcess pid)
382+ modifyValidLocalState_ (localState node) (cleanupProcess pid)
380383 writeChan (localCtrlChan node) NCMsg
381384 { ctrlMsgSender = ProcessIdentifier pid
382385 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
@@ -391,27 +394,30 @@ forkProcess node proc =
391394 -- TODO: this doesn't look right at all - how do we know
392395 -- that newUnique represents a process id that is available!?
393396 newUnique <- randomIO
394− return ( (localProcessWithId lpid ^= Just lproc)
397+ return ( LocalNodeValid
398+ $ (localProcessWithId lpid ^= Just lproc)
395399 . (localPidCounter ^= firstNonReservedProcessId)
396400 . (localPidUnique ^= newUnique)
397− $ st
401+ $ vst
398402 , pid
399403 )
400404 else
401− return ( (localProcessWithId lpid ^= Just lproc)
405+ return ( LocalNodeValid
406+ $ (localProcessWithId lpid ^= Just lproc)
402407 . (localPidCounter ^: (+ 1))
403− $ st
408+ $ vst
404409 , pid
405410 )
411+ startProcess LocalNodeClosed = throwIO $ userError "LocalNode closed"
406412
407− cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState
408− cleanupProcess pid st = do
413+ cleanupProcess :: ProcessId -> ValidLocalNodeState -> IO ValidLocalNodeState
414+ cleanupProcess pid vst = do
409415 let pid' = ProcessIdentifier pid
410− let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)
416+ let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
411417 mapM_ (NT.close . fst) (Map.elems affected)
412418 return $ (localProcessWithId (processLocalId pid) ^= Nothing)
413419 . (localConnections ^= unaffected)
414− $ st
420+ $ vst
415421
416422-- note [tracer/forkProcess races]
417423--
@@ -500,7 +506,7 @@ handleIncomingMessages node = go initConnectionState
500506 case decode (BSL.fromChunks payload) of
501507 ProcessIdentifier pid -> do
502508 let lpid = processLocalId pid
503− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
509+ mProc <- withValidLocalState state $ return . (^. localProcessWithId lpid)
504510 case mProc of
505511 Just proc ->
506512 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
@@ -509,7 +515,7 @@ handleIncomingMessages node = go initConnectionState
509515 SendPortIdentifier chId -> do
510516 let lcid = sendPortLocalId chId
511517 lpid = processLocalId (sendPortProcessId chId)
512− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
518+ mProc <- withValidLocalState state $ return . (^. localProcessWithId lpid)
513519 case mProc of
514520 Just proc -> do
515521 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
@@ -970,8 +976,8 @@ ncEffectGetInfo from pid =
970976 them = (ProcessIdentifier pid)
971977 in do
972978 node <- ask
973− mProc <- liftIO $
974− withMVar (localState node) $ return . (^. localProcessWithId lpid)
979+ mProc <- liftIO $ withValidLocalState (localState node)
980+ $ return . (^. localProcessWithId lpid)
975981 case mProc of
976982 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
977983 from node (ProcessInfoNone DiedUnknownId)
@@ -1014,17 +1020,17 @@ ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
10141020ncEffectGetNodeStats from _nid = do
10151021 node <- ask
10161022 ncState <- StateT.get
1017− nodeState <- liftIO $ readMVar (localState node)
1018− let localProcesses' = nodeState ^. localProcesses
1019− stats =
1023+ nodeState <- liftIO $ withValidLocalState (localState node) return
1024+ let stats =
10201025 NodeStats {
10211026 nodeStatsNode = localNodeId node
10221027 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
10231028 , nodeStatsMonitors = Map.size $ ncState ^. monitors
10241029 , nodeStatsLinks = Map.size $ ncState ^. links
1025− , nodeStatsProcesses = Map.size localProcesses'
1030+ , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
10261031 }
10271032 postAsMessage from stats
1033+
10281034--------------------------------------------------------------------------------
10291035-- Auxiliary --
10301036--------------------------------------------------------------------------------
@@ -1097,7 +1103,7 @@ unClosure closure = do
10971103isValidLocalIdentifier :: Identifier -> NC Bool
10981104isValidLocalIdentifier ident = do
10991105 node <- ask
1100− liftIO . withMVar (localState node) $ \nSt ->
1106+ liftIO . withValidLocalState (localState node) $ \nSt ->
11011107 case ident of
11021108 NodeIdentifier nid ->
11031109 return $ nid == localNodeId node
@@ -1135,8 +1141,8 @@ withLocalProc node pid p =
11351141 -- By [Unified: table 6, rule missing_process] messages to dead processes
11361142 -- can silently be dropped
11371143 let lpid = processLocalId pid in do
1138− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
1139− forM_ mProc p
1144+ withValidLocalState (localState node) $ \vst ->
1145+ forM_ (vst ^. localProcessWithId lpid) p
11401146
11411147--------------------------------------------------------------------------------
11421148-- Accessors --