e57e0e3 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Support code for spawn (but no implementation yet)

e57e0e3Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+32−14

compare with master

unifiedsplit

@@ -15,7 +15,7 @@ module Control.Distributed.Process

15 , expect

16 , send

17 , getSelfPid

18− , RemoteTable

18+ , spawn

19 -- * Matching messages

20 , Match

21 , match

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

38 , Closure

39 , Static

40 , unClosure

41+ , RemoteTable

41 -- * Auxiliary API

42 -- * Auxiliary API

42 , catch

43 , catch

43 , expectTimeout

44 , expectTimeout

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

78 , DidUnlink(..)

79 , DidUnlink(..)

79 , DiedReason(..)

80 , DiedReason(..)

80 , SpawnRef(..)

81 , SpawnRef(..)

82+ , DidSpawn(..)

81 , NCMsg(..)

83 , NCMsg(..)

82 , ProcessSignal(..)

84 , ProcessSignal(..)

83 , monitorCounter

85 , monitorCounter

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

87 )

89 )

88import Control.Distributed.Process.Internal.MessageT

90import Control.Distributed.Process.Internal.MessageT

89 ( sendMessage

91 ( sendMessage

92+ , sendBinary

90 , getLocalNode

93 , getLocalNode

91 )

94 )

@@ -171,21 +174,29 @@ getSelfPid = processId <$> ask

171monitor :: ProcessId -> Process MonitorRef

174monitor :: ProcessId -> Process MonitorRef

172monitor them = do

175monitor them = do

173 monitorRef <- getMonitorRefFor them

176 monitorRef <- getMonitorRefFor them

174− postCtrlMsg $ Monitor them monitorRef

177+ sendCtrlMsg Nothing $ Monitor them monitorRef

175 return monitorRef

178 return monitorRef

176

179

177-- | Link to a remote process

180-- | Link to a remote process

178link :: ProcessId -> Process ()

181link :: ProcessId -> Process ()

179−link = postCtrlMsg . Link

182+link = sendCtrlMsg Nothing . Link

180

183

181-- | Remove a monitor

184-- | Remove a monitor

182unmonitor :: MonitorRef -> Process ()

185unmonitor :: MonitorRef -> Process ()

183−unmonitor = postCtrlMsg . Unmonitor

186+unmonitor = sendCtrlMsg Nothing . Unmonitor

184

187

185-- | Remove a link

188-- | Remove a link

186unlink :: ProcessId -> Process ()

189unlink :: ProcessId -> Process ()

187−unlink = postCtrlMsg . Unlink

190+unlink = sendCtrlMsg Nothing . Unlink

188

191

192+-- | Spawn a process

193+spawn :: NodeId -> Closure (Process ()) -> Process ProcessId

194+spawn nid proc = do

195+ ref <- spawnAsync nid proc

196+ receiveWait [ matchIf (\(DidSpawn ref' _) -> ref == ref')

197+ (\(DidSpawn _ pid) -> return pid)

198+ ]

199+

189--------------------------------------------------------------------------------

200--------------------------------------------------------------------------------

190-- Matching --

201-- Matching --

191--------------------------------------------------------------------------------

202--------------------------------------------------------------------------------

@@ -257,9 +268,9 @@ expectTimeout timeout = receiveTimeout timeout [match return]

257--

268--

258-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')

269-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')

259spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef

270spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef

260−spawnAsync _nid proc = do

271+spawnAsync nid proc = do

261 spawnRef <- getSpawnRef

272 spawnRef <- getSpawnRef

262− postCtrlMsg (Spawn proc spawnRef)

273+ sendCtrlMsg (Just nid) $ Spawn proc spawnRef

263 return spawnRef

274 return spawnRef

264

275

265--------------------------------------------------------------------------------

276--------------------------------------------------------------------------------

@@ -284,14 +295,21 @@ getSpawnRef = do

284 , SpawnRef counter

295 , SpawnRef counter

285 )

296 )

286

297

287−-- | Post a control message on the local node controller

298+-- Send a control message

299+sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node

288−postCtrlMsg :: ProcessSignal -> Process ()

300+ -> ProcessSignal -- ^ Message to send

289−postCtrlMsg signal = do

301+ -> Process ()

302+sendCtrlMsg mNid signal = do

290 us <- getSelfPid

303 us <- getSelfPid

291− ctrlChan <- localCtrlChan <$> lift getLocalNode

