4d0f927 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Some infrastructure for typed channels

4d0f927Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+43−5

compare with master

unifiedsplit

@@ -15,6 +15,10 @@ module Control.Distributed.Process

15 -- * Basic messaging

16 , send

17 , expect

18+ -- * Channels

19+ , ReceivePort

20+ , SendPort

21+ , newChan

18 -- * Advanced messaging

22 -- * Advanced messaging

19 , Match

23 , Match

20 , receiveWait

24 , receiveWait

@@ -62,9 +66,13 @@ import Control.Exception (Exception, throw)

62import qualified Control.Exception as Exception (catch)

66import qualified Control.Exception as Exception (catch)

63import Control.Concurrent.MVar (modifyMVar)

67import Control.Concurrent.MVar (modifyMVar)

64import Control.Concurrent.Chan (writeChan)

68import Control.Concurrent.Chan (writeChan)

65−import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..))

69+import Control.Distributed.Process.Internal.CQueue

70+ ( newCQueue

71+ , dequeue

72+ , BlockSpec(..)

73+ )

66import Control.Distributed.Process.Serializable (Serializable, fingerprint)

74import Control.Distributed.Process.Serializable (Serializable, fingerprint)

67−import Data.Accessor ((^.), (^:))

75+import Data.Accessor ((^.), (^:), (^=))

68import Control.Distributed.Process.Internal.Types

76import Control.Distributed.Process.Internal.Types

69 ( RemoteTable

77 ( RemoteTable

70 , NodeId(..)

78 , NodeId(..)

@@ -90,6 +98,13 @@ import Control.Distributed.Process.Internal.Types

90 , spawnCounter

98 , spawnCounter

91 , MessageT

99 , MessageT

92 , Closure(..)

100 , Closure(..)

101+ , SendPort(..)

102+ , ReceivePort(..)

103+ , channelCounter

104+ , typedChannelWithId

105+ , TypedChannel(..)

106+ , ChannelId(..)

107+ , Identifier(..)

93 )

108 )

94import Control.Distributed.Process.Internal.MessageT

109import Control.Distributed.Process.Internal.MessageT

