acid-state 0.11.2 → 0.11.3
raw patch · 3 files changed
+218/−208 lines, 3 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- acid-state.cabal +3/−3
- benchmarks/loading/Benchmark.hs +10/−10
- src/Data/Acid/Log.hs +205/−195
acid-state.cabal view
@@ -1,5 +1,5 @@ Name: acid-state-Version: 0.11.2+Version: 0.11.3 Synopsis: Add ACID guarantees to any serializable Haskell data structure. Description: Use regular Haskell data structures as your database and get stronger ACID guarantees than most RDBMS offer. Homepage: http://acid-state.seize.it/@@ -17,8 +17,8 @@ src-unix/*.hs Source-repository head- type: darcs- location: http://hub.darcs.net/Lemmih/acid-state+ type: git+ location: https://github.com/acid-state/acid-state Library Exposed-Modules: Data.Acid,
benchmarks/loading/Benchmark.hs view
@@ -10,8 +10,8 @@ main :: IO () main = do- - workingPath <- do ++ workingPath <- do workingPath <- FS.getTemporaryDirectory rndStr <- replicateM 16 $ Random.randomRIO ('a', 'z') return $ workingPath <> "acid-state" <> "benchmarks" <> "loading" <> FS.decodeString rndStr@@ -22,10 +22,10 @@ defaultMain =<< sequence [+ prepareBenchmarksGroup workingPath $ 100, prepareBenchmarksGroup workingPath $ 200,- prepareBenchmarksGroup workingPath $ 400,- prepareBenchmarksGroup workingPath $ 600,- prepareBenchmarksGroup workingPath $ 800+ prepareBenchmarksGroup workingPath $ 300,+ prepareBenchmarksGroup workingPath $ 400 ] @@ -41,10 +41,10 @@ FS.createTree logsInstancePath FS.createTree checkpointInstancePath- + putStrLn "Initializing" inst <- initialize checkpointInstancePath size- + putStrLn "Copying" FS.copy checkpointInstancePath logsInstancePath FS.removeFile $ logsInstancePath <> "open.lock"@@ -55,11 +55,11 @@ putStrLn "Closing" Acid.closeAcidState inst - return $ bgroup (show size) + return $ bgroup (show size) [- bench "From Logs" $ nfIO $ + bench "From Logs" $ nfIO $ load logsInstancePath >>= Acid.closeAcidState,- bench "From Checkpoint" $ nfIO $ + bench "From Checkpoint" $ nfIO $ load checkpointInstancePath >>= Acid.closeAcidState ]
src/Data/Acid/Log.hs view
@@ -64,94 +64,97 @@ formatLogFile = printf "%s-%010d.log" 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 ]+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"+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 fLog = FileLog { logIdentifier = identifier- , logCurrent = currentState- , logNextEntryId = nextEntryRef- , logQueue = queue- , logThreads = [tid2] }- if null logFiles- then do let currentEntryId = 0- 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- handle <- open (logDirectory identifier </> formatLogFile (logPrefix identifier) currentEntryId)- putMVar currentState handle- return fLog+openFileLog identifier = do+ logFiles <- findLogFiles identifier+ saveVersionFile identifier+ currentState <- newEmptyMVar+ queue <- newTVarIO ([], [])+ nextEntryRef <- newTVarIO 0+ tid2 <- forkIO $ fileWriter currentState queue+ let fLog = FileLog { logIdentifier = identifier+ , logCurrent = currentState+ , logNextEntryId = nextEntryRef+ , logQueue = queue+ , logThreads = [tid2] }+ if null logFiles+ then do let currentEntryId = 0+ 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+ handle <- open (logDirectory identifier </> formatLogFile (logPrefix identifier) currentEntryId)+ putMVar currentState handle+ return fLog fileWriter :: MVar FHandle -> TVar ([Lazy.ByteString], [IO ()]) -> IO ()-fileWriter currentState queue- = forever $- do (entries, actions) <- atomically $ do (entries, actions) <- readTVar queue- when (null entries && null actions) retry- writeTVar queue ([], [])- return (reverse entries, reverse actions)- withMVar currentState $ \fd ->- do let arch = Archive.packEntries entries- writeToDisk fd (repack arch)- sequence_ actions- yield+fileWriter currentState queue = forever $ do+ (entries, actions) <- atomically $ do+ (entries, actions) <- readTVar queue+ when (null entries && null actions) retry+ writeTVar queue ([], [])+ return (reverse entries, reverse actions)+ withMVar currentState $ \fd -> do+ let arch = Archive.packEntries entries+ writeToDisk fd (repack arch)+ sequence_ actions+ yield -- Repack a lazy bytestring into larger blocks that can be efficiently written to disk. repack :: Lazy.ByteString -> [Strict.ByteString] repack = worker- where worker bs- | Lazy.null bs = []- | otherwise = Strict.concat (Lazy.toChunks (Lazy.take blockSize bs)) : worker (Lazy.drop blockSize bs)- blockSize = 4*1024+ where+ worker bs+ | Lazy.null bs = []+ | otherwise = Strict.concat (Lazy.toChunks (Lazy.take blockSize bs)) : worker (Lazy.drop blockSize bs)+ blockSize = 4*1024 writeToDisk :: FHandle -> [Strict.ByteString] -> IO () writeToDisk _ [] = return ()-writeToDisk handle xs- = do mapM_ worker xs- flush handle- where worker bs- = do let len = Strict.length bs- count <- Strict.unsafeUseAsCString bs $ \ptr -> write handle (castPtr ptr) (fromIntegral len)- when (fromIntegral count < len) $- worker (Strict.drop (fromIntegral count) bs)+writeToDisk handle xs = do+ mapM_ worker xs+ flush handle+ where+ worker bs = do+ let len = Strict.length bs+ count <- Strict.unsafeUseAsCString bs $ \ptr -> write handle (castPtr ptr) (fromIntegral len)+ when (fromIntegral count < len) $+ worker (Strict.drop (fromIntegral count) bs) closeFileLog :: FileLog object -> IO ()-closeFileLog fLog- = modifyMVar_ (logCurrent fLog) $ \handle ->- do close handle- _ <- forkIO $ forM_ (logThreads fLog) killThread- return $ error "FileLog has been closed"+closeFileLog fLog =+ modifyMVar_ (logCurrent fLog) $ \handle -> do+ close handle+ _ <- forkIO $ forM_ (logThreads fLog) 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 msg) = error msg+readEntities path = do+ archive <- Lazy.readFile path+ return $ worker (Archive.readEntries archive)+ where+ worker Done = []+ worker (Next entry next) = entry : worker next+ worker (Fail msg) = error msg ensureLeastEntryId :: FileLog object -> EntryId -> IO () ensureLeastEntryId fLog youngestEntry = do@@ -165,119 +168,123 @@ -- Note that entries written during or after this call won't -- be included in the returned list. readEntriesFrom :: SafeCopy object => FileLog object -> EntryId -> IO [object]-readEntriesFrom fLog youngestEntry- = do -- Cut the log so we can read written entries without interfering- -- with the writing of new entries.- entryCap <- cutFileLog fLog- -- We're interested in these entries: youngestEntry <= x < entryCap.- logFiles <- findLogFiles (logIdentifier fLog)- let sorted = sort logFiles- 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'- $ take (entryCap - youngestEntry) -- Take events under the eventCap.- $ drop (youngestEntry - firstEntryId) entries -- Drop entries that are too young.- where rangeStart (firstEntryId, _path) = firstEntryId+readEntriesFrom fLog youngestEntry = do+ -- Cut the log so we can read written entries without interfering+ -- with the writing of new entries.+ entryCap <- cutFileLog fLog+ -- We're interested in these entries: youngestEntry <= x < entryCap.+ logFiles <- findLogFiles (logIdentifier fLog)+ let sorted = sort logFiles+ relevant = filterLogFiles (Just youngestEntry) (Just entryCap) sorted+ firstEntryId = case relevant of+ [] -> 0+ ( logFile : _logFiles) -> rangeStart logFile+ -- XXX: Strict bytestrings are used due to a performance bug in+ -- cereal-0.3.5.2 and binary-0.7.1.0. The code should revert back+ -- to lazy bytestrings once the bug has been fixed.+ archive <- liftM Lazy.fromChunks $ mapM (Strict.readFile . 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 -- Obliterate log entries younger than or equal to the EventId. Very unsafe, can't be undone rollbackTo :: SafeCopy object => LogKey object -> EntryId -> IO ()-rollbackTo identifier youngestEntry- = do logFiles <- findLogFiles identifier- let sorted = sort logFiles- loop [] = return ()- loop ((rangeStart, path) : xs)- = if rangeStart >= youngestEntry- then do {-putStrLn ("Removing file: "++path); -}removeFile path; loop xs- else do archive <- Strict.readFile path- pathHandle <- openFile path WriteMode- let entries = entriesToList $ readEntries (Lazy.fromChunks [archive])- entriesToKeep = take (youngestEntry - rangeStart + 1) entries- lengthToKeep = Lazy.length (packEntries entriesToKeep)- --putStrLn $ "FileSize: " ++ show (lengthToKeep, Strict.length archive)- hSetFileSize pathHandle (fromIntegral lengthToKeep)- hClose pathHandle- loop (reverse sorted)+rollbackTo identifier youngestEntry = do+ logFiles <- findLogFiles identifier+ let sorted = sort logFiles+ loop [] = return ()+ loop ((rangeStart, path) : xs)+ | rangeStart >= youngestEntry = removeFile path >> loop xs+ | otherwise = do+ archive <- Strict.readFile path+ pathHandle <- openFile path WriteMode+ let entries = entriesToList $ readEntries (Lazy.fromChunks [archive])+ entriesToKeep = take (youngestEntry - rangeStart + 1) entries+ lengthToKeep = Lazy.length (packEntries entriesToKeep)+ hSetFileSize pathHandle (fromIntegral lengthToKeep)+ hClose pathHandle+ loop (reverse sorted) -- Obliterate log entries as long as the filterFn returns True. rollbackWhile :: SafeCopy object => LogKey object -> (object -> Bool) -> IO ()-rollbackWhile identifier filterFn- = do logFiles <- findLogFiles identifier- let sorted = sort logFiles- loop [] = return ()- loop ((_rangeStart, path) : xs)- = do archive <- Strict.readFile path- let entries = entriesToList $ readEntries (Lazy.fromChunks [archive])- entriesToSkip = takeWhile (filterFn . decode') $ reverse entries- skip_size = Lazy.length (packEntries entriesToSkip)- orig_size = fromIntegral $ Strict.length archive- new_size = orig_size - skip_size- if new_size == 0- then do removeFile path; loop xs- else do pathHandle <- openFile path WriteMode- hSetFileSize pathHandle (fromIntegral new_size)- hClose pathHandle- loop (reverse sorted)+rollbackWhile identifier filterFn = do+ logFiles <- findLogFiles identifier+ let sorted = sort logFiles+ loop [] = return ()+ loop ((_rangeStart, path) : xs) = do+ archive <- Strict.readFile path+ let entries = entriesToList $ readEntries (Lazy.fromChunks [archive])+ entriesToSkip = takeWhile (filterFn . decode') $ reverse entries+ skip_size = Lazy.length (packEntries entriesToSkip)+ orig_size = fromIntegral $ Strict.length archive+ new_size = orig_size - skip_size+ if new_size == 0+ then do removeFile path; loop xs+ else do pathHandle <- openFile path WriteMode+ hSetFileSize pathHandle (fromIntegral new_size)+ hClose pathHandle+ loop (reverse sorted) -- 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 -> (<= minEntryId)- ltMaxEntryId = case maxEntryIdMb of Nothing -> const True- Just maxEntryId -> (< maxEntryId)- rangeStart (firstEntryId, _path) = firstEntryId+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 -> (<= minEntryId)+ ltMaxEntryId = case maxEntryIdMb of Nothing -> const True+ Just maxEntryId -> (< 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+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"+ 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)+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 <- getNextDurableEntryId fLog- modifyMVar_ (logCurrent fLog) $ \old ->- do close old- open (logDirectory key </> formatLogFile (logPrefix key) currentEntryId)- putMVar mvar currentEntryId- pushAction fLog action- takeMVar mvar- where key = logIdentifier fLog+cutFileLog fLog = do+ mvar <- newEmptyMVar+ let action = do currentEntryId <- getNextDurableEntryId fLog+ modifyMVar_ (logCurrent fLog) $ \old ->+ do close old+ open (logDirectory key </> formatLogFile (logPrefix key) currentEntryId)+ putMVar mvar currentEntryId+ pushAction fLog action+ takeMVar mvar+ where+ key = logIdentifier fLog -- Finds the newest entry in the log. Doesn't work on open logs. -- Do not use after the log has been opened.@@ -285,49 +292,52 @@ -- containing at least one valid entry is found, -- return the last entry in that file. newestEntry :: SafeCopy object => LogKey object -> IO (Maybe object)-newestEntry identifier- = do logFiles <- findLogFiles identifier- let sorted = reverse $ sort logFiles- (_eventIds, files) = unzip sorted- worker files- where worker [] = return Nothing- worker (logFile:logFiles)- = do archive <- fmap Lazy.fromStrict $ Strict.readFile logFile- case Archive.readEntries archive of- Done -> worker logFiles- Next entry next -> return $ Just (decode' (lastEntry entry next))- Fail msg -> error msg- lastEntry entry Done = entry- lastEntry entry (Fail msg) = error msg- lastEntry _ (Next entry next) = lastEntry entry next+newestEntry identifier = do+ logFiles <- findLogFiles identifier+ let sorted = reverse $ sort logFiles+ (_eventIds, files) = unzip sorted+ worker files+ where+ worker [] = return Nothing+ worker (logFile:logFiles) = do+ -- XXX: Strict bytestrings are used due to a performance bug in+ -- cereal-0.3.5.2 and binary-0.7.1.0. The code should revert back+ -- to lazy bytestrings once the bug has been fixed.+ archive <- fmap Lazy.fromStrict $ Strict.readFile logFile+ case Archive.readEntries archive of+ Done -> worker logFiles+ Next entry next -> return $ Just (decode' (lastEntry entry next))+ Fail msg -> error msg+ lastEntry entry Done = entry+ lastEntry entry (Fail msg) = error msg+ 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 :: SafeCopy object => FileLog object -> object -> IO () -> IO ()-pushEntry fLog object finally- = atomically $- do tid <- readTVar (logNextEntryId fLog)- writeTVar (logNextEntryId fLog) (tid+1)- (entries, actions) <- readTVar (logQueue fLog)- writeTVar (logQueue fLog) ( encoded : entries, finally : actions )- where encoded = Lazy.fromChunks [ Strict.copy $ Put.runPut (safePut object) ]+pushEntry fLog object finally = atomically $ do+ tid <- readTVar (logNextEntryId fLog)+ writeTVar (logNextEntryId fLog) (tid+1)+ (entries, actions) <- readTVar (logQueue fLog)+ writeTVar (logQueue fLog) ( encoded : entries, finally : actions )+ where+ encoded = Lazy.fromChunks [ Strict.copy $ Put.runPut (safePut object) ] -- The given IO action is executed once all previous entries are durable. pushAction :: FileLog object -> IO () -> IO ()-pushAction fLog finally- = atomically $- do (entries, actions) <- readTVar (logQueue fLog)- writeTVar (logQueue fLog) (entries, finally : actions)+pushAction fLog finally = atomically $ do+ (entries, actions) <- readTVar (logQueue fLog)+ writeTVar (logQueue fLog) (entries, finally : actions) askCurrentEntryId :: FileLog object -> IO EntryId-askCurrentEntryId fLog- = atomically $ readTVar (logNextEntryId fLog)+askCurrentEntryId fLog = atomically $+ readTVar (logNextEntryId fLog) -- FIXME: Check for unused input. decode' :: SafeCopy object => Lazy.ByteString -> object-decode' inp- = case Get.runGetLazy safeGet inp of- Left msg -> error msg- Right val -> val+decode' inp =+ case Get.runGetLazy safeGet inp of+ Left msg -> error msg+ Right val -> val