git.iko.soy/distributed-process
on fd61d21 help sign in
distributed-process/distributed-process-simplelocalnet/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs
355 lines 13 kB history raw
1-- | Simple backend based on the TCP transport which offers node discovery
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.SimpleLocalnet
17-- >
18-- > master :: Backend -> [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
36--
37-- [Compiling and Running]
38--
39-- Save to @example.hs@ and compile using
40--
41-- > ghc -threaded example.hs
42--
43-- Fire up some slave nodes (for the example, we run them on a single machine):
44--
45-- > ./example slave localhost 8080 &
46-- > ./example slave localhost 8081 &
47-- > ./example slave localhost 8082 &
48-- > ./example slave localhost 8083 &
49--
50-- And start the master node:
51--
52-- > ./example master localhost 8084
53--
54-- which should then output:
55--
56-- > Slaves: [nid://localhost:8083:0,nid://localhost:8082:0,nid://localhost:8081:0,nid://localhost:8080:0]
57--
58-- at which point the slaves should exit.
59--
60-- To run the example on multiple machines, you could run
61--
62-- > ./example slave 198.51.100.1 8080 &
63-- > ./example slave 198.51.100.2 8080 &
64-- > ./example slave 198.51.100.3 8080 &
65-- > ./example slave 198.51.100.4 8080 &
66--
67-- on four different machines (with IP addresses 198.51.100.1..4), and run the
68-- master on a fifth node (or on any of the four machines that run the slave
69-- nodes).
70--
71-- It is important that every node has a unique (hostname, port number) pair,
72-- and that the hostname you use to initialize the node can be resolved by
73-- peer nodes. In other words, if you start a node and pass hostname @localhost@
74-- then peer nodes won't be able to reach it because @localhost@ will resolve
75-- to a different IP address for them.
76{-# OPTIONS_GHC -fno-warn-orphans #-}
77module Control.Distributed.Process.Backend.SimpleLocalnet
78 ( -- * Initialization
79 Backend(..)
80 , initializeBackend
81 -- * Slave nodes
82 , startSlave
83 , terminateSlave
84 , findSlaves
85 , terminateAllSlaves
86 -- * Master nodes
87 , startMaster
88 ) where
90import System.IO (fixIO)
91import Data.Maybe (catMaybes)
92import Data.Binary (Binary(get, put), getWord8, putWord8)
93import Data.Accessor (Accessor, accessor, (^:), (^.))
94import Data.Set (Set)
95import qualified Data.Set as Set (insert, empty, toList)
96import Data.Foldable (forM_)
97import Data.Typeable (Typeable)
98import Control.Applicative ((<$>))
99import Control.Exception (throw)
100import Control.Monad (forever, forM, replicateM)
101import Control.Monad.IO.Class (liftIO)
102import Control.Concurrent (forkIO, threadDelay, ThreadId)
103import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)
104import Control.Distributed.Process
105 ( RemoteTable
106 , NodeId
107 , Process
108 , WhereIsReply(..)
109 , whereis
110 , whereisRemoteAsync
111 , registerRemote
112 , getSelfPid
113 , register
114 , expect
115 , nsendRemote
116 , receiveWait
117 , match
118 , matchIf
119 , processNodeId
120 , monitorNode
121 , unmonitor
122 , NodeMonitorNotification(..)
123 )
124import qualified Control.Distributed.Process.Node as Node
125 ( LocalNode
126 , newLocalNode
127 , localNodeId
128 , runProcess
129 )
130import qualified Network.Transport.TCP as NT
131 ( createTransport
132 , defaultTCPParameters
133 )
134import qualified Network.Transport as NT (Transport)
135import qualified Network.Socket as N (HostName, ServiceName, SockAddr)
136import Control.Distributed.Process.Backend.SimpleLocalnet.Internal.Multicast (initMulticast)
138-- | Local backend
139data Backend = Backend {
140 -- | Create a new local node
141 newLocalNode :: IO Node.LocalNode
142 -- | @findPeers t@ broadcasts a /who's there?/ message on the local
143 -- network, waits 't' msec, and then collects and returns the answers.
144 -- You can use this to dynamically discover peer nodes.
145 , findPeers :: Int -> IO [NodeId]
146 -- | Make sure that all log messages are printed by the logger on the
147 -- current node
148 , redirectLogsHere :: Process ()
149 }
151data BackendState = BackendState {
152 _localNodes :: [Node.LocalNode]
153 , _peers :: Set NodeId
154 , discoveryDaemon :: ThreadId
155 }
157-- | Initialize the backend
158initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO Backend
159initializeBackend host port rtable = do
160 mTransport <- NT.createTransport host port NT.defaultTCPParameters
161 (recv, send) <- initMulticast "224.0.0.99" 9999 1024
162 (_, backendState) <- fixIO $ \ ~(tid, _) -> do
163 backendState <- newMVar BackendState
164 { _localNodes = []
165 , _peers = Set.empty
166 , discoveryDaemon = tid
167 }
168 tid' <- forkIO $ peerDiscoveryDaemon backendState recv send
169 return (tid', backendState)
170 case mTransport of
171 Left err -> throw err
172 Right transport ->
173 let backend = Backend {
174 newLocalNode = apiNewLocalNode transport rtable backendState
175 , findPeers = apiFindPeers send backendState
176 , redirectLogsHere = apiRedirectLogsHere backend
177 }
178 in return backend
180-- | Create a new local node
181apiNewLocalNode :: NT.Transport
182 -> RemoteTable
183 -> MVar BackendState
184 -> IO Node.LocalNode
185apiNewLocalNode transport rtable backendState = do
186 localNode <- Node.newLocalNode transport rtable
187 modifyMVar_ backendState $ return . (localNodes ^: (localNode :))
188 return localNode
190-- | Peer discovery
191apiFindPeers :: (PeerDiscoveryMsg -> IO ())
192 -> MVar BackendState
193 -> Int
194 -> IO [NodeId]
195apiFindPeers send backendState delay = do
196 send PeerDiscoveryRequest
197 threadDelay delay
198 Set.toList . (^. peers) <$> readMVar backendState
200data PeerDiscoveryMsg =
201 PeerDiscoveryRequest
202 | PeerDiscoveryReply NodeId
204instance Binary PeerDiscoveryMsg where
205 put PeerDiscoveryRequest = putWord8 0
206 put (PeerDiscoveryReply nid) = putWord8 1 >> put nid
207 get = do
208 header <- getWord8
209 case header of
210 0 -> return PeerDiscoveryRequest
211 1 -> PeerDiscoveryReply <$> get
212 _ -> fail "PeerDiscoveryMsg.get: invalid"
214-- | Respond to peer discovery requests sent by other nodes
215peerDiscoveryDaemon :: MVar BackendState
216 -> IO (PeerDiscoveryMsg, N.SockAddr)
217 -> (PeerDiscoveryMsg -> IO ())
218 -> IO ()
219peerDiscoveryDaemon backendState recv send = forever go
220 where
221 go = do
222 (msg, _) <- recv
223 case msg of
224 PeerDiscoveryRequest -> do
225 nodes <- (^. localNodes) <$> readMVar backendState
226 forM_ nodes $ send . PeerDiscoveryReply . Node.localNodeId
227 PeerDiscoveryReply nid ->
228 modifyMVar_ backendState $ return . (peers ^: Set.insert nid)
230--------------------------------------------------------------------------------
231-- Back-end specific primitives --
232--------------------------------------------------------------------------------
234-- | Make sure that all log messages are printed by the logger on this node
235apiRedirectLogsHere :: Backend -> Process ()
236apiRedirectLogsHere backend = do
237 mLogger <- whereis "logger"
238 forM_ mLogger $ \logger -> do
239 nids <- liftIO $ findPeers backend 1000000
240 forM_ nids $ \nid -> registerRemote nid "logger" logger
242--------------------------------------------------------------------------------
243-- Slaves --
244--------------------------------------------------------------------------------
246-- | Messages to slave nodes
247--
248-- This datatype is not exposed; instead, we expose primitives for dealing
249-- with slaves.
250data SlaveControllerMsg =
251 SlaveTerminate
252 deriving (Typeable, Show)
254instance Binary SlaveControllerMsg where
255 put SlaveTerminate = putWord8 0
256 get = do
257 header <- getWord8
258 case header of
259 0 -> return SlaveTerminate
260 _ -> fail "SlaveControllerMsg.get: invalid"
262-- | Calling 'slave' sets up a new local node and then waits. You start
263-- processes on the slave by calling 'spawn' from other nodes.
264--
265-- This function does not return. The only way to exit the slave is to CTRL-C
266-- the process or call terminateSlave from another node.
267startSlave :: Backend -> IO ()
268startSlave backend = do
269 node <- newLocalNode backend
270 Node.runProcess node slaveController
272-- | The slave controller interprets 'SlaveControllerMsg's
273slaveController :: Process ()
274slaveController = do
275 pid <- getSelfPid
276 register "slaveController" pid
277 go
278 where
279 go = do
280 msg <- expect
281 case msg of
282 SlaveTerminate -> return ()
284-- | Terminate the slave at the given node ID
285terminateSlave :: NodeId -> Process ()
286terminateSlave nid = nsendRemote nid "slaveController" SlaveTerminate
288-- | Find slave nodes
289findSlaves :: Backend -> Process [NodeId]
290findSlaves backend = do
291 nodes <- liftIO $ findPeers backend 1000000
292 -- Fire of asynchronous requests for the slave controller
293 refs <- forM nodes $ \nid -> do
294 whereisRemoteAsync nid "slaveController"
295 ref <- monitorNode nid
296 return (nid, ref)
297 -- Wait for the replies
298 catMaybes <$> replicateM (length nodes) (
299 receiveWait
300 [ matchIf (\(WhereIsReply label _) -> label == "slaveController")
301 (\(WhereIsReply _ mPid) ->
302 case mPid of
303 Nothing ->
304 return Nothing
305 Just pid -> do
306 let nid = processNodeId pid
307 Just ref = lookup nid refs
308 unmonitor ref
309 return (Just nid))
310 , match (\(NodeMonitorNotification {}) -> return Nothing)
311 ])
313-- | Terminate all slaves
314terminateAllSlaves :: Backend -> Process ()
315terminateAllSlaves backend = do
316 slaves <- findSlaves backend
317 forM_ slaves terminateSlave
318 liftIO $ threadDelay 1000000
320--------------------------------------------------------------------------------
321-- Master nodes
322--------------------------------------------------------------------------------
324-- | 'startMaster' finds all slaves /currently/ available on the local network,
325-- redirects all log messages to itself, and then calls the specified process,
326-- passing the list of slaves nodes.
327--
328-- Terminates when the specified process terminates. If you want to terminate
329-- the slaves when the master terminates, you should manually call
330-- 'terminateAllSlaves'.
331--
332-- If you start more slave nodes after having started the master node, you can
333-- discover them with later calls to 'findSlaves', but be aware that you will
334-- need to call 'redirectLogHere' to redirect their logs to the master node.
335--
336-- Note that you can use functionality of "SimpleLocalnet" directly (through
337-- 'Backend'), instead of using 'startMaster'/'startSlave', if the master/slave
338-- distinction does not suit your application.
339startMaster :: Backend -> ([NodeId] -> Process ()) -> IO ()
340startMaster backend proc = do
341 node <- newLocalNode backend
342 Node.runProcess node $ do
343 slaves <- findSlaves backend
344 redirectLogsHere backend
345 proc slaves
347--------------------------------------------------------------------------------
348-- Accessors --
349--------------------------------------------------------------------------------
351localNodes :: Accessor BackendState [Node.LocalNode]
352localNodes = accessor _localNodes (\ns st -> st { _localNodes = ns })
354peers :: Accessor BackendState (Set NodeId)
355peers = accessor _peers (\ps st -> st { _peers = ps })