95 ( sendMessage

110 ( sendMessage

@@ -164,12 +179,35 @@ import Control.Distributed.Process.Internal.MessageT

164send :: Serializable a => ProcessId -> a -> Process ()

179send :: Serializable a => ProcessId -> a -> Process ()

165-- This requires a lookup on every send. If we want to avoid that we need to

180-- This requires a lookup on every send. If we want to avoid that we need to

166-- modify serializable to allow for stateful (IO) deserialization

181-- modify serializable to allow for stateful (IO) deserialization

167−send them msg = lift $ sendMessage (Left them) msg

182+send them msg = lift $ sendMessage (ProcessIdentifier them) msg

168

183

169-- | Wait for a message of a specific type

184-- | Wait for a message of a specific type

170expect :: forall a. Serializable a => Process a

185expect :: forall a. Serializable a => Process a

171expect = receiveWait [match return]

186expect = receiveWait [match return]

172

187

188+--------------------------------------------------------------------------------

189+-- Channels --

190+--------------------------------------------------------------------------------

191+

192+-- | Create a new typed channel (WORK IN PROGRESS)

193+newChan :: Serializable a => Process (SendPort a, ReceivePort a)

194+newChan = do

195+ proc <- ask

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

197+ queue <- liftIO newCQueue

198+ let lcid = st ^. channelCounter

199+ cid = ChannelId { channelProcessId = processId proc

200+ , channelLocalId = lcid

201+ }

202+ sport = SendPort cid

203+ rport = ReceivePort queue

204+ tch = TypedChannel queue

205+ return ( (channelCounter ^: (+ 1))

206+ . (typedChannelWithId lcid ^= Just tch)

207+ $ st

208+ , (sport, rport)

209+ )

210+

173--------------------------------------------------------------------------------

211--------------------------------------------------------------------------------

174-- Advanced messaging --

212-- Advanced messaging --

175--------------------------------------------------------------------------------

213--------------------------------------------------------------------------------

@@ -320,7 +358,7 @@ sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node

320 -> Process ()

358 -> Process ()

321sendCtrlMsg mNid signal = do

359sendCtrlMsg mNid signal = do

322 us <- getSelfPid

360 us <- getSelfPid

323− let msg = NCMsg { ctrlMsgSender = Left us

361+ let msg = NCMsg { ctrlMsgSender = ProcessIdentifier us

324 , ctrlMsgSignal = signal

362 , ctrlMsgSignal = signal

325 }

363 }

326 case mNid of

364 case mNid of

@@ -328,7 +366,7 @@ sendCtrlMsg mNid signal = do

328 ctrlChan <- localCtrlChan <$> lift getLocalNode

366 ctrlChan <- localCtrlChan <$> lift getLocalNode

329 liftIO $ writeChan ctrlChan msg

367 liftIO $ writeChan ctrlChan msg

330 Just nid ->

368 Just nid ->

331− lift $ sendBinary (Right nid) msg

369+ lift $ sendBinary (NodeIdentifier nid) msg

332

370

333lookupStatic :: Typeable a => String -> Process a

371lookupStatic :: Typeable a => String -> Process a

334lookupStatic label = do

372lookupStatic label = do

distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+1−1

@@ -39,7 +39,7 @@ data BlockSpec =

40dequeue :: forall a b.

41 CQueue a -- ^ Queue

42− -> BlockSpec -- ^ Blocking behaviour

42+ -> BlockSpec -- ^ Blocking behaviour

43 -> [a -> Maybe b] -- ^ List of matches

44 -> IO (Maybe b) -- ^ 'Nothing' only on timeout

45dequeue (CQueue arrived incoming) blockSpec matches =

distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+10−13

@@ -22,10 +22,10 @@ import Control.Monad.State (gets, modify)

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

23import Control.Concurrent.Chan (writeChan)

24import Control.Distributed.Process.Internal.Types

25− ( Identifier

25+ ( Identifier(..)

26 , NodeId(nodeAddress)

27− , ProcessId(processNodeId, processLocalId)

27+ , ProcessId(processNodeId)

28− , idToPayload

28+ , ChannelId(channelProcessId)

29 , LocalNode(localCtrlChan, localEndPoint)

30 , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)

31 , DiedReason(DiedDisconnect)

@@ -75,12 +75,12 @@ setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)

75setupConnTo them = do

76 endPoint <- localEndPoint `liftM` getLocalNode

77 mConn <- liftIO $ NT.connect endPoint

78− addr

78+ (nodeAddress . identifierNode $ them)

79 NT.ReliableOrdered

80 NT.defaultConnectHints

81 case mConn of

82 Right conn -> do

83− didSend <- liftIO $ NT.send conn firstMsg

83+ didSend <- liftIO $ NT.send conn (BSL.toChunks . encode $ them)

84 case didSend of

85 Left _ ->

86 return Nothing

@@ -89,14 +89,6 @@ setupConnTo them = do

89 return $ Just conn

90 Left _ ->

91 return Nothing

92− where

93− (addr, firstMsg) = case them of

94− Left pid -> ( nodeAddress (processNodeId pid)

95− , idToPayload (Just $ processLocalId pid)

96− )

97− Right nid -> ( nodeAddress nid

98− , idToPayload Nothing

99− )

100

101connTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)

93connTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)

102connTo them = do

94connTo them = do

@@ -105,6 +97,11 @@ connTo them = do

105 Just conn -> return $ Just conn

97 Just conn -> return $ Just conn

106 Nothing -> setupConnTo them

98 Nothing -> setupConnTo them

107

100+identifierNode :: Identifier -> NodeId

101+identifierNode (NodeIdentifier nid) = nid

102+identifierNode (ProcessIdentifier pid) = processNodeId pid

103+identifierNode (ChannelIdentifier cid) = processNodeId (channelProcessId cid)

104+

108--------------------------------------------------------------------------------

105--------------------------------------------------------------------------------

109-- Accessors --

106-- Accessors --

110--------------------------------------------------------------------------------

107--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+61−29

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

