git.iko.soy/distributed-process
on master help sign in
Properly implement peer discovery
81ce9d5Edsko de Vries
distributed-process/distributed-process.cabal+23−3
compare with master
unifiedsplit
@@ -29,7 +29,7 @@ Library
29 Control.Distributed.Process.Closure,
29 Control.Distributed.Process.Closure,
30 Control.Distributed.Process.Node
30 Control.Distributed.Process.Node
31 Control.Distributed.Process.Backend.Local
31 Control.Distributed.Process.Backend.Local
32− Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs
32+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes
33 ghc-options: -Wall
33 ghc-options: -Wall
34 HS-Source-Dirs: src
34 HS-Source-Dirs: src
35
35
@@ -49,7 +49,7 @@ Test-Suite TestCH
49 stm,
49 stm,
50 ghc-prim,
50 ghc-prim,
51 network-transport >= 0.1
51 network-transport >= 0.1
52− Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs
52+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes
53 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
53 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
54 HS-Source-Dirs: tests src
54 HS-Source-Dirs: tests src
55
55
@@ -70,6 +70,26 @@ Test-Suite TestClosure
70 template-haskell,
70 template-haskell,
71 stm,
71 stm,
72 network-transport >= 0.1
72 network-transport >= 0.1
73− Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs
73+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes
74+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
75+ HS-Source-Dirs: tests src
76+
77+-- Not a proper test, but we want to use cabal to compile it
78+Test-Suite TestLocalBackend
79+ Type: exitcode-stdio-1.0
80+ Main-Is: TestLocalBackend.hs
81+ Build-Depends: base,
82+ bytestring,
83+ containers,
84+ mtl,
85+ binary,
86+ transformers,
87+ data-accessor,
88+ random,
89+ ansi-terminal,
90+ template-haskell,
91+ stm,
92+ network-transport >= 0.1
93+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell, ExistentialQuantification, RankNTypes, GADTs, ImpredicativeTypes
74 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
94 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
75 HS-Source-Dirs: tests src
95 HS-Source-Dirs: tests src
distributed-process/src/Control/Distributed/Process/Backend/Local.hs+80−36
@@ -1,18 +1,20 @@
1-- | Simple backend based on the TCP transport which offers node discovery
1-- | Simple backend based on the TCP transport which offers node discovery
2-- based on UDP multicast
2-- based on UDP multicast
3+{-# OPTIONS_GHC -fno-warn-orphans #-}
3module Control.Distributed.Process.Backend.Local
4module Control.Distributed.Process.Backend.Local
4 ( LocalBackend(..)
5 ( LocalBackend(..)
5 , initializeBackend
6 , initializeBackend
6 ) where
7 ) where
7
8
8−import Data.Binary (encode)
9+import Data.Binary (Binary(get, put), getWord8, putWord8)
9−import qualified Data.ByteString as BSS (concat, append)
10+import Data.Accessor (Accessor, accessor, (^:), (^.))
11+import Data.Set (Set)
10−import qualified Data.ByteString.Lazy as BSL (toChunks, length)
12+import qualified Data.Set as Set (insert, empty, toList)
11import Control.Applicative ((<$>))
13import Control.Applicative ((<$>))
12import Control.Exception (throw)
14import Control.Exception (throw)
13−import Control.Monad (forM_)
15+import Control.Monad (forM_, forever)
14−import Control.Concurrent (forkIO)
16+import Control.Concurrent (forkIO, threadDelay)
15−import Control.Concurrent.MVar (MVar, newMVar, readMVar)
17+import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)
16import Control.Distributed.Process.Internal.Types
18import Control.Distributed.Process.Internal.Types
17 ( LocalNode
19 ( LocalNode
18 , RemoteTable
20 , RemoteTable
@@ -26,55 +28,97 @@ import qualified Network.Transport.TCP as NT
26 ( createTransport
28 ( createTransport
27 , defaultTCPParameters
29 , defaultTCPParameters
28 )
30 )
29−import Network.Transport.Internal (encodeInt32, decodeInt32)
30−import qualified Network.Transport as NT ()
31+import qualified Network.Transport as NT (Transport)
31−import qualified Network.Socket as N (HostName, ServiceName, Socket, SockAddr)
32+import qualified Network.Socket as N (HostName, ServiceName, SockAddr)
32−import qualified Network.Socket.ByteString as NBS (recvFrom, sendTo)
33+import Control.Distributed.Process.Internal.Multicast (initMulticast)
33−import qualified Network.Multicast as NM (multicastSender, multicastReceiver)
34
34
35+-- | Local backend
35data LocalBackend = LocalBackend {
36data LocalBackend = LocalBackend {
37+ -- | Create a new local node
36 newLocalNode :: IO LocalNode
38 newLocalNode :: IO LocalNode
39+ -- | @findPeers t@ sends out a /who's there?/ request, waits 't' msec,
40+ -- and then collects and returns the answers
37− , findPeers :: IO [NodeId]
41+ , findPeers :: Int -> IO [NodeId]
38 }
42 }
39
43
40data LocalBackendState = LocalBackendState {
44data LocalBackendState = LocalBackendState {
41− localNodes :: [LocalNode]
45+ _localNodes :: [LocalNode]
46+ , _peers :: Set NodeId
42 }
47 }
43
48
44-- | Initialize the backend
49-- | Initialize the backend
45initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO LocalBackend
50initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO LocalBackend
46initializeBackend host port rtable = do
51initializeBackend host port rtable = do
47− mTransport <- NT.createTransport host port NT.defaultTCPParameters
52+ mTransport <- NT.createTransport host port NT.defaultTCPParameters
48− mcastSender <- NM.multicastSender "224.0.0.99" 9999
53+ (recv, send) <- initMulticast "224.0.0.99" 9999 1024
49− mcastReceiver <- NM.multicastReceiver "224.0.0.99" 9999
50− backendState <- newMVar LocalBackendState { localNodes = [] }
54+ backendState <- newMVar LocalBackendState
55+ { _localNodes = []
56+ , _peers = Set.empty
57+ }
51− forkIO $ peerDiscoveryDaemon backendState mcastReceiver mcastSender
58+ forkIO $ peerDiscoveryDaemon backendState recv send
52 case mTransport of
59 case mTransport of
53 Left err -> throw err
60 Left err -> throw err
54 Right transport -> return LocalBackend {
61 Right transport -> return LocalBackend {
55− newLocalNode = Node.newLocalNode transport rtable
62+ newLocalNode = apiNewLocalNode transport rtable backendState
56− , findPeers = apiFindPeers mcastSender
63+ , findPeers = apiFindPeers send backendState
57 }
64 }
58
65
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+
59-- | Peer discovery
76-- | Peer discovery
60−apiFindPeers :: (N.Socket, N.SockAddr) -> IO [NodeId]
77+apiFindPeers :: (PeerDiscoveryMsg -> IO ())
61−apiFindPeers (sock, addr) = do
78+ -> MVar LocalBackendState
62− NBS.sendTo sock (encodeInt32 (0xCDAECDAE :: Int)) addr
79+ -> Int
63− return []
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"
64
99
65-- | Respond to peer discovery requests sent by other nodes
100-- | Respond to peer discovery requests sent by other nodes
66peerDiscoveryDaemon :: MVar LocalBackendState
101peerDiscoveryDaemon :: MVar LocalBackendState
67− -> N.Socket
68− -> (N.Socket, N.SockAddr)
102+ -> IO (PeerDiscoveryMsg, N.SockAddr)
103+ -> (PeerDiscoveryMsg -> IO ())
69 -> IO ()
104 -> IO ()
70−peerDiscoveryDaemon backendState readSock (writeSock, addr) = do
105+peerDiscoveryDaemon backendState recv send = forever go
71− (msgLen, _remoteAddr) <- NBS.recvFrom readSock 4
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 })
72
122
73− case decodeInt32 msgLen of
123+peers :: Accessor LocalBackendState (Set NodeId)
74− (0xCDAECDAE :: Int) -> do
124+peers = accessor _peers (\ps st -> st { _peers = ps })
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
distributed-process/src/Control/Distributed/Process/Internal/Multicast.hsnew file+124
@@ -0,0 +1,124 @@
1+-- | Multicast utilities
2+{-# OPTIONS_GHC -fno-warn-orphans #-}
3+module Control.Distributed.Process.Internal.Multicast (initMulticast) where
4+
5+import Data.Function (on)
6+import Data.Map (Map)
7+import qualified Data.Map as Map (empty)
8+import Data.Binary (Binary, decode, encode)
9+import Data.IORef (IORef, newIORef, readIORef, modifyIORef)
10+import qualified Data.ByteString as BSS (ByteString, concat)
11+import qualified Data.ByteString.Lazy as BSL
12+ ( ByteString
13+ , empty
14+ , append
15+ , fromChunks
16+ , toChunks
17+ , length
18+ , splitAt
19+ )
20+import Data.Accessor (Accessor, (^:), (^.), (^=))
21+import qualified Data.Accessor.Container as DAC (mapDefault)
22+import Control.Applicative ((<$>))
23+import Network.Socket (HostName, PortNumber, Socket, SockAddr)
24+import qualified Network.Socket.ByteString as NBS (recvFrom, sendManyTo)
25+import Network.Transport.Internal (decodeInt32, encodeInt32)
26+import Network.Multicast (multicastSender, multicastReceiver)
27+
28+--------------------------------------------------------------------------------
29+-- Top-level API --
30+--------------------------------------------------------------------------------
31+
32+-- | Given a hostname and a port number, initialize the multicast system.
33+--
34+-- Note: it is important that you never send messages larger than the maximum
35+-- message size; if you do, all subsequent communication will probably fail.
36+--
37+-- Returns a reader and a writer.
38+initMulticast :: HostName -- ^ Multicast IP
39+ -> PortNumber -- ^ Port number
40+ -> Int -- ^ Maximum message size
41+ -> IO ( forall a. Binary a => IO (a, SockAddr)
42+ , forall a. Binary a => a -> IO ()
43+ )
44+initMulticast host port bufferSize = do
45+ (sendSock, sendAddr) <- multicastSender host port
46+ readSock <- multicastReceiver host port
47+ st <- newIORef Map.empty
48+ return (recvBinary readSock st bufferSize, writer sendSock sendAddr)
49+ where
50+ writer :: forall a. Binary a => Socket -> SockAddr -> a -> IO ()
51+ writer sock addr val = do
52+ let bytes = encode val
53+ len = encodeInt32 (BSL.length bytes)
54+ NBS.sendManyTo sock (len : BSL.toChunks bytes) addr
55+
56+--------------------------------------------------------------------------------
57+-- UDP multicast read, dealing with multiple senders --
58+--------------------------------------------------------------------------------
59+
60+type UDPState = Map SockAddr BSL.ByteString
61+
62+-- TODO: This is inefficient and an orphan instance.
63+-- Requested official instance (https://github.com/haskell/network/issues/38)
64+instance Ord SockAddr where
65+ compare = compare `on` show
66+
67+bufferFor :: SockAddr -> Accessor UDPState BSL.ByteString
68+bufferFor = DAC.mapDefault BSL.empty
69+
70+bufferAppend :: SockAddr -> BSS.ByteString -> UDPState -> UDPState
71+bufferAppend addr bytes =
72+ bufferFor addr ^: flip BSL.append (BSL.fromChunks [bytes])
73+
74+recvBinary :: Binary a => Socket -> IORef UDPState -> Int -> IO (a, SockAddr)
75+recvBinary sock st bufferSize = do
76+ (bytes, addr) <- recvWithLength sock st bufferSize
77+ return (decode bytes, addr)
78+
79+recvWithLength :: Socket
80+ -> IORef UDPState
81+ -> Int
82+ -> IO (BSL.ByteString, SockAddr)
83+recvWithLength sock st bufferSize = do
84+ (len, addr) <- recvExact sock 4 st bufferSize
85+ let n = decodeInt32 . BSS.concat . BSL.toChunks $ len
86+ bytes <- recvExactFrom addr sock n st bufferSize
87+ return (bytes, addr)
88+
89+-- Receive all bytes currently in the buffer
90+recvAll :: Socket -> IORef UDPState -> Int -> IO SockAddr
91+recvAll sock st bufferSize = do
92+ (bytes, addr) <- NBS.recvFrom sock bufferSize
93+ modifyIORef st $ bufferAppend addr bytes
94+ return addr
95+
96+recvExact :: Socket
97+ -> Int
98+ -> IORef UDPState
99+ -> Int
100+ -> IO (BSL.ByteString, SockAddr)
101+recvExact sock n st bufferSize = do
102+ addr <- recvAll sock st bufferSize
103+ bytes <- recvExactFrom addr sock n st bufferSize
104+ return (bytes, addr)
105+
106+recvExactFrom :: SockAddr
107+ -> Socket
108+ -> Int
109+ -> IORef UDPState
110+ -> Int
111+ -> IO BSL.ByteString
112+recvExactFrom addr sock n st bufferSize = go
113+ where
114+ go :: IO BSL.ByteString
115+ go = do
116+ accAddr <- (^. bufferFor addr) <$> readIORef st
117+ if BSL.length accAddr >= fromIntegral n
118+ then do
119+ let (bytes, accAddr') = BSL.splitAt (fromIntegral n) accAddr
120+ modifyIORef st $ bufferFor addr ^= accAddr'
121+ return bytes
122+ else do
123+ _ <- recvAll sock st bufferSize
124+ go
distributed-process/tests/TestLocalBackend.hs+6−2
@@ -2,14 +2,18 @@ import System.Environment (getArgs, getProgName)
2import Control.Distributed.Process.Node (initRemoteTable)
2import Control.Distributed.Process.Node (initRemoteTable)
3import Control.Distributed.Process.Backend.Local
3import Control.Distributed.Process.Backend.Local
4import Control.Concurrent (threadDelay)
4import Control.Concurrent (threadDelay)
5+import Control.Monad (replicateM)
5
6
6server :: LocalBackend -> IO ()
7server :: LocalBackend -> IO ()
7server backend = do
8server backend = do
9+ node <- newLocalNode backend
8− nodes <- findPeers backend
10+ nodes <- findPeers backend 1000000
9 print nodes
11 print nodes
10
12
11client :: LocalBackend -> IO ()
13client :: LocalBackend -> IO ()
14+client backend = do
15+ [node1, node2] <- replicateM 2 $ newLocalNode backend
12−client _ = threadDelay 30000000
16+ threadDelay 30000000
13
17
14main :: IO ()
18main :: IO ()
15main = do
19main = do
distributed-process/tests/runTestLocalBackend.hsnew file+8
@@ -0,0 +1,8 @@
1+#!/bin/bash
2+TestLocalBackend=dist/build/TestLocalBackend/TestLocalBackend
3+$TestLocalBackend client 127.0.0.1 8080 &
4+$TestLocalBackend client 127.0.0.1 8081 &
5+$TestLocalBackend client 127.0.0.1 8082 &
6+$TestLocalBackend client 127.0.0.1 8083 &
7+sleep 2 # give clients a change to start
8+$TestLocalBackend server 127.0.0.1 8084