bfacdc4 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Finish the local backend

bfacdc4Edsko de Vries14 years ago

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

compare with master

unifiedsplit

@@ -13,6 +13,9 @@ module Control.Distributed.Process

13 , NodeId

14 , Process

15 , SendPortId

16+ , processNodeId

17+ , sendPortProcessId

18+ , liftIO -- Reexported for convenience

16 -- * Basic messaging

19 -- * Basic messaging

17 , send

20 , send

18 , expect

21 , expect

@@ -74,7 +77,9 @@ module Control.Distributed.Process

74 , registerRemote

77 , registerRemote

75 , unregisterRemote

78 , unregisterRemote

76 , whereisRemote

79 , whereisRemote

80+ , whereisRemoteAsync

77 , nsendRemote

81 , nsendRemote

82+ , WhereIsReply(..)

78 -- * Auxiliary API

83 -- * Auxiliary API

79 , catch

84 , catch

80 , expectTimeout

85 , expectTimeout

@@ -89,6 +94,7 @@ import Prelude hiding (catch)

89import Data.Typeable (Typeable, typeOf)

94import Data.Typeable (Typeable, typeOf)

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

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

91import Control.Exception (throw)

96import Control.Exception (throw)

97+import Control.Monad.IO.Class (liftIO)

92import Control.Distributed.Process.Internal.MessageT (getLocalNode)

98import Control.Distributed.Process.Internal.MessageT (getLocalNode)

93import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)

99import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)

94import Control.Distributed.Process.Internal.Types

100import Control.Distributed.Process.Internal.Types

@@ -114,7 +120,8 @@ import Control.Distributed.Process.Internal.Types

114 , SerializableDict(..)

120 , SerializableDict(..)

115 , procMsg

121 , procMsg

116 , LocalNode(..)

122 , LocalNode(..)

117− , SendPortId

123+ , SendPortId(..)

124+ , WhereIsReply(..)

118 )

125 )

119import Control.Distributed.Process.Internal.Closure.BuiltIn

126import Control.Distributed.Process.Internal.Closure.BuiltIn

