kademlia 1.0.0.0 → 1.1.0.0

raw patch · 18 files changed

+1295/−296 lines, 18 filesdep +HUnitdep +QuickCheckdep +tastyPVP ok

version bump matches the API change (PVP)

Dependencies added: HUnit, QuickCheck, tasty, tasty-hunit, tasty-quickcheck

API changes (from Hackage documentation)

Files

• kademlia.cabal +30/−3
• src/Network/Kademlia.hs +24/−17
• src/Network/Kademlia/Implementation.hs +57/−17
• src/Network/Kademlia/Instance.hs +143/−31
• src/Network/Kademlia/Networking.hs +12/−16
• src/Network/Kademlia/Protocol/Parsing.hs +1/−1
• src/Network/Kademlia/ReplyQueue.hs +49/−47
• src/Network/Kademlia/Tree.hs +186/−157
• src/Network/Kademlia/Types.hs +2/−7
• test/Implementation.hs +122/−0
• test/Instance.hs +132/−0
• test/Networking.hs +94/−0
• test/Protocol.hs +35/−0
• test/ReplyQueue.hs +87/−0
• test/Test.hs +112/−0
• test/TestTypes.hs +95/−0
• test/Tree.hs +86/−0
• test/Types.hs +28/−0

@@ -1,5 +1,5 @@ name:                kademlia-version:             1.0.0.0+version:             1.1.0.0 homepage:            https://github.com/froozen/kademlia bug-reports:         https://github.com/froozen/kademlia/issues synopsis:            An implementation of the Kademlia DHT Protocol@@ -39,10 +39,10 @@                        Network.Kademlia.ReplyQueue,                        Network.Kademlia.Implementation -  build-depends:       base >= 4 && < 5,+  build-depends:       base >= 4.7 && < 5,                        network >=2.6 && <2.7,                        mtl >=2.1.3.1,-                       bytestring >=0.10 && <0.11,+                       bytestring >=0.10.2 && <0.11,                        transformers >=0.3,                        containers >=0.5.5.1,                        stm >=2.4.3,@@ -50,3 +50,30 @@    hs-source-dirs:      src   default-language:    Haskell2010++test-suite library-test+  type:                exitcode-stdio-1.0+  main-is:             Test.hs+  hs-source-dirs:      test, src+  other-modules:       Protocol, Networking, TestTypes, Types, Tree, Instance,+                       ReplyQueue, Implementation,+                       Network.Kademlia.Networking, Network.Kademlia.Types,+                       Network.Kademlia.Protocol, Network.Kademlia.Instance,+                       Network.Kademlia.Protocol.Parsing, Network.Kademlia.Tree,+                       Network.Kademlia.ReplyQueue,+                       Network.Kademlia.Implementation++  default-language:    Haskell2010+  build-depends:       base >= 4.7 && < 5,+                       network >=2.6 && <2.7,+                       mtl >=2.1.3.1,+                       bytestring >=0.10.2 && <0.11,+                       transformers >=0.3,+                       containers >=0.5.5.1,+                       stm >=2.4.3,+                       transformers-compat >=0.3.3,+                       tasty >= 0.10.1,+                       tasty-quickcheck >= 0.8.3.1,+                       QuickCheck >= 2.4,+                       tasty-hunit >= 0.9.0.1,+                       HUnit >= 1.2.5.2

@@ -71,18 +71,26 @@ >    -- network >    firstInstance <- K.create 12345 . KademliaID . C.pack $ "hello" >+>    -- Create a Node representing the first instance+>    let firstNode = Node (Peer "localhost" 12345) . KademliaID . C.pack $ "hello"+> >    -- Create the second instance and make it join the network >    secondInstance <- K.create 12346 . KademliaID . C.pack $ "uAleu"->    K.joinNetwork secondInstance ("localhost", 12345, "hello")+>    joinResult <- K.joinNetwork secondInstance firstNode >->    -- Store an example value in the network->    let exampleValue = Person 25 "Alan Turing"->    K.store secondInstance (KademliaID . C.pack $ "raxqT") exampleValue+>    -- Make sure the joining was successful+>    case joinResult of+>         JoinSuccess -> do+>             -- Store an example value in the network+>             let exampleValue = Person 25 "Alan Turing"+>             K.store secondInstance (KademliaID . C.pack $ "raxqT") exampleValue >->    -- Look up the value->    result <- K.lookup firstInstance . KademliaID . C.pack $ "raxqT"->    print result+>             -- Look up the value and it's source+>             (value, source) <- K.lookup firstInstance . KademliaID . C.pack $ "raxqT"+>             print value >+>         _ -> return ()+> >    -- Close the instances >    K.close firstInstance >    K.close secondInstance@@ -109,8 +117,13 @@     , close     , I.lookup     , I.store-    , Network.Kademlia.joinNetwork+    , I.lookupNode+    , I.joinNetwork+    , dumpPeers+    , JoinResult(..)     , Serialize(..)+    , Node(..)+    , Peer(..)     ) where  import Network.Kademlia.Networking@@ -128,17 +141,11 @@ create :: (Serialize i, Ord i, Serialize a, Eq a, Eq i) =>     Int -> i -> IO (KademliaInstance i a) create port id = do-    h <- openOn (show port) id+    rq <- emptyReplyQueue+    h <- openOn (show port) id rq     inst <- newInstance id h-    start inst+    start inst rq     return inst---- | Make a KademliaInstance join the network the supplied Node is a part of-joinNetwork :: (Serialize i, Ord i, Eq i, Serialize a) => KademliaInstance i a-     -> (String, Int, i) -> IO ()-joinNetwork inst (host, port, i) = let peer = Peer host . fromIntegral $ port-                                       node = Node peer i-                                   in  I.joinNetwork inst node  -- | Stop a KademliaInstance by closing it close :: KademliaInstance i a -> IO ()

