@@ -1,18 +1,20 @@
11-- | Simple backend based on the TCP transport which offers node discovery
22-- based on UDP multicast
3+{-# OPTIONS_GHC -fno-warn-orphans #-}
34module Control.Distributed.Process.Backend.Local
45 ( LocalBackend(..)
56 , initializeBackend
67 ) where
78
8−import Data.Binary (encode)
9−import qualified Data.ByteString as BSS (concat, append)
10−import qualified Data.ByteString.Lazy as BSL (toChunks, length)
9+import Data.Binary (Binary(get, put), getWord8, putWord8)
10+import Data.Accessor (Accessor, accessor, (^:), (^.))
11+import Data.Set (Set)
12+import qualified Data.Set as Set (insert, empty, toList)
1113import Control.Applicative ((<$>))
1214import Control.Exception (throw)
13−import Control.Monad (forM_)
14−import Control.Concurrent (forkIO)
15−import Control.Concurrent.MVar (MVar, newMVar, readMVar)
15+import Control.Monad (forM_, forever)
16+import Control.Concurrent (forkIO, threadDelay)
17+import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)
1618import Control.Distributed.Process.Internal.Types
1719 ( LocalNode
1820 , RemoteTable
@@ -26,55 +28,97 @@ import qualified Network.Transport.TCP as NT
2628 ( createTransport
2729 , defaultTCPParameters
2830 )
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)
31+import qualified Network.Transport as NT (Transport)
32+import qualified Network.Socket as N (HostName, ServiceName, SockAddr)
33+import Control.Distributed.Process.Internal.Multicast (initMulticast)
3434
35+-- | Local backend
3536data LocalBackend = LocalBackend {
37+ -- | Create a new local node
3638 newLocalNode :: IO LocalNode
37− , findPeers :: IO [NodeId]
39+ -- | @findPeers t@ sends out a /who's there?/ request, waits 't' msec,
40+ -- and then collects and returns the answers
41+ , findPeers :: Int -> IO [NodeId]
3842 }
3943
4044data LocalBackendState = LocalBackendState {
41− localNodes :: [LocalNode]
45+ _localNodes :: [LocalNode]
46+ , _peers :: Set NodeId
4247 }
4348
4449-- | Initialize the backend
4550initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO LocalBackend
4651initializeBackend 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+ mTransport <- NT.createTransport host port NT.defaultTCPParameters
53+ (recv, send) <- initMulticast "224.0.0.99" 9999 1024
54+ backendState <- newMVar LocalBackendState
55+ { _localNodes = []
56+ , _peers = Set.empty
57+ }
58+ forkIO $ peerDiscoveryDaemon backendState recv send
5259 case mTransport of
5360 Left err -> throw err
5461 Right transport -> return LocalBackend {
55− newLocalNode = Node.newLocalNode transport rtable
56− , findPeers = apiFindPeers mcastSender
62+ newLocalNode = apiNewLocalNode transport rtable backendState
63+ , findPeers = apiFindPeers send backendState
5764 }
5865
66+-- | Create a new local node
67+apiNewLocalNode :: NT.Transport
68+ -> RemoteTable
69+ -> MVar LocalBackendState
70+ -> IO LocalNode
71+apiNewLocalNode transport rtable backendState = do
72+ localNode <- Node.newLocalNode transport rtable
73+ modifyMVar_ backendState $ return . (localNodes ^: (localNode :))
74+ return localNode
75+
5976-- | 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 []
77+apiFindPeers :: (PeerDiscoveryMsg -> IO ())
78+ -> MVar LocalBackendState
79+ -> Int
80+ -> IO [NodeId]
81+apiFindPeers send backendState delay = do
82+ send PeerDiscoveryRequest
83+ threadDelay delay
84+ Set.toList . (^. peers) <$> readMVar backendState
85+
86+data PeerDiscoveryMsg =
87+ PeerDiscoveryRequest
88+ | PeerDiscoveryReply NodeId
89+
90+instance Binary PeerDiscoveryMsg where
91+ put PeerDiscoveryRequest = putWord8 0
92+ put (PeerDiscoveryReply nid) = putWord8 1 >> put nid
93+ get = do
94+ header <- getWord8
95+ case header of
96+ 0 -> return PeerDiscoveryRequest
97+ 1 -> PeerDiscoveryReply <$> get
98+ _ -> fail "PeerDiscoveryMsg.get: invalid"
6499
65100-- | Respond to peer discovery requests sent by other nodes
66101peerDiscoveryDaemon :: MVar LocalBackendState
67− -> N.Socket
68− -> (N.Socket, N.SockAddr)
102+ -> IO (PeerDiscoveryMsg, N.SockAddr)
103+ -> (PeerDiscoveryMsg -> IO ())
69104 -> IO ()
70−peerDiscoveryDaemon backendState readSock (writeSock, addr) = do
71− (msgLen, _remoteAddr) <- NBS.recvFrom readSock 4
105+peerDiscoveryDaemon backendState recv send = forever go
106+ where
107+ go = do
108+ (msg, _) <- recv
109+ case msg of
110+ PeerDiscoveryRequest -> do
111+ nodes <- (^. localNodes) <$> readMVar backendState
112+ forM_ nodes $ send . PeerDiscoveryReply . Node.localNodeId
113+ PeerDiscoveryReply nid ->
114+ modifyMVar_ backendState $ return . (peers ^: Set.insert nid)
115+
116+--------------------------------------------------------------------------------
117+-- Accessors --
118+--------------------------------------------------------------------------------
119+
120+localNodes :: Accessor LocalBackendState [LocalNode]
121+localNodes = accessor _localNodes (\ns st -> st { _localNodes = ns })
72122
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
123+peers :: Accessor LocalBackendState (Set NodeId)
124+peers = accessor _peers (\ps st -> st { _peers = ps })