120 ( linkClosure

127 ( linkClosure

@@ -167,6 +174,7 @@ import Control.Distributed.Process.Internal.Primitives

167 , registerRemote

174 , registerRemote

168 , unregisterRemote

175 , unregisterRemote

169 , whereisRemote

176 , whereisRemote

177+ , whereisRemoteAsync

170 , nsendRemote

178 , nsendRemote

171 -- Auxiliary API

179 -- Auxiliary API

172 , catch

180 , catch

distributed-process/src/Control/Distributed/Process/Backend/Local.hs+167−22

@@ -1,31 +1,87 @@

1-- | Simple backend based on the TCP transport which offers node discovery

2−-- based on UDP multicast

2+-- based on UDP multicast. This is a zero-configuration backend designed to

3+-- get you going with Cloud Haskell quickly without imposing any structure

4+-- on your application.

5+--

6+-- To simplify getting started we provide special support for /master/ and

7+-- /slave/ nodes (see 'startSlave' and 'startMaster'). Use of these functions

8+-- is completely optional; you can use the local backend without making use

9+-- of the predefined master and slave nodes.

10+--

11+-- [Minimal example]

12+--

13+-- > import System.Environment (getArgs)

14+-- > import Control.Distributed.Process

15+-- > import Control.Distributed.Process.Node (initRemoteTable)

16+-- > import Control.Distributed.Process.Backend.Local

17+-- >

18+-- > master :: LocalBackend -> [NodeId] -> Process ()

19+-- > master backend slaves = do

20+-- > -- Do something interesting with the slaves

21+-- > liftIO . putStrLn $ "Slaves: " ++ show slaves

22+-- > -- Terminate the slaves when the master terminates (this is optional)

23+-- > terminateAllSlaves backend

24+-- >

25+-- > main :: IO ()

26+-- > main = do

27+-- > args <- getArgs

28+-- >

29+-- > case args of

30+-- > ["master", host, port] -> do

31+-- > backend <- initializeBackend host port initRemoteTable

32+-- > startMaster backend (master backend)

33+-- > ["slave", host, port] -> do

34+-- > backend <- initializeBackend host port initRemoteTable

35+-- > startSlave backend

3{-# OPTIONS_GHC -fno-warn-orphans #-}

36{-# OPTIONS_GHC -fno-warn-orphans #-}

4module Control.Distributed.Process.Backend.Local

37module Control.Distributed.Process.Backend.Local

38+ ( -- * Initialization

5− ( LocalBackend(..)

39+ LocalBackend(..)

6 , initializeBackend

40 , initializeBackend

41+ -- * Slave nodes

42+ , startSlave

43+ , terminateSlave

44+ , findSlaves

45+ , terminateAllSlaves

46+ -- * Master nodes

47+ , startMaster

7 ) where

48 ) where

50+import System.IO (fixIO)

51+import Data.Maybe (catMaybes)

9import Data.Binary (Binary(get, put), getWord8, putWord8)

52import Data.Binary (Binary(get, put), getWord8, putWord8)

10import Data.Accessor (Accessor, accessor, (^:), (^.))

53import Data.Accessor (Accessor, accessor, (^:), (^.))

11import Data.Set (Set)

54import Data.Set (Set)

12import qualified Data.Set as Set (insert, empty, toList)

55import qualified Data.Set as Set (insert, empty, toList)

13import Data.Foldable (forM_)

56import Data.Foldable (forM_)

57+import Data.Typeable (Typeable)

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

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

15import Control.Exception (throw)

59import Control.Exception (throw)

16−import Control.Monad (forever)

60+import Control.Monad (forever, forM)

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

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

18−import Control.Concurrent (forkIO, threadDelay)

62+import Control.Concurrent (forkIO, threadDelay, ThreadId)

19import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)

63import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)

20−import Control.Distributed.Process.Internal.Types

64+import Control.Distributed.Process

21− ( LocalNode

22− , RemoteTable

65+ ( RemoteTable

23 , NodeId

66 , NodeId

24 , Process

67 , Process

68+ , WhereIsReply(..)

69+ , whereis

70+ , whereisRemoteAsync

71+ , registerRemote

72+ , getSelfPid

73+ , register

74+ , expect

75+ , nsendRemote

76+ , receiveWait

77+ , matchIf

78+ , processNodeId

25 )

79 )

26import qualified Control.Distributed.Process.Node as Node

80import qualified Control.Distributed.Process.Node as Node

81+ ( LocalNode

27− ( newLocalNode

82+ , newLocalNode

28 , localNodeId

83 , localNodeId

84+ , runProcess

29 )

85 )

30import qualified Network.Transport.TCP as NT

86import qualified Network.Transport.TCP as NT

31 ( createTransport

87 ( createTransport

@@ -34,12 +90,11 @@ import qualified Network.Transport.TCP as NT

34import qualified Network.Transport as NT (Transport)

90import qualified Network.Transport as NT (Transport)

35import qualified Network.Socket as N (HostName, ServiceName, SockAddr)

91import qualified Network.Socket as N (HostName, ServiceName, SockAddr)

36import Control.Distributed.Process.Internal.Multicast (initMulticast)

92import Control.Distributed.Process.Internal.Multicast (initMulticast)

37−import Control.Distributed.Process.Internal.Primitives (whereis, registerRemote)

39-- | Local backend

94-- | Local backend

40data LocalBackend = LocalBackend {

95data LocalBackend = LocalBackend {

41 -- | Create a new local node

96 -- | Create a new local node

42− newLocalNode :: IO LocalNode

97+ newLocalNode :: IO Node.LocalNode

43 -- | @findPeers t@ sends out a /who's there?/ request, waits 't' msec,

98 -- | @findPeers t@ sends out a /who's there?/ request, waits 't' msec,

44 -- and then collects and returns the answers

99 -- and then collects and returns the answers

45 , findPeers :: Int -> IO [NodeId]

100 , findPeers :: Int -> IO [NodeId]

@@ -49,8 +104,9 @@ data LocalBackend = LocalBackend {

49 }

104 }

105

51data LocalBackendState = LocalBackendState {

106data LocalBackendState = LocalBackendState {

52− _localNodes :: [LocalNode]

107+ _localNodes :: [Node.LocalNode]

53− , _peers :: Set NodeId

108+ , _peers :: Set NodeId

109+ , discoveryDaemon :: ThreadId

54 }

110 }

111

56-- | Initialize the backend

112-- | Initialize the backend

@@ -58,18 +114,21 @@ initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO LocalBacke

58initializeBackend host port rtable = do

114initializeBackend host port rtable = do

59 mTransport <- NT.createTransport host port NT.defaultTCPParameters

115 mTransport <- NT.createTransport host port NT.defaultTCPParameters

60 (recv, send) <- initMulticast "224.0.0.99" 9999 1024

116 (recv, send) <- initMulticast "224.0.0.99" 9999 1024

117+ (_, backendState) <- fixIO $ \ ~(tid, _) -> do

61− backendState <- newMVar LocalBackendState

118+ backendState <- newMVar LocalBackendState

62− { _localNodes = []

119+ { _localNodes = []

63− , _peers = Set.empty

120+ , _peers = Set.empty

121+ , discoveryDaemon = tid

64− }

122+ }

65− forkIO $ peerDiscoveryDaemon backendState recv send

123+ tid' <- forkIO $ peerDiscoveryDaemon backendState recv send

124+ return (tid', backendState)

66 case mTransport of

125 case mTransport of

67 Left err -> throw err

126 Left err -> throw err

68 Right transport ->

127 Right transport ->

69 let backend = LocalBackend {

128 let backend = LocalBackend {

70− newLocalNode = apiNewLocalNode transport rtable backendState

129+ newLocalNode = apiNewLocalNode transport rtable backendState

71− , findPeers = apiFindPeers send backendState

130+ , findPeers = apiFindPeers send backendState

72− , redirectLogsHere = apiRedirectLogsHere backend

131+ , redirectLogsHere = apiRedirectLogsHere backend

73 }

132 }

74 in return backend

133 in return backend

134

@@ -77,7 +136,7 @@ initializeBackend host port rtable = do

77apiNewLocalNode :: NT.Transport

136apiNewLocalNode :: NT.Transport

78 -> RemoteTable

137 -> RemoteTable

79 -> MVar LocalBackendState

138 -> MVar LocalBackendState

80− -> IO LocalNode

139+ -> IO Node.LocalNode

81apiNewLocalNode transport rtable backendState = do

140apiNewLocalNode transport rtable backendState = do

82 localNode <- Node.newLocalNode transport rtable

141 localNode <- Node.newLocalNode transport rtable

83 modifyMVar_ backendState $ return . (localNodes ^: (localNode :))

142 modifyMVar_ backendState $ return . (localNodes ^: (localNode :))

@@ -135,11 +194,97 @@ apiRedirectLogsHere backend = do

135 nids <- liftIO $ findPeers backend 1000000

194 nids <- liftIO $ findPeers backend 1000000

136 forM_ nids $ \nid -> registerRemote nid "logger" logger

195 forM_ nids $ \nid -> registerRemote nid "logger" logger

137

196

197+--------------------------------------------------------------------------------

198+-- Slaves --

199+--------------------------------------------------------------------------------

200+

201+-- | Messages to slave nodes

202+--

203+-- This datatype is not exposed; instead, we expose primitives for dealing

204+-- with slaves.

205+data SlaveControllerMsg =

206+ SlaveTerminate

207+ deriving (Typeable, Show)

208+

209+instance Binary SlaveControllerMsg where

210+ put SlaveTerminate = putWord8 0

211+ get = do

212+ header <- getWord8

213+ case header of

214+ 0 -> return SlaveTerminate

215+ _ -> fail "SlaveControllerMsg.get: invalid"

216+

217+-- | Calling 'slave' sets up a new local node and then waits. You start

218+-- processes on the slave by calling 'spawn' from other nodes.

219+--

220+-- This function does not return. The only way to exit the slave is to CTRL-C

221+-- the process or call terminateSlave from another node.

222+startSlave :: LocalBackend -> IO ()

223+startSlave backend = do

224+ node <- newLocalNode backend

225+ Node.runProcess node slaveController

226+

227+-- | The slave controller interprets 'SlaveControllerMsg's

228+slaveController :: Process ()

229+slaveController = do

230+ pid <- getSelfPid

231+ register "slaveController" pid

232+ go

233+ where

234+ go = do

235+ msg <- expect

236+ case msg of

237+ SlaveTerminate -> return ()

238+

239+-- | Terminate the slave at the given node ID

240+terminateSlave :: NodeId -> Process ()

241+terminateSlave nid = nsendRemote nid "slaveController" SlaveTerminate

242+

243+-- | Find slave nodes

244+findSlaves :: LocalBackend -> Process [NodeId]

245+findSlaves backend = do

246+ nodes <- liftIO $ findPeers backend 1000000

247+ -- Fire of asynchronous requests for the slave controller

248+ forM_ nodes $ \nid -> whereisRemoteAsync nid "slaveController"

249+ -- Wait for the replies

250+ catMaybes <$> forM nodes (\_ ->

251+ receiveWait

252+ [ matchIf (\(WhereIsReply label _) -> label == "slaveController")

253+ (\(WhereIsReply _ mPid) -> return (processNodeId <$> mPid))

254+ ])

255+

256+-- | Terminate all slaves

257+terminateAllSlaves :: LocalBackend -> Process ()

258+terminateAllSlaves backend = do

259+ slaves <- findSlaves backend

260+ forM_ slaves terminateSlave

261+ liftIO $ threadDelay 1000000

262+

263+--------------------------------------------------------------------------------

264+-- Master nodes

265+--------------------------------------------------------------------------------

266+

267+-- | 'startMaster' finds all slaves currently available on the local network

268+-- (which should therefore be started first), redirects all log messages to

269+-- itself, and then calls the specified process, passing the list of slaves

270+-- nodes.

271+--

272+-- Terminates when the specified process terminates. If you want to terminate

273+-- the slaves when the master terminates, you should manually call

274+-- 'terminateAllSlaves'.

275+startMaster :: LocalBackend -> ([NodeId] -> Process ()) -> IO ()

276+startMaster backend proc = do

277+ node <- newLocalNode backend

278+ Node.runProcess node $ do

279+ slaves <- findSlaves backend

280+ redirectLogsHere backend

281+ proc slaves

282+

138--------------------------------------------------------------------------------

283--------------------------------------------------------------------------------

139-- Accessors --

284-- Accessors --

140--------------------------------------------------------------------------------

285--------------------------------------------------------------------------------

141

286

142−localNodes :: Accessor LocalBackendState [LocalNode]

287+localNodes :: Accessor LocalBackendState [Node.LocalNode]

143localNodes = accessor _localNodes (\ns st -> st { _localNodes = ns })

288localNodes = accessor _localNodes (\ns st -> st { _localNodes = ns })

144

289

145peers :: Accessor LocalBackendState (Set NodeId)

290peers :: Accessor LocalBackendState (Set NodeId)

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+11−3

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

39 , registerRemote

40 , unregisterRemote

41 , whereisRemote

42+ , whereisRemoteAsync

42 , nsendRemote

43 , nsendRemote

43 -- * Auxiliary API

44 -- * Auxiliary API

44 , catch

45 , catch

@@ -54,7 +55,7 @@ module Control.Distributed.Process.Internal.Primitives

55import Prelude hiding (catch)

56import Prelude hiding (catch)

56import Data.Binary (decode)

57import Data.Binary (decode)

57−import Data.Typeable (Typeable,)

58+import Data.Typeable (Typeable)

58import Data.Time.Clock (getCurrentTime)

59import Data.Time.Clock (getCurrentTime)

59import Data.Time.Format (formatTime)

60import Data.Time.Format (formatTime)

60import System.Locale (defaultTimeLocale)

61import System.Locale (defaultTimeLocale)

@@ -422,11 +423,18 @@ whereis label = do

422-- | Query a remote process registry (synchronous)

423-- | Query a remote process registry (synchronous)

423whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

424whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

424whereisRemote nid label = do

425whereisRemote nid label = do

425− sendCtrlMsg (Just nid) (WhereIs label)

426+ whereisRemoteAsync nid label

426 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')

427 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')

427 (\(WhereIsReply _ mPid) -> return mPid)

428 (\(WhereIsReply _ mPid) -> return mPid)

428 ]

429 ]

429

430

431+-- | Query a remote process registry (asynchronous)

432+--

433+-- Reply will come in the form of a 'WhereIsReply' message

434+whereisRemoteAsync :: NodeId -> String -> Process ()

435+whereisRemoteAsync nid label =

436+ sendCtrlMsg (Just nid) (WhereIs label)

437+

430-- | Named send to a process in the local registry (asynchronous)

438-- | Named send to a process in the local registry (asynchronous)

431nsend :: Serializable a => String -> a -> Process ()

439nsend :: Serializable a => String -> a -> Process ()

432nsend label msg =

440nsend label msg =

@@ -434,7 +442,7 @@ nsend label msg =

434

442

435-- | Named send to a process in a remote registry (asynchronous)

443-- | Named send to a process in a remote registry (asynchronous)

436nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

444nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

437−nsendRemote nid label msg =

445+nsendRemote nid label msg =

438 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

446 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

439

447

440--------------------------------------------------------------------------------

448--------------------------------------------------------------------------------

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

@@ -106,6 +106,7 @@ import Control.Distributed.Process.Serializable

106 , encodeFingerprint

107 , sizeOfFingerprint

108 , decodeFingerprint

109+ , showFingerprint

109 )

110 )

110import Control.Distributed.Process.Internal.CQueue (CQueue)

111import Control.Distributed.Process.Internal.CQueue (CQueue)

111import Control.Distributed.Process.Internal.Dynamic (Dynamic)

112import Control.Distributed.Process.Internal.Dynamic (Dynamic)

@@ -313,7 +314,7 @@ data Message = Message

313 }

314 }