@@ -10,6 +10,8 @@     ( lookup     , store     , joinNetwork+    , JoinResult(..)+    , Network.Kademlia.Implementation.lookupNode     ) where  import Network.Kademlia.Networking@@ -27,9 +29,10 @@ import Data.Maybe (isJust, fromJust)  --- Lookup the value corresponding to a key in the DHT+-- | Lookup the value corresponding to a key in the DHT and return it, together+--   with the Node that was the first to answer the lookup lookup :: (Serialize i, Serialize a, Eq i, Ord i) => KademliaInstance i a -> i-       -> IO (Maybe a)+       -> IO (Maybe (a, Node i)) lookup inst id = runLookup go inst id     where go = startLookup sendS cancel checkSignal @@ -57,7 +60,7 @@                     liftIO . send (handle inst) cachePeer . STORE id $ value                  -- Return the value-                return . Just $ value+                return . Just $ (value, origin)            -- When receiving a RETURN_NODES command, throw the nodes into the           -- lookup loop and continue the lookup@@ -82,7 +85,7 @@                 finish           finishCheck _ = finish --- Store assign a value to a key and store it in the DHT+-- | Store assign a value to a key and store it in the DHT store :: (Serialize i, Serialize a, Eq i, Ord i) =>          KademliaInstance i a -> i -> a -> IO () store inst key val = runLookup go inst key@@ -106,38 +109,76 @@              unless (null polled) $ do                 let h = handle inst-                    -- Select the peer closest to the key-                    storePeer = peer . head . sortByDistanceTo polled $ key-                -- Send it a STORE command-                liftIO . send h storePeer . STORE key $ val+                    -- Don't select more than 7 peers+                    peerNum = if length polled > 7 then 7 else length polled+                    -- Select the peers closest to the key+                    storePeers =+                        map peer . take peerNum . sortByDistanceTo polled $ key +                -- Send them a STORE command+                forM_ storePeers $+                    \storePeer -> liftIO . send h storePeer . STORE key $ val++-- | The different possibel results of joinNetwork+data JoinResult = JoinSucces | NodeDown | IDClash deriving (Eq, Ord, Show)+ -- | Make a KademliaInstance join the network a supplied Node is in joinNetwork :: (Serialize i, Serialize a, Eq i, Ord i) => KademliaInstance i a-            -> Node i -> IO ()+            -> Node i -> IO JoinResult joinNetwork inst node = ownId >>= runLookup go inst     where go = do             -- Poll the supplied node             sendS node             -- Run a normal lookup from thereon out-            waitForReply cancel checkSignal+            waitForReply nodeDown checkSignal -          -- Do nothing upon failure or when the join operation has terminated-          cancel = return ()+          -- No answer to the first signal means, that that Node is down+          nodeDown = return NodeDown            -- Retrieve your own id           ownId =             fmap T.extractId . atomically . readTVar .  sTree . state $ inst -          -- Always add the nodes into the loop and continue the lookup-          checkSignal (Signal _ (RETURN_NODES _ nodes)) =-            continueLookup nodes sendS continue cancel+          -- Check wether the own id was encountered. If so, return a IDClash+          -- error, otherwise, continue the lookup.+          checkSignal (Signal _ (RETURN_NODES _ nodes)) = do+                tId <- gets targetId+                case find (\node -> nodeId node == tId) nodes of+                    Just _ -> return IDClash+                    _ -> continueLookup nodes sendS continue finish            -- Continuing always means waiting for the next signal-          continue = waitForReply cancel checkSignal+          continue = waitForReply finish checkSignal            -- Send a FIND_NODE command, looking up your own id           sendS node = liftIO ownId >>= flip sendSignal node . FIND_NODE +          -- Return a success, when the operation finished cleanly+          finish = return JoinSucces++-- | Lookup the Node corresponding to the supplied ID+lookupNode :: (Serialize i, Serialize a, Eq i, Ord i) => KademliaInstance i a -> i+           -> IO (Maybe (Node i))+lookupNode inst id = runLookup go inst id+    where go = startLookup sendS end checkSignal++          -- Return Nothing on lookup failure+          end = return Nothing++          -- Check wether the Node we are looking for was found. If so, return+          -- it, otherwise continue the lookup.+          checkSignal (Signal _ (RETURN_NODES _ nodes)) =+                case find (\(Node _ nId) -> nId == id) nodes of+                    Just node -> return . Just $ node+                    _ -> continueLookup nodes sendS continue end++          -- Continuing always means waiting for the next signal+          continue = waitForReply end checkSignal++          -- Send a FIND_NODE command looking for the Node corresponding to the+          -- id+          sendS = sendSignal (FIND_NODE id)+ -- | The state of a lookup data LookupState i a = LookupState {       inst :: KademliaInstance i a@@ -217,7 +258,6 @@             let node = fromJust . find (\n -> nodeId n == id) $ polled              -- Remove every trace of the node's existance-            liftIO . deleteNode inst $ id             modify $ \s -> s {                   pending = delete node sPending                 , known = delete node known

@@ -13,8 +13,8 @@     , start     , newInstance     , insertNode-    , deleteNode     , lookupNode+    , dumpPeers     ) where  import Control.Concurrent@@ -22,13 +22,15 @@ import Control.Concurrent.STM import Control.Monad (void, forever, when, join, forM_, forever) import Control.Monad.Trans-import Control.Monad.Trans.State+import Control.Monad.Trans.State hiding (state) import Control.Monad.Trans.Reader import Control.Monad.IO.Class (liftIO)+import Control.Applicative ((<$>), (<*>)) import System.IO.Error (catchIOError) import qualified Data.Map as M-import Data.Maybe (catMaybes)+import Data.Maybe (catMaybes, isJust, fromJust) import Data.Function (on)+import Data.Map (toList)  import Network.Kademlia.Networking import qualified Network.Kademlia.Tree as T@@ -39,6 +41,7 @@ data KademliaInstance i a = KI {       handle :: KademliaHandle i a     , state :: KademliaState i a+    , expirationThreads :: TVar (M.Map i ThreadId)     }  -- | Representation of the data the KademliaProcess carries@@ -53,43 +56,130 @@ newInstance id handle = do     tree <- atomically . newTVar . T.create $ id     values <- atomically . newTVar $ M.empty-    return . KI handle . KS tree $ values+    threads <- atomically . newTVar $ M.empty+    return . KI handle (KS tree values) $ threads +-- | Insert a Node into the NodeTree insertNode :: (Serialize i, Ord i) => KademliaInstance i a -> Node i -> IO ()-insertNode (KI _ (KS sTree _)) node = atomically $ do+insertNode (KI _ (KS sTree _) _) node = atomically $ do     tree <- readTVar sTree     writeTVar sTree . T.insert tree $ node -deleteNode :: (Serialize i, Ord i) => KademliaInstance i a -> i -> IO ()-deleteNode (KI _ (KS sTree _)) id = atomically $ do+-- | Signal a Node's timeout and retur wether it should be repinged+timeoutNode :: (Serialize i, Ord i) => KademliaInstance i a -> i -> IO Bool+timeoutNode (KI _ (KS sTree _) _) id = atomically $ do     tree <- readTVar sTree-    writeTVar sTree . T.delete tree $ id+    let (newTree, pingAgain) = T.handleTimeout tree id+    writeTVar sTree newTree+    return pingAgain +-- | Lookup a Node in the NodeTree lookupNode :: (Serialize i, Ord i) => KademliaInstance i a -> i -> IO (Maybe (Node i))-lookupNode (KI _ (KS sTree _)) id = atomically $ do+lookupNode (KI _ (KS sTree _) _) id = atomically $ do     tree <- readTVar sTree     return . T.lookup tree $ id +-- | Return all the Nodes an Instance has encountered so far+dumpPeers :: KademliaInstance i a -> IO [Node i]+dumpPeers (KI _ (KS sTree _) _) = atomically $ do+    tree <- readTVar sTree+    return . T.toList $ tree++-- | Insert a value into the store insertValue :: (Ord i) => i -> a -> KademliaInstance i a -> IO ()-insertValue key value (KI _ (KS _ values)) = atomically $ do+insertValue key value (KI _ (KS _ values) _) = atomically $ do     vals <- readTVar values     writeTVar values $ M.insert key value vals +-- | Delete a value from the store+deleteValue :: (Ord i) => i -> KademliaInstance i a -> IO ()+deleteValue key (KI _ (KS _ values) _) = atomically $ do+    vals <- readTVar values+    writeTVar values $ M.delete key vals++-- | Lookup a value in the store lookupValue :: (Ord i) => i -> KademliaInstance i a -> IO (Maybe a)-lookupValue key (KI _ (KS _ values)) = atomically $ do+lookupValue key (KI _ (KS _ values) _) = atomically $ do     vals <- readTVar values     return . M.lookup key $ vals  -- | Start the background process for a KademliaInstance start :: (Serialize i, Ord i, Serialize a, Eq i, Eq a) =>-    KademliaInstance i a -> IO ()-start inst = do-        chan <- newChan-        startRecvProcess (handle inst) chan-        pingId <- forkIO . pingProcess inst $ chan+         KademliaInstance i a -> ReplyQueue i a -> IO ()+start inst rq = do+        startRecvProcess . handle $ inst+        let rChan = timeoutChan rq+            dChan = defaultChan rq+        receivingId <- forkIO . receivingProcess inst rq rChan $ dChan+        pingId <- forkIO . pingProcess inst $ dChan         spreadId <- forkIO . spreadValueProcess $ inst-        void . forkIO $ backgroundProcess inst chan [pingId, spreadId]+        void . forkIO $ backgroundProcess inst dChan [pingId, spreadId, receivingId] +-- | The central process all Replys go trough+receivingProcess :: (Serialize i, Serialize a, Eq i, Ord i) =>+       KademliaInstance i a -> ReplyQueue i a -> Chan (Reply i a)+    -> Chan (Reply i a)-> IO ()+receivingProcess inst rq replyChan registerChan = forever $ do+    reply <- readChan replyChan++    case reply of+        -- Handle a timed out node+        Timeout registration -> do+            let origin = replyOrigin registration+                h = handle inst+                newRegistration = registration { replyTypes = [R_PONG] }++            -- Mark the node as timed out+            pingAgain <- timeoutNode inst origin+            -- If the node should be repinged+            when pingAgain $ do+                result <- lookupNode inst origin+                case result of+                    Nothing -> return ()+                    Just node -> do+                        -- Ping the node+                        send h (peer node) PING+                        expect h newRegistration registerChan++        -- Store values in newly encountered nodes that you are the closest to+        Answer (Signal node cmd) -> do+            let originId = nodeId node+            tree <- retrieve sTree++            -- This node is not yet known+            when (not . isJust . T.lookup tree $ originId) $ do+                let closestKnown = T.findClosest tree originId 1+                    ownId = T.extractId tree+                    self = node { nodeId = ownId }+                    bucket = self:closestKnown+                    -- Find out closest known node+                    closestId = nodeId . head . sortByDistanceTo bucket $ originId++                -- This node can be assumed to be closest to the new node+                when (ownId == closestId) $ do+                    storedValues <- toList <$> retrieve values+                    let h = handle inst+                        p = peer node+                    -- Store all stored values in the new node+                    forM_ storedValues (send h p . uncurry STORE)++            case cmd of+                -- Ping unknown Nodes that were returned by RETURN_NODES.+                -- Pinging them first is neccessary to prevent disconnected+                -- nodes from spreading through the networks NodeTrees.+                (RETURN_NODES _ nodes) -> forM_ nodes $ \node -> do+                    result <- lookupNode inst . nodeId $ node+                    case result of+                        Nothing -> send (handle inst) (peer node) PING+                        _ -> return ()+                _ -> return ()++        _ -> return ()++    dispatch reply rq+    where retrieve f = atomically . readTVar . f . state $ inst++ -- | The actual process running in the background backgroundProcess :: (Serialize i, Ord i, Serialize a, Eq i, Eq a) =>     KademliaInstance i a -> Chan (Reply i a) -> [ThreadId] -> IO ()@@ -109,39 +199,41 @@              backgroundProcess inst chan threadIds -        -- Delete timed out nodes-        Timeout registration -> deleteNode inst . replyOrigin $ registration+        -- Kill all other processes and stop on Closed+        Closed -> do+            mapM_ killThread threadIds -        -- Kill pingProcess and stop on Closed-        Closed -> mapM_ killThread threadIds+            eThreads <- atomically . readTVar . expirationThreads $ inst+            mapM_ killThread $ map snd (M.toList eThreads) +        _ -> return ()+ -- | Ping all known nodes every five minutes to make sure they are still present pingProcess :: (Serialize i, Serialize a, Eq i) => KademliaInstance i a             -> Chan (Reply i a) -> IO ()-pingProcess (KI h (KS sTree _)) chan = forever $ do+pingProcess (KI h (KS sTree _) _) chan = forever $ do     threadDelay fiveMinutes      tree <- atomically . readTVar $ sTree-    forM_ (allNodes tree) $ \node -> do+    forM_ (T.toList tree) $ \node -> do         -- Send PING and expect a PONG         send h (peer node) PING         expect h (RR [R_PONG] (nodeId node)) $ chan      where fiveMinutes = 300000000-          allNodes = join . catMaybes . map snd --- | Store all values stored in the node in the 7 closest known nodes every day+-- | Store all values stored in the node in the 7 closest known nodes every hour spreadValueProcess :: (Serialize i, Serialize a, Eq i) => KademliaInstance i a                    -> IO ()-spreadValueProcess (KI h (KS sTree sValues)) = forever $ do-    threadDelay day+spreadValueProcess (KI h (KS sTree sValues) _) = forever $ do+    threadDelay hour      values <- atomically . readTVar $ sValues     tree <- atomically . readTVar $ sTree      mapMWithKey (sendRequests tree) $ values -    where day = 24 * 60 * 60 * 1000000+    where hour = 60 * 60 * 1000000           sendRequests tree key val = do             let closest = T.findClosest tree key 7             forM_ closest $ \node -> send h (peer node) (STORE key val)@@ -149,6 +241,24 @@           mapMWithKey :: (k -> v -> IO a) -> M.Map k v -> IO [a]           mapMWithKey f m = sequence . map snd . M.toList . M.mapWithKey f $ m +-- | Delete a value after a certain amount of time has passed+expirationProcess :: (Ord i) => KademliaInstance i a -> i -> IO ()+expirationProcess inst@(KI _ _ valueTs) key = do+    -- Map own ThreadId to the key+    myTId <- myThreadId+    oldTId <- atomically $ do+        threadIds <- readTVar valueTs+        writeTVar valueTs $ M.insert key myTId threadIds+        return . M.lookup key $ threadIds++    -- Kill the old timeout thread, if it exists+    when (isJust oldTId) (killThread . fromJust $ oldTId)++    threadDelay hour+    deleteValue key inst++    where hour = 60 * 60 * 1000000+ -- | Handles the differendt Kademlia Commands appropriately handleCommand :: (Serialize i, Eq i, Ord i, Serialize a) =>     Command i a -> Peer -> KademliaInstance i a -> IO ()@@ -156,8 +266,10 @@ handleCommand PING peer inst = send (handle inst) peer PONG -- Return a KBucket with the closest Nodes handleCommand (FIND_NODE id) peer inst = returnNodes peer id inst--- Insert the value into the values Map-handleCommand (STORE key value) _ inst = insertValue key value inst+-- Insert the value into the values store and start the expiration process+handleCommand (STORE key value) _ inst = do+    insertValue key value inst+    void . forkIO . expirationProcess inst $ key -- Return the value, if known, or the closest other known Nodes handleCommand (FIND_VALUE key) peer inst = do     result <- lookupValue key inst@@ -169,7 +281,7 @@ -- | Return a KBucket with the closest Nodes to a supplied Id returnNodes :: (Serialize i, Eq i, Ord i, Serialize a) =>     Peer -> i -> KademliaInstance i a -> IO ()-returnNodes peer id (KI h (KS sTree _)) = do+returnNodes peer id (KI h (KS sTree _) _) = do     tree <- atomically . readTVar $ sTree     let nodes = T.findClosest tree id 7     liftIO $ send h peer (RETURN_NODES id nodes)

@@ -41,8 +41,9 @@  -- | Open a Kademlia connection on specified port and return a corresponding --   KademliaHandle-openOn :: (Serialize i, Serialize a) => String -> i -> IO (KademliaHandle i a)-openOn port id = withSocketsDo $ do+openOn :: (Serialize i, Serialize a) => String -> i -> ReplyQueue i a+       -> IO (KademliaHandle i a)+openOn port id rq = withSocketsDo $ do     -- Get addr to bind to     (serveraddr:_) <- getAddrInfo                  (Just (defaultHints {addrFlags = [AI_PASSIVE]}))@@ -54,7 +55,6 @@      chan <- newChan     tId <- forkIO . sendProcess sock id $ chan-    rq <- emptyReplyQueue     mvar <- newEmptyMVar      -- Return the handle@@ -82,8 +82,8 @@ -- --   This throws an exception if called a second time. startRecvProcess :: (Serialize i, Serialize a, Eq i, Eq a) => KademliaHandle i a-                 -> Chan (Reply i a) -> IO ()-startRecvProcess kh defaultChan = do+                 -> IO ()+startRecvProcess kh = do     tId <- forkIO $ (withSocketsDo . forever $ do         -- Read from socket         (received, addr) <- S.recvFrom (kSock kh) 1500@@ -95,18 +95,14 @@                 -- Try parsing the signal                 case parse p received of                     Left _    -> return ()-                    Right sig -> do-                        -- Try to dispatch the signal-                        success <- dispatch sig $ replyQueue kh--                        unless success $-                            -- Send it to the default channel-                            writeChan defaultChan $ Answer sig)+                    Right sig ->+                        -- Send the signal to the receivng process of instance+                        writeChan (timeoutChan . replyQueue $ kh) $ Answer sig)              -- Send Closed reply to all handlers             `finally` do                 flush . replyQueue $ kh-                writeChan defaultChan  Closed+                writeChan (timeoutChan . replyQueue $ kh) Closed      success <- tryPutMVar (recvThread kh) tId     unless success . ioError . userError $ "Receiving process already running"@@ -125,13 +121,13 @@ -- | Close the connection corresponding to a KademliaHandle closeK :: KademliaHandle i a -> IO () closeK kh = do-    -- Kill sendThread-    killThread . sendThread $ kh-     -- Kill recvThread     empty <- isEmptyMVar . recvThread $ kh     unless empty $ do         tId <- takeMVar . recvThread $ kh         killThread tId++    -- Kill sendThread+    killThread . sendThread $ kh      yield

@@ -115,7 +115,7 @@     return $ Node peer id  -- | Parses a trailing k-bucket-parseKBucket :: (Serialize i) => Parse (KBucket i)+parseKBucket :: (Serialize i) => Parse [Node i] parseKBucket = liftM2 (:) parseNode parseKBucket                    `catchE` \_ -> return [] 

@@ -12,7 +12,7 @@     ( ReplyType(..)     , ReplyRegistration(..)     , Reply(..)-    , ReplyQueue+    , ReplyQueue(..)     , emptyReplyQueue     , register     , dispatch@@ -22,7 +22,7 @@ import Control.Concurrent import Control.Concurrent.STM import Control.Concurrent.Chan-import Control.Monad (liftM, forM_)+import Control.Monad (liftM3, forM_) import Control.Monad.Trans.Maybe import Data.List (find, delete) @@ -36,20 +36,22 @@ data ReplyType i = R_PONG                  | R_RETURN_VALUE i                  | R_RETURN_NODES i-                   deriving (Eq)+                   deriving (Eq, Show)  -- | The representation of registered replies data ReplyRegistration i = RR {       replyTypes  :: [ReplyType i]     , replyOrigin :: i-    } deriving (Eq)+    } deriving (Eq, Show)  -- | Convert a Signal into its ReplyRegistration representation-toRegistration :: Signal i a -> Maybe (ReplyRegistration i)-toRegistration sig = case rType . command $ sig of-                        Nothing -> Nothing-                        Just rt -> Just (RR [rt] origin)-    where origin = nodeId . source $ sig+toRegistration :: Reply i a -> Maybe (ReplyRegistration i)+toRegistration Closed        = Nothing+toRegistration (Timeout reg) = Just reg+toRegistration (Answer sig)  = case rType . command $ sig of+            Nothing -> Nothing+            Just rt -> Just (RR [rt] (origin sig))+    where origin sig = nodeId . source $ sig            rType :: Command i a -> Maybe (ReplyType i)           rType  PONG               = Just  R_PONG@@ -66,77 +68,77 @@ data Reply i a = Answer (Signal i a)                | Timeout (ReplyRegistration i)                | Closed-                 deriving (Eq)+                 deriving (Eq, Show)  -- | The actual type representing a ReplyQueue-newtype ReplyQueue i a = RQ (TVar [(ReplyRegistration i, Chan (Reply i a), ThreadId)])+data ReplyQueue i a = RQ {+      queue :: (TVar [(ReplyRegistration i, Chan (Reply i a), ThreadId)])+    , timeoutChan :: Chan (Reply i a)+    , defaultChan :: Chan (Reply i a)+    } --- | Create an empty ReplyQueue+-- | Create a new ReplyQueue emptyReplyQueue :: IO (ReplyQueue i a)-emptyReplyQueue = atomically . liftM RQ $ newTVar []+emptyReplyQueue = liftM3 RQ (atomically . newTVar $ []) newChan $ newChan  -- | Register a channel as handler for a reply register :: (Eq i) => ReplyRegistration i -> ReplyQueue i a -> Chan (Reply i a)          -> IO ()-register reg (RQ rq) chan = do-    tId <- timeoutThread chan reg (RQ rq)+register reg rq chan = do+    tId <- timeoutThread reg rq     atomically $ do-        queue <- readTVar $ rq-        writeTVar rq $ queue ++ [(reg, chan, tId)]+        rQueue <- readTVar . queue $ rq+        writeTVar (queue rq) $ rQueue ++ [(reg, chan, tId)] -timeoutThread :: (Eq i) => Chan (Reply i a) -> ReplyRegistration i-              -> ReplyQueue i a -> IO ThreadId-timeoutThread chan reg (RQ rq) = forkIO $ do+timeoutThread :: (Eq i) => ReplyRegistration i -> ReplyQueue i a -> IO ThreadId+timeoutThread reg rq = forkIO $ do     -- Wait 5 seconds     threadDelay 5000000      -- Remove the ReplyRegistration from the ReplyQueue     myTId <- myThreadId-    atomically $ do-        queue <- readTVar $ rq-        case find (\(_, _, tId) -> tId == myTId) queue of-            Just rqElem -> writeTVar rq $ delete rqElem queue-            _ -> return ()      -- Send Timeout signal-    writeChan chan . Timeout $ reg+    writeChan (timeoutChan rq) . Timeout $ reg --- | Try to send a received Signal over the registered handler channel and---   return wether it succeeded-dispatch :: (Eq i) => Signal i a -> ReplyQueue i a -> IO Bool-dispatch sig (RQ rq) = do+-- | Dispatch a reply over a registered handler. If there is no handler,+--   dispatch it to the default one.+dispatch :: (Eq i) => Reply i a -> ReplyQueue i a -> IO ()+dispatch reply rq = do+    -- Try to find a registration matching the reply     result <- atomically $ do-        queue <- readTVar $ rq-        case toRegistration sig of-            Just regA -> case find (matches regA) queue of-                Just reg -> do+        rQueue <- readTVar . queue $ rq+        case toRegistration reply of+            Just repReg -> case find (matches repReg) rQueue of+                Just registration -> do                     -- Remove registration from queue-                    writeTVar rq $ delete reg queue-                    return . Just $ reg+                    writeTVar (queue rq) $ delete registration rQueue+                    return . Just $ registration                  Nothing -> return Nothing-            Nothing  -> return Nothing+            Nothing -> return Nothing      case result of         Just (_, chan, tId) -> do             -- Kill the timeout thread             killThread tId -            -- Send the signal-            writeChan chan $ Answer sig-            return True-        _ -> return False+            -- Send the reply+            writeChan chan reply +        -- Send the reply over the default channel+        Nothing -> writeChan (defaultChan rq) reply+     where matches regA (regB, _, _) = matchRegistrations regA regB  -- | Send Closed signal to all handlers and empty ReplyQueue flush :: ReplyQueue i a -> IO ()-flush (RQ rq) = do-    queue <- atomically $ do-        queue <- readTVar $ rq-        writeTVar rq $ []-        return queue+flush rq = do+    rQueue <- atomically $ do+        rQueue <- readTVar . queue $ rq+        writeTVar (queue rq) []+        return rQueue -    forM_ queue $ \(_, chan, tId) -> do+    forM_ rQueue $ \(_, chan, tId) -> do         killThread tId         writeChan chan Closed

@@ -14,189 +14,218 @@     , insert     , lookup     , delete-    , refresh+    , handleTimeout     , findClosest     , extractId+    , toList+    , fold     ) where +import Prelude hiding (lookup) import Network.Kademlia.Types-import qualified Data.List as L (delete, find)-import Prelude hiding (lookup, split)-import Control.Monad (liftM)-import Control.Arrow (first, second)-import Data.Function (on)---- | Type used for building the Node Storage Tree-type NodeTree i = [(Bool, Maybe (KBucket i))]+import qualified Data.List as L (find, delete) --- | Structure used for easier modification of the NodeTree-type Zipper i = (NodeTree i, NodeTree i)+data NodeTree i = NodeTree ByteStruct (NodeTreeElem i) --- | Move the Zipper along an Id-seek :: (Serialize i) => NodeTree i -> i -> Zipper i-seek tree id = go tree $ toByteStruct id-    where go [] _ = ([], [])-          go (pair@(bit, bucket):rest) (b:bs)-            | ends rest = ([], pair:rest)-            | bit == b  = first (pair:) $ go rest bs-            | otherwise = ([], pair:rest)+data NodeTreeElem i = Split (NodeTreeElem i) (NodeTreeElem i)+                    | Bucket ([(Node i, Int)], [Node i]) --- | Cheks wether a NodeTree ends-ends :: NodeTree i -> Bool-ends ((_, Just _):_)  = False-ends ((_, Nothing):_) = True+type NodeTreeFunction i a = Int -> Bool -> ([(Node i, Int)], [Node i]) -> a --- | Apply a function to the KBucket a Node with a given Id would be in-applyTo :: (Serialize i, Eq i) =>-           (KBucket i -> a) -- ^ Function to apply at matched position-        -> a                -- ^ Default value-        -> NodeTree i       -- ^ NodeTree to apply to-        -> i                -- ^ Position to apply at-        -> a-applyTo f end tree id = case seek tree id of-        (_, [])                 -> end-        (_, (_, Nothing):_)     -> end-        (_, (_, Just bucket):_) -> f bucket+-- | Modify the position in the tree where the supplied id would be+modifyAt :: (Serialize i) =>+            NodeTree i -> i -> NodeTreeFunction i (NodeTreeElem i)+         -> NodeTree i+modifyAt (NodeTree idStruct elem) id f =+    let targetStruct = toByteStruct id+        newElems     = go idStruct targetStruct 0 True elem+    in  NodeTree idStruct newElems+    where -- This function is partial, but we know that there will alwasys be a+          -- bucket at the end. Therefore, we don't have to check for empty+          -- ByteStructs+          --+          -- Apply the function to the position of the bucket+          go _ _ depth valid (Bucket b) = f depth valid b+          -- If the bit is a 0, go left+          go (i:is) (False:ts) depth valid (Split left right) =+               let new = go is ts (depth + 1) (valid && not i) left+               in  Split new right+          -- Otherwise, continue to the right+          go (i:is) (True:ts) depth valid (Split left right) =+               let new = go is ts (depth + 1) (valid && i) right+               in  Split left new --- | Modify a NodeTree at the position a Node with a given Id would have-modifyTreeAt :: (Serialize i, Eq i) =>-                ((Bool, Maybe (KBucket i)) -> (Bool, Maybe (KBucket i)))-                -- ^ Function to apply to corresponding TreeNode-             -> NodeTree i -- ^ NodeTree to modify-             -> i          -- ^ Position to modify at-             -> NodeTree i-modifyTreeAt f tree id = case seek tree id of-        (beg, [])       -> beg-        (beg, pair:end) -> beg ++ f pair : end+-- | Modify and apply a function at the position in the tree where the+--   supplied id would be+bothAt :: (Serialize i) =>+            NodeTree i -> i -> NodeTreeFunction i (NodeTreeElem i, a)+         -> (NodeTree i, a)+bothAt (NodeTree idStruct elem) id f =+    let targetStruct    = toByteStruct id+        (newElems, val) = go idStruct targetStruct 0 True elem+    in  (NodeTree idStruct newElems, val)+    where -- This function is partial, but we know that there will alwasys be a+          -- bucket at the end. Therefore, we don't have to check for empty+          -- ByteStructs+          --+          -- Apply the function to the position of the bucket+          go _ _ depth valid (Bucket b) = f depth valid b+          -- If the bit is a 0, go left+          go (i:is) (False:ts) depth valid (Split left right) =+               let (new, val) = go is ts (depth + 1) (valid && not i) left+               in  (Split new right, val)+          -- Otherwise, continue to the right+          go (i:is) (True:ts) depth valid (Split left right) =+               let (new, val) = go is ts (depth + 1) (valid && i) right+               in  (Split left new, val) --- | Modify the KBucket a node of a given Id would be in-modifyKBucket :: (Serialize i, Eq i) =>-                 (KBucket i -> KBucket i) -- ^ Modification funciton-              -> NodeTree i -- ^ Node tree to modify-              -> i          -- ^ Postition to modify at-              -> NodeTree i-modifyKBucket f = modifyTreeAt (second . fmap $ f)+-- | Apply a function to the bucket the supplied id would be located in+applyAt :: (Serialize i) => NodeTree i -> i -> NodeTreeFunction i a -> a+applyAt (NodeTree idStruct elem) id f =+    let targetStruct = toByteStruct id+    in go idStruct targetStruct 0 True elem+    where -- This function is partial for the same reason as in modifyAt+          --+          -- Apply the function+          go _ _ depth valid (Bucket b) = f depth valid b+          -- If the bit is a 0, go left+          go (i:is) (False:ts) depth valid (Split left _) =+               go is ts (depth + 1) (valid && not i) left+          -- Otherwise, continue to the right+          go (i:is) (True:ts) depth valid (Split _ right) =+               go is ts (depth + 1) (valid && i) right --- | Create a NodeTree corresponding to the Owner-Node's Id+-- | Create a NodeTree corresponding to the id create :: (Serialize i) => i -> NodeTree i-create id = zip (toByteStruct id) (repeat Nothing)---- | Insert a node into a NodeTree-insert :: (Serialize i, Eq i, Ord i) => NodeTree i -> Node i -> NodeTree i-insert tree node = case seek tree . nodeId $ node of-        -- The tree is empty, create first KBucket-        (_, (b, Nothing):xs)       -> (b, Just [node]):xs--        -- Normal case-        (beg, (b, Just bucket):xs)-            -- At least refresh the Node, as it has been active-            | node `elem` bucket -> refresh tree . nodeId $ node-            -- The last bucket may always be split-            | full bucket && ends xs -> let new = split tree . extractId $ tree-                                        in insert new node-            -- If the bucket is full and can't be split, the Node isn't inserted-            | full bucket -> tree-            -- Just insert the Node-            | otherwise -> beg ++ (b, Just $ node:bucket):xs+create id = NodeTree (toByteStruct id) . Bucket $ ([], []) -    where full b = length b >= 7+-- | Lookup a node within a NodeTree+lookup :: (Serialize i, Eq i) => NodeTree i -> i -> Maybe (Node i)+lookup tree id = applyAt tree id f+    where f _ _ = L.find (idMatches id) . map fst . fst --- Extract original Id from NodeTree-extractId :: (Serialize i) => NodeTree i -> i-extractId tree = fromByteStruct bs-    where bs = foldr (\x id -> fst x:id) [] tree+-- | Delete a Node corresponding to a supplied Id from a NodeTree+delete :: (Serialize i, Eq i) => NodeTree i -> i -> NodeTree i+delete tree id = modifyAt tree id f+    where f _ _ (nodes, cache) =+              let deleted = filter (not . idMatches id . fst) $ nodes+              in Bucket (deleted, cache) --- | Split the last bucket------   This function does some quite unsafe pattern matching for the sake of not---   ending up even longer than it already is. It is only used internally and---   all the assumptions made by those patterns are provable, so it's ok.-split :: (Serialize i, Ord i) => NodeTree i -> i -> NodeTree i-split tree id = let (begin, (b, Just bucket):xs) = seek tree id-                    (this, next) = doSplit bucket-                in begin ++ (b, Just this) : injectBucket next xs+-- | Handle a timed out node by incrementing its timeoutCount and deleting it+--  if the count exceeds the limit. Also, return wether it's reasonable to ping+--  the node again.+handleTimeout :: (Serialize i, Eq i) => NodeTree i -> i -> (NodeTree i, Bool)+handleTimeout tree id = bothAt tree id f+    where f _ _ (nodes, cache) = case L.find (idMatches id . fst) nodes of+            -- Delete a node that exceeded the limit. Don't contact it again+            --   as it is now considered dead+            Just x@(_, 4) -> (Bucket (L.delete x $ nodes, cache), False)+            -- Increment the timeoutCount+            Just x@(n, timeoutCount) ->+                 (Bucket ((n, timeoutCount + 1) : L.delete x nodes, cache), True)+            -- Don't contact an unknown node a second time+            Nothing -> (Bucket (nodes, cache), False) -    where doSplit []        = ([], [])-          doSplit (node:ns) =-            -- More matching bytes than the index means that a node can be-            -- moved to a later bucket.-            if countMatching (toByteStruct . nodeId $ node)-                             (toByteStruct id)               > index-              then second (node:) $ doSplit ns-              else first  (node:) $ doSplit ns+-- | Refresh the node corresponding to a supplied Id by placing it at the first+--   index of it's KBucket and reseting its timeoutCount, then return a Bucket+--   NodeTreeElem+refresh :: (Serialize i, Eq i) => Node i -> ([(Node i, Int)], [Node i]) -> NodeTreeElem i+refresh node (nodes, cache) =+         Bucket (case L.find (idMatches (nodeId node) . fst) nodes of+            Just x@(n, _) -> (n, 0) : L.delete x nodes+            _             -> nodes+            , cache) -          index = let (beg, _) = seek tree id in length beg-          countMatching [] [] = 0-          countMatching (a:as) (b:bs)-            | a == b    = 1 + countMatching as bs-            | otherwise = 0+-- | Insert a node into a NodeTree+insert :: (Serialize i, Eq i) => NodeTree i -> Node i -> NodeTree i+insert tree node = if applyAt tree (nodeId node) needsSplit+                   -- Split the tree before inserting, when it makes sense+                   then let splitTree = split tree . nodeId $ node+                        in insert splitTree node+                   -- Insert the node+                   else modifyAt tree (nodeId node) doInsert -          injectBucket bucket ((b, _):xs) = (b, Just bucket):xs+    where needsSplit depth valid (nodes, _) =+            let maxDepth = (length . toByteStruct . nodeId $ node) - 1+            in  -- A new node will be inserted+                node `notElem` map fst nodes &&+                -- The bucket is full+                length nodes >= 7 &&+                -- The bucket may be split+                (depth < 5 || valid) && depth <= maxDepth --- | Lookup a node within a NodeTree-lookup :: (Serialize i, Eq i) => NodeTree i -> i -> Maybe (Node i)-lookup tree id = applyTo f Nothing tree id-    where f = L.find $ idMatches id+          doInsert _ _ b@(nodes, cache)+            -- Refresh an already existing node+            | node `elem` map fst nodes = refresh node b+            -- Simply insert the node, if the bucket isn't full+            | length nodes < 7 = Bucket ((node, 0):nodes, cache)+            -- Move the node to the first spot, if it's already cached+            | node `elem` cache = Bucket (nodes, node : L.delete node cache)+            -- Cache the node and drop older ones, if necessary+            | otherwise = Bucket (nodes, node : take 4 cache) --- | Delete a Node corresponding to a supplied Id from a NodeTree-delete :: (Serialize i, Eq i) => NodeTree i -> i -> NodeTree i-delete tree id = modifyKBucket f tree id-    where f = filter $ not . idMatches id+-- | Split the KBucket the specified id would reside in into two and return a+--   Split NodeTreeElem+split :: (Serialize i) => NodeTree i -> i -> NodeTree i+split tree splitId = modifyAt tree splitId f+    where f depth _ (nodes, cache) =+            let (leftNodes, rightNodes) = splitBucket depth fst nodes+                (leftCache, rightCache) = splitBucket depth id cache+            in  Split+                   (Bucket (leftNodes, leftCache))+                   (Bucket (rightNodes, rightCache)) --- | Refresh the node corresponding to a supplied Id by placing it at the first---   index of it's KBucket-refresh :: (Serialize i, Eq i) => NodeTree i -> i -> NodeTree i-refresh tree id = modifyKBucket f tree id-    where f bucket = case L.find (idMatches id) bucket of-                Just node -> node : L.delete node bucket-                _         -> bucket+          -- Recursivly split the nodes into two buckets+          splitBucket _ _ []     = ([], [])+          splitBucket i f (n:ns) = let bs = toByteStruct . nodeId . f $ n+                                       bit = bs !! i+                                       (left, right) = splitBucket i f ns+                                   in if bit+                                      then (left, n:right)+                                      else (n:left, right)  -- | Find the k closest Nodes to a given Id------   Uset to implemenet RETURN_NODES-findClosest :: (Serialize i, Eq i) => NodeTree i -> i -> Int -> KBucket i-findClosest tree id n = case seek tree id of-    -- The tree is empty-    (_, (_, Nothing):xs) -> []--    -- Normal case-    (beg, (_, Just bk):xs)-        -- The bucket contains enough Nodes on its own-        | length bk == n -> bk-        -- We need to retrieve Nodes from other buckets as well-        | otherwise -> let-            missing = n - length bk-            in if ends xs-                    -- If it's the last one, take nodes from higher up in-                    -- the hierarchy-                    then let higher = next missing $ reverse beg-                         in take n . flip sortByDistanceTo id $ bk ++ higher-                    -- Else retrieve the missing amount of Nodes by calling-                    -- findClosest with an Id whose first differing bit doesn't-                    -- differ.-                    -- (Sounds complicated, but the tests prove that it actually-                    -- works this way)-                    else let treeId = extractId tree-                             newId  = id `alignedTo` treeId-                             other  = findClosest tree newId missing-                         in bk ++ other-    where -- Pick the n closest Nodes from the tree-          next _ [] = []-          next n ((_, Nothing):xs) = next n xs-          next n ((_, Just bk):xs)-            | length bk == n = bk-            | length bk <  n = bk ++ next (n - length bk) xs-            -- Take the n closest Nodes-            | otherwise = take n . sortByDistanceTo bk $ id+findClosest :: (Serialize i) => NodeTree i -> i -> Int -> [Node i]+findClosest (NodeTree idStruct elem) id n =+    let targetStruct = toByteStruct id+    in go idStruct targetStruct elem n+    where -- This function is partial for the same reason as in modifyAt+          --+          -- Take the n closest nodes+          go _ _ (Bucket (nodes, _)) n+            | length nodes <= n = map fst nodes+            | otherwise = take n . sortByDistanceTo (map fst nodes) $ id+          -- Take the closest nodes from the left child first, if those aren't+          -- enough, take the rest from the right+          go (i:is) (False:ts) (Split left right) n =+            let result = go is ts left n+            in if length result == n+                then result+                else result ++ go is ts right n+          -- Take the closest nodes from the right child first, if those aren't+          -- enough, take the rest from the left+          go (i:is) (True:ts) (Split left right) n =+            let result = go is ts right n+            in if length result == n+                then result+                else result ++ go is ts left n -          -- Change the first differing bit of idA to match idB-          idA `alignedTo` idB = fromByteStruct . alignF idA $ idB-          alignF = align `on` toByteStruct-          align [] [] = []-          align (a:as) (b:bs)-            | a == b    = a : align as bs-            | otherwise = b : as+-- Extract original Id from NodeTree+extractId :: (Serialize i) => NodeTree i -> i+extractId (NodeTree id _) = fromByteStruct id  -- | Helper function used for KBucket manipulation idMatches :: (Eq i) => i -> Node i -> Bool idMatches id node = id == nodeId node++-- | Turn the NodeTree into a list of nodes+toList :: NodeTree i -> [Node i]+toList (NodeTree _ elems) = go elems+    where go (Split left right) = go left ++ go right+          go (Bucket b) = map fst . fst $ b++-- | Fold over the buckets+fold :: ([Node i] -> a -> a) -> a -> NodeTree i -> a+fold f init (NodeTree _ elems) = go init elems+    where go a (Split left right) = let a' = go a left in go a' right+          go a (Bucket b) = f (map fst . fst $ b) a

