acid-state-dist (empty) → 0.1.0.0
raw patch · 19 files changed
+2057/−0 lines, 19 filesdep +acid-statedep +acid-state-distdep +basesetup-changed
Dependencies added: acid-state, acid-state-dist, base, bytestring, cereal, concurrent-extra, containers, criterion, directory, filepath, mtl, random, safe, safecopy, semigroups, stm, transformers, zeromq4-haskell
Files
- LICENSE +21/−0
- Setup.hs +2/−0
- acid-state-dist.cabal +211/−0
- benchmark/Local.hs +26/−0
- benchmark/MasterOnly.hs +28/−0
- benchmark/MasterSlave.hs +42/−0
- benchmark/Slave.hs +42/−0
- src/Data/Acid/Centered.hs +89/−0
- src/Data/Acid/Centered/Common.hs +182/−0
- src/Data/Acid/Centered/Master.hs +526/−0
- src/Data/Acid/Centered/Slave.hs +482/−0
- test/CRCFail.hs +51/−0
- test/CheckpointSync.hs +42/−0
- test/NReplication.hs +47/−0
- test/OrderingRandom.hs +85/−0
- test/Simple.hs +40/−0
- test/SlaveUpdates.hs +52/−0
- test/SyncTimeout.hs +38/−0
- test/UpdateError.hs +51/−0
+ LICENSE view
@@ -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.+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ acid-state-dist.cabal view
@@ -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: <max.voit+hdv@with-eyes.net>++-- 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
+ benchmark/Local.hs view
@@ -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
+ benchmark/MasterOnly.hs view
@@ -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
+ benchmark/MasterSlave.hs view
@@ -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
+ benchmark/Slave.hs view
@@ -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
+ src/Data/Acid/Centered.hs view
@@ -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+
+ src/Data/Acid/Centered/Common.hs view
@@ -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
+ src/Data/Acid/Centered/Master.hs view
@@ -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 (<mr) ns) $ sendUpdate zmqSocket mr Nothing event+ Right (reqID, reqNodeIdent) -> do+ let noReqSlaves = filter (/= reqNodeIdent) $ M.keys $ M.filter (<mr) ns+ sendUpdate zmqSocket mr (Just reqID) event reqNodeIdent+ forM_ noReqSlaves $ sendUpdate zmqSocket mr Nothing event+ return mr+ loop+ loop+ -- signal that we're done+ void $ takeMVar masterRepNThreadId+ where+ sendUpdate sock revision reqId encoded =+ sendToSlave sock (DoRep revision reqId encoded)+ sendCheckpoint sock revision = sendToSlave sock (DoCheckpoint revision)+ sendArchive sock revision = sendToSlave sock (DoArchive revision)++-- | Create a checkpoint (on all nodes, per request).+-- This is useful for faster resume of both the Master (at startup) and+-- Slaves (at startup and reconnect).+createMasterCheckpoint :: MasterState st -> 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+ }+
+ src/Data/Acid/Centered/Slave.hs view
@@ -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+ }
+ test/CRCFail.hs view
@@ -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+
+ test/CheckpointSync.hs view
@@ -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+
+ test/NReplication.hs view
@@ -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+
+ test/OrderingRandom.hs view
@@ -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+
+ test/Simple.hs view
@@ -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+
+ test/SlaveUpdates.hs view
@@ -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+
+ test/SyncTimeout.hs view
@@ -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+
+ test/UpdateError.hs view
@@ -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+