crdt 8.0 → 9.0
raw patch · 9 files changed
+260/−151 lines, 9 filesdep +hashablePVP ok
version bump matches the API change (PVP)
Dependencies added: hashable
API changes (from Hackage documentation)
+ CRDT.LamportClock: getTimes :: Clock m => Natural -> m LamportTime
Files
- crdt.cabal +4/−3
- lib/CRDT/Cv/RGA.hs +29/−30
- lib/CRDT/LamportClock.hs +36/−17
- lib/CRDT/LamportClock/Simulation.hs +17/−10
- test/Cm/RGA.hs +130/−0
- test/Cv/RGA.hs +31/−4
- test/LwwElementSet.hs +7/−2
- test/RGA.hs +0/−85
- test/Util.hs +6/−0
crdt.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: f6c5e8a56fae4a26b32be8d375082402b27412d45fd5f75b9d227c7f77374a7c+-- hash: 6b85645aa899cb940c1bf1caed7e902da66f4dd6c35c35fad4c33443f5606138 name: crdt-version: 8.0+version: 9.0 synopsis: Conflict-free replicated data types description: Definitions of CmRDT and CvRDT. Implementations for some classic CRDTs. category: Distributed Systems@@ -31,6 +31,7 @@ , binary , bytestring , containers+ , hashable , mtl , network-info , safe@@ -82,6 +83,7 @@ other-modules: ArbitraryOrphans Cm.ORSet+ Cm.RGA Cm.TwoPSet Counter Cv.ORSet@@ -94,7 +96,6 @@ LwwElementSet Max PNCounter- RGA Util Paths_crdt default-language: Haskell2010
lib/CRDT/Cv/RGA.hs view
@@ -1,6 +1,5 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE ParallelListComp #-}-{-# LANGUAGE TupleSections #-} module CRDT.Cv.RGA ( RGA (..)@@ -16,13 +15,14 @@ , unpack ) where -import Data.Algorithm.Diff (Diff (Both, First, Second), getDiffBy)+import Data.Algorithm.Diff (Diff (Both, First, Second),+ getGroupedDiffBy) import Data.Function (on) import Data.Semigroup (Semigroup, (<>)) import Data.Semilattice (Semilattice) import Data.Traversable (for) -import CRDT.LamportClock (Clock, LamportTime (LamportTime), getTime)+import CRDT.LamportClock (Clock, LamportTime (LamportTime), getTimes) type VertexId = LamportTime @@ -33,20 +33,19 @@ type RgaString = RGA Char merge :: Eq a => RGA a -> RGA a -> RGA a-merge (RGA vertices1) (RGA vertices2) =- RGA $ mergeVertexLists vertices1 vertices2+merge (RGA vertices1) (RGA vertices2) = RGA+ $ mergeVertexLists vertices1 vertices2 where- mergeVertexLists [] vs2 = vs2- mergeVertexLists vs1 [] = vs1- mergeVertexLists (v1@(id1, a1) : vs1) (v2@(id2, a2) : vs2) =+ mergeVertexLists [] vs2 = vs2+ mergeVertexLists vs1 [] = vs1+ mergeVertexLists (v1@(id1, a1):vs1) (v2@(id2, a2):vs2) = case compare id1 id2 of- LT -> v2 : mergeVertexLists (v1:vs1) vs2- GT -> v1 : mergeVertexLists vs1 (v2:vs2)- EQ -> (id1, mergeAtoms a1 a2) : mergeVertexLists vs1 vs2+ LT -> v2 : mergeVertexLists (v1 : vs1) vs2+ GT -> v1 : mergeVertexLists vs1 (v2 : vs2)+ EQ -> (id1, mergeAtoms a1 a2) : mergeVertexLists vs1 vs2 - mergeAtoms (Just a1) (Just a2)- | a1 == a2 = Just a1- mergeAtoms _ _ = Nothing+ mergeAtoms (Just a1) (Just a2) | a1 == a2 = Just a1+ mergeAtoms _ _ = Nothing instance Eq a => Semigroup (RGA a) where (<>) = merge@@ -59,32 +58,32 @@ mappend = (<>) toList :: RGA a -> [a]-toList (RGA rga) = [a | (_, Just a) <- rga]+toList (RGA rga) = [ a | (_, Just a) <- rga ] toString :: RgaString -> String toString = toList fromList :: Clock m => [a] -> m (RGA a)-fromList = fmap RGA . traverse makeVertex- where- makeVertex a = do- t <- getTime- pure (t, Just a)+fromList = fmap RGA . fromList' . map Just +fromList' :: Clock m => [Maybe a] -> m [(VertexId, Maybe a)]+fromList' xs = do+ LamportTime time0 pid <- getTimes . fromIntegral $ length xs+ pure [ (LamportTime time pid, x) | time <- [time0..] | x <- xs ]+ fromString :: Clock m => String -> m RgaString fromString = fromList -- | Replace content with specified, -- applying changed found by the diff algorithm edit :: (Eq a, Clock m) => [a] -> RGA a -> m (RGA a)-edit newList (RGA oldRga) =- fmap RGA $ for diff $ \case- First (vid, _) -> pure (vid, Nothing)- Both v _ -> pure v- Second (_, a) -> (, a) <$> getTime+edit newList (RGA oldRga) = fmap (RGA . concat) . for diff $ \case+ First removed -> pure [ (vid, Nothing) | (vid, _) <- removed ] -- :: m [(VertexId, Maybe a)]+ Both v _ -> pure v -- :: m [(VertexId, Maybe a)]+ Second added -> fromList' $ map snd added -- :: m [(VertexId, Maybe a)] where- newList' = [(undefined, Just a) | a <- newList]- diff = getDiffBy ((==) `on` snd) oldRga newList' -- TODO(cblp, 2018-02-07) getGroupedDiffBy+ newList' = [ (undefined, Just a) | a <- newList ]+ diff = getGroupedDiffBy ((==) `on` snd) oldRga newList' -- | Compact version of 'RGA'. -- For each 'VertexId', the corresponding sequence of vetices has the same 'Pid'@@ -92,17 +91,17 @@ type RgaPacked a = [(VertexId, [Maybe a])] pack :: RGA a -> RgaPacked a-pack (RGA []) = []+pack (RGA [] ) = [] pack (RGA ((first, atom):vs)) = go first [atom] 1 vs where -- TODO(cblp, 2018-02-08) buf :: DList go vid buf _ [] = [(vid, buf)] go vid buf dt ((wid, a):ws) | wid == next dt vid = go vid (buf ++ [a]) (succ dt) ws- | otherwise = (vid, buf) : go wid [a] 1 ws+ | otherwise = (vid, buf) : go wid [a] 1 ws next dt (LamportTime t p) = LamportTime (t + dt) p unpack :: RgaPacked a -> RGA a unpack packed = RGA $ do (LamportTime time pid, atoms) <- packed- [(LamportTime (time + i) pid, atom) | i <- [0..] | atom <- atoms]+ [ (LamportTime (time + i) pid, atom) | i <- [0..] | atom <- atoms ]
lib/CRDT/LamportClock.hs view
@@ -8,6 +8,7 @@ -- * Lamport timestamp (for a single process) , Clock (..) , LamportTime (..)+ , getTime , LocalTime , Process (..) -- * Real Lamport clock@@ -56,18 +57,33 @@ getMac :: IO MAC getMac = headDef (error "Can't get any non-zero MAC address of this machine")- . filter (/= minBound)- . map mac- <$> getNetworkInterfaces+ . filter (/= minBound)+ . map mac+ <$> getNetworkInterfaces decodeMac :: MAC -> Word64 decodeMac (MAC b5 b4 b3 b2 b1 b0) = decode $ BSL.pack [0, 0, b5, b4, b3, b2, b1, b0] class Process m => Clock m where- getTime :: m LamportTime+ -- | Get sequential timestamps.+ --+ -- Laws:+ -- 1. t1 <- getTimes n+ -- t2 <- getTime+ -- t2 >= t1 + n+ --+ -- 2. getTimes 0 == getTimes 1+ getTimes+ :: Natural -- ^ number of needed timestamps+ -> m LamportTime+ -- ^ Starting value of the range.+ -- So return value @t@ means range @[t .. t + n - 1]@. advance :: LocalTime -> m () +getTime :: Clock m => m LamportTime+getTime = getTimes 1+ -- TODO(cblp, 2018-01-06) benchmark and compare with 'atomicModifyIORef' newtype LamportClock a = LamportClock (ReaderT (TVar LocalTime) IO a) deriving (Applicative, Functor, Monad, MonadIO)@@ -83,17 +99,20 @@ timeVar <- ask lift $ atomically $ modifyTVar' timeVar $ max time - getTime = LamportClock $ do- timeVar <- ask- lift $ do- realTime <- getRealLocalTime- time1 <- atomically $ do- time0 <- readTVar timeVar- let time1 = max realTime (time0 + 1)- writeTVar timeVar time1- pure time1- pid <- getPidByMac- pure $ LamportTime time1 pid+ getTimes n' = do+ timeRangeStart <- LamportClock $ do+ timeVar <- ask+ lift $ do+ realTime <- getRealLocalTime+ atomically $ do+ timeCur <- readTVar timeVar+ let timeRangeStart = max realTime (timeCur + 1)+ writeTVar timeVar $ timeRangeStart + n - 1+ pure timeRangeStart+ pid <- getPid+ pure $ LamportTime timeRangeStart pid+ where+ n = max n' 1 instance Process m => Process (ReaderT r m) where getPid = lift getPid@@ -103,8 +122,8 @@ instance Clock m => Clock (ReaderT r m) where advance = lift . advance- getTime = lift getTime+ getTimes = lift . getTimes instance Clock m => Clock (StateT s m) where advance = lift . advance- getTime = lift getTime+ getTimes = lift . getTimes
lib/CRDT/LamportClock/Simulation.hs view
@@ -25,11 +25,15 @@ modify, state) import Control.Monad.Trans (MonadTrans, lift) import Data.Functor.Identity (Identity)+import Data.Hashable (hash) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map+import Data.Maybe (fromMaybe)+import Numeric.Natural (Natural) import CRDT.LamportClock (Clock, LamportTime (LamportTime), LocalTime,- Pid, Process, advance, getPid, getTime)+ Pid (Pid), Process, advance, getPid,+ getTimes) -- | Lamport clock simulation. Key is 'Pid'. -- Non-present value is equivalent to (0, initial).@@ -58,10 +62,12 @@ getPid = ProcessSim ask instance Monad m => Clock (ProcessSimT m) where- getTime = ProcessSim $ do+ getTimes n' = ProcessSim $ do pid <- ask- time <- lift $ preIncrementTime pid+ time <- lift $ preIncreaseTime n pid pure $ LamportTime time pid+ where+ n = max n' 1 advance time = ProcessSim $ do pid <- ask@@ -85,10 +91,11 @@ runProcessSimT :: Pid -> ProcessSimT m a -> LamportClockSimT m a runProcessSimT pid (ProcessSim action) = runReaderT action pid -preIncrementTime :: Monad m => Pid -> LamportClockSimT m LocalTime-preIncrementTime pid = LamportClockSim $ state $ \pss -> let- time = case Map.lookup pid pss of- Nothing -> 1- Just current -> succ current- in- (time, Map.insert pid time pss)+-- | Increase time by pid and return new value+preIncreaseTime :: Monad m => Natural -> Pid -> LamportClockSimT m LocalTime+preIncreaseTime n pid = LamportClockSim $ state $ \pss ->+ let time0 = fromMaybe 0 $ Map.lookup pid pss+ Pid p = pid+ d = fromIntegral . abs $ hash (time0, n, p)+ time = time0 + max 1 d+ in (time, Map.insert pid time pss)
+ test/Cm/RGA.hs view
@@ -0,0 +1,130 @@+{-# OPTIONS_GHC -Wno-missing-signatures #-}++{-# LANGUAGE ParallelListComp #-}+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE TypeApplications #-}++module Cm.RGA where++import Prelude hiding (fail)++import Control.Monad.State.Strict (execStateT, runStateT)+import Data.Foldable (toList)+import qualified Data.Map.Strict as Map+import Data.Maybe (isJust)+import qualified Data.Vector as Vector+import Test.QuickCheck (Property, conjoin, counterexample, (.&&.),+ (===))++import CRDT.Cm (apply, initial, makeAndApplyOp, makeOp)+import CRDT.Cm.RGA (RGA (OpAddAfter, OpRemove), RgaIntent (AddAfter),+ RgaPayload (RgaPayload), fromString, load,+ toString)+import CRDT.LamportClock (LamportTime (LamportTime), Pid (Pid),+ advance)+import CRDT.LamportClock.Simulation (ProcessSim, runLamportClockSim,+ runProcessSim)++import Laws (cmrdtLaw)+import Util (pattern (:-), expectRightK, fail, ok)++prop_makeOp = isJust $ makeOp @(RGA Char) @ProcessSim+ (AddAfter Nothing 'a')+ (initial @(RGA Char))++prop_makeAndApplyOp = conjoin+ [ counterexample "result3" $ expectRightK result3 $ \(op, payload) ->+ counterexample ("op = " ++ show op) (opsEqWoTime op op3)+ .&&. counterexample "payload" (payloadsEqWoTime payload payload3)+ , counterexample "result2" $ expectRightK result2 $ \(op, payload) ->+ counterexample ("op = " ++ show op) (opsEqWoTime op op2)+ .&&. counterexample "payload" (payloadsEqWoTime payload payload2)+ , counterexample "result1" $ expectRightK result1 $ \(op, payload) ->+ counterexample ("op = " ++ show op) (opsEqWoTime op op1)+ .&&. counterexample "payload" (payloadsEqWoTime payload payload1)+ , counterexample "result12" $ result12 === payload12+ , counterexample "results=" $ result21 === result12+ ]+ where+ time1 = LamportTime 4 $ Pid 1 -- TODO(cblp, 2018-02-11) arbitrary pids+ time2 = LamportTime 4 $ Pid 2+ time3 = LamportTime 3 $ Pid 3+ op1 = OpAddAfter Nothing '1' time1+ op2 = OpAddAfter Nothing '2' time2+ op3 = OpAddAfter Nothing '3' time3+ payload3 = load $ Vector.singleton (time3 :- Just '3')+ payload1 = load $ Vector.fromList [time1 :- Just '1', time3 :- Just '3']+ payload2 = load $ Vector.fromList [time2 :- Just '2', time3 :- Just '3']+ payload12 = load $ Vector.fromList+ [time2 :- Just '2', time1 :- Just '1', time3 :- Just '3']+ result3 =+ runLamportClockSim+ . runProcessSim (Pid 3)+ . (`runStateT` initial @(RGA Char))+ $ do+ advance 2+ makeAndApplyOp @(RGA Char) (AddAfter Nothing '3')+ result2 =+ runLamportClockSim . runProcessSim (Pid 2) . (`runStateT` payload3) $ do+ advance 1+ makeAndApplyOp @(RGA Char) (AddAfter Nothing '2')+ result1 =+ runLamportClockSim+ . runProcessSim (Pid 1)+ . (`runStateT` payload3)+ $ makeAndApplyOp @(RGA Char) (AddAfter Nothing '1')+ result12 = apply op2 payload1+ result21 = apply op1 payload2++prop_fromString s pid =+ expectRightK result $ payloadsEqWoTime $ load $ Vector.fromList+ [ LamportTime t pid :- Just c | t <- [1..] | c <- s ]+ where+ result =+ runLamportClockSim+ . runProcessSim pid+ . (`execStateT` initial @(RGA Char))+ $ fromString s++prop_fromString_toString s pid = expectRightK result $ \s' -> toString s' === s+ where+ result =+ runLamportClockSim+ . runProcessSim pid+ . (`execStateT` initial @(RGA Char))+ $ fromString s++prop_Cm = cmrdtLaw @(RGA Char)++-- | Ops equal without local times+opsEqWoTime (OpAddAfter parent1 atom1 id1) (OpAddAfter parent2 atom2 id2) =+ conjoin+ [ counterexample "parent" $ pidsMaybeEq parent1 parent2+ , counterexample "atom" $ atom1 === atom2+ , counterexample "id" $ pidsEqWoTime id1 id2+ ]+opsEqWoTime (OpRemove parent1) (OpRemove parent2) =+ counterexample "parent" $ pidsEqWoTime parent1 parent2+opsEqWoTime x y = fail $ show x ++ " /= " ++ show y++pidsEqWoTime (LamportTime _ pid1) (LamportTime _ pid2) = pid1 === pid2++pidsMaybeEq Nothing Nothing = ok+pidsMaybeEq (Just x) (Just y) = pidsEqWoTime x y+pidsMaybeEq x y = fail $ show x ++ " /= " ++ show y++payloadsEqWoTime :: (Eq a, Show a) => RgaPayload a -> RgaPayload a -> Property+payloadsEqWoTime (RgaPayload vertices1 vertexIxs1) (RgaPayload vertices2 vertexIxs2)+ = conjoin+ [ counterexample "vertices" $ conjoin+ [ counterexample ("[" ++ show i ++ "]")+ $ counterexample "id" (pidsEqWoTime id1 id2)+ .&&. counterexample "atom" (a1 === a2)+ | i <- [0 :: Int ..] | (id1, a1) <- toList vertices1 | (id2, a2) <- toList vertices2+ ]+ , counterexample "vertexIxs" $ conjoin+ [ counterexample "id" (pidsEqWoTime id1 id2)+ .&&. counterexample "ix" (ix1 === ix2)+ | (id1, ix1) <- Map.assocs vertexIxs1 | (id2, ix2) <- Map.assocs vertexIxs2+ ]+ ]
test/Cv/RGA.hs view
@@ -4,19 +4,22 @@ module Cv.RGA where -import Test.QuickCheck ((===))+import Prelude hiding (fail) +import Test.QuickCheck (conjoin, (.&&.), (.||.), (===))+ import CRDT.Cv.RGA (RgaString, edit, fromString, pack, toString, unpack)+import CRDT.LamportClock (LamportTime (LamportTime)) import CRDT.LamportClock.Simulation (runLamportClockSim, runProcessSim) import Laws (cvrdtLaws)-import Util (expectRight)+import Util (expectRight, fail) prop_fromString_toString s pid = expectRight $ do- s' <- runLamportClockSim . runProcessSim pid $ fromString s- pure $ toString s' === s+ v <- runLamportClockSim . runProcessSim pid $ fromString s+ pure $ toString v === s test_Cv = cvrdtLaws @RgaString @@ -25,3 +28,27 @@ pure $ toString v2 === s2 prop_pack_unpack rga = unpack (pack rga) == (rga :: RgaString)++prop_fromString_pack s pid = expectRight $ do+ v <- runLamportClockSim . runProcessSim pid $ fromString s+ pure $ case pack v of+ [(LamportTime _ pid', atoms)] -> atoms === map Just s .&&. pid' === pid+ [] -> s === ""+ p -> fail $ "cannot pack " ++ show p++prop_edit_pack s1 s2 pid1 pid2 = expectRight . runLamportClockSim $ do+ v1 <- runProcessSim pid1 $ fromString s1+ v2 <- runProcessSim pid2 $ edit (s1 ++ s2) v1+ pure $ case pack v2 of+ [(LamportTime _ pid1', atoms1), (LamportTime _ pid2', atoms2)] ->+ conjoin+ [ atoms1 === map Just s1+ , pid1' === pid1+ , atoms2 === map Just s2+ , pid2' === pid2+ ]+ [(LamportTime _ pid', atoms)] ->+ (atoms === map Just s1 .&&. pid' === pid1 .&&. s2 === "")+ .||. (atoms === map Just s2 .&&. pid' === pid2 .&&. s1 === "")+ [] -> s1 === "" .&&. s2 === ""+ p -> fail $ "cannot pack " ++ show p
test/LwwElementSet.hs view
@@ -16,6 +16,7 @@ import Data.Semigroup ((<>)) import CRDT.Cv.LwwElementSet (LwwElementSet, add, lookup, remove)+import CRDT.LamportClock (LamportTime (LamportTime), advance, getTime) import CRDT.LamportClock.Simulation (runLamportClockSim, runProcessSim) @@ -46,6 +47,10 @@ -- | Difference from 'ORSet' -- other replica can accidentally delete x prop_they_accidentally_delete_our_value (s :: LwwElementSet Char) x pid1 pid2 = expectRight . runLamportClockSim $ do- s1 <- runProcessSim pid1 $ add x s- s2 <- runProcessSim pid2 $ remove x =<< add x s+ (s1, LamportTime t0 _) <-+ runProcessSim pid1 $ (,) <$> add x s <*> getTime+ s2 <- runProcessSim pid2 $ do+ advance t0+ s' <- add x s+ remove x s' pure . not . lookup x $ s1 <> s2
− test/RGA.hs
@@ -1,85 +0,0 @@-{-# OPTIONS_GHC -Wno-missing-signatures #-}--{-# LANGUAGE ParallelListComp #-}-{-# LANGUAGE PatternSynonyms #-}-{-# LANGUAGE TypeApplications #-}--module RGA where--import Control.Monad.State.Strict (execStateT, runStateT)-import Data.Maybe (isJust)-import qualified Data.Vector as Vector-import Test.QuickCheck (conjoin, counterexample, (===))--import CRDT.Cm (apply, initial, makeAndApplyOp, makeOp)-import CRDT.Cm.RGA (RGA (OpAddAfter), RgaIntent (AddAfter),- fromString, load, toString)-import CRDT.LamportClock (LamportTime (LamportTime), Pid (Pid),- advance)-import CRDT.LamportClock.Simulation (ProcessSim, runLamportClockSim,- runProcessSim)--import Laws (cmrdtLaw)-import Util (pattern (:-), expectRightK)--prop_makeOp =- isJust- (makeOp @(RGA Char) (AddAfter Nothing 'a') (initial @(RGA Char))- :: Maybe (ProcessSim (RGA Char)))--prop_makeAndApplyOp = conjoin- [ counterexample "result3" $ result3 === Right (op3, payload3)- , counterexample "result2" $ result2 === Right (op2, payload2)- , counterexample "result1" $ result1 === Right (op1, payload1)- , counterexample "result12" $ result12 === payload12- , counterexample "results=" $ result21 === result12- ]- where- time1 = LamportTime 4 $ Pid 1- time2 = LamportTime 4 $ Pid 2- time3 = LamportTime 3 $ Pid 3- op1 = OpAddAfter Nothing '1' time1- op2 = OpAddAfter Nothing '2' time2- op3 = OpAddAfter Nothing '3' time3- payload3 = load $ Vector.singleton (time3 :- Just '3')- payload1 = load $ Vector.fromList [time1 :- Just '1', time3 :- Just '3']- payload2 = load $ Vector.fromList [time2 :- Just '2', time3 :- Just '3']- payload12 = load $ Vector.fromList- [time2 :- Just '2', time1 :- Just '1', time3 :- Just '3']- result3 =- runLamportClockSim .- runProcessSim (Pid 3) .- (`runStateT` initial @(RGA Char)) $ do- advance 2- makeAndApplyOp @(RGA Char) (AddAfter Nothing '3')- result2 =- runLamportClockSim . runProcessSim (Pid 2) . (`runStateT` payload3) $ do- advance 1- makeAndApplyOp @(RGA Char) (AddAfter Nothing '2')- result1 =- runLamportClockSim . runProcessSim (Pid 1) . (`runStateT` payload3) $- makeAndApplyOp @(RGA Char) (AddAfter Nothing '1')- result12 = apply op2 payload1- result21 = apply op1 payload2--prop_fromString s pid =- result ===- Right (load $ Vector.fromList- [LamportTime t pid :- Just c | t <- [1..] | c <- s])- where- result =- runLamportClockSim .- runProcessSim pid .- (`execStateT` initial @(RGA Char)) $- fromString s--prop_fromString_toString s pid =- expectRightK result $ \s' -> toString s' === s- where- result =- runLamportClockSim .- runProcessSim pid .- (`execStateT` initial @(RGA Char)) $- fromString s--prop_Cm = cmrdtLaw @(RGA Char)
test/Util.hs view
@@ -15,3 +15,9 @@ pattern (:-) :: a -> b -> (a, b) pattern a :- b = (a, b) infix 0 :-++ok :: Property+ok = property ()++fail :: String -> Property+fail s = counterexample s $ property False