acid-state 0.5.0 → 0.5.1
raw patch · 7 files changed
+169/−50 lines, 7 filesdep +Win32dep ~basePVP ok
version bump matches the API change (PVP)
Dependencies added: Win32
Dependency ranges changed: base
API changes (from Hackage documentation)
+ Data.Acid: createArchive :: AcidState st -> IO ()
+ Data.Acid.Local: createArchive :: AcidState st -> IO ()
Files
- acid-state.cabal +15/−4
- src-unix/FileIO.hs +31/−0
- src-win32/FileIO.hs +27/−0
- src/Data/Acid.hs +1/−0
- src/Data/Acid/Archive.hs +1/−1
- src/Data/Acid/Local.hs +27/−0
- src/Data/Acid/Log.hs +67/−45
acid-state.cabal view
@@ -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.5.0+Version: 0.5.1 -- A short (one-line) description of the package. Synopsis: Add ACID guarantees to any serializable Haskell data structure.@@ -37,7 +37,7 @@ -- Extra files to be distributed with the package, such as examples or -- a README.-Extra-source-files: examples/*.hs, examples/errors/*.hs+Extra-source-files: examples/*.hs, examples/errors/*.hs, src-win32/*.hs, src-unix/*.hs -- Constraint on the version of Cabal needed to build this package. Cabal-version: >=1.6@@ -52,13 +52,24 @@ -- Modules not exported by this package. Other-modules: Data.Acid.Log, Data.Acid.Archive, Data.Acid.CRC, Paths_acid_state,- Data.Acid.TemplateHaskell, Data.Acid.Common+ Data.Acid.TemplateHaskell, Data.Acid.Common, FileIO -- Packages needed in order to build this package. Build-depends: base >= 4 && < 5, cereal >= 0.3.2.0, safecopy >= 0.5, bytestring, stm,- filepath, directory, mtl, array, containers, template-haskell, unix+ filepath, directory, mtl, array, containers, template-haskell + if os(windows)+ Build-depends: Win32+ else+ Build-depends: unix+ Hs-Source-Dirs: src/++ if os(windows)+ Hs-Source-Dirs: src-win32/+ else+ Hs-Source-Dirs: src-unix/+ GHC-Options: -fwarn-unused-imports -fwarn-unused-binds
+ src-unix/FileIO.hs view
@@ -0,0 +1,31 @@+{-# LANGUAGE ForeignFunctionInterface #-}+module FileIO(FHandle,open,write,flush,close) where+import System.Posix(Fd(Fd),+ openFd,+ fdWriteBuf,+ closeFd,+ OpenMode(WriteOnly),+ defaultFileFlags,+ stdFileMode+ )+import Data.Word(Word8,Word32)+import Foreign(Ptr)+import Foreign.C(CInt)++data FHandle = FHandle Fd++-- should handle opening flags correctly+open :: FilePath -> IO FHandle+open filename = fmap FHandle $ openFd filename WriteOnly (Just stdFileMode) defaultFileFlags++write :: FHandle -> Ptr Word8 -> Word32 -> IO Word32+write (FHandle fd) data' length = fmap fromIntegral $ fdWriteBuf fd data' $ fromIntegral length++-- Handle error values?+flush :: FHandle -> IO ()+flush (FHandle (Fd c_fd)) = c_fsync c_fd >> return ()++foreign import ccall "fsync" c_fsync :: CInt -> IO CInt++close :: FHandle -> IO ()+close (FHandle fd) = closeFd fd
+ src-win32/FileIO.hs view
@@ -0,0 +1,27 @@+module FileIO(FHandle,open,write,flush,close) where+import System.Win32(HANDLE,+ createFile,+ gENERIC_WRITE,+ fILE_SHARE_NONE,+ cREATE_ALWAYS,+ fILE_ATTRIBUTE_NORMAL,+ win32_WriteFile,+ flushFileBuffers,+ closeHandle)+import Data.Word(Word8,Word32)+import Foreign(Ptr)++data FHandle = FHandle HANDLE++open :: FilePath -> IO FHandle+open filename =+ fmap FHandle $ createFile filename gENERIC_WRITE fILE_SHARE_NONE Nothing cREATE_ALWAYS fILE_ATTRIBUTE_NORMAL Nothing++write :: FHandle -> Ptr Word8 -> Word32 -> IO Word32+write (FHandle handle) data' length = win32_WriteFile handle data' length Nothing ++flush :: FHandle -> IO ()+flush (FHandle handle) = flushFileBuffers handle++close :: FHandle -> IO ()+close (FHandle handle) = closeHandle handle
src/Data/Acid.hs view
@@ -20,6 +20,7 @@ , closeAcidState , createCheckpoint , createCheckpointAndClose+ , createArchive , update , query , update'
src/Data/Acid/Archive.hs view
@@ -29,7 +29,7 @@ entriesToList :: Entries -> [Entry] entriesToList Done = [] entriesToList (Next entry next) = entry : entriesToList next-entriesToList (Fail msg) = fail msg+entriesToList (Fail msg) = error msg entriesToListNoFail :: Entries -> [Entry] entriesToListNoFail Done = []
src/Data/Acid/Local.hs view
@@ -28,6 +28,7 @@ , closeAcidState , createCheckpoint , createCheckpointAndClose+ , createArchive , update , scheduleUpdate , query@@ -231,3 +232,29 @@ closeFileLog (localEvents acidState) closeFileLog (localCheckpoints acidState) ++-- | Move all log files that are no longer necessary for state restoration into the 'Archive'+-- folder in the state directory. This folder can then be backed up or thrown out as you see fit.+-- Reverting to a state before the last checkpoint will not be possible if the 'Archive' folder+-- has been thrown out.+-- +-- This method is idempotent and does not block the normal operation of the AcidState.+createArchive :: AcidState st -> IO ()+createArchive state+ = do -- We need to look at the last checkpoint saved to disk. Since checkpoints can be written+ -- in parallel with this call, we can't guarantee that the checkpoint we get really is the+ -- last one but that's alright.+ currentCheckpointId <- cutFileLog (localCheckpoints state)+ -- 'currentCheckpointId' is the ID of the next checkpoint that will be written to disk.+ -- 'currentCheckpointId-1' must then be the ID of a checkpoint on disk (or -1, of course).+ let durableCheckpointId = currentCheckpointId-1+ checkpoints <- readEntriesFrom (localCheckpoints state) durableCheckpointId+ case checkpoints of+ [] -> return ()+ (Checkpoint entryId _content : _)+ -> do -- 'entryId' is the lowest entryId that didn't contribute to the checkpoint.+ -- 'archiveFileLog' moves all files that are lower than this entryId to the archive.+ archiveFileLog (localEvents state) entryId+ -- In the same style as above, we archive all log files that came before the log file+ -- which contains our checkpoint.+ archiveFileLog (localCheckpoints state) durableCheckpointId
src/Data/Acid/Log.hs view
@@ -14,15 +14,15 @@ , readEntriesFrom , newestEntry , askCurrentEntryId+ , cutFileLog+ , archiveFileLog ) where import Data.Acid.Archive as Archive import System.Directory import System.FilePath-import System.IO-import System.Posix ( handleToFd, Fd(..), fdWriteBuf- , closeFd )-import Foreign.C+import FileIO+ import Foreign.Ptr import Control.Monad import Control.Concurrent@@ -45,7 +45,7 @@ data FileLog object = FileLog { logIdentifier :: LogKey object- , logCurrent :: MVar Fd -- Handle+ , logCurrent :: MVar FHandle -- Handle , logNextEntryId :: TVar EntryId , logQueue :: TVar ([Lazy.ByteString], [IO ()]) , logThreads :: [ThreadId]@@ -92,21 +92,17 @@ , logThreads = [tid2] } if null logFiles then do let currentEntryId = 0- currentHandle <- openBinaryFile (logDirectory identifier </> formatLogFile (logPrefix identifier) currentEntryId) WriteMode- fd <- handleToFd currentHandle- putMVar currentState fd+ handle <- open (logDirectory identifier </> formatLogFile (logPrefix identifier) currentEntryId)+ putMVar currentState handle else do let (lastFileEntryId, lastFilePath) = maximum logFiles entries <- readEntities lastFilePath let currentEntryId = lastFileEntryId + length entries atomically $ writeTVar nextEntryRef currentEntryId- currentHandle <- openBinaryFile (logDirectory identifier </> formatLogFile (logPrefix identifier) currentEntryId) WriteMode- fd <- handleToFd currentHandle- putMVar currentState fd+ handle <- open (logDirectory identifier </> formatLogFile (logPrefix identifier) currentEntryId)+ putMVar currentState handle return fLog -foreign import ccall "fsync" c_fsync :: CInt -> IO CInt--fileWriter :: MVar Fd -> TVar ([Lazy.ByteString], [IO ()]) -> IO ()+fileWriter :: MVar FHandle -> TVar ([Lazy.ByteString], [IO ()]) -> IO () fileWriter currentState queue = forever $ do (entries, actions) <- atomically $ do (entries, actions) <- readTVar queue@@ -129,15 +125,15 @@ | otherwise = Strict.concat (Lazy.toChunks (Lazy.take blockSize bs)) : worker (Lazy.drop blockSize bs) blockSize = 4*1024 -writeToDisk :: Fd -> [Strict.ByteString] -> IO ()+writeToDisk :: FHandle -> [Strict.ByteString] -> IO () writeToDisk _ [] = return ()-writeToDisk fd@(Fd c_fd) xs+writeToDisk handle xs = do mapM_ worker xs- c_fsync c_fd+ flush handle return () where worker bs = do let len = Strict.length bs- count <- Strict.unsafeUseAsCString bs $ \ptr -> fdWriteBuf fd (castPtr ptr) (fromIntegral len)+ count <- Strict.unsafeUseAsCString bs $ \ptr -> write handle (castPtr ptr) (fromIntegral len) if fromIntegral count < len then worker (Strict.drop (fromIntegral count) bs) else return ()@@ -145,8 +141,8 @@ closeFileLog :: FileLog object -> IO () closeFileLog fLog- = modifyMVar_ (logCurrent fLog) $ \fd ->- do closeFd fd+ = modifyMVar_ (logCurrent fLog) $ \handle ->+ do close handle _ <- forkIO $ forM_ (logThreads fLog) killThread return $ error "FileLog has been closed" @@ -157,7 +153,7 @@ where worker Done = [] worker (Next entry next) = entry : worker next- worker Fail{} = []+ worker (Fail msg) = error msg -- Read all durable entries younger than the given EntryId. -- Note that entries written during or after this call won't@@ -170,25 +166,10 @@ -- We're interested in these entries: youngestEntry <= x < entryCap. logFiles <- findLogFiles (logIdentifier fLog) 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+ relevant = filterLogFiles (Just youngestEntry) (Just entryCap) sorted firstEntryId = case relevant of [] -> 0 ( logFile : _logFiles) -> rangeStart logFile- archive <- liftM Lazy.concat $ mapM (Lazy.readFile . snd) relevant let entries = entriesToList $ readEntries archive return $ map decode'@@ -197,17 +178,58 @@ where rangeStart (firstEntryId, _path) = firstEntryId +-- Filter out log files that are outside the min_entry/max_entry range.+-- minEntryId <= x < maxEntryId+filterLogFiles :: Maybe EntryId -> Maybe EntryId -> [(EntryId, FilePath)] -> [(EntryId, FilePath)]+filterLogFiles minEntryIdMb maxEntryIdMb logFiles+ = worker logFiles+ where worker [] = []+ worker [ logFile ]+ | ltMaxEntryId (rangeStart logFile) -- If the logfile starts before our maxEntryId then we're intersted.+ = [ logFile ]+ | otherwise+ = []+ worker ( left : right : xs)+ | ltMinEntryId (rangeStart right) -- If 'right' starts before our minEntryId then we can discard 'left'.+ = worker (right : xs)+ | ltMaxEntryId (rangeStart left) -- If 'left' starts before our maxEntryId then we're interested.+ = left : worker (right : xs)+ | otherwise -- If 'left' starts after our maxEntryId then we're done.+ = []+ ltMinEntryId = case minEntryIdMb of Nothing -> const False+ Just minEntryId -> \entryId -> entryId < minEntryId+ ltMaxEntryId = case maxEntryIdMb of Nothing -> const True+ Just maxEntryId -> \entryId -> entryId < maxEntryId+ rangeStart (firstEntryId, _path) = firstEntryId +-- Move all log files that do not contain entries equal or higher than the given entryId+-- into an Archive/ directory.+archiveFileLog :: FileLog object -> EntryId -> IO ()+archiveFileLog fLog entryId+ = do logFiles <- findLogFiles (logIdentifier fLog)+ let sorted = sort logFiles+ relevant = filterLogFiles Nothing (Just entryId) sorted \\+ filterLogFiles (Just (entryId+1)) (Just entryId) sorted++ createDirectoryIfMissing True archiveDir+ forM_ relevant $ \(_startEntry, logFilePath) ->+ renameFile logFilePath (archiveDir </> takeFileName logFilePath)+ where archiveDir = logDirectory (logIdentifier fLog) </> "Archive"++getNextDurableEntryId :: FileLog object -> IO EntryId+getNextDurableEntryId fLog+ = atomically $+ do (entries, _) <- readTVar (logQueue fLog)+ next <- readTVar (logNextEntryId fLog)+ return (next - length entries)+ cutFileLog :: FileLog object -> IO EntryId cutFileLog fLog = do mvar <- newEmptyMVar- let action = do currentEntryId <- atomically $- do (entries, _) <- readTVar (logQueue fLog)- next <- readTVar (logNextEntryId fLog)- return (next - length entries)+ let action = do currentEntryId <- getNextDurableEntryId fLog modifyMVar_ (logCurrent fLog) $ \old ->- do closeFd old- handleToFd =<< openBinaryFile (logDirectory key </> formatLogFile (logPrefix key) currentEntryId) WriteMode+ do close old+ open (logDirectory key </> formatLogFile (logPrefix key) currentEntryId) putMVar mvar currentEntryId pushAction fLog action takeMVar mvar@@ -230,9 +252,9 @@ = case Archive.readEntries archive of Done -> worker archives Next entry next -> Just (decode' (lastEntry entry next))- Fail{} -> worker archives+ Fail msg -> error msg lastEntry entry Done = entry- lastEntry entry Fail{} = entry+ lastEntry entry (Fail msg) = error msg lastEntry _ (Next entry next) = lastEntry entry next -- Schedule a new log entry. This call does not block