8 NodeId(..)

9 , LocalProcessId(..)

10 , ProcessId(..)

11− , Identifier

11+ , Identifier(..)

12 -- * Local nodes and processes

13 , LocalNode(..)

14 , LocalNodeState(..)

@@ -16,6 +16,12 @@ module Control.Distributed.Process.Internal.Types

16 , LocalProcessState(..)

17 , Process(..)

18 , runLocalProcess

19+ -- * Typed channels

20+ , LocalChannelId

21+ , ChannelId(..)

22+ , TypedChannel(..)

23+ , SendPort(..)

24+ , ReceivePort(..)

19 -- * Closures

25 -- * Closures

20 , Static(..)

26 , Static(..)

21 , Closure(..)

27 , Closure(..)

@@ -37,8 +43,6 @@ module Control.Distributed.Process.Internal.Types

37 , createMessage

43 , createMessage

38 , messageToPayload

44 , messageToPayload

39 , payloadToMessage

45 , payloadToMessage

40− , idToPayload

41− , payloadToId

42 -- * Accessors

46 -- * Accessors

43 , localProcesses

47 , localProcesses

44 , localPidCounter

48 , localPidCounter

@@ -46,6 +50,9 @@ module Control.Distributed.Process.Internal.Types

46 , localProcessWithId

50 , localProcessWithId

47 , monitorCounter

51 , monitorCounter

48 , spawnCounter

52 , spawnCounter

53+ , channelCounter

54+ , typedChannels

55+ , typedChannelWithId

49 -- * MessageT monad

56 -- * MessageT monad

50 , MessageT(..)

57 , MessageT(..)

51 , MessageState(..)

58 , MessageState(..)

@@ -67,9 +74,8 @@ import Data.ByteString.Lazy (ByteString)

67import Data.Accessor (Accessor, accessor)

74import Data.Accessor (Accessor, accessor)

68import qualified Data.Accessor.Container as DAC (mapMaybe)

75import qualified Data.Accessor.Container as DAC (mapMaybe)

69import qualified Data.ByteString.Lazy as BSL (ByteString, toChunks, fromChunks, splitAt)

76import qualified Data.ByteString.Lazy as BSL (ByteString, toChunks, fromChunks, splitAt)

70−import qualified Data.ByteString as BSS (ByteString, concat, splitAt)

77+import qualified Data.ByteString as BSS (ByteString, concat)

71import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

78import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

72−import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)

73import Control.Applicative ((<$>), (<*>))

79import Control.Applicative ((<$>), (<*>))

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

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

75import Control.Monad.IO.Class (MonadIO)

81import Control.Monad.IO.Class (MonadIO)

@@ -126,8 +132,12 @@ data ProcessId = ProcessId

126instance Show ProcessId where

132instance Show ProcessId where

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

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

128

134

129−-- | Node or process identifier

135+-- | Union of all kinds of identifiers

130−type Identifier = Either ProcessId NodeId

136+data Identifier =

137+ ProcessIdentifier ProcessId

138+ | NodeIdentifier NodeId

139+ | ChannelIdentifier ChannelId

140+ deriving (Show, Eq, Ord)

131

141

132--------------------------------------------------------------------------------

142--------------------------------------------------------------------------------

133-- Local nodes and processes --

143-- Local nodes and processes --

@@ -162,6 +172,8 @@ data LocalProcess = LocalProcess

162data LocalProcessState = LocalProcessState

172data LocalProcessState = LocalProcessState

163 { _monitorCounter :: Int32

173 { _monitorCounter :: Int32

164 , _spawnCounter :: Int32

174 , _spawnCounter :: Int32

175+ , _channelCounter :: Int32

176+ , _typedChannels :: Map LocalChannelId TypedChannel

165 }

177 }

166

178

167-- | The Cloud Haskell 'Process' type

179-- | The Cloud Haskell 'Process' type

@@ -174,6 +186,23 @@ newtype Process a = Process {

174runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a

186runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a

175runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc)

187runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc)

176

188

189+--------------------------------------------------------------------------------

190+-- Typed channels --

