10ccea3 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Support monitor/link for nodes and ports

10ccea3Edsko de Vries14 years ago

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

compare with master

unifiedsplit

@@ -18,6 +18,7 @@ module Control.Distributed.Process

18 -- * Channels

19 , ReceivePort

20 , SendPort

21+ , sendPortId

21 , newChan

22 , newChan

22 , sendChan

23 , sendChan

23 , receiveChan

24 , receiveChan

@@ -40,15 +41,23 @@ module Control.Distributed.Process

40 , getSelfNode

41 , getSelfNode

41 -- * Monitoring and linking

42 -- * Monitoring and linking

42 , link

43 , link

44+ , linkNode

45+ , linkPort

43 , unlink

46 , unlink

47+ , unlinkNode

48+ , unlinkPort

44 , monitor

49 , monitor

50+ , monitorNode

51+ , monitorPort

45 , unmonitor

52 , unmonitor

46− , LinkException(..)

53+ , ProcessLinkException(..)

54+ , NodeLinkException(..)

55+ , PortLinkException(..)

47 , MonitorRef -- opaque

56 , MonitorRef -- opaque

48− , MonitorNotification(..)

57+ , ProcessMonitorNotification(..)

58+ , NodeMonitorNotification(..)

59+ , PortMonitorNotification(..)

49 , DiedReason(..)

60 , DiedReason(..)

50− , DidUnmonitor(..)

51− , DidUnlink(..)

52 -- * Closures

61 -- * Closures

53 , Closure

62 , Closure

54 , Static

63 , Static

@@ -78,10 +87,12 @@ import Control.Distributed.Process.Internal.Types

78 , Closure(..)

87 , Closure(..)

79 , Static(..)

88 , Static(..)

80 , MonitorRef(..)

89 , MonitorRef(..)

81− , MonitorNotification(..)

90+ , ProcessMonitorNotification(..)

82− , LinkException(..)

91+ , NodeMonitorNotification(..)

83− , DidUnmonitor(..)

92+ , PortMonitorNotification(..)

84− , DidUnlink(..)

93+ , ProcessLinkException(..)

94+ , NodeLinkException(..)

95+ , PortLinkException(..)

85 , DiedReason(..)

96 , DiedReason(..)

86 , SpawnRef(..)

97 , SpawnRef(..)

87 , DidSpawn(..)

98 , DidSpawn(..)

@@ -122,8 +133,14 @@ import Control.Distributed.Process.Internal.Process.Primitives

122 , getSelfNode

133 , getSelfNode

123 -- Monitoring and linking

134 -- Monitoring and linking

124 , link

135 , link

136+ , linkNode

137+ , linkPort

125 , unlink

138 , unlink

139+ , unlinkNode

140+ , unlinkPort

126 , monitor

141 , monitor

142+ , monitorNode

143+ , monitorPort

127 , unmonitor

144 , unmonitor

128 -- Auxiliary API

145 -- Auxiliary API

129 , catch

146 , catch

distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+2−8

@@ -27,9 +27,8 @@ import Control.Monad.IO.Class (MonadIO, liftIO)

27import Control.Concurrent.Chan (writeChan)

28import Control.Distributed.Process.Internal.Types

29 ( Identifier(..)

30+ , nodeOf

30 , NodeId(nodeAddress)

31 , NodeId(nodeAddress)

31− , ProcessId(processNodeId)

32− , ChannelId(channelProcessId)

33 , LocalNode(localCtrlChan, localEndPoint)

32 , LocalNode(localCtrlChan, localEndPoint)

34 , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)

33 , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)

35 , DiedReason(DiedDisconnect)

34 , DiedReason(DiedDisconnect)

@@ -122,7 +121,7 @@ setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)

122setupConnTo them = do

121setupConnTo them = do

123 endPoint <- localEndPoint `liftM` getLocalNode

122 endPoint <- localEndPoint `liftM` getLocalNode

124 mConn <- liftIO $ NT.connect endPoint

123 mConn <- liftIO $ NT.connect endPoint

125− (nodeAddress . identifierNode $ them)

124+ (nodeAddress . nodeOf $ them)

126 NT.ReliableOrdered

125 NT.ReliableOrdered

127 NT.defaultConnectHints

126 NT.defaultConnectHints

128 case mConn of

127 case mConn of

@@ -144,11 +143,6 @@ connTo them = do

144 Just conn -> return $ Just conn

143 Just conn -> return $ Just conn

145 Nothing -> setupConnTo them

144 Nothing -> setupConnTo them

146

145

147−identifierNode :: Identifier -> NodeId

148−identifierNode (NodeIdentifier nid) = nid

149−identifierNode (ProcessIdentifier pid) = processNodeId pid

150−identifierNode (ChannelIdentifier cid) = processNodeId (channelProcessId cid)

151−

152--------------------------------------------------------------------------------

146--------------------------------------------------------------------------------

153-- Accessors --

147-- Accessors --

154--------------------------------------------------------------------------------

148--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/Process/Primitives.hs+100−19

@@ -36,6 +36,12 @@ module Control.Distributed.Process.Internal.Process.Primitives

36 , spawnAsync

37 , spawnLink

38 , spawnMonitor

39+ , linkNode

40+ , linkPort

41+ , unlinkNode

42+ , unlinkPort

43+ , monitorNode

44+ , monitorPort

39 ) where

45 ) where

41import Prelude hiding (catch)

47import Prelude hiding (catch)

@@ -82,9 +88,13 @@ import Control.Distributed.Process.Internal.Types

82 , channelCounter

88 , channelCounter

83 , typedChannelWithId

89 , typedChannelWithId

84 , TypedChannel(..)

90 , TypedChannel(..)

85− , ChannelId(..)

91+ , SendPortId(..)

86 , Identifier(..)

92 , Identifier(..)

87 , procMsg

93 , procMsg

94+ , DidUnmonitor(..)

95+ , DidUnlinkProcess(..)

96+ , DidUnlinkNode(..)

97+ , DidUnlinkPort(..)

88 )

98 )

89import Control.Distributed.Process.Internal.MessageT

99import Control.Distributed.Process.Internal.MessageT

