diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,31 @@ +Copyright Well-Typed LLP, 2011-2012 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of the owner nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/distributed-process-simplelocalnet.cabal b/distributed-process-simplelocalnet.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-simplelocalnet.cabal @@ -0,0 +1,63 @@ +Name: distributed-process-simplelocalnet +Version: 0.2.0 +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, dcoutts@well-typed.com +Stability: experimental +Homepage: http://github.com/haskell-distributed/distributed-process +Bug-Reports: mailto:edsko@well-typed.com +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 + backend designed to get you going with Cloud Haskell quickly + without imposing any structure on your application. +Tested-With: GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 +Category: Control + +Source-Repository head + Type: git + Location: https://github.com/haskell-distributed/distributed-process + SubDir: distributed-process-simplelocalnet + +Library + Build-Depends: base >= 4.4 && < 5, + bytestring >= 0.9 && < 0.10, + network >= 2.3 && < 2.4, + network-multicast >= 0.0 && < 0.1, + data-accessor >= 0.2 && < 0.3, + binary >= 0.5 && < 0.6, + containers >= 0.4 && < 0.5, + transformers >= 0.2 && < 0.4, + network-transport >= 0.2 && < 0.3, + network-transport-tcp >= 0.2 && < 0.3, + distributed-process >= 0.2 && < 0.3 + Exposed-modules: Control.Distributed.Process.Backend.SimpleLocalnet, + Control.Distributed.Process.Backend.SimpleLocalnet.Internal.Multicast + Extensions: RankNTypes, + DeriveDataTypeable + ghc-options: -Wall + HS-Source-Dirs: src + +-- Not a proper test, but we want to use cabal to compile it +Test-Suite TestSimpleLocalnet + Type: exitcode-stdio-1.0 + Main-Is: TestSimpleLocalnet.hs + Build-Depends: base >= 4.4 && < 5, + bytestring >= 0.9 && < 0.10, + network >= 2.3 && < 2.4, + network-multicast >= 0.0 && < 0.1, + data-accessor >= 0.2 && < 0.3, + binary >= 0.5 && < 0.6, + containers >= 0.4 && < 0.5, + transformers >= 0.2 && < 0.4, + network-transport >= 0.2 && < 0.3, + network-transport-tcp >= 0.2 && < 0.3, + distributed-process >= 0.2 && < 0.3 + Extensions: RankNTypes, + DeriveDataTypeable + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind + HS-Source-Dirs: tests src diff --git a/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs b/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs @@ -0,0 +1,291 @@ +-- | Simple backend based on the TCP transport which offers node discovery +-- based on UDP multicast. This is a zero-configuration backend designed to +-- 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 +-- /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.Local +-- > +-- > 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 +-- > startMaster backend (master backend) +-- > ["slave", host, port] -> do +-- > backend <- initializeBackend host port initRemoteTable +-- > startSlave backend +{-# OPTIONS_GHC -fno-warn-orphans #-} +module Control.Distributed.Process.Backend.SimpleLocalnet + ( -- * Initialization + Backend(..) + , initializeBackend + -- * Slave nodes + , startSlave + , terminateSlave + , findSlaves + , terminateAllSlaves + -- * Master nodes + , startMaster + ) where + +import System.IO (fixIO) +import Data.Maybe (catMaybes) +import Data.Binary (Binary(get, put), getWord8, putWord8) +import Data.Accessor (Accessor, accessor, (^:), (^.)) +import Data.Set (Set) +import qualified Data.Set as Set (insert, empty, toList) +import Data.Foldable (forM_) +import Data.Typeable (Typeable) +import Control.Applicative ((<$>)) +import Control.Exception (throw) +import Control.Monad (forever, forM) +import Control.Monad.IO.Class (liftIO) +import Control.Concurrent (forkIO, threadDelay, ThreadId) +import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_) +import Control.Distributed.Process + ( RemoteTable + , NodeId + , Process + , WhereIsReply(..) + , whereis + , whereisRemoteAsync + , registerRemote + , getSelfPid + , register + , expect + , nsendRemote + , receiveWait + , matchIf + , processNodeId + ) +import qualified Control.Distributed.Process.Node as Node + ( LocalNode + , newLocalNode + , localNodeId + , runProcess + ) +import qualified Network.Transport.TCP as NT + ( createTransport + , defaultTCPParameters + ) +import qualified Network.Transport as NT (Transport) +import qualified Network.Socket as N (HostName, ServiceName, SockAddr) +import Control.Distributed.Process.Backend.SimpleLocalnet.Internal.Multicast (initMulticast) + +-- | Local backend +data Backend = Backend { + -- | Create a new local node + newLocalNode :: IO Node.LocalNode + -- | @findPeers t@ sends out a /who's there?/ request, waits 't' msec, + -- and then collects and returns the answers + , findPeers :: Int -> IO [NodeId] + -- | Make sure that all log messages are printed by the logger on the + -- current node + , redirectLogsHere :: Process () + } + +data BackendState = BackendState { + _localNodes :: [Node.LocalNode] + , _peers :: Set NodeId + , discoveryDaemon :: ThreadId + } + +-- | 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 + (_, backendState) <- fixIO $ \ ~(tid, _) -> do + backendState <- newMVar BackendState + { _localNodes = [] + , _peers = Set.empty + , discoveryDaemon = tid + } + tid' <- forkIO $ peerDiscoveryDaemon backendState recv send + return (tid', backendState) + case mTransport of + Left err -> throw err + Right transport -> + let backend = Backend { + newLocalNode = apiNewLocalNode transport rtable backendState + , findPeers = apiFindPeers send backendState + , redirectLogsHere = apiRedirectLogsHere backend + } + in return backend + +-- | Create a new local node +apiNewLocalNode :: NT.Transport + -> RemoteTable + -> MVar BackendState + -> IO Node.LocalNode +apiNewLocalNode transport rtable backendState = do + localNode <- Node.newLocalNode transport rtable + modifyMVar_ backendState $ return . (localNodes ^: (localNode :)) + return localNode + +-- | Peer discovery +apiFindPeers :: (PeerDiscoveryMsg -> IO ()) + -> MVar BackendState + -> Int + -> IO [NodeId] +apiFindPeers send backendState delay = do + send PeerDiscoveryRequest + threadDelay delay + Set.toList . (^. peers) <$> readMVar backendState + +data PeerDiscoveryMsg = + PeerDiscoveryRequest + | PeerDiscoveryReply NodeId + +instance Binary PeerDiscoveryMsg where + put PeerDiscoveryRequest = putWord8 0 + put (PeerDiscoveryReply nid) = putWord8 1 >> put nid + get = do + 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 + -> IO (PeerDiscoveryMsg, N.SockAddr) + -> (PeerDiscoveryMsg -> IO ()) + -> IO () +peerDiscoveryDaemon backendState recv send = forever go + where + go = do + (msg, _) <- recv + case msg of + PeerDiscoveryRequest -> do + nodes <- (^. localNodes) <$> readMVar backendState + forM_ nodes $ send . PeerDiscoveryReply . Node.localNodeId + PeerDiscoveryReply nid -> + modifyMVar_ backendState $ return . (peers ^: Set.insert nid) + +-------------------------------------------------------------------------------- +-- Back-end specific primitives -- +-------------------------------------------------------------------------------- + +-- | Make sure that all log messages are printed by the logger on this node +apiRedirectLogsHere :: Backend -> Process () +apiRedirectLogsHere backend = do + mLogger <- whereis "logger" + forM_ mLogger $ \logger -> do + nids <- liftIO $ findPeers backend 1000000 + forM_ nids $ \nid -> registerRemote nid "logger" logger + +-------------------------------------------------------------------------------- +-- Slaves -- +-------------------------------------------------------------------------------- + +-- | Messages to slave nodes +-- +-- This datatype is not exposed; instead, we expose primitives for dealing +-- with slaves. +data SlaveControllerMsg = + SlaveTerminate + deriving (Typeable, Show) + +instance Binary SlaveControllerMsg where + put SlaveTerminate = putWord8 0 + get = do + header <- getWord8 + case header of + 0 -> return SlaveTerminate + _ -> fail "SlaveControllerMsg.get: invalid" + +-- | Calling 'slave' sets up a new local node and then waits. You start +-- processes on the slave by calling 'spawn' from other nodes. +-- +-- This function does not return. The only way to exit the slave is to CTRL-C +-- the process or call terminateSlave from another node. +startSlave :: Backend -> IO () +startSlave backend = do + node <- newLocalNode backend + Node.runProcess node slaveController + +-- | The slave controller interprets 'SlaveControllerMsg's +slaveController :: Process () +slaveController = do + pid <- getSelfPid + register "slaveController" pid + go + where + go = do + msg <- expect + case msg of + SlaveTerminate -> return () + +-- | 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 = do + nodes <- liftIO $ findPeers backend 1000000 + -- Fire of asynchronous requests for the slave controller + forM_ nodes $ \nid -> whereisRemoteAsync nid "slaveController" + -- Wait for the replies + catMaybes <$> forM nodes (\_ -> + receiveWait + [ matchIf (\(WhereIsReply label _) -> label == "slaveController") + (\(WhereIsReply _ mPid) -> return (processNodeId <$> mPid)) + ]) + +-- | Terminate all slaves +terminateAllSlaves :: Backend -> Process () +terminateAllSlaves backend = do + slaves <- findSlaves backend + forM_ slaves terminateSlave + liftIO $ threadDelay 1000000 + +-------------------------------------------------------------------------------- +-- Master nodes +-------------------------------------------------------------------------------- + +-- | 'startMaster' finds all slaves currently available on the local network +-- (which should therefore be started first), redirects all log messages to +-- itself, and then calls the specified process, 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 +-- 'terminateAllSlaves'. +startMaster :: Backend -> ([NodeId] -> Process ()) -> IO () +startMaster backend proc = do + node <- newLocalNode backend + Node.runProcess node $ do + slaves <- findSlaves backend + redirectLogsHere backend + proc slaves + +-------------------------------------------------------------------------------- +-- Accessors -- +-------------------------------------------------------------------------------- + +localNodes :: Accessor BackendState [Node.LocalNode] +localNodes = accessor _localNodes (\ns st -> st { _localNodes = ns }) + +peers :: Accessor BackendState (Set NodeId) +peers = accessor _peers (\ps st -> st { _peers = ps }) diff --git a/src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs b/src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs @@ -0,0 +1,126 @@ +-- | Multicast utilities +{-# OPTIONS_GHC -fno-warn-orphans #-} +module Control.Distributed.Process.Backend.SimpleLocalnet.Internal.Multicast (initMulticast) where + +import Data.Function (on) +import Data.Map (Map) +import qualified Data.Map as Map (empty) +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 + ( ByteString + , empty + , append + , fromChunks + , toChunks + , length + , splitAt + ) +import Data.Accessor (Accessor, (^:), (^.), (^=)) +import qualified Data.Accessor.Container as DAC (mapDefault) +import Control.Applicative ((<$>)) +import Network.Socket (HostName, PortNumber, Socket, SockAddr) +import qualified Network.Socket.ByteString as NBS (recvFrom, sendManyTo) +import Network.Transport.Internal (decodeInt32, encodeInt32) +import Network.Multicast (multicastSender, multicastReceiver) + +-------------------------------------------------------------------------------- +-- Top-level API -- +-------------------------------------------------------------------------------- + +-- | Given a hostname and a port number, initialize the multicast system. +-- +-- Note: it is important that you never send messages larger than the maximum +-- message size; if you do, all subsequent communication will probably fail. +-- +-- Returns a reader and a writer. +-- +-- NOTE: By rights the two functions should be "locally" polymorphic in 'a', +-- but this requires impredicative types. +initMulticast :: forall a. Binary a + => HostName -- ^ Multicast IP + -> PortNumber -- ^ Port number + -> Int -- ^ Maximum message size + -> IO (IO (a, SockAddr), a -> IO ()) +initMulticast host port bufferSize = do + (sendSock, sendAddr) <- multicastSender host port + readSock <- multicastReceiver host port + st <- newIORef Map.empty + 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 + len = encodeInt32 (BSL.length bytes) + NBS.sendManyTo sock (len : BSL.toChunks bytes) addr + +-------------------------------------------------------------------------------- +-- UDP multicast read, dealing with multiple senders -- +-------------------------------------------------------------------------------- + +type UDPState = Map SockAddr BSL.ByteString + +-- TODO: This is inefficient and an orphan instance. +-- Requested official instance (https://github.com/haskell/network/issues/38) +instance Ord SockAddr where + compare = compare `on` show + +bufferFor :: SockAddr -> Accessor UDPState BSL.ByteString +bufferFor = DAC.mapDefault BSL.empty + +bufferAppend :: SockAddr -> BSS.ByteString -> UDPState -> UDPState +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 + -> Int + -> IO (BSL.ByteString, SockAddr) +recvWithLength sock st bufferSize = do + (len, addr) <- recvExact sock 4 st bufferSize + let n = decodeInt32 . BSS.concat . BSL.toChunks $ len + bytes <- recvExactFrom addr sock n st bufferSize + return (bytes, addr) + +-- 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 + modifyIORef st $ bufferAppend addr bytes + return addr + +recvExact :: Socket + -> Int + -> IORef UDPState + -> Int + -> IO (BSL.ByteString, SockAddr) +recvExact sock n st bufferSize = do + addr <- recvAll sock st bufferSize + bytes <- recvExactFrom addr sock n st bufferSize + return (bytes, addr) + +recvExactFrom :: SockAddr + -> Socket + -> Int + -> IORef UDPState + -> Int + -> IO BSL.ByteString +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 + then do + let (bytes, accAddr') = BSL.splitAt (fromIntegral n) accAddr + modifyIORef st $ bufferFor addr ^= accAddr' + return bytes + else do + _ <- recvAll sock st bufferSize + go diff --git a/tests/TestSimpleLocalnet.hs b/tests/TestSimpleLocalnet.hs new file mode 100644 --- /dev/null +++ b/tests/TestSimpleLocalnet.hs @@ -0,0 +1,24 @@ +import System.Environment (getArgs, getProgName) +import Control.Distributed.Process (NodeId, Process, liftIO) +import Control.Distributed.Process.Node (initRemoteTable) +import Control.Distributed.Process.Backend.SimpleLocalnet + +master :: Backend -> [NodeId] -> Process () +master backend slaves = do + liftIO . putStrLn $ "Slaves: " ++ show slaves + terminateAllSlaves backend + +main :: IO () +main = do + prog <- getProgName + args <- getArgs + + case args of + ["master", host, port] -> do + backend <- initializeBackend host port initRemoteTable + startMaster backend (master backend) + ["slave", host, port] -> do + backend <- initializeBackend host port initRemoteTable + startSlave backend + _ -> + putStrLn $ "usage: " ++ prog ++ " (master | slave) host port"