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.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 diff --git a/src-unix/FileIO.hs b/src-unix/FileIO.hs new file mode 100644 --- /dev/null +++ b/src-unix/FileIO.hs @@ -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 diff --git a/src-win32/FileIO.hs b/src-win32/FileIO.hs new file mode 100644 --- /dev/null +++ b/src-win32/FileIO.hs @@ -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 diff --git a/src/Data/Acid.hs b/src/Data/Acid.hs --- a/src/Data/Acid.hs +++ b/src/Data/Acid.hs @@ -20,6 +20,7 @@ , closeAcidState , createCheckpoint , createCheckpointAndClose + , createArchive , update , query , update' diff --git a/src/Data/Acid/Archive.hs b/src/Data/Acid/Archive.hs --- a/src/Data/Acid/Archive.hs +++ b/src/Data/Acid/Archive.hs @@ -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 = [] diff --git a/src/Data/Acid/Local.hs b/src/Data/Acid/Local.hs --- a/src/Data/Acid/Local.hs +++ b/src/Data/Acid/Local.hs @@ -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 diff --git a/src/Data/Acid/Log.hs b/src/Data/Acid/Log.hs --- a/src/Data/Acid/Log.hs +++ b/src/Data/Acid/Log.hs @@ -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