@@ -10,7 +10,6 @@     ( Peer(..)     , toPeer     , Node(..)-    , KBucket     , sortByDistanceTo     , Serialize(..)     , Signal(..)@@ -39,12 +38,8 @@     , nodeId :: i     } deriving (Eq, Ord, Show) --- | Aliases to make the code more readable by using the same names as the---   papers-type KBucket i = [Node i]- -- | Sort a bucket by the closeness of its nodes to a give Id-sortByDistanceTo :: (Serialize i) => KBucket i -> i -> KBucket i+sortByDistanceTo :: (Serialize i) => [Node i] -> i -> [Node i] sortByDistanceTo bucket id = unpack . sort . pack $ bucket     where pack bk = zip bk $ map f bk           f = distance id . nodeId@@ -103,7 +98,7 @@                  | PONG                  | STORE        i a                  | FIND_NODE    i-                 | RETURN_NODES i (KBucket i)+                 | RETURN_NODES i [Node i]                  | FIND_VALUE   i                  | RETURN_VALUE i a                    deriving (Eq, Show)

@@ -0,0 +1,122 @@+{-|+Module      : Implementation+Description : Tests for Network.Kademlia.Implementation++Tests specific to Network.Kademlia.Implementation.+-}++module Implementation where++import Test.HUnit hiding (assert)+import Test.QuickCheck+import Test.QuickCheck.Monadic+import TestTypes++import qualified Network.Kademlia as K+import qualified Network.Kademlia.Tree as T+import Network.Kademlia.Types+import Network.Kademlia.Instance++import Control.Monad+import Control.Applicative+import Control.Concurrent.STM++import Data.Maybe (isJust, fromJust)+import qualified Data.ByteString.Char8 as C++constructNetwork :: IdBunch IdType -> PropertyM IO [KademliaInstance IdType String]+constructNetwork idBunch = run $ do+    let entryNode = Node (Peer "127.0.0.1" 1123) (head . getIds $ idBunch)+    instances <- zipWithM K.create [1123..] (getIds idBunch)+                        :: IO [KademliaInstance IdType String]++    forM_ (tail instances) (`K.joinNetwork` entryNode)+    return instances++joinCheck :: IdBunch IdType -> Property+joinCheck idBunch = monadicIO $ do+    instances <- constructNetwork idBunch+    present <- run $ do+        mapM_ K.close instances+        mapM filled instances+    assert . and $ present++    where filled inst = do+            tree <- atomically . readTVar . sTree . state $ inst+            return $ (length . T.toList $ tree) >= 7++-- | Make sure ID clashes are detected properly+idClashCheck :: IdType -> IdType -> Property+idClashCheck idA idB = monadicIO $ do+    let peers = map (Peer "127.0.0.1") [1123..]+        ids = [idA, idB, idA]+        entryNode = Node (Peer "127.0.0.1" 1124) idB++    joinResult <- run $ do+        insts@[kiA, _, kiB] <- zipWithM K.create [1123..] ids+                            :: IO [KademliaInstance IdType String]++        K.joinNetwork kiA $ entryNode+        joinResult <- K.joinNetwork kiB $ entryNode++        mapM_ K.close insts++        return joinResult++    assert $ joinResult == K.IDClash+++-- | Make sure an offline peer is detected+nodeDownCheck :: Assertion+nodeDownCheck = do+    let entryNode = Node (Peer "127.0.0.1" 1124) idB+    inst <- K.create 1123 idA :: IO (KademliaInstance IdType String)+    joinResult <- K.joinNetwork inst entryNode+    K.close inst++    assertEqual "" joinResult K.NodeDown++    where idA = IT . C.pack $ "hello"+          idB = IT . C.pack $ "herro"++storeAndLookupCheck :: IdBunch IdType -> IdBunch IdType -> Property+storeAndLookupCheck ids keys = monadicIO $ do+    let keyVal = zip (getIds keys) vals+    instances <- constructNetwork ids++    success <- run $ do+        mapM_ (doStore instances) keyVal++        success <- forM instances $ \inst ->+            and <$> mapM (tryLookup inst) keyVal++        mapM_ K.close instances++        return success++    assert . and $ success++    where vals = take 20 . map (replicate 5) $ ['a'..]+          doStore instances (key, val) = K.store (head instances) key val+          tryLookup inst (key, val) = do+            result <- K.lookup inst key+            case result of+                Just (v, _) -> return $ v == val+                _ -> return False++lookupNodesCheck :: IdBunch IdType -> Property+lookupNodesCheck ids = monadicIO $ do+    instances <- constructNetwork ids++    success <- run $ do+        success <- forM instances $ \inst ->+            and <$> (mapM (tryLookup inst) . getIds $ ids)++        mapM_ K.close instances++        return success++    assert . and $ success++    where tryLookup inst id = check id <$> K.lookupNode inst id+          check id result = isJust result && id == (nodeId . fromJust $ result)

@@ -0,0 +1,132 @@+{-|+Module      : Instance+Description : Tests for Network.Kademlia.Instance++Tests specific to Network.Kademlia.Instance.+-}++module Instance where++import Test.HUnit hiding (assert)+import Test.QuickCheck+import Test.QuickCheck.Monadic++import Network.Kademlia.Instance as I+import Network.Kademlia+import Network.Kademlia.Networking+import Network.Kademlia.Types+import Network.Kademlia.ReplyQueue+import qualified Data.ByteString.Char8 as C+import Control.Concurrent.Chan+import Control.Monad (liftM2)+import Data.Maybe (isJust, fromJust)++import TestTypes++-- | The default set of peers+peers :: (Peer, Peer)+peers = let pA = Peer "127.0.0.1" 1122+            pB = Peer "127.0.0.1" 1123+        in (pA, pB)++-- | A set of randomly generated Ids+ids :: (Monad m) => PropertyM m (IdType, IdType)+ids = liftM2 (,) (pick arbitrary) (pick arbitrary)++-- | Checks wether PINGs are handled appropriately+handlesPingCheck :: Assertion+handlesPingCheck = do+    let (pA, pB) = peers++    let (Right (idA, _)) = fromBS . C.replicate 32 $ 'a'+                           :: Either String (IdType, C.ByteString)+    let (Right (idB, _)) = fromBS . C.replicate 32 $ 'b'+                           :: Either String (IdType, C.ByteString)++    rq <- emptyReplyQueue++    khA <- openOn "1122" idA rq :: IO (KademliaHandle IdType String)+    kiB <- create 1123 idB   :: IO (KademliaInstance IdType String)++    startRecvProcess khA++    send khA pB PING+    (Answer sig) <- readChan . timeoutChan $ rq :: IO (Reply IdType String)++    closeK khA+    close kiB++    assertEqual "" (command sig) PONG+    assertEqual "" (peer . source $ sig) pB+    assertEqual "" (nodeId . source $ sig) idB++    return ()++-- | Make sure a stored value can be retrieved+storeAndFindValueCheck :: IdType -> String -> Property+storeAndFindValueCheck key value = monadicIO $ do+    let (pA, pB) = peers+    (idA, idB) <- ids++    receivedCmd <- run $ do+        rq <- emptyReplyQueue++        khA <- openOn "1122" idA rq+        kiB <- create 1123 idB :: IO (KademliaInstance IdType String)++        startRecvProcess khA++        send khA pB $ STORE key value+        send khA pB $ FIND_VALUE key++        -- There is a race condition, so the instance will sometimes try to store+        -- the value in the handle, before replying with a RETURN_VALUE+        (Answer sig) <- readChan . timeoutChan $ rq :: IO (Reply IdType String)+        sig <- case command sig of+                STORE _ _ -> do+                    (Answer sig) <- readChan . timeoutChan $ rq :: IO (Reply IdType String)+                    return sig+                _ -> return sig++        closeK khA+        close kiB++        return . command $ sig++    let cmd = RETURN_VALUE key value :: Command IdType String++    monitor . counterexample $ "Commands inequal: " ++ show cmd ++ " /= " ++ show receivedCmd+    assert $ cmd == receivedCmd++    return ()++-- | Assert that a peer is put into the NodeTree on first encounter+trackingKnownPeersCheck :: Property+trackingKnownPeersCheck = monadicIO $ do+    let (_, pB) = peers+    (idA, idB) <- ids++    (node, kiB) <- run $ do+        rq <- emptyReplyQueue :: IO (ReplyQueue IdType String)++        khA <- openOn "1122" idA rq+        kiB <- create 1123 idB :: IO (KademliaInstance IdType String)++        startRecvProcess khA++        send khA pB $ PING+        readChan . timeoutChan $ rq++        node <- I.lookupNode kiB idA++        closeK khA+        close kiB++        return (node, kiB)++    assert . isJust $ node++    nodes <- run . dumpPeers $ kiB+    assert $ nodes == [fromJust node]++    return ()

