05d2ebf · distributed-process

git.iko.soy/distributed-process

on master help sign in

Support unlink/unmonitor

05d2ebfEdsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+34−22

compare with master

unifiedsplit

@@ -64,12 +64,16 @@ module Control.Distributed.Process

64 , send

65 , getSelfPid

66 -- * Monitoring and linking

67+ , link

68+ , unlink

67 , monitor

69 , monitor

68− , MonitorNotification(..)

70+ , unmonitor

69− , MonitorRef

70 , LinkException(..)

71 , LinkException(..)

72+ , MonitorRef -- opaque

73+ , MonitorNotification(..)

71 , DiedReason(..)

74 , DiedReason(..)

72− , link

75+ , DidUnmonitor(..)

76+ , DidUnlink(..)

73 -- * Initialization

77 -- * Initialization

74 , newLocalNode

78 , newLocalNode

75 , forkProcess

79 , forkProcess

@@ -140,9 +144,11 @@ import Control.Distributed.Process.Internal ( NodeId(..)

140 , LocalProcess(..)

144 , LocalProcess(..)

141 , LocalProcessState(..)

145 , LocalProcessState(..)

142 , Message(..)

146 , Message(..)

143− , MonitorRef

147+ , MonitorRef(..)

144 , MonitorNotification(..)

148 , MonitorNotification(..)

145 , LinkException(..)

149 , LinkException(..)

150+ , DidUnmonitor(..)

151+ , DidUnlink(..)

146 , DiedReason(..)

152 , DiedReason(..)

147 , NCMsg(..)

153 , NCMsg(..)

148 , ProcessSignal(..)

154 , ProcessSignal(..)

@@ -190,20 +196,21 @@ getSelfPid = processId <$> ask

190-- | Monitor another process

196-- | Monitor another process

191monitor :: ProcessId -> Process MonitorRef

197monitor :: ProcessId -> Process MonitorRef

192monitor them = do

198monitor them = do

193− us <- getSelfPid

199+ monitorRef <- getMonitorRefFor them

194− monitorRef <- getMonitorRef

195− postCtrlMsg NCMsg { ctrlMsgSender = Left us

196− , ctrlMsgSignal = Monitor them monitorRef

200+ postCtrlMsg $ Monitor them monitorRef

197− }

198 return monitorRef

201 return monitorRef

199

202

200-- | Link to a remote process

203-- | Link to a remote process

201link :: ProcessId -> Process ()

204link :: ProcessId -> Process ()

202−link them = do

203− us <- getSelfPid

204− postCtrlMsg NCMsg { ctrlMsgSender = Left us

205+link = postCtrlMsg . Link

205− , ctrlMsgSignal = Link them

206+

206− }

207+-- | Remove a monitor

208+unmonitor :: MonitorRef -> Process ()

209+unmonitor = postCtrlMsg . Unmonitor

210+

211+-- | Remove a link

212+unlink :: ProcessId -> Process ()

213+unlink = postCtrlMsg . Unlink

207

214

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

215--------------------------------------------------------------------------------

209-- Auxiliary API --

216-- Auxiliary API --

@@ -319,8 +326,9 @@ handleIncomingMessages node = go [] Map.empty Set.empty

319 let msg = payloadToMessage payload

326 let msg = payloadToMessage payload

320 enqueue (processQueue proc) msg

327 enqueue (processQueue proc) msg

321 go uninitConns procConns ctrlConns

328 go uninitConns procConns ctrlConns

322− Nothing | cid `Set.member` ctrlConns ->

329+ Nothing | cid `Set.member` ctrlConns -> do

323 writeChan ctrlChan (decode . BSL.fromChunks $ payload)

330 writeChan ctrlChan (decode . BSL.fromChunks $ payload)

331+ go uninitConns procConns ctrlConns

324 Nothing | cid `elem` uninitConns ->

332 Nothing | cid `elem` uninitConns ->

325 case payloadToId payload of

333 case payloadToId payload of

326 Just lpid -> do

334 Just lpid -> do

@@ -425,12 +433,13 @@ remoteProcessFailed node them = do

425 , ctrlMsgSignal = Died nid DiedDisconnect

433 , ctrlMsgSignal = Died nid DiedDisconnect

426 }

