distributed-process-p2p 0.1.4.0 → 0.1.5.0

raw patch · 3 files changed

+28/−13 lines, 3 files

Files

• CHANGELOG +5/−0
• distributed-process-p2p.cabal +3/−4
• src/Control/Distributed/Backend/P2P.hs +20/−9

@@ -1,3 +1,8 @@+0.1.5.0:++    * Use async messages instead of channels for discovery.+      Should prevent deadlocks in peer exchange.+ 0.1.4.0:      * Limit network-transport-tcp to >=0.6

@@ -1,11 +1,11 @@--- This file has been generated from package.yaml by hpack version 0.20.0.+-- This file has been generated from package.yaml by hpack version 0.28.2. -- -- see: https://github.com/sol/hpack ----- hash: a01a000a7a5b07e8ffbf1a52a25f9e247795e9751b5c8894682ed84f6eccd127+-- hash: a8fa97b703de94703e2085a4835441a1b3f28ab0a9b2bf8caeb881d7ad0c77a5  name:                distributed-process-p2p-version:             0.1.4.0+version:             0.1.5.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/@@ -17,7 +17,6 @@ category:            Network build-type:          Simple cabal-version:       >= 1.10- extra-source-files:   CHANGELOG 

@@ -42,8 +42,11 @@  import qualified Data.ByteString.Char8 as BS import qualified Data.Set as S+import Data.Binary (Binary(..)) import Data.Maybe (isJust)+import Data.Typeable (Typeable) + -- * Peer-to-peer API  type Peers = S.Set ProcessId@@ -126,6 +129,7 @@     forever $ receiveWait [ matchIf isPeerDiscover $ onDiscover state                           , match $ onMonitor state                           , match $ onPeerRequest state+                          , match $ onPeerResponse state                           , match $ onPeerQuery state                           , match $ onPeerCapable                           ]@@ -161,22 +165,28 @@ isPeerDiscover (WhereIsReply service pid) =     service == peerControllerService && isJust pid +data GiveMePeers = GiveMePeers+  deriving (Typeable)++instance Binary GiveMePeers where+  put GiveMePeers = pure ()+  get = pure GiveMePeers+ onDiscover :: PeerState -> WhereIsReply -> Process () onDiscover _     (WhereIsReply _ Nothing) = return () onDiscover state (WhereIsReply _ (Just seedPid)) = do     say $ "Peer discovered: " ++ show seedPid--    (sp, rp) <- newChan     self <- getSelfPid-    send seedPid (self, sp :: SendPort Peers)-    say $ "Waiting for peers..."-    peers <- receiveChan rp+    send seedPid (self, GiveMePeers) -    known <- liftIO $ readMVar (p2pPeers state)+onPeerResponse :: PeerState -> (ProcessId, Peers) -> Process ()+onPeerResponse state (peer, peers) = do+    say $ "Got peers from: " ++ show peer+    known <- liftIO $ readMVar $ p2pPeers state     mapM_ (doRegister state) (S.toList $ S.difference peers known) -onPeerRequest :: PeerState -> (ProcessId, SendPort Peers) -> Process ()-onPeerRequest PeerState{..} (peer, replyTo) = do+onPeerRequest :: PeerState -> (ProcessId, GiveMePeers) -> Process ()+onPeerRequest PeerState{..} (peer, _) = do     say $ "Peer exchange with " ++ show peer     peers <- liftIO $ takeMVar p2pPeers     if S.member peer peers@@ -185,7 +195,8 @@             _ <- monitor peer             liftIO $ putMVar p2pPeers (S.insert peer peers) -    sendChan replyTo peers+    self <- getSelfPid+    send peer (self, peers)  onPeerQuery :: PeerState -> SendPort Peers -> Process () onPeerQuery PeerState{..} replyTo = do