diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +acid-state-dist - replication backend for acid-state +Copyright © 2015 Max Voit + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the "Software"), +to deal in the Software without restriction, including without limitation +the rights to use, copy, modify, merge, publish, distribute, sublicense, +and/or sell copies of the Software, and to permit persons to whom the +Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included +in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES +OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, +DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE +OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/acid-state-dist.cabal b/acid-state-dist.cabal new file mode 100644 --- /dev/null +++ b/acid-state-dist.cabal @@ -0,0 +1,211 @@ +-- Initial acid-state-dist.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/ + +-- The name of the package. +name: acid-state-dist + +-- The package version. See the Haskell package versioning policy (PVP) +-- for standards guiding when and how versions should be incremented. +-- http://www.haskell.org/haskellwiki/Package_versioning_policy +-- PVP summary: +-+------- breaking API changes +-- | | +----- non-breaking API additions +-- | | | +--- code changes with no API change +version: 0.1.0.0 + +-- A short (one-line) description of the package. +synopsis: Replication backend for acid-state + +-- A longer description of the package. +-- description: + +-- The license under which the package is released. +license: MIT + +-- The file containing the license text. +license-file: LICENSE + +-- The package author(s). +author: Max Voit + +-- An email address to which users can send suggestions, bug reports, and +-- patches. +maintainer: + +-- A copyright notice. +-- copyright: + +category: Database + +build-type: Simple + +-- Extra files to be distributed with the package, such as examples or a +-- README. +-- extra-source-files: + +-- Constraint on the version of Cabal needed to build this package. +cabal-version: >=1.10 + +-- Flag for controlling debug output +flag debug + description: enable debug output + default: True + +-- Source +source-repository head + type: git + location: http://git.with-eyes.net/gitweb/?p=acid-state-dist.git + +library + -- Modules exported by the library. + exposed-modules: Data.Acid.Centered, + Data.Acid.Centered.Master, + Data.Acid.Centered.Slave + + -- Modules included in this library but not exported. + other-modules: Data.Acid.Centered.Common + + -- LANGUAGE extensions used by modules in this package. + -- other-extensions: + + -- Other library packages from which modules are imported. + build-depends: base > 4.7 && < 4.9, + safecopy, + acid-state > 0.12 && < 0.13, + concurrent-extra, + cereal, + zeromq4-haskell, + bytestring, + containers, + transformers, + stm, + semigroups, + safe, + filepath, + mtl + + -- Directories containing source files. + hs-source-dirs: src + + -- Base language which the package is written in. + default-language: Haskell2010 + + default-extensions: CPP + ghc-options: -Wall -threaded + -- Switch on debugging by "-Unodebug", off by "-Dnodebug" + if flag(debug) + cpp-options: -Unodebug + else + cpp-options: -Dnodebug + + +---------------------------------------------------------------------- +-- Tests +test-suite Simple + main-is: Simple.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, + safecopy, acid-state, acid-state-dist + +test-suite SlaveUpdates + main-is: SlaveUpdates.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, + safecopy, acid-state, acid-state-dist + +test-suite CRCFail + main-is: CRCFail.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, + safecopy, acid-state, acid-state-dist + +test-suite CheckpointSync + main-is: CheckpointSync.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, + safecopy, acid-state, acid-state-dist + +test-suite OrderingRandom + main-is: OrderingRandom.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, random, + safecopy, acid-state, acid-state-dist + +test-suite NReplication + main-is: NReplication.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, + safecopy, acid-state, acid-state-dist + +test-suite UpdateError + main-is: UpdateError.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, + safecopy, acid-state, acid-state-dist + +test-suite SyncTimeout + main-is: SyncTimeout.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, + safecopy, acid-state, acid-state-dist + +---------------------------------------------------------------------- +-- Benchmarks +benchmark Local + main-is: Local.hs + type: exitcode-stdio-1.0 + hs-source-dirs: benchmark + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, criterion, + safecopy, acid-state + +benchmark MasterOnly + main-is: MasterOnly.hs + type: exitcode-stdio-1.0 + hs-source-dirs: benchmark + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, criterion, + safecopy, acid-state, acid-state-dist + +benchmark MasterSlave + main-is: MasterSlave.hs + type: exitcode-stdio-1.0 + hs-source-dirs: benchmark + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, criterion, + safecopy, acid-state, acid-state-dist + +benchmark Slave + main-is: Slave.hs + type: exitcode-stdio-1.0 + hs-source-dirs: benchmark + ghc-options: -Wall -threaded + default-language: Haskell2010 + build-depends: base, directory, mtl, criterion, + safecopy, acid-state, acid-state-dist diff --git a/benchmark/Local.hs b/benchmark/Local.hs new file mode 100644 --- /dev/null +++ b/benchmark/Local.hs @@ -0,0 +1,26 @@ +{-# LANGUAGE TypeFamilies #-} + +import Criterion.Main + +import Data.Acid + +import System.Exit (exitSuccess) + +-- common benchmarking stuff +import BenchCommon + +main :: IO () +main = do + -- init acid + cleanup "state/Local" + acid <- openLocalStateFrom "state/Local/m" (IntState 0) + + -- run benchmark + defaultMain + [ bench "Local" $ nfIO (masterBench acid) + , bench "Local-grouped" $ nfIO (masterBenchGrouped acid) + ] + + -- cleanup + closeAcidState acid + exitSuccess diff --git a/benchmark/MasterOnly.hs b/benchmark/MasterOnly.hs new file mode 100644 --- /dev/null +++ b/benchmark/MasterOnly.hs @@ -0,0 +1,28 @@ +{-# LANGUAGE TypeFamilies #-} + +import Criterion.Main + +import Data.Acid +import Data.Acid.Centered + +import System.Exit (exitSuccess) + +-- common benchmarking stuff +import BenchCommon + +main :: IO () +main = do + -- init acid + cleanup "state/MasterOnly" + acid <- openMasterStateFrom "state/MasterOnly/m" "127.0.0.1" 3333 (IntState 0) + delaySec 3 + + -- run benchmark + defaultMain + [ bench "MasterOnly" $ nfIO (masterBench acid) + , bench "MasterOnly-grouped" $ nfIO (masterBenchGrouped acid) + ] + + -- cleanup + closeAcidState acid + exitSuccess diff --git a/benchmark/MasterSlave.hs b/benchmark/MasterSlave.hs new file mode 100644 --- /dev/null +++ b/benchmark/MasterSlave.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE TypeFamilies #-} + +import Criterion.Main + +import Data.Acid +import Data.Acid.Centered + +import System.Exit (exitSuccess) +import Control.Monad (void) +import Control.Concurrent (forkIO) +import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar) + +-- common benchmarking stuff +import BenchCommon + +-- the slave +slave :: MVar () -> IO () +slave sync = do + acid <- enslaveStateFrom "state/MasterSlave/s1" "localhost" 3333 (IntState 0) + takeMVar sync + closeAcidState acid + +main :: IO () +main = do + -- init acid + cleanup "state/MasterSlave" + acid <- openMasterStateFrom "state/MasterSlave/m" "127.0.0.1" 3333 (IntState 0) + sync <- newEmptyMVar + void $ forkIO $ slave sync + delaySec 3 + + -- run benchmark + defaultMain + [ bench "MasterSlave" $ nfIO (masterBench acid) + , bench "MasterSlave-grouped" $ nfIO (masterBenchGrouped acid) + ] + + -- cleanup + putMVar sync () + delaySec 5 + closeAcidState acid + exitSuccess diff --git a/benchmark/Slave.hs b/benchmark/Slave.hs new file mode 100644 --- /dev/null +++ b/benchmark/Slave.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE TypeFamilies #-} + +import Criterion.Main + +import Data.Acid +import Data.Acid.Centered + +import System.Exit (exitSuccess) +import Control.Monad (void) +import Control.Concurrent (forkIO) +import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar) + +-- common benchmarking stuff +import BenchCommon + +-- the master +master :: MVar () -> IO () +master sync = do + acid <- openMasterStateFrom "state/Slave/m" "127.0.0.1" 3333 (IntState 0) + takeMVar sync + closeAcidState acid + +main :: IO () +main = do + -- init acid + cleanup "state/Slave" + sync <- newEmptyMVar + void $ forkIO $ master sync + acid <- enslaveStateFrom "state/Slave/s1" "localhost" 3333 (IntState 0) + delaySec 3 + + -- run benchmark + defaultMain + [ bench "Slave" $ nfIO (slaveBench acid) + , bench "Slave-grouped" $ nfIO (slaveBenchGrouped acid) + ] + + -- cleanup + delaySec 5 + closeAcidState acid + putMVar sync () + exitSuccess diff --git a/src/Data/Acid/Centered.hs b/src/Data/Acid/Centered.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/Centered.hs @@ -0,0 +1,89 @@ +-------------------------------------------------------------------------------- +{- | + Module : Data.Acid.Centered + Copyright : MIT + + Maintainer : max.voit+hdv@with-eyes.net + Portability : non-portable (uses GHC extensions) + + A replication backend for acid-state that is centered around a Master node. + Slave nodes connect to the Master, are synchronized and get updated by the + Master continuously. This backend offers two flavors of operation: + + [@Regular operation@] No redundancy guarantees but fast. + [@Redundant operation@] Guarantees redundant replication on /n/ nodes but + slower. + + In both cases Slaves' Updates block (and eventually time out) if the Master is + unreachable. + + Queries operate on the local state with the performance known from acid-state. + Note that state on all nodes is eventually consistent, i.e. it might differ + shortly (for Queries run concurrently to Updates being serialized). + +-} + +module Data.Acid.Centered + ( +-- * Usage +-- | +-- Open your AcidState using one of the functions below. Take care to use the +-- same initial-state on all nodes. Afterwards the usual interface of acid state +-- is available. +-- +-- Always make sure to have sensible exception management since naturally a lot +-- more error sources exist with this backend than do for a 'Data.Acid.Local' +-- AcidState. +-- Using 'Control.Exception.bracket' is recommended for achieving this +-- conveniently: +-- +-- > main = bracket +-- > (enslaveState ...) +-- > closeAcidState +-- > $ \acid -> do +-- > ... +-- +-- 'Data.Acid.createCheckpoint' issued on Master is a global operation, +-- while issued on a Slave it is not. Note that Checkpoints on Master reduce the +-- amount of Updates to be transferred when a Slave (re-)connects. +-- +-- 'Data.Acid.createArchive' is an operation local to each node since usually +-- further action is required. For global Archives see 'createArchiveGlobally'. + + + -- * Regular operation + -- | + -- Running Updates on Master works as known from acid-state and with + -- negligible performance loss. + -- On Slaves Updates are delayed by approximately one round trip time (RTT). + -- When no Slaves are connected Updates are only serialized on the Master, + -- i.e. there is no redundancy. + openMasterState + , openMasterStateFrom + , enslaveState + , enslaveStateFrom + -- * Redundant operation + -- | + -- When Updates are scheduled they are sent out and written to disk on all + -- nodes. However, an Update is visible to Queries (and its result returned) + -- /only/ as soon as at least /n/ nodes are done replicating it. Thus each + -- Update is delayed for at least one RTT. + -- + -- If less than /n-1/ Slave nodes are connected, all Updates are blocked + -- until enough nodes are available again. Queries are not affected and + -- operate on the last /n/-replicated state. + -- + -- /Note:/ Shutting down and restarting the Master resumes the last state + -- including even Updates that were not /n/-replicated. + , openRedMasterState + , openRedMasterStateFrom + , enslaveRedState + , enslaveRedStateFrom + -- * Types + , PortNumber + ) where + +import Data.Acid.Centered.Master +import Data.Acid.Centered.Slave +import Data.Acid.Centered.Common + diff --git a/src/Data/Acid/Centered/Common.hs b/src/Data/Acid/Centered/Common.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/Centered/Common.hs @@ -0,0 +1,182 @@ +{-# LANGUAGE OverloadedStrings #-} +-------------------------------------------------------------------------------- +{- | + Module : Data.Acid.Centered.Common + Copyright : MIT + + Maintainer : max.voit+hdv@with-eyes.net + Portability : non-portable (uses GHC extensions) + + Stuff common to Master and Slave in Centered systems. + +-} + +module Data.Acid.Centered.Common + ( + debug + , whenM + , waitPoll + , waitPollN + , crcOfState + , Crc + , NodeRevision + , Revision + , RequestID + , PortNumber + , SlaveMessage(..) + , MasterMessage(..) + , AcidException(..) + ) where + +import Data.Typeable (Typeable) +import Data.SafeCopy (safePut) + +import Data.Acid.Core (Tagged, withCoreState) +import Data.Acid.Local (localCore) +import Data.Acid.Abstract (downcast) +import Data.Acid (AcidState, IsAcidic) +import Data.Acid.CRC (crc16) + +import Control.Concurrent (threadDelay) + +import Control.Monad (liftM, liftM2, liftM3, + unless, when) +import Control.Exception (Exception) + +import Data.ByteString.Lazy.Char8 (ByteString) +import Data.Serialize (Serialize(..), put, get, + putWord8, getWord8, + runPutLazy) +import Data.Word (Word16) + +#ifdef nodebug +#else +import System.IO (stderr, hPutStrLn) +import qualified Control.Concurrent.Lock as L +import System.IO.Unsafe (unsafePerformIO) +#endif + +-------------------------------------------------------------------------------- + +-- | Number of a port for establishing a network connection. +type PortNumber = Int + +-- | (Current) Revision of a node. +type NodeRevision = Int + +-- | Revision an Update resembles. +type Revision = Int + +-- | ID of an Update Request. +type RequestID = Int + +-- | We use CRC16 for now. +type Crc = Word16 + +#ifdef nodebug +-- | Debugging disabled. +debug :: String -> IO () +debug _ = return () +#else +-- | Lock for non-interleaved debug output. +{-# NOINLINE debugLock #-} +debugLock :: L.Lock +debugLock = unsafePerformIO L.new + +-- | Debugging without interleaving output from different threads. +debug :: String -> IO () +debug = L.with debugLock . hPutStrLn stderr +#endif + +-- | Internally used for killing handler threads. +data AcidException = GracefulExit + deriving (Show, Typeable) + +instance Exception AcidException + +-- | Messages the Master sends to Slaves. +data MasterMessage = DoRep Revision (Maybe RequestID) (Tagged ByteString) + | DoSyncRep Revision (Tagged ByteString) + | SyncDone Crc + | DoCheckpoint Revision + | DoSyncCheckpoint Revision ByteString + | DoArchive Revision + | FullRep Revision + | FullRepTo Revision + | MayQuit + | MasterQuit + deriving (Show) + +-- | Messages Slaves sends to the Master. +data SlaveMessage = NewSlave Int + | RepDone Int + | RepError + | ReqUpdate RequestID (Tagged ByteString) + | SlaveQuit + deriving (Show) + +instance Serialize MasterMessage where + put msg = case msg of + DoRep r i d -> putWord8 0 >> put r >> put i >> put d + DoSyncRep r d -> putWord8 1 >> put r >> put d + SyncDone c -> putWord8 2 >> put c + DoCheckpoint r -> putWord8 3 >> put r + DoSyncCheckpoint r d -> putWord8 4 >> put r >> put d + DoArchive r -> putWord8 5 >> put r + FullRep r -> putWord8 6 >> put r + FullRepTo r -> putWord8 7 >> put r + MayQuit -> putWord8 8 + MasterQuit -> putWord8 9 + get = do + tag <- getWord8 + case tag of + 0 -> liftM3 DoRep get get get + 1 -> liftM2 DoSyncRep get get + 2 -> liftM SyncDone get + 3 -> liftM DoCheckpoint get + 4 -> liftM2 DoSyncCheckpoint get get + 5 -> liftM DoArchive get + 6 -> liftM FullRep get + 7 -> liftM FullRepTo get + 8 -> return MayQuit + 9 -> return MasterQuit + _ -> error $ "Data.Serialize.get failed for MasterMessage: invalid tag " ++ show tag + +instance Serialize SlaveMessage where + put msg = case msg of + NewSlave r -> putWord8 0 >> put r + RepDone r -> putWord8 1 >> put r + RepError -> putWord8 2 + ReqUpdate i d -> putWord8 3 >> put i >> put d + SlaveQuit -> putWord8 9 + get = do + tag <- getWord8 + case tag of + 0 -> liftM NewSlave get + 1 -> liftM RepDone get + 2 -> return RepError + 3 -> liftM2 ReqUpdate get get + 9 -> return SlaveQuit + _ -> error $ "Data.Serialize.get failed for SlaveMessage: invalid tag " ++ show tag + +-- | Compute the CRC of a state. +crcOfState :: (IsAcidic st, Typeable st) => AcidState st -> IO Crc +crcOfState state = do + let lst = downcast state + withCoreState (localCore lst) $ \st -> do + let encoded = runPutLazy (safePut st) + return $ crc16 encoded + +-- | By polling, wait until predicate fulfilled. +waitPoll :: Int -> IO Bool -> IO () +waitPoll t p = p >>= \e -> unless e $ threadDelay t >> waitPoll t p + +-- | By polling, wait until predicate fulfilled. Poll at max. n times. +waitPollN :: Int -> Int -> IO Bool -> IO () +waitPollN t n p + | n == 0 = return () + | otherwise = p >>= \e -> unless e $ threadDelay t >> waitPollN t (n-1) p + +-- | Monadic when +whenM :: Monad m => m Bool -> m () -> m () +whenM b a = b >>= flip when a diff --git a/src/Data/Acid/Centered/Master.hs b/src/Data/Acid/Centered/Master.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/Centered/Master.hs @@ -0,0 +1,526 @@ +{-# LANGUAGE DeriveDataTypeable, RecordWildCards, FlexibleContexts #-} +-------------------------------------------------------------------------------- +{- | + Module : Data.Acid.Centered.Master + Copyright : MIT + + Maintainer : max.voit+hdv@with-eyes.net + Portability : non-portable (uses GHC extensions) + + The Master part of the Centered replication backend for acid state. + +-} +{- big chunks still todo: + o authentification + o encryption +-} +module Data.Acid.Centered.Master + ( + openMasterState + , openMasterStateFrom + , openRedMasterState + , openRedMasterStateFrom + , createArchiveGlobally + , MasterState(..) + ) where + +import Data.Typeable +import Data.SafeCopy +import Data.Serialize (decode, encode, runPutLazy) + +import Data.Acid +import Data.Acid.Core +import Data.Acid.Abstract +import Data.Acid.Local +import Data.Acid.Log + +import Data.Acid.Centered.Common + +import Control.Concurrent (forkIO, ThreadId, myThreadId) +import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan, dupChan) +import Control.Concurrent.STM.TVar (readTVar) +import Control.Concurrent.MVar(MVar, newMVar, newEmptyMVar, + takeMVar, putMVar, tryPutMVar, isEmptyMVar, + modifyMVar, modifyMVar_, withMVar) + +import Control.Monad.STM (atomically) +import Control.Monad (when, unless, void, forM_, liftM2) +import Control.Exception (handle, throwTo, SomeException) + +import System.ZMQ4 (Context, Socket, Router(..), Receiver, + setReceiveHighWM, setSendHighWM, restrict, + context, term, socket, close, bind, unbind, + poll, Poll(..), Event(..), + sendMulti, receiveMulti) +import System.FilePath ( () ) + +import qualified Data.ByteString.Lazy.Char8 as CSL +import Data.ByteString.Lazy.Char8 (ByteString) +import qualified Data.ByteString.Char8 as CS + +import qualified Data.Map as M +import Data.Map (Map) +import qualified Data.IntMap as IM +import Data.IntMap (IntMap) +import qualified Data.List.NonEmpty as NEL +import Safe (headDef) + +-------------------------------------------------------------------------------- + +-- | Master state structure, for internal use. +data MasterState st + = MasterState { localState :: AcidState st + , nodeStatus :: MVar NodeStatus + , repRedundancy :: Int + , repFinalizers :: MVar (IntMap (IO ())) + , masterStateLock :: MVar () + , masterRevision :: MVar NodeRevision + , masterRevisionN :: MVar NodeRevision + , masterReplicationChan :: Chan ReplicationItem + , masterReplicationChanN :: Chan ReplicationItem + , masterReqThreadId :: MVar ThreadId + , masterRepLThreadId :: MVar ThreadId + , masterRepNThreadId :: MVar ThreadId + , masterParentThreadId :: ThreadId + , zmqContext :: Context + , zmqAddr :: String + , zmqSocket :: MVar (Socket Router) + } deriving (Typeable) + +type NodeIdentity = CS.ByteString +type NodeStatus = Map NodeIdentity NodeRevision +type Callback = IO (IO ()) -- an IO action that returns a finalizer +data ReplicationItem = + RIEnd + | RICheckpoint + | RIArchive + | RIUpdate (Tagged ByteString) (Either Callback (RequestID, NodeIdentity)) + +-- | The request handler on master node. Does +-- o handle receiving requests from nodes, +-- o answering as needed (old updates), +-- o bookkeeping on node states. +masterRequestHandler :: (IsAcidic st, Typeable st) => MasterState st -> IO () +masterRequestHandler masterState@MasterState{..} = do + mtid <- myThreadId + putMVar masterReqThreadId mtid + let loop = handle (\e -> throwTo masterParentThreadId (e :: SomeException)) $ + handle killHandler $ do + -- take one frame + -- waitRead =<< readMVar zmqSocket + -- FIXME: we needn't poll if not for strange zmq behaviour + re <- withMVar zmqSocket $ \sock -> poll 100 [Sock sock [In] Nothing] + unless (null $ head re) $ do + (ident, msg) <- withMVar zmqSocket receiveFrame + handleMessage ident msg + loop + loop + where + killHandler :: AcidException -> IO () + killHandler GracefulExit = return () + identityIsValid i = do + isMember <- withMVar nodeStatus $ return . (i `M.member`) + if isMember then return True + else do + debug $ "Request by unknown node [" ++ CS.unpack i ++ "]" + sendToSlave zmqSocket MayQuit i + return False + handleMessage i m = case m of + -- New Slave joined. + NewSlave r -> connectNode masterState i r + -- Slave is done replicating. + RepDone r -> whenM (identityIsValid i) $ + updateNodeStatus masterState i r + -- Slave sends an Udate. + ReqUpdate rid event -> whenM (identityIsValid i) $ + queueRepItem masterState (RIUpdate event (Right (rid, i))) + -- Slave quits. + SlaveQuit -> do + sendToSlave zmqSocket MayQuit i + removeFromNodeStatus nodeStatus i + RepError -> do + sendToSlave zmqSocket MayQuit i + removeFromNodeStatus nodeStatus i + -- no other messages possible + +-- | Remove a Slave node from NodeStatus. +removeFromNodeStatus :: MVar NodeStatus -> NodeIdentity -> IO () +removeFromNodeStatus nodeStatus ident = + modifyMVar_ nodeStatus $ return . M.delete ident + +-- | Update the NodeStatus after a node has replicated an Update. +updateNodeStatus :: MasterState st -> NodeIdentity -> Int -> IO () +updateNodeStatus MasterState{..} ident rev = + modifyMVar_ nodeStatus $ \ns -> do + when (ns M.! ident /= (rev - 1)) $ + error $ "Invalid increment of node status " + ++ show (ns M.! ident) ++ " -> " ++ show rev + let rns = M.adjust (+1) ident ns + -- only for redundant operation: + when ((repRedundancy > 1) && (M.size (M.filter (>=rev) rns) >= (repRedundancy - 1))) $ do + debug $ "Full replication of " ++ show rev + -- finalize local replication + runAndDelFinalizer rev + -- send out FullRep signal + forM_ (M.keys ns) $ sendToSlave zmqSocket (FullRep rev) + return rns + where + runAndDelFinalizer r = modifyMVar_ repFinalizers $ \rf -> do + rf IM.! r + return $ IM.delete r rf + +-- | Connect a new Slave by getting it up-to-date, +-- i.e. send all past events as Updates. This is fire&forget. +connectNode :: (IsAcidic st, Typeable st) => MasterState st -> NodeIdentity -> Revision -> IO () +connectNode MasterState{..} i revision = + -- locking masterRevision prohibits additional events written on disk + withMVar masterRevision $ \mr -> + modifyMVar_ nodeStatus $ \ns -> do + -- crc generated from localCore thus corresponds to disk + crc <- crcOfState localState + -- if there has been a checkpoint in between: + lastCp <- getLastCheckpointRev localState + let lastCpRev = cpRevision lastCp + debug $ "Found checkpoint at revision " ++ show lastCpRev + if lastCpRev > revision then do + -- send last checkpoint and newer events + sendSyncCheckpoint zmqSocket lastCp i + pastUpdates <- getPastUpdates localState lastCpRev + forM_ pastUpdates $ \(r, u) -> sendSyncUpdate zmqSocket r u i + else do + -- just the events + pastUpdates <- getPastUpdates localState revision + forM_ pastUpdates $ \(r, u) -> sendSyncUpdate zmqSocket r u i + -- now done, crc + sendToSlave zmqSocket (SyncDone crc) i + let nns = M.insert i mr ns + -- only for redundant operation: + when (repRedundancy > 1) $ checkRepStatus mr nns + return nns + where + cpRevision (Checkpoint r _) = r + sendSyncCheckpoint sock (Checkpoint cr encoded) = + sendToSlave sock (DoSyncCheckpoint cr encoded) + sendSyncUpdate sock r encoded = + sendToSlave sock (DoSyncRep r encoded) + -- FIXME: do this better (less than maxRev is possible in corner cases) + checkRepStatus maxRev pns = + when (M.size (M.filter (>= maxRev) pns) >= (repRedundancy-2)) $ do + debug $ "Full replication up to " ++ show maxRev + -- finalize local replication + modifyMVar_ repFinalizers $ \rf -> do + forM_ (filter (<= maxRev) (IM.keys rf)) $ \r -> rf IM.! r + return $ IM.filterWithKey (\k _ -> k > maxRev) rf + -- send out FullRep signal + forM_ (M.keys pns) $ sendToSlave zmqSocket (FullRepTo maxRev) + + +-- | Fetch past Updates from FileLog for replication. +getPastUpdates :: (Typeable st) => AcidState st -> Int -> IO [(Int, Tagged ByteString)] +getPastUpdates state startRev = + liftM2 zip (return [(startRev+1)..]) (readEntriesFrom (localEvents $ downcast state) startRev) + +-- | Get the revision at which the last checkpoint was taken. +getLastCheckpointRev :: (Typeable st) => AcidState st -> IO Checkpoint +getLastCheckpointRev state = do + let cplog = localCheckpoints $ downcast state + nextId <- atomically $ readTVar $ logNextEntryId cplog + cps <- readEntriesFrom cplog (nextId - 1) + return $ headDef (Checkpoint 0 CSL.empty) cps + +-- | Send a message to a Slave +sendToSlave :: MVar (Socket Router) -> MasterMessage -> NodeIdentity -> IO () +sendToSlave msock msg ident = withMVar msock $ \sock -> sendMulti sock $ NEL.fromList [ident, encode msg] + +-- | Receive one Frame. A Frame consists of two messages: +-- sender ID and actual content +receiveFrame :: (Receiver t) => Socket t -> IO (NodeIdentity, SlaveMessage) +receiveFrame sock = do + list <- receiveMulti sock + when (length list /= 2) $ error "Received invalid frame." + let ident = head list + let msg = list !! 1 + case decode msg of + Left str -> error $ "Data.Serialize.decode failed on SlaveMessage: " ++ show str + Right smsg -> do + debug $ "Received from [" ++ CS.unpack ident ++ "]: " + ++ take 20 (show smsg) + return (ident, smsg) + +-- | Open the Master state. +-- +-- The directory for the local state files is the default one ("state/[typeOf state]/"). +openMasterState :: (IsAcidic st, Typeable st) => + String -- ^ address to bind (useful to listen on specific interfaces only) + -> PortNumber -- ^ port to bind to + -> st -- ^ initial state + -> IO (AcidState st) +openMasterState address port initialState = + openMasterStateFrom ("state" show (typeOf initialState)) address port initialState + +-- | Open the master state from a specific location. +openMasterStateFrom :: (IsAcidic st, Typeable st) => + FilePath -- ^ location of the local state files + -> String -- ^ address to bind (useful to listen on specific interfaces only) + -> PortNumber -- ^ port to bind to + -> st -- ^ initial state + -> IO (AcidState st) +openMasterStateFrom directory address port = + openRedMasterStateFrom directory address port 0 + +-- | Open the master state with /n/-redundant replication. +-- +-- The directory for the local state files is the default one ("state/[typeOf +-- state]/"). +openRedMasterState :: (IsAcidic st, Typeable st) => + String -- ^ address to bind (useful to listen on specific interfaces only) + -> PortNumber -- ^ port to bind to + -> Int -- ^ guarantee n-redundant replication + -> st -- ^ initial state + -> IO (AcidState st) +openRedMasterState address port red initialState = + openRedMasterStateFrom ("state" show (typeOf initialState)) address port red initialState + +-- | Open the master state from a specific location with redundant replication. +openRedMasterStateFrom :: (IsAcidic st, Typeable st) => + FilePath -- ^ location of the local state files + -> String -- ^ address to bind (useful to listen on specific interfaces only) + -> PortNumber -- ^ port to bind to + -> Int -- ^ guarantee /n/-redundant replication + -> st -- ^ initial state + -> IO (AcidState st) +openRedMasterStateFrom directory address port red initialState = do + debug "opening master state" + -- local + lst <- openLocalStateFrom directory initialState + let levs = localEvents $ downcast lst + lrev <- atomically $ readTVar $ logNextEntryId levs + rev <- newMVar lrev + revN <- newMVar lrev + repChan <- newChan + repChanN <- dupChan repChan + repFin <- newMVar IM.empty + ns <- newMVar M.empty + repTidL <- newEmptyMVar + repTidN <- newEmptyMVar + reqTid <- newEmptyMVar + parTid <- myThreadId + sLock <- newEmptyMVar + -- remote + let addr = "tcp://" ++ address ++ ":" ++ show port + ctx <- context + sock <- socket ctx Router + setReceiveHighWM (restrict (100*1000 :: Int)) sock + setSendHighWM (restrict (100*1000 :: Int)) sock + bind sock addr + msock <- newMVar sock + + let masterState = MasterState { localState = lst + , nodeStatus = ns + , repRedundancy = red + , repFinalizers = repFin + , masterStateLock = sLock + , masterRevision = rev + , masterRevisionN = revN + , masterReplicationChan = repChan + , masterReplicationChanN = repChanN + , masterRepLThreadId = repTidL + , masterRepNThreadId = repTidN + , masterReqThreadId = reqTid + , masterParentThreadId = parTid + , zmqContext = ctx + , zmqAddr = addr + , zmqSocket = msock + } + void $ forkIO $ masterRequestHandler masterState + void $ forkIO $ masterReplicationHandlerL masterState + void $ forkIO $ masterReplicationHandlerN masterState + return $ toAcidState masterState + +-- | Close the master state. +closeMasterState :: MasterState st -> IO () +closeMasterState MasterState{..} = + -- disallow requests + whenM (tryPutMVar masterStateLock ()) $ do + debug "Closing master state." + -- send nodes quit + debug "Nodes quitting." + withMVar nodeStatus $ mapM_ (sendToSlave zmqSocket MasterQuit) . M.keys + -- wait all nodes done + waitPollN 100 1000 (withMVar nodeStatus (return . M.null)) + -- wait replication chan + debug "Waiting for repChans to empty." + writeChan masterReplicationChan RIEnd + mtid <- myThreadId + putMVar masterRepLThreadId mtid + putMVar masterRepNThreadId mtid + -- kill handler + debug "Killing request handler." + withMVar masterReqThreadId $ flip throwTo GracefulExit + -- cleanup zmq + debug "Closing down zmq." + withMVar zmqSocket $ \sock -> do + unbind sock zmqAddr + close sock + term zmqContext + -- cleanup local state + closeAcidState localState + +-- | Update on master site. +scheduleMasterUpdate :: (UpdateEvent event, Typeable (EventState event)) => MasterState (EventState event) -> event -> IO (MVar (EventResult event)) +scheduleMasterUpdate masterState@MasterState{..} event = do + debug "Update by Master." + unlocked <- isEmptyMVar masterStateLock + if not unlocked then error "State is locked!" + else do + result <- newEmptyMVar + let callback = if repRedundancy > 1 + then + -- the returned action fills in result when executed later + scheduleLocalUpdate' (downcast localState) event result + else do + hd <- scheduleUpdate localState event + void $ forkIO (putMVar result =<< takeMVar hd) + return (return ()) -- bogus finalizer + let encoded = runPutLazy (safePut event) + queueRepItem masterState (RIUpdate (methodTag event, encoded) (Left callback)) + return result + +-- | Cold Update on master site. +scheduleMasterColdUpdate :: Typeable st => MasterState st -> Tagged ByteString -> IO (MVar ByteString) +scheduleMasterColdUpdate masterState@MasterState{..} encoded = do + debug "Cold Update by Master." + unlocked <- isEmptyMVar masterStateLock + if not unlocked then error "State is locked!" + else do + result <- newEmptyMVar + let callback = if repRedundancy > 1 + then + -- the returned action fills in result when executed later + scheduleLocalColdUpdate' (downcast localState) encoded result + else do + hd <- scheduleColdUpdate localState encoded + void $ forkIO (putMVar result =<< takeMVar hd) + return (return ()) -- bogus finalizer + queueRepItem masterState (RIUpdate encoded (Left callback)) + return result + +-- | Queue an RepItem (originating from the Master itself of an Slave via zmq) +queueRepItem :: MasterState st -> ReplicationItem -> IO () +queueRepItem MasterState{..} = writeChan masterReplicationChan + +-- | The local replication handler. Takes care to run Updates locally. +masterReplicationHandlerL :: (Typeable st) => MasterState st -> IO () +masterReplicationHandlerL MasterState{..} = do + mtid <- myThreadId + putMVar masterRepLThreadId mtid + let loop = handle (\e -> throwTo masterParentThreadId (e :: SomeException)) $ do + debug "Replicating next item locally." + repItem <- readChan masterReplicationChan + case repItem of + RIEnd -> return () + RIArchive -> do + debug "Archive on master." + createArchive localState + loop + RICheckpoint -> do + debug "Checkpoint on master." + createCheckpoint localState + loop + RIUpdate event sink -> do + if repRedundancy > 1 then do + (rev, act) <- modifyMVar masterRevision $ \r -> do + a <- case sink of + Left callback -> callback + _ -> newEmptyMVar >>= scheduleLocalColdUpdate' (downcast localState) event + return (r+1,(r+1,a)) + -- act finalizes the transaction - will be run after full replication + modifyMVar_ repFinalizers $ return . IM.insert rev act + else + modifyMVar_ masterRevision $ \r -> do + case sink of + Left callback -> void callback + _ -> void $ scheduleColdUpdate localState event + return (r+1) + loop + loop + -- signal that we're done + void $ takeMVar masterRepLThreadId + +-- | The network replication handler. Takes care to run Updates on Slaves. +masterReplicationHandlerN :: MasterState st -> IO () +masterReplicationHandlerN MasterState{..} = do + mtid <- myThreadId + putMVar masterRepNThreadId mtid + let loop = handle (\e -> throwTo masterParentThreadId (e :: SomeException)) $ do + debug "Replicating next item in network." + repItem <- readChan masterReplicationChanN + case repItem of + RIEnd -> return () + RIArchive -> do + withMVar nodeStatus $ \ns -> do + debug "Sending archive request to Slaves." + withMVar masterRevisionN $ \mr -> + forM_ (M.keys ns) $ sendArchive zmqSocket mr + loop + RICheckpoint -> do + withMVar nodeStatus $ \ns -> do + debug "Sending Checkpoint Request to Slaves." + withMVar masterRevisionN $ \mr -> + forM_ (M.keys ns) $ sendCheckpoint zmqSocket mr + loop + RIUpdate event sink -> do + withMVar nodeStatus $ \ns -> do + debug $ "Sending Update to Slaves, there are " ++ show (M.size ns) + modifyMVar_ masterRevisionN $ \mrOld -> do + let mr = mrOld + 1 + case sink of + Left _ -> forM_ (M.keys $ M.filter ( do + let noReqSlaves = filter (/= reqNodeIdent) $ M.keys $ M.filter ( IO () +createMasterCheckpoint masterState@MasterState{..} = do + debug "Checkpoint." + unlocked <- isEmptyMVar masterStateLock + unless unlocked $ error "State is locked." + queueRepItem masterState RICheckpoint + +-- | Create an archive on all nodes. +-- Usually createArchive (local to each node) is appropriate. +-- Also take care: Nodes that are not connected at the time, will not create +-- an archive (on reconnect). +createArchiveGlobally :: (IsAcidic st, Typeable st) => AcidState st -> IO () +createArchiveGlobally acid = do + debug "Archive globally." + let masterState = downcast acid + queueRepItem masterState RIArchive + + +toAcidState :: (IsAcidic st, Typeable st) => MasterState st -> AcidState st +toAcidState master + = AcidState { _scheduleUpdate = scheduleMasterUpdate master + , scheduleColdUpdate = scheduleMasterColdUpdate master + , _query = query $ localState master + , queryCold = queryCold $ localState master + , createCheckpoint = createMasterCheckpoint master + , createArchive = createArchive $ localState master + , closeAcidState = closeMasterState master + , acidSubState = mkAnyState master + } + diff --git a/src/Data/Acid/Centered/Slave.hs b/src/Data/Acid/Centered/Slave.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/Centered/Slave.hs @@ -0,0 +1,482 @@ +{-# LANGUAGE DeriveDataTypeable, RecordWildCards, FlexibleContexts #-} +-------------------------------------------------------------------------------- +{- | + Module : Data.Acid.CenteredSlave.hs + Copyright : MIT + + Maintainer : max.voit+hdv@with-eyes.net + Portability : non-portable (uses GHC extensions) + + The Slave part of a the Centered replication backend for acid state. + +-} + +-------------------------------------------------------------------------------- +-- SLAVE part + +module Data.Acid.Centered.Slave + ( + enslaveState + , enslaveStateFrom + , enslaveRedState + , enslaveRedStateFrom + , SlaveState(..) + ) where + +import Data.Typeable +import Data.SafeCopy +import Data.Serialize (decode, encode, runPutLazy, runGetLazy) + +import Data.Acid +import Data.Acid.Core +import Data.Acid.Abstract +import Data.Acid.Local +import Data.Acid.Log + +import Data.Acid.Centered.Common + +import Control.Concurrent (forkIO, ThreadId, myThreadId, killThread, threadDelay, forkIOWithUnmask) +import Control.Concurrent.MVar (MVar, newMVar, newEmptyMVar, isEmptyMVar, + withMVar, modifyMVar, modifyMVar_, + takeMVar, putMVar, tryPutMVar) +import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan) +import Control.Concurrent.STM.TVar (readTVar, writeTVar) +import Data.IORef (writeIORef) +import qualified Control.Concurrent.Event as Event + +import Control.Monad.STM (atomically) +import Control.Monad (void, when, unless) +import Control.Exception (handle, throwTo, SomeException, ErrorCall(..)) + +import System.ZMQ4 (Context, Socket, Dealer(..), + setReceiveHighWM, setSendHighWM, setLinger, restrict, + poll, Poll(..), Event(..), + context, term, socket, close, + connect, disconnect, send, receive) +import System.FilePath ( () ) + +import Data.ByteString.Lazy.Char8 (ByteString) + +import Data.IntMap (IntMap) +import qualified Data.IntMap as IM + +-------------------------------------------------------------------------------- + +-- | Slave state structure, for internal use. +data SlaveState st + = SlaveState { slaveLocalState :: AcidState st + , slaveStateIsRed :: Bool + , slaveStateLock :: MVar () + , slaveRepFinalizers :: MVar (IntMap (IO ())) + , slaveRepChan :: Chan SlaveRepItem + , slaveSyncDone :: Event.Event + , slaveRevision :: MVar NodeRevision + , slaveRequests :: MVar SlaveRequests + , slaveLastRequestID :: MVar RequestID + , slaveRepThreadId :: MVar ThreadId + , slaveReqThreadId :: MVar ThreadId + , slaveParentThreadId :: ThreadId + , slaveZmqContext :: Context + , slaveZmqAddr :: String + , slaveZmqSocket :: MVar (Socket Dealer) + } deriving (Typeable) + +-- | Memory of own Requests sent to Master. +type SlaveRequests = IntMap (IO (IO ()),ThreadId) + +-- | One Update + Metainformation to replicate. +data SlaveRepItem = + SRIEnd + | SRICheckpoint Revision + | SRIArchive Revision + | SRIUpdate Revision (Maybe RequestID) (Tagged ByteString) + +-- | Open a local State as Slave for a Master. +-- +-- The directory for the local state files is the default one ("state/[typeOf +-- state]"). +enslaveState :: (IsAcidic st, Typeable st) => + String -- ^ hostname of the Master + -> PortNumber -- ^ port to connect to + -> st -- ^ initial state + -> IO (AcidState st) +enslaveState address port initialState = + enslaveStateFrom ("state" show (typeOf initialState)) address port initialState + +-- | Open a local State as Slave for a Master. +-- +-- The directory for the local state files is the default one ("state/[typeOf +-- state]"). +enslaveRedState :: (IsAcidic st, Typeable st) => + String -- ^ hostname of the Master + -> PortNumber -- ^ port to connect to + -> st -- ^ initial state + -> IO (AcidState st) +enslaveRedState address port initialState = + enslaveRedStateFrom ("state" show (typeOf initialState)) address port initialState + +-- | Open a local State as Slave for a Master. The directory of the local state +-- files can be specified. +enslaveStateFrom :: (IsAcidic st, Typeable st) => + FilePath -- ^ location of the local state files. + -> String -- ^ hostname of the Master + -> PortNumber -- ^ port to connect to + -> st -- ^ initial state + -> IO (AcidState st) +enslaveStateFrom = enslaveMayRedStateFrom False + +-- | Open a local State as Slave for a _redundant_ Master. The directory of the local state +-- files can be specified. +enslaveRedStateFrom :: (IsAcidic st, Typeable st) => + FilePath -- ^ location of the local state files. + -> String -- ^ hostname of the Master + -> PortNumber -- ^ port to connect to + -> st -- ^ initial state + -> IO (AcidState st) +enslaveRedStateFrom = enslaveMayRedStateFrom True + +-- | Open a local State as Slave for a Master, redundant or not. +-- The directory of the local state files can be specified. +enslaveMayRedStateFrom :: (IsAcidic st, Typeable st) => + Bool -- ^ is redundant + -> FilePath -- ^ location of the local state files. + -> String -- ^ hostname of the Master + -> PortNumber -- ^ port to connect to + -> st -- ^ initial state + -> IO (AcidState st) +enslaveMayRedStateFrom isRed directory address port initialState = do + -- local + lst <- openLocalStateFrom directory initialState + lrev <- getLocalRevision lst + rev <- newMVar lrev + debug $ "Opening enslaved state at revision " ++ show lrev + srs <- newMVar IM.empty + lastReqId <- newMVar 0 + repChan <- newChan + syncDone <- Event.new + reqTid <- newEmptyMVar + repTid <- newEmptyMVar + parTid <- myThreadId + repFin <- newMVar IM.empty + sLock <- newEmptyMVar + -- remote + let addr = "tcp://" ++ address ++ ":" ++ show port + ctx <- context + sock <- socket ctx Dealer + setReceiveHighWM (restrict (100*1000 :: Int)) sock + setSendHighWM (restrict (100*1000 :: Int)) sock + connect sock addr + msock <- newMVar sock + sendToMaster msock $ NewSlave lrev + + let slaveState = SlaveState { slaveLocalState = lst + , slaveStateIsRed = isRed + , slaveStateLock = sLock + , slaveRepFinalizers = repFin + , slaveRepChan = repChan + , slaveSyncDone = syncDone + , slaveRevision = rev + , slaveRequests = srs + , slaveLastRequestID = lastReqId + , slaveReqThreadId = reqTid + , slaveRepThreadId = repTid + , slaveParentThreadId = parTid + , slaveZmqContext = ctx + , slaveZmqAddr = addr + , slaveZmqSocket = msock + } + void $ forkIOWithUnmask $ slaveRequestHandler slaveState + void $ forkIO $ slaveReplicationHandler slaveState + return $ slaveToAcidState slaveState + where + getLocalRevision = + atomically . readTVar . logNextEntryId . localEvents . downcast + +-- | Replication handler of the Slave. +slaveRequestHandler :: (IsAcidic st, Typeable st) => SlaveState st -> (IO () -> IO ()) -> IO () +slaveRequestHandler slaveState@SlaveState{..} unmask = do + mtid <- myThreadId + putMVar slaveReqThreadId mtid + let loop = handle (\e -> throwTo slaveParentThreadId (e :: SomeException)) $ + unmask $ handle killHandler $ do + --waitRead =<< readMVar slaveZmqSocket + -- FIXME: we needn't poll if not for strange zmq behaviour + re <- withMVar slaveZmqSocket $ \sock -> poll 100 [Sock sock [In] Nothing] + unless (null $ head re) $ do + msg <- withMVar slaveZmqSocket receive + case decode msg of + Left str -> error $ "Data.Serialize.decode failed on MasterMessage: " ++ show str + Right mmsg -> handleMessage mmsg + loop + loop + where + killHandler :: AcidException -> IO () + killHandler GracefulExit = return () + handleMessage m = do + debug $ "Received: " ++ show m + case m of + -- We are sent an Update to replicate. + DoRep r i d -> queueRepItem slaveState (SRIUpdate r i d) + -- We are sent a Checkpoint for synchronization. + DoSyncCheckpoint r d -> replicateSyncCp slaveState r d + -- We are sent an Update to replicate for synchronization. + DoSyncRep r d -> replicateSyncUpdate slaveState r d + -- Master done sending all synchronization Updates. + SyncDone c -> onSyncDone slaveState c + -- We are sent a Checkpoint request. + DoCheckpoint r -> queueRepItem slaveState (SRICheckpoint r) + -- We are sent an Archive request. + DoArchive r -> queueRepItem slaveState (SRIArchive r) + -- Full replication of a revision + FullRep r -> modifyMVar_ slaveRepFinalizers $ \rf -> do + rf IM.! r + return $ IM.delete r rf + -- Full replication of events up to revision + FullRepTo r -> modifyMVar_ slaveRepFinalizers $ \rf -> do + let (ef, nrf) = IM.partitionWithKey (\k _ -> k <= r) rf + sequence_ (IM.elems ef) + return nrf + -- We are allowed to Quit. + MayQuit -> writeChan slaveRepChan SRIEnd + -- We are requested to Quit - shall be handled by + -- 'bracket' usage by user. + MasterQuit -> throwTo slaveParentThreadId $ + ErrorCall "Data.Acid.Centered.Slave: Master quit." + -- no other messages possible, enforced by type checker + +-- | After sync check CRC +onSyncDone :: (IsAcidic st, Typeable st) => SlaveState st -> Crc -> IO () +onSyncDone SlaveState{..} crc = do + localCrc <- crcOfState slaveLocalState + if crc /= localCrc then + error "Data.Acid.Centered.Slave: CRC mismatch after sync." + else do + debug "Sync Done, CRC fine." + Event.set slaveSyncDone + +-- | Queue Updates into Chan for replication. +-- We use the Chan so Sync-Updates and normal ones can be interleaved. +queueRepItem :: SlaveState st -> SlaveRepItem -> IO () +queueRepItem SlaveState{..} repItem = do + debug "Queuing RepItem." + writeChan slaveRepChan repItem + +-- | Replicates content of Chan. +slaveReplicationHandler :: Typeable st => SlaveState st -> IO () +slaveReplicationHandler slaveState@SlaveState{..} = do + mtid <- myThreadId + putMVar slaveRepThreadId mtid + + -- todo: timeout is magic variable, make customizable + noTimeout <- Event.waitTimeout slaveSyncDone $ 10*1000*1000 + unless noTimeout $ throwTo slaveParentThreadId $ + ErrorCall "Data.Acid.Centered.Slave: Took too long to sync. Timeout." + + let loop = handle (\e -> throwTo slaveParentThreadId (e :: SomeException)) $ do + mayRepItem <- readChan slaveRepChan + case mayRepItem of + SRIEnd -> return () + SRICheckpoint r -> repCheckpoint slaveState r >> loop + SRIArchive r -> repArchive slaveState r >> loop + SRIUpdate r i d -> replicateUpdate slaveState r i d False >> loop + loop + + -- signal that we're done + void $ takeMVar slaveRepThreadId + +-- | Replicate Sync-Checkpoints directly. +replicateSyncCp :: (IsAcidic st, Typeable st) => + SlaveState st -> Revision -> ByteString -> IO () +replicateSyncCp SlaveState{..} rev encoded = do + st <- decodeCheckpoint encoded + let lst = downcast slaveLocalState + let core = localCore lst + modifyMVar_ slaveRevision $ \sr -> do + when (sr > rev) $ error "Data.Acid.Centered.Slave: Revision mismatch for checkpoint: Slave is newer." + modifyCoreState_ core $ \_ -> do + writeIORef (localCopy lst) st + createCpFake lst encoded rev + adjustEventLogId lst rev + return st + return rev + where + adjustEventLogId l r = do + atomically $ writeTVar (logNextEntryId (localEvents l)) r + void $ cutFileLog (localEvents l) + createCpFake l e r = do + mvar <- newEmptyMVar + pushAction (localEvents l) $ + pushEntry (localCheckpoints l) (Checkpoint r e) (putMVar mvar ()) + takeMVar mvar + decodeCheckpoint e = + case runGetLazy safeGet e of + Left msg -> error $ "Data.Acid.Centered.Slave: Checkpoint could not be decoded: " ++ msg + Right val -> return val + +-- | Replicate Sync-Updates directly. +replicateSyncUpdate :: Typeable st => SlaveState st -> Revision -> Tagged ByteString -> IO () +replicateSyncUpdate slaveState rev event = replicateUpdate slaveState rev Nothing event True + +-- | Replicate an Update as requested by Master. +-- Updates that were requested by this Slave are run locally and the result +-- put into the MVar in SlaveRequests. +-- Other Updates are just replicated without using the result. +replicateUpdate :: Typeable st => SlaveState st -> Revision -> Maybe RequestID -> Tagged ByteString -> Bool -> IO () +replicateUpdate SlaveState{..} rev reqId event syncing = do + debug $ "Got an Update to replicate " ++ show rev + modifyMVar_ slaveRevision $ \nr -> if rev - 1 == nr + then do + -- commit / run it locally + case reqId of + Nothing -> replicateForeign + Just rid -> replicateOwn rid + -- send reply: we're done + unless syncing $ sendToMaster slaveZmqSocket $ RepDone rev + return rev + else do + sendToMaster slaveZmqSocket RepError + void $ error $ + "Data.Acid.Centered.Slave: Replication failed at revision " + ++ show nr ++ " -> " ++ show rev + return nr + where + replicateForeign = + if slaveStateIsRed then do + act <- newEmptyMVar >>= scheduleLocalColdUpdate' (downcast slaveLocalState) event + modifyMVar_ slaveRepFinalizers $ return . IM.insert rev act + else + void $ scheduleColdUpdate slaveLocalState event + replicateOwn rid = do + act <- modifyMVar slaveRequests $ \srs -> do + debug $ "This is the Update for Request " ++ show rid + let (icallback, timeoutId) = srs IM.! rid + callback <- icallback + killThread timeoutId + let nsrs = IM.delete rid srs + return (nsrs, callback) + when slaveStateIsRed $ + modifyMVar_ slaveRepFinalizers $ return . IM.insert rev act + +repCheckpoint :: SlaveState st -> Revision -> IO () +repCheckpoint SlaveState{..} rev = do + debug $ "Got Checkpoint request at revision: " ++ show rev + withMVar slaveRevision $ \_ -> + -- create checkpoint + createCheckpoint slaveLocalState + +repArchive :: SlaveState st -> Revision -> IO () +repArchive SlaveState{..} rev = do + debug $ "Got Archive request at revision: " ++ show rev + withMVar slaveRevision $ \_ -> + createArchive slaveLocalState + + +-- | Update on slave site. +-- The steps are: +-- - Request Update from Master +-- - Master issues Update with same RequestID +-- - repHandler replicates and puts result in MVar +scheduleSlaveUpdate :: (UpdateEvent e, Typeable (EventState e)) => SlaveState (EventState e) -> e -> IO (MVar (EventResult e)) +scheduleSlaveUpdate slaveState@SlaveState{..} event = do + unlocked <- isEmptyMVar slaveStateLock + if not unlocked then error "State is locked." + else do + debug "Update by Slave." + result <- newEmptyMVar + reqId <- getNextRequestId slaveState + modifyMVar_ slaveRequests $ \srs -> do + let encoded = runPutLazy (safePut event) + sendToMaster slaveZmqSocket $ ReqUpdate reqId (methodTag event, encoded) + timeoutID <- forkIO $ timeoutRequest slaveState reqId result + let callback = if slaveStateIsRed + then scheduleLocalUpdate' (downcast slaveLocalState) event result + else do + hd <- scheduleUpdate slaveLocalState event + void $ forkIO $ putMVar result =<< takeMVar hd + return (return ()) -- bogus finalizer + return $ IM.insert reqId (callback, timeoutID) srs + return result + +-- | Cold Update on slave site. This enables for using Remote. +scheduleSlaveColdUpdate :: Typeable st => SlaveState st -> Tagged ByteString -> IO (MVar ByteString) +scheduleSlaveColdUpdate slaveState@SlaveState{..} encoded = do + unlocked <- isEmptyMVar slaveStateLock + if not unlocked then error "State is locked." + else do + debug "Cold Update by Slave." + result <- newEmptyMVar + -- slaveLastRequestID is only modified here - and used for locking the state + reqId <- getNextRequestId slaveState + modifyMVar_ slaveRequests $ \srs -> do + sendToMaster slaveZmqSocket $ ReqUpdate reqId encoded + timeoutID <- forkIO $ timeoutRequest slaveState reqId result + let callback = if slaveStateIsRed + then scheduleLocalColdUpdate' (downcast slaveLocalState) encoded result + else do + hd <- scheduleColdUpdate slaveLocalState encoded + void $ forkIO $ putMVar result =<< takeMVar hd + return (return ()) -- bogus finalizer + return $ IM.insert reqId (callback, timeoutID) srs + return result + +-- | Generate ID for another request. +getNextRequestId :: SlaveState st -> IO RequestID +getNextRequestId SlaveState{..} = modifyMVar slaveLastRequestID $ \x -> return (x+1,x+1) + +-- | Ensures requests are actually answered or fail. +-- On timeout the Slave dies, not the thread that invoked the Update. +timeoutRequest :: SlaveState st -> RequestID -> MVar m -> IO () +timeoutRequest SlaveState{..} reqId mvar = do + threadDelay $ 5*1000*1000 + stillThere <- withMVar slaveRequests (return . IM.member reqId) + when stillThere $ do + putMVar mvar $ error "Data.Acid.Centered.Slave: Update-Request timed out." + throwTo slaveParentThreadId $ ErrorCall "Data.Acid.Centered.Slave: Update-Request timed out." + +-- | Send a message to Master. +sendToMaster :: MVar (Socket Dealer) -> SlaveMessage -> IO () +sendToMaster msock smsg = withMVar msock $ \sock -> send sock [] (encode smsg) + +-- | Close an enslaved State. +liberateState :: SlaveState st -> IO () +liberateState SlaveState{..} = + -- lock state against updates: disallow requests + whenM (tryPutMVar slaveStateLock ()) $ do + debug "Closing Slave state..." + -- check / wait unprocessed requests + debug "Waiting for Requests to finish." + waitPoll 100 (withMVar slaveRequests (return . IM.null)) + -- send master quit message + sendToMaster slaveZmqSocket SlaveQuit + -- wait replication chan, only if sync done + syncDone <- Event.isSet slaveSyncDone + when syncDone $ do + debug "Waiting for repChan to empty." + mtid <- myThreadId + putMVar slaveRepThreadId mtid + -- kill handler threads + debug "Killing request handler." + withMVar slaveReqThreadId $ flip throwTo GracefulExit + -- cleanup zmq + debug "Closing down zmq." + withMVar slaveZmqSocket $ \s -> do + -- avoid the socket hanging around + setLinger (restrict (1000 :: Int)) s + disconnect s slaveZmqAddr + close s + term slaveZmqContext + -- cleanup local state + debug "Closing local state." + closeAcidState slaveLocalState + +slaveToAcidState :: (IsAcidic st, Typeable st) => SlaveState st -> AcidState st +slaveToAcidState slaveState + = AcidState { _scheduleUpdate = scheduleSlaveUpdate slaveState + , scheduleColdUpdate = scheduleSlaveColdUpdate slaveState + , _query = query $ slaveLocalState slaveState + , queryCold = queryCold $ slaveLocalState slaveState + , createCheckpoint = createCheckpoint $ slaveLocalState slaveState + , createArchive = createArchive $ slaveLocalState slaveState + , closeAcidState = liberateState slaveState + , acidSubState = mkAnyState slaveState + } diff --git a/test/CRCFail.hs b/test/CRCFail.hs new file mode 100644 --- /dev/null +++ b/test/CRCFail.hs @@ -0,0 +1,51 @@ +{-# LANGUAGE TypeFamilies #-} + +import Data.Acid +import Data.Acid.Centered + +import Control.Monad (void, when) +import Control.Concurrent (threadDelay, forkIO) +import Control.Concurrent.MVar +import System.Exit (exitSuccess, exitFailure) +import System.Directory (doesDirectoryExist, removeDirectoryRecursive) + +import Control.Exception (handle, SomeException) + +-- state structures +import IntCommon + +-- helpers +delaySec :: Int -> IO () +delaySec n = threadDelay $ n*1000*1000 + +cleanup :: FilePath -> IO () +cleanup path = do + sp <- doesDirectoryExist path + when sp $ removeDirectoryRecursive path + +-- actual test +slave :: MVar () -> IO () +slave mBlock = handle eHandler $ do + acid <- enslaveStateFrom "state/CRCFail/s1" "localhost" 3333 (IntState 23) + -- at this point happens the crc fail - we check for an exception and + -- thereby determine whether the test was successful + delaySec 2 + -- this should never be executed + closeAcidState acid + exitFailure + where + eHandler :: SomeException -> IO () + eHandler e = when (show e == "Data.Acid.Centered.Slave: CRC mismatch after sync.") $ do + putStrLn "CRC mismatch, fine." + putMVar mBlock () + +main :: IO () +main = do + cleanup "state/CRCFail" + acid <- openMasterStateFrom "state/CRCFail/m" "127.0.0.1" 3333 (IntState 0) + mBlock <- newEmptyMVar + void $ forkIO $ slave mBlock + takeMVar mBlock + closeAcidState acid + exitSuccess + diff --git a/test/CheckpointSync.hs b/test/CheckpointSync.hs new file mode 100644 --- /dev/null +++ b/test/CheckpointSync.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE TypeFamilies #-} + +import Data.Acid +import Data.Acid.Centered + +import Control.Monad (void, when) +import Control.Concurrent (threadDelay, forkIO) +import System.Exit (exitSuccess, exitFailure) +import System.Directory (doesDirectoryExist, removeDirectoryRecursive) + +-- state structures +import IntCommon + +-- helpers +delaySec :: Int -> IO () +delaySec n = threadDelay $ n*1000*1000 + +cleanup :: FilePath -> IO () +cleanup path = do + sp <- doesDirectoryExist path + when sp $ removeDirectoryRecursive path + +-- actual test +slave :: IO () +slave = do + acid <- enslaveStateFrom "state/CheckpointSync/s1" "localhost" 3333 (IntState 23) + delaySec 5 + val <- query acid GetState + closeAcidState acid + when (val /= 1) $ putStrLn "Slave hasn't got value." >> exitFailure + +main :: IO () +main = do + cleanup "state/CheckpointSync" + acid <- openMasterStateFrom "state/CheckpointSync/m" "127.0.0.1" 3333 (IntState 0) + update acid IncrementState + createCheckpoint acid + void $ forkIO slave + delaySec 10 + closeAcidState acid + exitSuccess + diff --git a/test/NReplication.hs b/test/NReplication.hs new file mode 100644 --- /dev/null +++ b/test/NReplication.hs @@ -0,0 +1,47 @@ +{-# LANGUAGE TypeFamilies #-} + +import Data.Acid +import Data.Acid.Centered + +import Control.Monad (when, replicateM, void) +import Control.Concurrent (threadDelay, forkIO) +import System.Exit (exitSuccess, exitFailure) +import System.Directory (doesDirectoryExist, removeDirectoryRecursive) + +-- state structures +import IntCommon + +-- helpers +delaySec :: Int -> IO () +delaySec n = threadDelay $ n*1000*1000 + +cleanup :: FilePath -> IO () +cleanup path = do + sp <- doesDirectoryExist path + when sp $ removeDirectoryRecursive path + +-- actual test +slave :: IO () +slave = do + acid <- enslaveRedStateFrom "state/NReplication/s1" "localhost" 3333 (IntState 0) + delaySec 2 + val <- query acid GetState + closeAcidState acid + when (val /= 23) $ putStrLn "Slave hasn't got value." >> exitFailure + +main :: IO () +main = do + cleanup "state/NReplication" + acid <- openRedMasterStateFrom "state/NReplication/m" "127.0.0.1" 3333 2 (IntState 0) + void $ forkIO $ delaySec 2 >> slave + void $ forkIO $ update acid (SetState 23) -- this update blocks + vals <- replicateM 5 $ do + delaySec 1 + query acid GetState + closeAcidState acid + -- queries before the slave joined yield the old state, only after joining + -- the update is accepted + if head vals == 0 && last vals == 23 + then exitSuccess + else exitFailure + diff --git a/test/OrderingRandom.hs b/test/OrderingRandom.hs new file mode 100644 --- /dev/null +++ b/test/OrderingRandom.hs @@ -0,0 +1,85 @@ +{-# LANGUAGE TypeFamilies #-} + +import Data.Acid +import Data.Acid.Centered + +import Control.Monad (when, forM_) +import Control.Concurrent (forkIO,threadDelay) +import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar) +import System.Exit (exitSuccess, exitFailure) +import System.Directory (doesDirectoryExist, removeDirectoryRecursive) +import System.Random (mkStdGen, randomRs) + +-- state structures +import NcCommon + +-- helpers +delaySec :: Int -> IO () +delaySec n = threadDelay $ n*1000*1000 + +cleanup :: FilePath -> IO () +cleanup path = do + sp <- doesDirectoryExist path + when sp $ removeDirectoryRecursive path + +-- actual test +randRange :: (Int,Int) +randRange = (100,100000) + +numRands :: Int +numRands = 100 + +slave :: Int -> MVar [Int] -> MVar () -> MVar () -> IO () +slave ident res done alldone = do + let rs = randomRs randRange $ mkStdGen ident :: [Int] + acid <- enslaveStateFrom ("state/OrderingRandom/s" ++ show ident) "localhost" 3333 (NcState []) + forM_ (take numRands rs) $ \r -> do + threadDelay r + update acid $ NcOpState r + putMVar done () + -- wait for others + _ <- readMVar alldone + delaySec ident + val <- query acid GetState + putMVar res val + print $ "slave quit ident " ++ show ident + closeAcidState acid + +main :: IO () +main = do + cleanup "state/OrderingRandom" + acid <- openMasterStateFrom "state/OrderingRandom/m" "127.0.0.1" 3333 (NcState []) + allDone <- newEmptyMVar + -- start slaves + s1Res <- newEmptyMVar + s1Done <- newEmptyMVar + _ <- forkIO $ slave 1 s1Res s1Done allDone + threadDelay 1000 -- zmq-indentity could be the same if too fast + s2Res <- newEmptyMVar + s2Done <- newEmptyMVar + _ <- forkIO $ slave 2 s2Res s2Done allDone + -- manipulate state on master + let rs = randomRs randRange $ mkStdGen 23 :: [Int] + forM_ (take numRands rs) $ \r -> do + threadDelay r + update acid $ NcOpState r + -- wait for slaves + print "at wait" + _ <- takeMVar s1Done + _ <- takeMVar s2Done + -- signal slaves done + print "at done" + putMVar allDone () + -- collect results + print "at collect" + vs1 <- takeMVar s1Res + vs2 <- takeMVar s2Res + vm <- query acid GetState + -- check results + print "at results" + when (vs1 /= vs2) exitFailure + when (vm /= vs1) exitFailure + when (vm /= vs2) exitFailure + closeAcidState acid + exitSuccess + diff --git a/test/Simple.hs b/test/Simple.hs new file mode 100644 --- /dev/null +++ b/test/Simple.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE TypeFamilies #-} + +import Data.Acid +import Data.Acid.Centered + +import Control.Monad (when) +import Control.Concurrent (threadDelay) +import System.Exit (exitSuccess, exitFailure) +import System.Directory (doesDirectoryExist, removeDirectoryRecursive) + +-- state structures +import IntCommon + +-- helpers +delaySec :: Int -> IO () +delaySec n = threadDelay $ n*1000*1000 + +cleanup :: FilePath -> IO () +cleanup path = do + sp <- doesDirectoryExist path + when sp $ removeDirectoryRecursive path + +-- actual test +slave :: IO () +slave = do + acid <- enslaveStateFrom "state/Simple/s1" "localhost" 3333 (IntState 0) + delaySec 5 + val <- query acid GetState + closeAcidState acid + when (val /= 23) $ putStrLn "Slave hasn't got value." >> exitFailure + +main :: IO () +main = do + cleanup "state/Simple" + acid <- openMasterStateFrom "state/Simple/m" "127.0.0.1" 3333 (IntState 0) + update acid (SetState 23) + slave + closeAcidState acid + exitSuccess + diff --git a/test/SlaveUpdates.hs b/test/SlaveUpdates.hs new file mode 100644 --- /dev/null +++ b/test/SlaveUpdates.hs @@ -0,0 +1,52 @@ +{-# LANGUAGE TypeFamilies #-} + +import Data.Acid +import Data.Acid.Centered + +import Control.Monad (void, when) +import Control.Concurrent (forkIO, threadDelay) +import System.Exit (exitSuccess, exitFailure) +import System.Directory (doesDirectoryExist, removeDirectoryRecursive) + +-- state structures +import IntCommon + +-- helpers +delaySec :: Int -> IO () +delaySec n = threadDelay $ n*1000*1000 + +cleanup :: FilePath -> IO () +cleanup path = do + sp <- doesDirectoryExist path + when sp $ removeDirectoryRecursive path + +-- actual test +slave1 :: IO () +slave1 = do + acid <- enslaveStateFrom "state/SlaveUpdates/s1" "localhost" 3333 (IntState 0) + update acid (SetState 23) + val <- query acid GetState + closeAcidState acid + when (val /= 23) $ putStrLn "Slave 1 hasn't got value." >> exitFailure + +slave2 :: IO () +slave2 = do + acid <- enslaveStateFrom "state/SlaveUpdates/s2" "localhost" 3333 (IntState 0) + delaySec 5 + val <- query acid GetState + closeAcidState acid + when (val /= 23) $ putStrLn "Slave 2 hasn't got value." >> exitFailure + +main :: IO () +main = do + cleanup "state/SlaveUpdates" + + acid <- openMasterStateFrom "state/SlaveUpdates/m" "127.0.0.1" 3333 (IntState 0) + void $ forkIO slave1 + slave2 + val <- query acid GetState + closeAcidState acid + when (val /= 23) $ putStrLn "Master hasn't got value." >> exitFailure + + exitSuccess + diff --git a/test/SyncTimeout.hs b/test/SyncTimeout.hs new file mode 100644 --- /dev/null +++ b/test/SyncTimeout.hs @@ -0,0 +1,38 @@ +{-# LANGUAGE TypeFamilies, ScopedTypeVariables #-} + +import Data.Acid +import Data.Acid.Centered + +import Control.Monad (when) +import Control.Concurrent (threadDelay) +import System.Exit (exitSuccess, exitFailure) +import System.Directory (doesDirectoryExist, removeDirectoryRecursive) +import Control.Exception (catch, SomeException) + +-- state structures +import IntCommon + +-- helpers +delaySec :: Int -> IO () +delaySec n = threadDelay $ n*1000*1000 + +cleanup :: FilePath -> IO () +cleanup path = do + sp <- doesDirectoryExist path + when sp $ removeDirectoryRecursive path + +-- actual test +slave :: IO () +slave = do + acid <- enslaveStateFrom "state/SyncTimeout/s1" "localhost" 3333 (IntState 0) + delaySec 11 -- SyncTimeout happens at 10 seconds + closeAcidState acid + +main :: IO () +main = do + cleanup "state/SyncTimeout" + catch slave $ \(e :: SomeException) -> + if show e == "Data.Acid.Centered.Slave: Took too long to sync. Timeout." + then exitSuccess + else exitFailure + diff --git a/test/UpdateError.hs b/test/UpdateError.hs new file mode 100644 --- /dev/null +++ b/test/UpdateError.hs @@ -0,0 +1,51 @@ +{-# LANGUAGE TypeFamilies #-} + +import Data.Acid +import Data.Acid.Centered + +import Control.Monad (when, void) +import Control.Concurrent (threadDelay, forkIO) +import System.Exit (exitSuccess, exitFailure) +import System.Directory (doesDirectoryExist, removeDirectoryRecursive) +import Control.Exception (finally) + +-- state structures +import IntCommon + +-- helpers +delaySec :: Int -> IO () +delaySec n = threadDelay $ n*1000*1000 + +cleanup :: FilePath -> IO () +cleanup path = do + sp <- doesDirectoryExist path + when sp $ removeDirectoryRecursive path + +-- actual test +slave1 :: IO () +slave1 = do + acid <- enslaveStateFrom "state/UpdateError/s1" "localhost" 3333 (IntState 0) + delaySec 1 + finally -- the update fails; if not it's an error + (update acid (SetState (error "fail s1")) >> exitFailure) + (closeAcidState acid) + +slave2 :: IO () +slave2 = do + acid <- enslaveStateFrom "state/UpdateError/s2" "localhost" 3333 (IntState 0) + delaySec 1 + finally -- the update fails; if not it's an error + (update acid (error "fail s2" :: IncrementState) >> exitFailure) + (closeAcidState acid) + +main :: IO () +main = do + cleanup "state/UpdateError" + acid <- openMasterStateFrom "state/UpdateError/m" "127.0.0.1" 3333 (IntState 0) + update acid (SetState 23) + void $ forkIO slave1 + void $ forkIO slave2 + delaySec 2 + closeAcidState acid + exitSuccess +