packages feed

distributed-process-simplelocalnet (empty) → 0.2.0

raw patch · 6 files changed

+537/−0 lines, 6 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring, containers, data-accessor, distributed-process, network, network-multicast, network-transport, network-transport-tcp, transformers

Files

+ LICENSE view
@@ -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.+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ distributed-process-simplelocalnet.cabal view
@@ -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
+ src/Control/Distributed/Process/Backend/SimpleLocalnet.hs view
@@ -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 })
+ src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs view
@@ -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
+ tests/TestSimpleLocalnet.hs view
@@ -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"