packages feed

distributed-process-p2p (empty) → 0.1.0.0

raw patch · 5 files changed

+274/−0 lines, 5 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring, containers, distributed-process, distributed-process-p2p, mtl, network-transport, network-transport-tcp

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2012, Alexander Bondarenko++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 Alexander Bondarenko 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-p2p.cabal view
@@ -0,0 +1,41 @@+name:                distributed-process-p2p+version:             0.1.0.0+synopsis:            Peer-to-peer node discovery for Cloud Haskell+description:         Bootstraps a peer-to-peer connection network from a set of known hosts.+homepage:            https://bitbucket.org/dpwiz/distributed-process-p2p/+license:             BSD3+license-file:        LICENSE+author:              Alexander Bondarenko+maintainer:          aenor.realm@gmail.com+copyright:           Alexannder Bondarenko+category:            Network+build-type:          Simple+cabal-version:       >=1.8++source-repository head+  type:     mercurial+  location: http://bitbucket.org/dpwiz/distributed-process-p2p++library+  exposed-modules:     Control.Distributed.Backend.P2P+  -- other-modules:       +  hs-source-dirs:      src+  build-depends:+    base ==4.*,+    mtl ==2.1.*,+    bytestring ==0.9.*,+    containers ==0.4.*,+    binary ==0.5.*,++    distributed-process ==0.4.*,+    network-transport ==0.3.*,+    network-transport-tcp ==0.3.*++executable jollycloud+  hs-source-dirs:      tests/+  main-is:             JollyCloud.hs+  build-depends:+    base ==4.*,+    mtl ==2.*,+    distributed-process,+    distributed-process-p2p
+ src/Control/Distributed/Backend/P2P.hs view
@@ -0,0 +1,165 @@+{-# LANGUAGE OverloadedStrings, DeriveDataTypeable #-}++-- | Peer-to-peer node discovery backend for Cloud Haskell based on the TCP+-- transport. Provided with a known node address it discovers and maintains+-- the knowledge of it's peers.+--+-- > import qualified Control.Distributed.Backend.P2P as P2P+-- > import           Control.Monad.Trans (liftIO)+-- > import           Control.Concurrent (threadDelay)+-- > +-- > main = P2P.bootstrap "myhostname" "9001" (P2P.makeNodeId "seedhost:9000") $ do+-- >     liftIO $ threadDelay 1000000 -- give dispatcher a second to discover other nodes+-- >     P2P.nsendPeers "myService" ("some", "message")++module Control.Distributed.Backend.P2P (+    bootstrap,+    makeNodeId,+    getPeers,+    nsendPeers+) where++import           Control.Distributed.Process                as DP+import qualified Control.Distributed.Process.Node           as DPN+import qualified Control.Distributed.Process.Internal.Types as DPT+import           Control.Distributed.Process.Serializable (Serializable)+import           Network.Transport (EndPointAddress(..))+import           Network.Transport.TCP (createTransport, defaultTCPParameters)++import Control.Monad+import Control.Applicative+import Control.Monad.Trans+import Control.Concurrent.MVar++import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy.Char8 as BSL+import qualified Data.Set as S+import Data.Typeable+import Data.Binary++-- | Make a NodeId from "host:port" string.+makeNodeId :: String -> DPT.NodeId+makeNodeId addr = DPT.NodeId . EndPointAddress . BS.concat $ [BS.pack addr, ":0"]++-- | Start a peerController process and aquire connections to a swarm.+bootstrap :: String -> String -> [DPT.NodeId] -> DP.Process () -> IO ()+bootstrap host port seeds proc = do+    transport <- either (error . show) id `fmap` createTransport host port defaultTCPParameters+    node <- DPN.newLocalNode transport DPN.initRemoteTable++    pcPid <- DPN.forkProcess node $ do+        pid <- getSelfPid+        register "peerController" pid+        peerSet <- liftIO $ newMVar (S.singleton pid)++        forM_ seeds $ flip whereisRemoteAsync "peerController"++        forever $ receiveWait [ match $ onPeerMsg peerSet+                              , match $ onMonitor peerSet+                              , match $ onDiscover pid+                              , match $ onQuery peerSet+                              ]++    DPN.runProcess node proc++-- | Request and response to query peer controller for remote nodes.+data QueryMessage = QueryMessage (DPT.SendPort QueryMessage)+                  | QueryResult [DPT.NodeId]+                  deriving (Eq, Show, Typeable)++onQuery :: MVar (S.Set DPT.ProcessId) -> QueryMessage -> Process ()+onQuery peers (QueryMessage reply) = do+    ps <- liftIO $ readMVar peers+    sendChan reply $ QueryResult . map processNodeId . S.toList $ ps++instance Binary QueryMessage where+    put (QueryMessage port) = putWord8 0 >> put port+    put (QueryResult ns)    = putWord8 1 >> put ns+    get = do+        mt <- getWord8+        case mt of+            0 -> get >>= return . QueryMessage+            1 -> get >>= return . QueryResult++-- | Get a list of currently available peer nodes.+getPeers :: Process [DPT.NodeId]+getPeers = do+    (s, r) <- newChan+    nsend "peerController" (QueryMessage s)+    QueryResult nodes <- receiveChan r+    return nodes++-- | Broadcast a message to a specific service on all peers.+nsendPeers :: Serializable a => String -> a -> Process ()+nsendPeers service msg = getPeers >>= mapM_ (\peer -> nsendRemote peer service msg)++-- | A set of p2p messages.+data PeerMessage = PeerPing+                 | PeerExchange [DPT.ProcessId]+                 | PeerJoined DPT.ProcessId+                 | PeerLeft DPT.ProcessId+                 deriving (Eq, Show, Typeable)++instance Binary PeerMessage where+    put PeerPing = putWord8 0+    put (PeerExchange ps) = putWord8 1 >> put ps+    put (PeerJoined pid)  = putWord8 2 >> put pid+    put (PeerLeft pid)    = putWord8 3 >> put pid+    get = do+        mt <- getWord8+        case mt of+            0 -> return PeerPing+            1 -> PeerExchange <$> get+            2 -> PeerJoined <$> get+            3 -> PeerLeft <$> get++onPeerMsg :: MVar (S.Set DPT.ProcessId) -> PeerMessage -> Process ()++onPeerMsg _ PeerPing = say "Peer ping" >> return ()++onPeerMsg peers (PeerExchange ps) = do+    say $ "Peer exchange: " ++ show ps+    liftIO $ do+        current <- takeMVar peers+        putMVar peers $ S.union current (S.fromList ps)+    mapM_ monitor ps++onPeerMsg peers (PeerJoined pid) = do+    say $ "Peer joined: " ++ show pid+    (seen, mine) <- liftIO $ do+        mine <- takeMVar peers+        seen <- return $! S.member pid mine+        if seen then putMVar peers mine+                else putMVar peers (S.insert pid mine)+        return (seen, S.toList mine)++    send pid $ PeerExchange mine+    monitor pid++    if seen+        then return ()+        else do+            myPid <- getSelfPid+            forM_ mine $ \peer -> when (peer /= myPid) $ do+                send peer $ PeerJoined pid++onPeerMsg peers (PeerLeft pid) = do+    say $ "Peer left: " ++ show pid+    liftIO $ do+        current <- takeMVar peers+        putMVar peers $ S.delete pid current++onDiscover :: DPT.ProcessId -> WhereIsReply -> Process ()+onDiscover myPid (WhereIsReply "peerController" (Just seed)) = do+    say $ "Seed discovered: " ++ show seed+    send seed $ PeerJoined myPid++onMonitor :: MVar (S.Set DPT.ProcessId) -> ProcessMonitorNotification -> Process ()+onMonitor peerSet (ProcessMonitorNotification mref pid reason) = do+    say $ "Monitor event: " ++ show (pid, reason)+    peers <- liftIO $ do+        peers <- takeMVar peerSet+        putMVar peerSet $! S.delete pid peers+        return (S.toList peers)+    forM_ peers $ \peer -> send peer (PeerLeft pid)+
+ tests/JollyCloud.hs view
@@ -0,0 +1,36 @@+module Main where++import qualified Control.Distributed.Backend.P2P as P2P+import           Control.Distributed.Process as DP+import           Control.Distributed.Process.Node as DPN++import System.Environment (getArgs)++import Control.Monad+import Control.Monad.Trans (liftIO)+import Control.Concurrent (threadDelay)++main :: IO ()+main = do+  args <- getArgs++  case args of+    ["-h"] -> putStrLn "Usage: jollycloud addr port [<seed>..]"+    host:port:seeds -> P2P.bootstrap host port (map P2P.makeNodeId seeds) mainProcess++mainProcess :: Process ()+mainProcess = do+    spawnLocal logger++    forever $ do+        liftIO $ threadDelay (10 * 1000000)+        P2P.getPeers >>= liftIO . print++logger :: Process ()+logger = do+    unregister "logger"+    getSelfPid >>= register "logger"+    forever $ do+        (time, pid, msg) <- expect :: Process (String, ProcessId, String)+        liftIO $ putStrLn $ time ++ " " ++ show pid ++ " " ++ msg+        return ()