Starting to move stuff around to have more treatment happening in pure code.

Author: Matthieu Riou

src/Comm.hs

@@ -1,6 +1,6 @@
 module Comm
   ( IPPeer(..), parseHeader, localServer, serverDispatch, tunnel,
-    toCharArray, fromCharArray
+    buildHeader, toCharArray, fromCharArray
   ) where
 
 import Network.Socket
@@ -21,7 +21,7 @@ import Debug.Trace
 
 import KTable
 import Globals
-import Kad (storeReceive, nodeLookupReceive, nodeLookupCallback, valueLookupCallback)
+import Kad -- (storeReceive, nodeLookupReceive, nodeLookupCallback, valueLookupCallback)
 
 -- TODO shut out nodes that are too chatty
 -- Header: version - 1 byte
@@ -48,6 +48,13 @@ instance Peer IPPeer where
   sendLookupReply = sendLookupReplyIP
   sendStore = sendStoreIP
   sendValueReply = sendValueReplyIP
+  sendToPeer = sendToPeerIP
+  serPeer p = (serIP $ host p) ++ (serPort $ port p) ++ toCharArray (nodeId p) 20
+
+deserIP = intercalate "." . map (show . ord)
+serIP = map (chr . read) . split '.' 
+deserPort = (show ::Int -> String) . fromCharArray
+serPort = (flip toCharArray 2) . read
 
 instance Node IPPeer where
   nodeId = ipNodeId
@@ -65,35 +72,35 @@ data Header = Header { msgVersion ::Int, msgOp ::KadOp, msgUid ::Word64, sender
 -- operation itself is tracked by a secondary id stored locally. Supports both
 -- node and value lookup using the last boolean
 sendLookupIP peers nid lookupId valL = forM peers (\p -> do 
-  msgId <- liftIO newUid
+  msgId <- newUid
   me    <- askLocalId
   let msg = buildHeader (if valL then ValueLookupOp else NodeLookupOp) msgId me ++ toCharArray nid 20
 
   newWaitingReply p (if valL then ValueLookupReplyOp else NodeLookupReplyOp) msgId lookupId
-  liftIO $ sendToPeer msg p ) >> return ()
+  liftIO $ sendToPeerIP msg p ) >> return ()
 
 -- Sends the reply to a node lookup query, sending k nodes and reproducing the
 -- received message id.
 sendLookupReplyIP peer nodes msgId valL = do
   me <- askLocalId
   let msg = buildHeader (if valL then ValueLookupReplyOp else NodeLookupReplyOp) msgId me ++ serPeers nodes
-  liftIO $ sendToPeer msg peer
+  liftIO $ sendToPeerIP msg peer
 
 sendValueReplyIP peer val msgId = do
   me <- askLocalId
   let msg = buildHeader ValueLookupReplyOp msgId me ++ 
         if length val `mod` 26 == 0 then val ++ " " else val
-  liftIO $ sendToPeer msg peer
+  liftIO $ sendToPeerIP msg peer
 
 sendStoreIP peers key value storeId = forM peers (\p -> do
-  msgId <- liftIO newUid
+  msgId <- newUid
   me    <- askLocalId
   let msg = buildHeader StoreOp msgId me ++ toCharArray key 20 ++ value
 
   newWaitingReply p StoreReplyOp msgId storeId
-  liftIO $ sendToPeer msg p ) >> return ()
+  liftIO $ sendToPeerIP msg p ) >> return ()
 
-sendToPeer msg peer = do
+sendToPeerIP msg peer = do
   phandle <- openPeerHandle (host peer) (port peer)
   sendstr phandle msg
   closePeerHandle phandle
@@ -176,6 +183,16 @@ parseHeader msg = runState parseHeader' msg
           put r
           return $ fn v
 
