Quelea (empty) → 1.0.0

raw patch · 21 files changed

+3252/−0 lines, 21 filesdep +basedep +bytestringdep +cassandra-cqlsetup-changed

Dependencies added: base, bytestring, cassandra-cql, cereal, containers, derive, directory, lens, mtl, optparse-applicative, process, random, template-haskell, text, time, transformers, tuple, unix, uuid, z3, zeromq4-haskell

Files

• LICENSE +24/−0
• Quelea.cabal +89/−0
• Quelea/Client.hs +220/−0
• Quelea/ClientMonad.hs +97/−0
• Quelea/Consts.hs +30/−0
• Quelea/Contract.hs +47/−0
• Quelea/Contract/Language.hs +208/−0
• Quelea/Contract/TypeCheck.hs +556/−0
• Quelea/DBDriver.hs +296/−0
• Quelea/Marshall.hs +144/−0
• Quelea/NameService/LoadBalancingBroker.hs +66/−0
• Quelea/NameService/SimpleBroker.hs +95/−0
• Quelea/NameService/Types.hs +16/−0
• Quelea/Shim.hs +222/−0
• Quelea/ShimLayer/Cache.hs +227/−0
• Quelea/ShimLayer/GC.hs +220/−0
• Quelea/ShimLayer/Types.hs +53/−0
• Quelea/ShimLayer/UpdateFetcher.hs +410/−0
• Quelea/TH.hs +52/−0
• Quelea/Types.hs +178/−0
• Setup.hs +2/−0

@@ -0,0 +1,24 @@+Copyright (c) 2014, KC Sivaramakrishnan+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:+    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.+    * Redistributions in binary form must reproduce the above copyright+      notice, this list of conditions and the following disclaimer in the+      documentation and/or other materials provided with the distribution.+    * Neither the name of the <organization> nor the+      names of its contributors may be used to endorse or promote products+      derived from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER BE LIABLE FOR ANY+DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,89 @@+Name:           Quelea+Version:        1.0.0+Cabal-Version:  >= 1.2+License:        BSD3+License-File:   LICENSE+Author:         KC Sivaramakrishnan <http://kcsrk.info>+Maintainer:     Gowtham Kaki <http://gowthamk.github.io>+Copyright:      2014, KC Sivaramakrishnan+Category:       Experimental+Synopsis:       Programming with Eventual Consistency over Cassandra.+Description:+    Quelea is a Haskell library that helps programmers develop highly+    scalable applications on the top of eventually consistent NoSQL stores,+    such as Cassandra. You can think of Quelea as a mediator between the+    application programmer and the underlying NoSQL store. It understands both+    the application requirements and the store semantics, and helps+    programmers use NoSQL stores in such a way so as to maximize performance+    and ensure correct application behaviour, both at the same time!  Among+    other things, Quelea library implements:+    .+    A Domain-Specific Language (DSL) that lets NoSQL application programmers+    declare the consistency requirements of their applications as contracts+    (also called specifications)+    .+    A static contract classification procedure that maps high-level+    application contracts to appropriate low-level consistency guarantees+    provided by the underlying NoSQL store, and+    .+    A run-time that runs application operations after tuning the+    store consistency to appropriate levels as determined by the+    contract classification procedure.  While the+    implementations of DSL and the static contract+    classification components are largely independent of the+    actual NoSQL store used, the current implementation of+    run-time component is tailor-made to work with the Cassandra+    data store.+Homepage:       http://gowthamk.github.io/Quelea+build-type:     Simple++Library+  Build-Depends:+    cassandra-cql >= 0.5.0.1  && < 1,+    base > 3 && < 5,+    bytestring,+    zeromq4-haskell,+    cereal,+    containers,+    lens,+    template-haskell,+    z3 >= 4.0.0,+    mtl,+    random,+    uuid,+    text,+    transformers,+    time,+    unix,+    directory,+    tuple,+    derive,+    process,+    optparse-applicative+  Exposed-modules:+    Quelea.NameService.Types+    Quelea.NameService.SimpleBroker+    Quelea.NameService.LoadBalancingBroker+    Quelea.Marshall+    Quelea.Shim+    Quelea.Client+    Quelea.ClientMonad+    Quelea.Types+    Quelea.TH+    Quelea.Contract+    Quelea.DBDriver+  Other-modules:+    Quelea.Contract.Language+    Quelea.Contract.TypeCheck+    Quelea.Consts+    Quelea.ShimLayer.Types+    Quelea.ShimLayer.Cache+    Quelea.ShimLayer.UpdateFetcher+    Quelea.ShimLayer.GC+  Extensions: CPP+  ghc-options: -w ++-- Executable LWW_txn+-- main-is:+--    LWW_txn.hs+--  ghc-options: -prof -XCPP -O2 -threaded -osuf p_o -hisuf p_hi -fprof-auto "-with-rtsopts=-N -p -s -h -i0.1"

