crdt 4.2 → 5.0

raw patch · 9 files changed

+108/−73 lines, 9 filesdep +stmPVP ok

version bump matches the API change (PVP)

Dependencies added: stm

API changes (from Hackage documentation)

Files

• crdt.cabal +4/−3
• lib/CRDT/Cm.hs +4/−3
• lib/CRDT/Cv/LwwElementSet.hs +4/−4
• lib/CRDT/Cv/ORSet.hs +1/−1
• lib/CRDT/LamportClock.hs +56/−26
• test/LWW.hs +9/−8
• test/Laws.hs +12/−10
• test/LwwElementSet.hs +14/−14
• test/ORSet.hs +4/−4

@@ -2,17 +2,17 @@ -- -- see: https://github.com/sol/hpack ----- hash: c7858c1e62913fddff18ba02c93949c604e502f20eced23a4b1d57b0c89d199b+-- hash: 35a8e5d951dbc5cec14577596c73f14d73ebf453af76b242e01beea8cf404d5c  name:           crdt-version:        4.2+version:        5.0 synopsis:       Conflict-free replicated data types description:    Definitions of CmRDT and CvRDT. Implementations for some classic CRDTs. category:       Distributed Systems homepage:       https://github.com/cblp/crdt#readme bug-reports:    https://github.com/cblp/crdt/issues maintainer:     Yuriy Syrovetskiy <cblp@cblp.su>-copyright:      2017 Yuriy Syrovetskiy, Nikolay Loginov+copyright:      2017 Yuriy Syrovetskiy, Nikolay Loginov; 2018 Yuriy Syrovetskiy license:        BSD3 license-file:   LICENSE build-type:     Simple@@ -33,6 +33,7 @@     , mtl     , network-info     , safe+    , stm     , time   exposed-modules:       CRDT.Cm

@@ -8,7 +8,7 @@     , concurrent     ) where -import           CRDT.LamportClock (Process)+import           CRDT.LamportClock (Clock)  -- | Partial order for causal semantics. -- Values of some type may be ordered and causally-ordered different ways.@@ -64,10 +64,11 @@     -- | Generate an update to the local and remote replicas.     --     -- Returns 'Nothing' if the intended operation is not applicable.-    makeOp :: Intent op -> Payload op -> Maybe (Process op)+    makeOp :: Clock m => Intent op -> Payload op -> Maybe (m op)      default makeOp-        :: Intent op ~ op => Intent op -> Payload op -> Maybe (Process op)+        :: (Intent op ~ op, Applicative m)+        => Intent op -> Payload op -> Maybe (m op)     makeOp i _ = Just $ pure i      -- | Apply an update to the payload (downstream).

@@ -16,7 +16,7 @@ import qualified Data.Map.Strict as Map import           Data.Semigroup (Semigroup (..)) -import           CRDT.LamportClock (Process)+import           CRDT.LamportClock (Clock) import           CRDT.LWW (LWW, advanceFromLWW) import qualified CRDT.LWW as LWW import           Data.Semilattice (Semilattice)@@ -32,13 +32,13 @@ initial :: LwwElementSet a initial = LES Map.empty -add :: Ord a => a -> LwwElementSet a -> Process (LwwElementSet a)+add :: (Ord a, Clock m) => a -> LwwElementSet a -> m (LwwElementSet a) add value old@(LES m) = do     advanceFromLES old     tag <- LWW.initial True     pure . LES $ Map.insertWith (<>) value tag m -remove :: Ord a => a -> LwwElementSet a -> Process (LwwElementSet a)+remove :: (Ord a, Clock m) => a -> LwwElementSet a -> m (LwwElementSet a) remove value old@(LES m) = do     advanceFromLES old     tag <- LWW.initial False@@ -47,5 +47,5 @@ lookup :: Ord a => a -> LwwElementSet a -> Bool lookup value (LES m) = maybe False LWW.query $ Map.lookup value m -advanceFromLES :: LwwElementSet a -> Process ()+advanceFromLES :: Clock m => LwwElementSet a -> m () advanceFromLES (LES m) = for_ m advanceFromLWW

