@@ -1,31 +1,87 @@
11-- | 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
336{-# OPTIONS_GHC -fno-warn-orphans #-}
437module Control.Distributed.Process.Backend.Local
5− ( LocalBackend(..)
38+ ( -- * Initialization
39+ LocalBackend(..)
640 , initializeBackend
41+ -- * Slave nodes
42+ , startSlave
43+ , terminateSlave
44+ , findSlaves
45+ , terminateAllSlaves
46+ -- * Master nodes
47+ , startMaster
748 ) where
849
50+import System.IO (fixIO)
51+import Data.Maybe (catMaybes)
952import Data.Binary (Binary(get, put), getWord8, putWord8)
1053import Data.Accessor (Accessor, accessor, (^:), (^.))
1154import Data.Set (Set)
1255import qualified Data.Set as Set (insert, empty, toList)
1356import Data.Foldable (forM_)
57+import Data.Typeable (Typeable)
1458import Control.Applicative ((<$>))
1559import Control.Exception (throw)
16−import Control.Monad (forever)
60+import Control.Monad (forever, forM)
1761import Control.Monad.IO.Class (liftIO)
18−import Control.Concurrent (forkIO, threadDelay)
62+import Control.Concurrent (forkIO, threadDelay, ThreadId)
1963import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)
20−import Control.Distributed.Process.Internal.Types
21− ( LocalNode
22− , RemoteTable
64+import Control.Distributed.Process
65+ ( RemoteTable
2366 , NodeId
2467 , Process
68+ , WhereIsReply(..)
69+ , whereis
70+ , whereisRemoteAsync
71+ , registerRemote
72+ , getSelfPid
73+ , register
74+ , expect
75+ , nsendRemote
76+ , receiveWait
77+ , matchIf
78+ , processNodeId
2579 )
2680import qualified Control.Distributed.Process.Node as Node
27− ( newLocalNode
81+ ( LocalNode
82+ , newLocalNode
2883 , localNodeId
84+ , runProcess
2985 )
3086import qualified Network.Transport.TCP as NT
3187 ( createTransport
@@ -34,12 +90,11 @@ import qualified Network.Transport.TCP as NT
3490import qualified Network.Transport as NT (Transport)
3591import qualified Network.Socket as N (HostName, ServiceName, SockAddr)
3692import Control.Distributed.Process.Internal.Multicast (initMulticast)
37−import Control.Distributed.Process.Internal.Primitives (whereis, registerRemote)
3893
3994-- | Local backend
4095data LocalBackend = LocalBackend {
4196 -- | Create a new local node
42− newLocalNode :: IO LocalNode
97+ newLocalNode :: IO Node.LocalNode
4398 -- | @findPeers t@ sends out a /who's there?/ request, waits 't' msec,
4499 -- and then collects and returns the answers
45100 , findPeers :: Int -> IO [NodeId]
@@ -49,8 +104,9 @@ data LocalBackend = LocalBackend {
49104 }
50105
51106data LocalBackendState = LocalBackendState {
52− _localNodes :: [LocalNode]
53− , _peers :: Set NodeId
107+ _localNodes :: [Node.LocalNode]
108+ , _peers :: Set NodeId
109+ , discoveryDaemon :: ThreadId
54110 }
55111
56112-- | Initialize the backend
@@ -58,18 +114,21 @@ initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO LocalBacke
58114initializeBackend host port rtable = do
59115 mTransport <- NT.createTransport host port NT.defaultTCPParameters
60116 (recv, send) <- initMulticast "224.0.0.99" 9999 1024
61− backendState <- newMVar LocalBackendState
62− { _localNodes = []
63− , _peers = Set.empty
64− }
65− forkIO $ peerDiscoveryDaemon backendState recv send
117+ (_, backendState) <- fixIO $ \ ~(tid, _) -> do
118+ backendState <- newMVar LocalBackendState
119+ { _localNodes = []
120+ , _peers = Set.empty
121+ , discoveryDaemon = tid
122+ }
123+ tid' <- forkIO $ peerDiscoveryDaemon backendState recv send
124+ return (tid', backendState)
66125 case mTransport of
67126 Left err -> throw err
68127 Right transport ->
69128 let backend = LocalBackend {
70− newLocalNode = apiNewLocalNode transport rtable backendState
71− , findPeers = apiFindPeers send backendState
72− , redirectLogsHere = apiRedirectLogsHere backend
129+ newLocalNode = apiNewLocalNode transport rtable backendState
130+ , findPeers = apiFindPeers send backendState
131+ , redirectLogsHere = apiRedirectLogsHere backend
73132 }
74133 in return backend
75134
@@ -77,7 +136,7 @@ initializeBackend host port rtable = do
77136apiNewLocalNode :: NT.Transport
78137 -> RemoteTable
79138 -> MVar LocalBackendState
80− -> IO LocalNode
139+ -> IO Node.LocalNode
81140apiNewLocalNode transport rtable backendState = do
82141 localNode <- Node.newLocalNode transport rtable
83142 modifyMVar_ backendState $ return . (localNodes ^: (localNode :))
@@ -135,11 +194,97 @@ apiRedirectLogsHere backend = do
135194 nids <- liftIO $ findPeers backend 1000000
136195 forM_ nids $ \nid -> registerRemote nid "logger" logger
137196
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+
138283--------------------------------------------------------------------------------
139284-- Accessors --
140285--------------------------------------------------------------------------------
141286
142−localNodes :: Accessor LocalBackendState [LocalNode]
287+localNodes :: Accessor LocalBackendState [Node.LocalNode]
143288localNodes = accessor _localNodes (\ns st -> st { _localNodes = ns })
144289
145290peers :: Accessor LocalBackendState (Set NodeId)