434 }

427

435

428−getMonitorRef :: Process MonitorRef

436+getMonitorRefFor :: ProcessId -> Process MonitorRef

429−getMonitorRef = do

437+getMonitorRefFor pid = do

430 node <- processNode <$> ask

438 node <- processNode <$> ask

431− liftIO $ modifyMVar (localState node) $ \st ->

439+ liftIO $ modifyMVar (localState node) $ \st -> do

440+ let counter = st ^. localMonitorCounter

432 return ( localMonitorCounter ^: (+ 1) $ st

441 return ( localMonitorCounter ^: (+ 1) $ st

433− , st ^. localMonitorCounter

442+ , MonitorRef pid counter

434 )

443 )

435

444

436-- This is most definitely NOT exported

445-- This is most definitely NOT exported

@@ -438,7 +447,10 @@ runLocalProcess :: LocalProcess -> Process a -> IO a

438runLocalProcess lproc proc = runReaderT (unProcess proc) lproc

447runLocalProcess lproc proc = runReaderT (unProcess proc) lproc

439

448

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

449-- | Post a control message on the local node controller

441−postCtrlMsg :: NCMsg -> Process ()

450+postCtrlMsg :: ProcessSignal -> Process ()

442−postCtrlMsg msg = do

451+postCtrlMsg signal = do

452+ us <- getSelfPid

443 ctrlChan <- localCtrlChan . processNode <$> ask

453 ctrlChan <- localCtrlChan . processNode <$> ask

444− liftIO $ writeChan ctrlChan msg

454+ liftIO $ writeChan ctrlChan NCMsg { ctrlMsgSender = Left us

455+ , ctrlMsgSignal = signal

456+ }

distributed-process/src/Control/Distributed/Process/Internal.hs+50−15

@@ -11,10 +11,12 @@ module Control.Distributed.Process.Internal

11 -- * Messages

12 , Message(..)

13 -- * Monitoring and linking

14− , MonitorRef

14+ , MonitorRef(..)

15 , MonitorNotification(..)

16 , LinkException(..)

17 , DiedReason(..)

18+ , DidUnmonitor(..)

19+ , DidUnlink(..)

18 -- * Node controller data types

20 -- * Node controller data types

19 , Identifier

21 , Identifier

20 , NCMsg(..)

22 , NCMsg(..)

@@ -62,7 +64,10 @@ import Control.Distributed.Process.Internal.CQueue (CQueue)

63-- We identify node IDs and endpoint IDs

65-- We identify node IDs and endpoint IDs

64newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }

66newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }

65− deriving (Show, Eq, Ord, Binary)

67+ deriving (Eq, Ord, Binary)

68+

69+instance Show NodeId where

70+ show = show . nodeAddress

67-- | A local process ID consists of a seed which distinguishes processes from

72-- | A local process ID consists of a seed which distinguishes processes from

68-- different instances of the same local node and a counter

73-- different instances of the same local node and a counter

@@ -70,7 +75,10 @@ data LocalProcessId = LocalProcessId

70 { lpidUnique :: Int32

75 { lpidUnique :: Int32

71 , lpidCounter :: Int32

76 , lpidCounter :: Int32

72 }

77 }

73− deriving (Eq, Ord, Typeable, Show)

78+ deriving (Eq, Ord, Typeable)

79+

80+instance Show LocalProcessId where

81+ show = show . lpidCounter

75-- | A process ID combines a local process with with an endpoint address

83-- | A process ID combines a local process with with an endpoint address

76-- (in other words, we identify nodes and endpoints)

84-- (in other words, we identify nodes and endpoints)

@@ -78,7 +86,10 @@ data ProcessId = ProcessId

78 { processNodeId :: NodeId

86 { processNodeId :: NodeId

79 , processLocalId :: LocalProcessId

87 , processLocalId :: LocalProcessId

80 }

88 }

81− deriving (Eq, Ord, Typeable, Show)

89+ deriving (Eq, Ord, Typeable)

90+

91+instance Show ProcessId where

92+ show pid = show (processNodeId pid) ++ ":" ++ show (processLocalId pid)

83-- | Local nodes

94-- | Local nodes

84data LocalNode = LocalNode

