@@ -0,0 +1,80 @@
1+-- | Simple backend based on the TCP transport which offers node discovery
2+-- based on UDP multicast
3+module Control.Distributed.Process.Backend.Local
4+ ( LocalBackend(..)
5+ , initializeBackend
6+ ) where
7+
8+import Data.Binary (encode)
9+import qualified Data.ByteString as BSS (concat, append)
10+import qualified Data.ByteString.Lazy as BSL (toChunks, length)
11+import Control.Applicative ((<$>))
12+import Control.Exception (throw)
13+import Control.Monad (forM_)
14+import Control.Concurrent (forkIO)
15+import Control.Concurrent.MVar (MVar, newMVar, readMVar)
16+import Control.Distributed.Process.Internal.Types
17+ ( LocalNode
18+ , RemoteTable
19+ , NodeId
20+ )
21+import qualified Control.Distributed.Process.Node as Node
22+ ( newLocalNode
23+ , localNodeId
24+ )
25+import qualified Network.Transport.TCP as NT
26+ ( createTransport
27+ , defaultTCPParameters
28+ )
29+import Network.Transport.Internal (encodeInt32, decodeInt32)
30+import qualified Network.Transport as NT ()
31+import qualified Network.Socket as N (HostName, ServiceName, Socket, SockAddr)
32+import qualified Network.Socket.ByteString as NBS (recvFrom, sendTo)
33+import qualified Network.Multicast as NM (multicastSender, multicastReceiver)
34+
35+data LocalBackend = LocalBackend {
36+ newLocalNode :: IO LocalNode
37+ , findPeers :: IO [NodeId]
38+ }
39+
40+data LocalBackendState = LocalBackendState {
41+ localNodes :: [LocalNode]
42+ }
43+
44+-- | Initialize the backend
45+initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO LocalBackend
46+initializeBackend host port rtable = do
47+ mTransport <- NT.createTransport host port NT.defaultTCPParameters
48+ mcastSender <- NM.multicastSender "224.0.0.99" 9999
49+ mcastReceiver <- NM.multicastReceiver "224.0.0.99" 9999
50+ backendState <- newMVar LocalBackendState { localNodes = [] }
51+ forkIO $ peerDiscoveryDaemon backendState mcastReceiver mcastSender
52+ case mTransport of
53+ Left err -> throw err
54+ Right transport -> return LocalBackend {
55+ newLocalNode = Node.newLocalNode transport rtable
56+ , findPeers = apiFindPeers mcastSender
57+ }
58+
59+-- | Peer discovery
60+apiFindPeers :: (N.Socket, N.SockAddr) -> IO [NodeId]
61+apiFindPeers (sock, addr) = do
62+ NBS.sendTo sock (encodeInt32 (0xCDAECDAE :: Int)) addr
63+ return []
64+
65+-- | Respond to peer discovery requests sent by other nodes
66+peerDiscoveryDaemon :: MVar LocalBackendState
67+ -> N.Socket
68+ -> (N.Socket, N.SockAddr)
69+ -> IO ()
70+peerDiscoveryDaemon backendState readSock (writeSock, addr) = do
71+ (msgLen, _remoteAddr) <- NBS.recvFrom readSock 4
72+
73+ case decodeInt32 msgLen of
74+ (0xCDAECDAE :: Int) -> do
75+ nodes <- localNodes <$> readMVar backendState
76+ forM_ nodes $ \node -> do
77+ let nodeAddr = encode (Node.localNodeId node)
78+ msg = encodeInt32 (BSL.length nodeAddr) `BSS.append`
79+ (BSS.concat . BSL.toChunks $ nodeAddr)
80+ NBS.sendTo writeSock msg addr