happstack-state (empty) → 0.1
raw patch · 24 files changed
+2646/−0 lines, 24 filesdep +HUnitdep +QuickCheckdep +basesetup-changed
Dependencies added: HUnit, QuickCheck, base, binary, bytestring, containers, directory, extensible-exceptions, filepath, happstack-data, happstack-util, hslogger, hspread, mtl, old-time, random, stm, syb, template-haskell, unix
Files
- COPYING +29/−0
- Setup.hs +3/−0
- happstack-state.cabal +85/−0
- src/HAppS/State.hs +37/−0
- src/HAppS/State/Checkpoint.hs +172/−0
- src/HAppS/State/ComponentSystem.hs +120/−0
- src/HAppS/State/ComponentTH.hs +245/−0
- src/HAppS/State/Control.hs +151/−0
- src/HAppS/State/Monad.hs +112/−0
- src/HAppS/State/Saver.hs +39/−0
- src/HAppS/State/Saver/Impl/File.hs +98/−0
- src/HAppS/State/Saver/Impl/Memory.hs +68/−0
- src/HAppS/State/Saver/Impl/Queue.hs +70/−0
- src/HAppS/State/Saver/Types.hs +16/−0
- src/HAppS/State/Spread.hs +164/−0
- src/HAppS/State/Transaction.hs +497/−0
- src/HAppS/State/TxControl.hs +59/−0
- src/HAppS/State/Types.hs +100/−0
- src/HAppS/State/Util.hs +100/−0
- tests/HAppS/State/Tests.hs +24/−0
- tests/HAppS/State/Tests/CheckpointProperties.hs +236/−0
- tests/HAppS/State/Tests/Helpers.hs +108/−0
- tests/HAppS/State/Tests/SaverProperties.hs +94/−0
- tests/Test.hs +19/−0
+ COPYING view
@@ -0,0 +1,29 @@+Copyright (c) 2006, HAppS.org+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 HAppS.org; 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 OWNER OR+CONTRIBUTORS 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.
+ Setup.hs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+import Distribution.Simple+main = defaultMainWithHooks defaultUserHooks
+ happstack-state.cabal view
@@ -0,0 +1,85 @@+Name: happstack-state+Version: 0.1+Synopsis: Event-based distributed state.+Description: Unplug your machine and restart and have your app recover to exactly where it left off. HAppS-State spares you the need to deal with all the marshalling, consistency, and configuration headache that you would have if you used an external DBMS for this purpose. Its component model makes it easy to compose big applications from smaller reliable parts. Use event subscription to trigger IO actions and support comet-style or irc-bot applications. +License: BSD3+License-file: COPYING+Author: Happstack team, HAppS LLC+Maintainer: Happstack team <happs@googlegroups.com>+homepage: http://happstack.com+Category: Web, Distributed Computing+Cabal-Version: >= 1.2.3+Build-Type: Simple++flag base4++Flag tests+ Description: Build the testsuite, and include the tests in the library+ Default: True++Library++ Exposed-modules: + HAppS.State+-- HAppS.State.Logger+ HAppS.State.Saver+ HAppS.State.Control+ HAppS.State.Transaction+ HAppS.State.ComponentTH+ HAppS.State.ComponentSystem+ if flag(tests)+ Exposed-modules: + HAppS.State.Tests+ HAppS.State.Tests.Helpers+ HAppS.State.Tests.CheckpointProperties+ HAppS.State.Tests.SaverProperties+ Other-modules: + HAppS.State.Checkpoint+ HAppS.State.Spread+ HAppS.State.Monad+ HAppS.State.Saver.Impl.File+ HAppS.State.Saver.Impl.Memory+ HAppS.State.Saver.Impl.Queue+ HAppS.State.Saver.Types+-- HAppS.State.OperationModes+ HAppS.State.Types+ HAppS.State.Util+ HAppS.State.TxControl+ Build-Depends: mtl,stm, template-haskell, hslogger >= 1.0.2, happstack-util,+ happstack-data, bytestring, containers, random, old-time,+ directory, binary, filepath, hspread >=0.3, extensible-exceptions+ if flag(base4)+ Build-Depends: base >= 4 && < 5, syb+ else+ Build-Depends: base < 4+ if flag(tests)+ Build-Depends: QuickCheck, HUnit++ hs-source-dirs: src+ if flag(tests)+ hs-source-dirs: tests+ + if !os(windows)+ Build-Depends: unix+ cpp-options: -DUNIX++ Extensions: TemplateHaskell, FlexibleInstances, FlexibleContexts,+ MultiParamTypeClasses, DeriveDataTypeable,+ OverlappingInstances, UndecidableInstances, CPP,+ TypeFamilies, FunctionalDependencies, TypeOperators,+ GADTs, EmptyDataDecls, ExistentialQuantification, RankNTypes,+ ScopedTypeVariables, PatternSignatures, PatternGuards,+ MagicHash, TypeSynonymInstances+ ghc-options: -Wall+ GHC-Prof-Options: -auto-all++Executable happstack-state-tests+ Main-Is: Test.hs+ GHC-Options: -threaded+ Build-depends: HUnit+ hs-source-dirs: tests, src+ if flag(tests)+ Buildable: True+ else+ Buildable: False+
+ src/HAppS/State.hs view
@@ -0,0 +1,37 @@+module HAppS.State+ (-- * ACID monad+ Ev, AnyEv,+ TxControl, query, update, Update, Query,+ -- * Types+ TxId, EpochMilli, TxConfig(..),nullTxConfig,Saver(..),+ -- * Misc utilities+ module HAppS.State.Monad,+ getEventId, getTime, getEventClockTime, getEventStream, EventItem(..),+ module HAppS.State.Util,+ -- * Serialization+ module HAppS.Data.Serialize,+ module HAppS.Data.SerializeTH,++ module HAppS.State.Control,+ module HAppS.State.TxControl,+ module HAppS.State.ComponentTH,+ module HAppS.State.ComponentSystem,+ closeTxControl,+ createCheckpoint,+ -- * Unsafe things+ unsafeIOToEv+ ) where+++import HAppS.State.Monad+import HAppS.State.Saver+import HAppS.Data.Serialize+import HAppS.Data.SerializeTH+import HAppS.State.Transaction+import HAppS.State.Checkpoint+import HAppS.State.ComponentSystem+import HAppS.State.Types+import HAppS.State.Util+import HAppS.State.ComponentTH+import HAppS.State.TxControl+import HAppS.State.Control
+ src/HAppS/State/Checkpoint.hs view
@@ -0,0 +1,172 @@+{-# LANGUAGE CPP, TemplateHaskell, DeriveDataTypeable #-}+module HAppS.State.Checkpoint+ ( createTxControl+ , closeTxControl+ , restoreState+ , createCheckpoint+ ) where+++import HAppS.State.Saver+import HAppS.Data.Serialize+import HAppS.Data.SerializeTH+import HAppS.State.Transaction+import HAppS.State.ComponentSystem++import Data.Typeable+import Data.Maybe+import Control.Concurrent+import Control.Monad+import Control.Exception as E+import qualified Data.ByteString.Lazy.Char8 as L+import qualified Data.Map as M+import System.IO+import qualified Data.Map as M++import System.Log.Logger++logMC :: Priority -> String -> IO ()+logMC = logM "HAppS.State.Checkpoint"++{- State on disk:++* ${TXD}/events event files in ascending order+* ${TXD}/checkpoints checkpoint files in ascending order+* ${TXD}/current pointer to last checkpoint++-}++data State = State+ { stateVersion :: Int+ , stateCutoff :: Int+ } deriving (Typeable,Show)++instance Version State+$(deriveSerialize ''State)++createTxControl :: (Methods state, Component state) =>+ Saver -> Proxy state -> IO (MVar TxControl)+createTxControl saver prox+ = do -- The state hasn't been loaded yet. Ignore events.+ eventSaverVar <- newMVar =<< createWriter NullSaver "events" 0+ newMVar $ TxControl+ { ctlSaver = saver+ , ctlEventSaver = eventSaverVar+ , ctlAllComponents = allStateTypes prox+ , ctlComponentVersions = componentVersions prox+ , ctlChildren = [] }++closeTxControl :: MVar TxControl -> IO ()+closeTxControl ctlVar+ = do ctl <- takeMVar ctlVar+ writerClose =<< takeMVar (ctlEventSaver ctl)++-- FIXME: It may be nice to print out what components were saved on disk+-- compared to the components actually used in the application.+-- | Load state from disk and re-run any needed events to+-- fully restore the state. The returned function enables+-- event logging.+restoreState :: MVar TxControl -> IO (IO ())+restoreState ctlVar+ = withMVar ctlVar $ \ctl ->+ do -- Find the last saved cutoff point.+ mbState <- readState ctl+ case mbState of+ Nothing ->+ do writeState ctl (State 0 0)+ -- No events to replay. Switch to real saver.+ return $ do swapMVar (ctlEventSaver ctl) =<< createWriter (ctlSaver ctl) "events" 0+ return ()+ Just state ->+ do let cutoff = stateCutoff state+ -- Load state and replay events.+ loadState ctl cutoff+ offset <- loadEvents ctl cutoff+ -- We use a NullSaver when replaying events. Switch to real saver.+ return $ do swapMVar (ctlEventSaver ctl) =<< createWriter (ctlSaver ctl) "events" (cutoff+offset)+ return ()++-- Load state from disk.+loadState :: TxControl -> Int -> IO ()+loadState ctl cutoff+ = do logMC NOTICE $ "Loading components from storage."+ checkpoints <- withReader ctl "checkpoints" cutoff $ loadCheckpoints+ forM_ (ctlAllComponents ctl) $ \stateType+ -> case M.lookup stateType (ctlComponentVersions ctl) of+ Just versions -> forM_ versions $ \stateTypeVersion + -> case M.lookup (L.unpack stateTypeVersion) checkpoints of+ Just state -> setNewState stateType state+ Nothing -> return ()+ Nothing -> return ()+ -- FIXME: Prints stats about which components weren't found on disk+ -- and which components weren't found in the dependency tree.+ logMC NOTICE "All components successfully loaded"++-- Read and execute events since last checkpoint.+loadEvents :: TxControl -> Int -> IO Int+loadEvents ctl cutoff+ = do logMC NOTICE "Loading events from storage"+ (events, offset) <- withReader ctl "events" cutoff $ readerGet+ -- Execute events. Events that predate the last checkpoint aren't executed.+ forM_ events $ \(EventLogEntry context object) -> runColdEvent context object+ logMC NOTICE "All events successfully replayed."+ return offset++withReader :: (Serialize a) => TxControl -> String -> Int+ -> (ReaderStream a -> IO c) -> IO c+withReader ctl key cutoff+ = bracket (createReader (ctlSaver ctl) key cutoff)+ (readerClose)++withWriter :: (Serialize a) => TxControl -> String -> Int+ -> (WriterStream a -> IO c) -> IO c+withWriter ctl key cutoff+ = bracket (createWriter (ctlSaver ctl) key cutoff)+ (writerClose)++readState :: (Serialize a) => TxControl -> IO (Maybe a)+readState ctl+ = withReader ctl "current" 0 $ \s ->+ liftM listToMaybe $ readerGetUncut s++writeState :: TxControl -> State -> IO ()+writeState ctl state+ = bracket (createWriter (ctlSaver ctl) "current" 0)+ (writerClose)+ (\saver -> writerAtomicReplace saver state)++-- FIXME: Show process while loading state?+-- Might not be useful here since we just load the raw data.+-- The time consuming parsing takes place later on.+-- | Load a map from component types to serialized states.+loadCheckpoints :: ReaderStream (M.Map String L.ByteString) -> IO (M.Map String L.ByteString)+loadCheckpoints saver+ = do checkpointss <- readerGetUncut saver+ case checkpointss of+ [checkpoints] -> return checkpoints+ [] -> return M.empty+ _ -> error "Failed to read checkpoints."++saveCheckpoints :: WriterStream (M.Map String L.ByteString) -> M.Map String L.ByteString -> IO ()+saveCheckpoints saver checkpoints+ = do mv <- newEmptyMVar+ writerAdd saver checkpoints (putMVar mv ())+ takeMVar mv++-- Each checkpoint has a separate event queue. We don't want+-- to clear the queues since that would block event execution.+-- Hence, events may be written to the log after we make the cut+-- and before the state is saved. This means that some events+-- may need to be discarded next time the state is restored.+createCheckpoint :: MVar TxControl -> IO ()+createCheckpoint ctlVar+ = withMVar ctlVar $ \ctl ->+ do logMC NOTICE "Initiating checkpoint."+ newCut <- writerCut =<< readMVar (ctlEventSaver ctl)+ withWriter ctl "checkpoints" newCut $ \saver ->+ do allStates <- mapM getState (ctlAllComponents ctl)+ saveCheckpoints saver (M.fromList $ zip (ctlAllComponents ctl) allStates)+ writeState ctl $ State {stateVersion = 0+ ,stateCutoff = newCut}+ logMC NOTICE "Checkpoint successfully serialized."+
+ src/HAppS/State/ComponentSystem.hs view
@@ -0,0 +1,120 @@+{-# LANGUAGE TypeFamilies, FlexibleContexts, FunctionalDependencies, CPP,+ MultiParamTypeClasses, TypeOperators, DeriveDataTypeable, GADTs #-}+module HAppS.State.ComponentSystem where++import HAppS.Data.Serialize+import HAppS.Data.Proxy+import HAppS.State.Types++import Data.Typeable+import qualified Data.Map as Map+import Data.Map (Map)+import Control.Monad.State.Strict+import qualified Data.ByteString.Lazy.Char8 as L++--------------------------------------------------------------+-- Type level list+-------------------------------------------------------------+data End = End+data h :+: t = h :+: t+infixr 6 :+:+++--------------------------------------------------------------+-- Event classes+--------------------------------------------------------------++-- TH doesn't support type families yet.+class (Serialize ev, Serialize res) => UpdateEvent ev res | ev -> res+class (Serialize ev, Serialize res) => QueryEvent ev res | ev -> res++--------------------------------------------------------------+-- Methods contain the query and update handlers for a component+--------------------------------------------------------------++data Method st where+ Update :: (UpdateEvent ev res) => (ev -> Update st res) -> Method st+ Query :: (QueryEvent ev res) => (ev -> Query st res) -> Method st++instance Show (Method st) where+ show method = "Method: " ++ methodType method+++methodType :: Method t -> String+methodType m = case m of+ Update fn -> let ev :: (ev -> Update st res) -> ev+ ev _ = undefined+ in show (typeOf (ev fn))+ Query fn -> let ev :: (ev -> Query st res) -> ev+ ev _ = undefined+ in show (typeOf (ev fn))++class Methods a where+ methods :: Proxy a -> [Method a]++data MethodMap where+ MethodMap :: (Component st) => Map String (Method st) -> MethodMap++instance Show MethodMap where+ show (MethodMap m) = show m++-- State type -> method map+type ComponentTree = Map String MethodMap++-- State type -> all versions+type ComponentVersions = Map String [L.ByteString]++--------------------------------------------------------------+-- A component consists of a list of subcomponents, some+-- initiation code and some methods.+-- The methods are in a different class so they can be generated+-- separately.+--------------------------------------------------------------+class (SubHandlers (Dependencies a),Serialize a) => Component a where+ type Dependencies a+ initialValue :: a+ onLoad :: Proxy a -> IO ()+ onLoad _ = return ()++--------------------------------------------------------------+-- Class for walking the component tree.+--------------------------------------------------------------+class SubHandlers a where+ subHandlers :: a -> Collect ()++instance SubHandlers End where+ subHandlers ~End = return ()+instance (Methods a, Component a, SubHandlers b) => SubHandlers (a :+: b) where+ subHandlers ~(a :+: b) = do collectHandlers' (proxy a)+ subHandlers b+++data Collection = Collection ComponentTree ComponentVersions [IO ()]++addItem :: (MonadState Collection m) => + String -> MethodMap -> [L.ByteString] -> IO () -> m ()+addItem key item versions ioAction+ = do Collection tree allVersions ioActions <- get+ case Map.member key tree of+ False -> put $ Collection (Map.insert key item tree) (Map.insert key versions allVersions) (ioAction:ioActions)+ True -> dup key++type Collect = State Collection++collectHandlers :: (Methods a, Component a) => Proxy a -> (ComponentTree, ComponentVersions, [IO ()])+collectHandlers prox+ = case execState (collectHandlers' prox) (Collection Map.empty Map.empty []) of+ Collection tree versions ioActions -> (tree, versions, ioActions)++collectHandlers' :: (Methods a, Component a) => Proxy a -> Collect ()+collectHandlers' prox+ = let key = show (typeOf (unProxy prox))+ item = MethodMap $ Map.fromList [ (methodType m, m) | m <- methods prox ]+ versions = collectVersions prox+ in do addItem key item versions (onLoad prox)+ subHandlers (sub prox)+ where sub :: Component a => Proxy a -> Dependencies a+ sub _ = undefined++dup :: String -> b+dup key = error $ "Duplicate component: " ++ key
+ src/HAppS/State/ComponentTH.hs view
@@ -0,0 +1,245 @@+{-# LANGUAGE FlexibleInstances, TemplateHaskell, CPP, PatternGuards #-}+module HAppS.State.ComponentTH+ ( mkMethods+ ) where++import Data.Char+import Language.Haskell.TH++import HAppS.State.Types+import HAppS.Data.Serialize+import HAppS.State.ComponentSystem++import Control.Monad.State+import Control.Monad.Reader+import Control.Monad.Writer+import Data.Maybe++import Data.List++import Data.Generics.Basics++nubCxt :: [TypeQ] -> Q [Type]+nubCxt tsQ+ = do ts <- cxt tsQ+ return $ nub ts++{-+ Error cases:+ not all state keys are tyvars.+ method not using the keys in either the args or the result.+ Checked: component not being a data declaration with a single constructor.+ Checked: method using a tyvar that isn't a key.+ Checked: method is not a function.+-}+mkMethods :: Name -> [Name] -> Q [Dec]+mkMethods componentName componentMethods+ = do keys <- liftM (requireSimpleCon componentName) $ reify componentName+ methodInfos <- getMethodInfos keys componentMethods+ ds1 <- genEventInstances methodInfos+ let handlers = genComponentHandlers methodInfos+ stType = mkType componentName keys+ context' = mkKeyConstraints keys +++ concatMap (mkMethodConstraints keys) methodInfos+ ds2 <- instanceD (nubCxt context') (appT (conT ''Methods) stType) [ funD 'methods [clause [wildP] (normalB handlers) []] ]+ ds4 <- genMethodStructs [''Typeable] methodInfos+ ds5 <- genSerializeInstances methodInfos+ return (ds1 ++ [ds2] ++ ds4 ++ ds5 )++mkKeyConstraints :: [Name] -> [TypeQ]+mkKeyConstraints keys+ = [ appT (conT ''Typeable) (varT key) | key <- keys ] +++ [ appT (conT ''Serialize) (varT key) | key <- keys ]++mkMethodConstraints :: [Name] -> MethodInfo -> [TypeQ]+mkMethodConstraints keys method+ = map return (substMethodContext method keys)++substMethodContext :: MethodInfo -> [Name] -> [Type]+substMethodContext method keys+ = let relation = zip (methodKeys method) keys+ worker (VarT old) | Just new <- lookup old relation+ = VarT new+ worker (AppT l r) = AppT (worker l) (worker r)+ worker (ForallT c names t) = ForallT c names (worker t)+ worker t = t+ in map worker (methodContext method)++mkType :: Name -> [Name] -> TypeQ+mkType name args = foldl appT (conT name) (map varT args)++genSerializeInstances :: [MethodInfo] -> Q [Dec]+genSerializeInstances meths+ = liftM concat $ forM meths $ \method ->+ let constraints = nubCxt $ mkKeyConstraints (methodKeys method) ++ map return (methodContext method)+ upperMethod = upperName (methodName method)+ encode = do args <- replicateM (length (methodArgs method)) (newName "arg")+ lamE [conP upperMethod $ map varP args ] $+ doE $ [ noBindS $ appE (varE 'safePut) (varE arg) | arg <- args] +++ [ noBindS [| return () |] ]+ decode = do args <- replicateM (length (methodArgs method)) (newName "arg")+ doE $ [ bindS (varP arg) (varE 'safeGet) | arg <- args] +++ [ noBindS $ appE (varE 'return) $ foldl appE (conE upperMethod) $ map varE args ]+ in do s <- instanceD constraints+ (appT (conT ''Serialize) (mkType (upperName (methodName method)) (methodKeys method)))+ [funD 'putCopy [clause [] (normalB [| contain . $(encode) |]) []]+ ,funD 'getCopy [clause [] (normalB [| contain $(decode) |]) []]]+ v <- instanceD constraints (appT (conT ''Version) (mkType (upperName (methodName method)) (methodKeys method))) []+ return [s,v]+++{-+ [ Update $ \(SetComponent c) -> setComponent c+ , Query $ \GetComponent -> getComponent ]+-}+genComponentHandlers :: [MethodInfo] -> ExpQ+genComponentHandlers meths+ = do let localHandlers = flip map meths $ \method ->+ let upName = upperName (methodName method)+ in do args <- replicateM (length (methodArgs method)) (newName "arg")+ appE (conE (methodEv method)) $+ lamE [conP upName (map varP args)] $ foldl appE (varE (methodName method)) $ map varE args+ handlers = listE localHandlers+ handlers+++genEventInstances :: [MethodInfo] -> Q [Dec]+genEventInstances methodsInfo+ = mapM genEventInstance methodsInfo++genEventInstance :: MethodInfo -> Q Dec+genEventInstance method+ = do let keys = methodKeys method+ eventType = foldl appT (conT (upperName (methodName method))) (map varT keys)+ instanceD (nubCxt $ [appT (conT ''Serialize) eventType+ ,appT (conT ''Serialize) (return (methodResult method))]+ ++ mkKeyConstraints keys+ ++ mkMethodConstraints keys method+ )+ (appT (appT (conT (methodClass method)) eventType) (return (methodResult method)))+ []+++genMethodStructs :: [Name] -> [MethodInfo] -> Q [Dec]+genMethodStructs derv meths+ = liftM concat (mapM (genMethodStruct derv) meths)++-- FIXME: allow class constraints on keys.+genMethodStruct :: [Name] -> MethodInfo -> Q [Dec]+genMethodStruct derv method+ = do let c = NormalC (upperName (methodName method)) (zip (repeat NotStrict ) (methodArgs method))+ return [ DataD [] (upperName (methodName method)) (methodKeys method) [c] (derv) ]++upperName :: Name -> Name+upperName = mkName . upperFirst . nameBase++upperFirst :: String -> String+upperFirst (x:xs) = toUpper x : xs+upperFirst "" = error "ComponentTH.UpperFirst []"++data MethodInfo = Method { methodName :: Name+ , methodKeys :: [Name]+ , methodContext:: [Type]+ , methodArgs :: [Type]+ , methodClass :: Name+ , methodEv :: Name+ , methodResult :: Type+ }++-- get and validate method information.+getMethodInfos :: [Name] -> [Name] -> Q [MethodInfo]+getMethodInfos sessionKeys names+ = do ms <- mapM getMethodInfo names+ mapM worker ms+ where worker m | length (methodKeys m) /= length sessionKeys+ = error $ "Inconsistent keys: " ++ pprint (methodName m) ++ ": " ++ show (sessionKeys, methodKeys m)+ | otherwise = case compare (sort (methodTyVars m)) (sort (methodKeys m)) of+ EQ -> return m+ GT -> error $ "Method too general: " ++ pprint (methodName m)+ LT -> error $ "Method not general enough: " ++ pprint (methodName m)+ getArgKeys (AppT t1 t2) = getArgKeys t1 ++ getArgKeys t2+ getArgKeys (VarT key) = [key]+ getArgKeys _ = []+ methodTyVars m = nub $ concatMap getArgKeys (methodResult m:methodArgs m)++getMethodInfo :: Name -> Q MethodInfo+getMethodInfo method+ = do methodInfo <- reify method+ case methodInfo of+ VarI _name funcType _decl _fixity -> return (getTypes funcType){methodName = method}+ _ -> error $ "Method is not a function: " ++ nameBase method ++ " is a " ++ showInfo methodInfo++showInfo :: Info -> String+showInfo (ClassI _) = "class"+showInfo (TyConI _) = "type constructor"+showInfo (PrimTyConI _ _ _) = "primitive type constructor"+showInfo (DataConI _ _ _ _) = "data constructor"+showInfo (VarI _ _ _ _) = "variable"+showInfo (TyVarI _ _) = "type variable"+showInfo x = pprint x+++-- Cases:+-- forall m. MonadState state m => X -> m Y+-- forall key. key -> Update ()+-- forall key m. MonadState state m => key -> m ()+-- X -> Ev (ReaderT state STM) Y+-- X -> Ev (StateT state STM) Y+getTypes :: Type -> MethodInfo+getTypes (ForallT _ contxt t) = getTypes' contxt t+getTypes t = getTypes' [] t++-- FIXME: only allow type variables used by the component.+getTypes' :: Cxt -> Type -> MethodInfo+getTypes' contxt t+ = case runWriter (worker t) of+ ((keys,className, typeName, res), args) -> Method { methodName = error "Method name not set", methodKeys = keys+ , methodContext = filter (isRelevant keys) contxt+ , methodArgs = args [], methodClass = className+ , methodEv = typeName, methodResult = res}+ where -- recursive case: A -> B+ worker (AppT (AppT ArrowT t1) t2)+ = do tell (t1:)+ worker t2+ -- end case: Update state res || Query state res+ worker (AppT (AppT (ConT c) state) r)+ | c == ''Update = return (getStateKeys state,''UpdateEvent, 'Update, r)+ | c == ''Query = return (getStateKeys state,''QueryEvent, 'Query, r)+ -- end case: Ev (ReaderT state STM) res || Ev (StateT state STM) res+ worker (AppT (AppT (ConT _ev) (AppT (AppT (ConT m) state) (ConT _stm))) r)+ | m == ''StateT = return (getStateKeys state,''UpdateEvent, 'Update, r)+ | m == ''ReaderT = return (getStateKeys state,''QueryEvent, 'Query, r)+ -- end case: m res (check if m is an instance of MonadState)+ worker (AppT name r)+ | Just state <- isMonadState contxt name = return (getStateKeys state, ''UpdateEvent, 'Update, r)+ | Just state <- isMonadReader contxt name = return (getStateKeys state, ''QueryEvent, 'Query, r)+ -- error case+ worker c = error ("Unexpected method type: " ++ pprint c)++getStateKeys :: Type -> [Name]+getStateKeys (AppT r r') = getStateKeys r ++ getStateKeys r'+getStateKeys (VarT key) = [key]+getStateKeys (ConT _st) = []+getStateKeys v = error $ "Bad state type: " ++ pprint v ++ " (expected a constant, an application or a type variable)"++isMonadState :: [Type] -> Type -> Maybe Type+isMonadState contxt name = listToMaybe [ state | AppT (AppT (ConT m) state) mName <- contxt, + mName == name, m == ''MonadState ]++isMonadReader :: [Type] -> Type -> Maybe Type+isMonadReader contxt name = listToMaybe [ state | AppT (AppT (ConT m) state) mName <- contxt, + mName == name, m == ''MonadReader ]++isRelevant :: [Name] -> Type -> Bool+isRelevant keys t = isAcceptableContext t && any (`elem` keys) (getStateKeys t)++isAcceptableContext :: Type -> Bool+isAcceptableContext (AppT r r') = isAcceptableContext r && isAcceptableContext r'+isAcceptableContext (ConT con) = con `notElem` [''MonadState, ''MonadReader]+isAcceptableContext _ = True++requireSimpleCon :: Name -> Info -> [Name]+requireSimpleCon _ (TyConI (DataD _ _ names _ _derv)) = names+requireSimpleCon _ (TyConI (NewtypeD _ _ names _ _derv)) = names+requireSimpleCon _ (TyConI (TySynD _ names _)) = names+requireSimpleCon name _ = error $ "Cannot create component from '"++pprint name++"'. Expected a data structure."
+ src/HAppS/State/Control.hs view
@@ -0,0 +1,151 @@+{-# LANGUAGE CPP, EmptyDataDecls #-}++module HAppS.State.Control+ ( startSystemState+ , startSystemStateMultimaster+ , stdSaver+ , waitForTermination+ ) where++import System.Log.Logger+import qualified System.Log.Handler as SLH+import System.Log.Handler.Simple+import System.Log.Handler.Syslog++import Control.Monad+import Data.List+import Data.Char+import Data.Maybe+import System.Environment+import System.IO+import System.Exit+import System.Console.GetOpt+import Control.Monad.Trans+import Control.Concurrent++#ifdef UNIX+import System.Posix.Signals hiding (Handler)+import System.Posix.IO ( stdInput )+import System.Posix.Terminal ( queryTerminal )+#endif++import HAppS.State.Transaction+import HAppS.State.Saver+import HAppS.State.TxControl+import HAppS.State.ComponentSystem+import HAppS.Data.Proxy hiding (proxy)++startSystemState :: (Methods a, Component a) =>+ Proxy a -> IO (MVar TxControl)+startSystemState proxy+ = do _txConfig <- parseArgs+ saver <- stdSaver+ runTxSystem saver proxy++startSystemStateMultimaster :: (Methods a, Component a) =>+ Proxy a -> IO (MVar TxControl)+startSystemStateMultimaster proxy+ = do _txConfig <- parseArgs+ saver <- stdSaver+ runTxSystem' True saver proxy++stdSaver :: IO Saver+stdSaver = do pn <- getProgName+ return $ Queue (FileSaver ("_local/" ++pn++"_state"))++-- | Wait for a signal.+-- On unix, a signal is sigINT or sigTERM. On windows, the signal+-- is entering 'e'.+waitForTermination :: IO ()+waitForTermination+ = do+#ifdef UNIX+ istty <- queryTerminal stdInput+ mv <- newEmptyMVar+ installHandler softwareTermination (CatchOnce (putMVar mv ())) Nothing+ case istty of+ True -> do installHandler keyboardSignal (CatchOnce (putMVar mv ())) Nothing+ return ()+ False -> return ()+ takeMVar mv+#else+ let loop 'e' = return () + loop _ = getChar >>= loop+ loop 'c'+#endif++mkTxConfig :: [Flag] -> TxConfig+mkTxConfig = foldr worker nullTxConfig+ where worker (Cluster serv) c = c{txcOperationMode = ClusterMode (fromMaybe "" serv)}+ worker (ClusterPort port) c = c{txcClusterPort = port}+ worker _ c = c++data NullLogger+instance SLH.LogHandler NullLogger where+ setLevel = error "Don't do this! This logger should not be used!"+ getLevel = error "Don't do this! This logger should not be used!"+ emit = error "Don't do this! This logger should not be used!"+ close = error "This logger should not be used!"++setLoggingSettings :: [Flag] -> IO ()+setLoggingSettings flags = do updateGlobalLogger "" (setHandlers ([] :: [NullLogger]))+ s <- streamHandler stdout DEBUG+ updateGlobalLogger "HAppS" (setHandlers [s] . setLevel WARNING)+ mapM_ worker flags+ where worker (LogTarget SysLog) = do s <- openlog "HAppS" [PID] DAEMON DEBUG -- This priority seems to be ignored?+ updateGlobalLogger "HAppS" (setHandlers [s])+ worker (LogTarget StdOut) = do s <- streamHandler stdout DEBUG+ updateGlobalLogger "HAppS" (setHandlers [s])+ worker (LogTarget (File path)) = do s <- fileHandler path DEBUG -- This priority seems to be ignored?+ updateGlobalLogger "HAppS" (setHandlers [s])+ worker (LogLevel pri) = do updateGlobalLogger "HAppS" (setLevel pri)+ rlogger <- getLogger "HAppS"+ logM "" WARNING ("Set logging priority to " ++ show (getLevel rlogger))+ worker _ = return ()++-- order should not matter, though it does now.+-- we should ALSO allow multiple loggers at the same time! --log-target=stdout --log-target=syslog+options :: [OptDescr Flag]+options = [Option "" ["log-level"] (ReqArg (LogLevel . read . map toUpper) "level")+ "Log level: DEBUG, INFO, NOTICE, WARNING, ERROR, CRITICAL, ALERT, EMERGENCY. Default: WARNING"+ ,Option "" ["log-target"] (ReqArg (LogTarget . readTarget) "target")+ "Log target: stdout, syslog, or a FilePath such as /home/foo/bar.log . Default: stdout"+ -- FIXME: use a better flag name+ ,Option "" ["cluster-mode"] (OptArg (Cluster) "servers")+ "Start in multimaster mode. Will join cluster if optional arg is given."+ ,Option "" ["cluster-port"] (ReqArg (ClusterPort . read) "port")+ "Multimaster server will listen on this port."+ ]++data Target = File FilePath | StdOut | SysLog deriving (Read,Show,Eq,Ord)++data Flag = LogLevel Priority | LogTarget Target+ | Cluster (Maybe String) | ClusterPort Int deriving Show++readTarget :: String -> Target+readTarget arg = case map toLower arg of+ "stdout" -> StdOut+ "syslog" -> SysLog+ _ -> File arg++castOptions :: [OptDescr ()]+castOptions = flip map options $ \(Option c f desc help) -> Option c f (worker desc) help+ where worker (NoArg _) = (NoArg ())+ worker (ReqArg _ f) = ReqArg (const ()) f+ worker (OptArg _ f) = OptArg (const ()) f++parseArgs :: IO TxConfig+parseArgs = do+ args <- liftIO getArgs+ pn <- liftIO getProgName+ let err n ls = -- XXX these next lines should be written to stderr!+ do putStrLn ("Syntax error in command line - "++n)+ putStrLn $ unlines $ map (" "++) ls+ putStrLn ("Usage "++usageInfo pn castOptions)+ exitWith (ExitFailure 1)+ case getOpt' Permute options args of+ (flags,_fs,_args',[]) -> do setLoggingSettings flags+ -- FIXME: replace system args with fs++args'+ return (mkTxConfig flags)+ (_,_,_,es) -> err "errors" es+
+ src/HAppS/State/Monad.hs view
@@ -0,0 +1,112 @@+{-# LANGUAGE CPP, FlexibleInstances, TypeSynonymInstances,+ MultiParamTypeClasses #-}+module HAppS.State.Monad where++import Control.Exception+import Control.Concurrent.STM+import HAppS.State.Types+import HAppS.Data.Proxy++import Control.Monad.State+import Control.Monad.Reader+import Control.Monad++instance (Monad m) => Monad (Ev m) where+ return x = Ev $ \_ -> return x+ fail x = Ev $ \_ -> fail x+ ev >>= f = Ev $ \env -> unEv ev env >>= \x -> unEv (f x) env++instance MonadState st (Update st) where+ get = Ev $ \_ -> get+ put x = Ev $ \_ -> put x++instance MonadReader st (Query st) where+ ask = Ev $ \_ -> ask+ local l (Ev cmd) = Ev $ \env -> local l (cmd env)++instance MonadReader st (Update st) where+ ask = Ev $ \_ -> get+ local l (Ev cmd) = Ev $ \env -> StateT $ \s ->+ do (r,_s') <- runStateT (cmd env) (l s)+ return (r,s)++instance (Monad m) => Functor (Ev m) where fmap = liftM++-- | Use a proxy to force the type of an update action.+setUpdateType :: Proxy t -> Update t ()+setUpdateType _ = return ()++proxyUpdate :: Ev (StateT t STM) b -> Proxy t -> Ev (StateT t STM) b+proxyUpdate f prox = setUpdateType prox >> f++-- | Use a proxy to force the type of a query action.+setQueryType :: Proxy t -> Query t ()+setQueryType _ = return ()++proxyQuery :: Ev (ReaderT t STM) b -> Proxy t -> Ev (ReaderT t STM) b+proxyQuery f prox = setQueryType prox >> f++-- | Currying version of 'setUpdateType'.+asUpdate :: Update t a -> Proxy t -> Update t a+asUpdate upd _ = upd++-- | Currying version of 'setQueryType'.+asQuery :: Query t a -> Proxy t -> Query t a+asQuery query _ = query++-- | Specialized version of 'ask'+askState :: Query st st+askState = ask++-- | Specialized version of 'get'+getState :: Update st st+getState = get++-- | Specialized version of 'put'.+putState :: st -> Update st ()+putState = put+++-- | Lift an STM action into Ev.+liftSTM :: STM a -> AnyEv a+liftSTM = unsafeSTMToEv++class CatchEv m where+#if __GLASGOW_HASKELL__ < 610+ catchEv :: Ev m a -> (Exception -> a) -> Ev m a+#else+ catchEv :: Ev m a -> (SomeException -> a) -> Ev m a+#endif+instance CatchEv (ReaderT st STM) where+ catchEv (Ev cmd) fun = Ev $ \s -> ReaderT $ \r -> runReaderT (cmd s) r `catchSTM` (\a -> return (fun a))++instance CatchEv (StateT st STM) where+ catchEv (Ev cmd) fun = Ev $ \s -> StateT $ \r -> runStateT (cmd s) r `catchSTM` (\a -> return (fun a,r))++instance MonadPlus m => MonadPlus (Ev m) where+ mzero = Ev $ \_ -> mzero+ mplus (Ev fn1) (Ev fn2) = Ev $ \env -> fn1 env `mplus` fn2 env++-- | Select a part of the environment.+sel :: (Env -> b) -> AnyEv b+sel f = Ev $ \env -> return (f env)++-- FIXME: should the users see this function?+-- | Run a computation with local state. Changes to state will be visible to outside.+localState :: (outer -> inner) -> (inner -> outer -> outer) -> Ev (StateT inner STM) a -> Ev (StateT outer STM) a+localState ifun ufun (Ev cmd)+ = Ev $ \env -> StateT $ \s ->+ do (r,s') <- runStateT (cmd env) (ifun s)+ return (r, ufun s' s)++-- | Run a computation with local state.+localStateReader :: (outer -> inner) -> Ev (ReaderT inner STM) a -> Ev (ReaderT outer STM) a+localStateReader ifun (Ev cmd)+ = Ev $ \env -> ReaderT $ \s ->+ runReaderT (cmd env) (ifun s)++-- | Execute a Query action in the Update monad.+runQuery :: Query st a -> Update st a+runQuery fn = Ev $ \env -> StateT $ \st ->+ do a <- runReaderT (unEv fn env) st+ return (a,st)
+ src/HAppS/State/Saver.hs view
@@ -0,0 +1,39 @@+module HAppS.State.Saver+ ( module HAppS.State.Saver.Types+ , Saver(..)+ , createReader, createWriter ) where++import Control.Concurrent+import HAppS.State.Saver.Impl.File+import HAppS.State.Saver.Impl.Memory+import HAppS.State.Saver.Impl.Queue+import HAppS.State.Saver.Types+import HAppS.Data.Serialize++data Saver = NullSaver -- ^ A saver that discards all output+ | FileSaver String -- ^ A saver that operates on files. The parameter is the prefix for the files.+ -- Creates the prefix directory.+ | Queue Saver -- ^ Enable queueing.+ | Memory (MVar Store)++createReader :: Serialize a => Saver -> String -> Int -> IO (ReaderStream a)+createReader (FileSaver prefix) key cutoff = fileReader prefix key cutoff+createReader (Memory store) key cutoff = memoryReader store key cutoff+createReader (Queue saver) key cutoff = queueReader =<< createReader saver key cutoff+createReader NullSaver _key _cutoff+ = return $ ReaderStream+ { readerClose = return ()+ , readerGet = fail "NullSaver: readerGet"+ , readerGetUncut = fail "NullSaver: readerGetUncut" }++createWriter :: Serialize a => Saver -> String -> Int -> IO (WriterStream a)+createWriter (FileSaver prefix) key cutoff = fileWriter prefix key cutoff+createWriter (Memory store) key cutoff = memoryWriter store key cutoff+createWriter (Queue saver) key cutoff = queueWriter =<< createWriter saver key cutoff+createWriter NullSaver _key _cutoff+ = return $ WriterStream+ { writerClose = return ()+ , writerAdd = \_ io -> io+ , writerAtomicReplace = fail "NullSaver: writerAtomicReplace"+ , writerCut = fail "NullSaver: writerCut" }+
+ src/HAppS/State/Saver/Impl/File.hs view
@@ -0,0 +1,98 @@+{-# LANGUAGE CPP #-}+module HAppS.State.Saver.Impl.File+ ( fileReader, fileWriter+ ) where++import HAppS.State.Saver.Types+import HAppS.Data.Serialize++import Control.Concurrent+import Control.Exception.Extensible as E+import qualified Data.ByteString.Lazy.Char8 as L+import qualified Data.ByteString.Char8 as B+import System.Directory ( createDirectoryIfMissing, renameFile, doesFileExist )+import System.IO+import System.Random ( randomIO )+import System.Log.Logger+import Text.Printf+import Control.Monad+import System.FilePath++tryE :: IO a -> IO (Either SomeException a)+tryE = E.try++catchE :: IO a -> (SomeException -> IO a) -> IO a+catchE = E.catch++logMF :: Priority -> String -> IO ()+logMF = logM "HAppS.State.Saver.Impl.File"++formatFilePath :: Int -> String -> FilePath+formatFilePath n str = printf "%s-%010d" str n++fileReader :: Serialize a => FilePath -> String -> Int -> IO (ReaderStream a)+fileReader prefix key cutoff+ = do let file = prefix </> formatFilePath cutoff key+ tryE $ createDirectoryIfMissing True prefix+ return $ ReaderStream+ { readerClose = do return ()+ , readerGet = do logMF NOTICE "fileReader: readerGet"+ allFiles <- getAllFiles prefix key cutoff+ allData <- mapM B.readFile allFiles+ return $ (parseAll (L.fromChunks allData), length allFiles)+ , readerGetUncut = do logMF NOTICE "fileReader: readerGetUncut"+ allData <- B.readFile file `catchE` \_ -> return B.empty+ return $ parseAll (L.fromChunks [allData])+ }+++parseAll :: Serialize a => L.ByteString -> [a]+parseAll = loop+ where loop l | L.null l = []+ loop l = let (a,rest) = deserialize l+ in a:loop rest++fileWriter :: Serialize a => FilePath -> String -> Int -> IO (WriterStream a)+fileWriter prefix key cutoffInit = do+ cutoffVar <- newMVar cutoffInit+ let getFileName = do cutoff <- readMVar cutoffVar+ return $ prefix </> formatFilePath cutoff key+ file <- getFileName+ logMF NOTICE ("fileWrter: "++key++" @ "++prefix)+ tryE $ createDirectoryIfMissing True prefix+ hmv <- newMVar =<< openBinaryFile file WriteMode+ return $ WriterStream+ { writerClose = withMVar hmv hClose+ , writerAdd = \m f -> do logMF NOTICE "fileWriter: saverAdd"+ withMVar hmv (\h -> L.hPut h (serialize m) >> hFlush h)+ forkIO f+ return ()+ , writerAtomicReplace = \ss -> do h <- takeMVar hmv+ hClose h+ file' <- getFileName+ atomicWriteFile file' (serialize ss)+ putMVar hmv =<< openBinaryFile file' AppendMode+ , writerCut = do h <- takeMVar hmv+ hClose h+ cutoff <- takeMVar cutoffVar+ let file' = prefix </> formatFilePath (cutoff+1) key+ putMVar cutoffVar (cutoff+1)+ putMVar hmv =<< openBinaryFile file' WriteMode+ return (cutoff+1)+ }++getAllFiles :: FilePath -> String -> Int -> IO [FilePath]+getAllFiles prefix key cutoff+ = loop cutoff+ where loop n = do let file = prefix </> formatFilePath n key+ exist <- doesFileExist file+ if exist then liftM (file:) (loop (n+1))+ else return []++-- | Just to avoid a dependency.+atomicWriteFile :: String -> L.ByteString -> IO ()+atomicWriteFile path string = do+ r <- randomIO :: IO Int+ let p' = path ++ ".atomic-tmp-" ++ show (abs r)+ L.writeFile p' string+ renameFile p' path
+ src/HAppS/State/Saver/Impl/Memory.hs view
@@ -0,0 +1,68 @@+module HAppS.State.Saver.Impl.Memory where++import HAppS.Data.Serialize+import HAppS.State.Saver.Types++import Control.Concurrent++import qualified Data.ByteString.Lazy.Char8 as L+import Data.Maybe++import qualified Data.Map as M++type Store = M.Map String (M.Map Int L.ByteString)++newMemoryStore :: IO (MVar Store)+newMemoryStore = newMVar M.empty++memoryReader :: Serialize a => MVar Store -> String -> Int -> IO (ReaderStream a)+memoryReader store key cutoff+ = do return $ ReaderStream+ { readerClose = do return ()+ , readerGet = withMVar store $ \storeData ->+ let getAllData ((cut,_):xs) n | n > cut = getAllData xs n+ getAllData ((cut,dat):xs) n | n == cut = dat : getAllData xs (n+1)+ getAllData _ _ = []+ allData = getAllData (maybe [] M.toList $ M.lookup key storeData) cutoff+ in return $ (parseAll (L.concat allData), length allData)+ , readerGetUncut = withMVar store $ \storeData ->+ return $ parseAll $ fromMaybe L.empty $ M.lookup cutoff =<< M.lookup key storeData+ }++memoryWriter :: Serialize a => MVar Store -> String -> Int -> IO (WriterStream a)+memoryWriter store key cutoffInit+ = do cutoffVar <- newMVar cutoffInit+ modifyMVar_ store $ \storeData -> return (addToStore key cutoffInit L.empty storeData)+ return $ WriterStream+ { writerClose = return ()+ , writerAdd = \m f ->+ do cutoff <- readMVar cutoffVar+ modifyMVar_ store $ \storeData -> return $ addToStore key cutoff (serialize m) storeData+ forkIO f+ return ()+ , writerAtomicReplace = \ss -> do cutoff <- readMVar cutoffVar+ modifyMVar_ store $ \storeData -> return $ setStore key cutoff (serialize ss) storeData+ , writerCut = do newCut <- modifyMVar cutoffVar $ \cutoff -> return (cutoff+1, cutoff+1)+ modifyMVar_ store $ \storeData -> return (addToStore key newCut L.empty storeData)+ return newCut+ }++parseAll :: Serialize a => L.ByteString -> [a]+parseAll = loop+ where loop l | L.null l = []+ loop l = let (a,rest) = deserialize l+ in a:loop rest++addToStore :: (Ord a, Ord b) => a -> b -> L.ByteString+ -> M.Map a (M.Map b L.ByteString)+ -> M.Map a (M.Map b L.ByteString)+addToStore key cutoff val store+ = M.unionWith (M.unionWith L.append) store element+ where element = M.singleton key $ M.singleton cutoff val++setStore :: (Ord a) => a -> b -> c+ -> M.Map a (M.Map b c)+ -> M.Map a (M.Map b c)+setStore key cutoff val store+ = M.unionWith (\_ _ -> M.singleton cutoff val) store element+ where element = M.singleton key $ M.singleton cutoff val
+ src/HAppS/State/Saver/Impl/Queue.hs view
@@ -0,0 +1,70 @@+{-# LANGUAGE DeriveDataTypeable #-}+module HAppS.State.Saver.Impl.Queue+ ( queueReader+ , queueWriter+ ) where++import HAppS.State.Saver.Types+import HAppS.Data.Serialize++import Control.Concurrent.STM+import Control.Concurrent+import Control.Monad+import Data.Typeable++import Data.Binary++data Item = Close (IO ())+ | Add Put (IO ())++queueReader :: ReaderStream a -> IO (ReaderStream a)+queueReader stream = return stream++data Encoded = Encoded Put deriving Typeable+instance Version Encoded where mode = Primitive+instance Serialize Encoded where+ putCopy (Encoded out) = contain out+ getCopy = error "decoding from writer queue."++-- | A saver that bunches writes.+queueWriter :: Serialize a => WriterStream Encoded -> IO (WriterStream a)+queueWriter writer = do+ ch <- newCh+ let handler = do+ input <- getChs ch+ let il (Add p f : rest) a0 a1 = il rest (p:a0) (f:a1)+ il something a0 a1 = (reverse a0, sequence_ $ reverse a1, something)+ let (ps,io,rest) = il input [] []+ when (not (null ps)) $ writerAdd writer (Encoded $ sequence_ ps) io+ case rest of+ [] -> handler+ (Close io':_) -> writerClose writer >> io'+ _ -> fail "queueSaver: Invalid saver bunch!"++ forkIO handler+ return $ WriterStream+ { writerClose = do+ mv <- newEmptyMVar+ writeCh ch $ Close (writerClose writer >> putMVar mv ())+ takeMVar mv+ , writerAdd = \ps fin -> writeCh ch $ Add (safePut ps) fin+ , writerAtomicReplace = \a -> writerAtomicReplace writer (Encoded $ safePut a)+ , writerCut = writerCut writer+ }+++-- Sample variables/queues+newtype Ch a = Ch (TVar [a])+newCh :: IO (Ch a)+newCh = fmap Ch $ newTVarIO []+writeCh :: Ch a -> a -> IO ()+writeCh (Ch ch) x = atomically $ do vs <- readTVar ch+ writeTVar ch (x:vs)++getChs :: Ch a -> IO [a]+getChs (Ch ch) = atomically $ do vs <- readTVar ch+ guard (not (null vs))+ writeTVar ch []+ return (reverse vs)++
+ src/HAppS/State/Saver/Types.hs view
@@ -0,0 +1,16 @@+module HAppS.State.Saver.Types where++data ReaderStream a+ = ReaderStream+ { readerClose :: IO ()+ , readerGet :: IO ([a], Int)+ , readerGetUncut :: IO [a]+ }++data WriterStream a+ = WriterStream+ { writerClose :: IO ()+ , writerAdd :: a -> IO () -> IO ()+ , writerAtomicReplace :: a -> IO ()+ , writerCut :: IO Int+ }
+ src/HAppS/State/Spread.hs view
@@ -0,0 +1,164 @@+{-# LANGUAGE TemplateHaskell, DeriveDataTypeable #-}+module HAppS.State.Spread+ ( connectToCluster+ , changeEventMapping+ ) where++import HAppS.State.Transaction+import HAppS.Data.Serialize+import HAppS.Data.SerializeTH++import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Lazy.Char8 as L+import qualified Data.Map as Map++import Data.Typeable+import Spread.Client+import qualified Control.Concurrent.Chan.Closeable as Closeable++import System.Log.Logger+import Control.Monad hiding (join)+import Control.Concurrent+import System.Random++logSP :: Priority -> String -> IO ()+logSP = logM "HAppS.State.Spread"+++data Cluster = Cluster+ { clusterChan :: Closeable.Chan Closeable.R Message+ , clusterConnection :: Connection+ }++type EventId = Int++data ClusterMsg+ = RequestState+ | NewState B.ByteString -- A piece of state (map of all components)+ | StateTransferred -- The entire state has been transmitted.+ | ForeignEvent EventId Object+ | EventResponse EventId Object+ deriving (Typeable)+instance Version ClusterMsg -- version 0+$(deriveSerialize ''ClusterMsg)++connectToCluster :: IO Cluster+connectToCluster+ = do logSP NOTICE "Connecting to spread daemon on localhost"+ uniq <- randomRIO (0,30000) -- FIXME: this is a hack to get unique names.+ let name = mkPrivateName (B.pack $ "node" ++ show (uniq::Int))+ (chan, conn) <- connect defaultConf{desiredName = name}+ startReceive conn+ join receiverGroup conn+ return $ Cluster chan conn++receiverGroup :: Group+receiverGroup = let Just g = makeGroup "receiver"+ in g++changeEventMapping :: MVar TxControl -> EventMap -> Cluster -> IO EventMap+changeEventMapping ctlVar localEventMap cluster+ = do logSP NOTICE "Create new event mapper"+ mems <- getClusterMembers cluster+ responseIndex <- newMVar Map.empty+ ready <- newEmptyMVar+ eventQueue <- newChan+ stateVar <- newMVar L.empty+ eidStore <- newMVar 0 -- eid's are locally unique.+ let newEID = modifyMVar eidStore (\store -> return (store+1,store+1))+ pushStateBlock st = modifyMVar_ stateVar (\acc -> return $ acc `L.append` L.fromChunks [st])+ insertEID eid = do mv <- newEmptyMVar+ modifyMVar_ responseIndex $ \idx -> return $ Map.insert eid mv idx+ return $ takeMVar mv+ returnResponse eid object = modifyMVar_ responseIndex $ \idx ->+ case Map.lookup eid idx of+ Nothing -> return idx -- We're already received the result from another node.+ Just mv -> do putMVar mv object -- Notify the caller about the new response.+ return $ Map.delete eid idx+ listener = forever $+ do mbMsg <- Closeable.readChan (clusterChan cluster)+ case mbMsg of+ Nothing -> error "Disconnected from the spread daemon."+ Just (Regular msg) ->+ case fst (deserialize (L.fromChunks [inData msg])) of+ RequestState -> sendState ctlVar (inSender msg) cluster+ ForeignEvent eid object -> writeChan eventQueue (inSender msg, eid, object)+ NewState st -> pushStateBlock st+ -- FIXME: check that stateVar is non-empty?+ StateTransferred -> do restoreState =<< takeMVar stateVar+ putMVar ready ()+ EventResponse eid object-> returnResponse eid object+ Just _ -> return ()+ responder = forever $+ do (sender, eid, object) <- readChan eventQueue+ response <- runObjectEventFunc object localEventMap+ sendClusterMsg [sender] (EventResponse eid response) cluster+ forkIO listener+ forkIO responder+ case mems of+ [] -> putMVar ready () -- The cluster is empty, use state from disk.+ (x:_) -> do logSP NOTICE $ "Requesting state from: " ++ show x+ sendClusterMsg [x] RequestState cluster+ logSP NOTICE "Waiting for ready signal"+ takeMVar ready+ logSP NOTICE "Ready signal received"+ let newEventMap = flip Map.map localEventMap $ \handler ->+ case handler of+ UpdateHandler runCold _ parse+ -> let runHot ev = do eid <- newEID+ logSP NOTICE $ "New eid: " ++ show eid+ wait <- insertEID eid+ sendClusterMsg [receiverGroup] (ForeignEvent eid (mkObject ev)) cluster+ response <- wait+ logSP NOTICE $ "Received response for: " ++ show eid+ return $ parseObject response+ in UpdateHandler runCold runHot parse+ QueryHandler{} -> handler+ return newEventMap++sendState :: MVar TxControl -> Group -> Cluster -> IO ()+sendState ctlVar sender cluster+ = withMVar ctlVar $ \ctl ->+ do logSP NOTICE $ "Sending state to: " ++ show sender+ allStates <- mapM getState (ctlAllComponents ctl)+ let chunks = L.toChunks $ serialize $ zip (ctlAllComponents ctl) allStates+ forM_ chunks $ \chunk ->+ do sendClusterMsg [sender] (NewState chunk) cluster+ sendClusterMsg [sender] StateTransferred cluster+ logSP NOTICE $ "The state has been transmitted."++restoreState :: L.ByteString -> IO ()+restoreState stateObject+ = do logSP NOTICE $ "Loading components from network."+ forM_ state $ \(stateType, stateData) ->+ do setNewState stateType stateData+ logSP NOTICE $ "All components successfully loaded"+ where state = fst (deserialize stateObject)++sendClusterMsg :: [Group] -> ClusterMsg -> Cluster -> IO ()+sendClusterMsg to msg cluster+ = send outMsg (clusterConnection cluster)+ where outMsg = Outgoing { outOrdering = Safe+ , outDiscard = False+ , outData = B.concat (L.toChunks (serialize msg))+ , outGroups = to+ , outMsgType = 0 }++-- Cluster members not including self.+getClusterMembers :: Cluster -> IO [PrivateGroup]+getClusterMembers cluster+ = do mbMsg <- Closeable.readChan (clusterChan cluster)+ case mbMsg of+ Just (Membership membership)+ -> case membership of+ Reg{members=m} -> return $ filter (/= privateGroup (clusterConnection cluster)) m+ Transient{} -> error $ "Received a Transient message on connect."+ SelfLeave{} -> error $ "Received a SelfLeave message on connect."+ Just msg -> error $ "Expected membership message from cluster. Received a " ++ showMsgType msg ++ " message."+ Nothing -> error "Cluster channel unexpectedly closed."++showMsgType :: Message -> String+showMsgType (Regular _) = "regular"+showMsgType (Membership _) = "membership"+showMsgType (Rejected _) = "rejected"+
+ src/HAppS/State/Transaction.hs view
@@ -0,0 +1,497 @@+{-# LANGUAGE CPP, TemplateHaskell, MagicHash,+ ExistentialQuantification, DeriveDataTypeable,+ GADTs, MultiParamTypeClasses #-}+module HAppS.State.Transaction where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception.Extensible+import Control.Monad.State+import Control.Monad.Reader+import qualified Data.Map as M+import qualified Data.ByteString.Lazy as L+import Data.IORef+import System.IO.Unsafe+import System.Random+import System.Time(getClockTime,ClockTime(TOD))+import System.Log.Logger++import HAppS.State.ComponentSystem+import HAppS.State.Monad+import HAppS.State.Saver+import HAppS.Data.Serialize+import HAppS.Data.SerializeTH+import HAppS.State.Types+import HAppS.Util.Common (Seconds)++import Data.Dynamic+import GHC.Base++import Prelude hiding (catch)++import qualified Data.Binary as Binary++type ExceptionT = SomeException++logMT :: Priority -> String -> IO ()+logMT = logM "HAppS.State.Transaction"++getTime :: (Integral epochTime) => AnyEv epochTime+getTime = sel (fromIntegral . txTime . evContext)++getEventClockTime :: AnyEv ClockTime+getEventClockTime = do milliSeconds <- sel (txTime . evContext)+ return $ TOD (fromIntegral milliSeconds) 0++getEventId :: (Integral txId) => AnyEv txId+getEventId = sel (fromIntegral . txId . evContext)+++instance Version TxContext -- Default to version 0++$(deriveSerialize ''TxContext)++-- Isomorphic to (), we're interested in a descriptive name.+data GetCheckpointState = GetCheckpointState deriving (Typeable)+instance Version GetCheckpointState+$(deriveSerialize ''GetCheckpointState)++newtype SetCheckpointState = SetCheckpointState L.ByteString deriving (Typeable)+instance Version SetCheckpointState+$(deriveSerialize ''SetCheckpointState)+++instance Version StdGen+instance Serialize StdGen where+ getCopy = contain $ liftM read safeGet+ putCopy = contain . safePut . show++{- Durablity:+* Pending queue is TChan (TxContext, ev)+* Get events from the input sources in circular fashion+* Dump events on disk before adding to pending queue+* Checkpoints as follows:+ * check point event arrives from one of the input sources+ * write a new checkpoint file with:+ + list of pending transactions (all non-pending are out of system)+ + next txid+ + save state+ + rotare log files+ * resume transaction processing+++-}+++type TypeString = String++data EventHandler where+ UpdateHandler :: UpdateEvent ev res =>+ (TxContext -> ev -> IO ()) ->+ (ev -> IO res) ->+ (Object -> ev) ->+ EventHandler+ QueryHandler :: QueryEvent ev res =>+ (ev -> IO res) ->+ (Object -> ev) ->+ EventHandler++type EventMap = M.Map TypeString EventHandler+data EmitInternal = EmitInternal EventMap++{-# NOINLINE emitRef #-}+emitRef :: IORef EmitInternal+emitRef = unsafePerformIO $ newIORef (error "HAppS not initiated")++-- Low level function for emitting events. Very unsafe, do not expose.+emitFunc :: (Serialize ev, Typeable res) =>+ EventMap -> TypeString -> ev -> IO res+emitFunc eventMap eventType ev+ = case M.lookup eventType eventMap of+ Nothing -> error $ "Emitted event to unknown component. Ev: " ++ eventType+ Just (UpdateHandler _cold hot _parse) -> unsafeCoerce# hot ev+ Just (QueryHandler fn _parse) -> unsafeCoerce# fn ev++-- Wrapper around the global emitter map. Very unsafe, do not expose.+-- This function is only safe through 'query' and 'update'.+emitEvent' :: (Serialize ev, Typeable res) => TypeString -> ev -> IO res+emitEvent' eventType ev+ = do internal <- readIORef emitRef+ case internal of+ EmitInternal eventMap -> emitFunc eventMap eventType ev++emitEvent :: (Serialize ev, Typeable res) => ev -> IO res+emitEvent ev = emitEvent' (show (typeOf ev)) ev++setNewEventMap :: EventMap -> IO ()+setNewEventMap eventMap+ = writeIORef emitRef $ EmitInternal eventMap++data EventItem = EventItem+ { eventContext :: TxContext+ , eventData :: Dynamic }++{-# NOINLINE eventStreamRef #-}+eventStreamRef :: MVar (Chan EventItem)+eventStreamRef = unsafePerformIO $ newMVar $ error "eventStreamRef not initialized."+++initEventStream :: IO ()+initEventStream = do c <- newChan+ forkIO $ forever $ readChan c -- Immediately discard items from this channel.+ swapMVar eventStreamRef c+ return ()++pushEventItem :: Serialize ev => TxContext -> ev -> IO ()+pushEventItem context ev+ = do c <- readMVar eventStreamRef+ writeChan c $ EventItem context (toDyn ev)++getEventStream :: IO (IO EventItem)+getEventStream = do c <- dupChan =<< readMVar eventStreamRef+ return (readChan c)++{-+ Events for different components can be executed in parallel.+-}+-- Casting the event and the result type is safe. The types are kept sane due to the+-- EventUpdate and EventQuery classes.+createEventMap :: (Methods st, Component st) => MVar TxControl -> Proxy st -> IO EventMap+createEventMap ctlVar componentProxy+ = do initEventStream+ maps <- forM (M.elems componentTree) $ \(MethodMap m) ->+ do tx <- createNewTxRun+ ctl <- readMVar ctlVar+ runTxLoop (ctlEventSaver ctl) (txProcessQueue tx) initialValue+ return $ M.union (extraEvents tx) (M.map (eventHandler tx) m)+ return $ M.unions maps+ where (componentTree, _versions, _ioActions) = collectHandlers componentProxy+ eventHandler tx (Update fn)+ = let updateCold' cxt ev+ = do mv <- newEmptyMVar+ let handler = do lastCxt <- readTVar (txLastTxContext tx)+ if txId lastCxt < txId cxt+ then do writeTVar (txLastTxContext tx) cxt+ writeTChan (txProcessQueue tx) $ IHR cxt ev $+ handleUpdate (putMVar mv) $ fn ev+ return $ return ()+ else return $ putMVar mv $ Right $ error "Cold event not executed"+ join $ atomically $ handler+ me <- takeMVar mv+ case me of+ Left e -> throwIO e+ Right e -> return e+ updateCold cxt ev = do updateCold' cxt ev; return ()+ updateHot ev+ = do cxt <- atomically . addTxId tx =<< newTxContext+ updateCold' cxt ev+ in UpdateHandler updateCold updateHot parseObject+ eventHandler tx (Query fn)+ = let queryEmitter ev+ = do mv <- newEmptyMVar+ quickQuery' tx $ HR ev $ handleQuery (putMVar mv) (fn ev)+ me <- takeMVar mv+ case me of+ Left e -> throwIO e+ Right e -> return e+ in QueryHandler queryEmitter parseObject+++instance QueryEvent GetCheckpointState L.ByteString+instance UpdateEvent SetCheckpointState ()++extraEvents :: Serialize st => TxRun st -> EventMap+extraEvents tx+ = M.fromList [ (getStateType stateType, getStateHandler tx)+ , (setNewStateType stateType, setNewStateHandler tx)+ ]+ where t :: TxRun st -> st+ t _ = undefined+ stateType = show (typeOf (t tx))+ getStateHandler tx'+ = let fn :: GetCheckpointState -> IO L.ByteString+ fn ev = do mv <- newEmptyMVar+ quickQuery' tx' $ HR ev $ \context st ->+ return (Nothing, putMVar mv (serialize (context, st)))+ takeMVar mv+ in QueryHandler fn parseObject+ setNewStateHandler tx'+ = let fn :: SetCheckpointState -> IO ()+ fn (SetCheckpointState bs)+ = do ((context, newState), rest) <- evaluate $ deserialize bs+ unless (L.null rest) $ error $ "Junk after checkpoint for state: " ++ stateType+ mv <- newEmptyMVar+ quickQuery' tx' $ HR () $ \_context _oldState ->+ return (Just newState, putMVar mv ())+ takeMVar mv+ atomically $ writeTVar (txLastTxContext tx') context+ in UpdateHandler (error "No cold setState handler") fn parseObject+++allStateTypes :: (Methods a, Component a) => Proxy a -> [TypeString]+allStateTypes prox = let (componentTree, _versions, _ioActions) = collectHandlers prox+ in M.keys componentTree++componentVersions :: (Methods a, Component a) => Proxy a -> M.Map String [L.ByteString]+componentVersions prox = let (_componentTree, versions, _ioActions) = collectHandlers prox+ in versions++componentIO :: (Methods a, Component a) => Proxy a -> [IO ()]+componentIO prox = let (_componentTree, _versions, ioActions) = collectHandlers prox+ in ioActions++createNewTxRun :: IO (TxRun st)+createNewTxRun =+ atomically $+ do processQueue <- newTChan+ lastContext <- newTVar (TxContext 0 0 0 (mkStdGen 42))+ return $ TxRun processQueue lastContext++setNewStateType :: String -> String+setNewStateType str = "SetNewState: " ++ str++getStateType :: String -> String+getStateType str = "GetState: " ++ str++setNewState :: TypeString -> L.ByteString -> IO ()+setNewState stateType state+ = emitEvent' (setNewStateType stateType) (SetCheckpointState state)++getState :: TypeString -> IO L.ByteString+getState stateType+ = emitEvent' (getStateType stateType) GetCheckpointState++data SetNewState st = SetNewState L.ByteString deriving (Typeable)+data GetState st = GetState deriving (Typeable)++instance Version (SetNewState st)+instance Typeable st => Serialize (SetNewState st) where+ putCopy (SetNewState lbs) = contain $ Binary.put lbs+ getCopy = contain $ liftM SetNewState Binary.get+instance Version (GetState st)+instance Typeable st => Serialize (GetState st) where+ putCopy GetState = contain $ return ()+ getCopy = contain $ return GetState++instance Typeable st => UpdateEvent (SetNewState st) ()++instance Typeable st => QueryEvent (GetState st) L.ByteString++++++-- | Schedule an update and wait for it to complete. When this function returns, you're+-- guaranteed the update will be persistent.+update :: (MonadIO m, UpdateEvent ev res) => ev -> m res+update = liftIO . emitEvent++-- | Emit a state query and wait for the result.+query :: (MonadIO m, QueryEvent ev res) => ev -> m res+query = liftIO . emitEvent+++-- Execute a query immediately without giving it a unique timestamp & transaction ID.+quickQuery' :: (Serialize st) => TxRun st -> HR st -> IO ()+quickQuery' txrun (HR ev fun)+ = do now <- getEpochMilli+ atomically $+ do tx <- readTVar (txLastTxContext txrun)+ writeTChan (txProcessQueue txrun) $ IHR tx{txTime=now} ev fun+++type Runner ev res = IO (IO ev, res -> IO ())+type EH i o = i -> IO o++data Event = forall ev. Serialize ev => Event ev++data IHR st = forall ev. (Serialize ev)+ => IHR TxContext+ ev+ (RunHandler st ev)+data HR st = forall ev. (Serialize ev)+ => HR ev+ (RunHandler st ev)+type RunHandler st ev = TxContext -> st -> IO (Maybe st, IO ())+data Res a = Ok a | Error ExceptionT+type EventQueue st = TChan (HR st) -- Queue of local event not yet given a TxContext.+type ProcessQueue st = TChan (IHR st) -- Queue of events to be processed. TxContext'es have been asigned at this point.+data TxRun st = TxRun {txProcessQueue :: !(ProcessQueue st)+ ,txLastTxContext :: !(TVar TxContext)}+++type EvLoaders' st = M.Map String (ProcessQueue st -> L.ByteString -> IO (TxId,L.ByteString))+type EvLoaders = M.Map String (L.ByteString -> IO (TxId,L.ByteString))++setEvLoadersQueue :: ProcessQueue st -> EvLoaders' st -> EvLoaders+setEvLoadersQueue queue = M.map (\fn -> fn queue)++-- serialized object -> serialized response+runObjectEvent :: Object -> IO Object+runObjectEvent obj+ = do EmitInternal eventMap <- readIORef emitRef+ runObjectEventFunc obj eventMap++runObjectEventFunc :: Object -> EventMap -> IO Object+runObjectEventFunc obj eventMap+ = do handler <- lookupEventHandler (objectType obj) eventMap+ case handler of+ -- FIXME: This will have to be filled in when doing sharding.+ QueryHandler{} -> error $ "Cold event was a query: " ++ objectType obj+ UpdateHandler _runCold runHot parse+ -> do res <- runHot (parse obj)+ return $ mkObject res++runColdEvent :: TxContext -> Object -> IO ()+runColdEvent cxt obj+ = do EmitInternal eventMap <- readIORef emitRef+ runColdEventFunc cxt obj eventMap++runColdEventFunc :: TxContext -> Object -> EventMap -> IO ()+runColdEventFunc cxt obj eventMap+ = do handler <- lookupEventHandler (objectType obj) eventMap+ case handler of+ QueryHandler{} -> error $ "Cold event was a query: " ++ objectType obj+ UpdateHandler runCold _runHotObj parse+ -> do runCold cxt (parse obj)+ return ()++lookupEventHandler :: TypeString -> EventMap -> IO EventHandler+lookupEventHandler eventType eventMap+ = case M.lookup eventType eventMap of+ Nothing -> error $ "Couldn't find handler for event of type: " ++ eventType+ Just handler -> return handler+ ++eventTString :: Serialize ev => ev -> TypeString+eventTString ev = show (typeOf ev)++++handleEvent :: (st -> Env -> Ev m res -> STM intermediate) -> (st -> intermediate -> IO (Maybe st, res))+ -> (Either ExceptionT res -> IO ()) -> Ev m res -> RunHandler st ev+handleEvent runner stateCheck ofun action tx st+ = handle eh $+ do intermediate <- atomically $ runQuery'+ (newState, res) <- stateCheck st intermediate+ return (newState, ofun (Right res))+ where runQuery' = do rs <- newTVar (txStdGen tx)+ let env = Env { evContext = tx, evRandoms = rs }+ intermediate <- runner st env action+ return $ intermediate+ eh e = do logMT ERROR ("handleEvent FAIL: "++ show e)+ return (Nothing,ofun (Left e))++handleQuery :: (Either ExceptionT res -> IO ()) -> Query st res -> RunHandler st ev+handleQuery = handleEvent (\st env (Ev cmd) -> runReaderT (cmd env) st) (\_st res -> return (Nothing, res))++handleUpdate :: (Either ExceptionT res -> IO ()) -> Update st res -> RunHandler st ev+handleUpdate = handleEvent (\st env (Ev cmd) -> runStateT (cmd env) st) (\st (res,st') -> checkDiff st st' >>= \diff -> return (diff, res))++{- Some updates might not modify the state.+ Doing a pointer-check might be worth it.+ (as a side note, reallyUnsafePtrEquality# is orders of magnitude faster+ than comparing StableNames.)+-}+checkDiff :: a -> a -> IO (Maybe a)+checkDiff _old new+ = return (Just new)++getEpochMilli :: IO EpochMilli+getEpochMilli =+ do TOD sec pico <- getClockTime+ return $ fromIntegral $ sec * 1000 + pico `div` 10^9++newTxContext :: IO TxContext+newTxContext = do+ milli <- getEpochMilli+ let txid = -1 -- Not set yet.+ sgen <- modifyMVar globalRandomGen (return . split)+ let (rand, sgen') = random sgen+ return $ TxContext txid rand milli sgen'++addTxId :: TxRun st -> TxContext -> STM TxContext+addTxId tx context+ = do lastContext <- readTVar (txLastTxContext tx)+ let new = context{txId = txId lastContext + 1}+ writeTVar (txLastTxContext tx) new+ return new{txId = txId new + 1}++{-# NOINLINE globalRandomGen #-}+-- XXX: why are we using a global StdGen? Isn't there already one in System.Random?+globalRandomGen :: MVar StdGen+globalRandomGen = unsafePerformIO (newMVar =<< getStdGen)+++data TxConfig = TxConfig+ { txcCheckpointSeconds :: Seconds, -- ^ Perform checkpoint at least every N seconds.+ txcOperationMode :: OperationMode,+ txcClusterSize :: Int, -- ^ Number of active nodes in the cluster (not counting this node).+ txcClusterPort :: Int, --+ txcCommitFrequency :: Int -- ^ Commits per second. Only applies to cluster mode.+ }++data TxControl = TxControl+ { ctlSaver :: Saver -- ^ Saver given by the user.+ , ctlEventSaver :: MVar (WriterStream EventLogEntry)+ , ctlAllComponents :: [String] -- ^ Types of each component used.+ , ctlComponentVersions :: M.Map String [L.ByteString] -- ^ Map listing all versions of a component+ , ctlChildren :: [(ThreadId, MVar ())] -- + }++data EventLogEntry = EventLogEntry TxContext Object deriving (Typeable, Show)+instance Version EventLogEntry+instance Serialize EventLogEntry where+ putCopy (EventLogEntry context obj) = contain $ safePut (context,obj)+ getCopy = contain $ + do (context, obj) <- safeGet+ return $ EventLogEntry context obj++data OperationMode+ = SingleMode+ | ClusterMode String++nullTxConfig :: TxConfig+nullTxConfig = TxConfig { txcCheckpointSeconds = 60*60*24,+ txcOperationMode = SingleMode,+ txcClusterSize = 0,+ txcClusterPort = 8500,+ txcCommitFrequency = 50+ }++runTxLoop :: MVar (WriterStream EventLogEntry) -> ProcessQueue st -> st -> IO ()+runTxLoop eventSaverVar queue st0 =+ let loop st = do+ IHR context ev fun <- atomically $ readTChan queue+ pushEventItem context ev -- Notify the user that this event is about to be executed.+ -- FIXME: What if the user wants to know when the event has+ -- finished executing?+ let tstring = eventTString ev+ logMT NOTICE $ ("> Event "++show (txId context)++" of "++tstring)+ (mst,ra) <- fun context st+ case mst of+ -- State was not updated.+ --+ -- Thus the response can be executed immediately.+ Nothing -> do forkIO $ logMT NOTICE "> pure" >> ra+ loop st + -- There is a new State.+ --+ -- Note that saverAdd can return without yet writing the result+ -- as long as:+ -- 1) saverAdd calls honor the sequence in which they were made.+ -- 2) saverAdd calls execute the finalizers only after the value+ -- has been serialized. The finalizers typically return the+ -- result to the user so they should not be kept+ -- waiting too long.+ -- 3) This means that checkpoints need to flush the saver+ -- which will guarantee that all pending result/side-effects+ -- have been processed.+ -- 4) Savers must *not* block while running the finalizers+ Just st' -> do eventSaver <- readMVar eventSaverVar+ writerAdd eventSaver (EventLogEntry context (mkObject ev)) (logMT NOTICE "> disk " >> ra)+ loop st'+ in do forkIO $ (loop st0) `catch` (\ThreadKilled -> return ())+ `catch` (\BlockedIndefinitely -> return ())+ return ()+
+ src/HAppS/State/TxControl.hs view
@@ -0,0 +1,59 @@+module HAppS.State.TxControl+ ( runTxSystem+ , runTxSystem'+ , shutdownSystem+ ) where++import System.Log.Logger+import System.IO+import Control.Monad+import Control.Exception+import Control.Concurrent++import HAppS.State.Checkpoint+import HAppS.State.Saver+import HAppS.State.Transaction+import HAppS.State.Spread+import HAppS.State.ComponentSystem+import HAppS.Data.Proxy++logMM :: Priority -> String -> IO ()+logMM = logM "HAppS.State.TxControl"+++-- | Run a transaction system +runTxSystem :: (Methods st, Component st) => Saver -> Proxy st -> IO (MVar TxControl)+runTxSystem = runTxSystem' False++runTxSystem' :: (Methods st, Component st) => Bool -> Saver -> Proxy st -> IO (MVar TxControl)+runTxSystem' withMultimaster saver stateProxy =+ do logMM NOTICE "Initializing system control."+ ctl <- createTxControl saver stateProxy+ logMM NOTICE "Creating event mapper."+ localEventMap <- createEventMap ctl stateProxy+ setNewEventMap localEventMap+ logMM NOTICE "Restoring state."+ enableLogging <- restoreState ctl+ when (withMultimaster)+ $ do logMM NOTICE "Multimaster mode"+ cluster <- connectToCluster+ eventMap <- changeEventMapping ctl localEventMap cluster+ setNewEventMap eventMap+ enableLogging+ let ioActions = componentIO stateProxy+ logMM NOTICE "Forking children."+ children <- forM ioActions $ \action -> do mv <- newEmptyMVar+ tid <- forkIO (action `finally` putMVar mv ())+ return (tid,mv)+ modifyMVar_ ctl $ \c -> return c{ctlChildren = children}+ return ctl++shutdownSystem :: MVar TxControl -> IO ()+shutdownSystem ctl+ = do logMM NOTICE "Shutting down."+ children <- liftM ctlChildren $ readMVar ctl+ logMM NOTICE "Killing children."+ mapM_ (killThread . fst) children+ mapM_ (takeMVar . snd) children -- FIXME: Use a timeout.+ logMM NOTICE "Shutdown complete"+ closeTxControl ctl
+ src/HAppS/State/Types.hs view
@@ -0,0 +1,100 @@+{-# LANGUAGE UndecidableInstances, TemplateHaskell,+ ExistentialQuantification, RankNTypes,+ FlexibleInstances, DeriveDataTypeable,+ ScopedTypeVariables, PatternSignatures+ #-}+++module HAppS.State.Types where++import Control.Concurrent.STM+import Data.Int+import Data.Word+import qualified GHC.Conc(unsafeIOToSTM)+import System.Random -- (StdGen)++import Control.Monad.State+import Control.Monad.Reader+import Control.Monad.Trans+import Data.Generics+-- Monad things+++data Env = Env+ { evRandoms :: TVar StdGen+ , evContext :: TxContext }++type TxId = Int64+type EpochMilli= Int64++instance Typeable StdGen where typeOf _ = mkTyConApp (mkTyCon "System.Random.StdGen") []++instance Random Word64 where+ randomR = integralRandomR+ random = randomR (minBound,maxBound)++instance Random Int64 where+ randomR = integralRandomR+ random = randomR (minBound,maxBound)++integralRandomR :: (Integral a, RandomGen g) => (a,a) -> g -> (a,g)+integralRandomR (a,b) g = case randomR (fromIntegral a :: Integer,+ fromIntegral b :: Integer) g of+ (x,g') -> (fromIntegral x, g')+++data TxContext = TxContext+ { txId :: TxId,+ txRand :: Word64,+ txTime :: EpochMilli,+ txStdGen :: StdGen+ } deriving (Read,Show,Typeable)+++{-+ Is STM really be best backend monad?+ We don't use any of the STM features.+-}+-- | ACID computations that work with any state and event types.+type AnyEv a = forall t. (Monad (t STM), MonadTrans t) => Ev (t STM) a++-- | Monad for ACID event handlers.+newtype Ev m t = Ev { unEv :: Env -> m t }++instance (Typeable st, Typeable1 m) => Typeable1 (ReaderT st m) where+ typeOf1 x = mkTyConApp (mkTyCon "Control.Monad.Reader.ReaderT") [typeOf (undefined :: st), typeOf1 (m x)]+ where m :: ReaderT st m a -> m a+ m = undefined++instance (Typeable st, Typeable1 m) => Typeable1 (StateT st m) where+ typeOf1 x = mkTyConApp (mkTyCon "Control.Monad.State.StateT") [typeOf (undefined :: st), typeOf1 (m x)]+ where m :: StateT st m a -> m a+ m = undefined++instance (Typeable state, Typeable t) => Typeable (Ev (ReaderT state STM) t) where+ typeOf (Ev _cmd) = mkTyConApp (mkTyCon "HAppS.State.Types.Ev") [typeOf (u::ReaderT state STM t)]+ where u = undefined+instance (Typeable state, Typeable t) => Typeable (Ev (StateT state STM) t) where+ typeOf (Ev _cmd) = mkTyConApp (mkTyCon "HAppS.State.Types.Ev") [typeOf (u::StateT state STM t)]+ where u = undefined++type Query state = Ev (ReaderT state STM)+type Update state = Ev (StateT state STM)++-- unsafe lifting++unsafeIOToEv :: IO a -> AnyEv a+unsafeIOToEv c = unsafeSTMToEv (unsafeIOToSTM c)+unsafeSTMToEv :: STM a -> AnyEv a+unsafeSTMToEv c = Ev $ \_ -> lift c+unsafeIOToSTM :: IO a -> STM a+unsafeIOToSTM = GHC.Conc.unsafeIOToSTM++++-- Misc++newtype Shadow t a = Shadow { unShadow :: a } deriving Typeable++newtype UsingXml a = UsingXml { unXml :: a } deriving Typeable+
+ src/HAppS/State/Util.hs view
@@ -0,0 +1,100 @@+{-# LANGUAGE TemplateHaskell, CPP #-}+module HAppS.State.Util+ ( -- * Random numbers+ getRandom, getRandomR,+ -- * TH helpers+ inferRecordUpdaters+ ) where++import Control.Concurrent.STM+import Control.Monad.State+import System.Random++import HAppS.State.Monad+import HAppS.State.Types++import Data.Char(toUpper)+import Language.Haskell.TH+++-- Random numbers++-- | Get a random number.+getRandom :: Random a => AnyEv a+getRandom = do r <- sel evRandoms+ g <- liftSTM $ readTVar r+ let (x,g') = random g+ liftSTM $ writeTVar r g'+ return x++-- | Get a random number inside the range.+getRandomR :: Random a => (a,a) -> AnyEv a+getRandomR z = do r <- sel evRandoms+ g <- liftSTM $ readTVar r+ let (x,g') = randomR z g+ liftSTM $ writeTVar r g'+ return x++--------------------------------------------------------------+-- inferRecordUpdater+--------------------------------------------------------------+++-- FIXME: Throw a decent error message if the input isn't a record.+-- | Infer updating functions for a record @a_foo :: component -> record -> record@ and+-- @withFoo = localState foo a_foo@.+inferRecordUpdaters :: Name -> Q [Dec]+inferRecordUpdaters typeName = do+ con <- decToSimpleRecord =<< nameToDec typeName+ let c name upd s = + do let un = mkName ("a_"++ns)+ wn = mkName ("with"++(toUpper (head ns):tail ns))+ ns = nameBase name+ ud <- un `sdef` upd+ wd <- wn `sdef` (varE 'localState `appE` s `appE` varE un)+ return [ud, wd]+ xs <- sequence $ zipWith3 c (fieldNames con) (updFuns con) (selFuns con)+ return $ concat xs+++-- Utilities++decToSimpleRecord :: Dec -> Q Con+decToSimpleRecord (DataD _ _ _ [con] _) = return con+decToSimpleRecord (DataD _ n _ _ _) =+ fail ("Not a simple record (has multiple constructors): "++show n)+decToSimpleRecord (NewtypeD _ _ _ con _) = return con+decToSimpleRecord x = fail ("Wanted a simple record, got: "++show x)++nameToDec :: Name -> Q Dec+nameToDec ty = reify ty >>= un+ where un (TyConI d) = return $ d+ un _ = fail "nameToDec: expected TyCon"++-- | Create a list of selection functions for a record.+selFuns :: Con -> [ExpQ]+selFuns (RecC _ ts) = [ varE n | (n,_,_) <- ts ]+selFuns _ = error "Constructors other than RecC not handled in selFuns"++-- | Create a list of update functions for a record.+updFuns :: Con -> [ExpQ]+updFuns (RecC _ ts) = [ upd n | (n,_,_) <- ts ]+ where [x,y] = map mkName ["x","y"]+ upd f = lamE [varP x, varP y] $ rup f+ rup f = recUpdE (varE y) [return (f,VarE x)]+updFuns _ = error "Constructors other than RecC not handled in updFuns"++-- | Return field names+fieldNames :: Con -> [Name]+fieldNames (RecC _ ts) = [ n | (n,_,_) <- ts ]+fieldNames _ = error "Constructors other than RecC not handled in fieldNames"++-- | Simple definition+sdef :: Name -> ExpQ -> DecQ+sdef vn ve = valD (varP vn) (normalB ve) []+++--------------------------------------------------------------+-- inferRecordUpdater end+--------------------------------------------------------------+
+ tests/HAppS/State/Tests.hs view
@@ -0,0 +1,24 @@+module HAppS.State.Tests (allTests) where++import HAppS.State.Tests.SaverProperties (saverProperties)+import HAppS.State.Tests.CheckpointProperties (checkpointProperties, congestedCheckpoint, runRestoreCongestionKnownFailures)+import Test.HUnit++allTests :: Test+allTests + = "happstack-state" ~: + [ runRestoreCongestionKnownFailures+ , checkpointProperties+ , congestedCheckpoint+ , saverProperties+ ]+++{- List of properties++ * for all operations, memory saver == file saver+ * saverCut is only an optimization+ * checkpoint is only an optimization+ * events that predate the last checkpoint are ignored++-}
+ tests/HAppS/State/Tests/CheckpointProperties.hs view
@@ -0,0 +1,236 @@+{-# LANGUAGE DeriveDataTypeable, FlexibleContexts, FlexibleInstances, MultiParamTypeClasses, TemplateHaskell, TypeFamilies #-}+module HAppS.State.Tests.CheckpointProperties+ ( testCongestedCheckpoint+ , congestedCheckpoint+ , checkCheckpointProperties+ , checkpointProperties+ , runRestoreCongestionKnownFailures+ ) where++import HAppS.State+import HAppS.State.ComponentTH+import HAppS.State.TxControl+import HAppS.State.Tests.Helpers+import HAppS.Util.Testing (qccheck, qctest)++import Control.Monad+import Control.Monad.State (get,put)+import Control.Monad.Reader (ask)+import Control.Exception+import Control.Concurrent+import Data.Typeable+import Text.Printf++import Test.QuickCheck+import Test.QuickCheck.Batch+import Test.HUnit (Test(..),(~:), assertBool, assertFailure)++--------------------------------------------------------------+-- Checkpoint congestion+--------------------------------------------------------------++data RestoreTest = RestoreTest Int deriving (Typeable)+instance Version RestoreTest+$(deriveSerialize ''RestoreTest)++data Block = Block (Maybe (MVar ())) deriving (Typeable)++-- We don't want to block events that are loaded from disk.+instance Version Block+instance Serialize Block where+ putCopy _ = contain $ return ()+ getCopy = contain $ return $ Block Nothing++succValue :: Block -> Update RestoreTest ()+succValue block+ = do RestoreTest n <- get+ put $ RestoreTest (n+1)+ case block of+ Block (Just mv) -> unsafeIOToEv $ takeMVar mv+ _ -> return ()++getValue :: Query RestoreTest Int+getValue = do RestoreTest n <- ask+ return n++$(mkMethods ''RestoreTest [ 'succValue+ , 'getValue])++instance Component RestoreTest where+ type Dependencies RestoreTest = End+ initialValue = RestoreTest 0++restoreTestEntryPoint :: Proxy RestoreTest+restoreTestEntryPoint = Proxy++{-+ There is a small window after the saver has been cut+ and the checkpoints have been queried. Events that are+ executed in this windows must be ignored when the state+ is restored. That is, events with a transaction id lower+ than the saved checkpoint should not be replayed.+ In applications, this is bound to happen when the processing+ queue is congested (non-empty).++ We will trigger this occasion via a blocking event. The event+ will enter the processing queue and block before it has been written+ to disk. We will then schedule a checkpoint and unblock the event.++ The event will be written in the same file as the checkpoint. If+ the event isn't discarded when the state is restored, the value+ in the state will be '2' instead of '1'.+-}+testCongestedCheckpoint+ = withMemorySaver $ \saver ->+ do initCongestedCheckpoint (Queue saver)+ checkRestoredValue (Queue saver)++checkRestoredValue saver+ = bracket (runTxSystem saver restoreTestEntryPoint) (shutdownSystem) $ \ctl ->+ do val <- query $ GetValue+ when (val /= 1) $ error $ "testCongestedCheckpoint failed: " ++ show val+ printf "%25s : OK\n" "Checkpoint congestion"++congestedCheckpoint :: Test+congestedCheckpoint = TestCase $+ withMemorySaver $ \saver ->+ do initCongestedCheckpoint (Queue saver)+ bracket (runTxSystem saver restoreTestEntryPoint) (shutdownSystem) $ \ctl ->+ do val <- query $ GetValue+ when (val /= 1) $ assertFailure $ "testCongestedCheckpoint failed: " ++ show val++initCongestedCheckpoint saver+ = bracket (runTxSystem saver restoreTestEntryPoint) (shutdownSystem) $ \ctl ->+ do mv <- newEmptyMVar+ forkIO $ do update $ SuccValue (Block (Just mv))+ yield -- Relying on yields for correctness isn't a great idea.+ forkIO $ createCheckpoint ctl+ yield -- However, it's OK for now since tests aren't vital.+ putMVar mv ()++--------------------------------------------------------------+-- Run/Restore+--------------------------------------------------------------++data RunRestore = RunRestore [Int] deriving (Typeable,Show)+instance Version RunRestore+$(deriveSerialize ''RunRestore)++push :: Int -> Update RunRestore ()+push val = do RunRestore ls <- get+ put $ RunRestore (val:ls)++peek :: Query RunRestore [Int]+peek = do RunRestore ls <- ask+ return $ reverse ls++$(mkMethods ''RunRestore [ 'push+ , 'peek])++instance Component RunRestore where+ type Dependencies RunRestore = End+ initialValue = RunRestore []++runRestoreEntryPoint :: Proxy RunRestore+runRestoreEntryPoint = Proxy++genRunRestoreProp withSaver check action+ = ioProperty $ withSaver $ \saver ->+ do bracket (runTxSystem saver runRestoreEntryPoint) (shutdownSystem) $ \ctl ->+ do action ctl+ bracket (runTxSystem saver runRestoreEntryPoint) (shutdownSystem) $ \ctl ->+ do ls <- query Peek+ return (check ls)++-- State changes without a checkpoint should be lost.+prop_runRestoreId withSaver values+ = genRunRestoreProp withSaver (==values) $ \ctl ->+ forM_ values (update . Push)++-- Checkpoints can be placed arbitrarily.+prop_runRestoreCheckpoint withSaver before after+ = genRunRestoreProp withSaver (== (before++after)) $ \ctl ->+ do forM_ before (update . Push)+ createCheckpoint ctl+ forM_ after (update . Push)++-- Multiple checkpoints can be placed arbitrarily.+prop_runRestoreMultipleCheckpoint withSaver allData+ = genRunRestoreProp withSaver (== concat allData) $ \ctl ->+ do forM_ allData $ \dataList -> do createCheckpoint ctl+ forM_ dataList (update . Push)++-- 'update'/'update' and 'checkCheckpoint' are thread-safe.+prop_runRestoreAsync withSaver values+ = genRunRestoreProp withSaver (== values) $ \ctl ->+ do mv1 <- newEmptyMVar+ mv2 <- newEmptyMVar+ forkIO $ forM_ values (update . Push) `finally` putMVar mv1 ()+ forkIO $ createCheckpoint ctl `finally` putMVar mv2 ()+ takeMVar mv1+ takeMVar mv2++-- Send all events in parallel. Better at testing congestion.+prop_runRestoreCongestion withSaver values+ = genRunRestoreProp withSaver (\ls -> sum ls == sum values) $ \ctl ->+ do sem <- newQSem 0+ mv <- newEmptyMVar+ forM_ values $ \value -> forkIO (update (Push value) `finally` signalQSem sem)+ forkIO $ createCheckpoint ctl `finally` putMVar mv ()+ takeMVar mv+ replicateM_ (length values) (waitQSem sem)++checkCheckpointProperties :: IO ()+checkCheckpointProperties+ = tryTests ("runRestore") options [run (prop_runRestoreId saver)+ ,run (prop_runRestoreCheckpoint saver)+ ,run (prop_runRestoreMultipleCheckpoint saver)+ ,run (prop_runRestoreAsync saver)+ ,run (prop_runRestoreCongestion saver)]+ where options = defOpt{length_of_tests=5}+ saver = withQueueSaver withMemorySaver++checkpointProperties :: Test+checkpointProperties + = "checkpointProperties" ~: + [ "prop_runRestoreId" ~: qccheck config (prop_runRestoreId saver)+ , "prop_runRestoreCheckpoint" ~: qccheck config (prop_runRestoreCheckpoint saver)+ , "prop_runRestoreMultipleCheckpoint" ~: qccheck config (prop_runRestoreMultipleCheckpoint saver)+ , "prop_runRestoreAsync" ~: qccheck config (prop_runRestoreAsync saver)+ , "prop_runRestoreCongestion" ~: qccheck config (prop_runRestoreCongestion saver)+ ]+ where+ config = defaultConfig { configMaxTest = 100 }+ saver = withQueueSaver withMemorySaver+++{- ++I have seen prop_runRestoreCongestion fail several times. I tried+adding the failing cases as explicit tests, but they don't seem to+fail.++ ### Failure in: happstack-state:0:checkpointProperties:4:prop_runRestoreCongestion+ Falsifiable, after 56 tests:+ [-15,8,-11,-6,-3,-7,20,20,15,-16,13,20,-17,-19,14,-2,-17,-15,-1,-17]++ ### Failure in: happstack-state:0:checkpointProperties:4:prop_runRestoreCongestion+ Falsifiable, after 59 tests:+ [6,-10,3,14,2,1,-12,-8,-14,17,7,-14,10,-11]++ ### Failure in: happstack:3:happstack-state:0:checkpointProperties:4:prop_runRestoreCongestion+ Falsifiable, after 67 tests:+ [11,22,22,10,12,-13,-8,-21,9,19,3,11,-5]++-}+runRestoreCongestionKnownFailures :: Test+runRestoreCongestionKnownFailures+ = "runRestoreCongestionKnownFailures"+ ~: map (assertBool "" . prop_runRestoreCongestion saver) known+ where+ saver = withQueueSaver withMemorySaver+ known = [ [-15,8,-11,-6,-3,-7,20,20,15,-16,13,20,-17,-19,14,-2,-17,-15,-1,-17]+ , [6,-10,3,14,2,1,-12,-8,-14,17,7,-14,10,-11]+ , [11,22,22,10,12,-13,-8,-21,9,19,3,11,-5]+ ]+
+ tests/HAppS/State/Tests/Helpers.hs view
@@ -0,0 +1,108 @@+module HAppS.State.Tests.Helpers where++import HAppS.State.Saver+import HAppS.State.Saver.Impl.Memory++import System.Directory+import System.Random+import System.FilePath+import System.Exit+import System.IO.Unsafe+import Control.Monad+import Control.Exception+import Control.Concurrent++import Test.QuickCheck+import Test.QuickCheck.Batch+import Text.Printf++instance Arbitrary Char where+ arbitrary = choose ('a','z')+ coarbitrary = undefined++instance (Arbitrary a, Num a) => Arbitrary (Abs a) where+ arbitrary = liftM (Abs . abs) arbitrary+ coarbitrary = undefined++instance (Arbitrary a) => Arbitrary (NonEmpty a) where+ arbitrary = do x <- arbitrary+ xs <- arbitrary+ return $ NonEmpty (x:xs)+ coarbitrary = undefined++data Abs a = Abs a+instance (Show a) => Show (Abs a) where+ show (Abs a) = show a++data NonEmpty a = NonEmpty [a]+instance (Show a) => Show (NonEmpty a) where+ show (NonEmpty a) = show a++ioProperty = unsafePerformIO++withTemporaryDirectory :: (FilePath -> IO a) -> IO a+withTemporaryDirectory action+ = do tmp <- getTemporaryDirectory+ n <- randomIO+ let dir = tmp </> (show (abs n :: Int))+ exist <- doesDirectoryExist dir+ if exist+ then withTemporaryDirectory action+ else bracket_ (createDirectoryIfMissing False dir)+ (removeDirectoryRecursive dir)+ (action dir)++withMemorySaver action = action =<< liftM Memory newMemoryStore++withFileSaver action = withTemporaryDirectory $ \dir -> action (FileSaver dir)++withQueueSaver h action = h (action . Queue)++forEachSaverM :: (Monad m) => (String -> ((Saver -> IO a) -> IO a) -> m b) -> m [b]+forEachSaverM action+ = forM savers $ \(name, withSaver) ->+ action name withSaver+ where savers = [ ("Memory", withMemorySaver)+ , ("Queue Memory", \action -> withMemorySaver (action . Queue)) + , ("File", withFileSaver)+ , ("Queue File", \action -> withFileSaver (action . Queue))+ ]++forEachSaver :: (String -> ((Saver -> IO a) -> IO a) -> b) -> [b]+forEachSaver action+ = flip map savers $ \(name, withSaver) ->+ action name withSaver+ where savers = [ ("Memory", withMemorySaver)+ , ("Queue Memory", \action -> withMemorySaver (action . Queue)) + , ("File", withFileSaver)+ , ("Queue File", \action -> withFileSaver (action . Queue))+ ]++++forEachSaver_ action+ = forM_ savers $ \(name, withSaver) ->+ action name withSaver+ where savers = [ ("Memory", withMemorySaver)+ , ("Queue Memory", \action -> withMemorySaver (action . Queue)) + , ("File", withFileSaver)+ , ("Queue File", \action -> withFileSaver (action . Queue))+ ]++tryTests :: String -> TestOptions -> [TestOptions -> IO TestResult] -> IO ()+tryTests name opt tests+ = do printf "%25s : " name+ forM_ tests $ \test ->+ do res <- test opt+ case res of+ TestOk _ _ _ -> putStr "."+ TestExausted _ _ _ -> putStr "?"+ TestFailed err num -> do putStr "\n"+ printf " ** test %d of %s failed with the binding(s)\n" num name+ sequence_ [ putStrLn (" ** " ++ v) | v <- err ]+ exitWith (ExitFailure 1)+ TestAborted exp -> do putStr "\n"+ printf " ** test %s failed with the exception\n" name+ putStrLn (" ** " ++ show exp)+ exitWith (ExitFailure 1)+ putChar '\n'
+ tests/HAppS/State/Tests/SaverProperties.hs view
@@ -0,0 +1,94 @@+module HAppS.State.Tests.SaverProperties+ ( checkSaverProperties+ , saverProperties+ ) where++import HAppS.State.Saver+import HAppS.State.Saver.Impl.Memory+import HAppS.State.Tests.Helpers+import HAppS.Util.Testing (qctest, qcrun)++import System.Directory+import System.Random+import System.FilePath+import System.IO.Unsafe+import Control.Monad+import Control.Exception+import Control.Concurrent+import Data.List++import Test.QuickCheck+import Test.QuickCheck.Batch+import Test.HUnit (Test(TestCase),(~:), assertFailure)++--------------------------------------------------------------+-- Tests+--------------------------------------------------------------++-- read (write values) == values+prop_getSetId withSaver key cutoff values+ = not (null key) ==>+ cutoff > 0 ==>+ ioProperty $ withSaver $ \saver ->+ do writer <- createWriter saver key cutoff+ forM_ values $ \value -> writerAdd writer (value::Int) $ return ()+ writerClose writer++ reader <- createReader saver key cutoff+ out <- readerGetUncut reader+ return $ values == out++prop_seqReadWrite withSaver (NonEmpty key) (Abs cutoff) valss+ = ioProperty $ withSaver $ \saver ->+ do writer <- createWriter saver key cutoff+ forM_ valss $ \vals -> do forM_ vals $ \val -> writerAdd writer (val::Int) $ return ()+ writerCut writer+ writerClose writer++ reader <- createReader saver key cutoff+ (out,newCut) <- readerGet reader+ return $ out == concat valss && newCut == length valss + 1++prop_cutDrop withSaver (NonEmpty key) (Abs cutoff) a b+ = ioProperty $ withSaver $ \saver ->+ do writer <- createWriter saver key cutoff+ forM_ a $ \val -> writerAdd writer (val::Int) $ return ()+ newCut <- writerCut writer+ forM_ b $ \val -> writerAdd writer (val::Int) $ return ()+ writerClose writer++ reader <- createReader saver key newCut+ (out,_) <- readerGet reader+ return $ b `isSuffixOf` out++prop_atomic withSaver (NonEmpty key) (Abs cutoff) value+ = ioProperty $ withSaver $ \saver ->+ do writer <- createWriter saver key cutoff+ writerAtomicReplace writer (value::Int)+ writerClose writer++ reader <- createReader saver key cutoff+ out <- readerGetUncut reader+ readerClose reader+ return $ out == [value]++checkSaverProperties :: IO ()+checkSaverProperties+ = forEachSaver_ $ \name withSaver ->+ tryTests (name ++ " saver") options [run (prop_getSetId withSaver)+ ,run (prop_seqReadWrite withSaver)+ ,run (prop_cutDrop withSaver)+ ,run (prop_atomic withSaver)]+ where options = defOpt{length_of_tests=5}++saverProperties :: Test+saverProperties + = "saverProperties" ~:+ ((forEachSaver $ \name withSaver ->+ name ~:+ [ "prop_getSetId" ~: qcrun (prop_getSetId withSaver) options+ , "prop_seqReadWrite" ~: qcrun (prop_seqReadWrite withSaver) options+ , "prop_cutDrop" ~: qcrun (prop_cutDrop withSaver) options+ , "prop_atomic" ~: qcrun (prop_atomic withSaver) options+ ]))+ where options = defOpt{length_of_tests=5}
+ tests/Test.hs view
@@ -0,0 +1,19 @@+module Main where++import HAppS.State.Tests (allTests)+import Test.HUnit (errors, failures, putTextToShowS,runTestText, runTestTT)+import System.Exit (exitFailure)+import System.IO (hIsTerminalDevice, stdout)++-- |A simple driver for running the local test suite.+main :: IO ()+main =+ do c <- do istty <- hIsTerminalDevice stdout+ if istty+ then runTestTT allTests+ else do (c,st) <- runTestText putTextToShowS allTests+ putStrLn (st "")+ return c+ case (failures c) + (errors c) of+ 0 -> return ()+ n -> exitFailure