95data LocalNode = LocalNode

@@ -116,8 +127,12 @@ data Message = Message

116 , messageEncoding :: BSL.ByteString

127 , messageEncoding :: BSL.ByteString

117 }

128 }

118

129

119−-- | MonitorRef should be an opaque type

130+-- | MonitorRef is opaque for regular Cloud Haskell processes

120−type MonitorRef = Int32

131+data MonitorRef = MonitorRef

132+ { monitorRefPid :: ProcessId

133+ , monitorRefCounter :: Int32

134+ }

135+ deriving (Eq, Ord, Show)

121

136

122-- | Messages sent by monitors

137-- | Messages sent by monitors

123data MonitorNotification = MonitorNotification MonitorRef ProcessId DiedReason

138data MonitorNotification = MonitorNotification MonitorRef ProcessId DiedReason

@@ -129,6 +144,14 @@ data LinkException = LinkException ProcessId DiedReason

129

144

130instance Exception LinkException

145instance Exception LinkException

131

146

147+-- | (Asynchronous) reply from unmonitor

148+newtype DidUnmonitor = DidUnmonitor MonitorRef

149+ deriving (Typeable, Binary)

150+

151+-- | (Asynchronous) reply from unlink

152+newtype DidUnlink = DidUnlink ProcessId

153+ deriving (Typeable, Binary)

154+

132-- | Why did a process die?

155-- | Why did a process die?

133data DiedReason =

156data DiedReason =

134 DiedNormal

157 DiedNormal

@@ -136,7 +159,7 @@ data DiedReason =

136 | DiedDisconnect

159 | DiedDisconnect

137 | DiedNodeDown

160 | DiedNodeDown

138 | DiedNoProc

161 | DiedNoProc

139− deriving Show

162+ deriving (Show, Eq)

140

163

141-- | Node or process identifier

164-- | Node or process identifier

142type Identifier = Either ProcessId NodeId

165type Identifier = Either ProcessId NodeId

@@ -146,12 +169,16 @@ data NCMsg = NCMsg

146 { ctrlMsgSender :: Identifier

169 { ctrlMsgSender :: Identifier

147 , ctrlMsgSignal :: ProcessSignal

170 , ctrlMsgSignal :: ProcessSignal

148 }

171 }

172+ deriving Show

149

173

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

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

151data ProcessSignal =

175data ProcessSignal =

176+ Link ProcessId

177+ | Unlink ProcessId

152− Monitor ProcessId MonitorRef

178+ | Monitor ProcessId MonitorRef

153− | Link ProcessId

179+ | Unmonitor MonitorRef

154 | Died Identifier DiedReason

180 | Died Identifier DiedReason

181+ deriving Show

155

182

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

183--------------------------------------------------------------------------------

157-- Serialization/deserialization --

184-- Serialization/deserialization --

@@ -211,16 +238,24 @@ instance Binary NCMsg where

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

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

212 get = NCMsg <$> get <*> get

239 get = NCMsg <$> get <*> get

213

240

241+instance Binary MonitorRef where

242+ put ref = put (monitorRefPid ref) >> put (monitorRefCounter ref)

243+ get = MonitorRef <$> get <*> get

244+

214instance Binary ProcessSignal where

245instance Binary ProcessSignal where

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

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

246+ put (Link pid) = putWord8 0 >> put pid

247+ put (Unlink pid) = putWord8 1 >> put pid

248+ put (Monitor pid ref) = putWord8 2 >> put pid >> put ref

249+ put (Unmonitor ref) = putWord8 3 >> put ref

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

250+ put (Died who reason) = putWord8 4 >> put who >> put reason

218 get = do

251 get = do

219 header <- getWord8

252 header <- getWord8

220 case header of

253 case header of

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

254+ 0 -> Link <$> get

222− 1 -> Link <$> get

255+ 1 -> Unlink <$> get

256+ 2 -> Monitor <$> get <*> get

257+ 3 -> Unmonitor <$> get

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

258+ 4 -> Died <$> get <*> get

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

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

225

260

226instance Binary DiedReason where

261instance Binary DiedReason where

distributed-process/src/Control/Distributed/Process/Internal/NodeController.hs+71−29