292− liftIO $ writeChan ctrlChan NCMsg { ctrlMsgSender = Left us

304+ let msg = NCMsg { ctrlMsgSender = Left us

293− , ctrlMsgSignal = signal

305+ , ctrlMsgSignal = signal

294− }

306+ }

307+ case mNid of

308+ Nothing -> do

309+ ctrlChan <- localCtrlChan <$> lift getLocalNode

310+ liftIO $ writeChan ctrlChan msg

311+ Just nid ->

312+ lift $ sendBinary (Right nid) msg

295

313

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

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

297lookupStatic label = do

315lookupStatic label = do

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

@@ -105,7 +105,7 @@ nodeController = do

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

106 ncEffectProcessDied pid reason

107 NCMsg _from (Spawn _ _) ->

108− fail "spawn not implemented"

108+ fail $ show (localNodeId node) ++ ": spawn not implemented"

109 unexpected ->

110 error $ "Unexpected message " ++ show unexpected

111

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

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

29 , DidUnmonitor(..)

30 , DidUnlink(..)

31 , SpawnRef(..)

32+ , DidSpawn(..)

32 -- * Node controller internal data types

33 -- * Node controller internal data types

33 , NCMsg(..)

34 , NCMsg(..)

34 , ProcessSignal(..)

35 , ProcessSignal(..)

@@ -244,7 +245,11 @@ newtype DidUnlink = DidUnlink ProcessId

244

245

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

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

246newtype SpawnRef = SpawnRef Int32

247newtype SpawnRef = SpawnRef Int32

247− deriving (Show, Binary, Typeable)

248+ deriving (Show, Binary, Typeable, Eq)

249+

250+-- | (Asynchronius) reply from spawn

251+data DidSpawn = DidSpawn SpawnRef ProcessId

252+ deriving (Show, Typeable)

248

253

249--------------------------------------------------------------------------------

254--------------------------------------------------------------------------------

250-- Node controller internal data types --

255-- Node controller internal data types --

@@ -371,6 +376,10 @@ instance Binary (Closure a) where

371 put (Closure (Static label) env) = put label >> put env

376 put (Closure (Static label) env) = put label >> put env

372 get = Closure <$> (Static <$> get) <*> get

377 get = Closure <$> (Static <$> get) <*> get

373

378

379+instance Binary DidSpawn where

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

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

382+

374--------------------------------------------------------------------------------

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

375-- Accessors --

384-- Accessors --

376--------------------------------------------------------------------------------

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

distributed-process/tests/TestClosure.hs+42−14

@@ -19,15 +19,20 @@ putInt = flip putMVar

19sendInt :: ProcessId -> Int -> Process ()

20sendInt = send

22+sendPid :: ProcessId -> Process ()

23+sendPid toPid = do

24+ fromPid <- getSelfPid

25+ send toPid fromPid

26+

