8f5b623 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Allow LocalNode's to be in closed state.

8f5b623Mathieu Boespflug12 years ago

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

compare with master

unifiedsplit

@@ -13,7 +13,6 @@ import Data.Binary (Binary, encode)

13import qualified Data.Map as Map (partitionWithKey, elems)

14import qualified Data.ByteString.Lazy as BSL (toChunks)

15import qualified Data.ByteString as BSS (ByteString)

16−import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_)

17import Control.Distributed.Process.Serializable ()

16import Control.Distributed.Process.Serializable ()

19import Control.Concurrent.Chan (writeChan)

18import Control.Concurrent.Chan (writeChan)

@@ -30,6 +29,8 @@ import qualified Network.Transport as NT

30 )

29 )

31import Control.Distributed.Process.Internal.Types

30import Control.Distributed.Process.Internal.Types

32 ( LocalNode(localState, localEndPoint, localCtrlChan)

31 ( LocalNode(localState, localEndPoint, localCtrlChan)

32+ , withValidLocalState

33+ , modifyValidLocalState_

33 , Identifier

34 , Identifier

34 , localConnections

35 , localConnections

35 , localConnectionBetween

36 , localConnectionBetween

@@ -102,7 +103,7 @@ setupConnBetween :: LocalNode

102 -> ImplicitReconnect

103 -> ImplicitReconnect

103 -> IO (Maybe NT.Connection)

104 -> IO (Maybe NT.Connection)

104setupConnBetween node from to implicitReconnect = do

105setupConnBetween node from to implicitReconnect = do

105− mConn <- NT.connect endPoint

106+ mConn <- NT.connect (localEndPoint node)

106 (nodeAddress . nodeOf $ to)

107 (nodeAddress . nodeOf $ to)

107 NT.ReliableOrdered

108 NT.ReliableOrdered

108 NT.defaultConnectHints

109 NT.defaultConnectHints

@@ -113,14 +114,11 @@ setupConnBetween node from to implicitReconnect = do

113 Left _ ->

114 Left _ ->

114 return Nothing

115 return Nothing

115 Right () -> do

116 Right () -> do

116− modifyMVar_ nodeState $ return .

117+ modifyValidLocalState_ (localState node) $

117− (localConnectionBetween from to ^= Just (conn, implicitReconnect))

118+ return . (localConnectionBetween from to ^= Just (conn, implicitReconnect))

118 return $ Just conn

119 return $ Just conn

119 Left _ ->

120 Left _ ->

120 return Nothing

121 return Nothing

121− where

122− endPoint = localEndPoint node

123− nodeState = localState node

124

122

125connBetween :: LocalNode

123connBetween :: LocalNode

126 -> Identifier

124 -> Identifier

@@ -128,33 +126,33 @@ connBetween :: LocalNode

128 -> ImplicitReconnect

126 -> ImplicitReconnect

129 -> IO (Maybe NT.Connection)

127 -> IO (Maybe NT.Connection)

130connBetween node from to implicitReconnect = do

128connBetween node from to implicitReconnect = do

129+ mConn <- withValidLocalState (localState node) $

131− mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)

130+ return . (^. localConnectionBetween from to)

132 case mConn of

131 case mConn of

133 Just (conn, _) ->

132 Just (conn, _) ->

134 return $ Just conn

133 return $ Just conn

135 Nothing ->

134 Nothing ->

136 setupConnBetween node from to implicitReconnect

135 setupConnBetween node from to implicitReconnect

137− where

138− nodeState = localState node

139

136

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

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

141disconnect node from to =

138disconnect node from to =

142− modifyMVar_ (localState node) $ \st ->

139+ modifyValidLocalState_ (localState node) $ \vst ->

143− case st ^. localConnectionBetween from to of

140+ case vst ^. localConnectionBetween from to of

144 Nothing ->

141 Nothing ->

145− return st

142+ return vst

146 Just (conn, _) -> do

143 Just (conn, _) -> do

147 NT.close conn

144 NT.close conn

148− return (localConnectionBetween from to ^= Nothing $ st)

145+ return $ localConnectionBetween from to ^= Nothing $ vst

149

146

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

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

151closeImplicitReconnections node to =

148closeImplicitReconnections node to =

152− modifyMVar_ (localState node) $ \st -> do

149+ modifyValidLocalState_ (localState node) $ \vst -> do

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

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

154 shouldClose _ _ = False

151 shouldClose _ _ = False

152+ let (affected, unaffected) =

155− let (affected, unaffected) = Map.partitionWithKey shouldClose (st ^. localConnections)

153+ Map.partitionWithKey shouldClose (vst ^. localConnections)

156 mapM_ (NT.close . fst) (Map.elems affected)

154 mapM_ (NT.close . fst) (Map.elems affected)

157− return (localConnections ^= unaffected $ st)

155+ return $ localConnections ^= unaffected $ vst

158

156

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

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

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

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

20 , nullProcessId

21 -- * Local nodes and processes

22 , LocalNode(..)

23+ , LocalNodeState(..)

24+ , ValidLocalNodeState(..)

25+ , withValidLocalState

26+ , modifyValidLocalState_

23 , Tracer(..)

27 , Tracer(..)

24 , MxEventBus(..)

28 , MxEventBus(..)

25− , LocalNodeState(..)

26 , LocalProcess(..)

29 , LocalProcess(..)

27 , LocalProcessState(..)

30 , LocalProcessState(..)

28 , Process(..)

31 , Process(..)

@@ -101,7 +104,7 @@ import qualified Data.ByteString.Lazy.Internal as BSL (ByteString(..))

101import Data.Accessor (Accessor, accessor)

104import Data.Accessor (Accessor, accessor)

102import Control.Category ((>>>))

105import Control.Category ((>>>))

103import Control.DeepSeq (NFData(..))

106import Control.DeepSeq (NFData(..))

104−import Control.Exception (Exception)

107+import Control.Exception (Exception, throwIO)

105import Control.Concurrent (ThreadId)

108import Control.Concurrent (ThreadId)

106import Control.Concurrent.Chan (Chan)

109import Control.Concurrent.Chan (Chan)

107import Control.Concurrent.STM (STM)

110import Control.Concurrent.STM (STM)

@@ -121,7 +124,11 @@ import Control.Distributed.Process.Serializable

121 , showFingerprint

124 , showFingerprint

122 )