@@ -6,7 +6,7 @@ module Control.Distributed.Process.Internal.NodeController

6import Data.Map (Map)

7import qualified Data.Map as Map (empty, toList)

8import Data.Set (Set)

9−import qualified Data.Set as Set (empty, insert)

9+import qualified Data.Set as Set (empty, insert, delete)

10import qualified Data.ByteString as BSS (ByteString)

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

12import Data.Binary (encode)

@@ -21,7 +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)

24+import Control.Exception (Exception, throwTo)

25import qualified Network.Transport as NT ( EndPoint

26 , Connection

27 , Reliability(ReliableOrdered)

@@ -29,15 +29,18 @@ import qualified Network.Transport as NT ( EndPoint

29 , send

30 , connect

31 )

32+import Control.Distributed.Process.Serializable (Serializable)

32import Control.Distributed.Process.Internal.CQueue (enqueue)

33import Control.Distributed.Process.Internal.CQueue (enqueue)

33import Control.Distributed.Process.Internal ( NodeId(..)

34import Control.Distributed.Process.Internal ( NodeId(..)

34 , LocalProcessId(..)

35 , LocalProcessId(..)

35 , ProcessId(..)

36 , ProcessId(..)

36 , LocalNode(..)

37 , LocalNode(..)

37 , LocalProcess(..)

38 , LocalProcess(..)

38− , MonitorRef

39+ , MonitorRef(..)

39 , MonitorNotification(..)

40 , MonitorNotification(..)

40 , LinkException(..)

41 , LinkException(..)

42+ , DidUnmonitor(..)

43+ , DidUnlink(..)

41 , DiedReason(..)

44 , DiedReason(..)

42 , Identifier

45 , Identifier

43 , NCMsg(..)

46 , NCMsg(..)

@@ -89,18 +92,27 @@ nodeController = do

89 _ -> return ()

92 _ -> return ()

91 case msg of

94 case msg of

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

93− ncEffectMonitor from them (Just ref)

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

95− error "Monitor message from a node?"

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

95 NCMsg (Left from) (Link them) ->

97 ncEffectMonitor from them Nothing

96 ncEffectMonitor from them Nothing

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

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

99− error "Link message from a node?"

98+ ncEffectMonitor from them (Just ref)

99+ NCMsg (Left from) (Unlink them) ->

100+ ncEffectUnlink from them

101+ NCMsg (Left from) (Unmonitor ref) ->

102+ ncEffectUnmonitor from ref

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

103 NCMsg _from (Died (Right nid) reason) ->

101 ncEffectNodeDied nid reason

104 ncEffectNodeDied nid reason

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

105 NCMsg _from (Died (Left pid) reason) ->

103 ncEffectProcessDied pid reason

106 ncEffectProcessDied pid reason

107+ -- Error cases

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

109+ error "Link message from a node?"

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

111+ error "Monitor message from a node?"

112+ NCMsg (Right _) (Unlink _) ->

113+ error "Unlink message from a node?"

114+ NCMsg (Right _) (Unmonitor _) ->

115+ error "Unmonitor message from a node?"

104

116

105-- [Unified: Table 10]

117-- [Unified: Table 10]

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

118ncEffectMonitor :: ProcessId -- ^ Who's watching?

@@ -110,12 +122,11 @@ ncEffectMonitor :: ProcessId -- ^ Who's watching?

110ncEffectMonitor from them mRef = do

122ncEffectMonitor from them mRef = do

111 node <- ask

123 node <- ask

112 shouldLink <-

124 shouldLink <-

113− if processNodeId them /= localNodeId node

125+ if not (isLocal node them)

114 then return True

126 then return True

115 else liftIO . withMVar (localState node) $

127 else liftIO . withMVar (localState node) $

116 return . isJust . (^. localProcessWithId (processLocalId them))

128 return . isJust . (^. localProcessWithId (processLocalId them))

117− let localFrom = processNodeId from == localNodeId node

118− case (shouldLink, localFrom) of

129+ case (shouldLink, isLocal node from) of

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

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

120 case mRef of

131 case mRef of

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

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

@@ -128,6 +139,20 @@ ncEffectMonitor from them mRef = do

128 , ctrlMsgSignal = Died (Left them) DiedNoProc

139 , ctrlMsgSignal = Died (Left them) DiedNoProc

129 }

140 }

130

141

142+-- [Unified: Table 11]

143+ncEffectUnlink :: ProcessId -> ProcessId -> NC ()

144+ncEffectUnlink from them = do

145+ node <- ask

146+ when (isLocal node from) $ postMessage from $ DidUnlink them

147+ modify $ linksForProcess them ^: Set.delete from

148+

149+-- [Unified: Table 11]

150+ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()

151+ncEffectUnmonitor from ref = do

152+ node <- ask

153+ when (isLocal node from) $ postMessage from $ DidUnmonitor ref

154+ modify $ monitorsFor (monitorRefPid ref) from ^: Set.delete ref

155+

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

156-- [Unified: Table 12, bottom rule]

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

157ncEffectNodeDied :: NodeId -> DiedReason -> NC ()

133ncEffectNodeDied nid reason = do

158ncEffectNodeDied nid reason = do

@@ -139,13 +164,13 @@ ncEffectNodeDied nid reason = do

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

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

140 let pid = ProcessId nid them

165 let pid = ProcessId nid them

141 forM_ uss $ \us ->

166 forM_ uss $ \us ->

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

167+ when (isLocal node us) $

143 notifyDied us pid reason Nothing

168 notifyDied us pid reason Nothing

144

169

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

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

146 let pid = ProcessId nid them

171 let pid = ProcessId nid them

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

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

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

173+ when (isLocal node us) $

149 forM_ refs $ notifyDied us pid reason . Just

174 forM_ refs $ notifyDied us pid reason . Just

150

175

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

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

@@ -171,6 +196,10 @@ ncEffectProcessDied pid reason = do

171-- Connections --

196-- Connections --

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

197--------------------------------------------------------------------------------

173

198

199+-- | Check if a process is local to our own node

200+isLocal :: LocalNode -> ProcessId -> Bool

201+isLocal nid pid = processNodeId pid == localNodeId nid

202+

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

203notifyDied :: ProcessId -- ^ Who to notify?

175 -> ProcessId -- ^ Who died?

204 -> ProcessId -- ^ Who died?

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

205 -> DiedReason -- ^ How did they die?

@@ -178,25 +207,36 @@ notifyDied :: ProcessId -- ^ Who to notify?

178 -> NC ()

207 -> NC ()

179notifyDied dest src reason mRef = do

208notifyDied dest src reason mRef = do

180 node <- ask

209 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

210+ case (isLocal node dest, mRef) of

188− Just ref -> do

211+ (True, Just ref) ->

189− let msg = createMessage $ MonitorNotification ref src reason

212+ postMessage dest $ MonitorNotification ref src reason

190− enqueue (processQueue proc) msg

191− Nothing ->

213+ (True, Nothing) ->

192− throwTo (processThread proc) $ LinkException src reason

214+ throwException dest $ LinkException src reason

193− else

215+ (False, _) ->

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

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

195 -- (see [Unified: Table 10]

217 -- (see [Unified: Table 10]

196 sendCtrlMsg (processNodeId dest) NCMsg

218 sendCtrlMsg (processNodeId dest) NCMsg

197 { ctrlMsgSender = Right (localNodeId node)

219 { ctrlMsgSender = Right (localNodeId node)

198 , ctrlMsgSignal = Died (Left src) reason

220 , ctrlMsgSignal = Died (Left src) reason

199 }

221 }

222+

223+postMessage :: Serializable a => ProcessId -> a -> NC ()

224+postMessage pid msg = withLocalProc pid $ \p ->

225+ enqueue (processQueue p) (createMessage msg)

226+

227+throwException :: Exception e => ProcessId -> e -> NC ()

228+throwException pid e = withLocalProc pid $ \p ->

229+ throwTo (processThread p) e

230+

231+withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()

232+withLocalProc pid p = do

233+ node <- ask

234+ liftIO $ do

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

236+ -- can silently be dropped

237+ let lpid = processLocalId pid

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

239+ forM_ mProc p

200

240

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

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

202sendTo them payload = do

242sendTo them payload = do

@@ -259,8 +299,10 @@ createConnTo them = do

259--------------------------------------------------------------------------------

299--------------------------------------------------------------------------------

260

300

261destNid :: ProcessSignal -> Maybe NodeId

301destNid :: ProcessSignal -> Maybe NodeId

302+destNid (Link pid) = Just . processNodeId $ pid

303+destNid (Unlink pid) = Just . processNodeId $ pid

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

304+destNid (Monitor pid _) = Just . processNodeId $ pid

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

305+destNid (Unmonitor ref) = Just . processNodeId . monitorRefPid $ ref

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

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

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

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

266

308

distributed-process/tests/TestCH.hs+86−84

@@ -3,8 +3,10 @@ module Main where

3import Prelude hiding (catch)

4import Data.Binary (Binary)

5import Data.Typeable (Typeable)

6+import Data.Foldable (forM_)

6import Control.Concurrent (forkIO, threadDelay)

7import Control.Concurrent (forkIO, threadDelay)

7−import Control.Concurrent.MVar ( newEmptyMVar

8+import Control.Concurrent.MVar ( MVar

9+ , newEmptyMVar

8 , putMVar

10 , putMVar

9 , takeMVar

11 , takeMVar

10 , readMVar

12 , readMVar

@@ -64,13 +66,53 @@ testPing transport = do

64 takeMVar clientDone

66 takeMVar clientDone

66-- | Monitor or link to a remote node

68-- | Monitor or link to a remote node

69+monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link

70+ -> ProcessId -- Process to monitor/link to

71+ -> Maybe (MVar ()) -- MVar to signal on once the monitor has been set up

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

72+ -> Process (Maybe MonitorRef)

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

73+monitorOrLink mOrL pid mSignal = do

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

74+ result <- if mOrL then Just <$> monitor pid

75+ else link pid >> return Nothing

76+ -- Monitor is asynchronous, which usually does not matter but if we want a

77+ -- *specific* signal then it does. Therefore we wait an arbitrary delay and

78+ -- hope that this means the monitor has been set up

79+ forM_ mSignal $ \signal -> liftIO . forkIO $ threadDelay 100000 >> putMVar signal ()

80+ return result

81+

82+monitorTestProcess :: ProcessId -- Process to monitor/link to

83+ -> Bool -- 'True' for monitor, 'False' for link

84+ -> Bool -- Should we unmonitor?

85+ -> DiedReason -- Expected cause of death

86+ -> Maybe (MVar ()) -- Signal for 'monitor set up'

87+ -> MVar () -- Signal for successful termination

88+ -> Process ()

89+monitorTestProcess theirAddr mOrL un reason monitorSetup done =

90+ catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup

91+ case (un, mRef) of

92+ (True, Nothing) -> do

93+ unlink theirAddr

94+ DidUnlink pid <- expect

95+ True <- return $ pid == theirAddr

96+ liftIO $ putMVar done ()

97+ (True, Just ref) -> do

98+ unmonitor ref

99+ DidUnmonitor ref' <- expect

100+ True <- return $ ref == ref'

101+ liftIO $ putMVar done ()

102+ (False, ref) -> do

103+ MonitorNotification ref' pid reason' <- expect

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

105+ liftIO $ putMVar done ()

106+ )

107+ (\(LinkException pid reason') -> do

108+ True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'

109+ liftIO $ putMVar done ()

110+ )

111+

112

71-- | Monitor an unreachable node

113-- | Monitor an unreachable node

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

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

73−testMonitorUnreachable transport mOrL = do

115+testMonitorUnreachable transport mOrL un = do

74 deadProcess <- newEmptyMVar

116 deadProcess <- newEmptyMVar

75 done <- newEmptyMVar

117 done <- newEmptyMVar

118

@@ -84,19 +126,13 @@ testMonitorUnreachable transport mOrL = do

84 localNode <- newLocalNode transport

126 localNode <- newLocalNode transport

85 theirAddr <- readMVar deadProcess

127 theirAddr <- readMVar deadProcess

86 runProcess localNode $

128 runProcess localNode $

87− catch (do ref <- monitorOrLink mOrL theirAddr

129+ monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done

88− MonitorNotification ref' pid DiedDisconnect <- expect

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

90− liftIO $ putMVar done ())

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

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

93− liftIO $ putMVar done ())

130

95 takeMVar done

131 takeMVar done

132

97-- | Monitor a process which terminates normally

133-- | Monitor a process which terminates normally

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

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

99−testMonitorNormalTermination transport mOrL = do

135+testMonitorNormalTermination transport mOrL un = do

100 monitorSetup <- newEmptyMVar

136 monitorSetup <- newEmptyMVar

101 monitoredProcess <- newEmptyMVar

137 monitoredProcess <- newEmptyMVar

102 done <- newEmptyMVar

138 done <- newEmptyMVar

@@ -111,25 +147,13 @@ testMonitorNormalTermination transport mOrL = do

111 localNode <- newLocalNode transport

147 localNode <- newLocalNode transport

112 theirAddr <- readMVar monitoredProcess

148 theirAddr <- readMVar monitoredProcess

113 runProcess localNode $

149 runProcess localNode $

114− catch (do ref <- monitorOrLink mOrL theirAddr

150+ monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done

115− liftIO $ do

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

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

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

119− threadDelay 100000

120− putMVar monitorSetup ()

121− MonitorNotification ref' pid DiedNormal <- expect

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

123− liftIO $ putMVar done ())

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

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

126− liftIO $ putMVar done ())

127

151

128 takeMVar done

152 takeMVar done

129

153

130-- | Monitor a process which terminates abnormally

154-- | Monitor a process which terminates abnormally

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

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

132−testMonitorAbnormalTermination transport mOrL = do

156+testMonitorAbnormalTermination transport mOrL un = do

133 monitorSetup <- newEmptyMVar

157 monitorSetup <- newEmptyMVar

134 monitoredProcess <- newEmptyMVar

158 monitoredProcess <- newEmptyMVar

135 done <- newEmptyMVar

159 done <- newEmptyMVar

@@ -147,25 +171,13 @@ testMonitorAbnormalTermination transport mOrL = do

147 localNode <- newLocalNode transport

171 localNode <- newLocalNode transport

148 theirAddr <- readMVar monitoredProcess

172 theirAddr <- readMVar monitoredProcess

149 runProcess localNode $

173 runProcess localNode $

150− catch (do ref <- monitorOrLink mOrL theirAddr

174+ monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done

151− liftIO $ do

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

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

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

155− threadDelay 100000

156− putMVar monitorSetup ()

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

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

159− liftIO $ putMVar done ())

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

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

162− liftIO $ putMVar done ())

163

175

164 takeMVar done

176 takeMVar done

165

177

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

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

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

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

168−testMonitorLocalDeadProcess transport mOrL = do

180+testMonitorLocalDeadProcess transport mOrL un = do

169 processDead <- newEmptyMVar

181 processDead <- newEmptyMVar

170 processAddr <- newEmptyMVar

182 processAddr <- newEmptyMVar

171 localNode <- newLocalNode transport

183 localNode <- newLocalNode transport

@@ -179,19 +191,13 @@ testMonitorLocalDeadProcess transport mOrL = do

179 theirAddr <- readMVar processAddr

191 theirAddr <- readMVar processAddr

180 readMVar processDead

192 readMVar processDead

181 runProcess localNode $ do

193 runProcess localNode $ do

182− catch (do ref <- monitorOrLink mOrL theirAddr

194+ monitorTestProcess theirAddr mOrL un DiedNoProc Nothing done

183− MonitorNotification ref' pid DiedNoProc <- expect

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

185− liftIO $ putMVar done ())

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

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

188− liftIO $ putMVar done ())

189

195

190 takeMVar done

196 takeMVar done

191

197

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

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

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

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

194−testMonitorRemoteDeadProcess transport mOrL = do

200+testMonitorRemoteDeadProcess transport mOrL un = do

195 processDead <- newEmptyMVar

201 processDead <- newEmptyMVar

196 processAddr <- newEmptyMVar

202 processAddr <- newEmptyMVar

197 done <- newEmptyMVar

203 done <- newEmptyMVar

@@ -206,19 +212,13 @@ testMonitorRemoteDeadProcess transport mOrL = do

206 theirAddr <- readMVar processAddr

212 theirAddr <- readMVar processAddr

207 readMVar processDead

213 readMVar processDead

208 runProcess localNode $ do

214 runProcess localNode $ do

209− catch (do ref <- monitorOrLink mOrL theirAddr

215+ monitorTestProcess theirAddr mOrL un DiedNoProc Nothing done

210− MonitorNotification ref' pid DiedNoProc <- expect

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

212− liftIO $ putMVar done ())

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

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

215− liftIO $ putMVar done ())