22−$(remotable ['addInt, 'putInt, 'sendInt])

27+$(remotable ['addInt, 'putInt, 'sendInt, 'sendPid])

24testSendPureClosure :: Transport -> RemoteTable -> IO ()

29testSendPureClosure :: Transport -> RemoteTable -> IO ()

25−testSendPureClosure transport metaData = do

30+testSendPureClosure transport rtable = do

26 serverAddr <- newEmptyMVar

31 serverAddr <- newEmptyMVar

27 serverDone <- newEmptyMVar

32 serverDone <- newEmptyMVar

29 forkIO $ do

34 forkIO $ do

30− node <- newLocalNode transport metaData

35+ node <- newLocalNode transport rtable

31 addr <- forkProcess node $ do

36 addr <- forkProcess node $ do

32 cl <- expect

37 cl <- expect

33 fn <- unClosure cl :: Process (Int -> Int)

38 fn <- unClosure cl :: Process (Int -> Int)

@@ -36,19 +41,19 @@ testSendPureClosure transport metaData = do

36 putMVar serverAddr addr

41 putMVar serverAddr addr

38 forkIO $ do

43 forkIO $ do

39− node <- newLocalNode transport metaData

44+ node <- newLocalNode transport rtable

40 theirAddr <- readMVar serverAddr

45 theirAddr <- readMVar serverAddr

41 runProcess node $ send theirAddr ($(mkClosure 'addInt) 5)

46 runProcess node $ send theirAddr ($(mkClosure 'addInt) 5)

43 takeMVar serverDone

48 takeMVar serverDone

45testSendIOClosure :: Transport -> RemoteTable -> IO ()

50testSendIOClosure :: Transport -> RemoteTable -> IO ()

46−testSendIOClosure transport metaData = do

51+testSendIOClosure transport rtable = do

47 serverAddr <- newEmptyMVar

52 serverAddr <- newEmptyMVar

48 serverDone <- newEmptyMVar

53 serverDone <- newEmptyMVar

50 forkIO $ do

55 forkIO $ do

51− node <- newLocalNode transport metaData

56+ node <- newLocalNode transport rtable

52 addr <- forkProcess node $ do

57 addr <- forkProcess node $ do

53 cl <- expect

58 cl <- expect

54 io <- unClosure cl :: Process (MVar Int -> IO ())

59 io <- unClosure cl :: Process (MVar Int -> IO ())

@@ -60,19 +65,19 @@ testSendIOClosure transport metaData = do

60 putMVar serverAddr addr

65 putMVar serverAddr addr

62 forkIO $ do

67 forkIO $ do

63− node <- newLocalNode transport metaData

68+ node <- newLocalNode transport rtable

64 theirAddr <- readMVar serverAddr

69 theirAddr <- readMVar serverAddr

65 runProcess node $ send theirAddr ($(mkClosure 'putInt) 5)

70 runProcess node $ send theirAddr ($(mkClosure 'putInt) 5)

67 takeMVar serverDone

72 takeMVar serverDone

69testSendProcClosure :: Transport -> RemoteTable -> IO ()

74testSendProcClosure :: Transport -> RemoteTable -> IO ()

70−testSendProcClosure transport metaData = do

75+testSendProcClosure transport rtable = do

71 serverAddr <- newEmptyMVar

76 serverAddr <- newEmptyMVar

72 clientDone <- newEmptyMVar

77 clientDone <- newEmptyMVar

74 forkIO $ do

79 forkIO $ do

75− node <- newLocalNode transport metaData

80+ node <- newLocalNode transport rtable

76 addr <- forkProcess node $ do

81 addr <- forkProcess node $ do

77 cl <- expect

82 cl <- expect

78 pr <- unClosure cl :: Process (Int -> Process ())

83 pr <- unClosure cl :: Process (Int -> Process ())

@@ -80,7 +85,7 @@ testSendProcClosure transport metaData = do

80 putMVar serverAddr addr

85 putMVar serverAddr addr

82 forkIO $ do

87 forkIO $ do

83− node <- newLocalNode transport metaData

88+ node <- newLocalNode transport rtable

84 theirAddr <- readMVar serverAddr

89 theirAddr <- readMVar serverAddr

85 runProcess node $ do

90 runProcess node $ do

86 pid <- getSelfPid

91 pid <- getSelfPid

@@ -90,12 +95,35 @@ testSendProcClosure transport metaData = do

91 takeMVar clientDone

96 takeMVar clientDone

98+testSpawn :: Transport -> RemoteTable -> IO ()

99+testSpawn transport rtable = do

100+ serverNodeAddr <- newEmptyMVar

101+ clientDone <- newEmptyMVar

102+

103+ forkIO $ do

104+ node <- newLocalNode transport rtable

105+ print (localNodeId node)

106+ putMVar serverNodeAddr (localNodeId node)

107+

108+ forkIO $ do

109+ node <- newLocalNode transport rtable

110+ nid <- readMVar serverNodeAddr

111+ runProcess node $ do

112+ pid <- getSelfPid

113+ pid' <- spawn nid ($(mkClosure 'sendPid) pid)

114+ pid'' <- expect

115+ True <- return $ pid' == pid''

116+ liftIO $ putMVar clientDone ()

117+

118+ takeMVar clientDone

119+

93main :: IO ()

120main :: IO ()

94main = do

121main = do

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

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

96− let metaData = __remoteTable initRemoteTable

123+ let rtable = __remoteTable initRemoteTable

97 runTests

124 runTests

98− [ ("SendPureClosure", testSendPureClosure transport metaData)

125+ [ ("SendPureClosure", testSendPureClosure transport rtable)

99− , ("SendIOClosure", testSendIOClosure transport metaData)

126+ , ("SendIOClosure", testSendIOClosure transport rtable)

100− , ("SendProcClosure", testSendProcClosure transport metaData)

127+ , ("SendProcClosure", testSendProcClosure transport rtable)

128+ , ("Spawn", testSpawn transport rtable)

101 ]

129 ]