314

315

315instance Show Message where

316instance Show Message where

316− show = show . messageEncoding

317+ show (Message fp enc) = show enc ++ " :: " ++ showFingerprint fp []

317

318

318-- | Turn any serialiable term into a message

319-- | Turn any serialiable term into a message

319createMessage :: Serializable a => a -> Message

320createMessage :: Serializable a => a -> Message

distributed-process/src/Control/Distributed/Process/Node.hs+4−2

@@ -1,6 +1,7 @@

1-- | Local nodes

2module Control.Distributed.Process.Node

3+ ( LocalNode

3− ( newLocalNode

4+ , newLocalNode

4 , closeLocalNode

5 , closeLocalNode

5 , forkProcess

6 , forkProcess

6 , runProcess

7 , runProcess

@@ -262,7 +263,8 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

262 atomically $ writeTChan chan . decode . BSL.fromChunks $ payload

263 atomically $ writeTChan chan . decode . BSL.fromChunks $ payload

263 go uninitConns procs chans ctrls

264 go uninitConns procs chans ctrls

264 (_, _, True, _) -> do

265 (_, _, True, _) -> do

265− writeChan ctrlChan (decode . BSL.fromChunks $ payload)

266+ let ctrlMsg = decode . BSL.fromChunks $ payload

267+ writeChan ctrlChan ctrlMsg

266 go uninitConns procs chans ctrls

268 go uninitConns procs chans ctrls

267 (_, _, _, True) ->

269 (_, _, _, True) ->

268 case decode (BSL.fromChunks payload) of

270 case decode (BSL.fromChunks payload) of

distributed-process/src/Control/Distributed/Process/Serializable.hs+10−3

@@ -5,13 +5,15 @@ module Control.Distributed.Process.Serializable

5 , fingerprint

6 , sizeOfFingerprint

7 , Fingerprint

8+ , showFingerprint

8 ) where

9 ) where

