sirkel (empty) → 0.1
raw patch · 6 files changed
+1069/−0 lines, 6 filesdep +SHAdep +basedep +binarysetup-changed
Dependencies added: SHA, base, binary, bytestring, containers, hashtables, haskell98, random, remote, transformers
Files
- LICENSE +24/−0
- README.md +75/−0
- Remote/DHT/Chord.hs +569/−0
- Remote/DHT/DHash.hs +371/−0
- Setup.hs +6/−0
- sirkel.cabal +24/−0
+ LICENSE view
@@ -0,0 +1,24 @@+Copyright (c) 2011, Morten Olsen Lysgaard+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:+ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.+ * Redistributions in binary form must reproduce the above copyright+ notice, this list of conditions and the following disclaimer in the+ documentation and/or other materials provided with the distribution.+ * Neither the name of Morten Olsen Lysgaard, Sirkel nor the+ names of its contributors may be used to endorse or promote products+ derived from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL Morten Olsen Lysgaard BE LIABLE FOR ANY+DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,75 @@+Sirkel is a DHT based on Chord++To use, compile Main.hs with `ghc -threaded --make Main.hs`++afterwards run as many Sirkel instances you want on the network you have.+It only supports LAN or boxes directly connected to the internet yet.++to put data into the DHT write "put " preceded by the data and press enter.++example:+ put abc123++The data will the be saved in the DHT.+All nodes can now "get" the data using the SHA-1 hash of the data.++When you write the `put` the output will be a list of keys. These keys are the SHA-1+hashes of the blocks that your data was chunked intoo before it was put.++to get some data back, just write `get [key1, key2, key3]`+where `key1 ... key3` are the keys that where output from the `put`.+Note that to get your data correctly the order of the keys in the get+matters. Each key represents a block of data. First each of them are retrieved from+the DHT, then they are concatenated together to form the final data. If you mess up the order,+the concatenation will be messed up and therefore also the deserialization.++In Main.hs there is a initState. The fields in this state determines how your DHT works.++ initState = NodeState {+ self = undefined+ , fingerTable = Map.empty -- ^ The fingerTable+ , blockDir = "/tmp/"+ , predecessor = undefined+ , timeout = 10 -- ^ The timout latency of ping+ , m = 160 -- ^ The number of bits in a key, ususaly 160+ , r = 5 -- ^ the number of successors+ , b = 2 -- ^ the nuber of replicas+ , blockSize = 10 -- ^ the number of bytes a block is+ }++ * `self` is a reference to ourselves and should be left undefined, it is populated when you initialize the DHT.+ * `fingerTable` is our "address book". It tells us whom to ask what, and also who most likely to know what.+ * `blockDir` is currently not used since blocks are stored in RAM+ * `predecessor` is a reference to our predecessor in the Chord ring. It should be left empty for the same reason as self.+ * `timeout` is not used yet but will be used to determine how long to way for a reply from the DHT.+ * `m` is the number of bits that makes up the keyspace of the Chord ring. Note that it can not be changed without changing the hashing algorithm which is not supported yet.+ * `r` is one of the most important parameters. It determines how many immediate successors you keep. This has consequences that we'll look on soon.+ * `b` is the number of replicas of each block that should be kept, the node responsible for a block will make shure that its `b` successors has a copy of it's blocks so that in the event of node failure they can take over the role as owners of the block.+ * `blockSize` is the numbers of bytes a block should have. When you try to put data, it will first be chunked into blocks of this size, then each block will be hashed and put separately.++Now, more on the `r` number. Each node keeps `r` successors.+That means that if we are asked for the successor of a key that preceeds+our `r`th successor we know that our `r`th successor is the successor of+that key. We do not know anything about `r`s successors though, because+we only keep a list of the first `r` of us and the `r`th successor is the+last in that list. Therefore, in all calls that "get" something, `findSuccessors`,+`getBlock` and `getObject`, there is a `howMany` argument. This specifies how+many successors we need back. At most we can get `r` successors back since no node+keeps a list of more than that. But say if we only need the first successor of a key.+Then we call `findSuccessors` with `howMany = 1` which lets not just the immediate predecessor+of the key answer our call, but all the predecessors that have at-least one successor to that key.+This boils down to that the `r` nodes before the key can answer. Therefore the argument `howMany`+lets you specify how "exact" your query is. If you choose `howMany = r` then only one node in the+ring can answer that query. For `howMany <= r`, `r + 1 - howMany` nodes can answer your query.+This means that if you don't need that detailed information about the successors of a key, maybe+just the first one, the reliability and speed of your query increases.++This is especially important for small networks. If the `r` parameter is larger than the number+of Sirkel nodes then all nodes know all others and there will be no network traffic to resolve+a query. This off-course comes to a price. You have to store the contact information about `r`+nodes so for a million nodes network `r` must be kept at a reasonable level.++For questions or anything else:+ * Send me an e-mail at morten@lysgaard.no+ * [Projects webpage](mortenlysgaard.com)+ * [Projects Github page](https://github.com/molysgaard/Sirkel)
+ Remote/DHT/Chord.hs view
@@ -0,0 +1,569 @@+{-# LANGUAGE TemplateHaskell,BangPatterns,PatternGuards,DeriveDataTypeable #-}+module Remote.DHT.Chord (+ -- * Types+ NodeId, + NodeState(..),+ FingerTable,+ -- * Initialization+ bootstrap,+ -- * Lookup+ findSuccessors,+ successors,+ successor,+ -- * State+ getState,+ -- * Utility+ between, cNodeId,+ -- * Cloud haskell specific+ __remoteCallMetaData+ ) where++-- {{{ imports+import Remote+import Remote.Process+import Remote.Call++import Data.Typeable+import Data.Binary+import Data.Digest.Pure.SHA++import Control.Monad.IO.Class (liftIO)+import Control.Monad (liftM)+import Control.Concurrent (threadDelay)++import qualified Data.Map as Map+import qualified Data.List as List+import qualified Data.ByteString.Lazy.Char8 as BS++import Maybe (fromJust)+import Data.Int (Int64)+-- }}}++--{{{ Data type and type class definitions++-- | The successor list is from the closest to the farther away+data FingerEntry = SuccessorList [NodeId] | FingerNode NodeId deriving (Eq, Show, Typeable)++instance Binary FingerEntry where+ put (SuccessorList ns) = do put (0 :: Word8)+ put ns+ put (FingerNode n) = do put (1 :: Word8)+ put n+ get = do flag <- getWord8+ case flag of+ 0 -> get >>= (return . SuccessorList)+ 1 -> get >>= (return . FingerNode)++type FingerTable = Map.Map Integer FingerEntry++-- | NodeState, carries all important state for a Chord DHT to function+data NodeState = NodeState {+ self :: NodeId+ , fingerTable :: FingerTable -- ^ The fingerTable+ , blockDir :: FilePath -- ^ The dir to store the blocks, not currently used since everything is stored in a HashTable in memory+ , predecessor :: NodeId -- ^ The node directly before us in the ring+ , timeout :: Int -- ^ The timout latency of ping+ , m :: Integer -- ^ The number of bits in a key, ususaly 160+ , r :: Int -- ^ The number of in the successor list, eg. 16+ , b :: Int -- ^ The number of replicas of each block, (b <= r)+ , blockSize :: Int64 -- ^ The number of bytes each block fills+ } deriving (Show)++instance Binary NodeState where+ put a = do put (self a)+ put (fingerTable a)+ put (blockDir a)+ put (predecessor a)+ put (timeout a)+ put (m a)+ put (r a)+ put (b a)+ put (blockSize a)+ get = do se <- get+ ft <- get+ bd <- get+ pr <- get+ ti <- get+ m <- get+ r <- get+ b <- get+ blockSize <- get+ return (NodeState { self = se, fingerTable = ft, blockDir = bd, predecessor = pr, timeout = ti, m=m, r=r, b=b, blockSize=blockSize })++-- | 'successor' takes the state and tells us who our imidiate+-- successor is.+successor :: NodeState -> Maybe NodeId+successor st+ | null (successors st) = Nothing --error "No successors"+ | otherwise = Just . head . successors $ st++-- | 'successors' takes the state and tells us who+-- our 'r' imidiate successors are in rising order.+-- That is, node number 1 is the closest and node n+-- is the successor farthes away.+successors :: NodeState -> [NodeId]+successors st+ | Just (SuccessorList ns) <- Map.lookup 1 (fingerTable st)+ , length ns /= 0 -- We can't give an empty list back+ = ns+ | otherwise = []++-- | 'cNodeId' takes a 'NodeId' and return the SHA1 hash+-- of it. This is the ID/key of the NodeId.+cNodeId :: NodeId -> Integer+cNodeId n = integerDigest . sha1 $ encode n++data FingerResults = Has [NodeId] | HasNot | Empty+--}}}++-- {{{ algorithmic stuff+-- {{{ hasSuccessor+-- | 'hasSuccessor' tells us if we have the successors of a given key.+hasSuccessor :: NodeState -> Integer -> Int -> FingerResults+hasSuccessor st key howMany+ | List.null ss = Empty+ | between key (cNodeId (self st)) (cNodeId (head ss) + 1)+ = Has ss+ | otherwise = hasSuccessor' st key howMany (tail ss)+ where ss = successors st+++-- | If we have enough successors in our list we return them if they +-- are successors of the key+hasSuccessor' _ _ _ [] = HasNot+hasSuccessor' st key howMany succ@(s:ss)+ | length succ < howMany = HasNot+ | between key (cNodeId (self st)) ((cNodeId s) + 1) = Has succ+ | otherwise = hasSuccessor' st key howMany ss+-- }}}++-- {{{ closestPrecedingNode+-- | 'closestPreceding' is the node that has the 'NodeId' closest to the given key,+-- /but/ not above it that we know of in our 'NodeState'.+closestPreceding :: NodeState -> Integer -> [NodeId]+closestPreceding st key = closestPreceding' st (fingerTable st) key (m st) []++closestPreceding' :: NodeState -> FingerTable -> Integer -> Integer -> [NodeId] -> [NodeId]+closestPreceding' st fingers key 0 ns+ | length ns >= (r st) = ns+ | otherwise = (self st):ns+closestPreceding' st fingers key i ns+ | length ns >= (r st) = ns++ | (Just (SuccessorList hits)) <- lookopAndIf i fingers isBetween+ , length ns < (r st)+ , (length ns) + (length hits) > (r st)+ = closestPreceding' st fingers key (i-1) (nub ((take ((r st) - (length ns)) hits)++ns))-- we have to add a part of the successor list++ | (Just (SuccessorList hits)) <- lookopAndIf i fingers isBetween+ , length ns < r st -- We need more+ , (length ns) + (length hits) > r st+ = closestPreceding' st fingers key (i-1) (nub (hits++ns))-- we take the whole succesor list++ | (Just (FingerNode hit)) <- lookopAndIf i fingers isBetween+ , length ns < r st+ = closestPreceding' st fingers key (i-1) (nub (hit:ns))++ | otherwise = closestPreceding' st fingers key (i-1) ns+ where isBetween (FingerNode x) = between (cNodeId x) (cNodeId . self $ st) key -- isBetween is from the fingertable+ isBetween (SuccessorList []) = False+ isBetween (SuccessorList x) = between (cNodeId . head $ x) (cNodeId . self $ st) key -- isBetween is from the fingertable+ nub = (map head) . List.group+++-- | Lookup a value, if a predicate on it is true, return it, else Nothing+lookopAndIf k m f+ | (Just a) <- Map.lookup k m+ , f a = Just a -- a is the node from the fingertable+ | otherwise = Nothing+-- }}}++-- {{{ between+-- | Is n in the domain of (a, b) ?+--+--+-- NB: This domain is closed. It's the same as+-- a < n < b just it's circular as Chords keyspace.+between :: Integer -> Integer -> Integer -> Bool+between n a b+ | a == b = n /= a --error "n can't be between a and b when a == b" -- can't be alike+ | (a < b) = (a < n) && (n < b)+ | (b < a) = not $ between n (b-1) (a+1)+-- }}}++-- {{{ addFinger+-- | Adds a finger to the fingertable. If there already exists a+-- finger they are compared to see who's the best fit.+addFinger :: NodeId -> NodeState -> NodeState+addFinger newFinger st + | newFinger == (self st) = st+ | otherwise = st {fingerTable = List.foldl' pred (fingerTable st) [1..(m st)]}+ where pred :: FingerTable -> Integer -> FingerTable+ pred ft 1+ | Just (SuccessorList ns) <- Map.lookup 1 ft+ = Map.insert 1 (addToSuccessorList st newFinger [] ns) ft+ | otherwise = Map.insert 1 (addToSuccessorList st newFinger [] []) ft++ pred ft i+ | Just (FingerNode prevFinger) <- Map.lookup i ft -- there exists a node in the fingertable, is the new one closer?+ , let fv = fingerVal st i in (between c fv (cNodeId prevFinger)) && (between c fv n)+ = Map.insert i (FingerNode newFinger) ft++ | Nothing <- Map.lookup i ft -- there is no node, we just put it in if it fits+ , let fv = fingerVal st i in (between c fv n)+ = Map.insert i (FingerNode newFinger) ft+ | otherwise = ft++ c = cNodeId newFinger+ n = cNodeId (self st)++addToSuccessorList :: NodeState -> NodeId -> [NodeId] -> [NodeId] -> FingerEntry+addToSuccessorList st node prev [] = SuccessorList . (take (r st)) . nub $ (prev ++ [node])+ where nub = (map head) . List.group+addToSuccessorList st node prev (cur:next)+ | between (cNodeId node) (cNodeId . self $ st) (cNodeId cur) = SuccessorList . (take (r st)) . nub $ prev ++ [node] ++ next+ | otherwise = addToSuccessorList st node (prev ++ [cur]) next+ where nub = (map head) . List.group++addFingers :: [NodeId] -> NodeState -> NodeState+addFingers ns st = List.foldl' (\st n -> addFinger n st) st ns++fingerVal :: (Integral a) => NodeState -> a -> Integer+fingerVal st k = mod ((cNodeId . self $ st) + 2^(k-1)) (2^(m st))+-- }}}++-- {{{ removeFinger+-- 'removeFinger' takes a 'NodeId' and a 'NodeState' and removes all+-- occurences of it in the 'NodeState'. This is used when a node+-- leaves or times out.+removeFinger :: NodeId -> NodeState -> NodeState+removeFinger node st+ | node == (self st) = st+ | otherwise = remSuccList $ st { fingerTable = Map.filter (/= FingerNode node) (fingerTable st) }+ where remSuccList :: NodeState -> NodeState+ remSuccList st'+ | Just (SuccessorList ns) <- Map.lookup 1 (fingerTable st')+ = st' { fingerTable = Map.insert 1 (SuccessorList (List.filter (/= node) ns)) (fingerTable st') }+ | otherwise = st'+-- }}}+-- }}}++-- {{{ State stuff+-- | 'PassState' let's us pass state around between the+-- 'handleState' process and all the others.+data PassState = ReadState ProcessId+ | RetReadState NodeState+ | TakeState ProcessId+ | RetTakeState NodeState+ | PutState NodeState deriving (Show, Typeable)++instance Binary PassState where+ put (ReadState pid) = do put (0 :: Word8)+ put pid+ put (RetReadState i) = do put (1 :: Word8)+ put i++ put (TakeState pid) = do put (2 :: Word8)+ put pid+ put (RetTakeState i) = do put (3 :: Word8)+ put i+ put (PutState i) = do put (4 :: Word8)+ put i+ get = do+ flag <- getWord8+ case flag of+ 0 -> do+ pid <- get+ return (ReadState pid)+ 1 -> do+ i <- get+ return (RetReadState i)+ 2 -> do+ pid <- get+ return (TakeState pid)+ 3 -> do+ i <- get+ return (RetTakeState i)+ 4 -> do+ i <- get+ return (PutState i)++-- | 'getState' lets us retrive the current state from the+-- 'handleState' process.+getState :: ProcessM NodeState+getState = do statePid <- getStatePid+ pid <- getSelfPid+ send statePid (ReadState pid)+ ret <- receiveTimeout 10000000 [ match (\(RetReadState st) -> return st) ]+ case ret of+ Nothing -> say "asked for state but state process did not return within timeout, retrying" >> getState+ Just st -> return st++-- | 'modifyState' takes a 'NodeState'-modifying function and+-- runs it on the current state updating it.+modifyState :: (NodeState -> NodeState) -> ProcessM NodeState+modifyState f = do+ statePid <- getStatePid+ selfPid <- getSelfPid+ send statePid (TakeState selfPid)+ ret <- receiveTimeout 10000000 [ match (\(RetTakeState st) -> return st) ]+ case ret of+ Nothing -> say "asked for modify, timeout, retrying" >> liftIO (threadDelay 50000000) >> modifyState f+ Just st -> send statePid (PutState (f st)) >> return (f st)++-- | 'getStatePid' gives us the 'ProcessId' of the 'handleState' process.+getStatePid :: ProcessM ProcessId+getStatePid = do nid <- getSelfNode+ statePid <- nameQuery nid "CHORD-NODE-STATE"+ case statePid of+ Nothing -> say "State not initialized, state-process is not running" >> getStatePid+ Just pid -> return pid+-- }}}++-- {{{ relayFndSucc+-- | internal function: 'relayFndSucc' is called when a node does not know+-- the answer to a 'findSuccessors' call. For each call, we halve+-- the distance to the answer.+relayFndSucc :: NodeId -> ProcessId -> Integer -> Int -> ProcessM ()+relayFndSucc nid caller key howMany = do+ modifyState (\x -> addFinger (nodeFromPid caller) (addFinger nid x))+ st <- getState+ case (hasSuccessor st key howMany) of+ (Has suc) -> do + send caller suc -- we have the successor of the node+ HasNot -> do+ let recv = last $ closestPreceding st key -- find the next to relay to+ case recv == (self st) of+ False -> do+ let clos = $( mkClosureRec 'relayFndSucc )+ flag <- ptry $ spawn recv (clos (self st) caller key howMany) :: ProcessM (Either TransmitException ProcessId)+ case flag of+ Left _ -> say "could not spawn" >> modifyState (removeFinger recv) >> relayFndSucc nid caller key howMany -- spawning failed+ Right _ -> return ()+ True -> do+ say "THIS IS WRONG!" -- this should never happen because we should not be in the fingertable+ send caller (self st)+ Empty -> do+ send caller (self st)++getPred = do st <- getState+ return (predecessor st)+-- }}}++-- {{{ notify+-- | notifier is telling you he thinks he is your predecessor, check if it's true.+-- and update the 'NodeState' accordingly.+notify :: NodeId -> ProcessM ()+notify notifier = do+ st <- getState+ if between (cNodeId notifier) (cNodeId . predecessor $ st) (cNodeId . self $ st)+ then say "New predecessor" >> modifyState (\x -> x {predecessor = notifier}) >> return ()+ else return ()+-- }}}++-- {{{ ping+ping pid = send pid pid+-- }}}++-- | debug function, makes a remote node say a string.+remoteSay str = say str++$( remotable ['relayFndSucc, 'getPred, 'notify, 'ping, 'remoteSay] )++-- {{{ joinChord+-- | Joins a chord ring. Takes the id of a known node to bootstrap from.+-- This is not the function you would use to start a Chord Node, for that+-- use 'bootstrap'.+joinChord :: NodeId -> ProcessM ()+joinChord node = do+ st <- modifyState (addFinger node)+ say $ "Join on: " ++ (show node)+ succ <- liftM (head . fromJust) $ remoteFindSuccessor node (mod ((cNodeId . self $ st) + 1) (m $ st)) (r st)+ say $ "Ret self?: " ++ (show (succ == (self st))) ++ " Ret boot?: " ++ (show (succ == node))+ buildFingers succ+ sst <- getState+ --say $ "Finish join: " ++ (show . nils . fingerTable $ sst)+ let suc = successor sst+ case suc of+ (Just c) -> do ptry (spawn c (notify__closure (self st))) :: ProcessM (Either TransmitException ProcessId)+ return ()+ Nothing -> say "joining got us no successor!" >> return ()+-- }}}++-- {{{ checkAlive+-- | 'checkAlive' checks if a single node is alive, if it's+-- not we'll discard it from our 'fingerTable'.+checkAlive node = do pid <- getSelfPid+ flag <- ptry $ spawn node (ping__closure pid) :: ProcessM (Either TransmitException ProcessId)+ case flag of+ Left _ -> say "dropped node" >> modifyState (removeFinger node) >> return False+ Right _ -> do resp <- receiveTimeout 10000000 [match (\x -> return x)] :: ProcessM (Maybe ProcessId)+ case resp of+ Nothing -> do modifyState (removeFinger node)+ say "dropped node"+ return False+ Just pid -> return True+-- }}}++-- {{{ fingerNodes+-- | Utility function that takes a 'NodeState' and+-- returns all the 'NodeId's that we know+fingerNodes :: NodeState -> [NodeId]+fingerNodes st+ | Just (SuccessorList sl) <- Map.lookup 1 (fingerTable st)+ , fs <- (drop 1) . Map.elems $ (fingerTable st) :: [FingerEntry]+ = nub $ sl ++ (map strip fs)+ | otherwise = []+ where nub = (map head) . List.group+ strip (FingerNode n) = n+-- }}}++-- {{{ checkFingerTable+-- | 'checkFingerTable' checks if the nodes in our 'fingerTable' is alive+-- and discards them if they're not+checkFingerTable :: ProcessM ()+checkFingerTable = do st <- getState+ sequence $ List.map checkAlive (fingerNodes st)+ return ()+-- }}}++-- {{{ checkPred+-- | 'checkPred' checks if our 'predecessor' is alive.+checkPred :: ProcessM Bool+checkPred = do st <- getState+ flag <- checkAlive (predecessor st)+ if flag+ then return True+ else modifyState (\x -> x { predecessor = (self x) }) >> return False+-- }}}++-- {{{ stabilize+-- | This is run periodically to check if our fingertable is correct+-- It is the one that handles node failures, leaves and joins.+-- It's an internal function+stabilize = do+ liftIO $ threadDelay 5000000 -- 5 sec+ checkPred + checkFingerTable+ st <- getState+ case successor st of+ (Just succ) -> do alive <- checkAlive succ+ if alive+ then do succPred <- callRemote succ getPred__closure+ if between (cNodeId succPred) (cNodeId . self $ st) (cNodeId succ)+ then do modifyState (addFinger succPred)+ ptry $ spawn succ (notify__closure (self st)) :: ProcessM (Either TransmitException ProcessId)+ say ("New succ: " ++ (show succPred)) >> stabilize+ else do ptry (spawn succ (notify__closure (self st))) :: ProcessM (Either TransmitException ProcessId)+ stabilize+ else do say "Successor is dead, restabilizing"+ findSuccessors (mod ((cNodeId . self $ st) + 1) (2^(m st))) (r st)+ stabilize+ Nothing -> stabilize+-- }}}++-- {{{ findSuccessors+-- | This is the main function of the Chord DHT. It lets you lookup+-- what 'NodeId's has the responsebility for the given key+findSuccessors :: Integer -> Int -> ProcessM [NodeId]+findSuccessors key howMany = do+ st <- getState+ case (hasSuccessor st key howMany) of+ (Has suc) -> return suc+ Empty -> return [self st]+ HasNot -> do+ selfPid <- getSelfPid+ let recv = last $ closestPreceding st key+ case recv == (self st) of+ False -> do ret <- remoteFindSuccessor recv key howMany+ case ret of + Nothing -> modifyState (removeFinger recv) >> findSuccessors key howMany+ Just succ -> return succ+ True -> say "THIS IS WRONG, we should not be in our own fingertable! retrying" >> liftIO (threadDelay 5000000) >> findSuccessors key howMany+-- }}}++-- {{{ remoteFindSuccessor+-- | 'remoteFindSuccessor' takes a 'NodeId' the key to lookup and a number of how+-- many successors to get.+remoteFindSuccessor :: NodeId -> Integer -> Int -> ProcessM (Maybe [NodeId])+remoteFindSuccessor node key howMany = remoteFindSuccessor' node key howMany 2+remoteFindSuccessor' :: NodeId -> Integer -> Int -> Int -> ProcessM (Maybe [NodeId])+remoteFindSuccessor' _ _ _ 0 = return Nothing+remoteFindSuccessor' node key howMany tries = do+ st <- getState+ selfPid <- getSelfPid+ ptry $ spawn node (relayFndSucc__closure (self st) selfPid (key :: Integer) (howMany :: Int)) :: ProcessM (Either TransmitException ProcessId)+ -- maybe it should be checked for errors in the ptry, if the error is local something will not work?+ succ <- receiveTimeout 10000000 [match (\x -> return x)] :: ProcessM (Maybe [NodeId])+ case succ of+ Nothing -> say "RemFndSucc timed out, retrying" >> remoteFindSuccessor' node (key :: Integer) (howMany :: Int) (tries - 1)+ Just conts -> do+ modifyState (addFingers conts)+ return (Just conts)+-- }}}++-- {{{ buildFingers+-- |'buildFingers' takes a 'NodeId' and does a lookup+-- for sucessors to each index in our fingertable+buildFingers :: NodeId -> ProcessM ()+buildFingers buildNode = do+ say $ "buildNode is: " ++ (show buildNode)+ st <- getState+ nodeId <- getSelfNode+ let f :: Integer -> ProcessM (Maybe [NodeId])+ f i = remoteFindSuccessor buildNode i (r st)+ nodids = map (fingerVal st) nums+ nums = [1 .. (m $ st)]+ mapM_ f nodids+-- }}}++-- {{{ bootstrap+-- | Starts a Chord Node. It takes the initial state,+-- lookups other Chord-nodes+-- on the LAN, starts state handeling, stabilizing etc.+-- in the background and then runs 'joinChord' on the first+-- and best node that's altready a member in the ring.+bootstrap st = do+ selfN <- getSelfNode+ let st' = st { self = selfN, predecessor = selfN }+ peers <- getPeers+ let peers' = filter (/= selfN) $ findPeerByRole peers "NODE" -- remove ourselves from peerlist+ case length peers' >= 1 of+ True -> do+ spawnLocal $ handleState st'+ spawnLocal (joinChord (head peers'))+ spawnLocal stabilize+ + False -> do spawnLocal $ handleState (addFinger (self st') st')+ spawnLocal stabilize+-- }}}++-- {{{ handleState+-- | This is an internal function, it handles gets and puts for the state+-- this function is butt ugly and needs some hefty sugar+handleState st' = do+ nameSet "CHORD-NODE-STATE"+ loop st'++ where loop :: NodeState -> ProcessM ()+ loop st = do+ receiveWait+ [ matchIf (\x -> case x of+ (ReadState _) -> True+ _ -> False) (\(ReadState pid) -> getSt st pid)+ , matchIf (\x -> case x of+ (TakeState _) -> True+ _ -> False) (\(TakeState pid) -> modSt st pid) ]+ >>= loop+ getSt st pid = do+ send pid (RetReadState st)+ return st+ modSt st pid = do+ send pid (RetTakeState st)+ ret <- receiveTimeout 1000000 [ matchIf (\x -> case x of+ (PutState _) -> True+ _ -> False) (\(PutState st) -> return st) ]+ case ret of+ Nothing -> say "process asked for modify, but did not return new state" >> return st+ Just newSt -> return newSt+-- }}}
+ Remote/DHT/DHash.hs view
@@ -0,0 +1,371 @@+{-# LANGUAGE TemplateHaskell,BangPatterns,PatternGuards,DeriveDataTypeable #-}+module Remote.DHT.DHash (+ -- * Initialization+ initBlockStore,+ -- * Put/Get/Delete+ putObject,+ getObject,+ deleteBlock,+ -- * Utility+ encBlock,+ -- * Cloud haskell specific+ Remote.DHT.DHash.__remoteCallMetaData+ ) where++--TODO A block is put on node A and replicated on node B.+-- Then node B leaves.+-- Then the next node in the ring, node C, does not recieve a replicate command.+-- Only if node A leaves and there exits replicas will the replicas be correctly reinserted++import Remote+import Remote.Process+{--+import Remote.Call+import Remote.Channel+import Remote.Peer+import Remote.Init+import Remote.Encoding+import Remote.Reg+--}++import Control.Monad (liftM)+import Data.Typeable+import Control.Monad.IO.Class (liftIO)++import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar+import qualified Control.Exception as Ex++import qualified Data.Map as Map+import Data.List (foldl')+import Data.Maybe++import Data.Digest.Pure.SHA+import Data.Binary+import qualified Data.ByteString.Lazy.Char8 as BS++import qualified Data.HashTable.IO as HT+import Control.Monad.ST++import Remote.DHT.Chord++-- {{{ Block+-- | 'Block' lets us send blocks back when somone asks for one.+data Block = BlockError | BlockFound BS.ByteString deriving (Show, Typeable)+instance Binary Block where+ put BlockError = put (0 :: Word8)+ put (BlockFound bs) = do put (1 :: Word8)+ put bs+ get = do flag <- getWord8+ case flag of+ 0 -> return BlockError+ 1 -> do bs <- get+ return $ BlockFound bs+-- }}}++-- {{{ encBlock+-- | 'encBlock' takes a 'BS.ByteString' and returns the ID/key+-- of that block. This is for 'BS.ByteString's what 'cNodeId' is for 'NodeId's+encBlock :: BS.ByteString -> Integer +encBlock n = integerDigest . sha1 $ n+-- }}}++$( remotable [] )++-- {{{ getBlock+-- | 'getBlock' retrieves a block with a given ID/key.+getBlock :: Integer -> Int -> ProcessM (Maybe BS.ByteString)+getBlock key howMany = do+ succ <- findSuccessors key howMany+ getBlock' key howMany succ++-- getBlock' gets a block from a node we know has it+-- | internal function for 'getBlock'+getBlock' :: Integer -> Int -> [NodeId] -> ProcessM (Maybe BS.ByteString)+getBlock' _ _ [] = return Nothing+getBlock' key howMany (s:su) = do+ ret <- getBlockPid s+ case ret of+ Just blockPid -> do+ selfPid <- getSelfPid+ flag <- ptry $ send blockPid (Lookup key selfPid) :: ProcessM (Either TransmitException ())+ block <- receiveTimeout 10000000 [match (\x -> return x)] :: ProcessM (Maybe Block)+ case block of+ Nothing -> say "GetBlock timed out, retrying" >> liftIO (threadDelay 5000000) >> getBlock key howMany+ Just BlockError -> say "Block error" >> getBlock' key howMany su+ Just (BlockFound bs) -> if encBlock bs == key+ then return (Just bs)+ else return Nothing+ Nothing -> say "GetBlock timed out, retrying" >> liftIO (threadDelay 5000000) >> getBlock key howMany+-- }}}++-- {{{ putBlock+-- | 'putBlock', puts a block, returns the successor of that block. You will+-- also find the block replicated on the (r st) next nodes, but it is the node+-- responsible for the block that is responsible for delivering the block to the+-- replicators.+putBlock :: BS.ByteString -> ProcessM (Integer, NodeId)+putBlock bs = do+ let key = encBlock bs+ succs <- findSuccessors key 1+ putBlock' bs key (head succs)++-- | 'putBlock'' put a block on a node we know+putBlock' :: BS.ByteString -> Integer -> NodeId -> ProcessM (Integer, NodeId)+putBlock' bs key succ = do+ ret <- getBlockPid succ+ case ret of+ Just pid -> do+ flag <- ptry $ send pid (Insert bs) :: ProcessM (Either TransmitException ())+ case flag of+ Left _ -> say "put block failed, retrying" >> (liftIO (threadDelay 5000000)) >> putBlock bs+ Right _ -> return (key, succ)+ Nothing -> say "put block failed, retrying" >> (liftIO (threadDelay 5000000)) >> putBlock bs+-- }}}++-- {{{ deleteBlock+-- | 'deleteBlock' takes the ID/key of the block to delete+-- and sends a message to the node responsible for that block.+-- The message then propagates from the responsible to the+-- replicators.+deleteBlock :: Integer -> ProcessM Bool+deleteBlock key = do+ succs <- findSuccessors key 1+ ret <- getBlockPid (head succs)+ case ret of+ Just blockPid -> do+ flag <- ptry $ send blockPid (Delete key) :: ProcessM (Either TransmitException ())+ case flag of+ Left _ -> say "Delete failed" >> return False+ Right _ -> say "Delete sent" >> return True+ Nothing -> say "Delete failed" >> return False+-- }}}++-- {{{ getBlockPid+-- | 'getBlockPid' gets the 'ProcessId' for the 'initBlockStore' process.+getBlockPid :: NodeId -> ProcessM (Maybe ProcessId)+getBlockPid node = do + statePid <- ptry $ nameQuery node "DHASH-BLOCK-STORE" :: ProcessM (Either ServiceException (Maybe ProcessId))+ case statePid of+ Right (Just pid) -> return (Just pid)+ _ -> say "Dhash block store not initialized, state-process is not running" >> return Nothing+-- }}}++-- {{{ Datatypes+-- | 'Dhash' is a datatype encapsulating the things we can do with the HashTable+data DHash = Insert BS.ByteString | Lookup Integer ProcessId | Delete Integer | Janitor deriving (Eq, Show, Typeable)+instance Binary DHash where+ put (Insert a) = do put (0 :: Word8)+ put a+ put (Lookup key pid) = do put (1 :: Word8)+ put key+ put pid+ put (Delete key) = do put (2 :: Word8)+ put key+ put Janitor = do put (3 :: Word8)+ get = do flag <- getWord8+ case flag of+ 0 -> do val <- get+ return $ Insert val+ 1 -> do key <- get+ pid <- get+ return $ Lookup key pid+ 2 -> do key <- get+ return $ Delete key+ 3 -> return Janitor++-- | 'DHashTable' is the datastructure used to store all blocks.+-- This is designed so that in the future we can extend it to +-- other storage systems, eg. filesystem.+type DHashTable = HT.LinearHashTable Integer (Bool,BS.ByteString)+-- }}}++-- {{{ sendBlock+-- | sendBlock is a function to send a block from a lookup in the HashTable.+-- We do this in a separate thread because we don't want to block lookups etc.+-- while we are sending.+-- TODO there have to be some sort of queue here in the future, to limit the+-- upload+sendBlock :: Maybe (Bool, BS.ByteString) -> ProcessId -> ProcessM ()+sendBlock Nothing pid = do+ ptry (send pid BlockError) :: ProcessM (Either TransmitException ())+ return ()+sendBlock (Just (_, bs)) pid = do+ ptry (send pid (BlockFound bs)) :: ProcessM (Either TransmitException ())+ return ()+-- }}}++-- {{{ initBlockStore+-- | initBlockStore starts the BlockStore and handles requests to+-- insert, lookup and delete blocks as well as the janitor process+-- to check if ownership of any block has changed+initBlockStore :: DHashTable -> ProcessM ()+initBlockStore ht' = do+ nameSet "DHASH-BLOCK-STORE"+ spawnLocal janitorSceduler+ loop ht'+ where loop :: DHashTable -> ProcessM ()+ loop ht = do+ newHt <- receiveWait+ [ matchIf (\x -> case x of+ (Insert _) -> True+ _ -> False)+ (\(Insert val) -> insertBlock ht val)+ , matchIf (\x -> case x of+ (Lookup _ _) -> True+ _ -> False)+ (\(Lookup key pid) -> lookupBlock key pid ht)+ , matchIf (\x -> case x of+ Janitor -> True+ _ -> False)+ (\Janitor -> janitor ht)+ , matchIf (\x -> case x of+ (Delete _) -> True+ _ -> False)+ (\(Delete key) -> removeBlock ht key) ]+ loop newHt+-- }}}++-- {{{ lookupBlock+-- | looks in the hashtable for the block and returns it if it finds it.+lookupBlock :: Integer -> ProcessId -> DHashTable -> ProcessM DHashTable+lookupBlock key pid ht = do answ <- liftIO $ HT.lookup ht key+ spawnLocal (sendBlock answ pid)+ return ht+-- }}}++-- {{{ removeBlock+-- | removes a block from the hash table if it exists there.+removeBlock :: DHashTable -> Integer -> ProcessM DHashTable+removeBlock ht key = do + st <- getState+ -- if we are the owner of the block, also send delete+ -- to all the replicas+ -- else just delete our copy+ if between key (cNodeId . predecessor $ st) (cNodeId . self $ st)+ then do liftIO $ HT.delete ht key+ bs <- liftM catMaybes $ mapM getBlockPid (successors st)+ say "deleting replicas"+ mapM_ ((flip send) (Delete key)) bs+ return ht+ else do liftIO $ HT.delete ht key+ return ht+-- }}}++-- {{{ insertBlock+-- | Inserts a block to the 'DHashTable'. It also checks if we+-- are the node responsible for the block or just a replicator.+insertBlock :: DHashTable -> BS.ByteString -> ProcessM DHashTable+insertBlock ht val = do + st <- getState+ -- if the block is to big, we won't store it because somethings wrong+ if BS.null (BS.drop (blockSize st) val)+ then do+ let key = encBlock val+ -- if we are the right owner of this block+ -- or if we are just replicating+ -- TODO would be smart to check if we should be+ -- replicating but it is not trivial without a predecessor list+ -- wich is not implemented at this time+ if between key (cNodeId . predecessor $ st) (cNodeId . self $ st)+ then do liftIO $ HT.insert ht key (True, val)+ mapM_ (putBlock' val key) ((take (b st)) . successors $ st)+ return ht+ else do liftIO $ HT.insert ht key (False, val)+ say "replicating"+ return ht+ else return ht+-- }}}++-- {{{ janitor+-- | janitor takes the DHashTable and checks if the ownership of blocks+-- has changed sice last time. If so it updates replication etc.+janitor :: DHashTable -> ProcessM DHashTable+janitor ht = do ns <- liftIO $ HT.toList ht+ st <- getState+ ns' <- mapM (fix st) ns+ liftIO $ HT.fromList ns'+-- }}}++-- {{{ janitorSceduler+-- | janitorSceduler is a process that periodically sends a "janitor"+-- message to the block manager, this because we don't have MVars in+-- CloudHaskell+janitorSceduler :: ProcessM ()+janitorSceduler = do+ self <- getSelfNode+ ret <- getBlockPid self+ case ret of+ Just pid -> loop pid+ Nothing -> say "janitoring cant start before block store, retrying" >> liftIO (threadDelay 50000000) >> janitorSceduler+ where loop pid = do+ liftIO (threadDelay 5000000)+ send pid Janitor+ loop pid+-- }}}++-- {{{ fix+-- | fix function that looks on one block in the HashTable+-- and checks if ownership hash changed.+fix :: NodeState -> (Integer, (Bool,BS.ByteString)) -> ProcessM (Integer, (Bool,BS.ByteString))+fix st entry@(key, (True, bs))+ | between key (cNodeId . predecessor $ st) (cNodeId . self $ st)+ = return entry+ | otherwise = do + say "we are no longer responsible for this block"+ putBlock bs+ return (key,(False,bs))++fix st entry@(key, (False, bs))+ | between key (cNodeId . predecessor $ st) (cNodeId . self $ st)+ = do + say "we are the new block owner"+ mapM_ (putBlock' bs key) (successors st)+ return (key,(True,bs))+ | otherwise = return entry+-- }}}++-- {{{ chunkBs+-- | 'chunkBs' splits a 'BS.ByteString' into parts that+-- each has the size of 'blockSize' bytes or less.+chunkBs :: NodeState -> BS.ByteString -> [BS.ByteString]+chunkBs st bs+ | BS.null bs = []+ | otherwise = let (pre, post) = BS.splitAt (blockSize st) bs+ in pre : chunkBs st post+-- }}}++-- {{{ putObject+-- | 'putObject' is the main function of the DHash module.+-- It lets you put an arbitrary object that an instance of+-- Binary into the DHT. To retrieve it again, you'll have to call+-- 'getObject' with the list of IDs/keys that this function returns.+-- NB: The order of the IDs/keys matters. This is because an object is+-- chunked. Then each chunk is stored seperatly. When one calls 'getObject'+-- the blocks retrieved is concated in the order of the IDs/keys. +-- That means you'll get garbage if you mess up the order.+putObject :: (Binary a) => a -> ProcessM [(Integer, NodeId)]+putObject a = do st <- getState+ let bs = (chunkBs st) . encode $ a+ mapM putBlock bs+-- }}}++-- {{{ getObject+-- | 'getObject' takes a list of IDs/keys that represent an object that's+-- already been put with 'putObject'.+-- NB: The order of the IDs/keys matters. This is because an object is+-- chunked. Then each chunk is stored seperatly. When one calls 'getObject'+-- the blocks retrieved is concated in the order of the IDs/keys. +-- That means you'll get garbage if you mess up the order.+getObject :: (Binary a) => [Integer] -> Int -> ProcessM (Maybe a)+getObject keys howMany = liftM (liftM decode) $ liftM maybeConcatBS $ mapM (\k -> getBlock k howMany) keys++-- | 'maybeConcatBS' takes '[Maybe BS.ByteString]'s and concats+-- all the bytestrings into one if none of them are Nothing.+-- Else it returns Nothing+maybeConcatBS :: [Maybe BS.ByteString] -> Maybe BS.ByteString+maybeConcatBS blocks+ | any (== Nothing) blocks = Nothing+ | otherwise = Just . BS.concat . catMaybes $ blocks++-- }}}
+ Setup.hs view
@@ -0,0 +1,6 @@+module Main (main) where++import Distribution.Simple (defaultMain)++main :: IO ()+main = defaultMain
+ sirkel.cabal view
@@ -0,0 +1,24 @@+Name: sirkel+Version: 0.1+Cabal-Version: >=1.6+Description: An implementation of the Chord DHT with replication and faulth tolerance+synopsis: Sirkel, a Chord DHT+Category: Distributed Computing,Concurrency,Concurrent,Data Structures,Database+License: BSD3+License-file: LICENSE+Extra-Source-Files: README.md+Author: Morten Olsen Lysgaard <morten@lysgaard.no>+Maintainer: Morten Olsen Lysgaard <morten@lysgaard.no>+Stability: Experimental+Build-Type: Simple+tested-with: GHC == 6.12.3++library+ Build-Depends: base >= 4 && < 5, haskell98, random, bytestring, binary, containers, transformers, hashtables, remote, SHA+ ghc-options: -Wall+ Extensions: TemplateHaskell, DeriveDataTypeable+ Exposed-Modules: Remote.DHT.Chord, Remote.DHT.DHash++source-repository head+ type: git+ location: git://github.com/molysgaard/Sirkel.git