191+--------------------------------------------------------------------------------

192+

193+type LocalChannelId = Int32

194+

195+data ChannelId = ChannelId {

196+ channelProcessId :: ProcessId

197+ , channelLocalId :: LocalChannelId

198+ }

199+ deriving (Show, Eq, Ord)

200+

201+data TypedChannel = forall a. Serializable a => TypedChannel (CQueue a)

202+

203+newtype SendPort a = SendPort ChannelId

204+newtype ReceivePort a = ReceivePort (CQueue a)

205+

177--------------------------------------------------------------------------------

206--------------------------------------------------------------------------------

178-- Closures --

207-- Closures --

179--------------------------------------------------------------------------------

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

@@ -292,28 +321,6 @@ payloadToMessage payload = Message fp msg

292 $ BSL.fromChunks payload

321 $ BSL.fromChunks payload

293 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp

322 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp

294

323

295−

296−-- | The first message we send across a connection to indicate the intended

297−-- recipient. Pass Nothing for the remote node controller

298−idToPayload :: Maybe LocalProcessId -> [BSS.ByteString]

299−idToPayload Nothing = [ NTI.encodeInt32 (0 :: Int) ]

300−idToPayload (Just lpid) = [ NTI.encodeInt32 (1 :: Int)

301− , NTI.encodeInt32 (lpidCounter lpid)

302− , NTI.encodeInt32 (lpidUnique lpid)

303− ]

304−

305−-- | Inverse of 'idToPayload'

306−payloadToId :: [BSS.ByteString] -> Maybe LocalProcessId

307−payloadToId bss = let (bs1, bss') = BSS.splitAt 4 . BSS.concat $ bss

308− (bs2, bs3) = BSS.splitAt 4 bss' in

309− case NTI.decodeInt32 bs1 :: Int of

310− 0 -> Nothing

311− 1 -> Just LocalProcessId

312− { lpidCounter = NTI.decodeInt32 bs2

313− , lpidUnique = NTI.decodeInt32 bs3

314− }

315− _ -> fail "payloadToId"

316−

317--------------------------------------------------------------------------------

324--------------------------------------------------------------------------------

318-- Binary instances --

325-- Binary instances --

319--------------------------------------------------------------------------------

326--------------------------------------------------------------------------------

@@ -380,6 +387,22 @@ instance Binary DidSpawn where

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

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

381 get = DidSpawn <$> get <*> get

388 get = DidSpawn <$> get <*> get

382

389

390+instance Binary ChannelId where

391+ put cid = put (channelProcessId cid) >> put (channelLocalId cid)

392+ get = ChannelId <$> get <*> get

393+

394+instance Binary Identifier where

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

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

397+ put (ChannelIdentifier cid) = putWord8 2 >> put cid

398+ get = do

399+ header <- getWord8

400+ case header of

401+ 0 -> ProcessIdentifier <$> get

402+ 1 -> NodeIdentifier <$> get

403+ 2 -> ChannelIdentifier <$> get

404+ _ -> fail "Identifier.get: invalid"

405+

383--------------------------------------------------------------------------------

406--------------------------------------------------------------------------------

384-- Accessors --

407-- Accessors --

385--------------------------------------------------------------------------------

408--------------------------------------------------------------------------------

@@ -402,6 +425,15 @@ monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt

402spawnCounter :: Accessor LocalProcessState Int32

425spawnCounter :: Accessor LocalProcessState Int32

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

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

404

427

428+channelCounter :: Accessor LocalProcessState LocalChannelId

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

430+

431+typedChannels :: Accessor LocalProcessState (Map LocalChannelId TypedChannel)

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

433+

434+typedChannelWithId :: LocalChannelId -> Accessor LocalProcessState (Maybe TypedChannel)

435+typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid

436+

405--------------------------------------------------------------------------------

437--------------------------------------------------------------------------------

406-- MessageT monad --

438-- MessageT monad --

407--------------------------------------------------------------------------------

439--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Node.hs+78−49

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

71 , NCMsg(..)

72 , ProcessSignal(..)

73 , payloadToMessage

74− , payloadToId

75 , localPidCounter

74 , localPidCounter

76 , localPidUnique

75 , localPidUnique

77 , localProcessWithId

76 , localProcessWithId

@@ -88,6 +87,8 @@ import Control.Distributed.Process.Internal.Types

88 , createMessage

87 , createMessage

89 , MessageT

88 , MessageT

90 , runMessageT

89 , runMessageT

90+ , TypedChannel(..)

91+ , Identifier(..)

91 )

