distributed-process-simplelocalnet 0.2.0.8 → 0.2.0.9

raw patch · 4 files changed

+165/−107 lines, 4 filesdep ~distributed-processnew-uploaderPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: distributed-process

API changes (from Hackage documentation)

Files

• distributed-process-simplelocalnet.cabal +4/−4
• src/Control/Distributed/Process/Backend/SimpleLocalnet.hs +139/−81
• src/Control/Distributed/Process/Backend/SimpleLocalnet/Internal/Multicast.hs +19/−19
• tests/TestSimpleLocalnet.hs +3/−3

@@ -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,

@@ -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                                                                  --

@@ -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

@@ -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"