crdt 0.5 → 1.0

raw patch · 36 files changed

+753/−396 lines, 36 filesdep +enummapsetdep +latticesdep +microlensPVP ok

version bump matches the API change (PVP)

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

API changes (from Hackage documentation)

Files

• crdt.cabal +24/−11
• lib/CRDT/Cm.hs +60/−8
• lib/CRDT/Cm/Counter.hs +41/−0
• lib/CRDT/Cm/GSet.hs +43/−0
• lib/CRDT/Cm/LWW.hs +35/−0
• lib/CRDT/Cm/TPSet.hs +54/−0
• lib/CRDT/Cv.hs +2/−1
• lib/CRDT/Cv/GCounter.hs +42/−0
• lib/CRDT/Cv/GSet.hs +28/−0
• lib/CRDT/Cv/LWW.hs +48/−0
• lib/CRDT/Cv/Max.hs +19/−0
• lib/CRDT/Cv/PNCounter.hs +59/−0
• lib/CRDT/GCounter/Cm.hs +0/−22
• lib/CRDT/GCounter/Cv.hs +0/−31
• lib/CRDT/GCounter/Cv/Internal.hs +0/−16
• lib/CRDT/GSet/Cv.hs +0/−21
• lib/CRDT/GSet/Cv/Internal.hs +0/−12
• lib/CRDT/LWW.hs +0/−44
• lib/CRDT/PNCounter/Cm.hs +0/−23
• lib/CRDT/PNCounter/Cv.hs +0/−41
• lib/CRDT/PNCounter/Cv/Internal.hs +0/−22
• lib/CRDT/Timestamp.hs +0/−7
• lib/Data/Observe.hs +11/−0
• lib/Data/Semilattice.hs +5/−5
• lib/LamportClock.hs +75/−0
• lib/Lens/Micro/Extra.hs +29/−0
• test/ArbitraryOrphans.hs +47/−0
• test/Counter.hs +13/−0
• test/GCounter.hs +6/−27
• test/GSet.hs +7/−10
• test/LWW.hs +22/−41
• test/Laws.hs +32/−22
• test/Main.hs +6/−2
• test/Max.hs +23/−0
• test/PNCounter.hs +9/−30
• test/TPSet.hs +13/−0

@@ -3,7 +3,7 @@ -- see: https://github.com/sol/hpack  name:           crdt-version:        0.5+version:        1.0 synopsis:       Conflict-free replicated data types description:    Definitions of CmRDT and CvRDT. Implementations for some classic CRDTs. category:       Distributed Systems@@ -26,20 +26,29 @@   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.GCounter.Cm-      CRDT.GCounter.Cv-      CRDT.GCounter.Cv.Internal-      CRDT.GSet.Cv-      CRDT.GSet.Cv.Internal-      CRDT.LWW-      CRDT.PNCounter.Cm-      CRDT.PNCounter.Cv-      CRDT.PNCounter.Cv.Internal-      CRDT.Timestamp+      CRDT.Cv.GCounter+      CRDT.Cv.GSet+      CRDT.Cv.LWW+      CRDT.Cv.Max+      CRDT.Cv.PNCounter+      Data.Observe       Data.Semilattice+      LamportClock+  other-modules:+      Lens.Micro.Extra   default-language: Haskell2010  test-suite test@@ -55,9 +64,13 @@     , tasty-quickcheck     , crdt   other-modules:+      ArbitraryOrphans+      Counter       GCounter       GSet       Laws       LWW+      Max       PNCounter+      TPSet   default-language: Haskell2010

