diff --git a/distributed-process-simplelocalnet.cabal b/distributed-process-simplelocalnet.cabal --- a/distributed-process-simplelocalnet.cabal +++ b/distributed-process-simplelocalnet.cabal @@ -1,15 +1,15 @@ Name: distributed-process-simplelocalnet -Version: 0.2.0.8 +Version: 0.2.0.9 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 License-File: LICENSE Copyright: Well-Typed LLP Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -Maintainer: edsko@well-typed.com, duncan@well-typed.com +Maintainer: watson.timothy@gmail.com, edsko@well-typed.com, duncan@well-typed.com Stability: experimental Homepage: http://github.com/haskell-distributed/distributed-process -Bug-Reports: mailto:edsko@well-typed.com +Bug-Reports: http://github.com/haskell-distributed/distributed-process/issues Synopsis: Simple zero-configuration backend for Cloud Haskell Description: Simple backend based on the TCP transport which offers node discovery based on UDP multicast. This is a zero-configuration @@ -38,7 +38,7 @@ transformers >= 0.2 && < 0.4, network-transport >= 0.3 && < 0.4, network-transport-tcp >= 0.3 && < 0.4, - distributed-process >= 0.4.1 && < 0.5 + distributed-process >= 0.4.2 && < 0.5 Exposed-modules: Control.Distributed.Process.Backend.SimpleLocalnet, Control.Distributed.Process.Backend.SimpleLocalnet.Internal.Multicast Extensions: RankNTypes, diff --git a/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs b/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs --- a/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs +++ b/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs @@ -3,41 +3,41 @@ -- get you going with Cloud Haskell quickly without imposing any structure -- on your application. -- --- To simplify getting started we provide special support for /master/ and +-- To simplify getting started we provide special support for /master/ and -- /slave/ nodes (see 'startSlave' and 'startMaster'). Use of these functions -- is completely optional; you can use the local backend without making use -- of the predefined master and slave nodes. --- +-- -- [Minimal example] -- -- > import System.Environment (getArgs) -- > import Control.Distributed.Process -- > import Control.Distributed.Process.Node (initRemoteTable) -- > import Control.Distributed.Process.Backend.SimpleLocalnet --- > +-- > -- > master :: Backend -> [NodeId] -> Process () -- > master backend slaves = do -- > -- Do something interesting with the slaves -- > liftIO . putStrLn $ "Slaves: " ++ show slaves -- > -- Terminate the slaves when the master terminates (this is optional) -- > terminateAllSlaves backend --- > +-- > -- > main :: IO () -- > main = do -- > args <- getArgs --- > +-- > -- > case args of -- > ["master", host, port] -> do --- > backend <- initializeBackend host port initRemoteTable +-- > backend <- initializeBackend host port initRemoteTable -- > startMaster backend (master backend) -- > ["slave", host, port] -> do --- > backend <- initializeBackend host port initRemoteTable +-- > backend <- initializeBackend host port initRemoteTable -- > startSlave backend --- +-- -- [Compiling and Running] -- -- Save to @example.hs@ and compile using --- +-- -- > ghc -threaded example.hs -- -- Fire up some slave nodes (for the example, we run them on a single machine): @@ -68,7 +68,7 @@ -- master on a fifth node (or on any of the four machines that run the slave -- nodes). -- --- It is important that every node has a unique (hostname, port number) pair, +-- It is important that every node has a unique (hostname, port number) pair, -- and that the hostname you use to initialize the node can be resolved by -- peer nodes. In other words, if you start a node and pass hostname @localhost@ -- then peer nodes won't be able to reach it because @localhost@ will resolve @@ -79,10 +79,10 @@ -- If you try the above example and the master process cannot find any slaves, -- then it might be that your firewall settings do not allow for UDP multicast -- (in particular, the default iptables on some Linux distributions might not --- allow it). +-- allow it). {-# OPTIONS_GHC -fno-warn-orphans #-} module Control.Distributed.Process.Backend.SimpleLocalnet - ( -- * Initialization + ( -- * Initialization Backend(..) , initializeBackend -- * Slave nodes @@ -104,7 +104,7 @@ import Data.Typeable (Typeable) import Control.Applicative ((<$>)) import Control.Exception (throw) -import Control.Monad (forever, forM, replicateM) +import Control.Monad (forever, replicateM, replicateM_) import Control.Monad.IO.Class (liftIO) import Control.Concurrent (forkIO, threadDelay, ThreadId) import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_) @@ -112,29 +112,39 @@ ( RemoteTable , NodeId , Process + , ProcessId , WhereIsReply(..) , whereis , whereisRemoteAsync - , reregisterRemoteAsync , getSelfPid , register + , reregister , expect , nsendRemote , receiveWait , match - , matchIf , processNodeId , monitorNode + , monitor , unmonitor , NodeMonitorNotification(..) + , ProcessRegistrationException + , finally + , newChan + , receiveChan + , nsend + , SendPort + , bracket + , try + , send ) -import qualified Control.Distributed.Process.Node as Node +import qualified Control.Distributed.Process.Node as Node ( LocalNode , newLocalNode , localNodeId , runProcess ) -import qualified Network.Transport.TCP as NT +import qualified Network.Transport.TCP as NT ( createTransport , defaultTCPParameters ) @@ -142,7 +152,7 @@ import qualified Network.Socket as N (HostName, ServiceName, SockAddr) import Control.Distributed.Process.Backend.SimpleLocalnet.Internal.Multicast (initMulticast) --- | Local backend +-- | Local backend data Backend = Backend { -- | Create a new local node newLocalNode :: IO Node.LocalNode @@ -152,7 +162,7 @@ , findPeers :: Int -> IO [NodeId] -- | Make sure that all log messages are printed by the logger on the -- current node - , redirectLogsHere :: Process () + , redirectLogsHere :: [ProcessId] -> Process () } data BackendState = BackendState { @@ -164,73 +174,73 @@ -- | Initialize the backend initializeBackend :: N.HostName -> N.ServiceName -> RemoteTable -> IO Backend initializeBackend host port rtable = do - mTransport <- NT.createTransport host port NT.defaultTCPParameters - (recv, send) <- initMulticast "224.0.0.99" 9999 1024 + mTransport <- NT.createTransport host port NT.defaultTCPParameters + (recv, sendp) <- initMulticast "224.0.0.99" 9999 1024 (_, backendState) <- fixIO $ \ ~(tid, _) -> do - backendState <- newMVar BackendState - { _localNodes = [] + backendState <- newMVar BackendState + { _localNodes = [] , _peers = Set.empty , discoveryDaemon = tid } - tid' <- forkIO $ peerDiscoveryDaemon backendState recv send + tid' <- forkIO $ peerDiscoveryDaemon backendState recv sendp return (tid', backendState) case mTransport of Left err -> throw err - Right transport -> + Right transport -> let backend = Backend { - newLocalNode = apiNewLocalNode transport rtable backendState - , findPeers = apiFindPeers send backendState - , redirectLogsHere = apiRedirectLogsHere backend + newLocalNode = apiNewLocalNode transport rtable backendState + , findPeers = apiFindPeers sendp backendState + , redirectLogsHere = apiRedirectLogsHere backend } in return backend -- | Create a new local node -apiNewLocalNode :: NT.Transport - -> RemoteTable +apiNewLocalNode :: NT.Transport + -> RemoteTable -> MVar BackendState -> IO Node.LocalNode apiNewLocalNode transport rtable backendState = do - localNode <- Node.newLocalNode transport rtable + localNode <- Node.newLocalNode transport rtable modifyMVar_ backendState $ return . (localNodes ^: (localNode :)) return localNode -- | Peer discovery -apiFindPeers :: (PeerDiscoveryMsg -> IO ()) - -> MVar BackendState +apiFindPeers :: (PeerDiscoveryMsg -> IO ()) + -> MVar BackendState -> Int -> IO [NodeId] -apiFindPeers send backendState delay = do - send PeerDiscoveryRequest - threadDelay delay - Set.toList . (^. peers) <$> readMVar backendState +apiFindPeers sendfn backendState delay = do + sendfn PeerDiscoveryRequest + threadDelay delay + Set.toList . (^. peers) <$> readMVar backendState -data PeerDiscoveryMsg = - PeerDiscoveryRequest +data PeerDiscoveryMsg = + PeerDiscoveryRequest | PeerDiscoveryReply NodeId instance Binary PeerDiscoveryMsg where put PeerDiscoveryRequest = putWord8 0 put (PeerDiscoveryReply nid) = putWord8 1 >> put nid get = do - header <- getWord8 + header <- getWord8 case header of 0 -> return PeerDiscoveryRequest 1 -> PeerDiscoveryReply <$> get _ -> fail "PeerDiscoveryMsg.get: invalid" -- | Respond to peer discovery requests sent by other nodes -peerDiscoveryDaemon :: MVar BackendState +peerDiscoveryDaemon :: MVar BackendState -> IO (PeerDiscoveryMsg, N.SockAddr) - -> (PeerDiscoveryMsg -> IO ()) + -> (PeerDiscoveryMsg -> IO ()) -> IO () -peerDiscoveryDaemon backendState recv send = forever go +peerDiscoveryDaemon backendState recv sendfn = forever go where go = do (msg, _) <- recv case msg of PeerDiscoveryRequest -> do nodes <- (^. localNodes) <$> readMVar backendState - forM_ nodes $ send . PeerDiscoveryReply . Node.localNodeId + forM_ nodes $ sendfn . PeerDiscoveryReply . Node.localNodeId PeerDiscoveryReply nid -> modifyMVar_ backendState $ return . (peers ^: Set.insert nid) @@ -239,13 +249,28 @@ -------------------------------------------------------------------------------- -- | Make sure that all log messages are printed by the logger on this node -apiRedirectLogsHere :: Backend -> Process () -apiRedirectLogsHere backend = do +apiRedirectLogsHere :: Backend -> [ProcessId] -> Process () +apiRedirectLogsHere _backend slavecontrollers = do mLogger <- whereis "logger" + myPid <- getSelfPid + forM_ mLogger $ \logger -> do - nids <- liftIO $ findPeers backend 1000000 - forM_ nids $ \nid -> reregisterRemoteAsync nid "logger" logger -- ignore async response + bracket + (mapM monitor slavecontrollers) + (mapM unmonitor) + $ \_ -> do + + -- fire off redirect requests + forM_ slavecontrollers $ \pid -> send pid (RedirectLogsTo logger myPid) + + -- Wait for the replies + replicateM_ (length slavecontrollers) $ do + receiveWait + [ match (\(RedirectLogsReply {}) -> return ()) + , match (\(NodeMonitorNotification {}) -> return ()) + ] + -------------------------------------------------------------------------------- -- Slaves -- -------------------------------------------------------------------------------- @@ -254,18 +279,31 @@ -- -- This datatype is not exposed; instead, we expose primitives for dealing -- with slaves. -data SlaveControllerMsg = - SlaveTerminate +data SlaveControllerMsg + = SlaveTerminate + | RedirectLogsTo ProcessId ProcessId deriving (Typeable, Show) instance Binary SlaveControllerMsg where - put SlaveTerminate = putWord8 0 + put SlaveTerminate = putWord8 0 + put (RedirectLogsTo a b) = do putWord8 1; put (a,b) get = do header <- getWord8 case header of 0 -> return SlaveTerminate + 1 -> do (a,b) <- get; return (RedirectLogsTo a b) _ -> fail "SlaveControllerMsg.get: invalid" +data RedirectLogsReply + = RedirectLogsReply ProcessId Bool + deriving (Typeable, Show) + +instance Binary RedirectLogsReply where + put (RedirectLogsReply from ok) = put (from,ok) + get = do + (from,ok) <- get + return (RedirectLogsReply from ok) + -- | Calling 'slave' sets up a new local node and then waits. You start -- processes on the slave by calling 'spawn' from other nodes. -- @@ -273,8 +311,8 @@ -- the process or call terminateSlave from another node. startSlave :: Backend -> IO () startSlave backend = do - node <- newLocalNode backend - Node.runProcess node slaveController + node <- newLocalNode backend + Node.runProcess node slaveController -- | The slave controller interprets 'SlaveControllerMsg's slaveController :: Process () @@ -287,41 +325,49 @@ msg <- expect case msg of SlaveTerminate -> return () + RedirectLogsTo loggerPid from -> do + r <- try (reregister "logger" loggerPid) + ok <- case (r :: Either ProcessRegistrationException ()) of + Right _ -> return True + Left _ -> do + s <- try (register "logger" loggerPid) + case (s :: Either ProcessRegistrationException ()) of + Right _ -> return True + Left _ -> return False + pid <- getSelfPid + send from (RedirectLogsReply pid ok) + go -- | Terminate the slave at the given node ID terminateSlave :: NodeId -> Process () terminateSlave nid = nsendRemote nid "slaveController" SlaveTerminate -- | Find slave nodes -findSlaves :: Backend -> Process [NodeId] +findSlaves :: Backend -> Process [ProcessId] findSlaves backend = do - nodes <- liftIO $ findPeers backend 1000000 - -- Fire of asynchronous requests for the slave controller - refs <- forM nodes $ \nid -> do - whereisRemoteAsync nid "slaveController" - ref <- monitorNode nid - return (nid, ref) - -- Wait for the replies - catMaybes <$> replicateM (length nodes) ( - receiveWait - [ matchIf (\(WhereIsReply label _) -> label == "slaveController") - (\(WhereIsReply _ mPid) -> - case mPid of - Nothing -> - return Nothing - Just pid -> do - let nid = processNodeId pid - Just ref = lookup nid refs - unmonitor ref - return (Just nid)) - , match (\(NodeMonitorNotification {}) -> return Nothing) - ]) + nodes <- liftIO $ findPeers backend 1000000 + -- Fire off asynchronous requests for the slave controller + bracket + (mapM monitorNode nodes) + (mapM unmonitor) + $ \_ -> do + + -- fire off whereis requests + forM_ nodes $ \nid -> whereisRemoteAsync nid "slaveController" + + -- Wait for the replies + catMaybes <$> replicateM (length nodes) ( + receiveWait + [ match (\(WhereIsReply "slaveController" mPid) -> return mPid) + , match (\(NodeMonitorNotification {}) -> return Nothing) + ]) + -- | Terminate all slaves terminateAllSlaves :: Backend -> Process () terminateAllSlaves backend = do slaves <- findSlaves backend - forM_ slaves terminateSlave + forM_ slaves $ \pid -> send pid SlaveTerminate liftIO $ threadDelay 1000000 -------------------------------------------------------------------------------- @@ -330,15 +376,15 @@ -- | 'startMaster' finds all slaves /currently/ available on the local network, -- redirects all log messages to itself, and then calls the specified process, --- passing the list of slaves nodes. +-- passing the list of slaves nodes. -- -- Terminates when the specified process terminates. If you want to terminate --- the slaves when the master terminates, you should manually call +-- the slaves when the master terminates, you should manually call -- 'terminateAllSlaves'. -- -- If you start more slave nodes after having started the master node, you can -- discover them with later calls to 'findSlaves', but be aware that you will --- need to call 'redirectLogHere' to redirect their logs to the master node. +-- need to call 'redirectLogHere' to redirect their logs to the master node. -- -- Note that you can use functionality of "SimpleLocalnet" directly (through -- 'Backend'), instead of using 'startMaster'/'startSlave', if the master/slave @@ -348,8 +394,20 @@ node <- newLocalNode backend Node.runProcess node $ do slaves <- findSlaves backend - redirectLogsHere backend - proc slaves + redirectLogsHere backend slaves + proc (map processNodeId slaves) `finally` shutdownLogger + +-- +-- | shut down the logger process. This ensures that any pending +-- messages are flushed before the process exits. +-- +shutdownLogger :: Process () +shutdownLogger = do + (sport,rport) <- newChan + nsend "logger" (sport :: SendPort ()) + receiveChan rport + -- TODO: we should monitor the logger process so we don't deadlock if + -- it has already died. -------------------------------------------------------------------------------- -- Accessors -- diff --git a/src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs b/src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs --- a/src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs +++ b/src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs @@ -8,7 +8,7 @@ import Data.Binary (Binary, decode, encode) import Data.IORef (IORef, newIORef, readIORef, modifyIORef) import qualified Data.ByteString as BSS (ByteString, concat) -import qualified Data.ByteString.Lazy as BSL +import qualified Data.ByteString.Lazy as BSL ( ByteString , empty , append @@ -38,10 +38,10 @@ -- -- NOTE: By rights the two functions should be "locally" polymorphic in 'a', -- but this requires impredicative types. -initMulticast :: forall a. Binary a +initMulticast :: forall a. Binary a => HostName -- ^ Multicast IP -> PortNumber -- ^ Port number - -> Int -- ^ Maximum message size + -> Int -- ^ Maximum message size -> IO (IO (a, SockAddr), a -> IO ()) initMulticast host port bufferSize = do (sendSock, sendAddr) <- multicastSender host port @@ -50,8 +50,8 @@ return (recvBinary readSock st bufferSize, writer sendSock sendAddr) where writer :: forall a. Binary a => Socket -> SockAddr -> a -> IO () - writer sock addr val = do - let bytes = encode val + writer sock addr val = do + let bytes = encode val len = encodeInt32 (BSL.length bytes) NBS.sendManyTo sock (len : BSL.toChunks bytes) addr @@ -59,7 +59,7 @@ -- UDP multicast read, dealing with multiple senders -- -------------------------------------------------------------------------------- -type UDPState = Map SockAddr BSL.ByteString +type UDPState = Map SockAddr BSL.ByteString #if MIN_VERSION_network(2,4,0) -- network-2.4.0 provides the Ord instance for us @@ -72,16 +72,16 @@ bufferFor = DAC.mapDefault BSL.empty bufferAppend :: SockAddr -> BSS.ByteString -> UDPState -> UDPState -bufferAppend addr bytes = - bufferFor addr ^: flip BSL.append (BSL.fromChunks [bytes]) +bufferAppend addr bytes = + bufferFor addr ^: flip BSL.append (BSL.fromChunks [bytes]) recvBinary :: Binary a => Socket -> IORef UDPState -> Int -> IO (a, SockAddr) recvBinary sock st bufferSize = do (bytes, addr) <- recvWithLength sock st bufferSize return (decode bytes, addr) -recvWithLength :: Socket - -> IORef UDPState +recvWithLength :: Socket + -> IORef UDPState -> Int -> IO (BSL.ByteString, SockAddr) recvWithLength sock st bufferSize = do @@ -93,14 +93,14 @@ -- Receive all bytes currently in the buffer recvAll :: Socket -> IORef UDPState -> Int -> IO SockAddr recvAll sock st bufferSize = do - (bytes, addr) <- NBS.recvFrom sock bufferSize + (bytes, addr) <- NBS.recvFrom sock bufferSize modifyIORef st $ bufferAppend addr bytes return addr - -recvExact :: Socket - -> Int - -> IORef UDPState + +recvExact :: Socket -> Int + -> IORef UDPState + -> Int -> IO (BSL.ByteString, SockAddr) recvExact sock n st bufferSize = do addr <- recvAll sock st bufferSize @@ -113,16 +113,16 @@ -> IORef UDPState -> Int -> IO BSL.ByteString -recvExactFrom addr sock n st bufferSize = go +recvExactFrom addr sock n st bufferSize = go where go :: IO BSL.ByteString go = do - accAddr <- (^. bufferFor addr) <$> readIORef st - if BSL.length accAddr >= fromIntegral n + accAddr <- (^. bufferFor addr) <$> readIORef st + if BSL.length accAddr >= fromIntegral n then do let (bytes, accAddr') = BSL.splitAt (fromIntegral n) accAddr modifyIORef st $ bufferFor addr ^= accAddr' return bytes - else do + else do _ <- recvAll sock st bufferSize go diff --git a/tests/TestSimpleLocalnet.hs b/tests/TestSimpleLocalnet.hs --- a/tests/TestSimpleLocalnet.hs +++ b/tests/TestSimpleLocalnet.hs @@ -15,10 +15,10 @@ case args of ["master", host, port] -> do - backend <- initializeBackend host port initRemoteTable + backend <- initializeBackend host port initRemoteTable startMaster backend (master backend) ["slave", host, port] -> do - backend <- initializeBackend host port initRemoteTable + backend <- initializeBackend host port initRemoteTable startSlave backend - _ -> + _ -> putStrLn $ "usage: " ++ prog ++ " (master | slave) host port"