216

217 takeMVar done

218

219-- | Monitor a process that becomes disconnected

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

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

221−testMonitorDisconnect transport mOrL = do

221+testMonitorDisconnect transport mOrL un = do

222 processAddr <- newEmptyMVar

223 monitorSetup <- newEmptyMVar

224 done <- newEmptyMVar

@@ -234,14 +234,7 @@ testMonitorDisconnect transport mOrL = do

234 localNode <- newLocalNode transport

235 theirAddr <- readMVar processAddr

236 runProcess localNode $ do

237− catch (do ref <- monitorOrLink mOrL theirAddr

237+ monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done

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

239− MonitorNotification ref' pid DiedDisconnect <- expect

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

241− liftIO $ putMVar done ())

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

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

244− liftIO $ putMVar done ())

245

238

246 takeMVar done

239 takeMVar done

247

240

@@ -352,19 +345,28 @@ main :: IO ()

352main = do

345main = do

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

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

354 runTests

347 runTests

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

348+ [ ("Ping", testPing transport)

349+ -- The "missing" combinations in the list below don't make much sense, as

350+ -- we cannot guarantee that the monitor reply or link exception will not

351+ -- happen before the unmonitor or unlink

356− ("MonitorUnreachable", testMonitorUnreachable transport True)

352+ , ("MonitorUnreachable", testMonitorUnreachable transport True False)

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

353+ , ("MonitorNormalTermination", testMonitorNormalTermination transport True False)

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

354+ , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport True False)

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