@@ -1,24 +1,76 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE TypeFamilies #-}  module CRDT.Cm     ( CmRDT (..)+    , concurrent     ) where -import           Data.Kind (Type)+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++-- | Not comparable, i. e. ¬(a ≤ b) ∧ ¬(b ≤ a).+concurrent :: PartialOrd a => a -> a -> Bool+concurrent a b = not $ comparable a b+ {- | Operation-based, or commutative (Cm) replicated data type. -[Commutativity law]+== Implementation -    @'update' op1 . 'update' op2 == 'update' op2 . 'update' op1@+In Haskell, a CmRDT implementation consists of 3 types —+a __payload__, an __operation__ (@op@) and an __update__. +[Payload]+    Internal state of a replica.+[Operation]+    User's request to update.+[Update]+    Operation to be applied to other replicas.++For many types /operation/ and /update/ may be the same.+But for 'CRDT.Cm.LWW.LWW', for instance, this rule doesn't hold:+user can request only value, and type attaches a timestamp to it.++== Additional constraint — commutativity law++Concurrent updates are observed equally.++@+∀ up1 up2 s .+'concurrent' up1 up2 ==>+    'observe' ('updateDownstream' up1 . 'updateDownstream' up2 $ s) ==+    'observe' ('updateDownstream' up2 . 'updateDownstream' up1 $ s)+@+ Idempotency doesn't need to hold. -}-class CmRDT op where -    -- | The type of the target value-    type State op :: Type+class (Observe payload, Eq (Observed payload), PartialOrd update)+        => CmRDT payload op update+        | payload -> op, op -> update, update -> payload+        where -    -- | Apply operation to a value-    update :: op -> State op -> State op+    -- | Precondition for 'updateAtSource'.+    -- Calculates if the operation is applicable to the current state.+    updateAtSourcePre :: op -> payload -> 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++    -- | Apply an update to the payload.+    -- An invalid update must be ignored.+    updateDownstream :: update -> payload -> payload

@@ -0,0 +1,41 @@+{-# LANGUAGE FlexibleInstances #-}+{-# 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)++data CounterOp 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 Observe (Counter a) where+    type Observed (Counter a) = a+    observe (Counter c) = c++-- | 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

@@ -0,0 +1,43 @@+{-# OPTIONS_GHC -Wno-orphans #-}++{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}++module CRDT.Cm.GSet+    ( GSet+    , Add (..)+    , initial+    , lookup+    ) 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++newtype Add 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 Observe (GSet a) where+    type Observed (GSet a) = Set a+    observe = id++instance Eq a => PartialOrd (Add a) where+    leq _ _ = False

@@ -0,0 +1,35 @@+{-# 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

@@ -0,0 +1,54 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE TypeFamilies #-}++-- | 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)++data TPSetOp 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) (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}++instance Observe (TPSet a) where+    type Observed (TPSet a) = Set a+    observe = payload++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

@@ -4,7 +4,7 @@     ( CvRDT     ) where -import Data.Semilattice (Semilattice)+import           Data.Semilattice (Semilattice)  {- | State-based, or convergent (Cv) replicated data type.@@ -13,6 +13,7 @@  Query function is not needed. State itself is exposed. In other words, @query = 'id'@.+Some types may offer more convenient query functions.  Actually, a CvRDT is nothing more a 'Semilattice'. -}

@@ -0,0 +1,42 @@+module CRDT.Cv.GCounter+    ( GCounter (..)+    , initial+    , query+    -- * Operation+    , increment+    ) where++import           Data.IntMap.Strict (IntMap)+import qualified Data.IntMap.Strict as IntMap+import           Data.Semigroup (Semigroup ((<>)))++import           CRDT.Cv (CvRDT)++-- | Grow-only counter.+newtype GCounter a = GCounter (IntMap a)+    deriving (Eq, Show)++instance Ord a => Semigroup (GCounter a) where+    GCounter x <> GCounter y = GCounter $ IntMap.unionWith max x y++instance Ord a => CvRDT (GCounter a)++-- | Increment counter+increment+    :: Num a+    => Word -- ^ replica id+    -> GCounter a+    -> GCounter a+increment replicaId (GCounter imap) = GCounter (IntMap.insertWith (+) i 1 imap)+  where+    i = fromIntegral replicaId++-- | Initial state+initial :: GCounter a+initial = GCounter IntMap.empty++-- | Get value from the state+query :: Num a => GCounter a -> a+query (GCounter v) = mapSum v+  where+    mapSum = IntMap.foldr (+) 0

@@ -0,0 +1,28 @@+{-# OPTIONS_GHC -Wno-orphans #-}++module CRDT.Cv.GSet+    ( GSet+    , add+    , initial+    , query+    ) where++import           Data.Semilattice (Semilattice)+import           Data.Set (Set)+import qualified Data.Set as Set++-- | Grow-only set+type GSet = Set++instance Ord a => Semilattice (Set a)++-- | update+add :: Ord a => a -> GSet a -> GSet a+add = Set.insert++-- | initialization+initial :: GSet a+initial = Set.empty++query :: Ord a => a -> GSet a -> Bool+query = Set.member

@@ -0,0 +1,48 @@+{-# 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,19 @@+{-# OPTIONS_GHC -Wno-orphans #-}++module CRDT.Cv.Max+    ( Max (..)+    , point+    , query+    ) where++import           Data.Semigroup (Max (..))+import           Data.Semilattice (Semilattice)++instance Ord a => Semilattice (Max a)++-- | Construct new value+point :: a -> Max a+point = Max++query :: Max a -> a+query = getMax

@@ -0,0 +1,59 @@+{-# LANGUAGE NamedFieldPuns #-}++module CRDT.Cv.PNCounter+    ( PNCounter (..)+    , initial+    , query+    -- * Operations+    , decrement+    , increment+    ) where++import           Data.Semigroup (Semigroup (..))++import           CRDT.Cv (CvRDT)+import           CRDT.Cv.GCounter (GCounter)+import qualified CRDT.Cv.GCounter as GCounter++{- |+Positive-negative counter. Allows incrementing and decrementing.+Nice example of combining of existing CvRDT ('GCounter' in this case)+to create another CvRDT.+-}+data PNCounter a = PNCounter+    { positive :: !(GCounter a)+    , negative :: !(GCounter a)+    }+    deriving (Eq, Show)++instance Ord a => Semigroup (PNCounter a) where+    PNCounter p1 n1 <> PNCounter p2 n2 = PNCounter (p1 <> p2) (n1 <> n2)++instance Ord a => CvRDT (PNCounter a)++-- | Get value from the state+query :: Num a => PNCounter a -> a+query PNCounter{positive, negative} =+    GCounter.query positive - GCounter.query negative++-- | Decrement counter+decrement+    :: Num a+    => Word -- ^ replica id+    -> PNCounter a+    -> PNCounter a+decrement i pnc@PNCounter{negative} =+    pnc{negative = GCounter.increment i negative}++-- | Increment counter+increment+    :: Num a+    => Word -- ^ replica id+    -> PNCounter a+    -> PNCounter a+increment i pnc@PNCounter{positive} =+    pnc{positive = GCounter.increment i positive}++-- | Initial state+initial :: PNCounter a+initial = PNCounter{positive = GCounter.initial, negative = GCounter.initial}

@@ -1,22 +0,0 @@-{-# LANGUAGE TypeFamilies #-}--module CRDT.GCounter.Cm-    ( GCounter (..)-    , initial-    ) where--import           CRDT.Cm (CmRDT, State, update)---- | Grow-only counter.------ Commutativity: 'Increment' obviously commutes with itself.-data GCounter a = Increment-    deriving (Show)--instance Num a => CmRDT (GCounter a) where-    type State (GCounter a) = a-    update _ = (+1)---- | Initial state-initial :: Num a => a-initial = 0

@@ -1,31 +0,0 @@-module CRDT.GCounter.Cv-    ( GCounter-    , initial-    , query-    -- * Operation-    , increment-    ) where--import qualified Data.IntMap.Strict as IntMap--import           CRDT.GCounter.Cv.Internal---- | Increment counter-increment-    :: Num a-    => Word -- ^ replica id-    -> GCounter a-    -> GCounter a-increment replicaId (GCounter imap) = GCounter (IntMap.insertWith (+) i 1 imap)-  where-    i = fromIntegral replicaId---- | Initial state-initial :: GCounter a-initial = GCounter IntMap.empty---- | Get value from the state-query :: Num a => GCounter a -> a-query (GCounter v) = mapSum v-  where-    mapSum = IntMap.foldr (+) 0

@@ -1,16 +0,0 @@-module CRDT.GCounter.Cv.Internal where--import           Data.Semigroup (Semigroup ((<>)))-import           Data.IntMap.Strict (IntMap)-import qualified Data.IntMap.Strict as IntMap--import CRDT.Cv (CvRDT)---- | Grow-only counter.-newtype GCounter a = GCounter (IntMap a)-    deriving (Eq, Show)--instance Ord a => Semigroup (GCounter a) where-    GCounter x <> GCounter y = GCounter $ IntMap.unionWith max x y--instance Ord a => CvRDT (GCounter a)

@@ -1,21 +0,0 @@-module CRDT.GSet.Cv-    ( GSet-    , add-    , initial-    , query-    ) where--import qualified Data.Set as Set--import           CRDT.GSet.Cv.Internal---- | update-add :: Ord a => a -> GSet a -> GSet a-add = Set.insert---- | initialization-initial :: GSet a-initial = Set.empty--query :: Ord a => a -> GSet a -> Bool-query = Set.member

@@ -1,12 +0,0 @@-{-# OPTIONS_GHC -Wno-orphans #-}--module CRDT.GSet.Cv.Internal where--import Data.Set (Set)--import CRDT.Cv  (CvRDT)---- | Grow-only set-type GSet = Set--instance Ord a => CvRDT (Set a)

@@ -1,44 +0,0 @@-{-# LANGUAGE TypeFamilies #-}--module CRDT.LWW-    ( LWW (..)-    , point-    , write-    , query-    ) where--import           Data.Semigroup (Semigroup (..))--import           CRDT.Cm (CmRDT, State)-import qualified CRDT.Cm as Cm-import           CRDT.Cv (CvRDT)-import           CRDT.Timestamp (Timestamp)---- | Last write wins. Interesting, this type is both 'CmRDT' and 'CvRDT'.-data LWW a = Write-    { timestamp :: !Timestamp-    , value     :: !a-    }-    deriving (Eq, Ord, Show)---- | Merge by choosing more recent timestamp.-instance Ord a => Semigroup (LWW a) where-    (<>) = max--instance Ord a => CmRDT (LWW a) where-    type State (LWW a) = LWW a-    update = max--instance Ord a => CvRDT (LWW a)---- | Initialize state-point :: Timestamp -> a -> LWW a-point = Write---- | Change state-write :: Ord a => Timestamp -> a -> LWW a -> LWW a-write t v s = Write t v <> s---- | Query state-query :: LWW a -> a-query = value

@@ -1,23 +0,0 @@-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE TypeFamilies #-}--module CRDT.PNCounter.Cm-    ( PNCounter (..)-    , initial-    ) where--import CRDT.Cm (CmRDT, State, update)---- | Positive-negative counter. Allows incrementing and decrementing.-data PNCounter a = Increment | Decrement-    deriving (Bounded, Enum, Show)--instance Num a => CmRDT (PNCounter a) where-    type State (PNCounter a) = a-    update = \case-        Increment -> (+1)-        Decrement -> subtract 1---- | Initial state-initial :: Num a => a-initial = 0

@@ -1,41 +0,0 @@-{-# LANGUAGE NamedFieldPuns #-}--module CRDT.PNCounter.Cv-    ( PNCounter-    , initial-    , query-    -- * Operations-    , decrement-    , increment-    ) where--import qualified CRDT.GCounter.Cv as GCounter--import CRDT.PNCounter.Cv.Internal---- | Get value from the state-query :: Num a => PNCounter a -> a-query PNCounter{positive, negative} =-    GCounter.query positive - GCounter.query negative---- | Decrement counter-decrement-    :: Num a-    => Word -- ^ replica id-    -> PNCounter a-    -> PNCounter a-decrement i pnc@PNCounter{negative} =-    pnc{negative = GCounter.increment i negative}---- | Increment counter-increment-    :: Num a-    => Word -- ^ replica id-    -> PNCounter a-    -> PNCounter a-increment i pnc@PNCounter{positive} =-    pnc{positive = GCounter.increment i positive}---- | Initial state-initial :: PNCounter a-initial = PNCounter{positive = GCounter.initial, negative = GCounter.initial}

@@ -1,22 +0,0 @@-module CRDT.PNCounter.Cv.Internal where--import Data.Semigroup (Semigroup (..))--import CRDT.Cv          (CvRDT)-import CRDT.GCounter.Cv (GCounter)--{- |-Positive-negative counter. Allows incrementing and decrementing.-Nice example of combining of existing CvRDT ('GCounter' in this case)-to create another CvRDT.--}-data PNCounter a = PNCounter-    { positive :: !(GCounter a)-    , negative :: !(GCounter a)-    }-    deriving (Eq, Show)--instance Ord a => Semigroup (PNCounter a) where-    PNCounter p1 n1 <> PNCounter p2 n2 = PNCounter (p1 <> p2) (n1 <> n2)--instance Ord a => CvRDT (PNCounter a)

@@ -1,7 +0,0 @@-module CRDT.Timestamp-    ( Timestamp-    ) where--import           Numeric.Natural (Natural)--type Timestamp = Natural

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

@@ -1,6 +1,6 @@ module Data.Semilattice     ( Semilattice-    , slappend+    , merge     ) where  import           Data.Semigroup (Semigroup, (<>))@@ -25,7 +25,7 @@ class Semigroup a => Semilattice a  -- | Just ('Semigroup.<>'), specialized to 'Semilattice'.-slappend :: Semilattice a => a -> a -> a-slappend = (<>)-infixr 6 `slappend`-{-# INLINE slappend #-}+merge :: Semilattice a => a -> a -> a+merge = (<>)+infixr 6 `merge`+{-# INLINE merge #-}

@@ -0,0 +1,75 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TypeSynonymInstances #-}++module LamportClock+    ( Pid (..)+    , Time+    , Timestamp (..)+    , Clock (..)+    , LamportClock+    , runLamportClock+    , Process+    , runProcess+    , barrier+    ) where++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           Data.Functor (($>))+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)++-- | TODO(cblp, 2017-09-28) Use bounded-intmap+type LamportClock = State (EnumMap Pid Time)++-- | 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]+        in+        if null selectedClocks then+            clocks+        else+            EnumMap.union+                (selectedClocks $> succ (maximum selectedClocks))+                clocks++-- | Timestamps are assumed unique, totally ordered,+-- and consistent with causal order;+-- i.e., if assignment 1 happened-before assignment 2,+-- the former’s timestamp is less than the latter’s.+data Timestamp = Timestamp !Time !Pid+    deriving (Eq, Ord, Show)++class Applicative f => Clock f where+    -- | Get another unique timestamp+    newTimestamp :: f Timestamp++type Process = ReaderT Pid LamportClock++instance Clock Process where+    newTimestamp = do+        pid <- ask+        time <- at pid . non 0 <<+= 1+        pure $ Timestamp time pid++runLamportClock :: LamportClock a -> a+runLamportClock action = evalState action mempty++runProcess :: Pid -> Process a -> LamportClock a+runProcess pid action = runReaderT action pid

@@ -0,0 +1,29 @@+{-# 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 #-}

@@ -0,0 +1,47 @@+{-# OPTIONS_GHC -Wno-orphans #-}++{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE StandaloneDeriving #-}++module ArbitraryOrphans () where++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 qualified CRDT.Cm.TPSet as TPSet+import           CRDT.Cv.GCounter (GCounter (..))+import           CRDT.Cv.Max (Max (..))+import           CRDT.Cv.PNCounter (PNCounter (..))+import           LamportClock (Pid (..), Timestamp (..))++deriving instance Arbitrary a => Arbitrary (Counter a)++instance Arbitrary (CounterOp 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)++instance Arbitrary a => Arbitrary (TPSetOp a) where+    arbitrary = oneof [TPSet.Add <$> arbitrary, Remove <$> arbitrary]++deriving instance Arbitrary a => Arbitrary (GCounter a)++deriving instance Arbitrary a => Arbitrary (Max a)++instance Arbitrary a => Arbitrary (PNCounter a) where+    arbitrary = PNCounter <$> arbitrary <*> arbitrary++deriving instance Arbitrary Pid++instance Arbitrary Timestamp where+    arbitrary = Timestamp <$> arbitrary <*> arbitrary

@@ -0,0 +1,13 @@+{-# LANGUAGE TypeApplications #-}++module Counter+    ( counter+    ) where++import           CRDT.Cm.Counter (Counter)+import           Test.Tasty (TestTree, testGroup)++import           Laws (cmrdtLaw)++counter :: TestTree+counter = testGroup "Counter" [cmrdtLaw @(Counter Int)]

@@ -1,42 +1,21 @@-{-# OPTIONS_GHC -Wno-orphans #-}--{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TypeApplications #-}  module GCounter     ( gCounter     ) where -import           Test.QuickCheck (Arbitrary, arbitrary) import           Test.Tasty (TestTree, testGroup) import           Test.Tasty.QuickCheck (testProperty) -import           CRDT.Cm (update)-import qualified CRDT.GCounter.Cm as Cm-import qualified CRDT.GCounter.Cv as Cv-import qualified CRDT.GCounter.Cv.Internal as Cv--import           Laws (cmrdtLaws, cvrdtLaws)--instance Arbitrary (Cm.GCounter a) where-    arbitrary = pure Cm.Increment+import           CRDT.Cv.GCounter (GCounter (..), increment, query) -deriving instance Arbitrary a => Arbitrary (Cv.GCounter a)+import           Laws (cvrdtLaws)  gCounter :: TestTree gCounter = testGroup "GCounter"-    [ testGroup "Cv"-        [ cvrdtLaws @(Cv.GCounter Int)-        , testProperty "increment" $-            \(counter :: Cv.GCounter Int) i ->-                Cv.query (Cv.increment i counter)-                == succ (Cv.query counter)-        ]-    , testGroup "Cm"-        [ cmrdtLaws @(Cm.GCounter Int)-        , testProperty "increment" $-            \(counter :: Int) -> update Cm.Increment counter == succ counter-        ]+    [ cvrdtLaws @(GCounter Int)+    , testProperty "increment" $+        \(counter :: GCounter Int) i ->+            query (increment i counter) == succ (query counter)     ]

@@ -1,5 +1,3 @@-{-# OPTIONS_GHC -Wno-orphans #-}- {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeApplications #-} @@ -10,16 +8,15 @@ import           Test.Tasty (TestTree, testGroup) import           Test.Tasty.QuickCheck (testProperty, (==>)) -import qualified CRDT.GSet.Cv as Cv+import           CRDT.Cm.GSet (GSet)+import           CRDT.Cv.GSet (add, query) -import           Laws (cvrdtLaws)+import           Laws (cmrdtLaw, cvrdtLaws)  gSet :: TestTree gSet = testGroup "GSet"-    [ testGroup "Cv"-        [ cvrdtLaws @(Cv.GSet Int)-        , testProperty "add" $-            \(set :: Cv.GSet Int) i ->-                not (Cv.query i set) ==> Cv.query i (Cv.add i set)-        ]+    [ cmrdtLaw @(GSet Int)+    , cvrdtLaws @(GSet Int)+    , testProperty "Cv.add" $+        \(set :: GSet Int) i -> not (query i set) ==> query i (add i set)     ]

@@ -1,57 +1,38 @@-{-# OPTIONS_GHC -Wno-orphans #-}- {-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeApplications #-}  module LWW (lww) where -import           Test.QuickCheck (Arbitrary, arbitrary)+import           Data.Semigroup ((<>)) import           Test.Tasty (TestTree, testGroup)-import           Test.Tasty.QuickCheck (Positive (..), testProperty)--import           CRDT.Cm (update)-import           CRDT.LWW (LWW (Write), point, query, write)+import           Test.Tasty.QuickCheck (testProperty) -import           Laws (cmrdtLaws, cvrdtLaws)+import           CRDT.Cm.LWW (LWW (..))+import           CRDT.Cv.LWW (assign, initial, query)+import           LamportClock (barrier, runLamportClock, runProcess) -instance Arbitrary a => Arbitrary (LWW a) where-    arbitrary = Write <$> arbitrary <*> arbitrary+import           Laws (cmrdtLaw, cvrdtLaws)  lww :: TestTree lww = testGroup "LWW"-    [ testGroup "Cm"-        [ cmrdtLaws @(LWW Int)-        , testProperty "write latter" $-            \formerTime (formerValue :: Int) (Positive dt) latterValue -> let-                latterTime = formerTime + dt-                state = point formerTime formerValue-                state' = update (Write latterTime latterValue) state-                in-                query state' == latterValue-        , testProperty "write former" $-            \formerTime (formerValue :: Int) (Positive dt) latterValue -> let-                latterTime = formerTime + dt-                state = point latterTime latterValue-                state' = update (Write formerTime formerValue) state-                in-                query state' == latterValue-        ]+    [ cmrdtLaw @(LWW Int)     , testGroup "Cv"         [ cvrdtLaws @(LWW Int)-        , testProperty "write latter" $-            \formerTime (formerValue :: Int) (Positive dt) latterValue -> let-                latterTime = formerTime + dt-                state = point formerTime formerValue-                state' = write latterTime latterValue state-                in-                query state' == latterValue-        , testProperty "write former" $-            \formerTime (formerValue :: Int) (Positive dt) latterValue -> let-                latterTime = formerTime + dt-                state = point latterTime latterValue-                state' = write formerTime formerValue state-                in-                query state' == latterValue+        , testProperty "assign" $+            \pid1 pid2 (formerValue :: Int) latterValue ->+                runLamportClock $ do+                    state1 <- runProcess pid1 $ initial formerValue+                    barrier [pid1, pid2]+                    state2 <- runProcess pid2 $ assign latterValue state1+                    pure $ query state2 == latterValue+        , testProperty "merge with former" $+            \pid1 pid2 (formerValue :: Int) latterValue ->+                runLamportClock $ do+                    state1 <- runProcess pid1 $ initial formerValue+                    barrier [pid1, pid2]+                    state2 <- runProcess pid2 $ initial latterValue+                    pure $ query (state1 <> state2) == latterValue         ]     ]

@@ -1,30 +1,34 @@ {-# LANGUAGE AllowAmbiguousTypes #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeApplications #-}  module Laws-    ( cmrdtLaws+    ( cmrdtLaw     , cvrdtLaws-    , semigroupLaw-    , semilatticeLaws     ) where +import           Control.Monad (unless)+import           Data.Function ((&))+import           Data.Observe (Observe (..)) import           Data.Semigroup (Semigroup, (<>))-import           Data.Semilattice (Semilattice, slappend)-import           Test.QuickCheck (Arbitrary)+import           Data.Semilattice (Semilattice, merge) import           Test.Tasty (TestTree, testGroup)-import           Test.Tasty.QuickCheck (testProperty)+import           Test.Tasty.QuickCheck (Arbitrary (..), discard, testProperty,+                                        (===), (==>)) -import           CRDT.Cm (CmRDT, State, update)+import           CRDT.Cm (CmRDT (..), concurrent) import           CRDT.Cv (CvRDT)+import           LamportClock (Clock, runLamportClock, runProcess) +import           ArbitraryOrphans ()+ semigroupLaw :: forall a . (Arbitrary a, Semigroup a, Eq a, Show a) => TestTree semigroupLaw =     testGroup "Semigroup law" [testProperty "associativity" associativity]   where-    associativity :: a -> a -> a -> Bool-    associativity x y z = (x <> y) <> z == x <> (y <> z)+    associativity (x, y, z :: a) = (x <> y) <> z == x <> (y <> z)  semilatticeLaws     :: forall a . (Arbitrary a, Semilattice a, Eq a, Show a) => TestTree@@ -34,23 +38,29 @@     , testProperty "idempotency"   idempotency     ]   where-    commutativity :: a -> a -> Bool-    commutativity x y = x `slappend` y == y `slappend` x--    idempotency :: a -> Bool-    idempotency x = x `slappend` x == x+    commutativity (x, y :: a) = x `merge` y === y `merge` x+    idempotency (x :: a) = x `merge` x === x  cvrdtLaws :: forall a . (Arbitrary a, CvRDT a, Eq a, Show a) => TestTree cvrdtLaws = semilatticeLaws @a -cmrdtLaws-    :: forall op-    . ( Arbitrary op, CmRDT op, Show op-      , Arbitrary (State op), Eq (State op), Show (State op)+cmrdtLaw+    :: forall payload op up+    . ( CmRDT payload op up+      , Arbitrary payload, Show payload+      , Arbitrary op, Show op+      , Show (Observed payload)       )     => TestTree-cmrdtLaws = testProperty "CmRDT law: commutativity" commutativity+cmrdtLaw = testProperty "CmRDT law: concurrent ops commute" $+    \pid (state0 :: payload) (op1, op2 :: op) ->+        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   where-    commutativity :: op -> op -> State op -> Bool-    commutativity op1 op2 x =-        (update op1 . update op2) x == (update op2 . update op1) x+    updateAtSourceIfCan :: Clock m => op -> payload -> m up+    updateAtSourceIfCan op state =+        unless (updateAtSourcePre op state) discard *> updateAtSource op

@@ -1,9 +1,13 @@ import           Test.Tasty (defaultMain, testGroup) +import           Counter (counter) import           GCounter (gCounter)+import           GSet (gSet) import           LWW (lww)+import           Max (maxTest) import           PNCounter (pnCounter)-import           GSet      (gSet)+import           TPSet (tpSet)  main :: IO ()-main = defaultMain $ testGroup "" [gCounter, gSet, lww, pnCounter]+main = defaultMain $+    testGroup "" [counter, gCounter, gSet, lww, maxTest, pnCounter, tpSet]

@@ -0,0 +1,23 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module Max+    ( maxTest+    ) where++import           Test.Tasty (TestTree, testGroup)+import           Test.Tasty.QuickCheck (testProperty)++import           CRDT.Cv.Max (Max, point, query)+import           Data.Semilattice (merge)++import           Laws (cvrdtLaws)++maxTest :: TestTree+maxTest = testGroup "Max"+    [ testGroup "Cv"+        [ cvrdtLaws @(Max Int)+        , testProperty "merge" $+            \(m :: Max Int) i -> query (point i `merge` m) == max i (query m)+        ]+    ]

@@ -1,5 +1,3 @@-{-# OPTIONS_GHC -Wno-orphans #-}- {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeApplications #-} @@ -7,40 +5,21 @@     ( pnCounter     ) where -import           Test.QuickCheck (Arbitrary, arbitrary, arbitraryBoundedEnum) import           Test.Tasty (TestTree, testGroup) import           Test.Tasty.QuickCheck (testProperty) -import           CRDT.Cm (update)-import qualified CRDT.PNCounter.Cm as Cm-import qualified CRDT.PNCounter.Cv as Cv-import qualified CRDT.PNCounter.Cv.Internal as Cv+import           CRDT.Cv.PNCounter (PNCounter (..), decrement, increment, query)  import           GCounter ()-import           Laws (cmrdtLaws, cvrdtLaws)--instance Arbitrary (Cm.PNCounter a) where-    arbitrary = arbitraryBoundedEnum--instance Arbitrary a => Arbitrary (Cv.PNCounter a) where-    arbitrary = Cv.PNCounter <$> arbitrary <*> arbitrary+import           Laws (cvrdtLaws)  pnCounter :: TestTree pnCounter = testGroup "PNCounter"-    [ testGroup "Cv"-        [ cvrdtLaws @(Cv.PNCounter Int)-        , testProperty "increment" $-            \(counter :: Cv.PNCounter Int) i ->-                Cv.query (Cv.increment i counter) == succ (Cv.query counter)-        , testProperty "decrement" $-            \(counter :: Cv.PNCounter Int) i ->-                Cv.query (Cv.decrement i counter) == pred (Cv.query counter)-        ]-    , testGroup "Cm"-        [ cmrdtLaws @(Cm.PNCounter Int)-        , testProperty "increment" $-            \(counter :: Int) -> update Cm.Increment counter == succ counter-        , testProperty "decrement" $-            \(counter :: Int) -> update Cm.Decrement counter == pred counter-        ]+    [ cvrdtLaws @(PNCounter Int)+    , testProperty "increment" $+        \(counter :: PNCounter Int) i ->+            query (increment i counter) == succ (query counter)+    , testProperty "decrement" $+        \(counter :: PNCounter Int) i ->+            query (decrement i counter) == pred (query counter)     ]

@@ -0,0 +1,13 @@+{-# LANGUAGE TypeApplications #-}++module TPSet+    ( tpSet+    ) where++import           CRDT.Cm.TPSet (TPSet)+import           Test.Tasty (TestTree, testGroup)++import           Laws (cmrdtLaw)++tpSet :: TestTree+tpSet = testGroup "TPSet" [cmrdtLaw @(TPSet Int)]