+-- Deserializes peers from message strings using the opposite transformation from ser
+deserPeers = map deserPeer . splitEvery 26
+deserPeer s = 
+  let (h,rest) = splitAt 4 s
+      (p,nid)  = splitAt 2 rest
+  in IPPeer (deserIP h) (deserPort p) (fromCharArray nid)
+
+split delim = unfoldr (\b -> fmap (const . (second $ drop 1) . break (==delim) $ b) . listToMaybe $ b)
+splitEvery n = takeWhile (not . null) . unfoldr (Just . splitAt n)
+
 tunnel :: (SockAddr -> String -> ServerState p ()) -> ((SockAddr -> String -> IO ()) -> IO ()) -> ServerState p ()
 tunnel f k = do
   gs <- ask
@@ -196,61 +213,11 @@ localServer port handlerFn = do
           handlerFn addr msg
           procMessages sock
 
-
--- Utility functions to serialize messages
---
-
-buildHeader optype msgId sender = 
-  (chr 1) : optypeStr : toCharArray (toInteger msgId) 8 ++ serPeer sender
-  where optypeStr = case optype of
-                      PingOp              -> chr 1
-                      PingReplyOp         -> chr 2
-                      NodeLookupOp        -> chr 3
-                      NodeLookupReplyOp   -> chr 4
-                      StoreOp             -> chr 5
-                      StoreReplyOp        -> chr 6
-                      ValueLookupOp       -> chr 7
-                      ValueLookupReplyOp  -> chr 8
-
--- Convert an integer to a string of n bytes
-toCharArray:: Integer -> Int -> [Char]
-toCharArray num depth = map (chr . fromInteger) (toBytes num $ toInteger depth)
-
--- Converts a string to its numeric value by considering that each character is a
--- byte in a n byte number
-fromCharArray :: (Bits a) => [Char] -> a
-fromCharArray str = foldl (\acc ch -> shift acc 8 + fromIntegral (ord ch)) 0 str
-
--- Serializes peers
-serPeers  = concat . map serPeer
-serPeer p = (serIP $ host p) ++ (serPort $ port p) ++ toCharArray (nodeId p) 20
-
--- Deserializes peers from message strings using the opposite transformation from ser
-deserPeers = map deserPeer . splitEvery 26
-deserPeer s = 
-  let (h,rest) = splitAt 4 s
-      (p,nid)  = splitAt 2 rest
-  in IPPeer (deserIP h) (deserPort p) (fromCharArray nid)
-
-deserIP = intercalate "." . map (show . ord)
-serIP = map (chr . read) . split '.' 
-deserPort = (show ::Int -> String) . fromCharArray
-serPort = (flip toCharArray 2) . read
-
-split delim = unfoldr (\b -> fmap (const . (second $ drop 1) . break (==delim) $ b) . listToMaybe $ b)
-splitEvery n = takeWhile (not . null) . unfoldr (Just . splitAt n)
-
--- Decomposes a number into its successive byte values or, said differently, converts
--- a number to base 2^8.
-toBytes :: Integer -> Integer -> [Integer]
-toBytes num depth = unfoldr byteMod (num,depth-1)
-  where byteMod (num,exp) = let (d,m) = divMod num (2^(exp*8))
-                            in if exp < 0 then Nothing else Just (d, (m,exp-1))
-
 -- Build a new peer from packet address and header peer info
 toPeer addr pr = let (h,p) = span (/=':') (show addr)
                  in IPPeer h (port pr) (nodeId pr)
 
+
 -- Tried below but host is a Word 32...
 --   case addr of
 --     SockAddrInet port host -> Peer (show host) (show port) (-1)

src/Globals.hs

@@ -3,18 +3,19 @@
 
 module Globals (
     Peer(..),
-    RunningOps(..), ServerState, KadOp(..), GlobalData(..), HandlerFn(..),
+    RunningOps(..), ServerState, KadOp(..), GlobalData(..), HandlerFn(..), WaitingReply(..),
     runServer, askRoutingT, askRunningOpsT, askKTree, readKTree, askLocalId,
+    readRoutingT, readRunningOpsT,
     newRunningLookup, runningLookup, runningLookupDone, readStore, 
-    newWaitingReply, waitingReply, newUid, insertInKTree,
+    newWaitingReply, waitingReply, nrandoms, newUid, insertInKTree,
     insertInStore, lookupInStore, deleteInStore,
     modifyTVar
   ) where
 
 import Data.Word
 import Control.Monad.Reader
 import Control.Concurrent.STM
