@@ -25,6 +25,7 @@ import qualified Data.Map as Map
2525 , toList
2626 , partitionWithKey
2727 , filterWithKey
28+ , elems
2829 )
2930import Data.Set (Set)
3031import qualified Data.Set as Set (empty, insert, delete, member, filter)
@@ -64,6 +65,8 @@ import qualified Network.Transport as NT
6465 , address
6566 , closeEndPoint
6667 , ConnectionId
68+ , Connection
69+ , close
6770 )
6871import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
6972import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe)
@@ -88,6 +91,7 @@ import Control.Distributed.Process.Internal.Types
8891 , localPidCounter
8992 , localPidUnique
9093 , localProcessWithId
94+ , localConnections
9195 , MonitorRef(..)
9296 , ProcessMonitorNotification(..)
9397 , NodeMonitorNotification(..)
@@ -187,54 +191,65 @@ runProcess node proc = do
187191
188192-- | Spawn a new process on a local node
189193forkProcess :: LocalNode -> Process () -> IO ProcessId
190−forkProcess node proc = modifyMVar (localState node) $ \st -> do
191− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
192− , lpidUnique = st ^. localPidUnique
193− }
194− let pid = ProcessId { processNodeId = localNodeId node
195− , processLocalId = lpid
196− }
197− pst <- newMVar LocalProcessState { _monitorCounter = 0
198− , _spawnCounter = 0
199− , _channelCounter = 0
200− , _typedChannels = Map.empty
201− }
202− queue <- newCQueue
203− (_, lproc) <- fixIO $ \ ~(tid, _) -> do
204− let lproc = LocalProcess { processQueue = queue
205− , processId = pid
206− , processState = pst
207− , processThread = tid
208− , processNode = node
209− }
210− tid' <- forkIO $ do
211− reason <- Exception.catch
212− (runLocalProcess lproc proc >> return DiedNormal)
213− (return . DiedException . (show :: SomeException -> String))
214− -- [Unified: Table 4, rules termination and exiting]
215− modifyMVar_ (localState node) $
216− return . (localProcessWithId lpid ^= Nothing)
217− writeChan (localCtrlChan node) NCMsg
218− { ctrlMsgSender = ProcessIdentifier pid
219− , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
220− }
221− return (tid', lproc)
222−
223− if lpidCounter lpid == maxBound
224− then do
225− newUnique <- randomIO
226− return ( (localProcessWithId lpid ^= Just lproc)
227− . (localPidCounter ^= 0)
228− . (localPidUnique ^= newUnique)
229− $ st
230− , pid
231− )
232− else
233− return ( (localProcessWithId lpid ^= Just lproc)
234− . (localPidCounter ^: (+ 1))
194+forkProcess node proc = modifyMVar (localState node) startProcess
195+ where
196+ startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
197+ startProcess st = do
198+ let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
199+ , lpidUnique = st ^. localPidUnique
200+ }
201+ let pid = ProcessId { processNodeId = localNodeId node
202+ , processLocalId = lpid
203+ }
204+ pst <- newMVar LocalProcessState { _monitorCounter = 0
205+ , _spawnCounter = 0
206+ , _channelCounter = 0
207+ , _typedChannels = Map.empty
208+ }
209+ queue <- newCQueue
210+ (_, lproc) <- fixIO $ \ ~(tid, _) -> do
211+ let lproc = LocalProcess { processQueue = queue
212+ , processId = pid
213+ , processState = pst
214+ , processThread = tid
215+ , processNode = node
216+ }
217+ tid' <- forkIO $ do
218+ reason <- Exception.catch
219+ (runLocalProcess lproc proc >> return DiedNormal)
220+ (return . DiedException . (show :: SomeException -> String))
221+ -- [Unified: Table 4, rules termination and exiting]
222+ modifyMVar_ (localState node) (cleanupProcess pid)
223+ writeChan (localCtrlChan node) NCMsg
224+ { ctrlMsgSender = ProcessIdentifier pid
225+ , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
226+ }
227+ return (tid', lproc)
228+
229+ if lpidCounter lpid == maxBound
230+ then do
231+ newUnique <- randomIO
232+ return ( (localProcessWithId lpid ^= Just lproc)
233+ . (localPidCounter ^= 0)
234+ . (localPidUnique ^= newUnique)
235+ $ st
236+ , pid
237+ )
238+ else
239+ return ( (localProcessWithId lpid ^= Just lproc)
240+ . (localPidCounter ^: (+ 1))
241+ $ st
242+ , pid
243+ )
244+
245+ cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState
246+ cleanupProcess pid st = do
247+ let pid' = ProcessIdentifier pid
248+ let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)
249+ mapM_ NT.close (Map.elems affected)
250+ return $ (localProcessWithId (processLocalId pid) ^= Nothing)
251+ . (localConnections ^= unaffected)
235252 $ st
236− , pid
237− )
238253
239254handleIncomingMessages :: LocalNode -> IO ()
240255handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty