diff --git a/crdt.cabal b/crdt.cabal --- a/crdt.cabal +++ b/crdt.cabal @@ -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 diff --git a/lib/CRDT/Cv/RGA.hs b/lib/CRDT/Cv/RGA.hs --- a/lib/CRDT/Cv/RGA.hs +++ b/lib/CRDT/Cv/RGA.hs @@ -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 ] diff --git a/lib/CRDT/LamportClock.hs b/lib/CRDT/LamportClock.hs --- a/lib/CRDT/LamportClock.hs +++ b/lib/CRDT/LamportClock.hs @@ -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 diff --git a/lib/CRDT/LamportClock/Simulation.hs b/lib/CRDT/LamportClock/Simulation.hs --- a/lib/CRDT/LamportClock/Simulation.hs +++ b/lib/CRDT/LamportClock/Simulation.hs @@ -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) diff --git a/test/Cm/RGA.hs b/test/Cm/RGA.hs new file mode 100644 --- /dev/null +++ b/test/Cm/RGA.hs @@ -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 + ] + ] diff --git a/test/Cv/RGA.hs b/test/Cv/RGA.hs --- a/test/Cv/RGA.hs +++ b/test/Cv/RGA.hs @@ -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 diff --git a/test/LwwElementSet.hs b/test/LwwElementSet.hs --- a/test/LwwElementSet.hs +++ b/test/LwwElementSet.hs @@ -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 diff --git a/test/RGA.hs b/test/RGA.hs deleted file mode 100644 --- a/test/RGA.hs +++ /dev/null @@ -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) diff --git a/test/Util.hs b/test/Util.hs --- a/test/Util.hs +++ b/test/Util.hs @@ -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