92 )

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

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

93import Control.Distributed.Process.Internal.MessageT

94import Control.Distributed.Process.Internal.MessageT

@@ -152,6 +153,8 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do

152 }

153 }

153 pst <- newMVar LocalProcessState { _monitorCounter = 0

154 pst <- newMVar LocalProcessState { _monitorCounter = 0

154 , _spawnCounter = 0

155 , _spawnCounter = 0

156+ , _channelCounter = 0

157+ , _typedChannels = Map.empty

155 }

158 }

156 queue <- newCQueue

159 queue <- newCQueue

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

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

@@ -168,8 +171,8 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do

168 modifyMVar_ (localState node) $

171 modifyMVar_ (localState node) $

169 return . (localProcessWithId lpid ^= Nothing)

172 return . (localProcessWithId lpid ^= Nothing)

170 writeChan (localCtrlChan node) NCMsg

173 writeChan (localCtrlChan node) NCMsg

171− { ctrlMsgSender = Left pid

174+ { ctrlMsgSender = ProcessIdentifier pid

172− , ctrlMsgSignal = Died (Left pid) reason

175+ , ctrlMsgSignal = Died (ProcessIdentifier pid) reason

173 }

176 }

174 return (tid', lproc)

177 return (tid', lproc)

175

178

@@ -181,35 +184,45 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do

181 )

184 )

182

185

183handleIncomingMessages :: LocalNode -> IO ()

186handleIncomingMessages :: LocalNode -> IO ()

184−handleIncomingMessages node = go [] Map.empty Set.empty

187+handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

185 where

188 where

186 go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know

189 go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know

187 -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes

190 -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes

191+ -> Map NT.ConnectionId TypedChannel -- ^ Connections to typed channels

188 -> Set NT.ConnectionId -- ^ Connections to our controller

192 -> Set NT.ConnectionId -- ^ Connections to our controller

189 -> IO ()

193 -> IO ()

190− go uninitConns procConns ctrlConns = do

194+ go uninitConns procs chans ctrls = do

191 event <- NT.receive endpoint

195 event <- NT.receive endpoint

192 case event of

196 case event of

193 NT.ConnectionOpened cid _rel _theirAddr ->

197 NT.ConnectionOpened cid _rel _theirAddr ->

194− go (cid : uninitConns) procConns ctrlConns

198+ go (cid : uninitConns) procs chans ctrls

195 NT.Received cid payload ->

199 NT.Received cid payload ->

200+ case ( Map.lookup cid procs

196− case Map.lookup cid procConns of

201+ , Map.lookup cid chans

202+ , cid `Set.member` ctrls

203+ , cid `elem` uninitConns

204+ ) of

197− Just proc -> do

205+ (Just proc, _, _, _) -> do

198 let msg = payloadToMessage payload

206 let msg = payloadToMessage payload

199 enqueue (processQueue proc) msg

207 enqueue (processQueue proc) msg

200− go uninitConns procConns ctrlConns

208+ go uninitConns procs chans ctrls

201− Nothing | cid `Set.member` ctrlConns -> do

209+ (_, Just (TypedChannel queue), _, _) -> do

210+ enqueue queue . decode . BSL.fromChunks $ payload

211+ go uninitConns procs chans ctrls

212+ (_, _, True, _) -> do

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

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

203− go uninitConns procConns ctrlConns

214+ go uninitConns procs chans ctrls

204− Nothing | cid `elem` uninitConns ->

215+ (_, _, _, True) ->

205− case payloadToId payload of

216+ case decode (BSL.fromChunks payload) of

206− Just lpid -> do

217+ ProcessIdentifier pid -> do

