diff --git a/acid-state.cabal b/acid-state.cabal --- a/acid-state.cabal +++ b/acid-state.cabal @@ -7,7 +7,7 @@ -- 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.2 +Version: 0.3 -- A short (one-line) description of the package. Synopsis: Add ACID guarantees to any serializable Haskell data structure. @@ -45,15 +45,17 @@ Library -- Modules exported by the library. - Exposed-Modules: Data.State.Acid, Data.State.Acid.Local, Data.State.Acid.Core + Exposed-Modules: Data.Acid, Data.Acid.Core, + Data.Acid.Local -- Modules not exported by this package. - Other-modules: Data.State.Acid.Log, Data.State.Acid.Archive, - Data.State.Acid.CRC + Other-modules: Data.Acid.Log, Data.Acid.Archive, + Data.Acid.CRC, Paths_acid_state, + Data.Acid.TemplateHaskell -- Packages needed in order to build this package. Build-depends: base >= 4 && < 5, binary, bytestring, stm, filepath, directory, - mtl, array, containers + mtl, array, containers, template-haskell Hs-Source-Dirs: src/ diff --git a/examples/HelloDatabase.hs b/examples/HelloDatabase.hs new file mode 100644 --- /dev/null +++ b/examples/HelloDatabase.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE TypeFamilies, DeriveDataTypeable, TemplateHaskell #-} +module Main (main) where + +import Data.Acid + +import Control.Monad.State ( get, put ) +import Control.Monad.Reader ( ask ) +import Control.Applicative ( (<$>) ) +import System.Environment ( getArgs ) +import qualified Data.Binary as Binary + +type Message = String +data Database = Database [Message] + +instance Binary.Binary Database where + get = Database <$> Binary.get + put (Database msg) = Binary.put msg + +-- Transactions are defined to run in either the 'Update' monad +-- or the 'Query' monad. +addMessage :: Message -> Update Database () +addMessage msg + = do Database messages <- get + put $ Database (msg:messages) + +viewMessages :: Int -> Query Database [Message] +viewMessages limit + = do Database messages <- ask + return $ take limit messages + +-- This will define @ViewMessage@ and @AddMessage@ for us. +$(makeAcidic ''Database ['addMessage, 'viewMessages]) + +main :: IO () +main = do args <- getArgs + database <- openAcidStateFrom "myDatabase/" (Database ["Welcome to the acid-state database."]) + if null args + then do messages <- query database (ViewMessages 10) + putStrLn "Last 10 messages:" + mapM_ putStrLn [ " " ++ message | message <- messages ] + else do update database (AddMessage (unwords args)) + putStrLn "Your message has been added to the database." diff --git a/examples/HelloWorld.hs b/examples/HelloWorld.hs --- a/examples/HelloWorld.hs +++ b/examples/HelloWorld.hs @@ -1,8 +1,8 @@ -{-# LANGUAGE DeriveDataTypeable, TypeFamilies, StandaloneDeriving #-} +{-# LANGUAGE DeriveDataTypeable, TypeFamilies, TemplateHaskell #-} module Main (main) where -import Data.State.Acid.Core -import Data.State.Acid +import Data.Acid.Core +import Data.Acid import qualified Control.Monad.State as State import Control.Monad.Reader @@ -17,6 +17,9 @@ data HelloWorldState = HelloWorldState String deriving (Show, Typeable) +instance Binary HelloWorldState where + put (HelloWorldState state) = put state + get = liftM HelloWorldState get ------------------------------------------------------ -- The transaction we will execute over the state. @@ -29,49 +32,16 @@ queryState = do HelloWorldState string <- ask return string +$(makeAcidic ''HelloWorldState ['writeState, 'queryState]) ------------------------------------------------------ -- This is how AcidState is used: main :: IO () -main = do acid <- mkAcidState myEvents (HelloWorldState "Hello world") +main = do acid <- openAcidState (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/examples/HelloWorldNoTH.hs b/examples/HelloWorldNoTH.hs new file mode 100644 --- /dev/null +++ b/examples/HelloWorldNoTH.hs @@ -0,0 +1,79 @@ +{-# LANGUAGE DeriveDataTypeable, TypeFamilies, StandaloneDeriving #-} +module Main (main) where + +import Data.Acid.Core +import Data.Acid.Local + +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) + +instance Binary HelloWorldState where + put (HelloWorldState state) = put state + get = liftM HelloWorldState get + +------------------------------------------------------ +-- 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 <- openAcidState (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 = () + type MethodState WriteState = HelloWorldState +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 + type MethodState QueryState = HelloWorldState +instance QueryEvent QueryState + + +instance IsAcidic HelloWorldState where + acidEvents = [ UpdateEvent (\(WriteState newState) -> writeState newState) + , QueryEvent (\QueryState -> queryState) + ] diff --git a/examples/KeyValue.hs b/examples/KeyValue.hs new file mode 100644 --- /dev/null +++ b/examples/KeyValue.hs @@ -0,0 +1,64 @@ +{-# LANGUAGE DeriveDataTypeable, TypeFamilies, TemplateHaskell #-} +module Main (main) where + +import Data.Acid.Core +import Data.Acid + +import qualified Control.Monad.State as State +import Control.Monad.Reader +import Control.Applicative +import System.Environment +import System.IO +import Data.Binary + +import Data.Typeable + +import qualified Data.Map as Map + +------------------------------------------------------ +-- The Haskell structure that we want to encapsulate + +type Key = String +type Value = String + +data KeyValue = KeyValue !(Map.Map Key Value) + deriving (Typeable) + +instance Binary KeyValue where + put (KeyValue state) = put state + get = liftM KeyValue get + +------------------------------------------------------ +-- The transaction we will execute over the state. + +insertKey :: Key -> Value -> Update KeyValue () +insertKey key value + = do KeyValue m <- State.get + State.put (KeyValue (Map.insert key value m)) + +lookupKey :: Key -> Query KeyValue (Maybe Value) +lookupKey key + = do KeyValue m <- ask + return (Map.lookup key m) + +$(makeAcidic ''KeyValue ['insertKey, 'lookupKey]) + +------------------------------------------------------ +-- This is how AcidState is used: + +main :: IO () +main = do acid <- openAcidState (KeyValue Map.empty) + args <- getArgs + case args of + [key] + -> do mbKey <- query acid (LookupKey key) + case mbKey of + Nothing -> putStrLn $ key ++ " has no associated value." + Just value -> putStrLn $ key ++ " = " ++ value + [key,val] + -> do update acid (InsertKey key val) + putStrLn "Done." + _ -> do putStrLn "Usage:" + putStrLn " key Lookup the value of 'key'." + putStrLn " key value Set the value of 'key' to 'value'." + closeAcidState acid diff --git a/examples/KeyValueNoTH.hs b/examples/KeyValueNoTH.hs new file mode 100644 --- /dev/null +++ b/examples/KeyValueNoTH.hs @@ -0,0 +1,95 @@ +{-# LANGUAGE DeriveDataTypeable, TypeFamilies, StandaloneDeriving #-} +module Main (main) where + +import Data.Acid.Core +import Data.Acid.Local + +import qualified Control.Monad.State as State +import Control.Monad.Reader +import Control.Applicative +import System.Environment +import System.IO +import Data.Binary + +import Data.Typeable + +import qualified Data.Map as Map + +------------------------------------------------------ +-- The Haskell structure that we want to encapsulate + +type Key = String +type Value = String + +data KeyValue = KeyValue !(Map.Map Key Value) + deriving (Typeable) + +instance Binary KeyValue where + put (KeyValue state) = put state + get = liftM KeyValue get + +------------------------------------------------------ +-- The transaction we will execute over the state. + +insertKey :: Key -> Value -> Update KeyValue () +insertKey key value + = do KeyValue m <- State.get + State.put (KeyValue (Map.insert key value m)) + +lookupKey :: Key -> Query KeyValue (Maybe Value) +lookupKey key + = do KeyValue m <- ask + return (Map.lookup key m) + +------------------------------------------------------ +-- This is how AcidState is used: + +main :: IO () +main = do acid <- openAcidState (KeyValue Map.empty) + args <- getArgs + case args of + [key] + -> do mbKey <- query acid (LookupKey key) + case mbKey of + Nothing -> putStrLn $ key ++ " has no associated value." + Just value -> putStrLn $ key ++ " = " ++ value + [key,val] + -> do update acid (InsertKey key val) + putStrLn "Done." + _ -> do putStrLn "Usage:" + putStrLn " key Lookup the value of 'key'." + putStrLn " key value Set the value of 'key' to 'value'." + closeAcidState acid + + + +------------------------------------------------------ +-- The gritty details. These things may be done with +-- Template Haskell in the future. + +data InsertKey = InsertKey Key Value +data LookupKey = LookupKey Key + + +deriving instance Typeable InsertKey +instance Binary InsertKey where + put (InsertKey key value) = put key >> put value + get = InsertKey <$> get <*> get +instance Method InsertKey where + type MethodResult InsertKey = () + type MethodState InsertKey = KeyValue +instance UpdateEvent InsertKey + +deriving instance Typeable LookupKey +instance Binary LookupKey where + put (LookupKey key) = put key + get = LookupKey <$> get +instance Method LookupKey where + type MethodResult LookupKey = Maybe Value + type MethodState LookupKey = KeyValue +instance QueryEvent LookupKey + +instance IsAcidic KeyValue where + acidEvents = [ UpdateEvent (\(InsertKey key value) -> insertKey key value) + , QueryEvent (\(LookupKey key) -> lookupKey key) + ] diff --git a/examples/StressTest.hs b/examples/StressTest.hs new file mode 100644 --- /dev/null +++ b/examples/StressTest.hs @@ -0,0 +1,59 @@ +{-# LANGUAGE DeriveDataTypeable, TypeFamilies, TemplateHaskell #-} +module Main (main) where + +import Data.Acid.Core +import Data.Acid + +import qualified Control.Monad.State as State +import Control.Monad.Reader +import System.Environment +import System.IO +import Data.Binary + +import Data.Typeable + +------------------------------------------------------ +-- The Haskell structure that we want to encapsulate + +data StressState = StressState !Int + deriving (Show, Typeable) + +instance Binary StressState where + put (StressState state) = put state + get = liftM StressState get + +------------------------------------------------------ +-- The transaction we will execute over the state. + +pokeState :: Update StressState () +pokeState = do StressState i <- State.get + State.put (StressState (i+1)) + +queryState :: Query StressState Int +queryState = do StressState i <- ask + return i + +$(makeAcidic ''StressState ['pokeState, 'queryState]) + +------------------------------------------------------ +-- This is how AcidState is used: + +main :: IO () +main = do acid <- openAcidState (StressState 0) + args <- getArgs + case args of + ["checkpoint"] + -> createCheckpoint acid + ["query"] + -> do n <- query acid QueryState + putStrLn $ "State value: " ++ show n + ["poke"] + -> do putStr "Issuing 10k sequential transactions... " + hFlush stdout + replicateM_ 10000 (update acid PokeState) + putStrLn "Done" + _ -> do putStrLn $ "Commands:" + putStrLn $ " query Prints out the current state." + putStrLn $ " poke Spawn 10k transactions." + putStrLn $ " checkpoint Create a new checkpoint." + closeAcidState acid diff --git a/examples/StressTestNoTH.hs b/examples/StressTestNoTH.hs new file mode 100644 --- /dev/null +++ b/examples/StressTestNoTH.hs @@ -0,0 +1,91 @@ +{-# LANGUAGE DeriveDataTypeable, TypeFamilies, StandaloneDeriving #-} +module Main (main) where + +import Data.Acid.Core +import Data.Acid.Local + +import qualified Control.Monad.State as State +import Control.Monad.Reader +import System.Environment +import System.IO +import Data.Binary + +import Data.Typeable + +------------------------------------------------------ +-- The Haskell structure that we want to encapsulate + +data StressState = StressState !Int + deriving (Show, Typeable) + +instance Binary StressState where + put (StressState state) = put state + get = liftM StressState get + +------------------------------------------------------ +-- The transaction we will execute over the state. + +pokeState :: Update StressState () +pokeState = do StressState i <- State.get + State.put (StressState (i+1)) + +queryState :: Query StressState Int +queryState = do StressState i <- ask + return i + + +------------------------------------------------------ +-- This is how AcidState is used: + +main :: IO () +main = do acid <- openAcidState (StressState 0) + args <- getArgs + case args of + ["checkpoint"] + -> createCheckpoint acid + ["query"] + -> do n <- query acid QueryState + putStrLn $ "State value: " ++ show n + ["poke"] + -> do putStr "Issuing 10k sequential transactions... " + hFlush stdout + replicateM_ 10000 (update acid PokeState) + putStrLn "Done" + _ -> do putStrLn $ "Commands:" + putStrLn $ " query Prints out the current state." + putStrLn $ " poke Spawn 10k transactions." + putStrLn $ " checkpoint Create a new checkpoint." + closeAcidState acid + + + +------------------------------------------------------ +-- The gritty details. These things may be done with +-- Template Haskell in the future. + +data PokeState = PokeState +data QueryState = QueryState + + +deriving instance Typeable PokeState +instance Binary PokeState where + put PokeState = return () + get = return PokeState +instance Method PokeState where + type MethodResult PokeState = () + type MethodState PokeState = StressState +instance UpdateEvent PokeState + +deriving instance Typeable QueryState +instance Binary QueryState where + put QueryState = return () + get = return QueryState +instance Method QueryState where + type MethodResult QueryState = Int + type MethodState QueryState = StressState +instance QueryEvent QueryState + +instance IsAcidic StressState where + acidEvents = [ UpdateEvent (\PokeState -> pokeState) + , QueryEvent (\QueryState -> queryState) + ] diff --git a/src/Data/Acid.hs b/src/Data/Acid.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid.hs @@ -0,0 +1,34 @@ +----------------------------------------------------------------------------- +{- | + Module : Data.Acid + Copyright : PublicDomain + + Maintainer : lemmih@gmail.com + Portability : portable + + AcidState container using a transaction log on disk. + + To see how it all fits together, have a look at these example + . + +-} + +module Data.Acid + ( AcidState + , openAcidState + , openAcidStateFrom + , closeAcidState + , createCheckpoint + , update + , query + , EventResult + , UpdateEvent + , QueryEvent + , Update + , Query + , IsAcidic + , makeAcidic + ) where + +import Data.Acid.Local +import Data.Acid.TemplateHaskell diff --git a/src/Data/Acid/Archive.hs b/src/Data/Acid/Archive.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/Archive.hs @@ -0,0 +1,73 @@ +{- +Format: + |content length| crc16 | content | + |8 bytes | 2 bytes | n bytes | +-} +module Data.Acid.Archive + ( Entry + , Entries(..) + , putEntries + , packEntries + , readEntries + , entriesToList + , entriesToListNoFail + ) where + +import Data.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/Acid/CRC.hs b/src/Data/Acid/CRC.hs new file mode 100644 --- /dev/null +++ b/src/Data/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.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/Acid/Core.hs b/src/Data/Acid/Core.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/Core.hs @@ -0,0 +1,165 @@ +{-# LANGUAGE GADTs, OverloadedStrings, DeriveDataTypeable, TypeFamilies, + FlexibleContexts, BangPatterns #-} +----------------------------------------------------------------------------- +-- | +-- Module : Data.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.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 + type MethodState 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 (MethodState method) -> 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 (MethodState method) -> method -> State (MethodState method) (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 (MethodState method) (MethodResult method)) -> MethodContainer (MethodState method) + +-- | 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/Acid/Local.hs b/src/Data/Acid/Local.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/Local.hs @@ -0,0 +1,203 @@ +{-# LANGUAGE GADTs, OverloadedStrings, DeriveDataTypeable, TypeFamilies, + MagicHash, GeneralizedNewtypeDeriving #-} +----------------------------------------------------------------------------- +-- | +-- Module : Data.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.Acid.Core"). +-- + +module Data.Acid.Local + ( IsAcidic(..) + , AcidState + , Event(..) + , EventResult + , UpdateEvent + , QueryEvent + , Update + , Query + , openAcidState + , openAcidStateFrom + , closeAcidState + , createCheckpoint + , update + , query + ) where + +import Data.Acid.Log as Log +import Data.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 + +type EventState ev = MethodState 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 (EventState ev) (EventResult ev)) -> Event (EventState ev) + QueryEvent :: QueryEvent ev => (ev -> Query (EventState ev) (EventResult ev)) -> Event (EventState ev) + +-- | 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 (EventState event) -> 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 + -- to ensure that events are logged in the same order that they are executed. + 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 (EventState event) -> 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 + +class (Binary st) => IsAcidic st where + acidEvents :: [Event st] + -- ^ List of events capable of updating or querying the state. + +-- | Create an AcidState given an initial value. +-- +-- This will create or resume a log found in the \"state\/[typeOf state]\/\" directory. +openAcidState :: (Typeable st, IsAcidic st) + => st -- ^ Initial state value. This value is only used if no checkpoint is + -- found. + -> IO (AcidState st) +openAcidState initialState + = openAcidStateFrom ("state" show (typeOf initialState)) initialState + +-- | Create an AcidState given a log directory and an initial value. +-- +-- This will create or resume a log found in @directory@. +-- Running two AcidState's from the same directory is an error +-- but will not result in dataloss. +openAcidStateFrom :: (IsAcidic st) + => FilePath -- ^ Location of the checkpoint and transaction files. + -> st -- ^ Initial state value. This value is only used if no checkpoint is + -- found. + -> IO (AcidState st) +openAcidStateFrom directory initialState + = do core <- mkCore (eventsToMethods acidEvents) 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 + + eventsLog <- openFileLog eventsLogKey + events <- readEntriesFrom eventsLog n + mapM_ (runColdMethod core) events + 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/Acid/Log.hs b/src/Data/Acid/Log.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/Log.hs @@ -0,0 +1,221 @@ +-- A log is a stack of entries that supports efficient pushing of +-- new entries and fetching of old. It can be considered an +-- extendible array of entries. +-- +module Data.Acid.Log + ( FileLog + , LogKey(..) + , EntryId + , openFileLog + , closeFileLog + , pushEntry + , readEntriesFrom + , newestEntry + , askCurrentEntryId + ) where + +import Data.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 ( printf ) + +import Paths_acid_state ( version ) +import Data.Version ( showVersion ) + +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 ] + + +saveVersionFile :: LogKey object -> IO () +saveVersionFile key + = do exist <- doesFileExist versionFile + unless exist $ writeFile versionFile (showVersion version) + where versionFile = logDirectory key logPrefix key <.> "version" + +openFileLog :: LogKey object -> IO (FileLog object) +openFileLog identifier + = do logFiles <- findLogFiles identifier + saveVersionFile identifier + currentState <- newEmptyMVar + queue <- newTVarIO ([], []) + nextEntryRef <- newTVarIO 0 + tid2 <- forkIO $ fileWriter currentState queue + 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 + +fileWriter currentState queue + = forever $ + do (entries, actions) <- atomically $ do (entries, actions) <- readTVar queue + when (null entries && null actions) retry + writeTVar queue ([], []) + -- We don't actually have to reverse the actions + -- but I don't think it hurts performance much. + return (reverse entries, reverse actions) + withMVar currentState $ \handle -> + do let arch = Archive.packEntries entries + Lazy.hPutStr handle arch + hFlush handle + return () + sequence_ actions + yield + + +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{} = [] + +-- Read all durable entries younger than the given EntryId. +-- Note that entries written during or after this call won't +-- be included in the returned list. +readEntriesFrom :: Binary object => FileLog object -> EntryId -> IO [object] +readEntriesFrom log youngestEntry + = do -- Cut the log so we can read written entries without interfering + -- with the writing of new entries. + entryCap <- cutFileLog log + -- We're interested in these entries: youngestEntry <= x < entryCap. + logFiles <- findLogFiles (logIdentifier log) + let sorted = sort logFiles + findRelevant [] = [] + findRelevant [ logFile ] + | youngestEntry <= rangeStart logFile && rangeStart logFile < entryCap + = [ logFile ] + | otherwise + = [] + findRelevant ( left : right : xs ) + | youngestEntry >= rangeStart right -- All entries in 'path' must be too old if this is true + = findRelevant (right : xs) + | rangeStart left >= entryCap -- All files from now on contain entries that are too young. + = [] + | otherwise + = left : findRelevant (right : xs) + + relevant = findRelevant sorted + firstEntryId = case relevant of + [] -> 0 + ( logFile : _logFiles) -> rangeStart logFile + + archive <- liftM Lazy.concat $ mapM Lazy.readFile (map snd relevant) + let entries = entriesToList $ readEntries archive + return $ map decode + $ take (entryCap - youngestEntry) -- Take events under the eventCap. + $ drop (youngestEntry - firstEntryId) entries -- Drop entries that are too young. + + where rangeStart (firstEntryId, _path) = firstEntryId + + +cutFileLog :: FileLog object -> IO EntryId +cutFileLog log + = do mvar <- newEmptyMVar + let action = do currentEntryId <- atomically $ + do (entries, _) <- readTVar (logQueue log) + next <- readTVar (logNextEntryId log) + return (next - length entries) + modifyMVar_ (logCurrent log) $ \old -> + do hClose old + openFile (logDirectory key formatLogFile (logPrefix key) currentEntryId) WriteMode + putMVar mvar currentEntryId + atomically $ + do (entries, actions) <- readTVar (logQueue log) + writeTVar (logQueue log) (entries, action : actions) + takeMVar mvar + where key = logIdentifier log + +-- 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. This call does not block +-- The given IO action runs once the object is durable. The IO action +-- blocks the serialization of events so it should be swift. +pushEntry :: Binary object => FileLog object -> object -> IO () -> IO () +pushEntry log object finally + = atomically $ + do tid <- readTVar (logNextEntryId log) + writeTVar (logNextEntryId log) (tid+1) + (entries, actions) <- readTVar (logQueue log) + writeTVar (logQueue log) ( encoded : entries, finally : actions ) + where encoded = encode object + +askCurrentEntryId :: FileLog object -> IO EntryId +askCurrentEntryId log + = atomically $ readTVar (logNextEntryId log) diff --git a/src/Data/Acid/TemplateHaskell.hs b/src/Data/Acid/TemplateHaskell.hs new file mode 100644 --- /dev/null +++ b/src/Data/Acid/TemplateHaskell.hs @@ -0,0 +1,198 @@ +{-# LANGUAGE TemplateHaskell #-} +{- Holy crap this code is messy. -} +module Data.Acid.TemplateHaskell + ( makeAcidic + ) where + +import Language.Haskell.TH + +import Data.Acid.Core +import Data.Acid.Local + +import Data.Binary +import Data.Typeable +import Data.Char +import Control.Applicative +import Control.Monad + +{-| Create the control structures required for acid states + using Template Haskell. + +This code: + +@ +myUpdate :: Argument -> Update State Result +myUpdate arg = ... + +myQuery :: Argument -> Query State Result +myQuery arg = ... + +$(makeAcidic ''State ['myUpdate, 'myQuery]) +@ + +will make @State@ an instance of 'IsAcidic' and provide the following +events: + +@ +data MyUpdate = MyUpdate Argument +data MyQuery = MyQuery Argument +@ + +-} +makeAcidic :: Name -> [Name] -> Q [Dec] +makeAcidic stateName eventNames + = do stateInfo <- reify stateName + case stateInfo of + TyConI tycon + ->case tycon of + DataD _cxt _name tyvars constructors _derivs + -> makeAcidic' eventNames stateName tyvars constructors + NewtypeD _cxt _name tyvars constructor _derivs + -> makeAcidic' eventNames stateName tyvars [constructor] + _ -> error "Unsupported state type. Only 'data' and 'newtype' are supported." + _ -> error "Given state is not a type." + +makeEvent :: Name -> Name -> Q [Dec] +makeEvent eventName stateName + = do eventInfo <- reify eventName + eventType <- getEventType eventName + d <- makeEventDataType eventName eventType + b <- makeBinaryInstance eventName eventType + i <- makeMethodInstance eventName eventType + e <- makeEventInstance eventName eventType + return [d,b,i,e] + +getEventType :: Name -> Q Type +getEventType eventName + = do eventInfo <- reify eventName + case eventInfo of + VarI _name eventType _decl _fixity + -> return eventType + _ -> error $ "Events must be functions: " ++ show eventName + +--instance (Binary key, Typeable key, Binary val, Typeable val) => IsAcidic State where +-- acidEvents = [ UpdateEven (\(MyUpdateEvent arg1 arg2 -> myUpdateEvent arg1 arg2) ] +makeIsAcidic eventNames stateName tyvars constructors + = do types <- mapM getEventType eventNames + let preds = [ ''Binary, ''Typeable ] + ty = appT (conT ''IsAcidic) stateType + handlers = map (uncurry makeEventHandler) (zip eventNames types) + instanceD (mkCxtFromTyVars preds tyvars []) ty + [ valD (varP 'acidEvents) (normalB (listE handlers)) [] ] + where stateType = foldl appT (conT stateName) [ varT var | PlainTV var <- tyvars ] + +-- UpdateEvent (\(MyUpdateEvent arg1 arg2) -> myUpdateEvent arg1 arg2) +makeEventHandler :: Name -> Type -> ExpQ +makeEventHandler eventName eventType + = do vars <- replicateM (length args) (newName "arg") + let lamClause = conP eventStructName [varP var | var <- vars ] + conE constr `appE` lamE [lamClause] (foldl appE (varE eventName) (map varE vars)) + where constr = if isUpdate then 'UpdateEvent else 'QueryEvent + (tyvars, cxt, args, stateType, resultType, isUpdate) = analyseType eventName eventType + eventStructName = mkName (structName (nameBase eventName)) + structName [] = [] + structName (x:xs) = toUpper x : xs + +--data MyUpdateEvent = MyUpdateEvent Arg1 Arg2 +-- deriving (Typeable) +makeEventDataType eventName eventType + = do let con = normalC eventStructName [ strictType notStrict (return arg) | arg <- args ] + dataD (return cxt) eventStructName tyvars [con] [''Typeable] + where (tyvars, cxt, args, stateType, resultType, isUpdate) = analyseType eventName eventType + eventStructName = mkName (structName (nameBase eventName)) + structName [] = [] + structName (x:xs) = toUpper x : xs + +-- instance (Binary key, Binary val) => Binary (MyUpdateEvent key val) where +-- put (MyUpdateEvent a b) = do put a; put b +-- get = MyUpdateEvent <$> get <*> get +makeBinaryInstance eventName eventType + = do let preds = [ ''Binary ] + ty = AppT (ConT ''Binary) (foldl AppT (ConT eventStructName) [ VarT tyvar | PlainTV tyvar <- tyvars ]) + + getBase = appE (varE 'return) (conE eventStructName) + getArgs = foldl (\a b -> infixE (Just a) (varE '(<*>)) (Just (varE 'get))) getBase args + + putVars <- replicateM (length args) (newName "arg") + let putClause = conP eventStructName [varP var | var <- putVars ] + putExp = doE $ [ noBindS $ appE (varE 'put) (varE var) | var <- putVars ] ++ + [ noBindS $ appE (varE 'return) (tupE []) ] + + instanceD (mkCxtFromTyVars preds tyvars context) + (return ty) + [ funD 'put [clause [putClause] (normalB putExp) []] + , valD (varP 'get) (normalB getArgs) [] + ] + where (tyvars, context, args, stateType, resultType, isUpdate) = analyseType eventName eventType + eventStructName = mkName (structName (nameBase eventName)) + structName [] = [] + structName (x:xs) = toUpper x : xs + +mkCxtFromTyVars preds tyvars extraContext + = cxt $ [ classP classPred [varT tyvar] | PlainTV tyvar <- tyvars, classPred <- preds ] ++ + map return extraContext + +{- +instance (Binary key, Typeable key + ,Binary val, Typeable val) => Method (MyUpdateEvent key val) where + type MethodResult (MyUpdateEvent key val) = Return + type MethodState (MyUpdateEvent key val) = State key val +-} +makeMethodInstance eventName eventType + = do let preds = [ ''Binary, ''Typeable ] + ty = AppT (ConT ''Method) (foldl AppT (ConT eventStructName) [ VarT tyvar | PlainTV tyvar <- tyvars ]) + structType = foldl appT (conT eventStructName) [ varT tyvar | PlainTV tyvar <- tyvars ] + instanceD (cxt $ [ classP classPred [varT tyvar] | PlainTV tyvar <- tyvars, classPred <- preds ] ++ map return context) + (return ty) + [ tySynInstD ''MethodResult [structType] (return resultType) + , tySynInstD ''MethodState [structType] (return stateType) + ] + where (tyvars, context, args, stateType, resultType, isUpdate) = analyseType eventName eventType + eventStructName = mkName (structName (nameBase eventName)) + structName [] = [] + structName (x:xs) = toUpper x : xs + +--instance (Binary key, Typeable key +-- ,Binary val, Typeable val) => UpdateEvent (MyUpdateEvent key val) +makeEventInstance eventName eventType + = do let preds = [ ''Binary, ''Typeable ] + eventClass = if isUpdate then ''UpdateEvent else ''QueryEvent + ty = AppT (ConT eventClass) (foldl AppT (ConT eventStructName) [ VarT tyvar | PlainTV tyvar <- tyvars ]) + structType = foldl appT (conT eventStructName) [ varT tyvar | PlainTV tyvar <- tyvars ] + instanceD (cxt $ [ classP classPred [varT tyvar] | PlainTV tyvar <- tyvars, classPred <- preds ] ++ map return context) + (return ty) + [] + where (tyvars, context, args, stateType, resultType, isUpdate) = analyseType eventName eventType + eventStructName = mkName (structName (nameBase eventName)) + structName [] = [] + structName (x:xs) = toUpper x : xs + + +-- (tyvars, cxt, args, state type, result type, is update) +analyseType :: Name -> Type -> ([TyVarBndr], Cxt, [Type], Type, Type, Bool) +analyseType eventName t + = let (tyvars, cxt, t') = case t of + ForallT binds [] t' -> + (binds, [], t') + ForallT binds cxt t' -> + error $ "Context restrictions on events aren't supported yet: " ++ show eventName + _ -> ([], [], t) + args = getArgs t' + (stateType, resultType, isUpdate) = findMonad t' + in (tyvars, cxt, args, stateType, resultType, isUpdate) + where getArgs ForallT{} = error $ "Event has an invalid type signature: Nested forall: " ++ show eventName + getArgs (AppT (AppT ArrowT a) b) = a : getArgs b + getArgs _ = [] + + findMonad (AppT (AppT ArrowT a) b) + = findMonad b + findMonad (AppT (AppT (ConT con) state) result) + | con == ''Update = (state, result, True) + | con == ''Query = (state, result, False) + findMonad _ = error $ "Event has an invalid type signature: Not an Update or a Query: " ++ show eventName + +makeAcidic' :: [Name] -> Name -> [TyVarBndr] -> [Con] -> Q [Dec] +makeAcidic' eventNames stateName tyvars constructors + = do events <- sequence [ makeEvent eventName stateName | eventName <- eventNames ] + acidic <- makeIsAcidic eventNames stateName tyvars constructors + return $ acidic : concat events diff --git a/src/Data/State/Acid.hs b/src/Data/State/Acid.hs deleted file mode 100644 --- a/src/Data/State/Acid.hs +++ /dev/null @@ -1,5 +0,0 @@ -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 deleted file mode 100644 --- a/src/Data/State/Acid/Archive.hs +++ /dev/null @@ -1,73 +0,0 @@ -{- -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 deleted file mode 100644 --- a/src/Data/State/Acid/CRC.hs +++ /dev/null @@ -1,58 +0,0 @@ -{- 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 deleted file mode 100644 --- a/src/Data/State/Acid/Core.hs +++ /dev/null @@ -1,164 +0,0 @@ -{-# 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 deleted file mode 100644 --- a/src/Data/State/Acid/Local.hs +++ /dev/null @@ -1,183 +0,0 @@ -{-# 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 deleted file mode 100644 --- a/src/Data/State/Acid/Log.hs +++ /dev/null @@ -1,169 +0,0 @@ -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)