crdt 1.0 → 2.0

raw patch · 15 files changed

+195/−280 lines, 15 filesdep −enummapsetdep −latticesdep −microlensPVP ok

version bump matches the API change (PVP)

Dependencies removed: enummapset, lattices, microlens, microlens-ghc, microlens-mtl

API changes (from Hackage documentation)

Files

• crdt.cabal +2/−11
• lib/CRDT/Cm.hs +33/−16
• lib/CRDT/Cm/Counter.hs +10/−27
• lib/CRDT/Cm/GSet.hs +8/−26
• lib/CRDT/Cm/LWW.hs +0/−35
• lib/CRDT/Cm/TPSet.hs +12/−32
• lib/CRDT/Cv/LWW.hs +0/−48
• lib/CRDT/LWW.hs +77/−0
• lib/Data/Observe.hs +0/−11
• lib/LamportClock.hs +30/−16
• lib/Lens/Micro/Extra.hs +0/−29
• test/ArbitraryOrphans.hs +7/−11
• test/GSet.hs +6/−6
• test/LWW.hs +1/−2
• test/Laws.hs +9/−10

@@ -3,7 +3,7 @@ -- see: https://github.com/sol/hpack  name:           crdt-version:        1.0+version:        2.0 synopsis:       Conflict-free replicated data types description:    Definitions of CmRDT and CvRDT. Implementations for some classic CRDTs. category:       Distributed Systems@@ -26,29 +26,20 @@   build-depends:       base >= 4.9 && < 4.10     , containers-    , enummapset-    , lattices-    , microlens-    , microlens-ghc-    , microlens-mtl     , mtl   exposed-modules:       CRDT.Cm       CRDT.Cm.Counter       CRDT.Cm.GSet-      CRDT.Cm.LWW       CRDT.Cm.TPSet       CRDT.Cv       CRDT.Cv.GCounter       CRDT.Cv.GSet-      CRDT.Cv.LWW       CRDT.Cv.Max       CRDT.Cv.PNCounter-      Data.Observe+      CRDT.LWW       Data.Semilattice       LamportClock-  other-modules:-      Lens.Micro.Extra   default-language: Haskell2010  test-suite test

@@ -1,25 +1,29 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE DefaultSignatures #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FunctionalDependencies #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}  module CRDT.Cm-    ( CmRDT (..)+    ( CausalOrd (..)+    , CmRDT (..)     , concurrent     ) where -import           Algebra.PartialOrd (PartialOrd (leq))-import           Data.Observe (Observe (..))- import           LamportClock (Clock) --- | TODO(cblp, 2017-09-29) import from lattices >= 1.6-comparable :: PartialOrd a => a -> a -> Bool-comparable a b = a `leq` b || b `leq` a+-- | Partial order for causal semantics.+-- Values of some type may be ordered and causally-ordered different ways.+class CausalOrd a where+    before :: a -> a -> Bool +comparable :: CausalOrd a => a -> a -> Bool+comparable a b = a `before` b || b `before` a+ -- | Not comparable, i. e. ¬(a ≤ b) ∧ ¬(b ≤ a).-concurrent :: PartialOrd a => a -> a -> Bool+concurrent :: CausalOrd a => a -> a -> Bool concurrent a b = not $ comparable a b  {- |@@ -55,22 +59,35 @@ Idempotency doesn't need to hold. -} -class (Observe payload, Eq (Observed payload), PartialOrd update)-        => CmRDT payload op update-        | payload -> op, op -> update, update -> payload-        where+class (CausalOrd u, Eq (View u)) => CmRDT u where+    type Op u+    type Op u = u -- default case +    type Payload u++    type View u+    type View u = Payload u -- default case+     -- | Precondition for 'updateAtSource'.     -- Calculates if the operation is applicable to the current state.-    updateAtSourcePre :: op -> payload -> Bool+    updateAtSourcePre :: Op u -> Payload u -> Bool     updateAtSourcePre _ _ = True      -- | Generate an update to the local and remote replicas.     -- Doesn't have sense if 'updateAtSourcePre' is false.     --     -- May or may not use clock.-    updateAtSource :: Clock m => op -> m update+    updateAtSource :: Clock m => Op u -> m u +    default updateAtSource :: (Clock m, Op u ~ u) => Op u -> m u+    updateAtSource = pure+     -- | Apply an update to the payload.     -- An invalid update must be ignored.-    updateDownstream :: update -> payload -> payload+    updateDownstream :: u -> Payload u -> Payload u++    -- | Extract user-visible value from payload+    view :: Payload u -> View u++    default view :: (Payload u ~ View u) => Payload u -> View u+    view = id