125 )

123import Control.Distributed.Process.Internal.CQueue (CQueue)

126import 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+ )

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

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

126import Control.Distributed.Static (RemoteTable, Closure)

133import Control.Distributed.Static (RemoteTable, Closure)

127import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

134import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

@@ -253,7 +260,11 @@ data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

253 deriving (Eq, Show)

260 deriving (Eq, Show)

254

261

255-- | Local node state

262-- | Local node state

256−data LocalNodeState = LocalNodeState

263+data LocalNodeState =

264+ LocalNodeValid {-# UNPACK #-} !ValidLocalNodeState

265+ | LocalNodeClosed

266+

267+data ValidLocalNodeState = ValidLocalNodeState

257 { -- | Processes running on this node

268 { -- | Processes running on this node

258 _localProcesses :: !(Map LocalProcessId LocalProcess)

269 _localProcesses :: !(Map LocalProcessId LocalProcess)

259 -- | Counter to assign PIDs

270 -- | Counter to assign PIDs

@@ -266,6 +277,24 @@ data LocalNodeState = LocalNodeState

266 (NT.Connection, ImplicitReconnect))

277 (NT.Connection, ImplicitReconnect))

267 }

278 }

268

279

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+

269-- | Processes running on our local node

298-- | Processes running on our local node

270data LocalProcess = LocalProcess

299data LocalProcess = LocalProcess

