distributed-process-p2p 0.1.0.1 → 0.1.1.0

raw patch · 3 files changed

+165/−120 lines, 3 filesdep ~containersdep ~distributed-processPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: containers, distributed-process

API changes (from Hackage documentation)

Files

• distributed-process-p2p.cabal +3/−3
• src/Control/Distributed/Backend/P2P.hs +124/−115
• tests/JollyCloud.hs +38/−2

@@ -1,5 +1,5 @@ name:                distributed-process-p2p-version:             0.1.0.1+version:             0.1.1.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/@@ -24,10 +24,10 @@     base ==4.*,     mtl ==2.1.*,     bytestring ==0.9.*,-    containers ==0.4.*,+    containers >=0.4 && < 0.6,     binary ==0.5.*, -    distributed-process ==0.4.*,+    distributed-process >=0.4.1 && <0.5,     network-transport ==0.3.*,     network-transport-tcp ==0.3.* 

@@ -1,4 +1,4 @@-{-# LANGUAGE OverloadedStrings, DeriveDataTypeable #-}+{-# LANGUAGE OverloadedStrings, RecordWildCards, 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@@ -7,7 +7,7 @@ -- > 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")@@ -16,15 +16,17 @@     bootstrap,     makeNodeId,     getPeers,-    nsendPeers+    getCapable,+    nsendPeers,+    nsendCapable ) 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.Distributed.Process                as DP+import Control.Distributed.Process.Node           as DPN+import 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@@ -40,140 +42,147 @@  -- * Peer-to-peer API +peerControllerService = "P2P:Controller"++type Peers = S.Set ProcessId++data PeerState = PeerState { p2pPeers :: MVar Peers }++initPeerState :: Process PeerState+initPeerState = do+    self <- getSelfPid+    peers <- liftIO $ newMVar (S.singleton self)+    register peerControllerService self+    return $! PeerState peers+ -- ** Initialization  -- | Make a NodeId from "host:port" string.-makeNodeId :: String -> DPT.NodeId-makeNodeId addr = DPT.NodeId . EndPointAddress . BS.concat $ [BS.pack addr, ":0"]+makeNodeId :: String -> NodeId+makeNodeId addr = 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 ()+-- | Start a controller service process and aquire connections to a swarm.+bootstrap :: String -> String -> [NodeId] -> 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"+    node <- newLocalNode transport initRemoteTable -        forever $ receiveWait [ match $ onPeerMsg peerSet-                              , match $ onMonitor peerSet-                              , match $ onQuery peerSet-                              , matchIf isPeerDiscover $ onDiscover pid+    pcPid <- forkProcess node $ do+        state <- initPeerState+        mapM_ doDiscover seeds+        say "P2P controller started."+        forever $ receiveWait [ matchIf isPeerDiscover $ onDiscover state+                              , match $ onMonitor state+                              , match $ onPeerRequest state+                              , match $ onPeerQuery state+                              , match $ onPeerCapable                               ] -    DPN.runProcess node proc+    runProcess node proc  -- ** Discovery --- | Request and response to query peer controller for remote nodes.-data QueryMessage = QueryMessage (DPT.SendPort QueryMessage)-                  | QueryResult [DPT.NodeId]-                  deriving (Eq, Show, Typeable)+doDiscover :: NodeId -> Process ()+doDiscover node = do+    say $ "Examining node: " ++ show node+    whereisRemoteAsync node peerControllerService -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+doRegister :: PeerState -> ProcessId -> Process ()+doRegister state@PeerState{..} pid = do+    pids <- liftIO $ takeMVar p2pPeers+    if S.member pid pids+        then liftIO $ putMVar p2pPeers pids+        else do+            say $ "Registering peer:" ++ show pid+            monitor pid -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+            liftIO $ putMVar p2pPeers (S.insert pid pids)+            say $ "New node: " ++ show pid+            doDiscover $ processNodeId pid --- | 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+doUnregister :: PeerState -> Maybe MonitorRef -> ProcessId -> Process ()+doUnregister PeerState{..} mref pid = do+    say $ "Unregistering peer: " ++ show pid+    maybe (return ()) unmonitor mref+    peers <- liftIO $ takeMVar p2pPeers+    liftIO $ putMVar p2pPeers (S.delete pid peers) --- ** Messaging+isPeerDiscover :: WhereIsReply -> Bool+isPeerDiscover (WhereIsReply service pid) =+    service == peerControllerService && isJust pid --- | 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)+onDiscover :: PeerState -> WhereIsReply -> Process ()+onDiscover state (WhereIsReply _ (Just seedPid)) = do+    say $ "Peer discovered: " ++ show seedPid --- * Peer protocol+    (sp, rp) <- newChan+    self <- getSelfPid+    send seedPid (self, sp :: SendPort Peers)+    say $ "Waiting for peers..."+    peers <- receiveChan rp --- | A set of p2p messages.-data PeerMessage = PeerPing-                 | PeerExchange [DPT.ProcessId]-                 | PeerJoined DPT.ProcessId-                 | PeerLeft DPT.ProcessId-                 deriving (Eq, Show, Typeable)+    known <- liftIO $ readMVar (p2pPeers state)+    mapM_ (doRegister state) (S.toList $ S.difference known peers) -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+onPeerRequest :: PeerState -> (ProcessId, SendPort Peers) -> Process ()+onPeerRequest PeerState{..} (peer, replyTo) = do+    say $ "Peer exchange with " ++ show peer+    peers <- liftIO $ takeMVar p2pPeers+    if S.member peer peers+        then liftIO $ putMVar p2pPeers peers+        else do+            monitor peer+            liftIO $ putMVar p2pPeers (S.insert peer peers) -onPeerMsg :: MVar (S.Set DPT.ProcessId) -> PeerMessage -> Process ()+    sendChan replyTo peers -onPeerMsg _ PeerPing = say "Peer ping" >> return ()+onPeerQuery :: PeerState -> SendPort Peers -> Process ()+onPeerQuery PeerState{..} replyTo = do+    say $ "Local peer query."+    liftIO (readMVar p2pPeers) >>= sendChan replyTo -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+onPeerCapable :: (String, SendPort ProcessId) -> Process ()+onPeerCapable (service, replyTo) = do+    say $ "Capability request: " ++ service+    res <- whereis service+    case res of+        Nothing -> say "I can't."+        Just pid -> say "I can!" >> sendChan replyTo pid -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)+onMonitor :: PeerState -> ProcessMonitorNotification -> Process ()+onMonitor state (ProcessMonitorNotification mref pid reason) = do+    say $ "Monitor event: " ++ show (pid, reason)+    doUnregister state (Just mref) pid -    send pid $ PeerExchange mine-    monitor pid+-- ** Discovery -    if seen-        then return ()-        else do-            myPid <- getSelfPid-            forM_ mine $ \peer -> when (peer /= myPid) $ do-                send peer $ PeerJoined pid+-- | Get a list of currently available peer nodes.+getPeers :: Process [NodeId]+getPeers = do+    say $ "Requesting peer list from local controller..."+    (sp, rp) <- newChan+    nsend peerControllerService (sp :: SendPort Peers)+    receiveChan rp >>= return . map processNodeId . S.toList -onPeerMsg peers (PeerLeft pid) = do-    say $ "Peer left: " ++ show pid-    liftIO $ do-        current <- takeMVar peers-        putMVar peers $ S.delete pid current+-- | Poll a network for a list of specific service providers.+getCapable :: String -> Process [ProcessId]+getCapable service = do+    (sp, rp) <- newChan+    nsendPeers peerControllerService (service, sp)+    say "Waiting for capable nodes..."+    go rp [] -isPeerDiscover :: WhereIsReply -> Bool-isPeerDiscover (WhereIsReply service pid) = service == "peerController" && isJust pid+    where go rp acc = do res <- receiveChanTimeout 100000 rp+                         case res of Just pid -> say "cap hit" >> go rp (pid:acc)+                                     Nothing -> say "cap done" >> return acc -onDiscover :: DPT.ProcessId -> WhereIsReply -> Process ()-onDiscover myPid (WhereIsReply _ (Just seed)) = do-    say $ "Seed discovered: " ++ show seed-    send seed $ PeerJoined myPid+-- ** Messaging -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)+-- | 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)++-- | Broadcast a message to a service of on nodes currently running it.+nsendCapable :: Serializable a => String -> a -> Process ()+nsendCapable service msg = getCapable service >>= mapM_ (\pid -> send pid msg) 

@@ -23,8 +23,14 @@     spawnLocal logger      forever $ do-        liftIO $ threadDelay (10 * 1000000)-        P2P.getPeers >>= liftIO . print+        cmd <- liftIO getLine+        case words cmd of+            ["all"] -> listPeers+            ["in", r] -> listRoom r+            ["join", r] -> joinRoom r+            ["part", r] -> partRoom r+            "tell":r:msg -> tellRoom r (unwords msg)+            _ -> liftIO . putStrLn $ "all | in <r> | join <r> | part <r> | tell <r> <msg>"  logger :: Process () logger = do@@ -34,3 +40,33 @@         (time, pid, msg) <- expect :: Process (String, ProcessId, String)         liftIO $ putStrLn $ time ++ " " ++ show pid ++ " " ++ msg         return ()++listPeers = P2P.getPeers >>= (liftIO . print)++listRoom r = P2P.getCapable r >>= (liftIO . print)++joinRoom r = do+    pid <- whereis r+    case pid of+        Nothing -> spawnLocal (roomService r) >>= register r+        Just _ -> return ()++partRoom r = do+    pid <- whereis r+    case pid of+        Nothing -> return ()+        Just p -> send p (Nothing :: Maybe String)++tellRoom r msg = P2P.nsendCapable r (Just msg)++roomService :: String -> Process ()+roomService s = do+    msg <- expect :: Process (Maybe String)+    case msg of+        Nothing -> do+            liftIO . putStrLn $ "Leaving: " ++ s+            unregister s+        Just m -> do+            liftIO . putStrLn $ "<" ++ s ++ "> " ++ m+            roomService s+