119c411 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Support linking

119c411Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+59−41

compare with master

unifiedsplit

@@ -65,19 +65,23 @@ module Control.Distributed.Process

65 , getSelfPid

66 -- * Monitoring and linking

67 , monitor

68− , MonitorReply(..)

68+ , MonitorNotification(..)

69+ , MonitorRef

70+ , LinkException(..)

69 , DiedReason(..)

71 , DiedReason(..)

72+ , link

70 -- * Initialization

73 -- * Initialization

71 , newLocalNode

74 , newLocalNode

72 , forkProcess

75 , forkProcess

73 , runProcess

76 , runProcess

74 -- * Auxiliary API

77 -- * Auxiliary API

75 , closeLocalNode

78 , closeLocalNode

76− , pcatch

79+ , catch

77 , expectTimeout

80 , expectTimeout

78 ) where

81 ) where

80import Prelude hiding (catch)

83import Prelude hiding (catch)

84+import System.IO (fixIO)

81import qualified Data.ByteString as BSS (ByteString)

85import qualified Data.ByteString as BSS (ByteString)

82import qualified Data.ByteString.Lazy as BSL (fromChunks)

86import qualified Data.ByteString.Lazy as BSL (fromChunks)

83import Data.Binary (decode)

87import Data.Binary (decode)

@@ -91,7 +95,8 @@ import Control.Monad (void)

91import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

95import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

92import Control.Monad.IO.Class (MonadIO, liftIO)

96import Control.Monad.IO.Class (MonadIO, liftIO)

93import Control.Applicative ((<$>))

97import Control.Applicative ((<$>))

94−import Control.Exception (Exception, throwIO, catch, SomeException)

98+import Control.Exception (Exception, throwIO, SomeException)

99+import qualified Control.Exception as Exception (catch)

95import Control.Concurrent (forkIO)

100import Control.Concurrent (forkIO)

96import Control.Concurrent.MVar ( newMVar

101import Control.Concurrent.MVar ( newMVar

97 , withMVar

102 , withMVar

@@ -136,7 +141,8 @@ import Control.Distributed.Process.Internal ( NodeId(..)

136 , LocalProcessState(..)

141 , LocalProcessState(..)

137 , Message(..)

142 , Message(..)

138 , MonitorRef

143 , MonitorRef

139− , MonitorReply(..)

144+ , MonitorNotification(..)

145+ , LinkException(..)

140 , DiedReason(..)

146 , DiedReason(..)

141 , NCMsg(..)

147 , NCMsg(..)

142 , ProcessSignal(..)

148 , ProcessSignal(..)

@@ -186,13 +192,19 @@ monitor :: ProcessId -> Process MonitorRef

186monitor them = do

192monitor them = do

187 us <- getSelfPid

193 us <- getSelfPid

188 monitorRef <- getMonitorRef

194 monitorRef <- getMonitorRef

189− ourCtrlChan <- localCtrlChan . processNode <$> ask

190− liftIO . writeChan ourCtrlChan $ NCMsg

191− { ctrlMsgSender = Left us

195+ postCtrlMsg NCMsg { ctrlMsgSender = Left us

192− , ctrlMsgSignal = Monitor them monitorRef

196+ , ctrlMsgSignal = Monitor them monitorRef

193− }

197+ }

194 return monitorRef

198 return monitorRef

195

199

200+-- | Link to a remote process

201+link :: ProcessId -> Process ()

202+link them = do

203+ us <- getSelfPid

204+ postCtrlMsg NCMsg { ctrlMsgSender = Left us

205+ , ctrlMsgSignal = Link them

206+ }

207+

196--------------------------------------------------------------------------------

208--------------------------------------------------------------------------------

197-- Auxiliary API --

209-- Auxiliary API --

198--------------------------------------------------------------------------------

210--------------------------------------------------------------------------------

@@ -205,11 +217,10 @@ closeLocalNode node =

205 NT.closeEndPoint (localEndPoint node)

217 NT.closeEndPoint (localEndPoint node)

206

218

207-- Catch exceptions within a process

219-- Catch exceptions within a process

208−-- TODO: should this be called simply 'catch'?

209−pcatch :: Exception e => Process a -> (e -> Process a) -> Process a

220+catch :: Exception e => Process a -> (e -> Process a) -> Process a

210−pcatch p h = do

221+catch p h = do

211 run <- runLocalProcess <$> ask

222 run <- runLocalProcess <$> ask

212− liftIO $ catch (run p) (run . h)

223+ liftIO $ Exception.catch (run p) (run . h)

213

224

214expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)

225expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)

215expectTimeout timeout = do

226expectTimeout timeout = do

@@ -253,33 +264,27 @@ runProcess node proc = do

253 takeMVar done

264 takeMVar done

254

265

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

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

256−forkProcess node proc = do

257− queue <- newCQueue

258− lproc <- modifyMVar (localState node) $ \st -> do

267+forkProcess node proc = modifyMVar (localState node) $ \st -> do

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

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

260− , lpidUnique = st ^. localPidUnique

269+ , lpidUnique = st ^. localPidUnique

261− }