271 { processQueue :: !(CQueue Message)

300 { processQueue :: !(CQueue Message)

@@ -725,22 +754,22 @@ instance Binary ProcessInfoNone where

725-- Accessors --

754-- Accessors --

726--------------------------------------------------------------------------------

755--------------------------------------------------------------------------------

727

756

728−localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)

757+localProcesses :: Accessor ValidLocalNodeState (Map LocalProcessId LocalProcess)

729localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })

758localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })

730

759

731−localPidCounter :: Accessor LocalNodeState Int32

760+localPidCounter :: Accessor ValidLocalNodeState Int32

732localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr })

761localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr })

733

762

734−localPidUnique :: Accessor LocalNodeState Int32

763+localPidUnique :: Accessor ValidLocalNodeState Int32

735localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

764localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

736

765

737−localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect))

766+localConnections :: Accessor ValidLocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect))

738localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })

767localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })

739

768

740−localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

769+localProcessWithId :: LocalProcessId -> Accessor ValidLocalNodeState (Maybe LocalProcess)

741localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

770localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

742

771

743−localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (NT.Connection, ImplicitReconnect))

772+localConnectionBetween :: Identifier -> Identifier -> Accessor ValidLocalNodeState (Maybe (NT.Connection, ImplicitReconnect))

744localConnectionBetween from' to' = localConnections >>> DAC.mapMaybe (from', to')

773localConnectionBetween from' to' = localConnections >>> DAC.mapMaybe (from', to')

745

774

746monitorCounter :: Accessor LocalProcessState Int32

775monitorCounter :: Accessor LocalProcessState Int32

src/Control/Distributed/Process/Node.hs+32−26

@@ -76,7 +76,6 @@ import Control.Distributed.Process.Internal.StrictMVar

76 , newEmptyMVar

77 , putMVar

78 , takeMVar

79− , readMVar

80 )

79 )

81import Control.Concurrent.Chan (newChan, writeChan, readChan)

80import Control.Concurrent.Chan (newChan, writeChan, readChan)

82import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)

81import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)

@@ -119,6 +118,9 @@ import Control.Distributed.Process.Internal.Types

119 , LocalNode(..)

118 , LocalNode(..)

120 , MxEventBus(..)

119 , MxEventBus(..)

121 , LocalNodeState(..)

120 , LocalNodeState(..)

121+ , ValidLocalNodeState(..)

122+ , withValidLocalState

123+ , modifyValidLocalState_

122 , LocalProcess(..)

124 , LocalProcess(..)

123 , LocalProcessState(..)

125 , LocalProcessState(..)

124 , Process(..)

126 , Process(..)

@@ -231,7 +233,7 @@ newLocalNode transport rtable = do

231createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode

233createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode

232createBareLocalNode endPoint rtable = do

234createBareLocalNode endPoint rtable = do

233 unq <- randomIO

235 unq <- randomIO

234− state <- newMVar LocalNodeState

236+ state <- newMVar $ LocalNodeValid $ ValidLocalNodeState

235 { _localProcesses = Map.empty

237 { _localProcesses = Map.empty

236 , _localPidCounter = firstNonReservedProcessId

238 , _localPidCounter = firstNonReservedProcessId

237 , _localPidUnique = unq

239 , _localPidUnique = unq

@@ -322,7 +324,8 @@ startServiceProcesses node = do

322--

324--

323-- TODO: for now we just close the associated endpoint

325-- TODO: for now we just close the associated endpoint

324closeLocalNode :: LocalNode -> IO ()

326closeLocalNode :: LocalNode -> IO ()

325−closeLocalNode node =

327+closeLocalNode node = do

328+ modifyMVar_ (localState node) $ const $ return LocalNodeClosed

326 -- TODO: close all our processes, surely!?

329 -- TODO: close all our processes, surely!?

327 NT.closeEndPoint (localEndPoint node)

330 NT.closeEndPoint (localEndPoint node)

328

331

@@ -342,9 +345,9 @@ forkProcess node proc =

342 modifyMVarMasked (localState node) startProcess

345 modifyMVarMasked (localState node) startProcess

343 where

346 where

344 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)

347 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)