@@ -33,7 +33,7 @@ initial :: ORSet a initial = ORSet Map.empty -add :: Ord a => a -> ORSet a -> Process (ORSet a)+add :: (Ord a, Process m) => a -> ORSet a -> m (ORSet a) add a (ORSet s) = do     pid <- getPid     pure $ ORSet $ Map.alter (add1 pid) a s

@@ -5,16 +5,19 @@     -- * Lamport timestamp (for a single process)     , Clock (..)     , LamportTime (..)-    , getRealLamportTime     -- * Lamport clock (for a whole multi-process system)-    , LamportClock (..)-    , runLamportClock+    , LamportClockSim (..)+    , runLamportClockSim     -- * Process     , Process (..)-    , getPid-    , runProcess+    -- * ProcessSim+    , ProcessSim (..)+    , runProcessSim     ) where +import           Control.Concurrent.STM (TVar, atomically, modifyTVar',+                                         readTVar, writeTVar)+import           Control.Monad.IO.Class (MonadIO, liftIO) import           Control.Monad.Reader (ReaderT, ask, runReaderT) import           Control.Monad.State.Strict (State, evalState, modify, state) import           Control.Monad.Trans (lift)@@ -43,37 +46,36 @@ newtype Pid = Pid Word64     deriving (Eq, Ord, Show) --- | Key is 'Pid'. Non-present value is equivalent to 0.-newtype LamportClock a = LamportClock (State (Map Pid LocalTime) a)+-- | Lamport clock simpulation. Key is 'Pid'.+-- Non-present value is equivalent to 0.+newtype LamportClockSim a = LamportClockSim (State (Map Pid LocalTime) a)     deriving (Applicative, Functor, Monad) -newtype Process a = Process (ReaderT Pid LamportClock a)+-- | ProcessSim inside Lamport clock simpulation.+newtype ProcessSim a = ProcessSim (ReaderT Pid LamportClockSim a)     deriving (Applicative, Functor, Monad) -getPid :: Process Pid-getPid = Process ask+class Monad m => Process m where+    getPid :: m Pid -runLamportClock :: LamportClock a -> a-runLamportClock (LamportClock action) = evalState action mempty+runLamportClockSim :: LamportClockSim a -> a+runLamportClockSim (LamportClockSim action) = evalState action mempty -runProcess :: Pid -> Process a -> LamportClock a-runProcess pid (Process action) = runReaderT action pid+runProcessSim :: Pid -> ProcessSim a -> LamportClockSim a+runProcessSim pid (ProcessSim action) = runReaderT action pid -preIncrementAt :: Pid -> LamportClock LocalTime+preIncrementAt :: Pid -> LamportClockSim LocalTime preIncrementAt pid =-    LamportClock . state $ \m -> let+    LamportClockSim . state $ \m -> let         lt' = succ . fromMaybe 0 $ Map.lookup pid m         in (lt', Map.insert pid lt' m)  getRealLocalTime :: IO LocalTime getRealLocalTime = round . utcTimeToPOSIXSeconds <$> getCurrentTime --- TODO(cblp, 2018-01-05) monotonic-getRealLamportTime :: IO LamportTime-getRealLamportTime =-    LamportTime <$> getRealLocalTime <*> (Pid . decodeMac <$> getMac)+getPidByMac :: IO Pid+getPidByMac = Pid . decodeMac <$> getMac   where-     getMac :: IO MAC     getMac =         headDef (error "Can't get any non-zero MAC address of this machine")@@ -85,15 +87,43 @@     decodeMac (MAC b5 b4 b3 b2 b1 b0) =         decode $ BSL.pack [0, 0, b5, b4, b3, b2, b1, b0] -class Monad m => Clock m where+class Process m => Clock m where     getTime :: m LamportTime     advance :: LocalTime -> m () -instance Clock Process where-    getTime = Process $ do+instance Process ProcessSim where+    getPid = ProcessSim ask++instance Clock ProcessSim where+    getTime = ProcessSim $ do         pid <- ask         time <- lift $ preIncrementAt pid         pure $ LamportTime time pid-    advance time = Process $ do++    advance time = ProcessSim $ do         pid <- ask-        lift . LamportClock . modify $ Map.insertWith max pid time+        lift . LamportClockSim . modify $ Map.insertWith max pid time++-- TODO(cblp, 2018-01-06) benchmark and compare with 'atomicModifyIORef'+newtype LamportClock a = LamportClock (ReaderT (TVar LocalTime) IO a)+    deriving (Applicative, Functor, Monad, MonadIO)++instance Process LamportClock where+    getPid = liftIO getPidByMac++instance Clock LamportClock where+    advance time = LamportClock $ do+        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

@@ -14,7 +14,7 @@ import           Data.Semigroup ((<>)) import           Test.QuickCheck ((===)) -import           CRDT.LamportClock (runLamportClock, runProcess)+import           CRDT.LamportClock (runLamportClockSim, runProcessSim) import           CRDT.LWW (LWW, assign, initial, query)  import           Laws (cmrdtLaw, cvrdtLaws)@@ -23,13 +23,14 @@  test_Cv = cvrdtLaws @(LWW Char) -prop_assign pid1 pid2 (formerValue :: Char) latterValue = runLamportClock $ do-    state1 <- runProcess pid1 $ initial formerValue-    state2 <- runProcess pid2 $ assign latterValue state1-    pure $ query state2 === latterValue+prop_assign pid1 pid2 (formerValue :: Char) latterValue =+    runLamportClockSim $ do+        state1 <- runProcessSim pid1 $ initial formerValue+        state2 <- runProcessSim pid2 $ assign latterValue state1+        pure $ query state2 === latterValue  prop_merge_with_former pid1 pid2 (formerValue :: Char) latterValue =-    runLamportClock $ do-        state1 <- runProcess pid1 $ initial formerValue-        state2 <- runProcess pid2 $ assign latterValue state1+    runLamportClockSim $ do+        state1 <- runProcessSim pid1 $ initial formerValue+        state2 <- runProcessSim pid2 $ assign latterValue state1         pure $ query (state1 <> state2) === latterValue

@@ -23,7 +23,8 @@ import           CRDT.Cm.GSet (GSet) import           CRDT.Cm.TwoPSet (TwoPSet) import           CRDT.Cv (CvRDT)-import           CRDT.LamportClock (Process, runLamportClock, runProcess)+import           CRDT.LamportClock (Clock, ProcessSim, runLamportClockSim,+                                    runProcessSim) import           CRDT.LWW (LWW) import qualified CRDT.LWW as LWW import           Data.Semilattice (Semilattice, merge)@@ -53,9 +54,10 @@     type Initial op     type Initial op = Payload op -    initialize :: Initial op -> Process (Payload op)+    initialize :: Clock m => Initial op -> m (Payload op)     default initialize-        :: Initial op ~ Payload op => Initial op -> Process (Payload op)+        :: (Applicative m, Initial op ~ Payload op)+        => Initial op -> m (Payload op)     initialize = pure  instance Initialize (Counter a)@@ -82,18 +84,18 @@ cmrdtLaw = property concurrentOpsCommute   where     concurrentOpsCommute st1 st2 seed3 in1 in2 pid1 pid2 pid3 =-        let (op1, op2, state) = runLamportClock $ (,,)-                <$> runProcess pid1 (makeOp @op in1 st1 `orElse` discard)-                <*> runProcess pid2 (makeOp @op in2 st2 `orElse` discard)-                <*> runProcess pid3 (initialize @op seed3)+        let (op1, op2, state) = runLamportClockSim $ (,,)+                <$> runProcessSim pid1 (makeOp @op in1 st1 `orElse` discard)+                <*> runProcessSim pid2 (makeOp @op in2 st2 `orElse` discard)+                <*> runProcessSim pid3 (initialize @op seed3)         in  concurrent op1 op2 ==> opCommutativity (in1, op1) (in2, op2) state     opCommutativity (in1, op1) (in2, op2) state =-        isJust (makeOp @op in1 state) ==>-        isJust (makeOp @op in2 state) ==>+        isJust (makeOp @op @ProcessSim in1 state) ==>+        isJust (makeOp @op @ProcessSim in2 state) ==>             counterexample                 ( show in1 ++ " must be valid after " ++ show op2 ++                   " applied to " ++ show state )-                (isJust $ makeOp @op in1 $ apply op2 state)+                (isJust $ makeOp @op @ProcessSim in1 $ apply op2 state)             .&&.             (apply op1 . apply op2) state === (apply op2 . apply op1) state 

@@ -15,35 +15,35 @@  import           Data.Semigroup ((<>)) -import           CRDT.Cv.LwwElementSet (LwwElementSet (..), add, lookup, remove)-import           CRDT.LamportClock (runLamportClock, runProcess)+import           CRDT.Cv.LwwElementSet (LwwElementSet, add, lookup, remove)+import           CRDT.LamportClock (runLamportClockSim, runProcessSim)  import           Laws (cvrdtLaws)  test_Cv = cvrdtLaws @(LwwElementSet Char)  prop_add (s :: LwwElementSet Char) x pid1 =-    runLamportClock $ do-        s1 <- runProcess pid1 $ add x s+    runLamportClockSim $ do+        s1 <- runProcessSim pid1 $ add x s         pure $ lookup x s1  prop_remove (s :: LwwElementSet Char) x pid1 pid2 =-    runLamportClock $ do-        s1 <- runProcess pid1 $ add x s-        s2 <- runProcess pid2 $ remove x s1+    runLamportClockSim $ do+        s1 <- runProcessSim pid1 $ add x s+        s2 <- runProcessSim pid2 $ remove x s1         pure . not $ lookup x s2  -- | Difference from 'TwoPSet' -- no removal bias prop_no_removal_bias (s :: LwwElementSet Char) x pid1 pid2 pid3 =-    runLamportClock $ do-        s1 <- runProcess pid1 $ add x s-        s2 <- runProcess pid2 $ remove x s1-        s3 <- runProcess pid3 $ add x s2+    runLamportClockSim $ do+        s1 <- runProcessSim pid1 $ add x s+        s2 <- runProcessSim pid2 $ remove x s1+        s3 <- runProcessSim pid3 $ add x s2         pure $ lookup x s3  -- | Difference from 'ORSet' -- other replica can accidentally delete x prop_they_accidentally_delete_our_value (s :: LwwElementSet Char) x pid1 pid2 =-    runLamportClock $ do-        s1 <- runProcess pid1 $ add x s-        s2 <- runProcess pid2 $ remove x =<< add x s+    runLamportClockSim $ do+        s1 <- runProcessSim pid1 $ add x s+        s2 <- runProcessSim pid2 $ remove x =<< add x s         pure . not . lookup x $ s1 <> s2

@@ -9,17 +9,17 @@  import           Data.Semigroup (Semigroup ((<>))) -import           CRDT.Cv.ORSet (ORSet (..), add, lookup, remove)-import           CRDT.LamportClock (runLamportClock, runProcess)+import           CRDT.Cv.ORSet (ORSet, add, lookup, remove)+import           CRDT.LamportClock (runLamportClockSim, runProcessSim)  import           Laws (cvrdtLaws)  test_Cv = cvrdtLaws @(ORSet Int)  prop_add pid (x :: Char) s =-    runLamportClock $ runProcess pid $ not . lookup x . remove x <$> add x s+    runLamportClockSim $ runProcessSim pid $ not . lookup x . remove x <$> add x s  -- | Difference from 'LwwElementSet' -- -- other replica can not accidentally delete x prop_add_merge (x :: Char) pid s1 s0 =-    runLamportClock $ runProcess pid $ lookup x . (<> s1) <$> add x s0+    runLamportClockSim $ runProcessSim pid $ lookup x . (<> s1) <$> add x s0