90 ( sendMessage

100 ( sendMessage

@@ -118,9 +128,9 @@ newChan = do

118 liftIO . modifyMVar (processState proc) $ \st -> do

128 liftIO . modifyMVar (processState proc) $ \st -> do

119 chan <- liftIO . atomically $ newTChan

129 chan <- liftIO . atomically $ newTChan

120 let lcid = st ^. channelCounter

130 let lcid = st ^. channelCounter

121− cid = ChannelId { channelProcessId = processId proc

131+ cid = SendPortId { sendPortProcessId = processId proc

122− , channelLocalId = lcid

132+ , sendPortLocalId = lcid

123− }

133+ }

124 sport = SendPort cid

134 sport = SendPort cid

125 rport = ReceivePortSingle chan

135 rport = ReceivePortSingle chan

126 tch = TypedChannel chan

136 tch = TypedChannel chan

@@ -132,7 +142,7 @@ newChan = do

132

142

133-- | Send a message on a typed channel

143-- | Send a message on a typed channel

134sendChan :: Serializable a => SendPort a -> a -> Process ()

144sendChan :: Serializable a => SendPort a -> a -> Process ()

135−sendChan (SendPort them) msg = procMsg $ sendBinary (ChannelIdentifier them) msg

145+sendChan (SendPort cid) msg = procMsg $ sendBinary (SendPortIdentifier cid) msg

136

146

137-- | Wait for a message on a typed channel

147-- | Wait for a message on a typed channel

138receiveChan :: Serializable a => ReceivePort a -> Process a

148receiveChan :: Serializable a => ReceivePort a -> Process a

@@ -250,22 +260,43 @@ getSelfNode = localNodeId <$> procMsg getLocalNode

250-- Linking makes no distinction between normal and abnormal termination of

260-- Linking makes no distinction between normal and abnormal termination of

251-- the remote process.

261-- the remote process.

252link :: ProcessId -> Process ()

262link :: ProcessId -> Process ()

253−link = sendCtrlMsg Nothing . Link

263+link = sendCtrlMsg Nothing . Link . ProcessIdentifier

254−

255−-- | Remove a link (asynchronous)

256−unlink :: ProcessId -> Process ()

257−unlink = sendCtrlMsg Nothing . Unlink

258

264

259-- | Monitor another process (asynchronous)

265-- | Monitor another process (asynchronous)

260monitor :: ProcessId -> Process MonitorRef

266monitor :: ProcessId -> Process MonitorRef

261−monitor them = do

267+monitor = monitor' . ProcessIdentifier

262− monitorRef <- getMonitorRefFor them

263− sendCtrlMsg Nothing $ Monitor them monitorRef

264− return monitorRef

265

268

266−-- | Remove a monitor (asynchronous)

269+-- | Remove a link (synchronous)

270+unlink :: ProcessId -> Process ()

271+unlink pid = do

272+ unlinkAsync pid

273+ receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)

274+ (\_ -> return ())

275+ ]

276+

277+-- | Remove a node link (synchronous)

278+unlinkNode :: NodeId -> Process ()

279+unlinkNode nid = do

280+ unlinkNodeAsync nid

281+ receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)

282+ (\_ -> return ())

283+ ]

284+

285+-- | Remove a channel (send port) link (synchronous)

286+unlinkPort :: SendPort a -> Process ()

287+unlinkPort sport = do

288+ unlinkPortAsync sport

289+ receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)

290+ (\_ -> return ())

291+ ]

292+

293+-- | Remove a monitor (synchronous)

267unmonitor :: MonitorRef -> Process ()

294unmonitor :: MonitorRef -> Process ()

268−unmonitor = sendCtrlMsg Nothing . Unmonitor

295+unmonitor ref = do

296+ unmonitorAsync ref

297+ receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)

298+ (\_ -> return ())

299+ ]

269

300

270--------------------------------------------------------------------------------

301--------------------------------------------------------------------------------

271-- Auxiliary API --

302-- Auxiliary API --

@@ -311,17 +342,56 @@ spawnMonitor nid proc = do

311 ref <- monitor pid

342 ref <- monitor pid

312 return (pid, ref)

343 return (pid, ref)

313

344

345+-- | Monitor a node

346+monitorNode :: NodeId -> Process MonitorRef

347+monitorNode =

348+ monitor' . NodeIdentifier

349+

350+-- | Monitor a typed channel

351+monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef

352+monitorPort (SendPort cid) =

353+ monitor' (SendPortIdentifier cid)

354+

355+-- | Remove a monitor (asynchronous)

356+unmonitorAsync :: MonitorRef -> Process ()

357+unmonitorAsync =

358+ sendCtrlMsg Nothing . Unmonitor

359+

360+-- | Link to a node

361+linkNode :: NodeId -> Process ()

362+linkNode = link' . NodeIdentifier

363+

364+-- | Link to a channel (send port)

365+linkPort :: SendPort a -> Process ()

366+linkPort (SendPort cid) =

367+ link' (SendPortIdentifier cid)

368+

369+-- | Remove a process link (asynchronous)

370+unlinkAsync :: ProcessId -> Process ()

371+unlinkAsync =

372+ sendCtrlMsg Nothing . Unlink . ProcessIdentifier

373+

374+-- | Remove a node link (asynchronous)

375+unlinkNodeAsync :: NodeId -> Process ()

376+unlinkNodeAsync =

377+ sendCtrlMsg Nothing . Unlink . NodeIdentifier

378+

379+-- | Remove a channel (send port) link (asynchronous)

380+unlinkPortAsync :: SendPort a -> Process ()

381+unlinkPortAsync (SendPort cid) =

382+ sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid

383+

314--------------------------------------------------------------------------------

384--------------------------------------------------------------------------------

315-- Auxiliary functions --

385-- Auxiliary functions --

316--------------------------------------------------------------------------------

386--------------------------------------------------------------------------------

317

387

318−getMonitorRefFor :: ProcessId -> Process MonitorRef

388+getMonitorRefFor :: Identifier -> Process MonitorRef

319−getMonitorRefFor pid = do

389+getMonitorRefFor ident = do

320 proc <- ask

390 proc <- ask

321 liftIO $ modifyMVar (processState proc) $ \st -> do

391 liftIO $ modifyMVar (processState proc) $ \st -> do

322 let counter = st ^. monitorCounter

392 let counter = st ^. monitorCounter

323 return ( monitorCounter ^: (+ 1) $ st

393 return ( monitorCounter ^: (+ 1) $ st

324− , MonitorRef pid counter

394+ , MonitorRef ident counter

325 )

395 )

326

396

327getSpawnRef :: Process SpawnRef

397getSpawnRef :: Process SpawnRef

@@ -333,6 +403,17 @@ getSpawnRef = do

333 , SpawnRef counter

403 , SpawnRef counter

334 )

404 )

335

405

406+-- | Monitor a process/node/channel

407+monitor' :: Identifier -> Process MonitorRef

408+monitor' ident = do

409+ monitorRef <- getMonitorRefFor ident

410+ sendCtrlMsg Nothing $ Monitor monitorRef

411+ return monitorRef

412+

413+-- | Link to a process/node/channel

414+link' :: Identifier -> Process ()

415+link' = sendCtrlMsg Nothing . Link

416+

336-- Send a control message

417-- Send a control message

337sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node

418sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node

338 -> ProcessSignal -- ^ Message to send

419 -> ProcessSignal -- ^ Message to send

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+107−46

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

10 , LocalProcessId(..)

11 , ProcessId(..)

12 , Identifier(..)

13+ , nodeOf

13 -- * Local nodes and processes

14 -- * Local nodes and processes

14 , LocalNode(..)

15 , LocalNode(..)

15 , LocalNodeState(..)

16 , LocalNodeState(..)

@@ -18,8 +19,8 @@ module Control.Distributed.Process.Internal.Types

18 , Process(..)

19 , Process(..)

19 , procMsg

20 , procMsg

20 -- * Typed channels

21 -- * Typed channels

21− , LocalChannelId

22+ , LocalSendPortId

22− , ChannelId(..)

23+ , SendPortId(..)