@@ -0,0 +1,220 @@+{-# LANGUAGE TemplateHaskell, ScopedTypeVariables #-}++module Quelea.Client (+  Key,+  Session,+  Availability(..),+  TxnKind(..),++  beginSession,+  beginSessionStats,+  getStats,+  endSession,+  invoke,+  invokeAndGetDeps,+  newKey,+  mkKey,+  getServerAddr,+  beginTxn,+  endTxn,+  getLastEffect+) where++import Data.UUID+import Quelea.Types+import Quelea.NameService.Types+import Control.Lens+import Control.Monad (when)+import System.ZMQ4+import Data.Serialize+import Quelea.Marshall+import System.Random (randomIO)+import Control.Applicative+import Data.ByteString (ByteString, length)+import qualified Data.ByteString as B+import qualified Data.Map as M+import qualified Data.Set as S+import Data.Maybe (fromJust)+import Data.Tuple.Select+import Debug.Trace+import Data.Time++type Effect = ByteString+++data TxnState = TxnState {+  _txnidTS    :: TxnID,+  _txnKindTS  :: TxnKind,+  _txnEffsTS  :: S.Set TxnDep,+  _snapshotTS :: M.Map (ObjType, Key) (S.Set (Addr, Effect)),+  _seenTxnsTS :: S.Set TxnID+} deriving Eq++makeLenses ''TxnState++data Session = Session {+  _broker     :: Frontend,+  _server     :: Socket Req,+  _serverAddr :: String,+  _sessid     :: SessID,+  _seqMap     :: M.Map (ObjType, Key) SeqNo,+  _readObjs   :: S.Set (ObjType, Key),++  _collectStats :: Bool,+  _numOps       :: Int,+  _startTime    :: UTCTime,+  _avgLat       :: NominalDiffTime,++  _curTxn     :: Maybe TxnState,+  _lastEffect :: Maybe TxnDep++}++makeLenses ''Session++getStats :: Session -> IO (Double, NominalDiffTime)+getStats s =+  if s^.collectStats+  then do+    now <- getCurrentTime+    let diffTime = diffUTCTime now $ s^.startTime+    let d::Double = realToFrac $ (fromIntegral $ s^.numOps) / diffTime+    return (d, s^.avgLat)+  else return $ (0.0,0)++beginSessionStats :: NameService -> Bool -> IO Session+beginSessionStats ns getStats = do+  (serverAddr, sock) <- getClientJoin ns+  -- Create a session id+  sessid <- SessID <$> randomIO+  currentTime <- getCurrentTime+  -- Initialize session+  return $ Session (getFrontend ns) sock serverAddr sessid M.empty S.empty getStats 0 currentTime 0 Nothing Nothing++beginSession :: NameService -> IO Session+beginSession ns = beginSessionStats ns False++endSession :: Session -> IO ()+endSession s = disconnect (s ^. server) (s^.serverAddr)++beginTxn :: Session -> TxnKind -> IO Session+beginTxn s tk = do+  when (s^.curTxn /= Nothing) $ error "beginTxn: Nested transactions are not supported!"+  txnID <- TxnID <$> randomIO+  snapshot <-+    if (tk == RR)+    then do+      let req :: Request () = ReqSnapshot $ s^.readObjs+      let encodedReq = encode req+      -- putStrLn $ "beginTxn: req length=" ++ show (B.length encodedReq)+      send (s^.server) [] encodedReq+      responseBlob <- receive (s^.server)+      -- putStrLn $ "beginTxn: res length=" ++ show (B.length responseBlob)+      let (ResSnapshot s) = decodeResponse responseBlob+      return $ Just s+    else return Nothing+  return $ s {_curTxn = Just $ TxnState txnID tk S.empty M.empty S.empty}++endTxn :: S.Set TxnDep {- Extra dependencies -} -> Session -> IO Session+endTxn extraDeps s = do+  when (s^.curTxn == Nothing) $ error "endTxn: No transaction in progress!"+  let txnState = case s^.curTxn of {Nothing -> error "endTxn"; Just st -> st}+  let txnCommit :: Request () = ReqTxnCommit (txnState^.txnidTS) (S.union extraDeps $ txnState^.txnEffsTS)+  send (s^.server) [] $ encode txnCommit+  receive (s^.server)+  return $ s {_curTxn = Nothing}++getServerAddr :: Session -> String+getServerAddr s = s^.serverAddr++invoke :: (OperationClass on, Serialize arg, Serialize res)+       => Session -> Key -> on -> arg -> IO (res, Session)+invoke s k on arg = do+  startTime <- getCurrentTime+  (r, deps, s) <- invokeInternal False s k on arg+  endTime <- getCurrentTime+  let newOpsCnt = s^.numOps + 1+  let newAvgLat = (s^.avgLat * (fromIntegral $ s^.numOps) + (diffUTCTime endTime startTime)) / (fromIntegral $ s^.numOps + 1)+  return (r,s {_numOps = newOpsCnt, _avgLat = newAvgLat})++invokeAndGetDeps :: (OperationClass on, Serialize arg, Serialize res)+       => Session -> Key -> on -> arg -> IO (res, S.Set TxnDep, Session)+invokeAndGetDeps = invokeInternal True++invokeInternal :: (OperationClass on, Serialize arg, Serialize res)+       => Bool -> Session -> Key -> on -> arg -> IO (res, S.Set TxnDep, Session)+invokeInternal getDeps s key operName arg = do+  let ot = getObjType operName+  let seqNo = case M.lookup (ot, key) $ s^.seqMap of+                Nothing -> 0+                Just s -> s+  let txnReq = mkTxnReq ot key $ s^.curTxn+  let req = encode $ ReqOper $ OperationPayload ot key operName (encode arg)+                        (s ^. sessid) seqNo txnReq getDeps+  -- putStrLn $ "invokeInternal: req length=" ++ show (B.length req)+  send (s^.server) [] req+  responseBlob <- receive (s^.server)+  -- putStrLn $ "invokedInternal: res length=" ++ show (B.length responseBlob)+  let (ResOper newSeqNo resBlob mbNewEff mbTxns visAddrSet) = decodeResponse responseBlob+  let visSet = S.map (\(Addr sid sqn) -> TxnDep ot key sid sqn) visAddrSet+  case decode resBlob of+    Left s -> error $ "invoke : decode failure " ++ s+    Right res -> do+      let newSeqMap = M.insert (ot, key) newSeqNo $ s^.seqMap+      {- XXX KC -}+      let newReadObjs = if (S.size $ s^.readObjs) > 32+                        then S.insert (ot,key) $ S.deleteMin $ s^.readObjs+                        else S.insert (ot,key) $ s^.readObjs+      let partialSessRV = Session (s^.broker) (s^.server) (s^.serverAddr) (s^.sessid) newSeqMap newReadObjs+                                  (s^.collectStats) (s^.numOps) (s^.startTime) (s^.avgLat)+      let newLastEff = Just $ TxnDep ot key (s^.sessid) newSeqNo+      case s^.curTxn of+        Nothing -> {- This operation was not in a transaction -}+          if newSeqNo == seqNo {- This operation was read only -}+          then return (res, visSet, partialSessRV Nothing Nothing)+          else return (res, visSet, partialSessRV Nothing newLastEff)+        Just orig@(TxnState txid txnKind deps cache seenTxns) -> do+          case mbNewEff of+            Nothing -> {- This operation was read only -}+              return (res, visSet, partialSessRV (Just orig) Nothing)+            Just newEff -> do+              let newDeps = S.insert (TxnDep ot key (s^.sessid) newSeqNo) deps+              let (_, txnPayload) = case txnReq of {Nothing -> error "invokeInternal"; Just x -> x}+              let newSeenTxns = case mbTxns of+                              Nothing -> seenTxns+                              Just ts -> S.union ts seenTxns+              let newCache =+                    case getEffectSet txnPayload of+                      Nothing -> cache+                      Just es -> M.insert (ot, key) (S.insert (Addr (s^.sessid) newSeqNo, newEff) es) cache+              let newTxnState = (Just (TxnState txid txnKind newDeps newCache newSeenTxns))+              return (res, visSet, partialSessRV newTxnState newLastEff)+  where+    getEffectSet (RC_TxnPl es) = Just es+    getEffectSet (MAV_TxnPl es _) = Just es+    getEffectSet (RR_TxnPl es) = Just es++    mkTxnReq ot k Nothing = Nothing+    mkTxnReq ot k (Just ts) =+      let cache = ts^.snapshotTS+          txid = ts^.txnidTS+          es = case M.lookup (ot,k) cache of+                 Nothing -> S.empty+                 Just s -> s+      in Just $ case ts^.txnKindTS of+           RC ->+             (txid, RC_TxnPl es)+           RR ->+             (txid, RR_TxnPl es)+           MAV -> (txid, MAV_TxnPl es $ ts^.seenTxnsTS)++newKey :: IO Key+newKey = Key . encodeUUID <$> randomIO+  where+    encodeUUID (uuid :: UUID) = encode uuid++mkKey :: Serialize a => a -> Key+mkKey kv = Key $ encode kv++getLastEffect :: Session -> Maybe TxnDep+getLastEffect s = s^.lastEffect

@@ -0,0 +1,97 @@+{-# LANGUAGE TemplateHaskell #-}++module Quelea.ClientMonad (+  Key,+  Session,+  Availability(..),+  TxnKind(..),+  TxnDep,+  CSN,++  runSession,+  runSessionWithStats,+  getStats,+  invoke,+  invokeAndGetDeps,+  newKey,+  mkKey,+  atomically,+  atomicallyWith,+  getServerAddr,+  getLastEffect+) where++import Quelea.Types+import Quelea.Client hiding (invoke, getServerAddr, invokeAndGetDeps, getLastEffect, getStats)+import qualified Quelea.Client as CCLow+import Control.Monad.Trans.State+import Control.Monad.Trans (liftIO)+import Control.Lens+import Quelea.NameService.Types+import Data.Serialize hiding (get, put)+import qualified Data.Set as S+import qualified System.ZMQ4 as ZMQ4+import Data.Time++makeLenses ''Session++type CSN a = StateT Session IO a++runSession :: NameService -> CSN a -> IO a+runSession ns comp = do+  session <- beginSession ns+  res <- evalStateT comp session+  endSession session+  return res++runSessionWithStats :: NameService -> CSN a -> IO a+runSessionWithStats ns comp = do+  session <- beginSessionStats ns True+  res <- evalStateT comp session+  endSession session+  return res++getStats :: CSN (Double, NominalDiffTime)+getStats = do+  session <- get+  liftIO $ CCLow.getStats session++invoke :: (OperationClass on, Serialize arg, Serialize res)+       => Key -> on -> arg -> CSN res+invoke key operName arg = do+  session <- get+  (res, newSession) <- liftIO $ CCLow.invoke session key operName arg+  put newSession+  return res++invokeAndGetDeps :: (OperationClass on, Serialize arg, Serialize res)+       => Key -> on -> arg -> CSN (res, S.Set TxnDep)+invokeAndGetDeps key operName arg = do+  session <- get+  (res, deps, newSession) <- liftIO $ CCLow.invokeAndGetDeps session key operName arg+  put newSession+  return (res, deps)++getServerAddr :: CSN String+getServerAddr = do+  s <- use serverAddr+  return s++atomically :: TxnKind -> CSN a -> CSN a+atomically tk m = do+  get >>= liftIO . (flip beginTxn tk) >>= put+  r <- m+  get >>= liftIO . (endTxn S.empty) >>= put+  return r++atomicallyWith :: S.Set TxnDep -> TxnKind -> CSN a -> CSN a+atomicallyWith extraDeps tk m = do+  get >>= liftIO . (flip beginTxn tk) >>= put+  r <- m+  get >>= liftIO . (endTxn extraDeps) >>= put+  return r++getLastEffect :: CSN (Maybe TxnDep)+getLastEffect = do+  s <- get+  return $ s^.lastEffect

@@ -0,0 +1,30 @@+module Quelea.Consts (+  cCACHE_LWM,+  cDISK_LWM,+  cGC_WORKER_THREAD_DELAY,+  cCACHE_THREAD_DELAY,+  cCACHE_MAX_OBJS,+  cNUM_WORKERS,+  cLOCK_DELAY+) where++cCACHE_LWM :: Int+cCACHE_LWM = 128++cDISK_LWM :: Int+cDISK_LWM = 1024++cGC_WORKER_THREAD_DELAY :: Int+cGC_WORKER_THREAD_DELAY = 1000000++cCACHE_THREAD_DELAY :: Int+cCACHE_THREAD_DELAY = 1000000++cNUM_WORKERS :: Int+cNUM_WORKERS = 8++cLOCK_DELAY :: Int+cLOCK_DELAY = 10000++cCACHE_MAX_OBJS :: Int+cCACHE_MAX_OBJS = 1024

@@ -0,0 +1,47 @@+module Quelea.Contract (+  Rel(..),+  Prop(..),+  EffCol(..),+  Fol,+  Contract,+  Effect,++  true,+  vis,+  so,+  soo,+  sameEff,+  appRel,+  sameObj,+  sameObjList,+  trans,+  hb,+  hbo,+  (∩),+  (∪),+  (∧),+  (∨),+  (⇒),+  (^+),++  liftProp,+  forall_,+  forall2_,+  forall3_,+  forall4_,+  forallQ_,+  forallQ2_,+  forallQ3_,+  forallQ4_,+  isValid,+  isSat,+++  -- Only for DEBUG+  fol2Z3Ctrt,+  underMAV,+  dummyZ3Sort+) where++import Quelea.Contract.Language+import Quelea.Contract.TypeCheck

@@ -0,0 +1,208 @@+{-# LANGUAGE TemplateHaskell, ScopedTypeVariables #-}++module Quelea.Contract.Language (+  Rel(..),+  Prop(..),+  Fol(..),+  Contract,+  Effect(..),+  EffCol(..),+  Z3CtrtState(..),+  Z3Ctrt(..),++  true,+  vis,+  so,+  soo,+  hb,+  hbo,+  sameEff,+  atomDep,+  sameObj,+  sameObjList,+  (∩),+  (∪),+  (∧),+  (∨),+  (⇒),+  (^+),+  appRel,+  liftProp,+  forall_,+  forall2_,+  forall3_,+  forall4_,+  forallQ_,+  forallQ2_,+  forallQ3_,+  forallQ4_,++  trans+) where+++import Quelea.Types+import Z3.Monad+import qualified Data.Map as M+import Control.Monad.State++-------------------------------------------------------------------------------+-- Types++data Z3CtrtState = Z3CtrtState {+  -- Sorts+  _effSort  :: Sort,+  _operSort :: Sort,+  -- Relations+  _visRel     :: FuncDecl, -- Effect -> Effect -> Bool+  _soRel      :: FuncDecl, -- Effect -> Effect -> Bool+  _sameobjRel :: FuncDecl, -- Effect -> Effect -> Bool+  _atomDepRel  :: FuncDecl, -- Effect -> Effect -> Bool+  _curTxnRel  :: FuncDecl, -- Effect -> Bool+  _operRel    :: FuncDecl, -- Effect -> Oper+  -- Map+  _effMap :: M.Map Effect AST,+  -- Assertions+  _assertions :: [AST],+  -- TC Rels+  _tcRelMap :: M.Map Rel FuncDecl+}++data Z3Ctrt = Z3Ctrt { unZ3Ctrt :: StateT Z3CtrtState Z3 AST }++newtype Effect = Effect { unEffect :: Int } deriving (Eq, Ord)++data Rel = Vis | So | SameObj | TC Rel | SameEff | AtomDep+         | Union Rel Rel | Intersect Rel Rel deriving Ord++instance Eq Rel where+  rel1 == rel2 =+    case (rel1, rel2) of+      (Vis, Vis) -> True+      (So, So) -> True+      (SameObj, SameObj) -> True+      (AtomDep, AtomDep) -> True+      (TC r1, TC r2) -> r1 == r2+      (SameEff, SameEff) -> True+      (Union r1 r2, Union r3 r4) ->+        (r1 == r3 && r2 == r4) || (r2 == r3 && r1 == r4)+      (Intersect r1 r2, Intersect r3 r4) ->+        (r1 == r3 && r2 == r4) || (r2 == r3 && r1 == r4)+      otherwise -> False+++data OperationClass a => Prop a =+    PTrue | Not (Prop a) | AppRel Rel Effect Effect | Conj (Prop a) (Prop a) | CurTxn Effect+  | Disj (Prop a) (Prop a) | Impl (Prop a) (Prop a) | Oper Effect a | Raw Z3Ctrt+data OperationClass a => Fol a = Forall [a] (Effect -> Fol a) | Plain (Prop a)++type Contract a = Effect -> Fol a++-------------------------------------------------------------------------------+-- Contract builder++true :: Prop a+true = PTrue++vis :: Effect -> Effect -> Prop a+vis a b = AppRel Vis a b++so :: Effect -> Effect -> Prop a+so a b = AppRel So a b++soo :: Effect -> Effect -> Prop a+soo a b = AppRel (So ∩ SameObj) a b++hb :: Effect -> Effect -> Prop a+hb a b = AppRel (TC $ Union Vis So) a b++hbo :: Effect -> Effect -> Prop a+hbo a b = AppRel (TC $ Union Vis (So ∩ SameObj)) a b++atomDep :: Effect -> Effect -> Prop a+atomDep = AppRel AtomDep++sameObj :: Effect -> Effect -> Prop a+sameObj a b = AppRel SameObj a b++sameObjList :: OperationClass a => [Effect] -> Prop a+sameObjList [] = PTrue+sameObjList (x:[]) = PTrue+sameObjList (x:y:tail) =+  let p = AppRel SameObj x y+  in p ∧ (sameObjList $ y:tail)++(∪) :: Rel -> Rel -> Rel+a ∪ b = Union a b++(∩) :: Rel -> Rel -> Rel+a ∩ b = Intersect a b++(∧) :: OperationClass a => Prop a -> Prop a -> Prop a+(∧) = Conj++(∨) :: OperationClass a => Prop a -> Prop a -> Prop a+(∨) = Disj++(⇒) :: OperationClass a => Prop a -> Prop a -> Prop a+(⇒) = Impl++(^+) :: Rel -> Rel+(^+) = TC++appRel :: OperationClass a => Rel -> Effect -> Effect -> Prop a+appRel = AppRel++liftProp :: OperationClass a => Prop a -> Fol a+liftProp = Plain++forall_ :: OperationClass a => (Effect -> Fol a) -> Fol a+forall_ f = Forall [] f++forall2_ :: OperationClass a => (Effect -> Effect -> Fol a) -> Fol a+forall2_ f = forall_ $ \a -> forall_ $ \b -> f a b++forall3_ :: OperationClass a => (Effect -> Effect -> Effect -> Fol a) -> Fol a+forall3_ f = forall_ $ \a -> forall_ $ \b -> forall_ $ \c -> f a b c++forall4_ :: OperationClass a => (Effect -> Effect -> Effect -> Effect -> Fol a) -> Fol a+forall4_ f = forall_ $ \a -> forall_ $ \b -> forall_ $ \c -> forall_ $ \d -> f a b c d++forallQ_ :: OperationClass a => [a] -> (Effect -> Fol a) -> Fol a+forallQ_ q f = Forall q f++forallQ2_ :: OperationClass a => [a] -> [a] -> (Effect -> Effect -> Fol a) -> Fol a+forallQ2_ l1 l2 f = forallQ_ l1 $ \a -> forallQ_ l2 $ \b -> f a b++forallQ3_ :: OperationClass a => [a] -> [a] -> [a] -> (Effect -> Effect -> Effect -> Fol a) -> Fol a+forallQ3_ l1 l2 l3 f = forallQ_ l1 $ \a -> forallQ_ l2 $ \b -> forallQ_ l3 $ \c -> f a b c++forallQ4_ :: OperationClass a => [a] -> [a] -> [a] -> [a] -> (Effect -> Effect -> Effect -> Effect -> Fol a) -> Fol a+forallQ4_ l1 l2 l3 l4 f = forallQ_ l1 $ \a -> forallQ_ l2 $ \b -> forallQ_ l3 $ \c -> forallQ_ l4 $ \d -> f a b c d+++sameEff :: Effect -> Effect -> Prop a+sameEff a b = AppRel SameEff a b++data EffCol = DirDep Effect Effect | SameTxn Effect Effect | Single Effect++trans :: OperationClass a => EffCol -> EffCol -> Prop a+trans col1 col2 =+  mkCurTxn col1 ∧ mkDepRel col1 ∧ mkDepRel col2 ∧ mkNotDepRel col1 col2+  where+    mkCurTxn (DirDep e1 e2) = CurTxn e1 ∧ CurTxn e2+    mkCurTxn (SameTxn e1 e2) = CurTxn e1 ∧ CurTxn e2+    mkCurTxn (Single e) = CurTxn e++    mkDepRel (DirDep e1 e2) = atomDep e1 e2+    mkDepRel (SameTxn e1 e2) = atomDep e1 e2 ∧ atomDep e2 e1+    mkDepRel (Single _) = true++    getLastEff (DirDep e1 e2) = e2+    getLastEff (SameTxn e1 e2) = e2+    getLastEff (Single e) = e++    mkNotDepRel col1 col2 =+      let e1 = getLastEff col1+          e2 = getLastEff col2+      in (Not $ atomDep e1 e2 ∨ atomDep e2 e1)

@@ -0,0 +1,556 @@+{-# LANGUAGE TemplateHaskell, ScopedTypeVariables, DoAndIfThenElse, FlexibleContexts #-}++module Quelea.Contract.TypeCheck (+  classifyOperContract,+  classifyTxnContract,+  isValid,+  isSat,++  -- Debug Only+  fol2Z3Ctrt,+  dummyZ3Sort,+  underMAV+) where+++import Quelea.Types+import Quelea.Contract.Language+import Z3.Monad hiding (mkFreshFuncDecl, mkFreshConst, solverAssertCnstr, push, pop,+                        check, getModel)+import qualified Z3.Monad as Z3M (mkFreshFuncDecl, mkFreshConst, solverAssertCnstr,+                                  push, pop, check, getModel)+import qualified Data.Map as M+import qualified Data.Set as S+import Language.Haskell.TH+import Control.Lens+import Control.Monad.State+import Data.Maybe (fromJust)+import Data.List (find)+import Control.Applicative ((<$>))+import System.IO+import Data.Time++-------------------------------------------------------------------------------+-- Types++makeLenses ''Z3CtrtState++-------------------------------------------------------------------------------+-- Helper++-- #define DEBUG_SHOW+-- #define DEBUG_CHECK+-- #define DEBUG_SANITY++check :: Z3 Result+#ifdef DEBUG_SHOW+check = do+  liftIO $ do+    putStrLn "(check-sat)"+    hFlush stdout+    hFlush stderr+  Z3M.check+#else+check = Z3M.check+#endif++getModel :: Z3 (Result, Maybe Model)+#ifdef DEBUG_SHOW+getModel = do+  liftIO $ do+    putStrLn "(check-sat) ;; get-model"+    hFlush stdout+    hFlush stderr+  Z3M.getModel+#else+getModel = Z3M.getModel+#endif+++push :: Z3 ()+#ifdef DEBUG_SHOW+push = do+  liftIO $ do+    putStrLn "(push)"+    hFlush stdout+    hFlush stderr+  Z3M.push+#else+push = Z3M.push+#endif++pop :: Int -> Z3 ()+#ifdef DEBUG_SHOW+pop n | n /= 1 = error "pop"+pop 1 = do+  liftIO $ do+    putStrLn "(pop)"+    hFlush stdout+    hFlush stderr+  Z3M.pop 1+#else+pop = Z3M.pop+#endif++assertCnstr :: String -> AST -> Z3 ()+#ifdef DEBUG_SHOW+assertCnstr name ast = do+  setASTPrintMode Z3_PRINT_SMTLIB2_COMPLIANT+  astStr <- astToString ast+  liftIO $ do+    putStrLn $ ";; --------------------------------"+    putStrLn $ ";; Assert: " ++ name+    putStrLn $ "(assert " ++ astStr ++ ")"+    hFlush stdout+    hFlush stderr+  Z3M.solverAssertCnstr ast+  #ifdef DEBUG_CHECK+  push+  r <- check+  liftIO $ putStrLn $ ";; Assert Result: " ++ (show r)+  pop 1+  #endif+#else+assertCnstr s a = Z3M.solverAssertCnstr a+#endif++mkFreshFuncDecl :: String -> [Sort] -> Sort -> Z3 FuncDecl+#ifdef DEBUG_SHOW+mkFreshFuncDecl s args res = do+  setASTPrintMode Z3_PRINT_SMTLIB2_COMPLIANT+  fd <- Z3M.mkFreshFuncDecl s args res+  fdStr <- funcDeclToString fd+  liftIO $ putStrLn $ ";; --------------------------------\n" ++ fdStr ++ "\n"+  liftIO $ hFlush stdout+  liftIO $ hFlush stderr+  return fd+#else+mkFreshFuncDecl = Z3M.mkFreshFuncDecl+#endif++mkFreshConst :: String -> Sort -> Z3 AST+#ifdef DEBUG_SHOW+mkFreshConst str srt = do+  c <- Z3M.mkFreshConst str srt+  cstr <- astToString c+  srtstr <- sortToString srt+  liftIO $ putStrLn $ ";; --------------------------------"+  liftIO $ putStrLn $ "(declare-const " ++ cstr ++ " " ++ srtstr ++ ")\n"+  liftIO $ hFlush stdout+  liftIO $ hFlush stderr+  return c+#else+mkFreshConst = Z3M.mkFreshConst+#endif++#ifdef DEBUG_SANITY+debugCheck str =+  lift $ push >> check >>= (\r -> when (r == Unsat) $ error str) >> pop 1+#else+debugCheck str = return ()+#endif++lookupEff :: Effect -> StateT Z3CtrtState Z3 AST+lookupEff e = do+  em <- use effMap+  return $ fromJust $ em ^.at e++newEff :: StateT Z3CtrtState Z3 (Effect, App)+newEff = do+  em <- use effMap+  let newEff = Effect $ M.size em+  es <- use effSort+  qvConst <- lift $ mkFreshConst "E" es+  qv <- lift $ toApp qvConst+  let newEm = at newEff .~ Just qvConst $ em+  effMap .= newEm+  return $ (newEff, qv)++assertProp :: String -> Z3Ctrt -> StateT Z3CtrtState Z3 AST+assertProp str (Z3Ctrt m) = do+  ast <- m+  lift $ assertCnstr str ast+  asl <- use assertions+  assertions .= ast:asl+  return ast++-- Create a Z3 Event Sort that mirrors the Haskell Oper Type.+mkMkZ3OperSort :: Q (Z3 Sort)+mkMkZ3OperSort = do+  TyConI (DataD _ (typeName::Name) _ constructors _) <- reify $ mkName operationsTyConStr+  let typeNameStr = nameBase typeName+  let consNameStrList = map (\ (NormalC name _) -> nameBase name) constructors+  let makeCons consStr = do+      consSym <- mkStringSymbol consStr+      isConsSym <- mkStringSymbol $ "is_" ++ consStr+      mkConstructor consSym isConsSym []+  let makeDatatype = do+      consList <- sequence $ map makeCons consNameStrList+      dtSym <- mkStringSymbol typeNameStr+      mkDatatype dtSym consList+  return makeDatatype++dummyZ3Sort :: Z3 Sort+dummyZ3Sort = do+  let makeCons consStr = do+      consSym <- mkStringSymbol consStr+      isConsSym <- mkStringSymbol $ "is_" ++ consStr+      mkConstructor consSym isConsSym []+  let makeDatatype = do+      consList <- sequence $ map makeCons ["DummyCon"]+      dtSym <- mkStringSymbol "DummyTyCon"+      mkDatatype dtSym consList+  makeDatatype+++-------------------------------------------------------------------------------+-- Contract to Z3 translation++-- #define EXISTS++rel2Z3Ctrt :: Rel -> Effect -> Effect -> Z3Ctrt+rel2Z3Ctrt r e1 e2 = Z3Ctrt $ do+  case r of+    Vis -> mkApp1 visRel e1 e2+    So -> mkApp1 soRel e1 e2+    SameObj -> mkApp1 sameobjRel e1 e2+    AtomDep -> mkApp1 atomDepRel e1 e2+    SameEff -> do+      a1 <- lookupEff e1+      a2 <- lookupEff e2+      lift $ mkEq a1 a2+    Union r1 r2 -> do+      a1 <- unZ3Ctrt $ rel2Z3Ctrt r1 e1 e2+      a2 <- unZ3Ctrt $ rel2Z3Ctrt r2 e1 e2+      lift $ mkOr [a1, a2]+    Intersect r1 r2 -> do+      a1 <- unZ3Ctrt $ rel2Z3Ctrt r1 e1 e2+      a2 <- unZ3Ctrt $ rel2Z3Ctrt r2 e1 e2+      lift $ mkAnd [a1, a2]+    TC r -> do+      rm <- use tcRelMap+      case rm ^.at r of+        Nothing -> do+          es <- use effSort+          bs <- lift $ mkBoolSort+          newR <- lift $ mkFreshFuncDecl "TC" [es,es] bs+          -- Insert into rm+          let newRm = at r .~ Just newR $ rm+          tcRelMap .= newRm+          -- Prop 1+          let f1 :: Fol () = forall2_ $ \a b -> liftProp $+                  (Raw $ rel2Z3Ctrt r a b) ⇒ (Raw . Z3Ctrt $ mkApp2 newR a b)+          p1 <- unZ3Ctrt $ fol2Z3Ctrt f1+          -- Prop 2+          let f2 :: Fol () = forall3_ $ \a b c -> liftProp $+                  ((Raw . Z3Ctrt $ mkApp2 newR a b) ∧ (Raw .Z3Ctrt $ mkApp2 newR b c)) ⇒+                  (Raw . Z3Ctrt $ mkApp2 newR a c)+          p2 <- unZ3Ctrt $ fol2Z3Ctrt f2+#ifdef EXISTS+          -- Prop 3+          let f3 :: Fol () = forall2_ $ \a b -> liftProp $+                (Raw . Z3Ctrt $ mkApp2 newR a b) ⇒ ((Raw $ rel2Z3Ctrt r a b) ∨+                (exists $ \c -> (Raw $ rel2Z3Ctrt r a b) ∧ (Raw . Z3Ctrt $ mkApp2 newR b c)))+          p3 <- unZ3Ctrt $ fol2Z3Ctrt f3+          assertProp "TC" $ Z3Ctrt . lift $ mkAnd [p1,p2,p3]+#else+          assertProp "TC" $ Z3Ctrt . lift $ mkAnd [p1,p2]+#endif+          mkApp2 newR e1 e2+        Just savedR -> mkApp2 savedR e1 e2+  where+    mkApp1 idx e1 e2 = do+      r <- use idx+      mkApp2 r e1 e2+    mkApp2 r e1 e2 = do+      a1 <- lookupEff e1+      a2 <- lookupEff e2+      lift $ mkApp r [a1,a2]++prop2Z3Ctrt :: OperationClass a => Prop a -> Z3Ctrt+prop2Z3Ctrt PTrue = Z3Ctrt $ lift mkTrue+prop2Z3Ctrt (AppRel r e1 e2) = rel2Z3Ctrt r e1 e2+prop2Z3Ctrt (CurTxn e) = Z3Ctrt $ do+  ctRel <- use curTxnRel+  effAST <- lookupEff e+  lift $ mkApp ctRel [effAST]+prop2Z3Ctrt (Conj p1 p2) = Z3Ctrt $ do+  a1 <- unZ3Ctrt $ prop2Z3Ctrt p1+  a2 <- unZ3Ctrt $ prop2Z3Ctrt p2+  lift $ mkAnd [a1,a2]+prop2Z3Ctrt (Disj p1 p2) = Z3Ctrt $ do+  a1 <- unZ3Ctrt $ prop2Z3Ctrt p1+  a2 <- unZ3Ctrt $ prop2Z3Ctrt p2+  lift $ mkOr [a1,a2]+prop2Z3Ctrt (Impl p1 p2) = Z3Ctrt $ do+  a1 <- unZ3Ctrt $ prop2Z3Ctrt p1+  a2 <- unZ3Ctrt $ prop2Z3Ctrt p2+  lift $ mkImplies a1 a2+prop2Z3Ctrt (Oper eff operName) = Z3Ctrt $ do+  effAST <- lookupEff eff+  operRel <- use operRel+  operNameAST <- getZ3Operation operName+  a1 <- lift $ mkApp operRel [effAST]+  lift $ mkEq a1 operNameAST+  where+    getZ3Operation operName = do+      operSort <- use operSort+      constructors <- lift $ getDatatypeSortConstructors operSort+      nameList <- lift $ mapM getDeclName constructors+      strList <- lift $ mapM getSymbolString nameList+      let pList = zip strList constructors+      let Just (_,constructor) = find (\ (s,_) -> (show operName) == s) pList+      lift $ mkApp constructor []+prop2Z3Ctrt (Raw c) = c+prop2Z3Ctrt (Not p) = Z3Ctrt $ do+  a <- unZ3Ctrt $ prop2Z3Ctrt p+  lift $ mkNot a++fol2Z3Ctrt :: OperationClass a => Fol a -> Z3Ctrt+fol2Z3Ctrt (Plain p) = prop2Z3Ctrt p+fol2Z3Ctrt (Forall operNameList f) = Z3Ctrt $ do+  (effInt, effApp) <- newEff+  body <- unZ3Ctrt . fol2Z3Ctrt $ f effInt+  if length operNameList == 0 then+    lift $ mkForallConst [] [effApp] body+  else do+    l <- mapM (\on -> unZ3Ctrt . prop2Z3Ctrt $ Oper effInt on) operNameList+    ante <- lift $ mkOr l+    body2 <- lift $ mkImplies ante body+    lift $ mkForallConst [] [effApp] body2++exists :: OperationClass a => (Effect -> Prop a) -> Prop a+exists f = Raw $ Z3Ctrt $ do+  (effInt, effApp) <- newEff+  body <- unZ3Ctrt . prop2Z3Ctrt $ f effInt+  lift $ mkExistsConst [] [effApp] body++-------------------------------------------------------------------------------+-- Type checking helper++assertBasicAxioms :: StateT Z3CtrtState Z3 ()+assertBasicAxioms = do+  assertProp "ThinAir" $ fol2Z3Ctrt thinAir+  assertProp "doVis" $ fol2Z3Ctrt doVis+  assertProp "soTrans" $ fol2Z3Ctrt $ transRel so++  assertProp "reflRelAtomDep" $ fol2Z3Ctrt $ reflRel atomDep+  assertProp "transRelAtomDep" $ fol2Z3Ctrt $ transRel atomDep+  assertProp "txnFollowsSess" $ fol2Z3Ctrt $ txnFollowsSess++  assertProp "reflRelSameObj" $ fol2Z3Ctrt $ reflRel sameObj+  assertProp "symRelSameObj" $ fol2Z3Ctrt $ symRel sameObj+  assertProp "transRelSameObj" $ fol2Z3Ctrt $ transRel sameObj++  assertProp "curTxnAtomDep" $ fol2Z3Ctrt $ curTxnAtomDep+  assertProp "atomicVisibility" $ fol2Z3Ctrt $ readCommitted++  return ()+  where+    curTxnAtomDep :: Fol () = forall2_ $ \x y -> liftProp $ CurTxn x ∧ CurTxn y ⇒ atomDep x y+    thinAir :: Fol () = forall_ $ \x -> liftProp . Not $ hb x x+    doVis :: Fol () = forall2_ $ \a b -> liftProp $ vis a b ⇒ appRel SameObj a b+    txnFollowsSess :: Fol () = forall2_ $ \a b -> liftProp $+                                 (atomDep a b ∧ atomDep b a) ⇒  (so a b ∨ so b a ∨ sameEff a b)++    reflRel :: (Effect -> Effect -> Prop ()) -> Fol ()+    reflRel rel = forall_ $ \x -> liftProp $ rel x x++    symRel :: (Effect -> Effect -> Prop ()) -> Fol ()+    symRel rel = forall2_ $ \a b -> liftProp $ rel a b ⇒  rel b a++    transRel :: (Effect -> Effect -> Prop ()) -> Fol ()+    transRel rel = forall3_ $ \a b c -> liftProp $ rel a b ∧ rel b c ⇒  rel a c++    readCommitted :: Fol () = forall3_ $ \a b c -> liftProp $+                                trans (DirDep a b) (Single c) ∧ sameObjList [a,b,c] ∧ vis b c ⇒ vis a c+++++mkZ3CtrtState :: Sort -> Z3 Z3CtrtState+mkZ3CtrtState operSort = do+  effSort <- mkUninterpretedSort =<< mkStringSymbol "Effect"+  boolSort <- mkBoolSort++  visRel <- mkFreshFuncDecl "vis" [effSort, effSort] boolSort+  soRel <- mkFreshFuncDecl "so" [effSort, effSort] boolSort+  sameobjRel <- mkFreshFuncDecl "sameobj" [effSort, effSort] boolSort+  atomDepRel <- mkFreshFuncDecl "atomDep" [effSort, effSort] boolSort+  curTxnRel <- mkFreshFuncDecl "curTxn" [effSort] boolSort+  operRel <- mkFreshFuncDecl "oper" [effSort] operSort++  return $ Z3CtrtState effSort operSort visRel soRel sameobjRel atomDepRel curTxnRel operRel M.empty [] M.empty++res2Bool :: Result -> Bool+res2Bool Unsat = True+res2Bool Sat = False++not_ :: Z3Ctrt -> Z3Ctrt+not_ (Z3Ctrt m) = Z3Ctrt $ do+  ast <- m+  lift $ mkNot ast++typecheck :: Z3 Sort -> StateT Z3CtrtState Z3 Bool -> IO Bool+typecheck mkOperSort core = evalZ3 $ do+  operSort <- mkOperSort+  st <- mkZ3CtrtState operSort+  (res, _) <- runStateT core st+  return res++isWellTyped :: OperationClass a => Contract a -> Z3 Sort -> IO Bool+isWellTyped c mkOperSort = typecheck mkOperSort $ do+  (curEff, _) <- newEff++  let an1 = fol2Z3Ctrt $ sc curEff+  let cn1 = fol2Z3Ctrt $ c curEff+  let test1 = prop2Z3Ctrt $ mkRawImpl an1 cn1+  assertProp "SC_IMPL_CTRT" $ not_ test1+  lift $ res2Bool <$> check++sc :: Contract ()+sc x = forall_ $ \a ->  liftProp $ (hbo a x ∨ hbo x a ∨ AppRel SameEff a x) ∧+                                   (hbo a x ⇒ vis a x) ∧+                                   (hbo x a ⇒ vis x a)++cc :: Contract ()+cc x = forall_ $ \a -> liftProp $ hbo a x ⇒ vis a x++cv :: Contract ()+cv x = forall2_ $ \a b -> liftProp $ (hbo a b ∧ vis b x) ⇒ vis a x++mav :: Fol ()+mav = forall4_ $ \a b c d -> liftProp $ (trans (SameTxn a b) (DirDep c d) ∧ sameObj b c ∧ vis d a ∧ AppRel (So ∪ SameEff) a b ⇒ vis c b)++rr :: Fol ()+rr = forall4_ $ \a b c d -> liftProp $ (trans (SameTxn a b) (DirDep c d) ∧ sameObj b c ∧ vis d a ⇒ vis c b)++mkRawImpl :: Z3Ctrt -> Z3Ctrt -> Prop ()+mkRawImpl a b = (Raw a) ⇒ (Raw b)+++mkRawImpl2 :: (OperationClass a, OperationClass b) => Fol a -> Fol b -> Z3Ctrt+mkRawImpl2 a b =+  let p::Prop () = (Raw $ fol2Z3Ctrt a) ⇒ (Raw $ fol2Z3Ctrt b)+  in prop2Z3Ctrt p++isStronglyConsistent :: OperationClass a  => Contract a -> Z3 Sort -> IO Bool+isStronglyConsistent c mkOperSort =+  typecheck mkOperSort $ do+    assertBasicAxioms+    (curEff, _) <- newEff++    let an1 = fol2Z3Ctrt $ sc curEff+    let cn1 = fol2Z3Ctrt $ c curEff+    let test1 = prop2Z3Ctrt $ mkRawImpl an1 cn1+    assertProp "SC_IMPL_CTRT" $ not_ test1++    let an2 = fol2Z3Ctrt $ c curEff+    let cn2 = fol2Z3Ctrt $ cc curEff+    let test2 = prop2Z3Ctrt $ Not $ mkRawImpl an2 cn2+    assertProp "CTRT_NOT_IMPL_CC" $ not_ test2+    lift $ res2Bool <$> check++isCausallyConsistent :: OperationClass a  => Contract a -> Z3 Sort -> IO Bool+isCausallyConsistent c mkOperSort =+  typecheck mkOperSort $ do+    assertBasicAxioms+    (curEff, _) <- newEff++    let an1 = fol2Z3Ctrt $ cc curEff+    let cn1 = fol2Z3Ctrt $ c curEff+    let test1 = prop2Z3Ctrt $ mkRawImpl an1 cn1+    assertProp "CC_IMPL_CTRT" $ not_ test1++    let an2 = fol2Z3Ctrt $ c curEff+    let cn2 = fol2Z3Ctrt $ cv curEff+    let test2 = prop2Z3Ctrt $ Not $ mkRawImpl an2 cn2+    assertProp "CTRT_NOT_IMPL_CV" $ not_ test2+    lift $ res2Bool <$> check++isEventuallyConsistent :: OperationClass a  => Contract a -> Z3 Sort -> IO Bool+isEventuallyConsistent c mkOperSort =+  typecheck mkOperSort $ do+    assertBasicAxioms+    (curEff, _) <- newEff++    let an1 = fol2Z3Ctrt $ cv curEff+    let cn1 = fol2Z3Ctrt $ c curEff+    let test1 = prop2Z3Ctrt $ mkRawImpl an1 cn1+    assertProp "CV_IMPL_CTRT" $ not_ test1+    lift $ res2Bool <$> check++underRC :: OperationClass a => Fol a -> Z3 Sort -> IO Bool+underRC c mkOperSort =+  isValidProto mkOperSort "CTRT" c++underMAV :: OperationClass a => Fol a -> Z3 Sort -> IO Bool+underMAV c mkOperSort = do+    let test1 :: Fol () = liftProp . Raw $ mkRawImpl2 mav c+    isValidProto mkOperSort "MAV_IMPL_CTRT" test1++underRR :: OperationClass a => Fol a -> Z3 Sort -> IO Bool+underRR c mkOperSort = do+    let test1 :: Fol () = liftProp . Raw $ mkRawImpl2 rr c+    isValidProto mkOperSort "RR_IMPL_CTRT" test1++classifyTxnContract :: OperationClass a => Fol a -> String -> Q TxnKind+classifyTxnContract c info = do+  mkOperSort <- mkMkZ3OperSort+  t1 <- runIO getCurrentTime+  a <- runIO $ do+    res <- do+      res <- underRC c mkOperSort+      if res then return RC+      else do+        res <- underMAV c mkOperSort+        if res then return MAV+        else do+          res <- underRR c mkOperSort+          if res then return RR+          else fail $ info ++ " contract is not well-typed"+    return res+  t2 <- runIO getCurrentTime+  _ <- runIO $ putStrLn $ info ++ " classified as " ++ (show a) +      ++" in "++(show $ diffUTCTime t2 t1)++"."+  _ <- runIO $ hFlush stdout+  return a++classifyOperContract :: OperationClass a => Contract a -> String -> Q Availability+classifyOperContract c info = do+  mkOperSort <- mkMkZ3OperSort+  t1 <- runIO (getCurrentTime)+  a <- runIO $ do+    isWt <- isWellTyped c mkOperSort+    if not isWt then fail $ info ++ " contract is not well-typed"+    else do+      res <- isEventuallyConsistent c mkOperSort+      if res then return Eventual+      else do+        res <- isCausallyConsistent c mkOperSort+        if res then return Causal+        else do+          res <- isStronglyConsistent c mkOperSort+          if res then return Strong+          else fail $ info ++ " -- strange contract"+  t2 <- runIO (getCurrentTime)+  _ <- runIO $ putStrLn (info ++ " classified as "++(show a)++" in "+++            (show $ diffUTCTime t2 t1)++".")+  return a++isValidProto :: OperationClass a => Z3 Sort -> String -> Fol a -> IO Bool+isValidProto mkOperSort str c = typecheck mkOperSort $ do+  assertBasicAxioms+  assertProp str $ not_ $ fol2Z3Ctrt c+  lift $ res2Bool <$> check+++isValid :: OperationClass a => String -> Fol a -> IO Bool+isValid = isValidProto dummyZ3Sort++isSat :: OperationClass a => String -> Fol a -> IO Bool+isSat str c = typecheck dummyZ3Sort $ do+  assertBasicAxioms+  assertProp str $ fol2Z3Ctrt c+  r <- lift $ check+  return $ case r of {Unsat -> False; Sat -> True}

@@ -0,0 +1,296 @@+{-# LANGUAGE ScopedTypeVariables, EmptyDataDecls, TemplateHaskell,+    DataKinds, OverloadedStrings, DoAndIfThenElse  #-}++module Quelea.DBDriver (+  TableName(..),+  ReadRow,++  createTable,+  dropTable,++  cqlRead,+  cqlReadAfterTime,+  cqlReadWithTime,+  cqlReadAfterTimeWithTime,++  cqlInsert,+  -- cqlInsertWithTime,+  cqlDelete,++  getLock,+  releaseLock,++  getGCLock,+  releaseGCLock,++  createTxnTable,+  dropTxnTable,+  readTxn,+  insertTxn+) where+++import Quelea.Consts+import Control.Concurrent (threadDelay)+import Quelea.Types+import Quelea.NameService.SimpleBroker+import Quelea.Marshall+import Data.Serialize+import Control.Applicative ((<$>))+import Control.Monad (forever)+import Data.ByteString hiding (map, pack)+import Data.Either (rights)+import Data.Map (Map)+import Data.Time+import qualified Data.Map as Map+import System.ZMQ4+import Control.Lens+import Database.Cassandra.CQL+import Data.UUID+import Data.Int (Int64)+import qualified Data.Set as S+import Data.Text hiding (map)+import Control.Monad.Trans (liftIO)+import Data.Maybe (fromJust)+import Control.Monad (when)++-- Simply an alias for Types.ObjType+type TableName = String++type ReadRow = (SessID, SeqNo, S.Set Addr, Cell, Maybe TxnID)+type ReadRowInternal = (UUID, SeqNo, Deps, Cell, Maybe UUID)++type ReadRowWithTime = (SessID, SeqNo, UTCTime, S.Set Addr, Cell, Maybe TxnID)+type ReadRowWithTimeInternal = (UUID, SeqNo, UTCTime, Deps, Cell, Maybe UUID)++type WriteRowInternal = (UTCTime, Key, UUID, SeqNo, Deps, Cell, Maybe UUID)++--------------------------------------------------------------------------------+-- Cassandra Link Layer+--------------------------------------------------------------------------------+++-- A Row either corresponds to an effect (Cell is EffectVal bs) or a gc marker+-- (Cell is GCMarker). In case of GCMarker, the dependence set (deps value in+-- the row) is interpreted as a "Cursor". All effects that are encapsulated by+-- this cursor are considered to have been GC'ed.+mkCreateTable :: TableName -> Query Schema () ()+mkCreateTable tname = query $ pack $ "create table " ++ tname +++                      " ( objid blob, sessid uuid, seqno bigint, addedat timestamp, deps blob, value blob, txnid uuid, primary key (objid, addedat, sessid, seqno))"++mkDropTable :: TableName -> Query Schema () ()+mkDropTable tname = query $ pack $ "drop table " ++ tname++mkInsert :: TableName -> Query Write WriteRowInternal ()+mkInsert tname = query $ pack $ "insert into " ++ tname ++ " (addedat, objid, sessid, seqno, deps, value, txnid) values (?, ?, ?, ?, ?, ?, ?)"++mkDelete :: TableName -> Query Write (Key, UTCTime, UUID, SeqNo) ()+mkDelete tname = query $ pack $ "delete from " ++ tname ++ " where objid = ? and addedat = ? and sessid = ? and seqno = ?"++mkRead :: TableName -> Query Rows (Key) ReadRowInternal+mkRead tname = query $ pack $ "select sessid, seqno, deps, value, txnid from " ++ tname ++ " where objid = ?"++mkReadAfterTime :: TableName -> Query Rows (Key,UTCTime) ReadRowInternal+mkReadAfterTime tname = query $ pack $ "select sessid, seqno, deps, value, txnid from " ++ tname ++ " where objid = ? and addedat > ?"++mkReadWithTime :: TableName -> Query Rows (Key) ReadRowWithTimeInternal+mkReadWithTime tname = query $ pack $ "select sessid, seqno, addedat, deps, value, txnid from " ++ tname ++ " where objid = ?"++mkReadAfterTimeWithTime :: TableName -> Query Rows (Key, UTCTime) ReadRowWithTimeInternal+mkReadAfterTimeWithTime tname = query $ pack $ "select sessid, seqno, addedat, deps, value, txnid from " ++ tname ++ " where objid = ? and addedat > ?"++-------------------------------------------------------------------------------++mkCreateLockTable :: TableName -> Query Schema () ()+mkCreateLockTable tname = query $ pack $ "create table " ++ tname ++ "_LOCK (objid blob, sessid uuid, primary key (objid))"++mkDropLockTable :: TableName -> Query Schema () ()+mkDropLockTable tname = query $ pack $ "drop table " ++ tname ++ "_LOCK"++mkLockInsert :: TableName -> Query Write (Key, UUID) ()+mkLockInsert tname = query $ pack $ "insert into " ++ tname ++ "_LOCK (objid, sessid) values (?, ?) if not exists"++mkLockUpdate :: TableName -> Query Write (UUID {- New -}, Key, UUID {- Old -}) ()+mkLockUpdate tname = query $ pack $ "update " ++ tname ++ "_LOCK set sessid = ? where objid = ? if sessid = ?"++-------------------------------------------------------------------------------++mkCreateGCLockTable :: TableName -> Query Schema () ()+mkCreateGCLockTable tname = query $ pack $ "create table " ++ tname ++ "_GC_LOCK (objid blob, sessid uuid, primary key (objid))"++mkDropGCLockTable :: TableName -> Query Schema () ()+mkDropGCLockTable tname = query $ pack $ "drop table " ++ tname ++ "_GC_LOCK"++mkGCLockInsert :: TableName -> Query Write (Key, UUID) ()+mkGCLockInsert tname = query $ pack $ "insert into " ++ tname ++ "_GC_LOCK (objid, sessid) values (?, ?) if not exists"++mkGCLockUpdate :: TableName -> Query Write (UUID {- New -}, Key, UUID {- Old -}) ()+mkGCLockUpdate tname = query $ pack $ "update " ++ tname ++ "_GC_LOCK set sessid = ? where objid = ? if sessid = ?"+++-------------------------------------------------------------------------------++mkCreateTxnTable :: Query Schema () ()+mkCreateTxnTable = "create table Txns (txnid uuid, deps blob, primary key (txnid))"++mkDropTxnTable :: Query Schema () ()+mkDropTxnTable = "drop table Txns"++mkInsertTxnTable :: Query Write (UUID, TxnDepSet) ()+mkInsertTxnTable = "insert into Txns (txnid, deps) values (?, ?)"++mkReadTxnTable :: Query Rows (UUID) TxnDepSet+mkReadTxnTable = "select deps from Txns where txnid = ?"++-------------------------------------------------------------------------------++mkCreateGlobalLockTable :: Query Schema () ()+mkCreateGlobalLockTable = "create table GlobalLock (id uuid, txnid uuid, primary key id)"++mkDropGlobalLockTable :: Query Schema () ()+mkDropGlobalLockTable = "drop table GlobalLock"++mkGlobalLockInsert :: Query Write (UUID, UUID) ()+mkGlobalLockInsert = "insert into GlobalLock (id, txnid) values (?,?)"++mkGlobalLockUpdate :: Query Write (UUID {- New TxnID -}, UUID {- ID -}, UUID {- Old TxnID -}) ()+mkGlobalLockUpdate = "update GlobalLock set txnid = ? where id = ? if txnid = ?"++-------------------------------------------------------------------------------++cqlReadAfterTime :: TableName -> Consistency -> Key -> UTCTime -> Cas [ReadRow]+cqlReadAfterTime tname c k gcTime = do+  rows <- executeRows c (mkReadAfterTime tname) (k, gcTime)+  return $ map (\(sid, sqn, Deps deps, val, txid) -> (SessID sid, sqn, deps, val, TxnID <$> txid)) rows++cqlRead :: TableName -> Consistency -> Key -> Cas [ReadRow]+cqlRead tname c k = do+  rows <- executeRows c (mkRead tname) k+  return $ map (\(sid, sqn, Deps deps, val, txid) -> (SessID sid, sqn, deps, val, TxnID <$> txid)) rows++cqlReadAfterTimeWithTime :: TableName -> Consistency -> Key -> UTCTime -> Cas [ReadRowWithTime]+cqlReadAfterTimeWithTime tname c k gcTime = do+  rows <- executeRows c (mkReadAfterTimeWithTime tname) (k, gcTime)+  return $ map (\(sid, sqn, addedat, Deps deps, val, txid) -> (SessID sid, sqn, addedat, deps, val, TxnID <$> txid)) rows++cqlReadWithTime :: TableName -> Consistency -> Key -> Cas [ReadRowWithTime]+cqlReadWithTime tname c k = do+  rows <- executeRows c (mkReadWithTime tname) k+  return $ map (\(sid, sqn, addedat, Deps deps, val, txid) -> (SessID sid, sqn, addedat, deps, val, TxnID <$> txid)) rows++cqlInsertWithTime :: TableName -> Consistency -> Key -> ReadRow -> UTCTime -> Cas ()+cqlInsertWithTime tname c k (SessID sid, sqn, dep,val,txid) ct = do+  if sqn == 0+  then error "cqlInsert : sqn is 0"+  else do+    if S.size dep > 0+    then executeWrite c (mkInsert tname) (ct,k,sid,sqn,Deps dep,val,unTxnID <$> txid)+    else executeWrite c (mkInsert tname) (ct,k,sid,sqn,Deps $ S.singleton $ Addr (SessID sid) 0, val, unTxnID <$> txid)++cqlInsert :: TableName -> Consistency -> Key -> ReadRow -> Cas ()+cqlInsert tname c k row = do+  ct <- liftIO $ getCurrentTime+  cqlInsertWithTime tname c k row ct++cqlDelete :: TableName -> Key -> UTCTime -> SessID -> SeqNo -> Cas ()+cqlDelete tname k time (SessID sid) sqn =+  executeWrite ONE (mkDelete tname) (k,time,sid,sqn)++createTxnTable :: Cas ()+createTxnTable = liftIO . print =<< executeSchema ALL mkCreateTxnTable ()++dropTxnTable :: Cas ()+dropTxnTable = liftIO . print =<< executeSchema ALL mkDropTxnTable ()++insertTxn :: TxnID -> S.Set TxnDep -> Cas ()+insertTxn (TxnID txnid) deps = do+  when (S.size deps == 0) $ error "insertTxn: Txn has no actions"+  executeWrite ONE mkInsertTxnTable (txnid, TxnDepSet deps)++readTxn :: TxnID -> Cas (Maybe (S.Set TxnDep))+readTxn (TxnID txnid) = do+  result <- executeRow ONE mkReadTxnTable txnid+  case result of+    Nothing -> return Nothing+    Just (TxnDepSet s) -> return $ Just s++createTable :: TableName -> Cas ()+createTable tname = do+  liftIO . print =<< executeSchema ALL (mkCreateTable tname) ()+  liftIO . print =<< executeSchema ALL (mkCreateLockTable tname) ()+  liftIO . print =<< executeSchema ALL (mkCreateGCLockTable tname) ()++dropTable :: TableName -> Cas ()+dropTable tname = do+  liftIO . print =<< executeSchema ALL (mkDropTable tname) ()+  liftIO . print =<< executeSchema ALL (mkDropLockTable tname) ()+  liftIO . print =<< executeSchema ALL (mkDropGCLockTable tname) ()++----------------------------------------------------------------------------------++tryGetLock :: TableName -> Key -> SessID -> Bool {- tryInsert -} -> Cas Bool+tryGetLock tname k (SessID sid) True = do+  res <- executeTrans (mkLockInsert tname) (k, sid)+  if res then return True+  else tryGetLock tname k (SessID sid) False+tryGetLock tname k (SessID sid) False = do+  res <- executeTrans (mkLockUpdate tname) (sid, k, knownUUID)+  if res then return True+  else do+    liftIO $ threadDelay cLOCK_DELAY+    tryGetLock tname k (SessID sid) False++getLock :: TableName -> Key -> SessID -> Pool -> IO ()+getLock tname k sid pool = runCas pool $ do+  tryGetLock tname k sid True+  return ()++releaseLock :: TableName -> Key -> SessID -> Pool -> IO ()+releaseLock tname k (SessID sid) pool = runCas pool $ do+  res <- executeTrans (mkLockUpdate tname) (knownUUID, k, sid)+  if res then return ()+  else error $ "releaseLock : key=" ++ show k ++ " sid=" ++ show sid++--------------------------------------------------------------------------------++tryGetGCLock :: TableName -> Key -> SessID -> Bool {- tryInsert -} -> Cas Bool+tryGetGCLock tname k (SessID sid) True = do+  res <- executeTrans (mkGCLockInsert tname) (k, sid)+  if res then return True+  else tryGetGCLock tname k (SessID sid) False+tryGetGCLock tname k (SessID sid) False = do+  res <- executeTrans (mkGCLockUpdate tname) (sid, k, knownUUID)+  if res then return True+  else do+    liftIO $ threadDelay cLOCK_DELAY+    tryGetGCLock tname k (SessID sid) False++getGCLock :: TableName -> Key -> SessID -> Pool -> IO ()+getGCLock tname k sid pool = runCas pool $ do+  tryGetGCLock tname k sid True+  return ()++releaseGCLock :: TableName -> Key -> SessID -> Pool -> IO ()+releaseGCLock tname k (SessID sid) pool = runCas pool $ do+  res <- executeTrans (mkGCLockUpdate tname) (knownUUID, k, sid)+  if res then return ()+  else error $ "releaseGCLock : key=" ++ show k ++ " sid=" ++ show sid++--------------------------------------------------------------------------------++createGlobalLockTable :: Cas ()+createGlobalLockTable = do+  liftIO . print =<< executeSchema ALL mkCreateGlobalLockTable ()+  executeWrite ALL mkGlobalLockInsert (knownUUID, knownUUID)++getGlobalLock :: TxnID -> Pool -> IO ()+getGlobalLock (TxnID txnid) pool = runCas pool loop+  where+    loop = do+      success <- executeTrans mkGlobalLockUpdate (txnid, knownUUID, knownUUID)+      when (not success) loop++releaseGlobalLock :: TxnID -> Pool -> IO ()+releaseGlobalLock (TxnID txnid) pool = runCas pool $ do+  success <- executeTrans mkGlobalLockUpdate (knownUUID, knownUUID, txnid)+  when (not success) (error $ "releaseGlobalLock: key=" ++ show (TxnID txnid))

@@ -0,0 +1,144 @@+{-# LANGUAGE ScopedTypeVariables, TypeSynonymInstances, TemplateHaskell #-}++module Quelea.Marshall (+  mkGenOp,+  mkGenSum,+  decodeOperationPayload,+  decodeResponse+) where++import Quelea.Types+import Database.Cassandra.CQL+import Data.Serialize as S+import Control.Applicative ((<$>))+import Data.ByteString (ByteString, length, head, tail)+import Data.Either (rights)+import Data.Map (Map)+import qualified Data.Map as Map+import Data.ByteString.Char8 (pack, unpack)+import Data.UUID+import Data.Maybe (fromJust)+import Data.Word+import Data.DeriveTH+import Data.Time++$(derive makeSerialize ''OperationPayload)++$(derive makeSerialize ''Key)++$(derive makeSerialize ''SessID)++$(derive makeSerialize ''TxnID)++instance CasType TxnID where+  putCas = put+  getCas = get+  casType _ = CBlob++mkGenOp :: (Effectish eff, Serialize arg, Serialize res)+        => OpFun eff arg res+        -> ([eff] -> [eff])+        -> (GenOpFun, GenSumFun)+mkGenOp foo bar = (fun1 foo, mkGenSum bar)+  where+    fun1 foo ctxt arg =+      let ctxt2 = rights $ map decode ctxt+          arg2 = case decode arg of+                  Right v -> v+                  Left s -> error ("mkGenOp : " ++ s)+          (res, eff) = foo ctxt2 arg2+      in (encode res, encode <$> eff)++mkGenSum :: Effectish eff => ([eff] -> [eff]) -> GenSumFun+mkGenSum foo ctxt =+  let ctxt2 = rights $ map decode ctxt+      ctxt3 = foo ctxt2+  in encode <$> ctxt3++decodeOperationPayload :: OperationClass a => ByteString -> Request a+decodeOperationPayload b =+  case decode b of+    Left s -> error $ "decodeOperationPayload : " ++ s+    Right v -> v++instance Serialize UUID where+  put = putLazyByteString . toByteString+  get = do+    r <- fromByteString <$> getLazyByteString 16+    case r of+      Nothing -> error "serialize UUID"+      Just x -> return x++$(derive makeSerialize ''Response)++decodeResponse :: ByteString -> Response+decodeResponse b = case decode b of+                     Left s -> error $ "decodeResponse : " ++ s+                     Right v -> v++instance Serialize UTCTime where+  put = putCas+  get = getCas++$(derive makeSerialize ''Cell)++instance CasType Cell where+  putCas = put+  getCas = get+  casType _ = CBlob++{-+instance CasType Cell where+  putCas (EffectVal b) = do+    putWord8 0+    putWord32be $ fromIntegral $ Data.ByteString.length b+    putByteString b+  putCas GCMarker = putWord8 1+  getCas = do+    i <- getWord8+    case i of+      0 -> do+        l <- fromIntegral <$> getWord32be+        bs <- getByteString l+        return $ EffectVal bs+      1 -> return $ GCMarker+  casType _ = CBlob+-}++$(derive makeSerialize ''Addr)++instance CasType Addr where+  putCas = put+  getCas = get+  casType _ = CBlob+++$(derive makeSerialize ''Deps)++instance CasType Deps where+  putCas = put+  getCas = get+  casType _ = CBlob++$(derive makeSerialize ''TxnPayload)++$(derive makeSerialize ''TxnDep)++instance CasType TxnDep where+  putCas = put+  getCas = get+  casType _ = CBlob++$(derive makeSerialize ''TxnDepSet)++instance CasType TxnDepSet where+  putCas = put+  getCas = get+  casType _ = CBlob++instance CasType Key where+  putCas = put+  getCas = get+  casType _ = CBlob++$(derive makeSerialize ''Request)

@@ -0,0 +1,66 @@+{-# LANGUAGE OverloadedStrings #-}++module Quelea.NameService.LoadBalancingBroker (+  mkNameService,+  startBroker+) where++import qualified System.ZMQ4 as ZMQ4+import System.ZMQ4.Monadic+import Control.Concurrent+import Control.Monad+import Data.ByteString.Char8 (unpack, pack)+import System.Directory+import System.Posix.Process+import Control.Monad.Trans (liftIO)+import Quelea.NameService.Types++-- #define DEBUG++debugPrint :: String -> IO ()+#ifdef DEBUG+debugPrint s = do+  tid <- myThreadId+  putStrLn $ "[" ++ (show tid) ++ "] " ++ s+#else+debugPrint _ = return ()+#endif++startBroker :: Frontend -> Backend -> IO ()+startBroker f b  = runZMQ $ do+  fes <- socket Router+  bind fes $ unFE f+  bes <- socket Dealer+  bind bes $ unBE b+  proxy fes bes Nothing++clientJoin :: Frontend -> IO (String, ZMQ4.Socket ZMQ4.Req)+clientJoin f = do+  ctxt <- ZMQ4.context+  sock <- ZMQ4.socket ctxt ZMQ4.Req+  ZMQ4.connect sock serverAddr+  return (serverAddr, sock)+  where+    serverAddr = unFE f++serverJoin :: Backend -> String {- ip -} -> Int {- Port# -} -> IO ()+serverJoin b _ _ = runZMQ $ do+    {- Create a router socket and connect with the backend -}+    routerSock <- socket Router+    connect routerSock $ unBE b++    {- Create a dealer socket, bind it to a local ipc file -}+    dealerSock <- socket Dealer+    liftIO $ createDirectoryIfMissing False "/tmp/quelea"+    pid <- liftIO $ getProcessID+    bind dealerSock $ "ipc:///tmp/quelea/" ++ show pid++    {- Start proxy to distribute requests to workers -}+    proxy routerSock dealerSock Nothing++mkNameService :: Frontend -> Backend+              -> String {- Backend ip (only for sticky) -}+              -> Int {- Backend port (only for sticky) -}+              -> NameService+mkNameService fe be ip port =+  NameService fe (clientJoin fe) (serverJoin be ip port)

@@ -0,0 +1,95 @@+{-# LANGUAGE OverloadedStrings #-}++module Quelea.NameService.SimpleBroker (+  startBroker,+  mkNameService+) where++import qualified System.ZMQ4 as ZMQ4+import System.ZMQ4.Monadic+import Control.Concurrent+import Control.Monad+import Data.ByteString.Char8 (unpack, pack)+import System.Directory+import System.Posix.Process+import Control.Monad.Trans (liftIO)+import Quelea.NameService.Types++-- #define DEBUG++debugPrint :: String -> IO ()+#ifdef DEBUG+debugPrint s = do+  tid <- myThreadId+  putStrLn $ "[" ++ (show tid) ++ "] " ++ s+#else+debugPrint _ = return ()+#endif++startBroker :: Frontend -> Backend -> IO ()+startBroker f b  = runZMQ $ do+  fes <- socket Router+  bind fes $ unFE f+  bes <- socket Dealer+  bind bes $ unBE b+  proxy fes bes Nothing++clientJoin :: Frontend -> IO (String, ZMQ4.Socket ZMQ4.Req)+clientJoin f = do+  serverAddr <- runZMQ $ do+    requester <- socket Req+    liftIO $ debugPrint "clientJoin(1)"+    connect requester $ unFE f+    liftIO $ debugPrint "clientJoin(2)"+    send requester [] "Howdy Server! send your socket info"+    liftIO $ debugPrint "clientJoin(3)"+    msg <- receive requester+    liftIO $ debugPrint "clientJoin(4)"+    return $ unpack msg+  -- Connect to the shim layer node.+  ctxt <- ZMQ4.context+  sock <- ZMQ4.socket ctxt ZMQ4.Req+  ZMQ4.connect sock serverAddr+  return (serverAddr, sock)+++serverJoin :: Backend -> String {- ip -} -> Int {- Port# -} -> IO ()+serverJoin b ip port = do++  void $ forkIO $ runZMQ $ do+    liftIO $ debugPrint "serverJoin(5)"+    {- Create a router and a dealer -}+    routerSock <- socket Router+    let myaddr = "tcp://*:" ++ show port+    bind routerSock myaddr++    dealerSock <- socket Dealer+    liftIO $ createDirectoryIfMissing False "/tmp/quelea"+    pid <- liftIO $ getProcessID+    bind dealerSock $ "ipc:///tmp/quelea/" ++ show pid++    liftIO $ debugPrint "serverJoin(6): starting proxy"+    {- Start proxy to distribute requests to workers -}+    proxy routerSock dealerSock Nothing++  {- Fork a daemon thread that joins with the backend. The daemon shares the+   - servers address for every client request. The client then joins with the+   - server.+   -}+  runZMQ $ do+    responder <- socket Rep+    liftIO $ debugPrint "serverJoin(1)"+    connect responder $ unBE b+    liftIO $ debugPrint "serverJoin(2)"+    forever $ do+      message <- receive responder+      liftIO $ debugPrint $ "serverJoin(3) " ++ ip+      send responder [] $ pack $ "tcp://" ++ ip ++ ":" ++ show port+      liftIO $ debugPrint "serverJoin(4)"++mkNameService :: Frontend -> Backend+              -> String {- Backend ip (only for sticky) -}+              -> Int {- Backend port (only for sticky) -}+              -> NameService+mkNameService fe be ip port =+  NameService fe (clientJoin fe) (serverJoin be ip port)

@@ -0,0 +1,16 @@+module Quelea.NameService.Types (+  Frontend(..),+  Backend(..),+  NameService(..)+) where++import System.ZMQ4++newtype Frontend = Frontend { unFE :: String}+newtype Backend  = Backend  { unBE :: String}++data NameService = NameService {+  getFrontend   :: Frontend,+  getClientJoin :: IO (String, Socket Req),+  getServerJoin :: IO ()+}

@@ -0,0 +1,222 @@+{-# LANGUAGE ScopedTypeVariables, EmptyDataDecls,+    TemplateHaskell, DataKinds, OverloadedStrings,+    DoAndIfThenElse#-}++module Quelea.Shim (+ runShimNode,+ runShimNodeWithOpts,+ mkDtLib+) where++import Quelea.Consts+import Quelea.Types+import Quelea.Consts+import Quelea.NameService.Types+import Quelea.Marshall+import Quelea.DBDriver+import Quelea.ShimLayer.Cache+import Quelea.ShimLayer.GC+import Quelea.Contract.Language++import Control.Concurrent (threadDelay)+import Data.Serialize+import Control.Applicative ((<$>))+import Control.Monad (forever, replicateM, when)+import Data.ByteString hiding (map, pack, putStrLn)+import Data.Either (rights)+import Data.Map (Map)+import qualified Data.Map as M+import System.ZMQ4.Monadic+import qualified System.ZMQ4 as ZMQ+import Data.Maybe (fromJust)+import Control.Lens+import System.Posix.Process+import Database.Cassandra.CQL+import Data.UUID+import Data.Int (Int64)+import qualified Data.Set as S+import Data.Text hiding (map)+import Debug.Trace+import Control.Concurrent (forkIO, myThreadId, threadDelay)+import Data.Tuple.Select+++makeLenses ''Addr+makeLenses ''DatatypeLibrary+makeLenses ''OperationPayload++#define DEBUG++debugPrint :: String -> IO ()+#ifdef DEBUG+debugPrint s = do+  tid <- myThreadId+  putStrLn $ "[" ++ (show tid) ++ "] " ++ s+#else+debugPrint _ = return ()+#endif++runShimNodeWithOpts :: OperationClass a+                    => GCSetting+                    -> Int -- fetch update interval+                    -> DatatypeLibrary a+                    -> [Server] -> Keyspace -- Cassandra connection info+                    -> NameService+                    -> IO ()+runShimNodeWithOpts gcSetting fetchUpdateInterval dtLib serverList keyspace ns = do+  {- Connection to the Cassandra deployment -}+  pool <- newPool serverList keyspace Nothing+  {- Spawn cache manager -}+  cache <- initCacheManager pool fetchUpdateInterval+  {- Spawn a pool of workers -}+  replicateM cNUM_WORKERS (forkIO $ worker dtLib pool cache gcSetting)+  case gcSetting of+    No_GC -> getServerJoin ns+    GC_Mem_Only -> getServerJoin ns+    otherwise -> do+      {- Join the broker to serve clients -}+      forkIO $ getServerJoin ns+      {- Start gcWorker -}+      gcWorker dtLib cache++runShimNode :: OperationClass a+            => DatatypeLibrary a+            -> [Server] -> Keyspace -- Cassandra connection info+            -> NameService+            -> IO ()+runShimNode = runShimNodeWithOpts GC_Full cCACHE_THREAD_DELAY++worker :: OperationClass a => DatatypeLibrary a -> Pool -> CacheManager -> GCSetting -> IO ()+worker dtLib pool cache gcSetting = do+  ctxt <- ZMQ.context+  sock <- ZMQ.socket ctxt ZMQ.Rep+  pid <- getProcessID+  -- debugPrint "worker: connecting..."+  ZMQ.connect sock $ "ipc:///tmp/quelea/" ++ show pid+  -- debugPrint "worker: connected"+  {- loop forver servicing clients -}+  forever $ do+    binReq <- ZMQ.receive sock+    txns <- includedTxns cache+    case decodeOperationPayload binReq of+      ReqOper req -> do+        {- Fetch the operation from the datatype library using the object type and+        - operation name. -}+        let (op,av) =+              case dtLib ^. avMap ^.at (req^.objTypeReq, req^.opReq) of+                Nothing -> error $ "Not found in DatatypeLibrary:" ++ (show (req^.objTypeReq, req^.opReq))+                Just x -> x+        -- debugPrint $ "worker: before " ++ show (req^.objTypeReq, req^.opReq, av)+        (result, ctxtSize) <- case av of+          Eventual -> doOp op cache req ONE+          Causal -> processCausalOp req op cache+          Strong -> processStrongOp req op cache pool+        ZMQ.send sock [] $ encode result+        -- debugPrint $ "worker: after " ++ show (req^.objTypeReq, req^.opReq)+        -- Maybe perform summarization+        let gcFun =+              case dtLib ^. sumMap ^.at (req^.objTypeReq) of+                Nothing -> error "Worker(2)"+                Just x -> x+        case gcSetting of+          No_GC -> return ()+          otherwise -> maybeGCCache cache (req^.objTypeReq) (req^.keyReq) ctxtSize gcFun+        return ()+      ReqTxnCommit txid deps -> do+        -- debugPrint $ "Committing transaction " ++ show txid+        when (S.size deps > 0) $ runCas pool $ insertTxn txid deps+        ZMQ.send sock [] $ encode ResCommit+      ReqSnapshot objs -> do+        fetchUpdates cache ALL $ S.toList objs+        snapshot <- snapshotCache cache+        let filteredSnapshot = M.foldlWithKey (\m k v ->+              if S.member k objs then M.insert k v m else m) M.empty snapshot+        ZMQ.send sock [] $ encode $ ResSnapshot filteredSnapshot+  where+    processCausalOp req op cache =+      -- Check whether this is the first effect in the session <= previous+      -- sequence number is 0.+      if req^.sqnReq == 0+      then doOp op cache req ONE+      else do+        let ot = req^.objTypeReq+        let k = req^.keyReq+        -- Check whether the current cache includes the previous effect+        res <- doesCacheInclude cache ot k (req^.sidReq) (req^.sqnReq)+        if res+        then doOp op cache req ONE+        else do+          -- Read DB, and check cache for previous effect+          fetchUpdates cache ONE [(ot,k)]+          res <- doesCacheInclude cache ot k (req^.sidReq) (req^.sqnReq)+          if res+          then doOp op cache req ONE+          else do+            -- Wait till next cache refresh and repeat the process again+            waitForCacheRefresh cache ot k+            processCausalOp req op cache+    processStrongOp req op cache pool = do+      let (ot, k, sid) = (req^.objTypeReq, req^.keyReq, req^.sidReq)+      -- Get Lock+      getLock ot k sid pool+      -- debugPrint $ "processStrongOp: obtained lock"+      -- Read latest values at the key - under ALL+      fetchUpdates cache ALL [(ot,k)]+      -- Perform the op+      res <- doOp op cache req ALL+      -- Release Lock+      releaseLock ot k sid pool+      return res++doOp :: OperationClass a => GenOpFun -> CacheManager -> OperationPayload a -> Consistency -> IO (Response, Int)+doOp op cache request const = do+  let (OperationPayload objType key operName arg sessid seqno mbtxid getDeps) = request+  -- Build the context+  (ctxt, deps) <- buildContext objType key mbtxid+  -- Perform the operation on this context+  -- debugPrint $ "doOp: length of context = " ++ show (Prelude.length ctxt)+  let (res, effM) = op ctxt arg+  -- Add current location to the ones for which updates will be fetched+  addHotLocation cache objType key+  let resDeps = if getDeps then deps else S.empty+  result <- case effM of+    Nothing -> return $ ResOper seqno res Nothing Nothing resDeps+    Just eff -> do+      -- Write effect writes to DB, and potentially to cache+      writeEffect cache objType key (Addr sessid (seqno+1)) eff deps const $ sel1 <$> mbtxid+      case mbtxid of+        Nothing -> return $ ResOper (seqno + 1) res Nothing Nothing resDeps+        Just (_,MAV_TxnPl _ _) -> do+          txns <- getInclTxnsAt cache objType key+          return $ ResOper (seqno + 1) res (Just eff) (Just txns) resDeps+        otherwise -> return $ ResOper (seqno + 1) res (Just eff) Nothing resDeps+  -- return response+  return (result, Prelude.length ctxt)+  where+    buildContext ot k Nothing = getContext cache ot k+    buildContext ot k (Just (_, RC_TxnPl l)) = do+      (ctxtVanilla, depsVanilla) <- buildContext ot k Nothing+      let (el, as) = S.foldl (\(el,as) (addr, eff) ->+                      (eff:el, S.insert addr as)) ([],S.empty) l+      return (el ++ ctxtVanilla, S.union as depsVanilla)+    buildContext ot k (Just (txid, MAV_TxnPl l txndeps)) = do+      res <- doesCacheIncludeTxns cache txndeps+      if res then buildContext ot k (Just (txid,RC_TxnPl l))+      else do+        fetchTxns cache txndeps+        buildContext ot k (Just (txid, MAV_TxnPl l txndeps))+    buildContext ot k (Just (_,RR_TxnPl effSet)) = return $+      S.foldl (\(el,as) (addr, eff) -> (eff:el, S.insert addr as))+              ([], S.empty) effSet+++mkDtLib :: OperationClass a => [(a, (GenOpFun, GenSumFun), Availability)] -> DatatypeLibrary a+mkDtLib l =+  let (m1, m2) = Prelude.foldl core (M.empty, M.empty) l+  in DatatypeLibrary m1 m2+  where+    core (m1, m2) (op, (fun1, fun2), av) =+      (M.insert (getObjType op, op) (fun1, av) m1,+       M.insert (getObjType op) fun2 m2)++

@@ -0,0 +1,227 @@+{-# LANGUAGE ScopedTypeVariables, EmptyDataDecls, TemplateHaskell, DataKinds, OverloadedStrings, DoAndIfThenElse, BangPatterns  #-}++module Quelea.ShimLayer.Cache (+  CacheManager,++  initCacheManager,+  getContext,+  addHotLocation,+  writeEffect,+  doesCacheInclude,+  waitForCacheRefresh,+  fetchUpdates,+  includedTxns,+  doesCacheIncludeTxns,+  fetchTxns,+  snapshotCache,+  getInclTxnsAt,+) where++import Quelea.Consts+import Control.Concurrent+import Control.Concurrent.MVar+import Data.ByteString hiding (map, pack, putStrLn, foldl, length, filter)+import Control.Lens+import qualified Data.Map as M+import qualified Data.Set as S+import System.Posix.Process (getProcessID)+import Data.Map.Lens+import Control.Monad (forever, when, replicateM, foldM)+import Data.Maybe (fromJust)+import Database.Cassandra.CQL+import Control.Monad.State+import System.IO+import Control.Applicative ((<$>))+import Data.Tuple.Select++import Quelea.Types+import Quelea.ShimLayer.Types+import Quelea.DBDriver+import Quelea.ShimLayer.UpdateFetcher++makeLenses ''CacheManager++#define DEBUG++debugPrint :: String -> IO ()+#ifdef DEBUG+debugPrint s = do+  tid <- myThreadId+  pid <- getProcessID+  putStrLn $ "[" ++ (show pid) ++ "," ++ (show tid) ++ "] " ++ s+  hFlush stdout+#else+debugPrint _ = return ()+#endif+++initCacheManager :: Pool -> Int -> IO CacheManager+initCacheManager pool fetchUpdateInterval = do+  cache <- newMVar M.empty+  cursor <- newMVar M.empty+  nearestDeps <- newMVar M.empty+  lastGCAddr <- newMVar M.empty+  lastGCTime <- newMVar M.empty+  seenTxns <- newMVar (S.empty, M.empty)+  hwm <- newMVar M.empty+  drc <- newMVar M.empty+  hotLocs <- newMVar S.empty+  sem <- newEmptyMVar+  blockedList <- newMVar []+  let cm = CacheManager cache cursor nearestDeps lastGCAddr lastGCTime+                        seenTxns hwm drc hotLocs sem blockedList pool+  forkIO $ cacheMgrCore cm+  forkIO $ signalGenerator sem fetchUpdateInterval+  return $ cm+  where+    signalGenerator semMVar fetchUpdateInterval = forever $ do+      isEmpty <- isEmptyMVar semMVar+      if isEmpty+      then tryPutMVar semMVar ()+      else return True+      threadDelay fetchUpdateInterval++getInclTxnsAt :: CacheManager -> ObjType -> Key -> IO (S.Set TxnID)+getInclTxnsAt cm ot k = do+  inclTxns <- readMVar $ cm^.includedTxnsMVar+  case M.lookup (ot,k) $ sel2 inclTxns of+    Nothing -> return S.empty+    Just s -> return s++addHotLocation :: CacheManager -> ObjType -> Key -> IO ()+addHotLocation cm ot k = do+  hotLocs <- takeMVar $ cm^.hotLocsMVar+  putMVar (cm^.hotLocsMVar) $ S.insert (ot,k) hotLocs++cacheMgrCore :: CacheManager -> IO ()+cacheMgrCore cm = forever $ do+  takeMVar $ cm^.semMVar+  -- Woken up. Read the current list of hot locations, and empty the MVar.+  locs <- takeMVar $ cm^.hotLocsMVar+  putMVar (cm^.hotLocsMVar) S.empty+  -- Fetch updates+  fetchUpdates cm ONE $ S.toList locs+  -- Wakeup threads that are waiting for the cache to be refreshed+  blockedList <- takeMVar $ cm^.blockedMVar+  putMVar (cm^.blockedMVar) []+  mapM_ (\mv -> putMVar mv ()) blockedList++-- Print stats+printStats :: CacheManager -> ObjType -> Key -> IO ()+printStats cm ot k = do+  cacheMap <- readMVar $ cm^.cacheMVar+  cursorMap <- readMVar $ cm^.cursorMVar+  depsMap <- readMVar $ cm^.depsMVar+  lgca <- readMVar $ cm^.lastGCAddrMVar+  (inclTxns,_) <- readMVar $ cm^.includedTxnsMVar+  hwm <- readMVar $ cm^.hwmMVar+  hotLocs <- readMVar $ cm^.hotLocsMVar+  tq <- readMVar $ cm^.blockedMVar++  let cache = case M.lookup (ot,k) cacheMap of {Nothing -> S.empty; Just s -> s}+  let cursor = case M.lookup (ot,k) cursorMap of {Nothing -> M.empty; Just s -> s}+  let deps = case M.lookup (ot,k) depsMap of {Nothing -> S.empty; Just s -> s}++  putStrLn $ "Stats : cache=" ++ show (S.size cache) ++ " cursor=" ++ show (M.size cursor) +++             " deps=" ++ show (S.size deps) ++ " lgca=" ++ show (M.size lgca) +++             " incTxns=" ++ show (S.size inclTxns) ++ " hwm=" ++ show (M.size hwm) +++             " hotLocs=" ++ show (S.size hotLocs) ++ " tq=" ++ show (length tq)++-- Returns the set of effects at the location and a set of nearest dependencies+-- for this location.+getContext :: CacheManager -> ObjType -> Key -> IO ([Effect], S.Set Addr)+getContext cm ot k = do+  !cache <- takeMVar $ cm^.cacheMVar+  !deps <- takeMVar $ cm^.depsMVar+  putMVar (cm^.cacheMVar) cache+  putMVar (cm^.depsMVar) deps+  let !v1 = case M.lookup (ot,k) cache of+             Nothing -> []+             Just s -> Prelude.map (\(a,e) -> e) (S.toList s)+  let !v2 = case M.lookup (ot,k) deps of {Nothing -> S.empty; Just s -> s}+  -- printStats cm ot k+  return (v1, v2)++writeEffect :: CacheManager -> ObjType -> Key -> Addr -> Effect -> S.Set Addr+            -> Consistency -> Maybe TxnID -> IO ()+writeEffect cm ot k addr eff deps const mbtxnid = do+  let Addr sid sqn = addr+  -- Does cache include the previous effect?+  isPrevEffectAvailable <- doesCacheInclude cm ot k sid (sqn - 1)+  let isTxn = case mbtxnid of {Nothing -> False; otherwise -> True}+  -- Only write to cache if the previous effect is available in the cache. This+  -- maintains the cache to be a causally consistent cut of the updates. But do+  -- not update cache if the effect is in a transaction. This prevents+  -- uncommitted effects from being made visible.+  if ((not isTxn) && (sqn == 1 || isPrevEffectAvailable))+  then do+    !cache <- takeMVar $ cm^.cacheMVar+    !cursor <- takeMVar $ cm^.cursorMVar+    -- curDeps may be different from the deps seen before the operation was performed.+    !curDeps <- takeMVar $ cm^.depsMVar+    -- Update cache+    let !newCache = M.insertWith S.union (ot,k) (S.singleton (addr, eff)) cache+    putMVar (cm^.cacheMVar) newCache+    -- Update cursor+    let !cursorAtKey = case M.lookup (ot,k) cursor of {Nothing -> M.empty; Just m -> m}+    let !newCursorAtKey = M.insert sid sqn cursorAtKey+    putMVar (cm^.cursorMVar) $ M.insert (ot,k) newCursorAtKey cursor+    -- Update dependence+    -- the deps seen by the effect is a subset of the curDeps. We over+    -- approximate the dependence set; only means that effects might take longer+    -- to converge, but importantly does not affect correctness.+    let newDeps = case M.lookup (ot,k) curDeps of {Nothing -> S.empty; Just s -> s}+    putMVar (cm^.depsMVar) $ M.insert (ot,k) (S.singleton addr) curDeps+    -- Write to database+    runCas (cm^.pool) $ cqlInsert ot const k (sid, sqn, newDeps, EffectVal eff, mbtxnid)+  else do+    -- Write to database+    runCas (cm^.pool) $ cqlInsert ot const k (sid, sqn, deps, EffectVal eff, mbtxnid)++doesCacheInclude :: CacheManager -> ObjType -> Key -> SessID -> SeqNo -> IO Bool+doesCacheInclude cm ot k sid sqn = do+  cursor <- readMVar $ cm^.cursorMVar+  case M.lookup (ot,k) cursor of+    Nothing -> return False+    Just cursorAtKey ->+      case M.lookup sid cursorAtKey of+        Nothing -> return False+        Just curSqn -> return $ (==) sqn curSqn++waitForCacheRefresh :: CacheManager -> ObjType -> Key -> IO ()+waitForCacheRefresh cm ot k = do+  hotLocs <- takeMVar $ cm^.hotLocsMVar+  blockedList <- takeMVar $ cm^.blockedMVar+  mv <- newEmptyMVar+  putMVar (cm^.hotLocsMVar) $ S.insert (ot,k) hotLocs+  putMVar (cm^.blockedMVar) $ mv:blockedList+  takeMVar mv++includedTxns :: CacheManager -> IO (S.Set TxnID)+includedTxns cm = do+  txns <- readMVar (cm^.includedTxnsMVar)+  return $ sel1 txns++doesCacheIncludeTxns :: CacheManager -> S.Set TxnID -> IO Bool+doesCacheIncludeTxns cm deps = do+  incl <- includedTxns cm+  return $ deps `S.isSubsetOf` incl++fetchTxns :: CacheManager -> S.Set TxnID -> IO ()+fetchTxns cm deps = do+  incl <- includedTxns cm+  let diffSet = S.difference deps incl+  objs <- foldM (\acc txid -> do+            objs <- getObjs txid+            return $ S.union acc objs) S.empty $ S.toList diffSet+  fetchUpdates cm ONE (S.toList objs)+  where+    getObjs txid = do+      res <- runCas (cm^.pool) $ readTxn txid+      case res of+        Nothing -> return $ S.empty+        Just s -> return $ S.map (\(TxnDep ot k _ _) -> (ot,k)) s++snapshotCache :: CacheManager -> IO CacheMap+snapshotCache cm = do+  readMVar $ cm^.cacheMVar

@@ -0,0 +1,220 @@+{-# LANGUAGE ScopedTypeVariables, TemplateHaskell, DoAndIfThenElse, BangPatterns  #-}++module Quelea.ShimLayer.GC (+  maybeGCCache,+  gcWorker+) where++import Quelea.Consts+import Quelea.Types+import Quelea.ShimLayer.Types+import Quelea.DBDriver+import Quelea.ShimLayer.UpdateFetcher++import Control.Monad.State+import Control.Lens+import qualified Data.Set as S+import qualified Data.Map as M+import Control.Concurrent (threadDelay, myThreadId)+import Control.Concurrent.MVar+import System.Random (randomIO)+import Database.Cassandra.CQL+import Control.Applicative ((<$>))+import Data.Maybe (fromJust)+import Data.Time++makeLenses ''CacheManager+makeLenses ''DatatypeLibrary++-- #define DEBUG++debugPrint :: String -> IO ()+#ifdef DEBUG+debugPrint s = do+  tid <- myThreadId+  putStrLn $ "[" ++ (show tid) ++ "] " ++ s+#else+debugPrint _ = return ()+#endif+++data VisitedState = Visited Bool  -- Boolean indicates whether the effect is resolved+                  | NotVisited (S.Set Addr)++data ResolutionState = ResolutionState {+  _keyCursor    :: M.Map SessID SeqNo,+  _visitedState :: M.Map Addr VisitedState+}++makeLenses ''ResolutionState++{- How to GC a transactional effect?+ -----------------------------------------+ - When GCing a set of effects {a@t1,b@t2} belonging to transactions t1 and t2,+ - one needs to ensure that the transaction markers are preserved after the GC.+ - Assume c is the new effect that summarizes a and b, and GCM is the gc+ - marker. Now the final state after GC will be:++    {c} --introduced by--> {GCM@{t1,t2}} --GCs--> {a,b}++ - where the set of transactions of the effects that were GCed are included in+ - the GC marker. Now, if c is included in any shim layer node, the+ - transactions t1 and t2 also need to be processed.+ -}+++gcDB :: CacheManager -> ObjType -> Key -> GenSumFun -> IO ()+gcDB cm ot k gc = gcDBCore cm ot k gc 1++gcDBCore :: CacheManager -> ObjType -> Key -> GenSumFun -> Int -> IO ()+gcDBCore cm ot k gc 0 = return ()+gcDBCore cm ot k gc repeat = do+  debugPrint $ "gcDB: start"+  -- Allocate new session id+  gcSid <- SessID <$> randomIO+  -- Get GC lock+  getGCLock ot k gcSid $ cm^.pool+  -- Get time at the start of GC+  currentTime <- getCurrentTime+  let gcTime = currentTime+  lgctMap <- readMVar $ cm^.lastGCTimeMVar+  rows <- case M.lookup (ot,k) lgctMap of+            Nothing -> runCas (cm^.pool) $ cqlReadWithTime ot ALL k+            Just lastGCTime -> runCas (cm^.pool) $ cqlReadAfterTimeWithTime ot ALL k lastGCTime+  -- Split the rows into effects and gc markers+  let (effRows, gcMarker) = foldl (\(effAcc,gcAcc) (sid,sqn,time,deps,val,txnid) ->+        case txnid of+          Just _ -> (effAcc, gcAcc) -- XXX KC; Handle transactions+          Nothing ->+            case val of+              EffectVal bs -> ((sid,sqn,time,deps,bs):effAcc, gcAcc)+              GCMarker _ -> case gcAcc of+                            Nothing -> (effAcc, Just (sid,sqn,time,deps))+                            Just _ -> error "Multiple GC Markers") ([], Nothing) rows+  -- Build the GC cursor+  let gcCursor = case gcMarker of+                   Nothing -> M.empty+                   Just (sid, sqn, time, deps) ->+                     S.foldl (\m (Addr sid sqn) -> M.insert sid sqn m)+                             (M.singleton sid sqn) deps+  -- Build datastructure for filtering out unresolved effects+  let (effSet, depsMap) =+        foldl (\(setAcc, mapAcc) (sid, sqn, time, deps, eff) ->+                  (S.insert (Addr sid sqn, eff, time) setAcc,+                   M.insert (Addr sid sqn) (NotVisited deps) mapAcc))+          (S.empty, M.empty) effRows+  -- filter unresolved effects i,e) effects whose causal cut is also not visible.+  let filteredSet = filterUnresolved gcCursor depsMap effSet+  let (addrList, effList, timeList) = unzip3 (S.toList filteredSet)+  let gcedEffList = gc effList+  -- Allocate new gc address+  let gcAddr = Addr gcSid 1+  let (outRows, count) =+        foldl (\(acc, idx) eff ->+                  ((gcSid, idx, S.singleton gcAddr, EffectVal eff, Nothing):acc, idx+1))+              ([],2) gcedEffList+  runCas (cm^.pool) $ do+    -- Insert new effects+    mapM_ (\r -> cqlInsert ot ALL k r) outRows+    -- Delete previous marker if it exists+    case gcMarker of+      Nothing -> return ()+      Just (sid, sqn, time, _) -> cqlDelete ot k time sid sqn+    -- Insert marker+    let am = foldl (\m (Addr sid sqn) ->+               case M.lookup sid m of+                 Nothing -> M.insert sid sqn m+                 Just oldSqn -> M.insert sid (max oldSqn sqn) m) M.empty addrList+    let newCursor = M.unionWith max am gcCursor+    let newDeps = S.fromList $ map (\(sid,sqn) -> Addr sid sqn) $ M.toList newCursor+    cqlInsert ot ALL k (gcSid, 1, newDeps, GCMarker gcTime, Nothing)+    -- Update lastGCTime before deleting+    liftIO $ do+      lgctMap <- takeMVar $ cm^.lastGCTimeMVar+      putMVar (cm^.lastGCTimeMVar) $ M.insert (ot,k) gcTime lgctMap+    -- Delete old rows+    mapM_ (\(Addr sid sqn, time) -> cqlDelete ot k time sid sqn) $ zip addrList timeList+  {- info -}+  debugPrint $ "gcDB: inserted Rows=" ++ show (length outRows)+  debugPrint $ "gcDB: deleted Rows=" ++ show (length addrList)+  releaseGCLock ot k gcSid $ cm^.pool+  {- Remove the current object from diskRowCount map -}+  drc <- takeMVar $ cm^.diskRowCntMVar+  putMVar (cm^.diskRowCntMVar) $ M.delete (ot,k) drc+  debugPrint $ "gcDB: end"+  gcDBCore cm ot k gc $ repeat - 1++maybeGCCache :: CacheManager -> ObjType -> Key -> Int -> GenSumFun -> IO ()+maybeGCCache cm ot k curSize gc | curSize < cCACHE_LWM = return ()+                                | otherwise = do+  hwmMap <- readMVar $ cm^.hwmMVar+  let hwm = case M.lookup (ot,k) hwmMap of+              Nothing -> cCACHE_LWM+              Just x -> x+  when (curSize > hwm) $ do+    cache <- takeMVar $ cm^.cacheMVar+    let ctxt = case M.lookup (ot, k) cache of+                Nothing -> []+                Just s -> map (\(a,e) -> e) $ S.toList s+    let !newCtxt = gc ctxt+    newUUID <- SessID <$> randomIO+    let (newCache,_) = foldl (\(s,i) e -> (S.insert (Addr newUUID i, e) s, i+1)) (S.empty, 1) newCtxt+    hwm <- takeMVar $ cm^.hwmMVar+    putMVar (cm^.hwmMVar) $ M.insert (ot, k) (length newCtxt * 2) hwm+    putMVar (cm^.cacheMVar) $ M.insert (ot, k) newCache cache++filterUnresolved :: CursorAtKey                   -- Key Cursor+                 -> M.Map Addr VisitedState       -- Input visited state+                 -> S.Set (Addr, Effect, UTCTime) -- Unfiltered input+                 -> S.Set (Addr, Effect, UTCTime) -- Filtered result+filterUnresolved kc m1 s2 =+  let addrList = M.keys m1+      ResolutionState _ vs =+        foldl (\vs addr -> execState (isResolved addr) vs) (ResolutionState kc m1) addrList+  in S.filter (\(addr,_,_) ->+        case M.lookup addr vs of+          Nothing -> error "filterUnresolved: unexpected state(1)"+          Just (Visited True) -> True+          Just (Visited False) -> False+          Just (NotVisited _) -> error "filterUnresolved: unexpected state(2)") s2++isResolved :: Addr -> State ResolutionState Bool+isResolved addr = do+  vs <- use visitedState+  case M.lookup addr vs of+    Nothing -> do -- Might be an effect already in the cache+      let Addr sid sqn = addr+      if sqn == 0 -- Special case to please Cassandra+      then do+        visitedState .= M.insert addr (Visited True) vs+        return True+      else do+        kc <- use keyCursor+        case M.lookup sid kc of+          -- Session of effect not found in cache => effect unseen+          Nothing -> do+            visitedState .= M.insert addr (Visited False) vs+            return False+          -- Session found in cache, cache is sufficiently recent?+          Just maxSqn -> do+            let res = sqn <= maxSqn+            visitedState .= M.insert addr (Visited res) vs+            return res+    Just (Visited res) -> return res+    Just (NotVisited deps) -> do+      res <- foldM (\acc addr -> isResolved addr >>= \r -> return (r && acc)) True $ S.toList deps+      newVs <- use visitedState+      visitedState .= M.insert addr (Visited res) newVs+      return res++gcWorker :: OperationClass a => DatatypeLibrary a -> CacheManager -> IO ()+gcWorker dtLib cm = forever $ do+  threadDelay cGC_WORKER_THREAD_DELAY+  drc <- readMVar $ cm^.diskRowCntMVar+  let todoObjs = M.foldlWithKey (\todoObjs (ot,k) rowCount ->+                                    if rowCount > cDISK_LWM+                                    then (ot,k):todoObjs+                                    else todoObjs) [] drc+  mapM_ (\(ot,k) ->+    let gcFun = case dtLib ^. sumMap ^.at ot of {Nothing -> error "gcWorker"; Just x -> x}+    in gcDB cm ot k gcFun) todoObjs

@@ -0,0 +1,53 @@+{-# Language TemplateHaskell, EmptyDataDecls, ScopedTypeVariables,+    TypeSynonymInstances, FlexibleInstances #-}++module Quelea.ShimLayer.Types (+  CacheManager(..),+  CacheMap,+  NearestDeps,+  NearestDepsMap,+  CursorMap,+  CursorAtKey,+  Effect+) where++import qualified Data.Map as M+import qualified Data.Set as S+import Data.ByteString+import Control.Concurrent.MVar+import Database.Cassandra.CQL+import Data.Time++import Quelea.Types++type Effect = ByteString++type CacheMap    = (M.Map (ObjType, Key) (S.Set (Addr, Effect)))+type HwmMap      = M.Map (ObjType, Key) Int+type DiskRowCount = M.Map (ObjType, Key) Int+type Cache       = MVar CacheMap+type CursorAtKey = M.Map SessID SeqNo+type CursorMap   = (M.Map (ObjType, Key) CursorAtKey)+type Cursor      = MVar CursorMap+type NearestDepsMap = (M.Map (ObjType, Key) (S.Set Addr))+type NearestDeps  = MVar NearestDepsMap+type HotLocs     = MVar (S.Set (ObjType, Key))+type Semaphore   = MVar ()+type ThreadQueue = MVar ([MVar ()])+++data CacheManager = CacheManager {+  _cacheMVar        :: Cache,+  _cursorMVar       :: Cursor,+  _depsMVar         :: NearestDeps,+  _lastGCAddrMVar   :: MVar (M.Map (ObjType, Key) SessID),+  _lastGCTimeMVar   :: MVar (M.Map (ObjType, Key) UTCTime),+  _includedTxnsMVar :: MVar (S.Set TxnID, M.Map (ObjType,Key) (S.Set TxnID)),++  _hwmMVar          :: MVar HwmMap,+  _diskRowCntMVar   :: MVar DiskRowCount,+  _hotLocsMVar      :: HotLocs,+  _semMVar          :: Semaphore,+  _blockedMVar      :: ThreadQueue,+  _pool             :: Pool+}

@@ -0,0 +1,410 @@+{-# LANGUAGE ScopedTypeVariables, TemplateHaskell, DoAndIfThenElse, BangPatterns, FlexibleContexts #-}++module Quelea.ShimLayer.UpdateFetcher (+  fetchUpdates+) where++import Control.Concurrent.MVar+import qualified Data.Map as M+import qualified Data.Set as S+import Control.Lens+import Database.Cassandra.CQL+import Control.Monad.Trans.State+import Control.Monad.Trans (liftIO)+import Control.Monad (mapM_, when, foldM)+import Data.ByteString (empty)+import Data.Maybe (fromJust)+import Control.Concurrent (myThreadId)+import Debug.Trace+import System.IO (hFlush, stdout)+import System.Posix.Process (getProcessID)+import Data.Tuple.Select+import Data.Time++import Quelea.Consts+import Quelea.Types+import Quelea.ShimLayer.Types+import Quelea.DBDriver++makeLenses ''CacheManager+makeLenses ''Addr++data CollectRowsState = CollectRowsState {+  _inclTxnsCRS :: S.Set TxnID,+  _todoObjsCRS :: S.Set (ObjType, Key),+  _rowsMapCRS  :: M.Map (ObjType, Key) [ReadRow],+  _newTxnsCRS  :: M.Map TxnID (S.Set TxnDep)+}++makeLenses ''CollectRowsState+makeLenses ''TxnDep++data VisitedState = Visited (Maybe (Effect, Maybe TxnID))+                  | NotVisited { effect  :: Effect,+                                 deps    :: S.Set Addr,+                                 txnid   :: Maybe TxnID,+                                 txndeps :: S.Set TxnDep }+                  | Visiting deriving Show++data ResolutionState = ResolutionState {+  _visitedState :: M.Map (ObjType, Key) (M.Map Addr VisitedState)+}++makeLenses ''ResolutionState++#define DEBUG++debugPrint :: String -> IO ()+#ifdef DEBUG+debugPrint s = do+  tid <- myThreadId+  pid <- getProcessID+  putStrLn $ "[" ++ (show pid) ++ "," ++ (show tid) ++ "] " ++ s+  hFlush stdout+#else+debugPrint _ = return ()+#endif++data CacheUpdateState = CacheUpdateState {+  _cacheCUS      :: CacheMap,+  _cursorCUS     :: CursorMap,+  _depsCUS       :: NearestDepsMap,+  _lastGCAddrCUS :: M.Map (ObjType, Key) SessID,+  _lastGCTimeCUS :: M.Map (ObjType, Key) UTCTime,+  _inclTxnsCUS   :: (S.Set TxnID, M.Map (ObjType, Key) (S.Set TxnID))+}++makeLenses ''CacheUpdateState++fetchUpdates :: CacheManager -> Consistency -> [(ObjType, Key)] -> IO ()+fetchUpdates cm const todoList = do+  -- Recursively read the DB and collect all the rows+  !cursor <- readMVar $ cm^.cursorMVar+  !inclTxns <- readMVar $ cm^.includedTxnsMVar+  !lgctMap <- readMVar $ cm^.lastGCTimeMVar++  let todoObjsCRS = S.fromList todoList+  let crs = CollectRowsState (sel1 inclTxns) todoObjsCRS M.empty M.empty+  CollectRowsState _ _ !rowsMapCRS !newTransMap <- execStateT (collectTransitiveRows (cm^.pool) const lgctMap) crs++  -- Update disk row count for later use by gcDB+  drc <- takeMVar $ cm^.diskRowCntMVar+  let newDrc = M.foldlWithKey (\iDrc (ot,k) rows -> M.insert (ot,k) (length rows) iDrc) drc rowsMapCRS+  putMVar (cm^.diskRowCntMVar) newDrc++  -- First collect the GC markers+  let gcMarkerMap = M.foldlWithKey (\gcmm (ot,k) rowList ->+         let gcm = foldl (\gcm (sid::SessID,sqn,deps,val,mbTxid) ->+                case val of+                  EffectVal bs -> gcm+                  GCMarker atTime ->+                    case gcm of+                      Nothing -> Just (sid, sqn, deps, atTime)+                      Just _ -> error "Multiple GC Markers") Nothing rowList+         in M.insert (ot,k) gcm gcmm) M.empty rowsMapCRS++  -- Build new cursor with GC markers. The idea here is that if we did find a+  -- GC marker & we have not seen this marker already, then keep hold of all+  -- the rows except the ones explicitly covered by the GC. The reason is that+  -- subsequently, we will clear our cache and rebuild it. Here, it is not+  -- correct to ignore those rows, which might have already been seen, but was+  -- not GCed.+  !lastGCAddrMap <- readMVar $ cm^.lastGCAddrMVar+  let !newCursor = M.foldlWithKey (\c (ot,k) gcMarker ->+                    let lastGCAddr = M.lookup (ot,k) lastGCAddrMap+                    in if newGCHasOccurred lastGCAddr gcMarker+                         then case gcMarker of+                                Nothing -> c+                                Just marker ->+                                  let gcCursor = buildCursorFromGCMarker marker+                                  in M.insert (ot,k) gcCursor c+                         else c)+                     cursor gcMarkerMap++  -- Once we have built the cursor, filter those rows which are covered.+  let effRowsMap = M.foldlWithKey (\erm (ot,k) rowList ->+         let er = foldl (\er (sid,sqn,deps,val,mbTxid) ->+                case val of+                  EffectVal bs ->+                    if isCovered newCursor ot k sid sqn+                    then er+                    else+                      let mkRetVal txid txdeps = M.insert (Addr sid sqn) (NotVisited bs deps txid txdeps) er+                      in case mbTxid of+                           Nothing -> mkRetVal Nothing S.empty+                           Just txid -> case M.lookup txid newTransMap of+                                          Nothing -> mkRetVal (Just txid) S.empty {- Eventual consistency (or) txn in progress! -}+                                          Just txnDeps -> mkRetVal (Just txid) txnDeps+                  GCMarker _ -> er) M.empty rowList+         in M.insert (ot,k) er erm) M.empty rowsMapCRS++  -- debugPrint $ "newCursor"+  -- mapM_ (\((ot,k), m) -> mapM_ (\(sid,sqn) -> debugPrint $ show $ Addr sid sqn) $ M.toList m) $ M.toList newCursor++  -- Now filter those rows which are unresolved i.e) those rows whose+  -- dependencies are not visible.+  let !filteredMap = filterUnresolved newCursor effRowsMap++  -- debugPrint $ "filteredMap"+  -- mapM_ (\((ot,k), s) -> mapM_ (\(addr,_,_) -> debugPrint $ show addr) $ S.toList s) $ M.toList filteredMap++  -- Update state. First obtain locks...+  !cache      <- takeMVar $ cm^.cacheMVar+  !cursor     <- takeMVar $ cm^.cursorMVar+  !deps       <- takeMVar $ cm^.depsMVar+  !lastGCAddr <- takeMVar $ cm^.lastGCAddrMVar+  !lastGCTime <- takeMVar $ cm^.lastGCTimeMVar+  !inclTxns   <- takeMVar $ cm^.includedTxnsMVar++  let core =+        mapM_ (\((ot,k), filteredSet) ->+          updateCache ot k filteredSet+            (case M.lookup (ot,k) newCursor of {Nothing -> M.empty; Just m -> m})+            (case M.lookup (ot,k) gcMarkerMap of {Nothing -> error "fetchUpdates(1)"; Just x -> x})) $ M.toList filteredMap++  let CacheUpdateState !newCache !new2Cursor !newDeps !newLastGCAddr !newLastGCTime !newInclTxns =+        execState core (CacheUpdateState cache cursor deps lastGCAddr lastGCTime inclTxns)++  -- debugPrint $ "finalCursor"+  -- mapM_ (\((ot,k), m) -> mapM_ (\(sid,sqn) -> debugPrint $ show $ Addr sid sqn) $ M.toList m) $ M.toList new2Cursor++  -- Flush cache if necessary {- Catch: XXX KC: A query cannot desire to see more than 1024 objects -}+  if M.size newCache < cCACHE_MAX_OBJS+  then do+    putMVar (cm^.cacheMVar) newCache+    putMVar (cm^.cursorMVar) new2Cursor+    putMVar (cm^.depsMVar) newDeps+    putMVar (cm^.lastGCAddrMVar) newLastGCAddr+    putMVar (cm^.lastGCTimeMVar) newLastGCTime+    putMVar (cm^.includedTxnsMVar) newInclTxns+  else do+    -- Reset almost everything+    putMVar (cm^.cacheMVar) M.empty+    putMVar (cm^.cursorMVar) M.empty+    putMVar (cm^.depsMVar) M.empty+    putMVar (cm^.lastGCAddrMVar) M.empty+    putMVar (cm^.lastGCTimeMVar) M.empty+    putMVar (cm^.includedTxnsMVar) (S.empty, M.empty)++    takeMVar (cm^.hwmMVar)+    putMVar (cm^.hwmMVar) M.empty+    takeMVar (cm^.diskRowCntMVar)+    putMVar (cm^.diskRowCntMVar) M.empty+    takeMVar (cm^.hotLocsMVar)+    putMVar (cm^.hotLocsMVar) S.empty++    -- fetch updates again+    fetchUpdates cm const todoList+  where+    buildCursorFromGCMarker (sid, sqn, deps,_) =+      S.foldl (\m (Addr sid sqn) ->+            case M.lookup sid m of+              Nothing -> M.insert sid sqn m+              Just oldSqn -> M.insert sid (max oldSqn sqn) m) (M.singleton sid sqn) deps+    updateCache ot k filteredSet gcCursor gcMarker = do+      -- Handle GC+      lgca <- use lastGCAddrCUS+      let cacheGCId = M.lookup (ot,k) lgca+      -- If a new GC has occurred, flush the cache and get the new effects+      -- inserted by the GC.+      when (newGCHasOccurred cacheGCId gcMarker) $ do+        -- Update GC information+        let (newGCSessID, _, _, atTime) =+              case gcMarker of+                Nothing -> error "fetchUpdates(2)"+                Just x -> x+        lgca <- use lastGCAddrCUS+        lastGCAddrCUS .= M.insert (ot,k) newGCSessID lgca+        lgct <- use lastGCTimeCUS+        lastGCTimeCUS .= M.insert (ot,k) atTime lgct+        -- empty cache+        cache <- use cacheCUS+        cacheCUS .= M.insert (ot,k) S.empty cache+        -- reset cursor+        cursor <- use cursorCUS+        cursorCUS .= M.insert (ot,k) M.empty cursor+        -- reset deps+        deps <- use depsCUS+        depsCUS .= M.insert (ot,k) S.empty deps++      -- Update cache+      cache <- use cacheCUS+      let newEffs :: CacheMap = M.singleton (ot,k) (S.map (\(a,e,_) -> (a,e)) filteredSet)+      cacheCUS .= M.unionWith S.union cache newEffs+      -- Update cursor+      cursor <- use cursorCUS+      let cursorAtKey = case M.lookup (ot, k) cursor of+                          Nothing -> gcCursor+                          Just m -> mergeCursorsAtKey m gcCursor+      let newCursorAtKey = S.foldl (\m (Addr sid sqn, _, _) ->+                              case M.lookup sid m of+                                Nothing -> M.insert sid sqn m+                                Just oldSqn -> if oldSqn < sqn+                                              then M.insert sid sqn m+                                              else m) cursorAtKey filteredSet+      cursorCUS .= M.insert (ot,k) newCursorAtKey cursor++      -- Update dependence+      deps <- use depsCUS+      let curDepsMap = case M.lookup (ot,k) deps of+                         Nothing -> M.empty+                         Just s -> S.foldl (\m (Addr sid sqn) ->+                                     case M.lookup sid m of+                                       Nothing -> M.insert sid sqn m+                                       Just oldSqn -> if oldSqn < sqn+                                                      then M.insert sid sqn m+                                                      else m) M.empty s+      let maxSqnMap = S.foldl (\m (Addr sid sqn,_,_) ->+                                  case M.lookup sid m of+                                    Nothing -> M.insert sid sqn m+                                    Just oldSqn -> if oldSqn < sqn+                                                   then M.insert sid sqn m+                                                   else m) curDepsMap filteredSet+      -- Just convert "Map sid sqn" to "Set (Addr sid sqn)"+      let maxSqnSet = M.foldlWithKey (\s sid sqn -> S.insert (Addr sid sqn) s) S.empty maxSqnMap+      -- Insert into deps to create newDeps+      -- OLD CODE : let newDeps = M.unionWith S.union deps $ M.singleton (ot,k) maxSqnSet+      let newDeps = M.insert (ot,k) maxSqnSet deps+      depsCUS .= newDeps++      -- Update included transactions+      (inclTxnsSet, inclTxnsMap) <- use inclTxnsCUS+      let newTxns = S.foldl (\acc (_,_,mbTxid) ->+                        case mbTxid of+                          Nothing -> acc+                          Just txid -> S.insert txid acc) S.empty filteredSet+      let newInclTxns = (S.union inclTxnsSet newTxns, M.insertWith S.union (ot,k) newTxns inclTxnsMap)+      inclTxnsCUS .= newInclTxns++    newGCHasOccurred :: Maybe SessID -> Maybe (SessID, SeqNo, S.Set Addr, UTCTime) -> Bool+    newGCHasOccurred Nothing Nothing = False+    newGCHasOccurred Nothing (Just _) = True+    newGCHasOccurred (Just _) Nothing = error "newGCHasOccurred: unexpected state"+    newGCHasOccurred (Just fromCache) (Just (fromDB,_,_,_)) = fromCache /= fromDB+++isCovered :: CursorMap -> ObjType -> Key -> SessID -> SeqNo -> Bool+isCovered cursor ot k sid sqn =+  case M.lookup (ot,k) cursor of+    Nothing -> False+    Just cursorAtKey ->+      case M.lookup sid cursorAtKey of+        Nothing -> False+        Just curSqn -> sqn <= curSqn++filterUnresolved :: CursorMap+                 -> M.Map (ObjType, Key) (M.Map Addr VisitedState)+                 -> M.Map (ObjType, Key) (S.Set (Addr, Effect, Maybe TxnID))+filterUnresolved cm vs1 =+  let core = mapM_ (\((ot,k), vsObj) ->+               mapM_ (\(Addr sid sqn, _) ->+                 resolve cm ot k sid sqn) $ M.toList vsObj) $ M.toList vs1+      ResolutionState vs2 = execState core (ResolutionState vs1)+  in M.map (M.foldlWithKey (\s addr vs ->+       case vs of+         Visited (Just (eff, mbTxid)) -> S.insert (addr, eff, mbTxid) s+         otherwise -> s) S.empty) vs2++resolve :: CursorMap -> ObjType -> Key -> SessID -> SeqNo -> State ResolutionState Bool+resolve cursor ot k sid sqn = do+  vs <- lookupVisitedState cursor ot k sid sqn+  case vs of+    Visited Nothing -> return False+    Visited (Just _) -> return True+    Visiting -> return True+    NotVisited eff deps mbTxid txnDeps -> do+      -- First mark this node as visiting+      updateVisitedState ot k sid sqn Visiting+      -- Process local dependences+      res1 <- foldM (\acc (Addr sid sqn) ->+                resolve cursor ot k sid sqn >>= return . ((&&) acc)) True $ S.toList deps+      -- Process remote dependences+      res2 <- case mbTxid of+        Nothing -> return res1+        Just txid ->+          if S.size txnDeps == 0+          then return False {- Txn in progress (or) eventual consistency -}+          else foldM (\acc (TxnDep ot k sid sqn) ->+                 resolve cursor ot k sid sqn >>= return . ((&&) acc)) res1 $ S.toList txnDeps+      -- Update final state+      if res2+      then updateVisitedState ot k sid sqn (Visited $ Just (eff, mbTxid))+      else updateVisitedState ot k sid sqn (Visited Nothing)+      return res2+  where+    trueVal = Visiting+    falseVal = Visited Nothing+    lookupVisitedState cursor ot k sid sqn | sqn == 0 =+      return trueVal+    lookupVisitedState cursor ot k sid sqn | sqn > 0 = do+      if isCovered cursor ot k sid sqn+      then return trueVal+      else do+        vs <- use visitedState+        case M.lookup (ot,k) vs of+          Nothing -> return falseVal+          Just vsObj -> case M.lookup (Addr sid sqn) vsObj of+                          Nothing -> return $ Visited Nothing+                          Just val -> return $ val+    updateVisitedState ot k sid sqn val = do+      vs <- use visitedState+      let newVsObj = case M.lookup (ot,k) vs of+            Nothing -> M.insert (Addr sid sqn) val M.empty+            Just vsObj -> M.insert (Addr sid sqn) val vsObj+      visitedState .= M.insert (ot,k) newVsObj vs++-- Combines curors at a particular key. Since a cursor at a given (key, sessid)+-- records the largest sequence number seen so far, given two cursors at some+-- key k, the merge operation picks the larger of the sequence numbers for each+-- sessid.+mergeCursorsAtKey :: CursorAtKey -> CursorAtKey -> CursorAtKey+mergeCursorsAtKey = M.unionWith max++collectTransitiveRows :: Pool -> Consistency+                      -> M.Map (ObjType, Key) UTCTime+                      -> StateT CollectRowsState IO ()+collectTransitiveRows pool const lgct = do+  to <- use todoObjsCRS+  case S.minView to of+    Nothing -> return ()+    Just (x@(ot,k), xs) -> do+      -- Update todo list+      todoObjsCRS .= xs+      rm <- use rowsMapCRS+      case M.lookup x rm of+        Just _ -> return ()+        Nothing -> do -- Work+          -- Read this (ot,k)+          rows <- case M.lookup (ot,k) lgct of+                  Nothing -> liftIO $ do+                    runCas pool $ cqlRead ot const k+                  Just gcTime -> liftIO $ do+                    runCas pool $ cqlReadAfterTime ot const k gcTime+          -- Mark as read+          rowsMapCRS .= M.insert (ot,k) rows rm+          mapM_ processRow rows+      collectTransitiveRows pool const lgct+  where+    processRow (_,_,_,_,Nothing) = return ()+    processRow (sid, sqn, deps, val, Just txid) = do+      includedTxns <- use inclTxnsCRS+      -- Is this transaction id already included in the cache?+      when (not $ S.member txid includedTxns) $ do+        procTxns <- use newTxnsCRS+        -- Is this transaction already seen in this fetchUpdate?+        when (not $ M.member txid procTxns) $ do+          maybeDeps <- liftIO $ runCas pool $ readTxn txid+          case maybeDeps of+            Nothing -> -- Eventual consistency!+              return ()+            Just deps -> do+              newTxnsCRS .= M.insert txid deps procTxns+              mapM_ maybeAddObjToTodo $ S.toList deps+    maybeAddObjToTodo dep = do+      to <- use todoObjsCRS+      rm <- use rowsMapCRS+      let TxnDep ot k sid sqn = dep+      if S.member (ot,k) to || M.member (ot,k) rm+      then return ()+      else do+        todoObjsCRS .= S.insert (ot, k) to

@@ -0,0 +1,52 @@+{-# Language TemplateHaskell, EmptyDataDecls, ScopedTypeVariables #-}++module Quelea.TH (+  mkOperations,+  checkOp,+  checkTxn+) where+++import Quelea.Types+import Quelea.Contract.Language+import Quelea.Contract.TypeCheck+import Language.Haskell.TH+import Quelea.Marshall+import Quelea.Shim+import Language.Haskell.TH.Syntax (lift)+import Debug.Trace++mkOperations :: [Name] -> Q [Dec]+mkOperations l = do+  pl <- procNameList l+  let (_,consList) = unzip pl+  d1 <- dataD (return []) (mkName operationsTyConStr) [] consList [mkName "Show", mkName "Eq", mkName "Ord", mkName "Read", mkName "Enum"]+  let ap = appT ([t| OperationClass |]) (conT $ mkName operationsTyConStr)+  d2 <- instanceD (return []) ap [funD 'getObjType $ map mkGetObjType pl]+  return $ [d1,d2]+  where+    procNameList :: [Name] -> Q [(String,ConQ)]+    procNameList [] = return []+    procNameList (x:xs) = do+      TyConI (DataD _ (typeName::Name) _ constructors _) <- reify x+      let typeNameStr = nameBase typeName+      let consNameStrList = map (\ (NormalC name _) -> nameBase name) constructors+      let consList = map (\s -> normalC (mkName $ take (length s - 1) s) []) consNameStrList+      let pairList = map (\c -> (typeNameStr, c)) consList+      rest <- procNameList xs+      return $ pairList ++ rest++    mkGetObjType :: (String, ConQ) -> ClauseQ+    mkGetObjType (objType, con) = do+      NormalC conName _ <- con+      return $ Clause [ConP conName []] (NormalB (LitE (StringL objType))) []++checkOp :: OperationClass a => a -> Contract a -> ExpQ+checkOp kind c = do+  a <- classifyOperContract c $ show kind+  lift a++checkTxn :: OperationClass a => String -> Fol a -> ExpQ+checkTxn str c = do+  a <- classifyTxnContract c str+  lift a

@@ -0,0 +1,178 @@+{-# Language TemplateHaskell, EmptyDataDecls, ScopedTypeVariables,+    TypeSynonymInstances, FlexibleInstances #-}++module Quelea.Types (+  Cell(..),+  Deps(..),+  Effectish(..),+  Availability(..),+  DatatypeLibrary(..),+  GenOpFun(..),+  GenSumFun(..),+  ObjType(..),+  OpFun(..),+  OperationClass(..),+  OperationPayload(..),+  Request(..),+  Response(..),+  TxnPayload(..),+  TxnKind(..),++  Key(..),+  Addr(..),+  SessID(..),+  TxnID(..),+  TxnDep(..),+  TxnDepSet(..),+  GCSetting(..),+  SeqNo,+  knownUUID,++  operationsTyConStr+) where++import Database.Cassandra.CQL+import Data.Serialize as S+import Control.Applicative ((<$>))+import Data.ByteString (ByteString, head)+import Data.Either (rights)+import Data.Map (Map)+import qualified Data.Map as Map+import Data.ByteString.Char8 (pack, unpack)+import Language.Haskell.TH+import Language.Haskell.TH.Syntax+import Data.UUID hiding (show)+import Data.Int (Int64)+import Data.Maybe (fromJust)+import qualified Data.Set as S+import qualified Data.Map as M+import Data.Tuple.Select (sel1)+import Data.Time++class (CasType a, Serialize a) => Effectish a where+  summarize :: [a] -> [a]++type OpFun eff arg res = [eff] -> arg -> (res, Maybe eff)+type GenOpFun = [ByteString] -> ByteString -> (ByteString, Maybe ByteString)+type GenSumFun = [ByteString] -> [ByteString]+data Availability = Eventual | Causal | Strong deriving (Show, Eq, Ord)++data TxnKind = RC+             | MAV+             | RR deriving (Show, Eq, Ord, Read)++instance Show GenOpFun where+  show f = "GenOpFun"++instance Lift TxnKind where+  lift RC = [| RC |]+  lift MAV = [| MAV |]+  lift RR = [| RR |]++instance Lift Availability where+  lift Eventual = [| Eventual |]+  lift Causal = [| Causal |]+  lift Strong = [| Strong |]++type ObjType = String+class (Show a, Read a, Eq a, Ord a, Serialize a) => OperationClass a where+  getObjType :: a -> String++instance OperationClass () where+  getObjType _ = fail "requesting ObjType of ()"++type AvailabilityMap a = Map (ObjType, a) (GenOpFun, Availability)+type SummaryMap = Map ObjType ([ByteString] -> [ByteString])+type DependenceMap a = Map a (S.Set a)++data DatatypeLibrary a = DatatypeLibrary {+  _avMap  :: AvailabilityMap a,+  _sumMap :: SummaryMap+}++newtype Key = Key { unKey :: ByteString } deriving (Eq, Ord)++instance Show Key where+  show (Key kv) = "Key " ++ (show $ Data.ByteString.head kv)+++type SeqNo = Int64++newtype TxnID = TxnID { unTxnID :: UUID } deriving (Eq, Ord)++instance Show TxnID where+  show (TxnID uuid) = "TxnID " ++ (show . sel1 . toWords $ uuid)++newtype SessID = SessID { unSessID :: UUID } deriving (Eq, Ord)++instance Show SessID where+  show (SessID uuid) = "SessID " ++ (show . sel1 . toWords $ uuid)++type EffectVal = ByteString++data TxnPayload = RC_TxnPl  {writeBuffer :: S.Set (Addr, EffectVal)}+                | MAV_TxnPl {writeBuffer :: S.Set (Addr, EffectVal),+                       txnDepsMAV :: S.Set TxnID}+                | RR_TxnPl {cacheSI :: S.Set (Addr, EffectVal)}++data OperationPayload a = OperationPayload {+  _objTypeReq :: ObjType,+  _keyReq     :: Key,+  _opReq      :: a,+  _valReq     :: ByteString,+  _sidReq     :: SessID,+  _sqnReq     :: SeqNo,+  _txnReq     :: Maybe (TxnID, TxnPayload),+  _getDepsReq :: Bool+}++data Request a =+    ReqOper (OperationPayload a)+  | ReqTxnCommit TxnID (S.Set TxnDep)+  | ReqSnapshot (S.Set (ObjType,Key))++data Response = ResOper {+                  seqno  :: SeqNo, {- if an effect was produce (effect = Just _),+                                      then seqno is the sequence number of this+                                      effect. Otherwise, it is the seq no passed+                                      in the corresponding Request (See OperationPayload) -}+                  result :: ByteString,+                  effect :: Maybe EffectVal, {- Only relevant for RC, MAV and RR transactions -}+                  coveredTxns :: Maybe (S.Set TxnID), {- Only relevant for MAV transactions -}+                  visibilitySet :: S.Set Addr }+              | ResSnapshot (M.Map (ObjType, Key) (S.Set (Addr, EffectVal)))+              | ResCommit++operationsTyConStr :: String+operationsTyConStr = "Operation"++data Addr = Addr {+  _sessid :: SessID,+  _seqno  :: SeqNo+} deriving (Eq, Ord, Show)++data TxnDep = TxnDep {+  _objTypeTx :: ObjType,+  _keyTx     :: Key,+  _sidTx     :: SessID,+  _sqnTx     :: SeqNo+} deriving (Eq, Ord, Show)++newtype TxnDepSet = TxnDepSet (S.Set TxnDep) deriving (Show, Eq)++-- The type of value stored in a row of the cassandra table+data Cell = EffectVal ByteString -- An effect value+          | GCMarker UTCTime     -- Marks a GC with GC start time+          deriving (Show, Eq)++newtype Deps = Deps (S.Set Addr) deriving (Show, Eq)++{- TODO: GCMarker should include a set of transaction identifiers corresponding+ - to the transactions to which the GC'ed effects belonged to. Otherwise, do+ - not GC effects that belong to a transaction.+ -}++knownUUID :: UUID+knownUUID = fromJust $ fromString $ "123e4567-e89b-12d3-a456-426655440000"++data GCSetting = No_GC | GC_Mem_Only | GC_Full deriving (Read, Show)

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain