diff --git a/COPYING b/COPYING new file mode 100644 --- /dev/null +++ b/COPYING @@ -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. diff --git a/HAppS-State.cabal b/HAppS-State.cabal --- a/HAppS-State.cabal +++ b/HAppS-State.cabal @@ -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 diff --git a/src/HAppS/State.hs b/src/HAppS/State.hs --- a/src/HAppS/State.hs +++ b/src/HAppS/State.hs @@ -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, diff --git a/src/HAppS/State/Checkpoint.hs b/src/HAppS/State/Checkpoint.hs --- a/src/HAppS/State/Checkpoint.hs +++ b/src/HAppS/State/Checkpoint.hs @@ -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. diff --git a/src/HAppS/State/ComponentSystem.hs b/src/HAppS/State/ComponentSystem.hs --- a/src/HAppS/State/ComponentSystem.hs +++ b/src/HAppS/State/ComponentSystem.hs @@ -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 -} diff --git a/src/HAppS/State/Control.hs b/src/HAppS/State/Control.hs --- a/src/HAppS/State/Control.hs +++ b/src/HAppS/State/Control.hs @@ -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")) diff --git a/src/HAppS/State/Monad.hs b/src/HAppS/State/Monad.hs --- a/src/HAppS/State/Monad.hs +++ b/src/HAppS/State/Monad.hs @@ -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)) diff --git a/src/HAppS/State/Saver/Impl/File.hs b/src/HAppS/State/Saver/Impl/File.hs --- a/src/HAppS/State/Saver/Impl/File.hs +++ b/src/HAppS/State/Saver/Impl/File.hs @@ -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 diff --git a/src/HAppS/State/Spread.hs b/src/HAppS/State/Spread.hs new file mode 100644 --- /dev/null +++ b/src/HAppS/State/Spread.hs @@ -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" + diff --git a/src/HAppS/State/Transaction.hs b/src/HAppS/State/Transaction.hs --- a/src/HAppS/State/Transaction.hs +++ b/src/HAppS/State/Transaction.hs @@ -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 diff --git a/src/HAppS/State/TxControl.hs b/src/HAppS/State/TxControl.hs --- a/src/HAppS/State/TxControl.hs +++ b/src/HAppS/State/TxControl.hs @@ -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 diff --git a/src/HAppS/State/Types.hs b/src/HAppS/State/Types.hs --- a/src/HAppS/State/Types.hs +++ b/src/HAppS/State/Types.hs @@ -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