218+ let lpid = processLocalId pid

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

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

208 case mProc of

220 case mProc of

209 Just proc ->

221 Just proc ->

210 go (List.delete cid uninitConns)

222 go (List.delete cid uninitConns)

211− (Map.insert cid proc procConns)

223+ (Map.insert cid proc procs)

212− ctrlConns

224+ chans

225+ ctrls

213 Nothing ->

226 Nothing ->

214 -- Request for an unknown process.

227 -- Request for an unknown process.

215 -- TODO: We should close the incoming connection here, but

228 -- TODO: We should close the incoming connection here, but

@@ -217,22 +230,27 @@ handleIncomingMessages node = go [] Map.empty Set.empty

217 -- functionality. Not sure what the right approach is.

230 -- functionality. Not sure what the right approach is.

218 -- For now, we just drop the incoming messages

231 -- For now, we just drop the incoming messages

219 go (List.delete cid uninitConns)

232 go (List.delete cid uninitConns)

220− procConns

233+ procs

221− ctrlConns

234+ chans

222− Nothing ->

235+ ctrls

236+ ChannelIdentifier _chId ->

237+ fail "Connecting to typed channels not yet implemented"

238+ NodeIdentifier _ ->

223 go (List.delete cid uninitConns)

239 go (List.delete cid uninitConns)

224− procConns

240+ procs

241+ chans

225− (Set.insert cid ctrlConns)

242+ (Set.insert cid ctrls)

226− _ ->

243+ _ ->

227 -- Unexpected message. We just drop it.

244 -- Unexpected message. We just drop it.

228− go uninitConns procConns ctrlConns

245+ go uninitConns procs chans ctrls

229 NT.ConnectionClosed cid ->

246 NT.ConnectionClosed cid ->

230 go (List.delete cid uninitConns)

247 go (List.delete cid uninitConns)

231− (Map.delete cid procConns)

248+ (Map.delete cid procs)

249+ (Map.delete cid chans)

232− (Set.delete cid ctrlConns)

250+ (Set.delete cid ctrls)

233 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do

251 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do

234 -- [Unified table 9, rule node_disconnect]

252 -- [Unified table 9, rule node_disconnect]

235− let nid = Right $ NodeId theirAddr

253+ let nid = NodeIdentifier $ NodeId theirAddr

236 writeChan ctrlChan NCMsg

254 writeChan ctrlChan NCMsg