@@ -1,41 +1,24 @@ {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE TypeFamilies #-}  module CRDT.Cm.Counter     ( Counter (..)-    , CounterOp (..)-    , initial     ) where -import           Algebra.PartialOrd (PartialOrd (..))-import           Data.Observe (Observe (..))--import           CRDT.Cm (CmRDT (..))--newtype Counter a = Counter a-    deriving (Show)+import           CRDT.Cm (CausalOrd (..), CmRDT (..)) -data CounterOp a = Increment | Decrement+data Counter a = Increment | Decrement     deriving (Bounded, Enum, Eq, Show) -instance (Num a, Eq a) => CmRDT (Counter a) (CounterOp a) (CounterOp a) where-    updateAtSource = pure-    updateDownstream = opToFunc+instance (Num a, Eq a) => CmRDT (Counter a) where+    type Payload  (Counter a) = a -instance Observe (Counter a) where-    type Observed (Counter a) = a-    observe (Counter c) = c+    updateDownstream = \case+        Increment -> (+ 1)+        Decrement -> subtract 1  -- | Empty order, allowing arbitrary reordering-instance PartialOrd (CounterOp a) where-    leq _ _ = False--initial :: Num a => Counter a-initial = Counter 0--opToFunc :: Num a => CounterOp a -> Counter a -> Counter a-opToFunc op (Counter c) =-    Counter $ case op of-        Increment -> c + 1-        Decrement -> c - 1+instance CausalOrd (Counter a) where+    before _ _ = False

@@ -1,43 +1,25 @@-{-# OPTIONS_GHC -Wno-orphans #-}- {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE TypeFamilies #-}  module CRDT.Cm.GSet-    ( GSet-    , Add (..)-    , initial-    , lookup+    ( GSet (..)     ) where  import           Prelude hiding (lookup) -import           Algebra.PartialOrd (PartialOrd (leq))-import           Data.Observe (Observe (..)) import           Data.Set (Set) import qualified Data.Set as Set -import           CRDT.Cm (CmRDT (..))-import           CRDT.Cv.GSet (GSet)--initial :: GSet a-initial = Set.empty---- | query lookup-lookup :: Ord a => a -> GSet a -> Bool-lookup = Set.member+import           CRDT.Cm (CausalOrd (..), CmRDT (..)) -newtype Add a = Add a+newtype GSet a = Add a     deriving (Eq, Show) -instance Ord a => CmRDT (Set a) (Add a) (Add a) where-    updateAtSource = pure-    updateDownstream (Add a) = Set.insert a+instance Ord a => CmRDT (GSet a) where+    type Payload  (GSet a) = Set a -instance Observe (GSet a) where-    type Observed (GSet a) = Set a-    observe = id+    updateDownstream (Add a) = Set.insert a -instance Eq a => PartialOrd (Add a) where-    leq _ _ = False+instance Eq a => CausalOrd (GSet a) where+    before _ _ = False

@@ -1,35 +0,0 @@-{-# OPTIONS_GHC -Wno-orphans #-} -- TODO(cblp, 2017-10-02) join with Cv.LWW--{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE TypeFamilies #-}--module CRDT.Cm.LWW-    ( LWW (..)-    , Assign (..)-    ) where--import           Algebra.PartialOrd (PartialOrd (..))-import           Data.Observe (Observe (..))-import           Data.Semigroup ((<>))-import           Lens.Micro ((<&>))--import           CRDT.Cm (CmRDT (..))-import           CRDT.Cv.LWW (LWW (..))-import           LamportClock (Clock (newTimestamp))--instance PartialOrd (LWW a) where-    leq _ _ = False--newtype Assign a = Assign a-    deriving (Eq, Show)--instance Eq a => CmRDT (LWW a) (Assign a) (LWW a) where-    updateAtSource (Assign value) =-        newTimestamp <&> \t -> LWW{timestamp = t, value}-    updateDownstream = (<>)--instance Observe (LWW a) where-    type Observed (LWW a) = a-    observe = value

@@ -1,3 +1,4 @@+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE TypeFamilies #-}@@ -5,50 +6,29 @@ -- | TODO(cblp, 2017-09-29) USet? module CRDT.Cm.TPSet     ( TPSet (..)-    , TPSetOp (..)-    , initial-    , lookup     , updateAtSource     , updateDownstream     ) where -import           Prelude hiding (lookup)--import           Algebra.PartialOrd (PartialOrd (..))-import           Data.Observe (Observe (..)) import           Data.Set (Set) import qualified Data.Set as Set -import           CRDT.Cm (CmRDT (..))--newtype TPSet a = TPSet{payload :: Set a}-    deriving (Show)+import           CRDT.Cm (CausalOrd (..), CmRDT (..)) -data TPSetOp a = Add a | Remove a+data TPSet a = Add a | Remove a     deriving (Eq, Show) -initial :: TPSet a-initial = TPSet Set.empty---- | query lookup-lookup :: Ord a => a -> TPSet a -> Bool-lookup a TPSet{payload} = Set.member a payload+instance Ord a => CmRDT (TPSet a) where+    type Payload  (TPSet a) = Set a -instance Ord a => CmRDT (TPSet a) (TPSetOp a) (TPSetOp a) where     updateAtSourcePre op payload = case op of         Add _     -> True-        Remove a  -> lookup a payload--    updateAtSource = pure--    updateDownstream op TPSet{payload} = case op of-        Add a     -> TPSet{payload = Set.insert a payload}-        Remove a  -> TPSet{payload = Set.delete a payload}+        Remove a  -> Set.member a payload -instance Observe (TPSet a) where-    type Observed (TPSet a) = Set a-    observe = payload+    updateDownstream = \case+        Add a     -> Set.insert a+        Remove a  -> Set.delete a -instance Eq a => PartialOrd (TPSetOp a) where-    leq (Remove a) (Add b) = a == b -- `Remove e` can occur only after `Add e`-    leq _ _ = False -- Any other are not ordered+instance Eq a => CausalOrd (TPSet a) where+    Remove a `before` Add b = a == b -- `Remove e` can occur only after `Add e`+    _        `before` _     = False  -- Any other are not ordered

@@ -1,48 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE TypeFamilies #-}--module CRDT.Cv.LWW-    ( LWW (..)-    , initial-    , assign-    , query-    ) where--import           Data.Function (on)-import           Data.Semigroup (Semigroup, (<>))-import           Lens.Micro ((<&>))--import           CRDT.Cv (CvRDT)-import           LamportClock (Clock (newTimestamp), Timestamp)---- | Last write wins. Assuming timestamp is unique.-data LWW a = LWW-    { timestamp :: !Timestamp-    , value     :: !a-    }-    deriving (Show)--instance Eq (LWW a) where-    (==) = (==) `on` timestamp--instance Ord (LWW a) where-    (<=) = (<=) `on` timestamp---- | Merge by choosing more recent timestamp.-instance Semigroup (LWW a) where-    (<>) = max--instance CvRDT (LWW a)---- | Initialize state-initial :: Clock f => a -> f (LWW a)-initial value = newTimestamp <&> \timestamp -> LWW{timestamp = timestamp, value}---- | Change state-assign :: Clock f => a -> LWW a -> f (LWW a)-assign value cur = newTimestamp <&> \timestamp -> cur <> LWW{timestamp, value}---- | Query state-query :: LWW a -> a-query = value

@@ -0,0 +1,77 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE TypeFamilies #-}++module CRDT.LWW+    ( LWW (..)+      -- * CvRDT+    , initial+    , assign+    , query+      -- * CmRDT+    , Assign (..)+    ) where++import           Data.Function (on)+import           Data.Semigroup (Semigroup, (<>))++import           CRDT.Cm (CausalOrd (..), CmRDT (..))+import           CRDT.Cv (CvRDT)+import           LamportClock (Clock (newTimestamp), Timestamp)++-- | Last write wins. Assuming timestamp is unique.+-- This type is both 'CmRDT' and 'CvRDT'.+data LWW a = LWW+    { value     :: !a+    , timestamp :: !Timestamp+    }+    deriving (Show)++--------------------------------------------------------------------------------+-- CvRDT -----------------------------------------------------------------------++instance Eq (LWW a) where+    (==) = (==) `on` timestamp++instance Ord (LWW a) where+    (<=) = (<=) `on` timestamp++-- | Merge by choosing more recent timestamp.+instance Semigroup (LWW a) where+    (<>) = max++instance CvRDT (LWW a)++-- | Initialize state+initial :: Clock f => a -> f (LWW a)+initial value = LWW value <$> newTimestamp++-- | Change state as CvRDT operation.+-- Current value is ignored, because new timestamp is always greater.+assign :: Clock f => a -> LWW a -> f (LWW a)+assign value _ = LWW value <$> newTimestamp++-- | Query state+query :: LWW a -> a+query = value++--------------------------------------------------------------------------------+-- CmRDT -----------------------------------------------------------------------++instance CausalOrd (LWW a) where+    before _ _ = False++-- | Change state as CmRDT operation+newtype Assign a = Assign a+    deriving (Eq, Show)++instance Eq a => CmRDT (LWW a) where+    type Op       (LWW a) = Assign a+    type Payload  (LWW a) = LWW a+    type View     (LWW a) = a++    updateAtSource (Assign value) = LWW value <$> newTimestamp++    updateDownstream = (<>)++    view = value

@@ -1,11 +0,0 @@-{-# LANGUAGE TypeFamilies #-}--module Data.Observe-    ( Observe (..)-    ) where--import           Data.Kind (Type)--class Observe a where-    type Observed a :: Type-    observe :: a -> Observed a

@@ -1,7 +1,5 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE TypeSynonymInstances #-}  module LamportClock     ( Pid (..)@@ -15,37 +13,42 @@     , barrier     ) where +import           Control.Arrow (first) import           Control.Monad.Reader (ReaderT, ask, runReaderT)-import           Control.Monad.State.Strict (State, evalState, modify)-import           Data.EnumMap.Strict (EnumMap)-import qualified Data.EnumMap.Strict as EnumMap+import           Control.Monad.State.Strict (MonadState, State, evalState,+                                             modify, state) import           Data.Functor (($>))+import           Data.IntMap.Strict (IntMap)+import qualified Data.IntMap.Strict as IntMap import           Data.Maybe (fromMaybe)-import           Data.Word (Word32)-import           Lens.Micro (at, non)-import           Lens.Micro.Extra ((<<+=))  type Time = Word  -- | Unique process identifier-newtype Pid = Pid Word32-    deriving (Enum, Eq, Ord, Show)+newtype Pid = Pid Int+    deriving (Eq, Ord, Show) --- | TODO(cblp, 2017-09-28) Use bounded-intmap-type LamportClock = State (EnumMap Pid Time)+unPid :: Pid -> Int+unPid (Pid pid) = pid +-- | Key is 'Pid'. Non-present value is equivalent to 0.+-- TODO(cblp, 2017-09-28) Use bounded-intmap+type LamportTime = IntMap Time++type LamportClock = State LamportTime+ -- | XXX Make sure all subsequent calls to 'newTimestamp' return timestamps -- greater than all prior calls. barrier :: [Pid] -> LamportClock () barrier pids =     modify $ \clocks -> let-        selectedClocks = EnumMap.fromList-            [(pid, fromMaybe 0 $ EnumMap.lookup pid clocks) | pid <- pids]+        selectedClocks = lamportTimeFromList+            [(pid, fromMaybe 0 $ lamportTimeLookup pid clocks) | pid <- pids]         in         if null selectedClocks then             clocks         else-            EnumMap.union+            IntMap.union                 (selectedClocks $> succ (maximum selectedClocks))                 clocks @@ -65,7 +68,7 @@ instance Clock Process where     newTimestamp = do         pid <- ask-        time <- at pid . non 0 <<+= 1+        time <- postIncrementAt pid         pure $ Timestamp time pid  runLamportClock :: LamportClock a -> a@@ -73,3 +76,14 @@  runProcess :: Pid -> Process a -> LamportClock a runProcess pid action = runReaderT action pid++postIncrementAt :: MonadState LamportTime m => Pid -> m Time+postIncrementAt (Pid pid) = state $ \m ->+    let v = fromMaybe 0 $ IntMap.lookup pid m+    in  (v, IntMap.insert pid (v + 1) m)++lamportTimeFromList :: [(Pid, Time)] -> LamportTime+lamportTimeFromList = IntMap.fromList . map (first unPid)++lamportTimeLookup :: Pid -> LamportTime -> Maybe Time+lamportTimeLookup = IntMap.lookup . unPid

@@ -1,29 +0,0 @@-{-# OPTIONS_GHC -Wno-orphans #-}--{-# LANGUAGE TypeFamilies #-}--module Lens.Micro.Extra ((<<+=)) where--import           Control.Monad.State (MonadState)-import           Data.EnumMap.Strict (EnumMap, enumMapToIntMap, intMapToEnumMap)-import           Lens.Micro (LensLike', at, ix)-import           Lens.Micro.GHC ()-import           Lens.Micro.Internal (At, Index, IxValue, Ixed)-import           Lens.Micro.Mtl ((<<%=))--(<<+=) :: (Num a, MonadState s m) => LensLike' ((,) a) s a -> a -> m a-alens <<+= n = alens <<%= (+ n)-infix 4 <<+=-{-# INLINE (<<+=) #-}--type instance Index (EnumMap k v) = k--type instance IxValue (EnumMap k v) = v--instance Enum k => Ixed (EnumMap k v) where-    ix k f m = intMapToEnumMap <$> ix (fromEnum k) f (enumMapToIntMap m)-    {-# INLINE ix #-}--instance Enum k => At (EnumMap k v) where-    at k f m = intMapToEnumMap <$> at (fromEnum k) f (enumMapToIntMap m)-    {-# INLINE at #-}

@@ -7,31 +7,27 @@  import           Test.QuickCheck (Arbitrary (..), arbitraryBoundedEnum, oneof) -import           CRDT.Cm.Counter (Counter (..), CounterOp (..))-import           CRDT.Cm.GSet (Add (..))-import           CRDT.Cm.LWW (Assign (..), LWW (..))-import           CRDT.Cm.TPSet (TPSet (..), TPSetOp (..))+import           CRDT.Cm.Counter (Counter (..))+import           CRDT.Cm.GSet (GSet (..))+import           CRDT.Cm.TPSet (TPSet (..)) import qualified CRDT.Cm.TPSet as TPSet import           CRDT.Cv.GCounter (GCounter (..)) import           CRDT.Cv.Max (Max (..)) import           CRDT.Cv.PNCounter (PNCounter (..))+import           CRDT.LWW (Assign (..), LWW (..)) import           LamportClock (Pid (..), Timestamp (..)) -deriving instance Arbitrary a => Arbitrary (Counter a)--instance Arbitrary (CounterOp a) where+instance Arbitrary (Counter a) where     arbitrary = arbitraryBoundedEnum -deriving instance Arbitrary a => Arbitrary (Add a)- deriving instance Arbitrary a => Arbitrary (Assign a)  instance Arbitrary a => Arbitrary (LWW a) where     arbitrary = LWW <$> arbitrary <*> arbitrary -deriving instance (Arbitrary a, Ord a) => Arbitrary (TPSet a)+deriving instance Arbitrary a => Arbitrary (GSet a) -instance Arbitrary a => Arbitrary (TPSetOp a) where+instance Arbitrary a => Arbitrary (TPSet a) where     arbitrary = oneof [TPSet.Add <$> arbitrary, Remove <$> arbitrary]  deriving instance Arbitrary a => Arbitrary (GCounter a)

@@ -8,15 +8,15 @@ import           Test.Tasty (TestTree, testGroup) import           Test.Tasty.QuickCheck (testProperty, (==>)) -import           CRDT.Cm.GSet (GSet)-import           CRDT.Cv.GSet (add, query)+import qualified CRDT.Cm.GSet as Cm+import qualified CRDT.Cv.GSet as Cv  import           Laws (cmrdtLaw, cvrdtLaws)  gSet :: TestTree gSet = testGroup "GSet"-    [ cmrdtLaw @(GSet Int)-    , cvrdtLaws @(GSet Int)-    , testProperty "Cv.add" $-        \(set :: GSet Int) i -> not (query i set) ==> query i (add i set)+    [ cmrdtLaw @(Cm.GSet Int)+    , cvrdtLaws @(Cv.GSet Int)+    , testProperty "Cv.add" $ \(set :: Cv.GSet Int) i ->+        not (Cv.query i set) ==> Cv.query i (Cv.add i set)     ]

@@ -9,8 +9,7 @@ import           Test.Tasty (TestTree, testGroup) import           Test.Tasty.QuickCheck (testProperty) -import           CRDT.Cm.LWW (LWW (..))-import           CRDT.Cv.LWW (assign, initial, query)+import           CRDT.LWW (LWW (..), assign, initial, query) import           LamportClock (barrier, runLamportClock, runProcess)  import           Laws (cmrdtLaw, cvrdtLaws)

@@ -11,7 +11,6 @@  import           Control.Monad (unless) import           Data.Function ((&))-import           Data.Observe (Observe (..)) import           Data.Semigroup (Semigroup, (<>)) import           Data.Semilattice (Semilattice, merge) import           Test.Tasty (TestTree, testGroup)@@ -45,22 +44,22 @@ cvrdtLaws = semilatticeLaws @a  cmrdtLaw-    :: forall payload op up-    . ( CmRDT payload op up-      , Arbitrary payload, Show payload-      , Arbitrary op, Show op-      , Show (Observed payload)+    :: forall u+    . ( CmRDT u+      , Arbitrary (Op u), Show (Op u)+      , Arbitrary (Payload u), Show (Payload u)+      , Show (View u)       )     => TestTree cmrdtLaw = testProperty "CmRDT law: concurrent ops commute" $-    \pid (state0 :: payload) (op1, op2 :: op) ->+    \pid (state0 :: Payload u) (op1, op2 :: Op u) ->         runLamportClock $ runProcess pid $ do             up1 <- updateAtSourceIfCan op1 state0             up2 <- updateAtSourceIfCan op2 state0             let state12 = state0 & updateDownstream up1 & updateDownstream up2             let state21 = state0 & updateDownstream up2 & updateDownstream up1-            pure $ concurrent up1 up2 ==> observe state12 === observe state21+            pure $ concurrent up1 up2 ==> view @u state12 === view @u state21   where-    updateAtSourceIfCan :: Clock m => op -> payload -> m up+    updateAtSourceIfCan :: Clock m => Op u -> Payload u -> m u     updateAtSourceIfCan op state =-        unless (updateAtSourcePre op state) discard *> updateAtSource op+        unless (updateAtSourcePre @u op state) discard *> updateAtSource op