10import Data.Binary (Binary)

11import Data.Binary (Binary)

11import Data.Typeable (Typeable(..))

12import Data.Typeable (Typeable(..))

12import Data.Typeable.Internal (TypeRep(TypeRep))

13import Data.Typeable.Internal (TypeRep(TypeRep))

14+import Numeric (showHex)

13import Control.Exception (throw)

15import Control.Exception (throw)

14−import GHC.Fingerprint.Type (Fingerprint)

16+import GHC.Fingerprint.Type (Fingerprint(..))

15import Data.ByteString (ByteString)

17import Data.ByteString (ByteString)

16import qualified Data.ByteString as BS

18import qualified Data.ByteString as BS

17import qualified Data.ByteString.Internal as BSI ( unsafeCreate

19import qualified Data.ByteString.Internal as BSI ( unsafeCreate

@@ -38,8 +40,8 @@ decodeFingerprint bs

38 | BS.length bs /= sizeOfFingerprint =

40 | BS.length bs /= sizeOfFingerprint =

39 throw $ userError "decodeFingerprint: Invalid length"

41 throw $ userError "decodeFingerprint: Invalid length"

40 | otherwise = BSI.inlinePerformIO $ do

42 | otherwise = BSI.inlinePerformIO $ do

41− let (fp, _, _) = BSI.toForeignPtr bs

43+ let (fp, offset, _) = BSI.toForeignPtr bs

42− withForeignPtr fp $ \p -> peekByteOff p 0

44+ withForeignPtr fp $ \p -> peekByteOff p offset

44-- | Size of a fingerprint

46-- | Size of a fingerprint

45sizeOfFingerprint :: Int

47sizeOfFingerprint :: Int

@@ -48,3 +50,8 @@ sizeOfFingerprint = sizeOf (undefined :: Fingerprint)

48-- | The fingerprint of the typeRep of the argument

50-- | The fingerprint of the typeRep of the argument

49fingerprint :: Typeable a => a -> Fingerprint

51fingerprint :: Typeable a => a -> Fingerprint

50fingerprint a = let TypeRep fp _ _ = typeOf a in fp

52fingerprint a = let TypeRep fp _ _ = typeOf a in fp

53+

54+-- | Show fingerprint (for debugging purposes)

55+showFingerprint :: Fingerprint -> ShowS

56+showFingerprint (Fingerprint hi lo) =

57+ showString "(" . showHex hi . showString "," . showHex lo . showString ")"

distributed-process/tests/TestCH.hs+18−1

@@ -13,7 +13,6 @@ import Control.Concurrent.MVar

13 , readMVar

14 )

15import Control.Monad (replicateM_, replicateM)

16−import Control.Monad.IO.Class (liftIO)

17import Control.Exception (throwIO)

16import Control.Exception (throwIO)

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

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

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

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

@@ -540,6 +539,23 @@ testRegistry transport = do

540 True <- return $ pingServer == pid'

539 True <- return $ pingServer == pid'

541 return ()

540 return ()

542

541

542+testRemoteRegistry :: NT.Transport -> IO ()

543+testRemoteRegistry transport = do

544+ node1 <- newLocalNode transport initRemoteTable

545+ node2 <- newLocalNode transport initRemoteTable

546+

547+ pingServer <- forkProcess node1 ping

548+

549+ runProcess node2 $ do

550+ let nid1 = localNodeId node1

551+ registerRemote nid1 "ping" pingServer

552+ Just pid <- whereisRemote nid1 "ping"

553+ True <- return $ pingServer == pid

554+ us <- getSelfPid

555+ nsendRemote nid1 "ping" (Pong us)

556+ Ping pid' <- expect

557+ True <- return $ pingServer == pid'

558+ return ()

543

559

544main :: IO ()

560main :: IO ()

545main = do

561main = do

@@ -554,6 +570,7 @@ main = do

554 , ("MergeChannels", testMergeChannels transport)

570 , ("MergeChannels", testMergeChannels transport)

555 , ("Terminate", testTerminate transport)

571 , ("Terminate", testTerminate transport)

556 , ("Registry", testRegistry transport)

572 , ("Registry", testRegistry transport)

573+ , ("RemoteRegistry", testRemoteRegistry transport)

557 -- Monitoring processes

574 -- Monitoring processes

558 --

575 --

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

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

distributed-process/tests/TestClosure.hs−1

@@ -2,7 +2,6 @@ module Main where

3import qualified Data.ByteString.Lazy as BSL (empty)

4import Control.Monad (join, replicateM)

5−import Control.Monad.IO.Class (liftIO)

6import Control.Exception (IOException, throw)

5import Control.Exception (IOException, throw)

7import Control.Concurrent (forkIO, threadDelay)

6import Control.Concurrent (forkIO, threadDelay)

8import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)