237 { ctrlMsgSender = nid

255 { ctrlMsgSender = nid

238 , ctrlMsgSignal = Died nid DiedDisconnect

256 , ctrlMsgSignal = Died nid DiedDisconnect

@@ -291,23 +309,25 @@ nodeController = do

291

309

292 -- [Unified: Table 7, rule nc_forward]

310 -- [Unified: Table 7, rule nc_forward]

293 case destNid (ctrlMsgSignal msg) of

311 case destNid (ctrlMsgSignal msg) of

294− Just nid' | nid' /= localNodeId node -> lift $ sendBinary (Right nid') msg

312+ Just nid' | nid' /= localNodeId node ->

313+ lift $ sendBinary (NodeIdentifier nid') msg

314+ _ ->

295− _ -> return ()

315+ return ()

296

316

297 case msg of

317 case msg of

298− NCMsg (Left from) (Link them) ->

318+ NCMsg (ProcessIdentifier from) (Link them) ->

299 ncEffectMonitor from them Nothing

319 ncEffectMonitor from them Nothing

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

320+ NCMsg (ProcessIdentifier from) (Monitor them ref) ->

301 ncEffectMonitor from them (Just ref)

321 ncEffectMonitor from them (Just ref)

302− NCMsg (Left from) (Unlink them) ->

322+ NCMsg (ProcessIdentifier from) (Unlink them) ->

303 ncEffectUnlink from them

323 ncEffectUnlink from them

304− NCMsg (Left from) (Unmonitor ref) ->

324+ NCMsg (ProcessIdentifier from) (Unmonitor ref) ->

305 ncEffectUnmonitor from ref

325 ncEffectUnmonitor from ref

306− NCMsg _from (Died (Right nid) reason) ->

326+ NCMsg _from (Died (NodeIdentifier nid) reason) ->

307 ncEffectNodeDied nid reason

327 ncEffectNodeDied nid reason

308− NCMsg _from (Died (Left pid) reason) ->

328+ NCMsg _from (Died (ProcessIdentifier pid) reason) ->

309 ncEffectProcessDied pid reason

329 ncEffectProcessDied pid reason

310− NCMsg (Left from) (Spawn proc ref) ->

330+ NCMsg (ProcessIdentifier from) (Spawn proc ref) ->

311 ncEffectSpawn from proc ref

331 ncEffectSpawn from proc ref

312 unexpected ->

332 unexpected ->

313 error $ "Unexpected message " ++ show unexpected

333 error $ "Unexpected message " ++ show unexpected

@@ -332,9 +352,9 @@ ncEffectMonitor from them mRef = do

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

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

333 notifyDied from them DiedNoProc mRef

353 notifyDied from them DiedNoProc mRef

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

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

335− lift $ sendBinary (Right $ processNodeId from) NCMsg

355+ lift $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg

336− { ctrlMsgSender = Right (localNodeId node)

356+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

337− , ctrlMsgSignal = Died (Left them) DiedNoProc

357+ , ctrlMsgSignal = Died (ProcessIdentifier them) DiedNoProc

338 }

358 }

339

359

340-- [Unified: Table 11]

360-- [Unified: Table 11]

@@ -398,7 +418,7 @@ ncEffectSpawn pid cProc ref = do

398 forM_ mProc $ \proc -> do

418 forM_ mProc $ \proc -> do

399 node <- lift getLocalNode

419 node <- lift getLocalNode

400 pid' <- liftIO $ forkProcess node proc

420 pid' <- liftIO $ forkProcess node proc

401− lift $ sendMessage (Left pid) (DidSpawn ref pid')

421+ lift $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')

402

422

403--------------------------------------------------------------------------------

423--------------------------------------------------------------------------------

404-- Auxiliary --

424-- Auxiliary --

@@ -419,20 +439,29 @@ notifyDied dest src reason mRef = do

419 (False, _) ->

439 (False, _) ->

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

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

421 -- (see [Unified: Table 10]

441 -- (see [Unified: Table 10]

422− lift $ sendBinary (Right $ processNodeId dest) NCMsg

442+ lift $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg

423− { ctrlMsgSender = Right (localNodeId node)

443+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

424− , ctrlMsgSignal = Died (Left src) reason

444+ , ctrlMsgSignal = Died (ProcessIdentifier src) reason

425 }

445 }

426

446

427-- | [Unified: Table 8]

447-- | [Unified: Table 8]

428destNid :: ProcessSignal -> Maybe NodeId

448destNid :: ProcessSignal -> Maybe NodeId

449+destNid (Link pid) =

429−destNid (Link pid) = Just . processNodeId $ pid

450+ Just . processNodeId $ pid

451+destNid (Unlink pid) =

430−destNid (Unlink pid) = Just . processNodeId $ pid

452+ Just . processNodeId $ pid

453+destNid (Monitor pid _) =

431−destNid (Monitor pid _) = Just . processNodeId $ pid

454+ Just . processNodeId $ pid

455+destNid (Unmonitor ref) =

432−destNid (Unmonitor ref) = Just . processNodeId . monitorRefPid $ ref

456+ Just . processNodeId . monitorRefPid $ ref

433−destNid (Died (Right _) _) = Nothing

457+destNid (Died (NodeIdentifier _) _) =

458+ Nothing

459+destNid (Died (ProcessIdentifier _pid) _) =

460+ fail "destNid: TODO"

461+destNid (Died (ChannelIdentifier _cid) _) =

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

462+ fail "destNid: TODO"

435−destNid (Spawn _ _) = Nothing

463+destNid (Spawn _ _) =

464+ Nothing

436

465

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

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

438isLocal :: LocalNode -> ProcessId -> Bool

467isLocal :: LocalNode -> ProcessId -> Bool