345− startProcess st = do

348+ startProcess (LocalNodeValid vst) = do

346− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter

349+ let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter

347− , lpidUnique = st ^. localPidUnique

350+ , lpidUnique = vst ^. localPidUnique

348 }

351 }

349 let pid = ProcessId { processNodeId = localNodeId node

352 let pid = ProcessId { processNodeId = localNodeId node

350 , processLocalId = lpid

353 , processLocalId = lpid

@@ -376,7 +379,7 @@ forkProcess node proc =

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

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

377

380

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

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

379− modifyMVar_ (localState node) (cleanupProcess pid)

382+ modifyValidLocalState_ (localState node) (cleanupProcess pid)

380 writeChan (localCtrlChan node) NCMsg

383 writeChan (localCtrlChan node) NCMsg

381 { ctrlMsgSender = ProcessIdentifier pid

384 { ctrlMsgSender = ProcessIdentifier pid

382 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason

385 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason

@@ -391,27 +394,30 @@ forkProcess node proc =

391 -- TODO: this doesn't look right at all - how do we know

394 -- TODO: this doesn't look right at all - how do we know

392 -- that newUnique represents a process id that is available!?

395 -- that newUnique represents a process id that is available!?

393 newUnique <- randomIO

396 newUnique <- randomIO

397+ return ( LocalNodeValid

394− return ( (localProcessWithId lpid ^= Just lproc)

398+ $ (localProcessWithId lpid ^= Just lproc)

395 . (localPidCounter ^= firstNonReservedProcessId)

399 . (localPidCounter ^= firstNonReservedProcessId)

396 . (localPidUnique ^= newUnique)

400 . (localPidUnique ^= newUnique)

397− $ st

401+ $ vst

398 , pid

402 , pid

399 )

403 )

400 else

404 else

405+ return ( LocalNodeValid

401− return ( (localProcessWithId lpid ^= Just lproc)

406+ $ (localProcessWithId lpid ^= Just lproc)

402 . (localPidCounter ^: (+ 1))

407 . (localPidCounter ^: (+ 1))

403− $ st

408+ $ vst

404 , pid

409 , pid

405 )

410 )

411+ startProcess LocalNodeClosed = throwIO $ userError "LocalNode closed"

406

412

407− cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState

413+ cleanupProcess :: ProcessId -> ValidLocalNodeState -> IO ValidLocalNodeState

408− cleanupProcess pid st = do

414+ cleanupProcess pid vst = do

409 let pid' = ProcessIdentifier pid

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

411 mapM_ (NT.close . fst) (Map.elems affected)

417 mapM_ (NT.close . fst) (Map.elems affected)

412 return $ (localProcessWithId (processLocalId pid) ^= Nothing)

418 return $ (localProcessWithId (processLocalId pid) ^= Nothing)

413 . (localConnections ^= unaffected)

419 . (localConnections ^= unaffected)

414− $ st

420+ $ vst

415

421

416-- note [tracer/forkProcess races]

422-- note [tracer/forkProcess races]

417--

423--

@@ -500,7 +506,7 @@ handleIncomingMessages node = go initConnectionState

500 case decode (BSL.fromChunks payload) of

506 case decode (BSL.fromChunks payload) of

501 ProcessIdentifier pid -> do

507 ProcessIdentifier pid -> do

502 let lpid = processLocalId pid

508 let lpid = processLocalId pid

503− mProc <- withMVar state $ return . (^. localProcessWithId lpid)

509+ mProc <- withValidLocalState state $ return . (^. localProcessWithId lpid)

504 case mProc of

510 case mProc of

505 Just proc ->

511 Just proc ->

506 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)

512 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)

@@ -509,7 +515,7 @@ handleIncomingMessages node = go initConnectionState

509 SendPortIdentifier chId -> do

515 SendPortIdentifier chId -> do

510 let lcid = sendPortLocalId chId

516 let lcid = sendPortLocalId chId

511 lpid = processLocalId (sendPortProcessId chId)

517 lpid = processLocalId (sendPortProcessId chId)

512− mProc <- withMVar state $ return . (^. localProcessWithId lpid)

518+ mProc <- withValidLocalState state $ return . (^. localProcessWithId lpid)

513 case mProc of

519 case mProc of

514 Just proc -> do

520 Just proc -> do

515 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

521 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

@@ -970,8 +976,8 @@ ncEffectGetInfo from pid =

970 them = (ProcessIdentifier pid)

976 them = (ProcessIdentifier pid)

971 in do

977 in do

972 node <- ask

978 node <- ask

973− mProc <- liftIO $

979+ mProc <- liftIO $ withValidLocalState (localState node)

974− withMVar (localState node) $ return . (^. localProcessWithId lpid)

980+ $ return . (^. localProcessWithId lpid)

975 case mProc of

981 case mProc of

976 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

982 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

977 from node (ProcessInfoNone DiedUnknownId)

983 from node (ProcessInfoNone DiedUnknownId)

@@ -1014,17 +1020,17 @@ ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()

1014ncEffectGetNodeStats from _nid = do

1020ncEffectGetNodeStats from _nid = do

1015 node <- ask

1021 node <- ask

1016 ncState <- StateT.get

1022 ncState <- StateT.get

1017− nodeState <- liftIO $ readMVar (localState node)

1023+ nodeState <- liftIO $ withValidLocalState (localState node) return

1018− let localProcesses' = nodeState ^. localProcesses

1019− stats =

1024+ let stats =

1020 NodeStats {

1025 NodeStats {

1021 nodeStatsNode = localNodeId node

1026 nodeStatsNode = localNodeId node

1022 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere

1027 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere

1023 , nodeStatsMonitors = Map.size $ ncState ^. monitors

1028 , nodeStatsMonitors = Map.size $ ncState ^. monitors

1024 , nodeStatsLinks = Map.size $ ncState ^. links

1029 , nodeStatsLinks = Map.size $ ncState ^. links

1025− , nodeStatsProcesses = Map.size localProcesses'

1030+ , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)

1026 }

1031 }

1027 postAsMessage from stats

1032 postAsMessage from stats

1033+

1028--------------------------------------------------------------------------------

1034--------------------------------------------------------------------------------

1029-- Auxiliary --

1035-- Auxiliary --

1030--------------------------------------------------------------------------------

1036--------------------------------------------------------------------------------

@@ -1097,7 +1103,7 @@ unClosure closure = do

1097isValidLocalIdentifier :: Identifier -> NC Bool

1103isValidLocalIdentifier :: Identifier -> NC Bool

1098isValidLocalIdentifier ident = do

1104isValidLocalIdentifier ident = do

1099 node <- ask

1105 node <- ask

1100− liftIO . withMVar (localState node) $ \nSt ->

1106+ liftIO . withValidLocalState (localState node) $ \nSt ->

1101 case ident of

1107 case ident of

1102 NodeIdentifier nid ->

1108 NodeIdentifier nid ->

1103 return $ nid == localNodeId node

1109 return $ nid == localNodeId node

@@ -1135,8 +1141,8 @@ withLocalProc node pid p =

1135 -- By [Unified: table 6, rule missing_process] messages to dead processes

1141 -- By [Unified: table 6, rule missing_process] messages to dead processes

1136 -- can silently be dropped

1142 -- can silently be dropped

1137 let lpid = processLocalId pid in do

1143 let lpid = processLocalId pid in do

1144+ withValidLocalState (localState node) $ \vst ->

1138− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)

1145+ forM_ (vst ^. localProcessWithId lpid) p

1139− forM_ mProc p

1140

1146

1141--------------------------------------------------------------------------------

1147--------------------------------------------------------------------------------

1142-- Accessors --

1148-- Accessors --