packages feed

acid-state (empty) → 0.1

raw patch · 9 files changed

+793/−0 lines, 9 filesdep +arraydep +basedep +binarysetup-changed

Dependencies added: array, base, binary, bytestring, containers, directory, filepath, monads-fd, stm

Files

+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ acid-state.cabal view
@@ -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 <lemmih@gmail.com>++-- 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:         +
+ examples/HelloWorld.hs view
@@ -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)+           ]
+ src/Data/State/Acid.hs view
@@ -0,0 +1,5 @@+module Data.State.Acid+    ( module Data.State.Acid.Local+    ) where++import Data.State.Acid.Local
+ src/Data/State/Acid/Archive.hs view
@@ -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+
+ src/Data/State/Acid/CRC.hs view
@@ -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)+
+ src/Data/State/Acid/Core.hs view
@@ -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)++
+ src/Data/State/Acid/Local.hs view
@@ -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)+
+ src/Data/State/Acid/Log.hs view
@@ -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)