git.iko.soy/distributed-process
on ryan help sign in
distributed-process/distributed-process-simplelocalnet/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs
332 lines 12 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)
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 , matchIf
118 , processNodeId
119 )
120import qualified Control.Distributed.Process.Node as Node
121 ( LocalNode
122 , newLocalNode
123 , localNodeId
124 , runProcess
125 )
126import qualified Network.Transport.TCP as NT
127 ( createTransport
128 , defaultTCPParameters
129 )
130import qualified Network.Transport as NT (Transport)
131import qualified Network.Socket as N (HostName, ServiceName, SockAddr)
132import Control.Distributed.Process.Backend.SimpleLocalnet.Internal.Multicast (initMulticast)
134-- | Local backend
135data Backend = Backend {
136 -- | Create a new local node
137 newLocalNode :: IO Node.LocalNode
138 -- | @findPeers t@ broadcasts a /who's there?/ message on the local
139 -- network, waits 't' msec, and then collects and returns the answers.
140 -- You can use this to dynamically discover peer nodes.
141 , findPeers :: Int -> IO [NodeId]
142 -- | Make sure that all log messages are printed by the logger on the
143 -- current node
144 , redirectLogsHere :: Process ()
145 }
147data BackendState = BackendState {
148 _localNodes :: [Node.LocalNode]
149 , _peers :: Set NodeId
150 , discoveryDaemon :: ThreadId
151 }
153-- | Initialize the backend
154initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO Backend
155initializeBackend host port rtable = do
156 mTransport <- NT.createTransport host port NT.defaultTCPParameters
157 (recv, send) <- initMulticast "224.0.0.99" 9999 1024
158 (_, backendState) <- fixIO $ \ ~(tid, _) -> do
159 backendState <- newMVar BackendState
160 { _localNodes = []
161 , _peers = Set.empty
162 , discoveryDaemon = tid
163 }
164 tid' <- forkIO $ peerDiscoveryDaemon backendState recv send
165 return (tid', backendState)
166 case mTransport of
167 Left err -> throw err
168 Right transport ->
169 let backend = Backend {
170 newLocalNode = apiNewLocalNode transport rtable backendState
171 , findPeers = apiFindPeers send backendState
172 , redirectLogsHere = apiRedirectLogsHere backend
173 }
174 in return backend
176-- | Create a new local node
177apiNewLocalNode :: NT.Transport
178 -> RemoteTable
179 -> MVar BackendState
180 -> IO Node.LocalNode
181apiNewLocalNode transport rtable backendState = do
182 localNode <- Node.newLocalNode transport rtable
183 modifyMVar_ backendState $ return . (localNodes ^: (localNode :))
184 return localNode
186-- | Peer discovery
187apiFindPeers :: (PeerDiscoveryMsg -> IO ())
188 -> MVar BackendState
189 -> Int
190 -> IO [NodeId]
191apiFindPeers send backendState delay = do
192 send PeerDiscoveryRequest
193 threadDelay delay
194 Set.toList . (^. peers) <$> readMVar backendState
196data PeerDiscoveryMsg =
197 PeerDiscoveryRequest
198 | PeerDiscoveryReply NodeId
200instance Binary PeerDiscoveryMsg where
201 put PeerDiscoveryRequest = putWord8 0
202 put (PeerDiscoveryReply nid) = putWord8 1 >> put nid
203 get = do
204 header <- getWord8
205 case header of
206 0 -> return PeerDiscoveryRequest
207 1 -> PeerDiscoveryReply <$> get
208 _ -> fail "PeerDiscoveryMsg.get: invalid"
210-- | Respond to peer discovery requests sent by other nodes
211peerDiscoveryDaemon :: MVar BackendState
212 -> IO (PeerDiscoveryMsg, N.SockAddr)
213 -> (PeerDiscoveryMsg -> IO ())
214 -> IO ()
215peerDiscoveryDaemon backendState recv send = forever go
216 where
217 go = do
218 (msg, _) <- recv
219 case msg of
220 PeerDiscoveryRequest -> do
221 nodes <- (^. localNodes) <$> readMVar backendState
222 forM_ nodes $ send . PeerDiscoveryReply . Node.localNodeId
223 PeerDiscoveryReply nid ->
224 modifyMVar_ backendState $ return . (peers ^: Set.insert nid)
226--------------------------------------------------------------------------------
227-- Back-end specific primitives --
228--------------------------------------------------------------------------------
230-- | Make sure that all log messages are printed by the logger on this node
231apiRedirectLogsHere :: Backend -> Process ()
232apiRedirectLogsHere backend = do
233 mLogger <- whereis "logger"
234 forM_ mLogger $ \logger -> do
235 nids <- liftIO $ findPeers backend 1000000
236 forM_ nids $ \nid -> registerRemote nid "logger" logger
238--------------------------------------------------------------------------------
239-- Slaves --
240--------------------------------------------------------------------------------
242-- | Messages to slave nodes
243--
244-- This datatype is not exposed; instead, we expose primitives for dealing
245-- with slaves.
246data SlaveControllerMsg =
247 SlaveTerminate
248 deriving (Typeable, Show)
250instance Binary SlaveControllerMsg where
251 put SlaveTerminate = putWord8 0
252 get = do
253 header <- getWord8
254 case header of
255 0 -> return SlaveTerminate
256 _ -> fail "SlaveControllerMsg.get: invalid"
258-- | Calling 'slave' sets up a new local node and then waits. You start
259-- processes on the slave by calling 'spawn' from other nodes.
260--
261-- This function does not return. The only way to exit the slave is to CTRL-C
262-- the process or call terminateSlave from another node.
263startSlave :: Backend -> IO ()
264startSlave backend = do
265 node <- newLocalNode backend
266 Node.runProcess node slaveController
268-- | The slave controller interprets 'SlaveControllerMsg's
269slaveController :: Process ()
270slaveController = do
271 pid <- getSelfPid
272 register "slaveController" pid
273 go
274 where
275 go = do
276 msg <- expect
277 case msg of
278 SlaveTerminate -> return ()
280-- | Terminate the slave at the given node ID
281terminateSlave :: NodeId -> Process ()
282terminateSlave nid = nsendRemote nid "slaveController" SlaveTerminate
284-- | Find slave nodes
285findSlaves :: Backend -> Process [NodeId]
286findSlaves backend = do
287 nodes <- liftIO $ findPeers backend 1000000
288 -- Fire of asynchronous requests for the slave controller
289 forM_ nodes $ \nid -> whereisRemoteAsync nid "slaveController"
290 -- Wait for the replies
291 catMaybes <$> forM nodes (\_ ->
292 receiveWait
293 [ matchIf (\(WhereIsReply label _) -> label == "slaveController")
294 (\(WhereIsReply _ mPid) -> return (processNodeId <$> mPid))
295 ])
297-- | Terminate all slaves
298terminateAllSlaves :: Backend -> Process ()
299terminateAllSlaves backend = do
300 slaves <- findSlaves backend
301 forM_ slaves terminateSlave
302 liftIO $ threadDelay 1000000
304--------------------------------------------------------------------------------
305-- Master nodes
306--------------------------------------------------------------------------------
308-- | 'startMaster' finds all slaves currently available on the local network
309-- (which should therefore be started first), redirects all log messages to
310-- itself, and then calls the specified process, passing the list of slaves
311-- nodes.
312--
313-- Terminates when the specified process terminates. If you want to terminate
314-- the slaves when the master terminates, you should manually call
315-- 'terminateAllSlaves'.
316startMaster :: Backend -> ([NodeId] -> Process ()) -> IO ()
317startMaster backend proc = do
318 node <- newLocalNode backend
319 Node.runProcess node $ do
320 slaves <- findSlaves backend
321 redirectLogsHere backend
322 proc slaves
324--------------------------------------------------------------------------------
325-- Accessors --
326--------------------------------------------------------------------------------
328localNodes :: Accessor BackendState [Node.LocalNode]
329localNodes = accessor _localNodes (\ns st -> st { _localNodes = ns })
331peers :: Accessor BackendState (Set NodeId)
332peers = accessor _peers (\ps st -> st { _peers = ps })