355+ , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport True False)

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

356+ , ("MonitorRemoteDeadProcess", testMonitorRemoteDeadProcess transport True False)

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

357+ , ("MonitorDisconnect", testMonitorDisconnect transport True False)

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

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

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

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

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

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

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

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

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

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

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

364+ , ("UnmonitorNormalTermination", testMonitorNormalTermination transport True True)

365+ , ("UnmonitorAbnormalTermination", testMonitorAbnormalTermination transport True True)

366+ , ("UnmonitorDisconnect", testMonitorDisconnect transport True True)

367+ , ("UnlinkNormalTermination", testMonitorNormalTermination transport False True)

368+ , ("UnlinkAbnormalTermination", testMonitorAbnormalTermination transport False True)

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

369+ , ("UnlinkDisconnect", testMonitorDisconnect transport False True)

368 {-

370 {-

369 , ("Monitor2", testMonitor2 transport)

371 , ("Monitor2", testMonitor2 transport)

370 , ("Monitor3", testMonitor3 transport)

372 , ("Monitor3", testMonitor3 transport)

network-transport/src/Network/Transport.hs+2−1

@@ -23,6 +23,7 @@ module Network.Transport ( -- * Types

23 ) where

25import Data.ByteString (ByteString)

26+import qualified Data.ByteString.Char8 as BSC (unpack)

26import Control.Exception (Exception)

27import Control.Exception (Exception)

27import Data.Typeable (Typeable)

28import Data.Typeable (Typeable)

28import Data.Binary (Binary)

29import Data.Binary (Binary)

@@ -113,7 +114,7 @@ newtype EndPointAddress = EndPointAddress { endPointAddressToByteString :: ByteS

113 deriving (Eq, Ord, Typeable, Binary)

114 deriving (Eq, Ord, Typeable, Binary)

114

115

115instance Show EndPointAddress where

116instance Show EndPointAddress where

116− show = show . endPointAddressToByteString

117+ show = BSC.unpack . endPointAddressToByteString

117

118

118-- | EndPointAddress of a multicast group.

119-- | EndPointAddress of a multicast group.

119newtype MulticastAddress = MulticastAddress { multicastAddressToByteString :: ByteString }

120newtype MulticastAddress = MulticastAddress { multicastAddressToByteString :: ByteString }