23 , TypedChannel(..)

24 , TypedChannel(..)

24 , SendPort(..)

25 , SendPort(..)

25 , ReceivePort(..)

26 , ReceivePort(..)

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

34 , Message(..)

35 , Message(..)

35 -- * Node controller user-visible data types

36 -- * Node controller user-visible data types

36 , MonitorRef(..)

37 , MonitorRef(..)

37− , MonitorNotification(..)

38+ , ProcessMonitorNotification(..)

38− , LinkException(..)

39+ , NodeMonitorNotification(..)

40+ , PortMonitorNotification(..)

41+ , ProcessLinkException(..)

42+ , NodeLinkException(..)

43+ , PortLinkException(..)

39 , DiedReason(..)

44 , DiedReason(..)

40 , DidUnmonitor(..)

45 , DidUnmonitor(..)

41− , DidUnlink(..)

46+ , DidUnlinkProcess(..)

47+ , DidUnlinkNode(..)

48+ , DidUnlinkPort(..)

42 , SpawnRef(..)

49 , SpawnRef(..)

43 , DidSpawn(..)

50 , DidSpawn(..)

44 -- * Node controller internal data types

51 -- * Node controller internal data types

@@ -121,10 +128,20 @@ instance Show ProcessId where

121

128

122-- | Union of all kinds of identifiers

129-- | Union of all kinds of identifiers

123data Identifier =

130data Identifier =

131+ NodeIdentifier NodeId

124− ProcessIdentifier ProcessId

132+ | ProcessIdentifier ProcessId

125− | NodeIdentifier NodeId

133+ | SendPortIdentifier SendPortId

126− | ChannelIdentifier ChannelId

127− deriving (Show, Eq, Ord)

134+ deriving (Eq, Ord)

135+

136+instance Show Identifier where

137+ show (NodeIdentifier nid) = show nid

138+ show (ProcessIdentifier pid) = show pid

139+ show (SendPortIdentifier cid) = show cid

140+

141+nodeOf :: Identifier -> NodeId

142+nodeOf (NodeIdentifier nid) = nid

143+nodeOf (ProcessIdentifier pid) = processNodeId pid

144+nodeOf (SendPortIdentifier cid) = processNodeId (sendPortProcessId cid)

128

145

129--------------------------------------------------------------------------------

146--------------------------------------------------------------------------------

130-- Local nodes and processes --

147-- Local nodes and processes --

@@ -165,7 +182,7 @@ data LocalProcessState = LocalProcessState

165 { _monitorCounter :: Int32

182 { _monitorCounter :: Int32

166 , _spawnCounter :: Int32

183 , _spawnCounter :: Int32

167 , _channelCounter :: Int32

184 , _channelCounter :: Int32

168− , _typedChannels :: Map LocalChannelId TypedChannel

185+ , _typedChannels :: Map LocalSendPortId TypedChannel

169 }

186 }

170

187

171-- | The Cloud Haskell 'Process' type

188-- | The Cloud Haskell 'Process' type

@@ -181,18 +198,22 @@ procMsg = Process . Trans.lift

181-- Typed channels --

198-- Typed channels --

182--------------------------------------------------------------------------------

199--------------------------------------------------------------------------------

183

200

184−type LocalChannelId = Int32

201+type LocalSendPortId = Int32

185

202

186−data ChannelId = ChannelId {

203+data SendPortId = SendPortId {

187− channelProcessId :: ProcessId

204+ sendPortProcessId :: ProcessId

188− , channelLocalId :: LocalChannelId

205+ , sendPortLocalId :: LocalSendPortId

189 }

206 }

190− deriving (Show, Eq, Ord)

207+ deriving (Eq, Ord)

208+

209+instance Show SendPortId where

210+ show cid = show (sendPortProcessId cid) ++ ":" ++ show (sendPortLocalId cid)

191

211

192data TypedChannel = forall a. Serializable a => TypedChannel (TChan a)

212data TypedChannel = forall a. Serializable a => TypedChannel (TChan a)

193

213

194-- | The send send of a typed channel (serializable)

214-- | The send send of a typed channel (serializable)

215+newtype SendPort a = SendPort { sendPortId :: SendPortId }

195−newtype SendPort a = SendPort ChannelId deriving (Typeable, Binary)

216+ deriving (Typeable, Binary, Show, Eq, Ord)

196

217

197-- | The receive end of a typed channel (not serializable)

218-- | The receive end of a typed channel (not serializable)

198data ReceivePort a =

219data ReceivePort a =

@@ -274,22 +295,46 @@ data Message = Message

274

295

275-- | MonitorRef is opaque for regular Cloud Haskell processes

296-- | MonitorRef is opaque for regular Cloud Haskell processes

276data MonitorRef = MonitorRef

297data MonitorRef = MonitorRef

277− { -- | PID of the process to be monitored (for routing purposes)

298+ { -- | ID of the entity to be monitored

278− monitorRefPid :: ProcessId

299+ monitorRefIdent :: Identifier

279 -- | Unique to distinguish multiple monitor requests by the same process

300 -- | Unique to distinguish multiple monitor requests by the same process

280 , monitorRefCounter :: Int32

301 , monitorRefCounter :: Int32

281 }

302 }

282 deriving (Eq, Ord, Show)

303 deriving (Eq, Ord, Show)

283

304

284−-- | Messages sent by monitors

305+-- | Message sent by process monitors

306+data ProcessMonitorNotification =

285−data MonitorNotification = MonitorNotification MonitorRef ProcessId DiedReason

307+ ProcessMonitorNotification MonitorRef ProcessId DiedReason

308+ deriving (Typeable, Show)

309+

310+-- | Message sent by node monitors

311+data NodeMonitorNotification =

312+ NodeMonitorNotification MonitorRef NodeId DiedReason

313+ deriving (Typeable, Show)

314+

315+-- | Message sent by channel (port) monitors

316+data PortMonitorNotification =

317+ PortMonitorNotification MonitorRef SendPortId DiedReason

286 deriving (Typeable, Show)

318 deriving (Typeable, Show)

287

319

288-- | Exceptions thrown when a linked process dies

320-- | Exceptions thrown when a linked process dies

321+data ProcessLinkException =

289−data LinkException = LinkException ProcessId DiedReason

322+ ProcessLinkException ProcessId DiedReason

290 deriving (Typeable, Show)

323 deriving (Typeable, Show)

291

324

325+-- | Exception thrown when a linked node dies

326+data NodeLinkException =

327+ NodeLinkException NodeId DiedReason

328+ deriving (Typeable, Show)

329+

330+-- | Exception thrown when a linked channel (port) dies

331+data PortLinkException =

332+ PortLinkException SendPortId DiedReason

333+ deriving (Typeable, Show)

334+

335+instance Exception ProcessLinkException

292−instance Exception LinkException

336+instance Exception NodeLinkException

337+instance Exception PortLinkException

293

338

294-- | Why did a process die?

339-- | Why did a process die?

295data DiedReason =

340data DiedReason =

@@ -302,8 +347,8 @@ data DiedReason =

302 | DiedDisconnect

347 | DiedDisconnect

303 -- | The process node died