7import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)

distributed-process/tests/TestLocalBackend.hs+11−29

@@ -1,30 +1,12 @@

1import System.Environment (getArgs, getProgName)

2−import Control.Distributed.Process (say)

2+import Control.Distributed.Process (NodeId, Process, liftIO)

3−import Control.Distributed.Process.Node

3+import Control.Distributed.Process.Node (initRemoteTable)

4− ( initRemoteTable

5− , forkProcess

6− , runProcess

7− )

8import Control.Distributed.Process.Backend.Local

4import Control.Distributed.Process.Backend.Local

9−import Control.Concurrent (threadDelay)

10−import Control.Monad (forever)

11−import Control.Monad.IO.Class (liftIO)

13−server :: LocalBackend -> IO ()

6+master :: LocalBackend -> [NodeId] -> Process ()

14−server backend = do

15− node <- newLocalNode backend

16− runProcess node $ do

17− redirectLogsHere backend

18− say "this is the server"

19− threadDelay 30000000

20−

21−client :: LocalBackend -> IO ()

22−client backend = do

7+master backend slaves = do

23− node <- newLocalNode backend

8+ liftIO . putStrLn $ "Slaves: " ++ show slaves

24− forkProcess node . forever $ do

9+ terminateAllSlaves backend

25− say "this is a client"

