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.cabal b/acid-state.cabal new file mode 100644 --- /dev/null +++ b/acid-state.cabal @@ -0,0 +1,62 @@ +-- acid-state.cabal auto-generated by cabal init. For additional +-- options, see +-- http://www.haskell.org/cabal/release/cabal-latest/doc/users-guide/authors.html#pkg-descr. +-- The name of the package. +Name: acid-state + +-- The package version. See the Haskell package versioning policy +-- (http://www.haskell.org/haskellwiki/Package_versioning_policy) for +-- standards guiding when and how versions should be incremented. +Version: 0.1 + +-- A short (one-line) description of the package. +Synopsis: Add ACID guarantees to any serializable Haskell data structure. + +-- A longer description of the package. +Description: Use regular Haskell data structures as your database and get stronger ACID guarantees than most RDBMS offer. + +-- URL for the project homepage or repository. +Homepage: http://acid-state.seize.it/ + +-- The license under which the package is released. +License: PublicDomain + +-- The package author(s). +Author: David Himmelstrup + +-- An email address to which users can send suggestions, bug reports, +-- and patches. +Maintainer: Lemmih + +-- 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: examples/*.hs + +-- Constraint on the version of Cabal needed to build this package. +Cabal-version: >=1.6 + + +Library + -- Modules exported by the library. + Exposed-Modules: Data.State.Acid, Data.State.Acid.Local, Data.State.Acid.Core + + -- Modules not exported by this package. + Other-modules: Data.State.Acid.Log, Data.State.Acid.Archive, + Data.State.Acid.CRC + + -- Packages needed in order to build this package. + Build-depends: base >= 4 && < 5, binary, bytestring, stm, filepath, directory, + monads-fd, array, containers + + Hs-Source-Dirs: src/ + + -- Extra tools (e.g. alex, hsc2hs, ...) needed to build the source. + -- Build-tools: + diff --git a/examples/HelloWorld.hs b/examples/HelloWorld.hs new file mode 100644 --- /dev/null +++ b/examples/HelloWorld.hs @@ -0,0 +1,77 @@ +{-# LANGUAGE DeriveDataTypeable, TypeFamilies, StandaloneDeriving #-} +module Main (main) where + +import Data.State.Acid.Core +import Data.State.Acid + +import qualified Control.Monad.State as State +import Control.Monad.Reader +import System.Environment +import Data.Binary + +import Data.Typeable + +------------------------------------------------------ +-- The Haskell structure that we want to encapsulate + +data HelloWorldState = HelloWorldState String + deriving (Show, Typeable) + + +------------------------------------------------------ +-- The transaction we will execute over the state. + +writeState :: String -> Update HelloWorldState () +writeState newValue + = State.put (HelloWorldState newValue) + +queryState :: Query HelloWorldState String +queryState = do HelloWorldState string <- ask + return string + + +------------------------------------------------------ +-- This is how AcidState is used: + +main :: IO () +main = do acid <- mkAcidState myEvents (HelloWorldState "Hello world") + args <- getArgs + if null args + then do string <- query acid QueryState + putStrLn $ "The state is: " ++ string + else do update acid (WriteState (unwords args)) + putStrLn $ "The state has been modified!" + + +------------------------------------------------------ +-- The gritty details. These things may be done with +-- Template Haskell in the future. + +data WriteState = WriteState String +data QueryState = QueryState + + +deriving instance Typeable WriteState +instance Binary WriteState where + put (WriteState st) = put st + get = liftM WriteState get +instance Method WriteState where + type MethodResult WriteState = () +instance UpdateEvent WriteState + +deriving instance Typeable QueryState +instance Binary QueryState where + put QueryState = return () + get = return QueryState +instance Method QueryState where + type MethodResult QueryState = String +instance QueryEvent QueryState + +instance Binary HelloWorldState where + put (HelloWorldState state) = put state + get = liftM HelloWorldState get + +myEvents :: [Event HelloWorldState] +myEvents = [ UpdateEvent (\(WriteState newState) -> writeState newState) + , QueryEvent (\QueryState -> queryState) + ] diff --git a/src/Data/State/Acid.hs b/src/Data/State/Acid.hs new file mode 100644 --- /dev/null +++ b/src/Data/State/Acid.hs @@ -0,0 +1,5 @@ +module Data.State.Acid + ( module Data.State.Acid.Local + ) where + +import Data.State.Acid.Local diff --git a/src/Data/State/Acid/Archive.hs b/src/Data/State/Acid/Archive.hs new file mode 100644 --- /dev/null +++ b/src/Data/State/Acid/Archive.hs @@ -0,0 +1,73 @@ +{- +Format: + |content length| crc16 | content | + |8 bytes | 2 bytes | n bytes | +-} +module Data.State.Acid.Archive + ( Entry + , Entries(..) + , putEntries + , packEntries + , readEntries + , entriesToList + , entriesToListNoFail + ) where + +import Data.State.Acid.CRC + +import qualified Data.ByteString.Lazy as Lazy +import qualified Data.ByteString as Strict +import Data.Binary.Get +import Data.Binary.Builder +import Data.Monoid + +type Entry = Lazy.ByteString +data Entries = Done | Next Entry Entries | Fail String + deriving (Show) + +entriesToList :: Entries -> [Entry] +entriesToList Done = [] +entriesToList (Next entry next) = entry : entriesToList next +entriesToList (Fail msg) = fail msg + +entriesToListNoFail :: Entries -> [Entry] +entriesToListNoFail Done = [] +entriesToListNoFail (Next entry next) = entry : entriesToListNoFail next +entriesToListNoFail Fail{} = [] + +putEntry :: Entry -> Builder +putEntry content + = putWord64le contentLength `mappend` + putWord16le contentHash `mappend` + fromLazyByteString content + where contentLength = fromIntegral $ Lazy.length content + contentHash = crc16 content + +putEntries :: [Entry] -> Builder +putEntries = mconcat . map putEntry + +packEntries :: [Entry] -> Lazy.ByteString +packEntries = toLazyByteString . putEntries + +readEntries :: Lazy.ByteString -> Entries +readEntries bs + | Lazy.null bs + = Done + | Lazy.length header < headerSize + = Fail "Incomplete header." + | Lazy.length content /= fromIntegral contentLength + = Fail "Insuficient content." + | crc16 content /= contentHash + = Fail "Invalid hash" + | otherwise + = Next content (readEntries rest) + where header = Lazy.take headerSize bs + headerSize = 10 + contentLength = fromIntegral $ runGet getWord64le header + contentHash = runGet getWord16le $ Lazy.drop 8 header + content = Lazy.take contentLength $ Lazy.drop headerSize bs + rest = Lazy.drop (contentLength+headerSize) bs + +lazyToStrict :: Lazy.ByteString -> Strict.ByteString +lazyToStrict = Strict.concat . Lazy.toChunks + diff --git a/src/Data/State/Acid/CRC.hs b/src/Data/State/Acid/CRC.hs new file mode 100644 --- /dev/null +++ b/src/Data/State/Acid/CRC.hs @@ -0,0 +1,58 @@ +{- CRC16 checksum inspired by http://hackage.haskell.org/package/crc16-table + As of 2011-04-13, this module is about 20x faster than crc16-table. +-} +module Data.State.Acid.CRC + ( crc16 + ) where + +import Data.Word ( Word16 ) +import Data.Array.Unboxed ( UArray, listArray ) +import Data.Array.Base ( unsafeAt ) +import Data.Bits ( Bits(..) ) + +import qualified Data.ByteString.Lazy as Lazy ( ByteString, foldl' ) + + +tableList :: [Word16] +tableList = + [0x00000,0x01189,0x02312,0x0329B,0x04624,0x057AD,0x06536,0x074BF, + 0x08C48,0x09DC1,0x0AF5A,0x0BED3,0x0CA6C,0x0DBE5,0x0E97E,0x0F8F7, + 0x01081,0x00108,0x03393,0x0221A,0x056A5,0x0472C,0x075B7,0x0643E, + 0x09CC9,0x08D40,0x0BFDB,0x0AE52,0x0DAED,0x0CB64,0x0F9FF,0x0E876, + 0x02102,0x0308B,0x00210,0x01399,0x06726,0x076AF,0x04434,0x055BD, + 0x0AD4A,0x0BCC3,0x08E58,0x09FD1,0x0EB6E,0x0FAE7,0x0C87C,0x0D9F5, + 0x03183,0x0200A,0x01291,0x00318,0x077A7,0x0662E,0x054B5,0x0453C, + 0x0BDCB,0x0AC42,0x09ED9,0x08F50,0x0FBEF,0x0EA66,0x0D8FD,0x0C974, + 0x04204,0x0538D,0x06116,0x0709F,0x00420,0x015A9,0x02732,0x036BB, + 0x0CE4C,0x0DFC5,0x0ED5E,0x0FCD7,0x08868,0x099E1,0x0AB7A,0x0BAF3, + 0x05285,0x0430C,0x07197,0x0601E,0x014A1,0x00528,0x037B3,0x0263A, + 0x0DECD,0x0CF44,0x0FDDF,0x0EC56,0x098E9,0x08960,0x0BBFB,0x0AA72, + 0x06306,0x0728F,0x04014,0x0519D,0x02522,0x034AB,0x00630,0x017B9, + 0x0EF4E,0x0FEC7,0x0CC5C,0x0DDD5,0x0A96A,0x0B8E3,0x08A78,0x09BF1, + 0x07387,0x0620E,0x05095,0x0411C,0x035A3,0x0242A,0x016B1,0x00738, + 0x0FFCF,0x0EE46,0x0DCDD,0x0CD54,0x0B9EB,0x0A862,0x09AF9,0x08B70, + 0x08408,0x09581,0x0A71A,0x0B693,0x0C22C,0x0D3A5,0x0E13E,0x0F0B7, + 0x00840,0x019C9,0x02B52,0x03ADB,0x04E64,0x05FED,0x06D76,0x07CFF, + 0x09489,0x08500,0x0B79B,0x0A612,0x0D2AD,0x0C324,0x0F1BF,0x0E036, + 0x018C1,0x00948,0x03BD3,0x02A5A,0x05EE5,0x04F6C,0x07DF7,0x06C7E, + 0x0A50A,0x0B483,0x08618,0x09791,0x0E32E,0x0F2A7,0x0C03C,0x0D1B5, + 0x02942,0x038CB,0x00A50,0x01BD9,0x06F66,0x07EEF,0x04C74,0x05DFD, + 0x0B58B,0x0A402,0x09699,0x08710,0x0F3AF,0x0E226,0x0D0BD,0x0C134, + 0x039C3,0x0284A,0x01AD1,0x00B58,0x07FE7,0x06E6E,0x05CF5,0x04D7C, + 0x0C60C,0x0D785,0x0E51E,0x0F497,0x08028,0x091A1,0x0A33A,0x0B2B3, + 0x04A44,0x05BCD,0x06956,0x078DF,0x00C60,0x01DE9,0x02F72,0x03EFB, + 0x0D68D,0x0C704,0x0F59F,0x0E416,0x090A9,0x08120,0x0B3BB,0x0A232, + 0x05AC5,0x04B4C,0x079D7,0x0685E,0x01CE1,0x00D68,0x03FF3,0x02E7A, + 0x0E70E,0x0F687,0x0C41C,0x0D595,0x0A12A,0x0B0A3,0x08238,0x093B1, + 0x06B46,0x07ACF,0x04854,0x059DD,0x02D62,0x03CEB,0x00E70,0x01FF9, + 0x0F78F,0x0E606,0x0D49D,0x0C514,0x0B1AB,0x0A022,0x092B9,0x08330, + 0x07BC7,0x06A4E,0x058D5,0x0495C,0x03DE3,0x02C6A,0x01EF1,0x00F78] + +table :: UArray Word16 Word16 +table = listArray (0,255) tableList + +crc16 :: Lazy.ByteString -> Word16 +crc16 = table `seq` complement . Lazy.foldl' worker 0xFFFF + where worker acc x = (acc `shiftR` 8) `xor` (table `unsafeAt` idx) + where idx = fromIntegral ((acc `xor` fromIntegral x) .&. 0xFF) + diff --git a/src/Data/State/Acid/Core.hs b/src/Data/State/Acid/Core.hs new file mode 100644 --- /dev/null +++ b/src/Data/State/Acid/Core.hs @@ -0,0 +1,164 @@ +{-# LANGUAGE GADTs, OverloadedStrings, DeriveDataTypeable, TypeFamilies, + FlexibleContexts, BangPatterns #-} +----------------------------------------------------------------------------- +-- | +-- Module : Data.State.Acid.Core +-- Copyright : PublicDomain +-- +-- Maintainer : lemmih@gmail.com +-- Portability : portable +-- +-- Low-level controls for transaction-based state changes. This module defines +-- structures and tools for running state modifiers indexed either by an Method +-- or a serialized Method. This module should rarely be used directly although +-- the 'Method' class is needed when defining events manually. +-- +-- The term \'Event\' is loosely used for transactions with ACID guarantees. +-- \'Method\' is loosely used for state operations without ACID guarantees +-- +module Data.State.Acid.Core + ( Core + , Method(..) + , MethodContainer(..) + , Tagged + , mkCore + , closeCore + , modifyCoreState + , modifyCoreState_ + , withCoreState + , lookupHotMethod + , lookupColdMethod + , runHotMethod + , runColdMethod + ) where + +import Control.Concurrent +import Control.Monad +import Control.Monad.State (State, runState ) +import qualified Data.Map as Map +import qualified Data.ByteString.Lazy as Lazy +import qualified Data.ByteString.Lazy.Char8 as Lazy.Char8 + +import Data.Binary + +import Data.Typeable +import Unsafe.Coerce (unsafeCoerce) + + +-- | The basic Method class. Each Method has an indexed result type +-- and a unique tag. +class ( Typeable ev, Binary ev + , Typeable (MethodResult ev), Binary (MethodResult ev)) => + Method ev where + type MethodResult ev + methodTag :: ev -> Tag + methodTag ev = Lazy.Char8.pack (show (typeOf ev)) + +-- | The control structure at the very center of acid-state. +-- This module provides access to a mutable state through +-- methods. No efforts towards durability, checkpointing or +-- sharding happens at this level. +-- Important things to keep in mind in this module: +-- * We don't distinguish between updates and queries. +-- * We allow direct access to the core state as well +-- as through events. +data Core st + = Core { coreState :: MVar st + , coreMethods :: MethodMap st + } + +-- | Construct a new Core using an initial state and a list of Methods. +mkCore :: [MethodContainer st] -- ^ List of methods capable of modifying the state. + -> st -- ^ Initial state value. + -> IO (Core st) +mkCore methods initialValue + = do mvar <- newMVar initialValue + return Core{ coreState = mvar + , coreMethods = mkMethodMap methods } + +-- | Mark Core as closed. Any subsequent use will throw an exception. +closeCore :: Core st -> IO () +closeCore core + = do swapMVar (coreState core) errorMsg + return () + where errorMsg = error "Access failure: Core closed." + +-- | Modify the state component. The resulting state is ensured to be in +-- WHNF. +modifyCoreState :: Core st -> (st -> IO (st, a)) -> IO a +modifyCoreState core action + = modifyMVar (coreState core) $ \st -> do (!st, a) <- action st + return (st, a) + +-- | Modify the state component. The resulting state is ensured to be in +-- WHNF. +modifyCoreState_ :: Core st -> (st -> IO st) -> IO () +modifyCoreState_ core action + = modifyMVar_ (coreState core) $ \st -> do !st' <- action st + return st' + +-- | Access the state component. +withCoreState :: Core st -> (st -> IO a) -> IO a +withCoreState core action + = withMVar (coreState core) action + +-- | Execute a method as given by a type identifier and an encoded string. +-- The exact format of the encoded string depends on the type identifier. +-- Results are encoded and type tagged before they're handed back out. +-- This function is used when running events from a log-file or from another +-- server. Events that originate locally are most likely executed with +-- the faster 'runHotMethod'. +runColdMethod :: Core st -> Tagged Lazy.ByteString -> IO Lazy.ByteString +runColdMethod core taggedMethod + = modifyCoreState core $ \st -> + do let (a, st') = runState (lookupColdMethod core taggedMethod) st + return ( st', a) + +-- | Find the state action that corresponds to a tagged and serialized method. +lookupColdMethod :: Core st -> Tagged Lazy.ByteString -> (State st Lazy.ByteString) +lookupColdMethod core (methodTag, methodContent) + = case Map.lookup methodTag (coreMethods core) of + Nothing -> error $ "Method tag doesn't exist: " ++ show methodTag + Just (Method method) + -> liftM encode (method (decode methodContent)) + +-- | Apply an in-memory method to the state. +runHotMethod :: Method method => Core st -> method -> IO (MethodResult method) +runHotMethod core method + = modifyCoreState core $ \st -> + do let (a, st') = runState (lookupHotMethod core method) st + return ( st', a) + +-- | Find the state action that corresponds to an in-memory method. +lookupHotMethod :: Method method => Core st -> method -> State st (MethodResult method) +lookupHotMethod core method + = case Map.lookup (methodTag method) (coreMethods core) of + Nothing -> error $ "Method type doesn't exist: " ++ show (typeOf method) + Just (Method methodHandler) + -> -- If the methodTag doesn't index the right methodHandler then we're in deep + -- trouble. Luckly, it would take deliberate malevolence for that to happen. + unsafeCoerce methodHandler method + +-- | Method tags must be unique and are most commenly generated automatically. +type Tag = Lazy.ByteString +type Tagged a = (Tag, a) + +-- | Method container structure that hides the exact type of the method. +data MethodContainer st where + Method :: Method method => (method -> State st (MethodResult method)) -> MethodContainer st + +-- | Collection of Methods indexed by a Tag. +type MethodMap st = Map.Map Tag (MethodContainer st) + +-- | Construct a 'MethodMap' from a list of Methods using their associated tag. +mkMethodMap :: [MethodContainer st] -> MethodMap st +mkMethodMap methods + = Map.fromList [ (methodType method, method) | method <- methods ] + where -- A little bit of ugliness is required to access the methodTags. + methodType :: MethodContainer st -> Tag + methodType m = case m of + Method fn -> let ev :: (ev -> State st res) -> ev + ev _ = undefined + in methodTag (ev fn) + + diff --git a/src/Data/State/Acid/Local.hs b/src/Data/State/Acid/Local.hs new file mode 100644 --- /dev/null +++ b/src/Data/State/Acid/Local.hs @@ -0,0 +1,183 @@ +{-# LANGUAGE GADTs, OverloadedStrings, DeriveDataTypeable, TypeFamilies, + MagicHash, GeneralizedNewtypeDeriving #-} +----------------------------------------------------------------------------- +-- | +-- Module : Data.State.Acid.Local +-- Copyright : PublicDomain +-- +-- Maintainer : lemmih@gmail.com +-- Portability : portable +-- +-- AcidState container using a transaction log on disk. The term \'Event\' is +-- loosely used for transactions with ACID guarantees. \'Method\' is loosely +-- used for state operations without ACID guarantees (see "Data.State.Acid.Core"). +-- + +module Data.State.Acid.Local + ( AcidState + , Event(..) + , EventResult + , UpdateEvent + , QueryEvent + , Update + , Query + , mkAcidState + , closeAcidState + , createCheckpoint + , update + , query + ) where + +import Data.State.Acid.Log as Log +import Data.State.Acid.Core + +import Control.Concurrent +import qualified Control.Monad.State as State +import Control.Monad.Reader +import Control.Applicative +import qualified Data.ByteString.Lazy as Lazy + +import Data.Binary +import Data.Typeable +import System.FilePath + +-- | Events return the same thing as Methods. The exact type of 'EventResult' +-- depends on the event. +type EventResult ev = MethodResult ev + +-- | We distinguish between events that modify the state and those that do not. +-- +-- UpdateEvents are executed in a MonadState context and have to be serialized +-- to disk before they are considered durable. +-- +-- QueryEvents are executed in a MonadReader context and obviously do not have +-- to be serialized to disk. +data Event st where + UpdateEvent :: UpdateEvent ev => (ev -> Update st (EventResult ev)) -> Event st + QueryEvent :: QueryEvent ev => (ev -> Query st (EventResult ev)) -> Event st + +-- | All UpdateEvents are also Methods. +class Method ev => UpdateEvent ev +-- | All QueryEvents are also Methods. +class Method ev => QueryEvent ev + + +eventsToMethods :: [Event st] -> [MethodContainer st] +eventsToMethods = map worker + where worker (UpdateEvent fn) = Method (unUpdate . fn) + worker (QueryEvent fn) = Method (\ev -> do st <- State.get + return (runReader (unQuery $ fn ev) st) + ) +{-| State container offering full ACID (Atomicity, Consistency, Isolation and Durability) + guarantees. + + [@Atomicity@] State changes are all-or-nothing. This is what you'd expect of any state + variable in Haskell and AcidState doesn't change that. + + [@Consistency@] No event or set of events will break your data invariants. This includes + power outages, + + [@Isolation@] Transactions cannot interfere with each other even when issued in parallel. + + [@Durability@] Successful transaction are guaranteed to survive system failure (both + hardware and software). +-} +data AcidState st + = AcidState { localCore :: Core st + , localEvents :: FileLog (Tagged Lazy.ByteString) + , localCheckpoints :: FileLog Checkpoint + } + +-- | Context monad for Update events. +newtype Update st a = Update { unUpdate :: State.State st a } + deriving (Monad, State.MonadState st) + +-- | Context monad for Query events. +newtype Query st a = Query { unQuery :: Reader st a } + deriving (Monad, MonadReader st) + +-- | Issue an Update event and wait for its result. Once this call returns, you are +-- guaranteed that the changes to the state are durable. Events may be issued in +-- parallel. +-- +-- It's a run-time error to issue events that aren't supported by the AcidState. +update :: UpdateEvent event => AcidState st -> event -> IO (EventResult event) +update acidState event + = do mvar <- newEmptyMVar + modifyCoreState_ (localCore acidState) $ \st -> + do let (result, st') = State.runState hotMethod st + -- schedule the log entry. Very important that it happens when 'localCore' is locked. + pushEntry (localEvents acidState) (methodTag event, encode event) $ putMVar mvar result + return st' + takeMVar mvar + where hotMethod = lookupHotMethod (localCore acidState) event + +-- | Issue a Query event and wait for its result. Events may be issued in parallel. +query :: QueryEvent event => AcidState st -> event -> IO (EventResult event) +query acidState event + = runHotMethod (localCore acidState) event + +-- | Take a snapshot of the state and save it to disk. Creating checkpoints +-- makes it faster to resume AcidStates and you're free to create them as +-- often or seldom as fits your needs. Transactions can run concurrently +-- with this call. +-- +-- This call will not return until the operation has succeeded. +createCheckpoint :: Binary st => AcidState st -> IO () +createCheckpoint acidState + = do mvar <- newEmptyMVar + withCoreState (localCore acidState) $ \st -> + do eventId <- askCurrentEntryId (localEvents acidState) + pushEntry (localCheckpoints acidState) (Checkpoint eventId (encode st)) (putMVar mvar ()) + takeMVar mvar + + + +data Checkpoint = Checkpoint EntryId Lazy.ByteString + +instance Binary Checkpoint where + put (Checkpoint eventEntryId content) + = do put eventEntryId + put content + get = Checkpoint <$> get <*> get + + +-- | Create an AcidState given a list of events (aka. transactions) and an initial value. +-- +-- This will create or resume a log found in the \"state\/[typeOf state]\/\" directory. +mkAcidState :: (Typeable st, Binary st) + => [Event st] -- ^ List of events capable of updating or querying the state. + -> st -- ^ Initial state value. This value is only used if no checkpoint is + -- found. + -> IO (AcidState st) +mkAcidState events initialState + = do core <- mkCore (eventsToMethods events) initialState + let directory = "state" show (typeOf initialState) + let eventsLogKey = LogKey { logDirectory = directory + , logPrefix = "events" } + checkpointsLogKey = LogKey { logDirectory = directory + , logPrefix = "checkpoints" } + mbLastCheckpoint <- Log.newestEntry checkpointsLogKey + n <- case mbLastCheckpoint of + Nothing + -> return 0 + Just (Checkpoint eventCutOff content) + -> do modifyCoreState_ core (\_oldState -> return (decode content)) + return eventCutOff + events <- entriesAfterCutoff eventsLogKey n + mapM_ (runColdMethod core) events + eventsLog <- openFileLog eventsLogKey + checkpointsLog <- openFileLog checkpointsLogKey + return AcidState { localCore = core + , localEvents = eventsLog + , localCheckpoints = checkpointsLog + } + +-- | Close an AcidState and associated logs. +-- Any subsequent usage of the AcidState will throw an exception. +closeAcidState :: AcidState st -> IO () +closeAcidState acidState + = do closeCore (localCore acidState) + closeFileLog (localEvents acidState) + closeFileLog (localCheckpoints acidState) + diff --git a/src/Data/State/Acid/Log.hs b/src/Data/State/Acid/Log.hs new file mode 100644 --- /dev/null +++ b/src/Data/State/Acid/Log.hs @@ -0,0 +1,169 @@ +module Data.State.Acid.Log + ( FileLog + , LogKey(..) + , EntryId + , openFileLog + , closeFileLog + , pushEntry + , newestEntry + , entriesAfterCutoff + , askCurrentEntryId + ) where + +import Data.State.Acid.Archive as Archive +import System.Directory +import System.FilePath +import System.IO +import Control.Monad +import Control.Concurrent +import Control.Concurrent.STM +import qualified Data.ByteString.Lazy as Lazy +--import qualified Data.ByteString as Strict +import Data.List +import Data.Maybe +import Data.Binary + +import Text.Printf + +type EntryId = Int + +data FileLog object + = FileLog { logIdentifier :: LogKey object + , logCurrent :: MVar (Handle) + , logNextEntryId :: TVar EntryId + , logQueue :: TVar [(Lazy.ByteString,IO ())] + , logThreads :: [ThreadId] + } + +data LogKey object + = LogKey + { logDirectory :: FilePath + , logPrefix :: String + } + +formatLogFile :: String -> EntryId -> String +formatLogFile tag n + = printf "%s-%010d.log" tag n + +findLogFiles :: LogKey object -> IO [(EntryId, FilePath)] +findLogFiles identifier + = do createDirectoryIfMissing True (logDirectory identifier) + files <- getDirectoryContents (logDirectory identifier) + return [ (tid, logDirectory identifier file) + | file <- files + , logFile <- maybeToList (stripPrefix (logPrefix identifier ++ "-") file) + , (tid, ".log") <- reads logFile ] + +openFileLog :: LogKey object -> IO (FileLog object) +openFileLog identifier + = do logFiles <- findLogFiles identifier + currentState <- newEmptyMVar + queue <- newTVarIO [] + nextEntryRef <- newTVarIO 0 + tid2 <- forkIO $ forever $ do pairs <- atomically $ do vals <- readTVar queue + guard (not $ null vals) + writeTVar queue [] + return (reverse vals) + let (entries, actions) = unzip pairs + withMVar currentState $ \handle -> + do let arch = Archive.packEntries entries + seq (Lazy.length arch) (return ()) + Lazy.hPutStr handle arch + hFlush handle + return () + sequence_ actions + yield + let log = FileLog { logIdentifier = identifier + , logCurrent = currentState + , logNextEntryId = nextEntryRef + , logQueue = queue + , logThreads = [tid2] } + if null logFiles + then do let currentEntryId = 0 + currentHandle <- openBinaryFile (logDirectory identifier formatLogFile (logPrefix identifier) currentEntryId) WriteMode + putMVar currentState currentHandle + else do let (lastFileEntryId, lastFilePath) = maximum logFiles + entries <- readEntities lastFilePath + let currentEntryId = lastFileEntryId + length entries + atomically $ writeTVar nextEntryRef currentEntryId + currentHandle <- openFile (logDirectory identifier formatLogFile (logPrefix identifier) currentEntryId) WriteMode + putMVar currentState currentHandle + return log + +closeFileLog :: FileLog object -> IO () +closeFileLog log + = modifyMVar_ (logCurrent log) $ \handle -> + do hClose handle + forkIO $ forM_ (logThreads log) killThread + return $ error "FileLog has been closed" + +readEntities :: FilePath -> IO [Lazy.ByteString] +readEntities path + = do archive <- Lazy.readFile path + return $ worker (Archive.readEntries archive) + where worker Done = [] + worker (Next entry next) + = entry : worker next + worker Fail{} = [] + +-- Return entries newer than or equal to the cutoff. +-- Do not use after the log has been opened. +-- Implementation: 1) find the files that /may/ contain entries +-- younger than the cutoff. +-- 2) parse all the entries in those files. +-- 3) drop the entries that are too old. +entriesAfterCutoff :: Binary object => LogKey object -> EntryId -> IO [object] +entriesAfterCutoff identifier cutoff + = do logFiles <- findLogFiles identifier + let sorted = reverse $ sort logFiles -- newest files first + relevant = reverse $ takeRelevant sorted -- oldest files first + (entryIds, files) = unzip relevant + case entryIds of + [] -> return [] + (firstEntryId : _) + -> do archive <- liftM Lazy.concat $ mapM Lazy.readFile files + let events = entriesToList $ readEntries archive + return $ map decode $ drop (cutoff - firstEntryId) events + where takeRelevant [] = [] + takeRelevant ((firstEntryId, file) : rest) + | firstEntryId < cutoff + = [ (firstEntryId, file) ] + | otherwise + = (firstEntryId, file) : takeRelevant rest + + +-- Finds the newest entry in the log. Doesn't work on open logs. +-- Do not use after the log has been opened. +-- Implementation: Search the newest log files first. Once a file +-- containing at least one valid entry is found, +-- return the last entry in that file. +newestEntry :: Binary object => LogKey object -> IO (Maybe object) +newestEntry identifier + = do logFiles <- findLogFiles identifier + let sorted = reverse $ sort logFiles + (eventIds, files) = unzip sorted + archives <- mapM Lazy.readFile files + return $ worker archives + where worker [] = Nothing + worker (archive:archives) + = case Archive.readEntries archive of + Done -> worker archives + Next entry next -> Just (decode (lastEntry entry next)) + Fail{} -> worker archives + lastEntry entry Done = entry + lastEntry entry Fail{} = entry + lastEntry _ (Next entry next) = lastEntry entry next + +-- Schedule a new log entry. May not block. +pushEntry :: Binary object => FileLog object -> object -> IO () -> IO () +pushEntry log object finally + = atomically $ + do tid <- readTVar (logNextEntryId log) + writeTVar (logNextEntryId log) (tid+1) + pairs <- readTVar (logQueue log) + writeTVar (logQueue log) ((encoded, finally) : pairs) + where encoded = encode object + +askCurrentEntryId :: FileLog object -> IO EntryId +askCurrentEntryId log + = atomically $ readTVar (logNextEntryId log)