348 -- | The process node died

304 | DiedNodeDown

349 | DiedNodeDown

305− -- | Invalid process ID

350+ -- | Invalid (process/node/channel) identifier

306− | DiedNoProc

351+ | DiedUnknownId

307 deriving (Show, Eq)

352 deriving (Show, Eq)

308

353

309-- | (Asynchronous) reply from unmonitor

354-- | (Asynchronous) reply from unmonitor

@@ -311,7 +356,15 @@ newtype DidUnmonitor = DidUnmonitor MonitorRef

311 deriving (Typeable, Binary)

356 deriving (Typeable, Binary)

312

357

313-- | (Asynchronous) reply from unlink

358-- | (Asynchronous) reply from unlink

314−newtype DidUnlink = DidUnlink ProcessId

359+newtype DidUnlinkProcess = DidUnlinkProcess ProcessId

360+ deriving (Typeable, Binary)

361+

362+-- | (Asynchronous) reply from unlinkNode

363+newtype DidUnlinkNode = DidUnlinkNode NodeId

364+ deriving (Typeable, Binary)

365+

366+-- | (Asynchronous) reply from unlinkPort

367+newtype DidUnlinkPort = DidUnlinkPort SendPortId

315 deriving (Typeable, Binary)

368 deriving (Typeable, Binary)

316

369

317-- | 'SpawnRef' are used to return pids of spawned processes

370-- | 'SpawnRef' are used to return pids of spawned processes

@@ -335,9 +388,9 @@ data NCMsg = NCMsg

335

388

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

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

337data ProcessSignal =

390data ProcessSignal =

338− Link ProcessId

391+ Link Identifier

339− | Unlink ProcessId

392+ | Unlink Identifier

340− | Monitor ProcessId MonitorRef

393+ | Monitor MonitorRef

341 | Unmonitor MonitorRef

394 | Unmonitor MonitorRef

342 | Died Identifier DiedReason

395 | Died Identifier DiedReason

343 | Spawn (Closure (Process ())) SpawnRef

396 | Spawn (Closure (Process ())) SpawnRef

@@ -355,22 +408,30 @@ instance Binary ProcessId where

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

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

356 get = ProcessId <$> get <*> get

409 get = ProcessId <$> get <*> get

357

410

358−instance Binary MonitorNotification where

411+instance Binary ProcessMonitorNotification where

412+ put (ProcessMonitorNotification ref pid reason) = put ref >> put pid >> put reason

413+ get = ProcessMonitorNotification <$> get <*> get <*> get

414+

415+instance Binary NodeMonitorNotification where

416+ put (NodeMonitorNotification ref pid reason) = put ref >> put pid >> put reason

417+ get = NodeMonitorNotification <$> get <*> get <*> get

418+

419+instance Binary PortMonitorNotification where

359− put (MonitorNotification ref pid reason) = put ref >> put pid >> put reason

420+ put (PortMonitorNotification ref pid reason) = put ref >> put pid >> put reason

360− get = MonitorNotification <$> get <*> get <*> get

421+ get = PortMonitorNotification <$> get <*> get <*> get

361

422

362instance Binary NCMsg where

423instance Binary NCMsg where

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

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

364 get = NCMsg <$> get <*> get

425 get = NCMsg <$> get <*> get

365

426

366instance Binary MonitorRef where

427instance Binary MonitorRef where

367− put ref = put (monitorRefPid ref) >> put (monitorRefCounter ref)

428+ put ref = put (monitorRefIdent ref) >> put (monitorRefCounter ref)

368 get = MonitorRef <$> get <*> get

429 get = MonitorRef <$> get <*> get

369

430

370instance Binary ProcessSignal where

431instance Binary ProcessSignal where

371 put (Link pid) = putWord8 0 >> put pid

432 put (Link pid) = putWord8 0 >> put pid

372 put (Unlink pid) = putWord8 1 >> put pid

433 put (Unlink pid) = putWord8 1 >> put pid

373− put (Monitor pid ref) = putWord8 2 >> put pid >> put ref

434+ put (Monitor ref) = putWord8 2 >> put ref

374 put (Unmonitor ref) = putWord8 3 >> put ref

435 put (Unmonitor ref) = putWord8 3 >> put ref

375 put (Died who reason) = putWord8 4 >> put who >> put reason

436 put (Died who reason) = putWord8 4 >> put who >> put reason

376 put (Spawn proc ref) = putWord8 5 >> put proc >> put ref

437 put (Spawn proc ref) = putWord8 5 >> put proc >> put ref

@@ -379,7 +440,7 @@ instance Binary ProcessSignal where

379 case header of

440 case header of

380 0 -> Link <$> get

441 0 -> Link <$> get

381 1 -> Unlink <$> get

442 1 -> Unlink <$> get

382− 2 -> Monitor <$> get <*> get

443+ 2 -> Monitor <$> get

383 3 -> Unmonitor <$> get

444 3 -> Unmonitor <$> get

384 4 -> Died <$> get <*> get

445 4 -> Died <$> get <*> get

385 5 -> Spawn <$> get <*> get

446 5 -> Spawn <$> get <*> get

@@ -390,7 +451,7 @@ instance Binary DiedReason where

390 put (DiedException e) = putWord8 1 >> put e

451 put (DiedException e) = putWord8 1 >> put e

391 put DiedDisconnect = putWord8 2

452 put DiedDisconnect = putWord8 2

392 put DiedNodeDown = putWord8 3

453 put DiedNodeDown = putWord8 3

393− put DiedNoProc = putWord8 4

454+ put DiedUnknownId = putWord8 4

394 get = do

455 get = do

395 header <- getWord8

456 header <- getWord8

396 case header of

457 case header of

@@ -398,7 +459,7 @@ instance Binary DiedReason where

398 1 -> DiedException <$> get

459 1 -> DiedException <$> get

399 2 -> return DiedDisconnect

460 2 -> return DiedDisconnect

400 3 -> return DiedNodeDown

461 3 -> return DiedNodeDown

401− 4 -> return DiedNoProc

462+ 4 -> return DiedUnknownId

402 _ -> fail "DiedReason.get: invalid"

463 _ -> fail "DiedReason.get: invalid"

403

464

404instance Binary (Closure a) where

465instance Binary (Closure a) where

@@ -409,20 +470,20 @@ instance Binary DidSpawn where

409 put (DidSpawn ref pid) = put ref >> put pid

470 put (DidSpawn ref pid) = put ref >> put pid

410 get = DidSpawn <$> get <*> get

471 get = DidSpawn <$> get <*> get

411

472

412−instance Binary ChannelId where

473+instance Binary SendPortId where

413− put cid = put (channelProcessId cid) >> put (channelLocalId cid)

474+ put cid = put (sendPortProcessId cid) >> put (sendPortLocalId cid)

414− get = ChannelId <$> get <*> get

475+ get = SendPortId <$> get <*> get

415

476

416instance Binary Identifier where

477instance Binary Identifier where

417− put (ProcessIdentifier pid) = putWord8 0 >> put pid

478+ put (ProcessIdentifier pid) = putWord8 0 >> put pid