26− liftIO $ threadDelay 2000000

27− threadDelay 30000000

29main :: IO ()

11main :: IO ()

30main = do

12main = do

@@ -32,11 +14,11 @@ main = do

32 args <- getArgs

14 args <- getArgs

34 case args of

16 case args of

35− ["server", host, port] -> do

17+ ["master", host, port] -> do

36 backend <- initializeBackend host port initRemoteTable

18 backend <- initializeBackend host port initRemoteTable

37− server backend

19+ startMaster backend (master backend)

38− ["client", host, port] -> do

20+ ["slave", host, port] -> do

39 backend <- initializeBackend host port initRemoteTable

21 backend <- initializeBackend host port initRemoteTable

40− client backend

22+ startSlave backend

41 _ ->

23 _ ->

42− putStrLn $ "usage: " ++ prog ++ " (server | client) host port"

24+ putStrLn $ "usage: " ++ prog ++ " (master | slave) host port"

distributed-process/tests/runTestLocalBackend.hs+9−6

@@ -1,8 +1,11 @@

1#!/bin/bash

2TestLocalBackend=dist/build/TestLocalBackend/TestLocalBackend

3−$TestLocalBackend client 127.0.0.1 8080 &

3+$TestLocalBackend slave 127.0.0.1 8080 &

4+sleep 1

4−$TestLocalBackend client 127.0.0.1 8081 &

5+$TestLocalBackend slave 127.0.0.1 8081 &

6+sleep 1

5−$TestLocalBackend client 127.0.0.1 8082 &

7+$TestLocalBackend slave 127.0.0.1 8082 &

8+sleep 1

6−$TestLocalBackend client 127.0.0.1 8083 &

9+$TestLocalBackend slave 127.0.0.1 8083 &

7−sleep 2 # give clients a change to start

10+sleep 1

8−$TestLocalBackend server 127.0.0.1 8084

11+$TestLocalBackend master 127.0.0.1 8084