2b1a8d9 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Close connections on process exit

2b1a8d9Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process/Node.hs+62−47

compare with master

unifiedsplit

@@ -25,6 +25,7 @@ import qualified Data.Map as Map

25 , toList

26 , partitionWithKey

27 , filterWithKey

28+ , elems

28 )

29 )

29import Data.Set (Set)

30import Data.Set (Set)

30import qualified Data.Set as Set (empty, insert, delete, member, filter)

31import qualified Data.Set as Set (empty, insert, delete, member, filter)

@@ -64,6 +65,8 @@ import qualified Network.Transport as NT

64 , address

65 , address

65 , closeEndPoint

66 , closeEndPoint

66 , ConnectionId

67 , ConnectionId

68+ , Connection

69+ , close

67 )

70 )

68import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

71import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

69import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe)

72import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe)

@@ -88,6 +91,7 @@ import Control.Distributed.Process.Internal.Types

88 , localPidCounter

91 , localPidCounter

89 , localPidUnique

92 , localPidUnique

90 , localProcessWithId

93 , localProcessWithId

94+ , localConnections

91 , MonitorRef(..)

95 , MonitorRef(..)

92 , ProcessMonitorNotification(..)

96 , ProcessMonitorNotification(..)

93 , NodeMonitorNotification(..)

97 , NodeMonitorNotification(..)

@@ -187,54 +191,65 @@ runProcess node proc = do

187

191

188-- | Spawn a new process on a local node

192-- | Spawn a new process on a local node

189forkProcess :: LocalNode -> Process () -> IO ProcessId

193forkProcess :: LocalNode -> Process () -> IO ProcessId

190−forkProcess node proc = modifyMVar (localState node) $ \st -> do

194+forkProcess node proc = modifyMVar (localState node) startProcess

195+ where

196+ startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)

197+ startProcess st = do

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

198+ let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter

192− , lpidUnique = st ^. localPidUnique

199+ , lpidUnique = st ^. localPidUnique

193− }

200+ }

194− let pid = ProcessId { processNodeId = localNodeId node

201+ let pid = ProcessId { processNodeId = localNodeId node

195− , processLocalId = lpid

202+ , processLocalId = lpid

196− }

203+ }

197− pst <- newMVar LocalProcessState { _monitorCounter = 0

204+ pst <- newMVar LocalProcessState { _monitorCounter = 0

198− , _spawnCounter = 0

205+ , _spawnCounter = 0

199− , _channelCounter = 0

206+ , _channelCounter = 0

200− , _typedChannels = Map.empty

207+ , _typedChannels = Map.empty

201− }

208+ }

202− queue <- newCQueue

209+ queue <- newCQueue

203− (_, lproc) <- fixIO $ \ ~(tid, _) -> do

210+ (_, lproc) <- fixIO $ \ ~(tid, _) -> do

204− let lproc = LocalProcess { processQueue = queue

211+ let lproc = LocalProcess { processQueue = queue

205− , processId = pid

212+ , processId = pid

206− , processState = pst

213+ , processState = pst

207− , processThread = tid

214+ , processThread = tid

208− , processNode = node

215+ , processNode = node

209− }

216+ }

210− tid' <- forkIO $ do

217+ tid' <- forkIO $ do

211− reason <- Exception.catch

218+ reason <- Exception.catch

212− (runLocalProcess lproc proc >> return DiedNormal)

219+ (runLocalProcess lproc proc >> return DiedNormal)

213− (return . DiedException . (show :: SomeException -> String))

220+ (return . DiedException . (show :: SomeException -> String))

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

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

215− modifyMVar_ (localState node) $

222+ modifyMVar_ (localState node) (cleanupProcess pid)

216− return . (localProcessWithId lpid ^= Nothing)

217− writeChan (localCtrlChan node) NCMsg

223+ writeChan (localCtrlChan node) NCMsg

218− { ctrlMsgSender = ProcessIdentifier pid

224+ { ctrlMsgSender = ProcessIdentifier pid

219− , ctrlMsgSignal = Died (ProcessIdentifier pid) reason

225+ , ctrlMsgSignal = Died (ProcessIdentifier pid) reason

220− }

226+ }

221− return (tid', lproc)

227+ return (tid', lproc)

222−

228+

223− if lpidCounter lpid == maxBound

229+ if lpidCounter lpid == maxBound

224− then do

230+ then do

225− newUnique <- randomIO

231+ newUnique <- randomIO

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

232+ return ( (localProcessWithId lpid ^= Just lproc)

227− . (localPidCounter ^= 0)

233+ . (localPidCounter ^= 0)

228− . (localPidUnique ^= newUnique)

234+ . (localPidUnique ^= newUnique)

229− $ st

235+ $ st

230− , pid

236+ , pid

231− )

237+ )

232− else

238+ else

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

239+ return ( (localProcessWithId lpid ^= Just lproc)

234− . (localPidCounter ^: (+ 1))

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)

235 $ st

252 $ st

236− , pid

237− )

238

253

239handleIncomingMessages :: LocalNode -> IO ()

254handleIncomingMessages :: LocalNode -> IO ()

240handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty

255handleIncomingMessages node = go Set.empty Map.empty Map.empty Set.empty