418− put (NodeIdentifier nid) = putWord8 1 >> put nid

479+ put (NodeIdentifier nid) = putWord8 1 >> put nid

419− put (ChannelIdentifier cid) = putWord8 2 >> put cid

480+ put (SendPortIdentifier cid) = putWord8 2 >> put cid

420 get = do

481 get = do

421 header <- getWord8

482 header <- getWord8

422 case header of

483 case header of

423 0 -> ProcessIdentifier <$> get

484 0 -> ProcessIdentifier <$> get

424 1 -> NodeIdentifier <$> get

485 1 -> NodeIdentifier <$> get

425− 2 -> ChannelIdentifier <$> get

486+ 2 -> SendPortIdentifier <$> get

426 _ -> fail "Identifier.get: invalid"

487 _ -> fail "Identifier.get: invalid"

427

488

428instance Binary StaticLabel where

489instance Binary StaticLabel where

@@ -485,13 +546,13 @@ monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt

485spawnCounter :: Accessor LocalProcessState Int32

546spawnCounter :: Accessor LocalProcessState Int32

486spawnCounter = accessor _spawnCounter (\cnt st -> st { _spawnCounter = cnt })

547spawnCounter = accessor _spawnCounter (\cnt st -> st { _spawnCounter = cnt })

487

548

488−channelCounter :: Accessor LocalProcessState LocalChannelId

549+channelCounter :: Accessor LocalProcessState LocalSendPortId

489channelCounter = accessor _channelCounter (\cnt st -> st { _channelCounter = cnt })

550channelCounter = accessor _channelCounter (\cnt st -> st { _channelCounter = cnt })

490

551

491−typedChannels :: Accessor LocalProcessState (Map LocalChannelId TypedChannel)

552+typedChannels :: Accessor LocalProcessState (Map LocalSendPortId TypedChannel)

492typedChannels = accessor _typedChannels (\cs st -> st { _typedChannels = cs })

553typedChannels = accessor _typedChannels (\cs st -> st { _typedChannels = cs })

493

554

494−typedChannelWithId :: LocalChannelId -> Accessor LocalProcessState (Maybe TypedChannel)

555+typedChannelWithId :: LocalSendPortId -> Accessor LocalProcessState (Maybe TypedChannel)

495typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid

556typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid

496

557

497remoteTableLabels :: Accessor RemoteTable (Map String Dynamic)

558remoteTableLabels :: Accessor RemoteTable (Map String Dynamic)

distributed-process/src/Control/Distributed/Process/Node.hs+165−105

@@ -13,7 +13,14 @@ import System.IO (fixIO)

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

14import Data.Binary (decode)

15import Data.Map (Map)

16−import qualified Data.Map as Map (empty, lookup, insert, delete, toList)

16+import qualified Data.Map as Map

17+ ( empty

18+ , lookup

19+ , insert

20+ , delete

21+ , toList

22+ , partitionWithKey

23+ )

17import qualified Data.List as List (delete)

24import qualified Data.List as List (delete)

18import Data.Set (Set)

25import Data.Set (Set)

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

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

@@ -73,18 +80,26 @@ import Control.Distributed.Process.Internal.Types

73 , localPidUnique

80 , localPidUnique

74 , localProcessWithId

81 , localProcessWithId

75 , MonitorRef(..)

82 , MonitorRef(..)

76− , MonitorNotification(..)

83+ , ProcessMonitorNotification(..)

77− , LinkException(..)

84+ , NodeMonitorNotification(..)

85+ , PortMonitorNotification(..)

86+ , ProcessLinkException(..)

87+ , NodeLinkException(..)

88+ , PortLinkException(..)

78 , DidUnmonitor(..)

89 , DidUnmonitor(..)

79− , DidUnlink(..)

90+ , DidUnlinkProcess(..)

91+ , DidUnlinkNode(..)

92+ , DidUnlinkPort(..)

80 , SpawnRef

93 , SpawnRef

81 , DidSpawn(..)

94 , DidSpawn(..)

82 , Closure(..)

95 , Closure(..)

83 , Static(..)

96 , Static(..)

97+ , Message

84 , MessageT

98 , MessageT

85 , TypedChannel(..)

99 , TypedChannel(..)

86 , Identifier(..)

100 , Identifier(..)

87− , ChannelId(..)

101+ , nodeOf

102+ , SendPortId(..)

88 , typedChannelWithId

103 , typedChannelWithId

89 )

104 )

90import Control.Distributed.Process.Serializable (Serializable)

105import Control.Distributed.Process.Serializable (Serializable)

@@ -249,9 +264,9 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

249 -- incoming connections (this requires a Transport layer

264 -- incoming connections (this requires a Transport layer

250 -- extension). (**)

265 -- extension). (**)

251 go (List.delete cid uninitConns) procs chans ctrls

266 go (List.delete cid uninitConns) procs chans ctrls

252− ChannelIdentifier chId -> do

267+ SendPortIdentifier chId -> do

253− let lcid = channelLocalId chId

268+ let lcid = sendPortLocalId chId

254− lpid = processLocalId (channelProcessId chId)

269+ lpid = processLocalId (sendPortProcessId chId)

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

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

256 case mProc of

271 case mProc of

257 Just proc -> do

272 Just proc -> do

@@ -325,9 +340,9 @@ runNodeController node =

325

340

326data NCState = NCState

341data NCState = NCState

327 { -- Mapping from remote processes to linked local processes

342 { -- Mapping from remote processes to linked local processes

328− _links :: Map NodeId (Map LocalProcessId (Set ProcessId))

343+ _links :: Map Identifier (Set ProcessId)

329 -- Mapping from remote processes to monitoring local processes

344 -- Mapping from remote processes to monitoring local processes

330− , _monitors :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))

345+ , _monitors :: Map Identifier (Set (ProcessId, MonitorRef))

331 }

346 }

332

347

333newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }

348newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }

@@ -362,24 +377,22 @@ nodeController = do

362 case msg of

377 case msg of

363 NCMsg (ProcessIdentifier from) (Link them) ->

378 NCMsg (ProcessIdentifier from) (Link them) ->

364 ncEffectMonitor from them Nothing

379 ncEffectMonitor from them Nothing

365− NCMsg (ProcessIdentifier from) (Monitor them ref) ->

380+ NCMsg (ProcessIdentifier from) (Monitor ref) ->

366− ncEffectMonitor from them (Just ref)

381+ ncEffectMonitor from (monitorRefIdent ref) (Just ref)

367 NCMsg (ProcessIdentifier from) (Unlink them) ->

382 NCMsg (ProcessIdentifier from) (Unlink them) ->

368 ncEffectUnlink from them

383 ncEffectUnlink from them

369 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->

384 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->

370 ncEffectUnmonitor from ref

385 ncEffectUnmonitor from ref

371− NCMsg _from (Died (NodeIdentifier nid) reason) ->

386+ NCMsg _from (Died ident reason) ->

372− ncEffectNodeDied nid reason

387+ ncEffectDied ident reason

373− NCMsg _from (Died (ProcessIdentifier pid) reason) ->

374− ncEffectProcessDied pid reason

375 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->

388 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->