270+ }

262− let pid = ProcessId { processNodeId = localNodeId node

271+ let pid = ProcessId { processNodeId = localNodeId node

263− , processLocalId = lpid

272+ , processLocalId = lpid

264− }

273+ }

265− pst <- newMVar LocalProcessState { _connections = Map.empty

274+ pst <- newMVar LocalProcessState { _connections = Map.empty

266− }

275+ }

276+ queue <- newCQueue

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

267− let lproc = LocalProcess { processQueue = queue

278+ let lproc = LocalProcess { processQueue = queue

268− , processNode = node

279+ , processNode = node

269− , processId = pid

280+ , processId = pid

270− , processState = pst

281+ , processState = pst

271− }

282+ , processThread = tid

272− -- TODO: if the counter overflows we should pick a new unique

283+ }

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

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

275− $ st

276− , lproc

277− )

278− void . forkIO $ do

284+ tid' <- forkIO $ do

279− let pid = processId lproc

285+ reason <- Exception.catch

280− lpid = processLocalId pid

281− reason <- catch (runLocalProcess lproc proc >> return DiedNormal)

286+ (runLocalProcess lproc proc >> return DiedNormal)

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

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

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

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

284 modifyMVar_ (localState node) $

289 modifyMVar_ (localState node) $

285 return . (localProcessWithId lpid ^= Nothing)

290 return . (localProcessWithId lpid ^= Nothing)

@@ -287,7 +292,14 @@ forkProcess node proc = do

287 { ctrlMsgSender = Left pid

292 { ctrlMsgSender = Left pid

288 , ctrlMsgSignal = Died (Left pid) reason

293 , ctrlMsgSignal = Died (Left pid) reason

289 }

294 }

290− return (processId lproc)

295+ return (tid', lproc)

296+

297+ -- TODO: if the counter overflows we should pick a new unique

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

299+ . (localPidCounter ^: (+ 1))

300+ $ st

301+ , pid

302+ )

291

303

292handleIncomingMessages :: LocalNode -> IO ()

304handleIncomingMessages :: LocalNode -> IO ()

293handleIncomingMessages node = go [] Map.empty Set.empty

305handleIncomingMessages node = go [] Map.empty Set.empty

@@ -424,3 +436,9 @@ getMonitorRef = do

424-- This is most definitely NOT exported

436-- This is most definitely NOT exported

425runLocalProcess :: LocalProcess -> Process a -> IO a

437runLocalProcess :: LocalProcess -> Process a -> IO a

426runLocalProcess lproc proc = runReaderT (unProcess proc) lproc

438runLocalProcess lproc proc = runReaderT (unProcess proc) lproc

439+

440+-- | Post a control message on the local node controller

441+postCtrlMsg :: NCMsg -> Process ()

442+postCtrlMsg msg = do

443+ ctrlChan <- localCtrlChan . processNode <$> ask

444+ liftIO $ writeChan ctrlChan msg

distributed-process/src/Control/Distributed/Process/Internal.hs+25−12

@@ -10,9 +10,10 @@ module Control.Distributed.Process.Internal

10 , LocalProcessState(..)

11 -- * Messages

12 , Message(..)

13− -- * Monitoring

13+ -- * Monitoring and linking

14 , MonitorRef

15− , MonitorReply(..)

15+ , MonitorNotification(..)

16+ , LinkException(..)

16 , DiedReason(..)

17 , DiedReason(..)

17 -- * Node controller data types

18 -- * Node controller data types

18 , Identifier

19 , Identifier

@@ -34,9 +35,11 @@ module Control.Distributed.Process.Internal

34 , connectionTo

35 , connectionTo

35 ) where

36 ) where

38+import Control.Concurrent (ThreadId)

37import Control.Concurrent.MVar (MVar)

39import Control.Concurrent.MVar (MVar)

38import Control.Concurrent.Chan (Chan)

40import Control.Concurrent.Chan (Chan)

39import Control.Category ((>>>))

41import Control.Category ((>>>))

42+import Control.Exception (Exception)

40import Data.Map (Map)

43import Data.Map (Map)

41import Data.Int (Int32)

44import Data.Int (Int32)

42import Data.Typeable (Typeable)

45import Data.Typeable (Typeable)

@@ -95,10 +98,11 @@ data LocalNodeState = LocalNodeState

96-- | Processes running on our local node

99-- | Processes running on our local node

97data LocalProcess = LocalProcess

100data LocalProcess = LocalProcess

98− { processQueue :: CQueue Message

101+ { processQueue :: CQueue Message

99− , processNode :: LocalNode

102+ , processNode :: LocalNode

100− , processId :: ProcessId

103+ , processId :: ProcessId

101− , processState :: MVar LocalProcessState

104+ , processState :: MVar LocalProcessState

105+ , processThread :: ThreadId

102 }

106 }

103

107

104-- | Local process state

108-- | Local process state

@@ -116,9 +120,15 @@ data Message = Message

116type MonitorRef = Int32

120type MonitorRef = Int32

117

121

118-- | Messages sent by monitors

122-- | Messages sent by monitors

119−data MonitorReply = ProcessDied MonitorRef ProcessId DiedReason

123+data MonitorNotification = MonitorNotification MonitorRef ProcessId DiedReason

120 deriving (Typeable)

124 deriving (Typeable)

121

125

126+-- | Exceptions thrown when a linked process dies

127+data LinkException = LinkException ProcessId DiedReason

128+ deriving (Typeable, Show)

129+

130+instance Exception LinkException

131+

122-- | Why did a process die?

132-- | Why did a process die?

123data DiedReason =

133data DiedReason =

124 DiedNormal

134 DiedNormal

@@ -140,6 +150,7 @@ data NCMsg = NCMsg

140-- | Signals to the node controller (see 'NCMsg')

150-- | Signals to the node controller (see 'NCMsg')

141data ProcessSignal =

151data ProcessSignal =

142 Monitor ProcessId MonitorRef

152 Monitor ProcessId MonitorRef

153+ | Link ProcessId

143 | Died Identifier DiedReason

154 | Died Identifier DiedReason

144

155

145--------------------------------------------------------------------------------

156--------------------------------------------------------------------------------

@@ -192,9 +203,9 @@ instance Binary ProcessId where

192 put pid = put (processNodeId pid) >> put (processLocalId pid)

203 put pid = put (processNodeId pid) >> put (processLocalId pid)

193 get = ProcessId <$> get <*> get

204 get = ProcessId <$> get <*> get

194

205

195−instance Binary MonitorReply where

206+instance Binary MonitorNotification where

196− put (ProcessDied ref pid reason) = put ref >> put pid >> put reason

207+ put (MonitorNotification ref pid reason) = put ref >> put pid >> put reason

197− get = ProcessDied <$> get <*> get <*> get

208+ get = MonitorNotification <$> get <*> get <*> get

198

209

199instance Binary NCMsg where

210instance Binary NCMsg where

200 put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)

211 put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)

@@ -202,12 +213,14 @@ instance Binary NCMsg where

202

213

203instance Binary ProcessSignal where

214instance Binary ProcessSignal where

204 put (Monitor pid ref) = putWord8 0 >> put pid >> put ref

215 put (Monitor pid ref) = putWord8 0 >> put pid >> put ref

205− put (Died who reason) = putWord8 1 >> put who >> put reason

216+ put (Link pid) = putWord8 1 >> put pid

217+ put (Died who reason) = putWord8 2 >> put who >> put reason

206 get = do

218 get = do

207 header <- getWord8

219 header <- getWord8

208 case header of

220 case header of

209 0 -> Monitor <$> get <*> get

221 0 -> Monitor <$> get <*> get

222+ 1 -> Link <$> get

210− 1 -> Died <$> get <*> get

223+ 2 -> Died <$> get <*> get

211 _ -> fail "ProcessSignal.get: invalid"

224 _ -> fail "ProcessSignal.get: invalid"

212

225

213instance Binary DiedReason where

226instance Binary DiedReason where

distributed-process/src/Control/Distributed/Process/Internal/NodeController.hs+72−47

@@ -21,6 +21,7 @@ import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)

21import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

22import Control.Concurrent.MVar (withMVar)

23import Control.Concurrent.Chan (readChan, writeChan)

24+import Control.Exception (throwTo)