-import System.Random
+import Control.Applicative((<$>))
 
 import qualified Data.Map as M
 import Debug.Trace
@@ -26,6 +27,8 @@ class Peer p where
   sendLookupReply :: p -> [p] -> Word64 -> Bool -> ServerState p () 
   sendStore       :: [p] -> Integer -> String -> Word64 -> ServerState p ()
   sendValueReply  :: p -> String -> Word64 -> ServerState p ()
+  sendToPeer      :: String -> p -> IO ()
+  serPeer         :: p -> String
 
 type RoutingTable p = M.Map Word64 (WaitingReply p)
 
@@ -50,21 +53,22 @@ data GlobalData p = GlobalData {
   runningOpsTable :: TVar (RunningOpsTable p),
   globalKTree     :: TVar (KTree p),
   localPeer       :: p,
-  localStore      :: TVar (M.Map Integer String)
+  localStore      :: TVar (M.Map Integer String),
+  randoms         :: TVar [Integer]
 }
 
 newtype ServerState p a = ServerState {
   runSS:: ReaderT (GlobalData p) IO a 
-} deriving (Monad, MonadIO, MonadReader (GlobalData p))
+} deriving (Monad, MonadIO, MonadReader (GlobalData p), Functor)
 
 -- runServer rt rot kt localId st = runReaderT (runSS st) (rt, rot, kt, localId)
 runServer gd st = runReaderT (runSS st) gd
 
-askRoutingT = liftM routingTable ask
-askRunningOpsT = liftM runningOpsTable ask
-askKTree = liftM globalKTree ask
-askLocalId = liftM localPeer ask
-askLocalStore = liftM localStore ask
+askRoutingT = routingTable <$> ask
+askRunningOpsT = runningOpsTable <$> ask
+askKTree = globalKTree <$> ask
+askLocalId = localPeer <$> ask
+askLocalStore = localStore <$> ask
 
 readRoutingT = do { rt <- askRoutingT; liftIO . atomically $ readTVar rt }
 readRunningOpsT = do { rot <- askRunningOpsT; liftIO . atomically $ readTVar rot }
@@ -74,13 +78,21 @@ readStore = do { s <- askLocalStore; liftIO . atomically $ readTVar s }
 modifyTVar :: (a -> a) -> TVar a -> STM ()
 modifyTVar f tv = readTVar tv >>= writeTVar tv . f
 
+nrandoms n = do 
+  rst <- randoms <$> ask
+  liftIO . atomically $ do
+    rs <- readTVar rst
+    let nrs = take n rs
+    writeTVar rst (drop n rs)
+    return $ map fromInteger nrs
+
 -- Helper to execute atomically a ST action on a value asked in ServerState
 atomicOnAsk :: (MonadIO m) => m a -> (a -> STM b) -> m b
 atomicOnAsk asked action = asked >>= liftIO . atomically . action
 
 insertInStore k v = atomicOnAsk askLocalStore (modifyTVar $ M.insert k v)
 