376 ncEffectSpawn from proc ref

389 ncEffectSpawn from proc ref

377 unexpected ->

390 unexpected ->

378− error $ "Unexpected message " ++ show unexpected

391+ error $ "nodeController: unexpected message " ++ show unexpected

379

392

380-- [Unified: Table 10]

393-- [Unified: Table 10]

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

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

382− -> ProcessId -- ^ Who's being watched?

395+ -> Identifier -- ^ Who's being watched?

383 -> Maybe MonitorRef -- ^ 'Nothing' to link

396 -> Maybe MonitorRef -- ^ 'Nothing' to link

384 -> NC ()

397 -> NC ()

385ncEffectMonitor from them mRef = do

398ncEffectMonitor from them mRef = do

@@ -387,73 +400,62 @@ ncEffectMonitor from them mRef = do

387 shouldLink <-

400 shouldLink <-

388 if not (isLocal node them)

401 if not (isLocal node them)

389 then return True

402 then return True

390− else liftIO . withMVar (localState node) $

403+ else isValidLocalIdentifier them

391− return . isJust . (^. localProcessWithId (processLocalId them))

392− case (shouldLink, isLocal node from) of

404+ case (shouldLink, isLocal node (ProcessIdentifier from)) of

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

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

394 case mRef of

406 case mRef of

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

407+ Just ref -> modify $ monitorsFor them ^: Set.insert (from, ref)

396− Nothing -> modify $ linksForProcess them ^: Set.insert from

408+ Nothing -> modify $ linksFor them ^: Set.insert from

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

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

398− notifyDied from them DiedNoProc mRef

410+ notifyDied from them DiedUnknownId mRef

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

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

400 ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg

412 ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg

401 { ctrlMsgSender = NodeIdentifier (localNodeId node)

413 { ctrlMsgSender = NodeIdentifier (localNodeId node)

402− , ctrlMsgSignal = Died (ProcessIdentifier them) DiedNoProc

414+ , ctrlMsgSignal = Died them DiedUnknownId

403 }

415 }

404

416

405-- [Unified: Table 11]

417-- [Unified: Table 11]

406−ncEffectUnlink :: ProcessId -> ProcessId -> NC ()

418+ncEffectUnlink :: ProcessId -> Identifier -> NC ()

407ncEffectUnlink from them = do

419ncEffectUnlink from them = do

408 node <- ncMsg getLocalNode

420 node <- ncMsg getLocalNode

409− when (isLocal node from) $ postMessage from $ DidUnlink them

421+ when (isLocal node (ProcessIdentifier from)) $

422+ case them of

423+ ProcessIdentifier pid ->

424+ postAsMessage from $ DidUnlinkProcess pid

425+ NodeIdentifier nid ->

426+ postAsMessage from $ DidUnlinkNode nid

427+ SendPortIdentifier cid ->

428+ postAsMessage from $ DidUnlinkPort cid

410− modify $ linksForProcess them ^: Set.delete from

429+ modify $ linksFor them ^: Set.delete from

411

430

412-- [Unified: Table 11]

431-- [Unified: Table 11]

413ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()

432ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()

414ncEffectUnmonitor from ref = do

433ncEffectUnmonitor from ref = do

415 node <- ncMsg getLocalNode

434 node <- ncMsg getLocalNode

435+ when (isLocal node (ProcessIdentifier from)) $

416− when (isLocal node from) $ postMessage from $ DidUnmonitor ref

436+ postAsMessage from $ DidUnmonitor ref

417− modify $ monitorsFor (monitorRefPid ref) from ^: Set.delete ref

437+ modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)

418−

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

420−ncEffectNodeDied :: NodeId -> DiedReason -> NC ()

421−ncEffectNodeDied nid reason = do

422− node <- ncMsg getLocalNode

423− lnks <- gets (^. linksForNode nid)

424− mons <- gets (^. monitorsForNode nid)

425−

426− -- We only need to notify local processes

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

428− let pid = ProcessId nid them

429− forM_ uss $ \us ->

430− when (isLocal node us) $

431− notifyDied us pid reason Nothing

432

438

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

439+-- [Unified: Table 12]

434− let pid = ProcessId nid them

440+ncEffectDied :: Identifier -> DiedReason -> NC ()

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

441+ncEffectDied ident reason = do

436− when (isLocal node us) $

442+ node <- ncMsg getLocalNode

437− forM_ refs $ notifyDied us pid reason . Just

443+ (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))

438−

444+ (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))

439− modify $ (linksForNode nid ^= Map.empty)

440− . (monitorsForNode nid ^= Map.empty)

441

445

442−-- [Unified: Table 12, top rule]

446+ let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False

443−ncEffectProcessDied :: ProcessId -> DiedReason -> NC ()

444−ncEffectProcessDied pid reason = do

445− lnks <- gets (^. linksForProcess pid)

446− mons <- gets (^. monitorsForProcess pid)

447

448− forM_ lnks $ \us ->

448+ forM_ (Map.toList affectedLinks) $ \(them, uss) ->

449+ forM_ uss $ \us ->

450+ when (localOnly <= isLocal node (ProcessIdentifier us)) $

449− notifyDied us pid reason Nothing

451+ notifyDied us them reason Nothing

450

452

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

453+ forM_ (Map.toList affectedMons) $ \(them, refs) ->

452− forM_ refs $

454+ forM_ refs $ \(us, ref) ->

455+ when (localOnly <= isLocal node (ProcessIdentifier us)) $

453− notifyDied us pid reason . Just

456+ notifyDied us them reason (Just ref)

454

457

455− modify $ (linksForProcess pid ^= Set.empty)

458+ modify $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)

456− . (monitorsForProcess pid ^= Map.empty)

457

459

458-- [Unified: Table 13]

460-- [Unified: Table 13]

459ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()

461ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()

@@ -471,46 +473,48 @@ ncEffectSpawn pid cProc ref = do

471--------------------------------------------------------------------------------

473--------------------------------------------------------------------------------

472

474

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

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

474− -> ProcessId -- ^ Who died?

476+ -> Identifier -- ^ Who died?

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

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

476 -> Maybe MonitorRef -- ^ 'Nothing' for linking

478 -> Maybe MonitorRef -- ^ 'Nothing' for linking

477 -> NC ()

479 -> NC ()

478notifyDied dest src reason mRef = do

480notifyDied dest src reason mRef = do

479 node <- ncMsg getLocalNode

481 node <- ncMsg getLocalNode

480− case (isLocal node dest, mRef) of

482+ case (isLocal node (ProcessIdentifier dest), mRef, src) of

481− (True, Just ref) ->

483+ (True, Just ref, ProcessIdentifier pid) ->

482− postMessage dest $ MonitorNotification ref src reason

484+ postAsMessage dest $ ProcessMonitorNotification ref pid reason

483− (True, Nothing) ->

485+ (True, Just ref, NodeIdentifier nid) ->

486+ postAsMessage dest $ NodeMonitorNotification ref nid reason

487+ (True, Just ref, SendPortIdentifier cid) ->

488+ postAsMessage dest $ PortMonitorNotification ref cid reason

489+ (True, Nothing, ProcessIdentifier pid) ->

