diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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) diff --git a/Remote/DHT/Chord.hs b/Remote/DHT/Chord.hs new file mode 100644 --- /dev/null +++ b/Remote/DHT/Chord.hs @@ -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 +-- }}} diff --git a/Remote/DHT/DHash.hs b/Remote/DHT/DHash.hs new file mode 100644 --- /dev/null +++ b/Remote/DHT/DHash.hs @@ -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 + +-- }}} diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,6 @@ +module Main (main) where + +import Distribution.Simple (defaultMain) + +main :: IO () +main = defaultMain diff --git a/sirkel.cabal b/sirkel.cabal new file mode 100644 --- /dev/null +++ b/sirkel.cabal @@ -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 +Maintainer: Morten Olsen Lysgaard +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