24import qualified Network.Transport as NT ( EndPoint

25import qualified Network.Transport as NT ( EndPoint

25 , Connection

26 , Connection

26 , Reliability(ReliableOrdered)

27 , Reliability(ReliableOrdered)

@@ -34,9 +35,9 @@ import Control.Distributed.Process.Internal ( NodeId(..)

34 , ProcessId(..)

35 , ProcessId(..)

35 , LocalNode(..)

36 , LocalNode(..)

36 , LocalProcess(..)

37 , LocalProcess(..)

37− , Message(..)

38 , MonitorRef

39− , MonitorReply(..)

39+ , MonitorNotification(..)

40+ , LinkException(..)

40 , DiedReason(..)

41 , DiedReason(..)

41 , Identifier

42 , Identifier

42 , NCMsg(..)

43 , NCMsg(..)

@@ -87,13 +88,26 @@ nodeController = do

87 Just nid' | nid' /= localNodeId node -> sendCtrlMsg nid' msg

88 Just nid' | nid' /= localNodeId node -> sendCtrlMsg nid' msg

88 _ -> return ()

89 _ -> return ()

90− ncEffect msg

91+ case msg of

91−

92+ NCMsg (Left from) (Monitor them ref) ->

92−-- [Unified: ncEffect]

93+ ncEffectMonitor from them (Just ref)

93−ncEffect :: NCMsg -> NC ()

94+ NCMsg (Right _) (Monitor _ _) ->

95+ error "Monitor message from a node?"

96+ NCMsg (Left from) (Link them) ->

97+ ncEffectMonitor from them Nothing

98+ NCMsg (Right _) (Link _) ->

99+ error "Link message from a node?"

100+ NCMsg _from (Died (Right nid) reason) ->

101+ ncEffectNodeDied nid reason

102+ NCMsg _from (Died (Left pid) reason) ->

103+ ncEffectProcessDied pid reason

104

95-- [Unified: Table 10]

105-- [Unified: Table 10]

96−ncEffect (NCMsg (Left from) (Monitor them ref)) = do

106+ncEffectMonitor :: ProcessId -- ^ Who's watching?

107+ -> ProcessId -- ^ Who's being watched?

108+ -> Maybe MonitorRef -- ^ 'Nothing' to link

109+ -> NC ()

110+ncEffectMonitor from them mRef = do

97 node <- ask

111 node <- ask

98 shouldLink <-

112 shouldLink <-

99 if processNodeId them /= localNodeId node

113 if processNodeId them /= localNodeId node

@@ -103,60 +117,52 @@ ncEffect (NCMsg (Left from) (Monitor them ref)) = do

103 let localFrom = processNodeId from == localNodeId node

117 let localFrom = processNodeId from == localNodeId node

104 case (shouldLink, localFrom) of

118 case (shouldLink, localFrom) of

105 (True, _) -> -- [Unified: first rule]

119 (True, _) -> -- [Unified: first rule]

120+ case mRef of

106− modify $ monitorsFor them from ^: Set.insert ref

121+ Just ref -> modify $ monitorsFor them from ^: Set.insert ref

122+ Nothing -> modify $ linksForProcess them ^: Set.insert from

107 (False, True) -> -- [Unified: second rule]

123 (False, True) -> -- [Unified: second rule]

108− sendLocal (processLocalId from) . createMessage $

109− ProcessDied ref them DiedNoProc

124+ notifyDied from them DiedNoProc mRef

110 (False, False) -> -- [Unified: third rule]

125 (False, False) -> -- [Unified: third rule]

111 sendCtrlMsg (processNodeId from) NCMsg

126 sendCtrlMsg (processNodeId from) NCMsg

112 { ctrlMsgSender = Right (localNodeId node)

127 { ctrlMsgSender = Right (localNodeId node)

113 , ctrlMsgSignal = Died (Left them) DiedNoProc

128 , ctrlMsgSignal = Died (Left them) DiedNoProc

114 }

129 }

115−

116−ncEffect (NCMsg (Right _) (Monitor _ _)) =

117− error "Monitor message from a node?"

118

130

119−-- [Unified: Table 12, bottom rule]

131+-- [Unified: Table 12, bottom rule]

120−ncEffect (NCMsg _from (Died (Right nid) reason)) = do

132+ncEffectNodeDied :: NodeId -> DiedReason -> NC ()

133+ncEffectNodeDied nid reason = do

121 node <- ask

134 node <- ask

122 lnks <- gets (^. linksForNode nid)

135 lnks <- gets (^. linksForNode nid)

123 mons <- gets (^. monitorsForNode nid)

136 mons <- gets (^. monitorsForNode nid)

124

137

138+ -- We only need to notify local processes

125− forM_ (Map.toList lnks) $ \(_them, _uss) ->

139+ forM_ (Map.toList lnks) $ \(them, uss) -> do

126− fail "ncEffect: linking not implemented"

140+ let pid = ProcessId nid them

141+ forM_ uss $ \us ->

142+ when (processNodeId us == localNodeId node) $

143+ notifyDied us pid reason Nothing

127

144

128− forM_ (Map.toList mons) $ \(them, uss) ->

145+ forM_ (Map.toList mons) $ \(them, uss) -> do

146+ let pid = ProcessId nid them

129 forM_ (Map.toList uss) $ \(us, refs) ->

147 forM_ (Map.toList uss) $ \(us, refs) ->

130− -- We only need to notify local processes

131− when (processNodeId us == localNodeId node) $ do

148+ when (processNodeId us == localNodeId node) $

132− let lpid = processLocalId us

149+ forM_ refs $ notifyDied us pid reason . Just

133− rpid = ProcessId nid them

134− forM_ refs $ \ref ->

135− sendLocal lpid . createMessage $ ProcessDied ref rpid reason

136

150

137 modify $ (linksForNode nid ^= Map.empty)

151 modify $ (linksForNode nid ^= Map.empty)

138 . (monitorsForNode nid ^= Map.empty)

152 . (monitorsForNode nid ^= Map.empty)

139

153

140-- [Unified: Table 12, top rule]

154-- [Unified: Table 12, top rule]

141−ncEffect (NCMsg _from (Died (Left pid) reason)) = do

155+ncEffectProcessDied :: ProcessId -> DiedReason -> NC ()

142− node <- ask

156+ncEffectProcessDied pid reason = do

143 lnks <- gets (^. linksForProcess pid)

157 lnks <- gets (^. linksForProcess pid)

144 mons <- gets (^. monitorsForProcess pid)

158 mons <- gets (^. monitorsForProcess pid)

145

159

146− forM_ lnks $ \_us ->

160+ forM_ lnks $ \us ->

147− fail "ncEffect: linking not implemented"

161+ notifyDied us pid reason Nothing

148

162

149 forM_ (Map.toList mons) $ \(us, refs) ->

163 forM_ (Map.toList mons) $ \(us, refs) ->

150− forM_ refs $ \ref -> do

164+ forM_ refs $

151− let msg = createMessage $ ProcessDied ref pid reason

165+ notifyDied us pid reason . Just

152− if processNodeId us == localNodeId node

153− then

154− sendLocal (processLocalId us) msg

155− else

156− sendCtrlMsg (processNodeId us) NCMsg

157− { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?

158− , ctrlMsgSignal = Died (Left pid) reason

159− }

160

166

161 modify $ (linksForProcess pid ^= Set.empty)

167 modify $ (linksForProcess pid ^= Set.empty)

162 . (monitorsForProcess pid ^= Map.empty)

168 . (monitorsForProcess pid ^= Map.empty)

@@ -165,6 +171,33 @@ ncEffect (NCMsg _from (Died (Left pid) reason)) = do

165-- Connections --

171-- Connections --

166--------------------------------------------------------------------------------

172--------------------------------------------------------------------------------

167

173

174+notifyDied :: ProcessId -- ^ Who to notify?

175+ -> ProcessId -- ^ Who died?

176+ -> DiedReason -- ^ How did they die?

177+ -> Maybe MonitorRef -- ^ 'Nothing' for linking

178+ -> NC ()

179+notifyDied dest src reason mRef = do

180+ node <- ask

181+ if processNodeId dest == localNodeId node

182+ then liftIO $ do

183+ let lpid = processLocalId dest

184+ mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)

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

186+ -- can silently be dropped

187+ forM_ mProc $ \proc -> case mRef of

188+ Just ref -> do

189+ let msg = createMessage $ MonitorNotification ref src reason

190+ enqueue (processQueue proc) msg

191+ Nothing ->

192+ throwTo (processThread proc) $ LinkException src reason

193+ else

194+ -- TODO: why the change in sender? How does that affect 'reconnect' semantics?

195+ -- (see [Unified: Table 10]

196+ sendCtrlMsg (processNodeId dest) NCMsg

197+ { ctrlMsgSender = Right (localNodeId node)

198+ , ctrlMsgSignal = Died (Left src) reason

199+ }

200+

168sendTo :: Identifier -> [BSS.ByteString] -> NC ()

201sendTo :: Identifier -> [BSS.ByteString] -> NC ()

169sendTo them payload = do

202sendTo them payload = do

170 mConn <- connTo them

203 mConn <- connTo them

@@ -187,15 +220,6 @@ sendTo them payload = do

187sendCtrlMsg :: NodeId -> NCMsg -> NC ()

220sendCtrlMsg :: NodeId -> NCMsg -> NC ()

188sendCtrlMsg dest = sendTo (Right dest) . BSL.toChunks . encode

221sendCtrlMsg dest = sendTo (Right dest) . BSL.toChunks . encode

189

222

190−sendLocal :: LocalProcessId -> Message -> NC ()

191−sendLocal lpid msg = do

192− node <- ask

193− liftIO $ do

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

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

196− -- can silently be dropped

197− forM_ mProc $ \proc -> enqueue (processQueue proc) msg

198−

199connTo :: Identifier -> NC (Maybe NT.Connection)

223connTo :: Identifier -> NC (Maybe NT.Connection)

200connTo them = do

224connTo them = do

201 mConn <- gets (^. connectionTo them)

225 mConn <- gets (^. connectionTo them)

@@ -236,6 +260,7 @@ createConnTo them = do

236

260

237destNid :: ProcessSignal -> Maybe NodeId

261destNid :: ProcessSignal -> Maybe NodeId

238destNid (Monitor pid _) = Just $ processNodeId pid

262destNid (Monitor pid _) = Just $ processNodeId pid

263+destNid (Link pid) = Just $ processNodeId pid

239destNid (Died (Right _) _) = Nothing

264destNid (Died (Right _) _) = Nothing

240destNid (Died (Left _pid) _) = fail "destNid: TODO"

265destNid (Died (Left _pid) _) = fail "destNid: TODO"

241

266

distributed-process/tests/TestCH.hs+89−66

@@ -12,6 +12,7 @@ import Control.Concurrent.MVar ( newEmptyMVar

12import Control.Monad (replicateM_)

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

14import Control.Exception (throwIO)

15+import Control.Applicative ((<$>))

15import qualified Network.Transport as NT (Transport, closeEndPoint)

16import qualified Network.Transport as NT (Transport, closeEndPoint)

16import Network.Transport.TCP (createTransport, defaultTCPParameters)

17import Network.Transport.TCP (createTransport, defaultTCPParameters)

17import Control.Distributed.Process

18import Control.Distributed.Process

@@ -62,10 +63,14 @@ testPing transport = do

63 takeMVar clientDone

64 takeMVar clientDone

66+-- | Monitor or link to a remote node

67+monitorOrLink :: Bool -> ProcessId -> Process (Maybe MonitorRef)

68+monitorOrLink True pid = Just <$> monitor pid

69+monitorOrLink False pid = link pid >> return Nothing

66-- | Monitor an unreachable node

71-- | Monitor an unreachable node

67−testMonitorUnreachable :: NT.Transport -> IO ()

72+testMonitorUnreachable :: NT.Transport -> Bool -> IO ()

68−testMonitorUnreachable transport = do

73+testMonitorUnreachable transport mOrL = do

69 deadProcess <- newEmptyMVar

74 deadProcess <- newEmptyMVar

70 done <- newEmptyMVar

75 done <- newEmptyMVar

@@ -78,18 +83,20 @@ testMonitorUnreachable transport = do

78 forkIO $ do

83 forkIO $ do

79 localNode <- newLocalNode transport

84 localNode <- newLocalNode transport

80 theirAddr <- readMVar deadProcess

85 theirAddr <- readMVar deadProcess

81− runProcess localNode $ do

86+ runProcess localNode $

82− ref <- monitor theirAddr

87+ catch (do ref <- monitorOrLink mOrL theirAddr

83− ProcessDied ref' pid DiedDisconnect <- expect

88+ MonitorNotification ref' pid DiedDisconnect <- expect

84− True <- return $ ref' == ref && pid == theirAddr

89+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL

85− return ()

90+ liftIO $ putMVar done ())

91+ (\(LinkException pid DiedDisconnect) -> do

92+ True <- return $ pid == theirAddr && not mOrL

86− putMVar done ()

93+ liftIO $ putMVar done ())

87−

94+

88 takeMVar done

95 takeMVar done

90-- | Monitor a process which terminates normally

97-- | Monitor a process which terminates normally

91−testMonitorNormalTermination :: NT.Transport -> IO ()

98+testMonitorNormalTermination :: NT.Transport -> Bool -> IO ()

92−testMonitorNormalTermination transport = do

99+testMonitorNormalTermination transport mOrL = do

93 monitorSetup <- newEmptyMVar

100 monitorSetup <- newEmptyMVar

94 monitoredProcess <- newEmptyMVar

101 monitoredProcess <- newEmptyMVar

95 done <- newEmptyMVar

102 done <- newEmptyMVar

@@ -103,24 +110,26 @@ testMonitorNormalTermination transport = do

103 forkIO $ do

110 forkIO $ do

104 localNode <- newLocalNode transport

111 localNode <- newLocalNode transport

105 theirAddr <- readMVar monitoredProcess

112 theirAddr <- readMVar monitoredProcess

106− runProcess localNode $ do

113+ runProcess localNode $

107− ref <- monitor theirAddr

114+ catch (do ref <- monitorOrLink mOrL theirAddr

108− liftIO $ do

115+ liftIO $ do

109− -- Monitor is asynchronous, but we want to make sure the monitor has

116+ -- Monitor is asynchronous, but we want to make sure the monitor has

110− -- been fully created before allowing the remote process to terminate,

117+ -- been fully created before allowing the remote process to terminate,

111− -- otherwise we might get a different signal here

118+ -- otherwise we might get a different signal here

112− threadDelay 100000

119+ threadDelay 100000

113− putMVar monitorSetup ()

120+ putMVar monitorSetup ()

114− ProcessDied ref' pid DiedNormal <- expect

121+ MonitorNotification ref' pid DiedNormal <- expect

115− True <- return $ ref' == ref && pid == theirAddr

122+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL

116− return ()

123+ liftIO $ putMVar done ())

124+ (\(LinkException pid DiedNormal) -> do

125+ True <- return $ pid == theirAddr && not mOrL

117− putMVar done ()

126+ liftIO $ putMVar done ())

118

127

119 takeMVar done

128 takeMVar done

120

129

121-- | Monitor a process which terminates abnormally

130-- | Monitor a process which terminates abnormally

122−testMonitorAbnormalTermination :: NT.Transport -> IO ()

131+testMonitorAbnormalTermination :: NT.Transport -> Bool -> IO ()

123−testMonitorAbnormalTermination transport = do

132+testMonitorAbnormalTermination transport mOrL = do

124 monitorSetup <- newEmptyMVar

133 monitorSetup <- newEmptyMVar

125 monitoredProcess <- newEmptyMVar

134 monitoredProcess <- newEmptyMVar

126 done <- newEmptyMVar

135 done <- newEmptyMVar

@@ -137,24 +146,26 @@ testMonitorAbnormalTermination transport = do

137 forkIO $ do

146 forkIO $ do

138 localNode <- newLocalNode transport

147 localNode <- newLocalNode transport

139 theirAddr <- readMVar monitoredProcess

148 theirAddr <- readMVar monitoredProcess

140− runProcess localNode $ do

149+ runProcess localNode $

141− ref <- monitor theirAddr

150+ catch (do ref <- monitorOrLink mOrL theirAddr

142− liftIO $ do

151+ liftIO $ do

143− -- Monitor is asynchronous, but we want to make sure the monitor has

152+ -- Monitor is asynchronous, but we want to make sure the monitor has

144− -- been fully created before allowing the remote process to terminate,

153+ -- been fully created before allowing the remote process to terminate,

145− -- otherwise we might get a different signal here

154+ -- otherwise we might get a different signal here

146− threadDelay 100000

155+ threadDelay 100000

147− putMVar monitorSetup ()

156+ putMVar monitorSetup ()

148− ProcessDied ref' pid (DiedException err') <- expect

157+ MonitorNotification ref' pid (DiedException err') <- expect

149− True <- return $ ref' == ref && pid == theirAddr && err' == show err

158+ True <- return $ Just ref' == ref && pid == theirAddr && err' == show err && mOrL

150− return ()

159+ liftIO $ putMVar done ())

160+ (\(LinkException pid (DiedException err')) -> do

161+ True <- return $ pid == theirAddr && err' == show err && not mOrL

151− putMVar done ()

162+ liftIO $ putMVar done ())

152

163

153 takeMVar done

164 takeMVar done

154

165

155-- | Monitor a local process that is already dead

166-- | Monitor a local process that is already dead

156−testMonitorLocalDeadProcess :: NT.Transport -> IO ()

167+testMonitorLocalDeadProcess :: NT.Transport -> Bool -> IO ()

157−testMonitorLocalDeadProcess transport = do

168+testMonitorLocalDeadProcess transport mOrL = do

158 processDead <- newEmptyMVar

169 processDead <- newEmptyMVar

159 processAddr <- newEmptyMVar

170 processAddr <- newEmptyMVar

160 localNode <- newLocalNode transport

171 localNode <- newLocalNode transport

@@ -168,17 +179,19 @@ testMonitorLocalDeadProcess transport = do

168 theirAddr <- readMVar processAddr

179 theirAddr <- readMVar processAddr

169 readMVar processDead

180 readMVar processDead

170 runProcess localNode $ do

181 runProcess localNode $ do

171− ref <- monitor theirAddr

182+ catch (do ref <- monitorOrLink mOrL theirAddr

172− ProcessDied ref' pid DiedNoProc <- expect

183+ MonitorNotification ref' pid DiedNoProc <- expect

173− True <- return $ ref' == ref && pid == theirAddr

184+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL

174− return ()

185+ liftIO $ putMVar done ())

186+ (\(LinkException pid DiedNoProc) -> do

187+ True <- return $ pid == theirAddr && not mOrL

175− putMVar done ()

188+ liftIO $ putMVar done ())

176

189

177 takeMVar done

190 takeMVar done

178

191

179-- | Monitor a remote process that is already dead

192-- | Monitor a remote process that is already dead

180−testMonitorRemoteDeadProcess :: NT.Transport -> IO ()

193+testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> IO ()

181−testMonitorRemoteDeadProcess transport = do

194+testMonitorRemoteDeadProcess transport mOrL = do

182 processDead <- newEmptyMVar

195 processDead <- newEmptyMVar

183 processAddr <- newEmptyMVar

196 processAddr <- newEmptyMVar

184 done <- newEmptyMVar

197 done <- newEmptyMVar

@@ -193,17 +206,19 @@ testMonitorRemoteDeadProcess transport = do

193 theirAddr <- readMVar processAddr

206 theirAddr <- readMVar processAddr

194 readMVar processDead

207 readMVar processDead

195 runProcess localNode $ do

208 runProcess localNode $ do

196− ref <- monitor theirAddr

209+ catch (do ref <- monitorOrLink mOrL theirAddr

197− ProcessDied ref' pid DiedNoProc <- expect

210+ MonitorNotification ref' pid DiedNoProc <- expect

198− True <- return $ ref' == ref && pid == theirAddr

211+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL

199− return ()

212+ liftIO $ putMVar done ())

213+ (\(LinkException pid DiedNoProc) -> do

214+ True <- return $ pid == theirAddr && not mOrL

200− putMVar done ()

215+ liftIO $ putMVar done ())

201

216

202 takeMVar done

217 takeMVar done

203

218

204-- | Monitor a process that becomes disconnected

219-- | Monitor a process that becomes disconnected

205−testMonitorDisconnect :: NT.Transport -> IO ()

220+testMonitorDisconnect :: NT.Transport -> Bool -> IO ()

206−testMonitorDisconnect transport = do

221+testMonitorDisconnect transport mOrL = do

207 processAddr <- newEmptyMVar

222 processAddr <- newEmptyMVar

208 monitorSetup <- newEmptyMVar

223 monitorSetup <- newEmptyMVar

209 done <- newEmptyMVar

224 done <- newEmptyMVar

@@ -219,13 +234,15 @@ testMonitorDisconnect transport = do

219 localNode <- newLocalNode transport

234 localNode <- newLocalNode transport

220 theirAddr <- readMVar processAddr

235 theirAddr <- readMVar processAddr

221 runProcess localNode $ do

236 runProcess localNode $ do

222− ref <- monitor theirAddr

237+ catch (do ref <- monitorOrLink mOrL theirAddr

223− liftIO $ threadDelay 100000 >> putMVar monitorSetup ()

238+ liftIO $ threadDelay 100000 >> putMVar monitorSetup ()

224− ProcessDied ref' pid DiedDisconnect <- expect

239+ MonitorNotification ref' pid DiedDisconnect <- expect

225− True <- return $ ref' == ref && pid == theirAddr

240+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL

226− return ()

241+ liftIO $ putMVar done ())

242+ (\(LinkException pid DiedDisconnect) -> do

243+ True <- return $ pid == theirAddr && not mOrL

227− putMVar done ()

244+ liftIO $ putMVar done ())

228−

245+

229 takeMVar done

246 takeMVar done

230

247

231

248

@@ -336,12 +353,18 @@ main = do

336 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

353 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

337 runTests

354 runTests

338 [ --("Ping", testPing transport)

355 [ --("Ping", testPing transport)

339− ("MonitorUnreachable", testMonitorUnreachable transport)

356+ ("MonitorUnreachable", testMonitorUnreachable transport True)

340− , ("MonitorNormalTermination", testMonitorNormalTermination transport)

357+ , ("MonitorNormalTermination", testMonitorNormalTermination transport True)

341− , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport)

358+ , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport True)

342− , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport)

359+ , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport True)

343− , ("MonitorRemoteDeadProcess", testMonitorRemoteDeadProcess transport)

360+ , ("MonitorRemoteDeadProcess", testMonitorRemoteDeadProcess transport True)

344− , ("MonitorDisconnect", testMonitorDisconnect transport)

361+ , ("MonitorDisconnect", testMonitorDisconnect transport True)

362+ , ("LinkUnreachable", testMonitorUnreachable transport False)

363+ , ("LinkNormalTermination", testMonitorNormalTermination transport False)

364+ , ("LinkAbnormalTermination", testMonitorAbnormalTermination transport False)

365+ , ("LinkLocalDeadProcess", testMonitorLocalDeadProcess transport False)

366+ , ("LinkRemoteDeadProcess", testMonitorRemoteDeadProcess transport False)

367+ , ("LinkDisconnect", testMonitorDisconnect transport False)

345 {-

368 {-

346 , ("Monitor2", testMonitor2 transport)

369 , ("Monitor2", testMonitor2 transport)

347 , ("Monitor3", testMonitor3 transport)

370 , ("Monitor3", testMonitor3 transport)