@@ -0,0 +1,94 @@+{-|+Module      : Networking+Description : Tests for Network.Kademlia.Networking++Tests specific to Network.Kademlia.Networking.+-}++module Networking where++import Test.QuickCheck+import Test.QuickCheck.Monadic++import Network.Kademlia.Networking+import Network.Kademlia.Types+import Network.Kademlia.ReplyQueue+import Control.Monad+import Control.Concurrent.Chan+import Control.Concurrent.STM+import qualified Data.ByteString.Char8 as C+import Data.Maybe (isJust)++import TestTypes++valueSet :: (Monad m) => PropertyM m (Peer, Peer, IdType, IdType)+valueSet = do+    let pA = Peer "127.0.0.1" 1122+        pB = Peer "127.0.0.1" 1123++    idA <- pick (arbitrary :: Gen IdType)+    idB <- pick (arbitrary :: Gen IdType)++    return (pA, pB, idA, idB)++-- | Make sure sending and receiving works+sendCheck :: Command IdType String -> Property+sendCheck cmd = monadicIO $ do+    (pA, pB, idA, idB) <- valueSet++    sig <- run $ do+        rqA <- emptyReplyQueue+        rqB <- emptyReplyQueue++        khA <- openOn "1122" idA rqA+        khB <- (openOn "1123" idB rqB :: IO (KademliaHandle IdType String))++        startRecvProcess khB++        send khA pB cmd+        (Answer sig) <- readChan . timeoutChan $ rqB :: IO (Reply IdType String)++        closeK khA+        closeK khB++        return sig++    assert $ command sig == cmd+    assert $ (peer . source $ sig) == pA+    assert $ (nodeId . source $ sig) == idA++    return ()++-- | Make sure expect works the way it's supposed to+expectCheck :: Signal IdType String -> IdType -> Property+expectCheck sig idA = monadicIO $ do+    let rtM = rType . command $ sig+    pre . isJust $ rtM+    let (Just rt) = rtM+        rr = RR [rt] . nodeId . source $ sig++    replySig <- run $ do+        rqA <- emptyReplyQueue++        khA <- openOn "1122" idA rqA++        startRecvProcess khA++        testChan <- newChan :: IO (Chan (Reply IdType String))+        expect khA rr testChan+        dispatch (Answer sig) rqA++        (Answer replySig) <- readChan testChan :: IO (Reply IdType String)++        closeK khA++        return replySig++    assert $ replySig == sig++-- | Convert a command into a ReplyType+rType :: Command i a -> Maybe (ReplyType i)+rType  PONG               = Just  R_PONG+rType (RETURN_VALUE id _) = Just (R_RETURN_VALUE id)+rType (RETURN_NODES id _) = Just (R_RETURN_NODES id)+rType _ = Nothing

@@ -0,0 +1,35 @@+{-|+Module      : Protocol+Description : Test for Network.Kademlia.Protocol++Tests specific to Network.Kademlia.Protocol.+-}++module Protocol+    ( parseCheck+    , lengthCheck+    , IdType(..)+    ) where++import Test.QuickCheck++import qualified Data.ByteString as B+import Network.Kademlia.Types+import Network.Kademlia.Protocol+import TestTypes++-- | A signal is the same as its serialized form parsed+parseCheck :: Signal IdType String -> Property+parseCheck s = test . parse (peer . source $ s) . serialize id . command $ s+    where id = nodeId . source $ s+          test (Left err) = counterexample "Parsing failed" False+          test (Right s') = counterexample+            ("Signals differ:\nIn:  " ++ show s ++ "\nOut: "+                 ++ show s' ++ "\n") $ s === s'++-- | A serialized signal's length is no longer than the max. UDP packet size+--   (or at least what I believe to be the max UDP packet size)+lengthCheck :: Signal IdType String -> Property+lengthCheck s = counterexample err $ length <= 1500+    where length = B.length . serialize (nodeId . source $ s) . command $ s+          err = "Serialized signal is too long: " ++ show length ++ " bytes"

@@ -0,0 +1,87 @@+{-|+Module      : ReplyQueue+Description : Tests for Network.Kademlia.ReplyQueue++Tests specific to Network.Kademlia.ReplyQueue+-}++module ReplyQueue where++import Test.QuickCheck+import Test.QuickCheck.Monadic++import Control.Concurrent.Chan+import Control.Concurrent.STM+import Data.Maybe (isJust)++import Network.Kademlia.ReplyQueue+import Network.Kademlia.Types++import TestTypes++-- | Check wether registered reply handlers a used+repliesCheck :: Signal IdType String -> Signal IdType String -> Property+repliesCheck sig1 sig2 = monadicIO $ do+    let reg1 = toRegistration sig1+    let reg2 = toRegistration sig2++    pre $ isJust reg1 && isJust reg2++    let (Just replyReg1) = reg1+    let (Just replyReg2) = reg2++    contents <- run $ do+        rq <- emptyReplyQueue+        chan <- newChan :: IO (Chan (Reply IdType String))++        register replyReg1 rq chan+        register replyReg2 rq chan++        dispatch (Answer sig1) rq+        dispatch (Answer sig2) rq++        contents <- getChanContents chan++        return contents++    assert . not . null $ contents++    let [reply1, reply2] = take 2 contents++    assert $ reply1 /= Closed+    assert $ reply2 /= Closed++    let (Answer unwrapped1) = reply1+    let (Answer unwrapped2) = reply2++    assert $ unwrapped1 == sig1+    assert $ unwrapped2 == sig2++-- | Check wether registered reply handlers are removed after usage+removedCheck :: Signal IdType String -> Property+removedCheck sig = monadicIO $ do+    let reg = toRegistration sig+    case reg of+        -- Discard the test case+        Nothing -> pre False+        Just reg -> do+            removed <- run $ do+                rq <- emptyReplyQueue+                chan <- newChan :: IO (Chan (Reply IdType String))+                register reg rq chan+                dispatch (Answer sig) rq+                fmap null (atomically . readTVar . queue $ rq)+            assert removed++-- | Convert a Signal into its ReplyRegistration representation+toRegistration :: Signal i a -> Maybe (ReplyRegistration i)+toRegistration sig = case rType . command $ sig of+                        Nothing -> Nothing+                        Just rt -> Just (RR [rt] origin)+    where origin = nodeId . source $ sig++          rType :: Command i a -> Maybe (ReplyType i)+          rType  PONG               = Just  R_PONG+          rType (RETURN_VALUE id _) = Just (R_RETURN_VALUE id)+          rType (RETURN_NODES id _) = Just (R_RETURN_NODES id)+          rType _ = Nothing

@@ -0,0 +1,112 @@+{-|+Module      : Tests+Description : Tests for the modules++A few tests using QuickCheck and Tasty to make sure everything works+the way it's supposed to.+-}++module Main where++import Test.Tasty+import Test.Tasty.QuickCheck as QC+import Test.Tasty.HUnit as HU++import Types+import Protocol+import Networking+import Tree+import Instance+import ReplyQueue+import Implementation++main = defaultMain tests++tests :: TestTree+tests = testGroup "Tests" [quickCheckTests, hUnitTests]++quickCheckTests = testGroup "QuickCheck" [+      typeChecks+    , protocolChecks+    , networkingChecks+    , treeChecks+    , instanceChecks+    , replyQueueChecks+    , implementationChecks+    ]++typeChecks = testGroup "Network.Kademlia.Types" [+      QC.testProperty "ByteString to ByteStruct conversion works"+         toByteStructCheck+    , QC.testProperty "ByteStruct to ByteString conversion works"+         fromByteStructCheck+    ]++protocolChecks = testGroup "Network.Kademlia.Protocol" [+      QC.testProperty "Protocol Serializing and Parsing works"+         parseCheck+    , QC.testProperty "Protocol messages are within the max UDP packet size"+         lengthCheck+    ]++networkingChecks = testGroup "Network.Kademlia.Networking" [+      QC.testProperty "Sending and Receiving works"+         sendCheck+    , QC.testProperty "Expecting works the way it's supposed to"+         expectCheck+    ]++treeChecks = testGroup "Network.Kademlia.Tree" [+      QC.testProperty "Inserting into the Tree works"+         insertCheck+    , QC.testProperty "Deleting from the Tree works"+         deleteCheck+    , QC.testProperty "Splitting works as expected"+         splitCheck+    , QC.testProperty "Buckets are within the size limit"+         bucketSizeCheck+    , QC.testProperty "Refreshing works as expected"+         refreshCheck+    , QC.testProperty "Finding closest works"+         findClosestCheck+    ]++instanceChecks = testGroup "Network.Kademlia.Instance" [+      QC.testProperty "Storing and Retrieving values works"+         storeAndFindValueCheck+    , QC.testProperty "Peers are put into the tree on first encounter"+         trackingKnownPeersCheck+    ]++replyQueueChecks = testGroup "Network.Kademlia.ReplyQueue" [+      QC.testProperty "Registering replies works"+          repliesCheck+    , QC.testProperty "Registrations are removed after being dispatched"+         removedCheck+    ]++implementationChecks = testGroup "Network.Kademlia.Implementation" [+      QC.testProperty "Joining the Network works"+         joinCheck+    , QC.testProperty "ID clashes are detected"+         idClashCheck+    , QC.testProperty "Storing and looking up values works"+        storeAndLookupCheck+    , QC.testProperty "Looking up Nodes works"+        lookupNodesCheck+    ]++hUnitTests = testGroup "HUnit" [+      instanceCases+    , implementationCases+    ]++instanceCases = testGroup "Network.Kademlia.Instance" [+      HU.testCase "PINGs are automaticly handled"+         handlesPingCheck+    ]++implementationCases = testGroup "Network.Kademlia.Implementation" [+      HU.testCase "Trying to join over an offline Node is detected"+         nodeDownCheck+    ]

@@ -0,0 +1,95 @@+{-# LANGUAGE TypeSynonymInstances, FlexibleInstances #-}++{-|+Module      : TestTypes+Description : Types and Generators needed for general testing+-}++module TestTypes where++import Test.QuickCheck++import Control.Monad (liftM, liftM2, liftM3)+import Control.Arrow (first)+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as C+import Network.Socket (PortNumber)+import Data.Word(Word16)+import Data.List (nubBy)+import Data.Function (on)++import Network.Kademlia.Types++newtype IdType = IT { getBS :: B.ByteString } deriving (Eq, Ord)++-- Custom show instance+instance Show IdType where+    show = show . C.unpack . getBS++-- A simple 5-byte ByteString+instance Serialize IdType where+    toBS = getBS+    fromBS bs = if B.length bs >= 5+        then Right $ first IT . B.splitAt 5 $ bs+        else Left "ByteString to short."++instance Serialize String where+    toBS = C.pack . show+    fromBS s =+        case (reads :: ReadS String) . C.unpack $ s of+            []               -> Left "Failed to parse string."+            (result, rest):_ -> Right (result, C.pack rest)++instance Arbitrary IdType where+    arbitrary = do+        str <- vectorOf 5 arbitrary+        return $ IT $ C.pack str++instance Arbitrary PortNumber where+    arbitrary = liftM fromIntegral (arbitrary :: Gen Word16)++instance Arbitrary Peer where+    arbitrary = do+        host <- arbitrary `suchThat` \s -> ' ' `notElem` s && not (null s)+                                           && length s < 20+        port <- arbitrary+        return $ Peer host port++instance (Arbitrary i, Arbitrary v) => Arbitrary (Command i v) where+    arbitrary = oneof [+          return PING+        , return PONG+        , liftM2 STORE arbitrary arbitrary+        , liftM FIND_NODE arbitrary+        , liftM2 RETURN_NODES arbitrary $ vectorOf 15 arbitrary+        , liftM FIND_VALUE arbitrary+        , liftM2 RETURN_VALUE arbitrary arbitrary+        ]++instance (Arbitrary i, Arbitrary v) => Arbitrary (Signal i v) where+    arbitrary = liftM2 Signal arbitrary arbitrary++instance (Arbitrary i) => Arbitrary (Node i) where+    arbitrary = liftM2 Node arbitrary arbitrary++-- | This enables me to specifiy a new Arbitrary instance+newtype NodeBunch i = NB {+      nodes :: [Node i]+    } deriving (Show)++-- | Make sure all Ids are unique+instance (Arbitrary i, Eq i) => Arbitrary (NodeBunch i) where+    arbitrary = liftM NB $ vectorOf 20 arbitrary `suchThat` individualIds+        where individualIds = individual ((==) `on` nodeId)++individual :: (a -> a -> Bool) -> [a] -> Bool+individual eq s = length s == (length . clear $ s)+    where clear = nubBy eq++-- | This is needed for the Implementation tests+newtype IdBunch i = IB {+      getIds :: [i]+    } deriving (Show)++instance (Arbitrary i, Eq i) => Arbitrary (IdBunch i) where+    arbitrary = liftM IB $ vectorOf 20 arbitrary `suchThat` individual (==)

@@ -0,0 +1,86 @@+{-|+Module      : Tree+Description : Tests for Network.Kademlia.Tree++Tests specific to Network.Kademlia.Tree.+-}++module Tree where++import Test.QuickCheck++import qualified Network.Kademlia.Tree as T+import Network.Kademlia.Types+import Control.Monad (liftM)+import Data.List (sortBy)+import Data.Maybe (isJust)++import TestTypes++-- | Helper method for lookup checking+lookupCheck :: (Serialize i, Eq i) => T.NodeTree i -> Node i -> Bool+lookupCheck tree node = T.lookup tree (nodeId node) == Just node++-- | Check wether an inserted Node is retrievable+insertCheck :: IdType -> Node IdType -> Bool+insertCheck id node = lookupCheck tree node+    where tree = T.insert (T.create id) node++-- | Make sure a deleted Node can't be retrieved anymore+deleteCheck :: IdType -> Node IdType -> Bool+deleteCheck id node = not . lookupCheck tree $ node+    where tree = T.delete origin . nodeId $ node+          origin = T.insert (T.create id) node++withTree :: (T.NodeTree IdType -> [Node IdType] -> a) ->+            NodeBunch IdType -> IdType -> a+withTree f bunch id = f tree $ nodes bunch+    where tree = foldr (flip T.insert) (T.create id) $ nodes bunch++splitCheck :: NodeBunch IdType -> IdType -> Property+splitCheck = withTree f+    where f tree nodes = conjoin . foldr (foldingFunc tree) [] $ nodes++          tree `contains` node = node `elem` T.toList tree++          foldingFunc tree node props = prop : props+            where prop =+                    counterexample ("Failed to find " ++ show node) $+                  -- There is the possibiliy that nodes weren't inserted+                  -- because of full buckets.+                    lookupCheck tree node || not (tree `contains` node)++-- | Make sure the bucket sizes end up correct+bucketSizeCheck :: NodeBunch IdType -> IdType -> Bool+bucketSizeCheck = withTree $ \tree _ -> T.fold foldingFunc True tree+    where foldingFunc _ False = False+          foldingFunc b  _    = length b <= 7++-- | Make sure refreshed Nodes are actually refreshed+refreshCheck :: NodeBunch IdType -> IdType -> Bool+refreshCheck = withTree f+    where f tree nodes = T.fold foldingFunc True refreshed+            where refreshed = T.insert tree node+                  node = last nodes+                  foldingFunc _  False = False+                  foldingFunc b _      = node `notElem` b+                                         || head b == node++-- | Make sure findClosest returns the Node with the closest Ids of all nodes+--   in the tree.+findClosestCheck :: IdType -> NodeBunch IdType -> IdType -> Property+findClosestCheck id = withTree f+    where f tree nodes = conjoin . foldr g [] $ manualClosest+           where g node props = counterexample (text node) (prop node):props+                  where prop node = node `elem` treeClosest+                        text node = "Failed to find: " ++ show node++                 treeClosest = T.findClosest tree id 7++                 contained = filter contains nodes+                 contains node = isJust . T.lookup tree . nodeId $ node++                 manualClosest = map fst . take 7 . sort $ packed+                 packed = zip contained $ map distanceF contained+                 distanceF = distance id . nodeId+                 sort = sortBy $ \(_, a) (_, b) -> compare a b

@@ -0,0 +1,28 @@+{-|+Module      : Types+Description : Tests for Network.Kademlia.Types++Tests specific to Network.Kademlia.Types.+-}++module Types where++import Test.QuickCheck++import qualified Data.ByteString as B+import Data.Bits (testBit)++import TestTypes+import Network.Kademlia.Types++-- | Checks wether toByteStruct converts corretctly+toByteStructCheck :: IdType -> Bool+toByteStructCheck id = foldl foldingFunc True [0..length converted - 1]+    where converted = toByteStruct id+          words = B.unpack . toBS $ id+          foldingFunc b i = b && (converted !! i == access words i)+          access ws i = testBit (ws !! (i `div` 8)) (i `mod` 8)++-- | Checks wether fromByteStruct converts corretctly+fromByteStructCheck :: IdType -> Bool+fromByteStructCheck id = id == (fromByteStruct . toByteStruct $ id)