484− throwException dest $ LinkException src reason

490+ throwException dest $ ProcessLinkException pid reason

491+ (True, Nothing, NodeIdentifier pid) ->

492+ throwException dest $ NodeLinkException pid reason

493+ (True, Nothing, SendPortIdentifier pid) ->

494+ throwException dest $ PortLinkException pid reason

485− (False, _) ->

495+ (False, _, _) ->

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

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

487 -- (see [Unified: Table 10]

497 -- (see [Unified: Table 10]

488 ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg

498 ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg

489 { ctrlMsgSender = NodeIdentifier (localNodeId node)

499 { ctrlMsgSender = NodeIdentifier (localNodeId node)

490− , ctrlMsgSignal = Died (ProcessIdentifier src) reason

500+ , ctrlMsgSignal = Died src reason

491 }

501 }

492

502

493-- | [Unified: Table 8]

503-- | [Unified: Table 8]

494destNid :: ProcessSignal -> Maybe NodeId

504destNid :: ProcessSignal -> Maybe NodeId

495−destNid (Link pid) =

505+destNid (Link ident) = Just $ nodeOf ident

496− Just . processNodeId $ pid

497−destNid (Unlink pid) =

506+destNid (Unlink ident) = Just $ nodeOf ident

498− Just . processNodeId $ pid

499−destNid (Monitor pid _) =

507+destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)

500− Just . processNodeId $ pid

501−destNid (Unmonitor ref) =

508+destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)

502− Just . processNodeId . monitorRefPid $ ref

509+destNid (Spawn _ _) = Nothing

503-- We don't need to forward 'Died' signals; if monitoring/linking is setup,

510-- We don't need to forward 'Died' signals; if monitoring/linking is setup,

504-- then when a local process dies the monitoring/linking machinery will take

511-- then when a local process dies the monitoring/linking machinery will take

505-- care of notifying remote nodes

512-- care of notifying remote nodes

506−destNid (Died _ _) =

513+destNid (Died _ _) = Nothing

507− Nothing

508−destNid (Spawn _ _) =

509− Nothing

510

514

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

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

512−isLocal :: LocalNode -> ProcessId -> Bool

516+isLocal :: LocalNode -> Identifier -> Bool

513−isLocal nid pid = processNodeId pid == localNodeId nid

517+isLocal nid ident = nodeOf ident == localNodeId nid

514

518

515-- | Lookup a local closure

519-- | Lookup a local closure

516unClosure :: Typeable a => Closure a -> NC (Maybe a)

520unClosure :: Typeable a => Closure a -> NC (Maybe a)

@@ -518,13 +522,35 @@ unClosure (Closure (Static label) env) = do

518 rtable <- remoteTable <$> ncMsg getLocalNode

522 rtable <- remoteTable <$> ncMsg getLocalNode

519 return (resolveClosure rtable label env >>= fromDynamic)

523 return (resolveClosure rtable label env >>= fromDynamic)

520

524

525+-- | Check if an identifier refers to a valid local object

526+isValidLocalIdentifier :: Identifier -> NC Bool

527+isValidLocalIdentifier ident = do

528+ node <- ncMsg getLocalNode

529+ liftIO . withMVar (localState node) $ \nSt ->

530+ case ident of

531+ NodeIdentifier nid ->

532+ return $ nid == localNodeId node

533+ ProcessIdentifier pid -> do

534+ let mProc = nSt ^. localProcessWithId (processLocalId pid)

535+ return $ isJust mProc

536+ SendPortIdentifier cid -> do

537+ let pid = sendPortProcessId cid

538+ mProc = nSt ^. localProcessWithId (processLocalId pid)

539+ case mProc of

540+ Nothing -> return False

541+ Just proc -> withMVar (processState proc) $ \pSt -> do

542+ let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)

543+ return $ isJust mCh

544+

521--------------------------------------------------------------------------------

545--------------------------------------------------------------------------------

522-- Messages to local processes --

546-- Messages to local processes --

523--------------------------------------------------------------------------------

547--------------------------------------------------------------------------------

524

548

525−postMessage :: Serializable a => ProcessId -> a -> NC ()

549+postAsMessage :: Serializable a => ProcessId -> a -> NC ()

550+postAsMessage pid = postMessage pid . createMessage

551+

552+postMessage :: ProcessId -> Message -> NC ()

526−postMessage pid msg = withLocalProc pid $ \p ->

553+postMessage pid msg = withLocalProc pid $ \p -> enqueue (processQueue p) msg

527− enqueue (processQueue p) (createMessage msg)

528

554

529throwException :: Exception e => ProcessId -> e -> NC ()

555throwException :: Exception e => ProcessId -> e -> NC ()

530throwException pid e = withLocalProc pid $ \p ->

556throwException pid e = withLocalProc pid $ \p ->

@@ -544,23 +570,57 @@ withLocalProc pid p = do

544-- Accessors --

570-- Accessors --

545--------------------------------------------------------------------------------

571--------------------------------------------------------------------------------

546

572

547−links :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))

573+links :: Accessor NCState (Map Identifier (Set ProcessId))

548−links = accessor _links (\lnks st -> st { _links = lnks })

574+links = accessor _links (\ls st -> st { _links = ls })

549−

550−monitors :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))

551−monitors = accessor _monitors (\mons st -> st { _monitors = mons })

552

575

553−linksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))

576+monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))

554−linksForNode nid = links >>> DAC.mapDefault Map.empty nid

577+monitors = accessor _monitors (\ms st -> st { _monitors = ms })

555

578

556−monitorsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))

579+linksFor :: Identifier -> Accessor NCState (Set ProcessId)

557−monitorsForNode nid = monitors >>> DAC.mapDefault Map.empty nid

580+linksFor ident = links >>> DAC.mapDefault Set.empty ident

558

581

559−linksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)

582+monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))

560−linksForProcess pid = linksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)

583+monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident

561

584

562−monitorsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))

585+-- | @splitNotif ident@ splits a notifications map into those

563−monitorsForProcess pid = monitorsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)

586+-- notifications that should trigger when 'ident' fails and those links that

564−

587+-- should not.

565−monitorsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)

588+--

566−monitorsFor them us = monitorsForProcess them >>> DAC.mapDefault Set.empty us

589+-- There is a hierarchy between identifiers: failure of a node implies failure

590+-- of all processes on that node, and failure of a process implies failure of

591+-- all typed channels to that process. In other words, if 'ident' refers to a

592+-- node, then the /should trigger/ set will include

593+--

594+-- * the notifications for the node specifically

595+-- * the notifications for processes on that node, and

596+-- * the notifications for typed channels to processes on that node.

597+--

598+-- Similarly, if 'ident' refers to a process, the /should trigger/ set will

599+-- include

600+--

601+-- * the notifications for that process specifically and

602+-- * the notifications for typed channels to that process.

603+splitNotif :: Identifier

604+ -> Map Identifier a

605+ -> (Map Identifier a, Map Identifier a)

606+splitNotif ident = Map.partitionWithKey (const . impliesDeathOf ident)

607+ where

608+ -- | Does the death of one entity (node, project, channel) imply the death