-lookupInStore k = liftM (M.lookup k) $ atomicOnAsk askLocalStore readTVar 
+lookupInStore k = M.lookup k <$> atomicOnAsk askLocalStore readTVar 
 
 deleteInStore k = atomicOnAsk askLocalStore (\lst -> do
   ls  <- readTVar lst
@@ -94,7 +106,7 @@ newRunningLookup nid rs ps qs lookupId handlerFn =
   let nrl = RunningLookup nid rs ps qs handlerFn
   in atomicOnAsk askRunningOpsT $ modifyTVar (M.insert lookupId nrl)
 
-runningLookup lookupId = liftM (M.lookup lookupId) readRunningOpsT
+runningLookup lookupId = M.lookup lookupId <$> readRunningOpsT
 
 runningLookupDone:: Word64 -> ServerState p (Maybe (RunningOps p))
 runningLookupDone lookupId = atomicOnAsk askRunningOpsT (\rot -> do
@@ -123,6 +135,6 @@ insertInKTree peer = do
   lid <- askLocalId
   atomicOnAsk askKTree $ modifyTVar (flip (kinsert $ nodeId lid) $ peer)
 
-newUid:: IO Word64
-newUid = liftM fromInteger $ randomRIO (0, 2^64 - 1)
+newUid:: (ServerState p) Word64
+newUid = fromInteger <$> head <$> nrandoms 1
 

src/Kad.hs

@@ -2,7 +2,8 @@
 
 module Kad 
   ( startNode, nodeLookup, valueLookup, store, 
-    nodeLookupReceive, storeReceive, nodeLookupCallback, valueLookupCallback
+    nodeLookupReceive, storeReceive, nodeLookupCallback, valueLookupCallback,
+    fromCharArray, toCharArray, buildHeader, serPeers
   ) where
 
 import Prelude hiding (error)
@@ -11,10 +12,13 @@ import Data.Word
 import Data.Maybe
 import Data.List(sortBy, (\\), delete, nub)
 import Data.Bits
+import Data.Char(chr,ord)
+import Data.List (genericDrop,unfoldr,intercalate)
 
 import Control.Monad(liftM, liftM2, liftM3, forM)
 import Control.Concurrent.STM
 import Control.Monad.Trans(liftIO)
+import Control.Arrow(second)
 
 import System.Random
 import System.Log.Logger
@@ -49,7 +53,7 @@ valueLookup = genericLookup True
 
 genericLookup valL nid handlerFn = do
   debug $ "Starting a node lookup for node " ++ show nid
-  uid   <- liftIO newUid
+  uid   <- newUid
   ktree <- readKTree
 
   -- k closest elements in the k tree
@@ -64,9 +68,36 @@ genericLookup valL nid handlerFn = do
       -- sending a lookup on the alpha nodes picked
       sendLookup ps nid uid valL
 
+genericLookup2 valL nid handlerFn = do
+  debug $ "Starting a node lookup for node " ++ show nid
+  uids <- nrandoms (alpha + 1)
+  let (lookupId, msgIds) = splitAt 1 uids
+  ktree <- readKTree
+  me    <- askLocalId
+  routeT <- readRoutingT
+  opsT <- readRunningOpsT
+
+  let (msgs, nopsT, nrouteT) = prepareLookups valL nid handlerFn opsT routeT me ktree (head lookupId) msgIds
+  -- TODO update maps
+  liftIO . mapM (uncurry sendToPeer) $ msgs
+
+prepareLookups valL nid handlerFn opsT routeT me ktree lookupId msgIds = (mps, newOpsT, newRouteT)  
+  where (mps, newRouteT) = foldr (\(mid,p) acc -> 
+          comb (prepareLookup valL nid p routeT me lookupId mid) acc) ([], routeT) (zip msgIds ps)
+        comb (mp, rt) (arr, ort) = (mp:arr, rt)
+        kc = kclosest ktree nid
+        ps = pickAlphaNodes kc
+        newOpsT = M.insert lookupId (RunningLookup nid kc ps [] handlerFn) opsT
+
+prepareLookup valL nid peer routeT me lookupId msgId = ((msg, peer), newRouteT)
+  where msg = buildHeader lookpOp msgId me ++ toCharArray nid 20
+        newRouteT = M.insert msgId (WaitingReply peer replyOp lookupId) routeT
+        lookpOp = if valL then ValueLookupOp else NodeLookupOp
+        replyOp = if valL then ValueLookupReplyOp else NodeLookupReplyOp
+
 -- Stores a key / value pair by doing a node lookup and sending a store command
 -- to the closest nodes found
-store key kdata = nodeLookup key $ PeersHandler (\peers -> (liftIO . putStrLn $ "Sending store to peers " ++ show peers) >> liftIO newUid >>= sendStore peers key kdata)
+store key kdata = nodeLookup key $ PeersHandler (\peers -> (liftIO . putStrLn $ "Sending store to peers " ++ show peers) >> newUid >>= sendStore peers key kdata)
 
 -- Received a node lookup request, queries the KTable for the k closest and
 -- sends the information back.
@@ -138,7 +169,7 @@ storeReceive msgId key val peer = (liftIO . putStrLn $ "local store for " ++ sho
 -- in Vivaldi coordinates.
 pickAlphaNodes kc =  
   let idx = length kc - alpha
-  in snd $ splitAt idx kc
+  in drop idx kc
 
 closestNode pivot p1 p2 = 
   let d1 = nodeId p1 `xor` pivot
@@ -151,3 +182,37 @@ closestNodes pivot = take kdepth . sortBy (closestNode pivot)
 debug s = liftM ((++ " " ++ s) . show . nodeId) askLocalId >>= liftIO . debugM "Kad"
 error s = liftM ((++ " " ++ s) . show . nodeId) askLocalId >>= liftIO . errorM "Kad"
 
+-- Utility functions to serialize messages
+--
+
+buildHeader optype msgId sender = 
+  (chr 1) : optypeStr : toCharArray (toInteger msgId) 8 ++ serPeer sender
+  where optypeStr = case optype of
+                      PingOp              -> chr 1
+                      PingReplyOp         -> chr 2
+                      NodeLookupOp        -> chr 3
+                      NodeLookupReplyOp   -> chr 4
+                      StoreOp             -> chr 5
+                      StoreReplyOp        -> chr 6
+                      ValueLookupOp       -> chr 7
+                      ValueLookupReplyOp  -> chr 8
+
+-- Convert an integer to a string of n bytes
+toCharArray:: Integer -> Int -> [Char]
+toCharArray num depth = map (chr . fromInteger) (toBytes num $ toInteger depth)
+
+-- Converts a string to its numeric value by considering that each character is a
+-- byte in a n byte number
+fromCharArray :: (Bits a) => [Char] -> a
+fromCharArray str = foldl (\acc ch -> shift acc 8 + fromIntegral (ord ch)) 0 str
+
+-- Serializes peers
+serPeers  = concat . map serPeer
+
+-- Decomposes a number into its successive byte values or, said differently, converts
+-- a number to base 2^8.
+toBytes :: Integer -> Integer -> [Integer]
+toBytes num depth = unfoldr byteMod (num,depth-1)
+  where byteMod (num,exp) = let (d,m) = divMod num (2^(exp*8))
+                            in if exp < 0 then Nothing else Just (d, (m,exp-1))
+

src/Main.hs

@@ -10,8 +10,10 @@ import Control.Concurrent(forkIO,threadDelay)
 import Control.Concurrent.STM
 import Control.Monad.Trans(liftIO)
 import Control.Monad(forM)
+import Control.Applicative((<$>))
 
 import System.Log.Logger
+import System.Random
 import Debug.Trace
 
 import KTable
@@ -26,8 +28,10 @@ newNode myPort otherPort = do
   trot <- newTVarIO M.empty
   tkt  <- newTVarIO $ kbleaf S.empty
   ls   <- newTVarIO M.empty
+  rs   <- randomRs (0, 2^64 - 1) <$> newStdGen
+  rsv  <- newTVarIO rs
   let myId = intSHA1 myPort
-  let gd   = GlobalData trt trot tkt (IPPeer "127.0.0.1" (show myPort) myId) ls
+  let gd   = GlobalData trt trot tkt (IPPeer "127.0.0.1" (show myPort) myId) ls rsv
   -- nodeId = sha1 port
   debug $ "Starting on port " ++ show myPort
   forkIO $ runServer gd (start myPort otherPort)