HAppS-State 0.9.2.1 → 0.9.3
raw patch · 12 files changed
+480/−144 lines, 12 filesdep +hspreaddep +sybdep ~HAppS-Datadep ~HAppS-Utildep ~base
Dependencies added: hspread, syb
Dependency ranges changed: HAppS-Data, HAppS-Util, base
Files
- COPYING +29/−0
- HAppS-State.cabal +45/−28
- src/HAppS/State.hs +1/−1
- src/HAppS/State/Checkpoint.hs +22/−13
- src/HAppS/State/ComponentSystem.hs +15/−9
- src/HAppS/State/Control.hs +6/−0
- src/HAppS/State/Monad.hs +10/−2
- src/HAppS/State/Saver/Impl/File.hs +13/−2
- src/HAppS/State/Spread.hs +161/−0
- src/HAppS/State/Transaction.hs +161/−85
- src/HAppS/State/TxControl.hs +15/−3
- src/HAppS/State/Types.hs +2/−1
+ 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.
HAppS-State.cabal view
@@ -1,38 +1,55 @@ Name: HAppS-State-Version: 0.9.2.1+Version: 0.9.3 Synopsis: Event-based distributed state. Description: Web framework License: BSD3-Copyright: 2007 HAppS LLC+License-file: COPYING+Copyright: 2007-2008 HAppS LLC Author: HAppS LLC Maintainer: AlexJacobson@HAppS.org Category: Web, Distributed Computing-Build-Depends: base, HaXml >= 1.13 && < 1.14, mtl, network,- stm, template-haskell, hslogger >= 1.0.2, HAppS-Util>=0.9.2,- HAppS-Data>=0.9.2, bytestring, containers, random, old-time,- old-locale, unix, directory, binary, filepath+Cabal-Version: >= 1.2 Build-Type: Simple-hs-source-dirs: src-Exposed-modules: - HAppS.State++flag base4++Library++ Exposed-modules: + HAppS.State -- HAppS.State.Logger- HAppS.State.Saver- HAppS.State.Control- HAppS.State.Transaction- HAppS.State.ComponentTH- HAppS.State.ComponentSystem-Other-modules: - HAppS.State.Checkpoint- HAppS.State.Monad- HAppS.State.Saver.Impl.File- HAppS.State.Saver.Impl.Memory- HAppS.State.Saver.Impl.Queue- HAppS.State.Saver.Types+ HAppS.State.Saver+ HAppS.State.Control+ HAppS.State.Transaction+ HAppS.State.ComponentTH+ HAppS.State.ComponentSystem+ 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-Extensions: CPP-cpp-options: -DUNIX-ghc-options: -W -fno-warn-incomplete-patterns-GHC-Prof-Options: -auto-all+ HAppS.State.Types+ HAppS.State.Util+ HAppS.State.TxControl+ Build-Depends: HaXml >= 1.13 && < 1.14, mtl, network,+ stm, template-haskell, hslogger >= 1.0.2, HAppS-Util>=0.9.3,+ HAppS-Data>=0.9.3, bytestring, containers, random, old-time,+ old-locale, directory, binary, filepath, hspread >=0.3+ if flag(base4)+ Build-Depends: base >= 4 && < 5, syb+ else+ Build-Depends: base < 4++ hs-source-dirs: src+ + if !os(windows)+ Build-Depends: unix+ cpp-options: -DUNIX++ Extensions: CPP + ghc-options: -W -fno-warn-incomplete-patterns+ GHC-Prof-Options: -auto-all
src/HAppS/State.hs view
@@ -6,7 +6,7 @@ TxId, EpochMilli, TxConfig(..),nullTxConfig,Saver(..), -- * Misc utilities module HAppS.State.Monad,- getEventId, getTime, getEventClockTime,+ getEventId, getTime, getEventClockTime, getEventStream, EventItem(..), module HAppS.State.Util, -- * Serialization module HAppS.Data.Serialize,
src/HAppS/State/Checkpoint.hs view
@@ -1,4 +1,5 @@-{-# OPTIONS -fglasgow-exts -cpp -fth #-}+{-# OPTIONS -fglasgow-exts #-}+{-# LANGUAGE CPP, TemplateHaskell #-} module HAppS.State.Checkpoint ( createTxControl , closeTxControl@@ -54,10 +55,11 @@ = do -- The state hasn't been loaded yet. Ignore events. eventSaverVar <- newMVar =<< createWriter NullSaver "events" 0 newMVar $ TxControl- { ctlSaver = saver- , ctlEventSaver = eventSaverVar- , ctlAllComponents = allStateTypes proxy- , ctlChildren = [] }+ { ctlSaver = saver+ , ctlEventSaver = eventSaverVar+ , ctlAllComponents = allStateTypes proxy+ , ctlComponentVersions = componentVersions proxy+ , ctlChildren = [] } closeTxControl :: MVar TxControl -> IO () closeTxControl ctlVar@@ -67,8 +69,9 @@ -- 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.-restoreState :: MVar TxControl -> IO ()+-- 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.@@ -77,15 +80,16 @@ Nothing -> do writeState ctl (State 0 0) -- No events to replay. Switch to real saver.- swapMVar (ctlEventSaver ctl) =<< createWriter (ctlSaver ctl) "events" 0+ 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.- swapMVar (ctlEventSaver ctl) =<< createWriter (ctlSaver ctl) "events" (cutoff+offset)- return ()+ return $ do swapMVar (ctlEventSaver ctl) =<< createWriter (ctlSaver ctl) "events" (cutoff+offset)+ return () -- Load state from disk. loadState :: TxControl -> Int -> IO ()@@ -93,9 +97,14 @@ = do logMC NOTICE $ "Loading components from storage." checkpoints <- withReader ctl "checkpoints" cutoff $ loadCheckpoints forM_ (ctlAllComponents ctl) $ \stateType- -> case M.lookup stateType checkpoints of- Just state -> setNewState stateType state- Nothing -> return ()+ -> 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.
src/HAppS/State/ComponentSystem.hs view
@@ -10,6 +10,7 @@ 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@@ -42,6 +43,7 @@ instance Show (Method st) where show method = "Method: " ++ methodType method + methodType m = case m of Update fn -> let ev :: (ev -> Update st res) -> ev ev _ = undefined@@ -64,6 +66,9 @@ -- State type -> method map type ComponentTree = Map String MethodMap +-- State type -> all versions+type ComponentVersions = Map String [L.ByteString]+ #ifndef __HADDOCK__ -------------------------------------------------------------- -- A component consists of a list of subcomponents, some@@ -91,26 +96,27 @@ subHandlers b #endif -data Collection = Collection ComponentTree [IO ()]+data Collection = Collection ComponentTree ComponentVersions [IO ()] -addItem key item ioAction- = do Collection tree ioActions <- get+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) (ioAction:ioActions)+ 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, [IO ()])+collectHandlers :: (Methods a, Component a) => Proxy a -> (ComponentTree, ComponentVersions, [IO ()]) collectHandlers proxy- = case execState (collectHandlers' proxy) (Collection Map.empty []) of- Collection tree ioActions -> (tree, ioActions)+ = case execState (collectHandlers' proxy) (Collection Map.empty Map.empty []) of+ Collection tree versions ioActions -> (tree, versions, ioActions) collectHandlers' :: (Methods a, Component a) => Proxy a -> Collect () collectHandlers' proxy = let key = show (typeOf (unProxy proxy)) item = MethodMap $ Map.fromList [ (methodType m, m) | m <- methods proxy ]- in do addItem key item (onLoad proxy)+ versions = collectVersions proxy+ in do addItem key item versions (onLoad proxy) subHandlers (sub proxy) where sub :: Component a => Proxy a -> Dependencies a sub _ = undefined@@ -162,7 +168,7 @@ mainState v = collectHandlers (proxy v) -runTest = let (tree, io) = mainState Test+runTest = let (tree, _, io) = mainState Test in do sequence_ io print tree -}
src/HAppS/State/Control.hs view
@@ -1,6 +1,7 @@ {-# OPTIONS -cpp -fglasgow-exts #-} module HAppS.State.Control ( startSystemState+ , startSystemStateMultimaster , stdSaver , waitForTermination ) where@@ -37,6 +38,11 @@ = do _txConfig <- parseArgs saver <- stdSaver runTxSystem saver proxy++startSystemStateMultimaster proxy+ = do _txConfig <- parseArgs+ saver <- stdSaver+ runTxSystem' True saver proxy stdSaver = do pn <- getProgName return $ Queue (FileSaver ("_local/" ++pn++"_state"))
src/HAppS/State/Monad.hs view
@@ -1,7 +1,11 @@-{-# OPTIONS -fglasgow-exts #-}+{-# OPTIONS_GHC -fglasgow-exts #-}+{-# LANGUAGE CPP #-} module HAppS.State.Monad where -import Control.Exception(Exception)+import Control.Exception+#if __GLASGOW_HASKELL__ >= 610+--import GHC.Exception+#endif import Control.Concurrent.STM import HAppS.State.Types import HAppS.Data.Proxy@@ -73,7 +77,11 @@ 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))
src/HAppS/State/Saver/Impl/File.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} module HAppS.State.Saver.Impl.File ( fileReader, fileWriter ) where@@ -7,6 +8,9 @@ import Control.Concurrent import Control.Exception ( try )+#if __GLASGOW_HASKELL__ >= 610+import Control.Exception ( IOException )+#endif import qualified Data.ByteString.Lazy.Char8 as L import qualified Data.ByteString.Char8 as B import System.Directory ( createDirectoryIfMissing, renameFile, doesFileExist )@@ -19,13 +23,20 @@ logMF = logM "HAppS.State.Saver.Impl.File" +#if __GLASGOW_HASKELL__ < 610+tryE = try+#else+tryE :: IO a -> IO (Either IOException a)+tryE = try+#endif+ 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- try $ createDirectoryIfMissing True prefix+ tryE $ createDirectoryIfMissing True prefix return $ ReaderStream { readerClose = do return () , readerGet = do logMF NOTICE "fileReader: readerGet"@@ -51,7 +62,7 @@ return $ prefix </> formatFilePath cutoff key file <- getFileName logMF NOTICE ("fileWrter: "++key++" @ "++prefix)- try $ createDirectoryIfMissing True prefix+ tryE $ createDirectoryIfMissing True prefix hmv <- newMVar =<< openBinaryFile file WriteMode return $ WriterStream { writerClose = withMVar hmv hClose
+ src/HAppS/State/Spread.hs view
@@ -0,0 +1,161 @@+{-# LANGUAGE TemplateHaskell, DeriveDataTypeable #-}+module HAppS.State.Spread+ ( connectToCluster+ , changeEventMapping+ ) where++import HAppS.State.Transaction+import HAppS.State.Types+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 = 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 = 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"+ members <- 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 members 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 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 (Regular _) = "regular"+showMsgType (Membership _) = "membership"+showMsgType (Rejected _) = "rejected"+
src/HAppS/State/Transaction.hs view
@@ -1,9 +1,14 @@-{-# OPTIONS -fglasgow-exts -cpp -fth #-}+{-# OPTIONS -fglasgow-exts #-}+{-# LANGUAGE CPP, TemplateHaskell #-} module HAppS.State.Transaction where import Control.Concurrent import Control.Concurrent.STM-import Control.Exception(handle,throw,Exception(..),AsyncException(..),throwIO,evaluate)+#if __GLASGOW_HASKELL__ < 610+import Control.Exception(handle,Exception(..),AsyncException(..),throwIO,evaluate)+#else+import Control.Exception+#endif import Control.Monad.State import Control.Monad.Reader import qualified Data.Map as M@@ -23,11 +28,21 @@ import HAppS.State.Types import HAppS.Util.Common (Seconds) -import Data.Typeable+import Data.Dynamic import GHC.Base +import Prelude hiding (catch)+ import qualified Data.Binary as Binary +#if __GLASGOW_HASKELL__ < 610+type ExceptionT = Exception+#else+type ExceptionT = SomeException+#endif+++ logMT = logM "HAppS.State.Transaction" --getTime :: AnyEv EpochTime@@ -50,6 +65,16 @@ instance Serialize TxContext #endif +-- 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@@ -77,7 +102,8 @@ #ifndef __HADDOCK__ data EventHandler where UpdateHandler :: UpdateEvent ev res =>- (Maybe TxContext -> ev -> IO res) ->+ (TxContext -> ev -> IO ()) ->+ (ev -> IO res) -> (Object -> ev) -> EventHandler QueryHandler :: QueryEvent ev res =>@@ -101,8 +127,8 @@ emitFunc eventMap eventType ev = case M.lookup eventType eventMap of Nothing -> error $ "Emitted event to unknown component. Ev: " ++ eventType- Just (UpdateHandler fn _) -> unsafeCoerce# fn Nothing ev- Just (QueryHandler fn _) -> unsafeCoerce# fn ev+ 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'.@@ -115,56 +141,83 @@ emitEvent :: (Serialize ev, Typeable res) => ev -> IO res emitEvent ev = emitEvent' (show (typeOf ev)) ev -initEventMap :: (Methods st, Component st) => MVar TxControl -> Proxy st -> IO ()-initEventMap ctlVar componentProxy- = do eventMap <- createEventMap ctlVar componentProxy- writeIORef emitRef $ EmitInternal eventMap+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 maps <- forM (M.elems componentTree) $ \(MethodMap m) ->+ = do initEventStream+ maps <- forM (M.elems componentTree) $ \(MethodMap m) -> do tx <- createNewTxRun- forkIO $ forever $ atomically $- do HR ev fn <- readTChan (txEventQueue tx)- tev <- addTxId tx =<< newTxContext- writeTChan (txProcessQueue tx) (IHR tev ev fn) 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, _ioActions) = collectHandlers componentProxy+ where (componentTree, _versions, _ioActions) = collectHandlers componentProxy eventHandler tx (Update fn)- = let updateEmitter mbContext ev- = let realEv = ev- in do mv <- newEmptyMVar- let handler fn =- case mbContext of- Nothing -> writeTChan (txEventQueue tx) (HR realEv fn)- Just cxt -> do lastCxt <- readTVar (txLastTxContext tx)- writeTChan (txProcessQueue tx) $ IHR cxt realEv $- if txId lastCxt < txId cxt- then fn- else handleUpdate (putMVar mv) $ return $ error "Forced result from dated event."- atomically $ handler $ handleUpdate (putMVar mv) (fn realEv)- takeMVar mv- in UpdateHandler updateEmitter parseObject+ = 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- = let realEv = ev- in do mv <- newEmptyMVar- quickQuery' tx $ HR realEv $ handleQuery (putMVar mv) (fn realEv)- takeMVar mv+ = 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 () L.ByteString-instance UpdateEvent L.ByteString () +instance QueryEvent GetCheckpointState L.ByteString+instance UpdateEvent SetCheckpointState ()+ extraEvents :: Serialize st => TxRun st -> EventMap extraEvents tx = M.fromList [ (getStateType stateType, getStateHandler tx)@@ -174,49 +227,54 @@ t _ = undefined stateType = show (typeOf (t tx)) getStateHandler tx- = let fn :: () -> IO L.ByteString- fn () = do mv <- newEmptyMVar- quickQuery' tx $ HR () $ \context st ->+ = 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 (error "No parser for GetState event")+ in QueryHandler fn parseObject setNewStateHandler tx- = let fn :: L.ByteString -> IO ()- fn bs = do ((context, newState), _) <- evaluate $ deserialize bs- mv <- newEmptyMVar- quickQuery' tx $ HR () $ \_context _oldState ->- return (Just newState, putMVar mv ())- takeMVar mv- atomically $ writeTVar (txLastTxContext tx) context- in UpdateHandler (const fn) (error "No parser for SetNewState event")+ = 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 proxy = let (componentTree, _ioActions) = collectHandlers proxy+allStateTypes proxy = let (componentTree, _versions, _ioActions) = collectHandlers proxy in M.keys componentTree +componentVersions :: (Methods a, Component a) => Proxy a -> M.Map String [L.ByteString]+componentVersions proxy = let (_componentTree, versions, _ioActions) = collectHandlers proxy+ in versions+ componentIO :: (Methods a, Component a) => Proxy a -> [IO ()]-componentIO proxy = let (_componentTree, ioActions) = collectHandlers proxy+componentIO proxy = let (_componentTree, _versions, ioActions) = collectHandlers proxy in ioActions createNewTxRun :: IO (TxRun st) createNewTxRun = atomically $- do eventQueue <- newTChan- processQueue <- newTChan+ do processQueue <- newTChan lastContext <- newTVar (TxContext 0 0 0 (mkStdGen 42))- return $ TxRun eventQueue processQueue lastContext+ return $ TxRun processQueue lastContext setNewStateType str = "SetNewState: " ++ str getStateType str = "GetState: " ++ str setNewState :: TypeString -> L.ByteString -> IO () setNewState stateType state- = emitEvent' (setNewStateType stateType) state+ = emitEvent' (setNewStateType stateType) (SetCheckpointState state) getState :: TypeString -> IO L.ByteString getState stateType- = emitEvent' (getStateType stateType) ()+ = emitEvent' (getStateType stateType) GetCheckpointState data SetNewState st = SetNewState L.ByteString deriving (Typeable) data GetState st = GetState deriving (Typeable)@@ -270,11 +328,10 @@ => HR ev (RunHandler st ev) type RunHandler st ev = TxContext -> st -> IO (Maybe st, IO ())-data Res a = Ok a | Error Exception+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 {txEventQueue :: !(EventQueue st)- ,txProcessQueue :: !(ProcessQueue st)+data TxRun st = TxRun {txProcessQueue :: !(ProcessQueue st) ,txLastTxContext :: !(TVar TxContext)} @@ -284,7 +341,22 @@ 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@@ -292,41 +364,43 @@ runColdEventFunc :: TxContext -> Object -> EventMap -> IO () runColdEventFunc cxt obj eventMap- = case M.lookup eventType eventMap of- Nothing -> error $ "Couldn't find handler for cold event of type: " ++ eventType- Just (QueryHandler run parse)- -> do run (parse obj)- return ()- Just (UpdateHandler run parse)- -> do run (Just cxt) (parse obj)- return ()- where eventType = objectType obj+ = 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))- -> (res -> IO ()) -> Ev m res -> RunHandler st ev+ -> (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 res)+ 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 (throw e))+ return (Nothing,ofun (Left e)) -handleQuery :: (res -> IO ()) -> Query st res -> RunHandler st ev+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 :: (res -> IO ()) -> Update st res -> RunHandler st ev+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.@@ -338,20 +412,13 @@ checkDiff _old new = return (Just new) -processEvent :: (Serialize ev) =>- TxRun st -> ev - -> (RunHandler st ev)- -> IO ()-processEvent txrun ev runHandler- = atomically $ writeTChan (txEventQueue txrun) $ HR ev (runHandler)- getEpochMilli :: IO EpochMilli getEpochMilli = do TOD sec pico <- getClockTime return $ fromIntegral $ sec * 1000 + pico `div` 10^9 -newTxContext :: STM TxContext-newTxContext = unsafeIOToSTM $ do+newTxContext :: IO TxContext+newTxContext = do milli <- getEpochMilli let txid = -1 -- Not set yet. sgen <- modifyMVar globalRandomGen (return . split)@@ -363,7 +430,7 @@ = do lastContext <- readTVar (txLastTxContext tx) let new = context{txId = txId lastContext + 1} writeTVar (txLastTxContext tx) new- return 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?@@ -383,6 +450,7 @@ { 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 ())] -- } @@ -410,6 +478,9 @@ 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@@ -435,9 +506,14 @@ Just st' -> do eventSaver <- readMVar eventSaverVar writerAdd eventSaver (EventLogEntry context (mkObject ev)) (logMT NOTICE "> disk " >> ra) loop st'+#if __GLASGOW_HASKELL__ < 610 in do forkIO $ handle excHandler $ loop st0 return () where excHandler (AsyncException ThreadKilled) = return () excHandler BlockedIndefinitely = return () excHandler e = throwIO e-+#else+ in do forkIO $ (loop st0) `catch` (\ThreadKilled -> return ())+ `catch` (\BlockedIndefinitely -> return ())+ return ()+#endif
src/HAppS/State/TxControl.hs view
@@ -1,5 +1,6 @@ module HAppS.State.TxControl ( runTxSystem+ , runTxSystem' , shutdownSystem ) where @@ -12,6 +13,7 @@ import HAppS.State.Checkpoint import HAppS.State.Saver import HAppS.State.Transaction+import HAppS.State.Spread import HAppS.State.Types import HAppS.State.ComponentSystem import HAppS.Data.Proxy@@ -21,13 +23,23 @@ -- | Run a transaction system runTxSystem :: (Methods st, Component st) => Saver -> Proxy st -> IO (MVar TxControl)-runTxSystem saver stateProxy =+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."- initEventMap ctl stateProxy+ localEventMap <- createEventMap ctl stateProxy+ setNewEventMap localEventMap logMM NOTICE "Restoring state."- restoreState ctl+ 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
src/HAppS/State/Types.hs view
@@ -1,4 +1,5 @@-{-# OPTIONS -fglasgow-exts -fth -cpp -fallow-undecidable-instances #-}+{-# OPTIONS -fglasgow-exts #-}+{-# LANGUAGE CPP, UndecidableInstances, TemplateHaskell #-} module HAppS.State.Types where import Control.Concurrent.STM