609+ -- of another?

610+ impliesDeathOf :: Identifier -- ^ Who died

611+ -> Identifier -- ^ Who's being watched

612+ -> Bool -- ^ Does this death implies the death of the watchee?

613+ NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' =

614+ nid' == nid

615+ NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =

616+ processNodeId pid == nid

617+ NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =

618+ processNodeId (sendPortProcessId cid) == nid

619+ ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =

620+ pid' == pid

621+ ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =

622+ sendPortProcessId cid == pid

623+ SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =

624+ cid' == cid

625+ _ `impliesDeathOf` _ =

626+ False

distributed-process/tests/TestCH.hs+42−10

@@ -94,26 +94,22 @@ monitorTestProcess theirAddr mOrL un reason monitorSetup done =

94 case (un, mRef) of

95 (True, Nothing) -> do

96 unlink theirAddr

97− DidUnlink pid <- expect

98− True <- return $ pid == theirAddr

99 liftIO $ putMVar done ()

97 liftIO $ putMVar done ()

100 (True, Just ref) -> do

98 (True, Just ref) -> do

101 unmonitor ref

99 unmonitor ref

102− DidUnmonitor ref' <- expect

103− True <- return $ ref == ref'

104 liftIO $ putMVar done ()

100 liftIO $ putMVar done ()

105 (False, ref) -> do

101 (False, ref) -> do

106− MonitorNotification ref' pid reason' <- expect

102+ ProcessMonitorNotification ref' pid reason' <- expect

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

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

108 liftIO $ putMVar done ()

104 liftIO $ putMVar done ()

109 )

105 )

110− (\(LinkException pid reason') -> do

106+ (\(ProcessLinkException pid reason') -> do

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

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

112 liftIO $ putMVar done ()

108 liftIO $ putMVar done ()

113 )

109 )

114

110

115

111

116−-- | Monitor an unreachable node

112+-- | Monitor a process on an unreachable node

117testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> IO ()

113testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> IO ()

118testMonitorUnreachable transport mOrL un = do

114testMonitorUnreachable transport mOrL un = do

119 deadProcess <- newEmptyMVar

115 deadProcess <- newEmptyMVar

@@ -194,7 +190,7 @@ testMonitorLocalDeadProcess transport mOrL un = do

194 theirAddr <- readMVar processAddr

190 theirAddr <- readMVar processAddr

195 readMVar processDead

191 readMVar processDead

196 runProcess localNode $ do

192 runProcess localNode $ do

197− monitorTestProcess theirAddr mOrL un DiedNoProc Nothing done

193+ monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done

198

194

199 takeMVar done

195 takeMVar done

200

196

@@ -215,7 +211,7 @@ testMonitorRemoteDeadProcess transport mOrL un = do

215 theirAddr <- readMVar processAddr

211 theirAddr <- readMVar processAddr

216 readMVar processDead

212 readMVar processDead

217 runProcess localNode $ do

213 runProcess localNode $ do

218− monitorTestProcess theirAddr mOrL un DiedNoProc Nothing done

214+ monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done

219

215

220 takeMVar done

216 takeMVar done

221

217

@@ -493,10 +489,41 @@ testTerminate transport = do

493

489

494 runProcess localNode $ do

490 runProcess localNode $ do

495 ref <- monitor pid

491 ref <- monitor pid

496− MonitorNotification ref' pid' (DiedException ex) <- expect

492+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect

497 True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException

493 True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException

498 return ()

494 return ()

499

495

496+testMonitorNode :: NT.Transport -> IO ()

497+testMonitorNode transport = do

498+ [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable

499+

500+ closeLocalNode node1

501+

502+ runProcess node2 $ do

503+ ref <- monitorNode (localNodeId node1)

504+ NodeMonitorNotification ref' nid DiedDisconnect <- expect

505+ True <- return $ ref == ref' && nid == localNodeId node1

506+ return ()

507+

508+testMonitorChannel :: NT.Transport -> IO ()

509+testMonitorChannel transport = do

510+ [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable

511+ gotNotification <- newEmptyMVar

512+

513+ pid <- forkProcess node1 $ do

514+ sport <- expect :: Process (SendPort ())

515+ ref <- monitorPort sport

516+ PortMonitorNotification ref' port' DiedNormal <- expect

517+ return $ ref' == ref && port' == sendPortId sport

518+ liftIO $ putMVar gotNotification ()

519+

520+ runProcess node2 $ do

521+ (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())

522+ send pid sport

523+ liftIO $ threadDelay 100000

524+

525+ takeMVar gotNotification

526+

500main :: IO ()

527main :: IO ()

501main = do

528main = do

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

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

@@ -509,6 +536,8 @@ main = do

509 , ("TypedChannnels", testTypedChannels transport)

536 , ("TypedChannnels", testTypedChannels transport)

510 , ("MergeChannels", testMergeChannels transport)

537 , ("MergeChannels", testMergeChannels transport)

511 , ("TestTerminate", testTerminate transport)

538 , ("TestTerminate", testTerminate transport)

539+ -- Monitoring processes

540+ --

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

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

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

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

514 -- happen before the unmonitor or unlink

543 -- happen before the unmonitor or unlink

@@ -530,4 +559,7 @@ main = do

530 , ("UnlinkNormalTermination", testMonitorNormalTermination transport False True)

559 , ("UnlinkNormalTermination", testMonitorNormalTermination transport False True)

531 , ("UnlinkAbnormalTermination", testMonitorAbnormalTermination transport False True)

560 , ("UnlinkAbnormalTermination", testMonitorAbnormalTermination transport False True)

532 , ("UnlinkDisconnect", testMonitorDisconnect transport False True)

561 , ("UnlinkDisconnect", testMonitorDisconnect transport False True)

562+ -- Monitoring nodes and channels

563+ , ("MonitorNode", testMonitorNode transport)

564+ , ("MonitorChannel", testMonitorChannel transport)

533 ]

565 ]

distributed-process/tests/TestClosure.hs+3−3

@@ -235,10 +235,10 @@ testSpawnSupervised transport rtable = do

235 forkProcess node2 $ do

236 [super, child] <- liftIO $ mapM readMVar [superPid, childPid]

237 ref <- monitor child

238− MonitorNotification ref' pid' (DiedException e) <- expect

238+ ProcessMonitorNotification ref' pid' (DiedException e) <- expect

239 True <- return $ ref' == ref

240 && pid' == child

241− && e == show (LinkException super (DiedException (show supervisorDeath)))

241+ && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))

242 liftIO $ putMVar thirdProcessDone ()

243

244 takeMVar thirdProcessDone

@@ -251,7 +251,7 @@ testSpawnInvalid transport rtable = do

251 node <- newLocalNode transport rtable

252 runProcess node $ do

253 (pid, ref) <- spawnMonitor (localNodeId node) (Closure (Static (UserStatic "ThisDoesNotExist")) BSL.empty)

254− MonitorNotification ref' pid' _reason <- expect

254+ ProcessMonitorNotification ref' pid' _reason <- expect

255 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure

256 True <- return $ ref' == ref && pid == pid'

257 return ()