diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -2,7 +2,6 @@ ====== [![Travis](https://travis-ci.org/haskell-haskey/haskey.svg?branch=master)](https://travis-ci.org/haskell-haskey/haskey) -[![Coverage](https://coveralls.io/repos/github/haskell-haskey/haskey/badge.svg?branch=master)](https://coveralls.io/github/haskell-haskey/haskey?branch=master) [![Hackage](https://img.shields.io/hackage/v/haskey.svg?maxAge=2592000)](https://hackage.haskell.org/package/haskey) [![Stackage Nightly](http://stackage.org/package/haskey/badge/nightly)](http://stackage.org/nightly/package/haskey) [![Stackage LTS](http://stackage.org/package/haskey/badge/lts)](http://stackage.org/lts/package/haskey) @@ -12,9 +11,15 @@ [soh2017]: https://summer.haskell.org/news/2017-05-24-accepted-projects.html -An introductory blog post on Haskey can be found [here][introduction]. +## Tutorial - [introduction]: https://deliquus.com/posts/2017-08-24-introducing-haskey.html +A full tutorial can be [found in the haskey-mtl library](https://github.com/haskell-haskey/haskey-mtl/blob/master/docs/tutorial.md), along with a [full code example](https://github.com/haskell-haskey/haskey-mtl/tree/feature/tutorial/example). -_Disclaimer: Haskey is not production ready yet. We are still actively making -changes to the public API and the internals, as well as the binary format._ +## Historical blog posts +Some blog posts have been written on Haskey's design an internals. These give an insight in the inner workings of Haskey, but the used APIs might be a bit outdated. + + - An introductory blog post on Haskey can be found [here][introduction]. + - An blog post on user-defined schemas and using the `HaskeyT` monad can be found [here][haskey-mtl-post] + + [introduction]: https://deliquus.com/posts/2017-08-24-introducing-haskey.html + [haskey-mtl-post]: https://deliquus.com/posts/2017-09-14-haskey-user-defined-schemas-and-monad-transformers.html diff --git a/example/Main.hs b/example/Main.hs --- a/example/Main.hs +++ b/example/Main.hs @@ -5,11 +5,11 @@ import Control.Monad (void, replicateM) import Control.Monad.Catch (bracket_, finally) -import Data.BTree.Impure (Tree, toList, insertTree) +import Data.BTree.Impure (Tree) import Data.ByteString (ByteString) import Data.Int (Int32) import Data.Text.Encoding (encodeUtf8) -import qualified Data.BTree.Impure as Tree +import qualified Data.BTree.Impure as B import qualified Data.Text as Text import Database.Haskey.Alloc.Concurrent (ConcurrentDb, @@ -64,7 +64,7 @@ where bs = encodeUtf8 $ Text.pack (show i) - tx tree = insertTree i bs tree >>= commit_ + tx tree = B.insert i bs tree >>= commit_ reader :: MemoryFiles FilePath -> ConcurrentDb Root @@ -72,14 +72,14 @@ -> IO () reader files db delay = void $ replicateM 10 $ do threadDelay delay - runDatabase files $ transactReadOnly toList db + runDatabase files $ transactReadOnly B.toList db openOrCreate :: MemoryFiles FilePath -> IO (ConcurrentDb Root) openOrCreate store = runDatabase store $ do maybeDb <- openConcurrentDb handles case maybeDb of - Nothing -> createConcurrentDb handles Tree.empty + Nothing -> createConcurrentDb handles B.empty Just db -> return db runDatabase :: MemoryFiles FilePath @@ -111,20 +111,20 @@ where bs = encodeUtf8 $ Text.pack (show i) - tx tree = insertTree i bs tree >>= commit_ + tx tree = B.insert i bs tree >>= commit_ reader :: ConcurrentDb Root -> Int -> IO () reader db delay = void $ replicateM 10 $ do threadDelay delay - runDatabase $ transactReadOnly toList db + runDatabase $ transactReadOnly B.toList db openOrCreate :: IO (ConcurrentDb Root) openOrCreate = runDatabase $ do maybeDb <- openConcurrentDb handles case maybeDb of - Nothing -> createConcurrentDb handles Tree.empty + Nothing -> createConcurrentDb handles B.empty Just db -> return db runDatabase :: Monad m diff --git a/haskey.cabal b/haskey.cabal --- a/haskey.cabal +++ b/haskey.cabal @@ -1,5 +1,5 @@ name: haskey -version: 0.2.0.1 +version: 0.3.0.0 synopsis: A transactional, ACID compliant, embeddable key-value store. description: Haskey is a transactional, ACID compliant, embeddable, scalable key-value @@ -22,14 +22,14 @@ library exposed-modules: Database.Haskey.Alloc.Concurrent - Database.Haskey.Alloc.Concurrent.Database - Database.Haskey.Alloc.Concurrent.Environment - Database.Haskey.Alloc.Concurrent.FreePages.Query - Database.Haskey.Alloc.Concurrent.FreePages.Save - Database.Haskey.Alloc.Concurrent.FreePages.Tree - Database.Haskey.Alloc.Concurrent.Meta - Database.Haskey.Alloc.Concurrent.Monad - Database.Haskey.Alloc.Concurrent.Overflow + Database.Haskey.Alloc.Concurrent.Internal.Database + Database.Haskey.Alloc.Concurrent.Internal.Environment + Database.Haskey.Alloc.Concurrent.Internal.FreePages.Query + Database.Haskey.Alloc.Concurrent.Internal.FreePages.Save + Database.Haskey.Alloc.Concurrent.Internal.FreePages.Tree + Database.Haskey.Alloc.Concurrent.Internal.Meta + Database.Haskey.Alloc.Concurrent.Internal.Monad + Database.Haskey.Alloc.Concurrent.Internal.Overflow Database.Haskey.Alloc.Transaction Database.Haskey.Store Database.Haskey.Store.Class @@ -65,7 +65,7 @@ exceptions >=0.8.3 && <0.9, filepath >=1.4 && <2, focus >=0.1.2 && <0.2, - haskey-btree >=0.2.0.1 && <1, + haskey-btree >=0.3.0.0 && <0.4, list-t >=0.2 && <2, lz4 >=0.2 && <1, mtl >=2.1 && <3, diff --git a/src/Database/Haskey/Alloc/Concurrent.hs b/src/Database/Haskey/Alloc/Concurrent.hs --- a/src/Database/Haskey/Alloc/Concurrent.hs +++ b/src/Database/Haskey/Alloc/Concurrent.hs @@ -25,7 +25,7 @@ , ConcurrentMetaStoreM(..) ) where -import Database.Haskey.Alloc.Concurrent.Database -import Database.Haskey.Alloc.Concurrent.Meta -import Database.Haskey.Alloc.Concurrent.Monad +import Database.Haskey.Alloc.Concurrent.Internal.Database +import Database.Haskey.Alloc.Concurrent.Internal.Meta +import Database.Haskey.Alloc.Concurrent.Internal.Monad import Database.Haskey.Alloc.Transaction diff --git a/src/Database/Haskey/Alloc/Concurrent/Database.hs b/src/Database/Haskey/Alloc/Concurrent/Database.hs deleted file mode 100644 --- a/src/Database/Haskey/Alloc/Concurrent/Database.hs +++ /dev/null @@ -1,441 +0,0 @@ -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE RecordWildCards #-} --- | This module implements data structures and functions related to the database. -module Database.Haskey.Alloc.Concurrent.Database where - -import Control.Applicative ((<$>)) -import Control.Concurrent.STM -import Control.Monad (void, unless) -import Control.Monad.IO.Class -import Control.Monad.Catch (MonadCatch, MonadMask, SomeException, - catch, mask, onException, bracket, bracket_) -import Control.Monad.State -import Control.Monad.Trans (lift) - -import Data.Proxy (Proxy(..)) -import Data.List.NonEmpty (NonEmpty((:|))) -import Data.Maybe (fromMaybe) - -import STMContainers.Map (Map) -import qualified STMContainers.Map as Map - -import Data.BTree.Alloc.Class -import Data.BTree.Impure -import Data.BTree.Primitives - -import Database.Haskey.Alloc.Concurrent.Environment -import Database.Haskey.Alloc.Concurrent.FreePages.Save -import Database.Haskey.Alloc.Concurrent.Meta -import Database.Haskey.Alloc.Concurrent.Monad -import Database.Haskey.Alloc.Concurrent.Overflow -import Database.Haskey.Alloc.Transaction -import Database.Haskey.Store -import Database.Haskey.Utils.RLock -import qualified Database.Haskey.Utils.STM.Map as Map - --- | An active concurrent database. --- --- This can be shared amongst threads. -data ConcurrentDb root = ConcurrentDb - { concurrentDbHandles :: ConcurrentHandles - , concurrentDbWriterLock :: RLock - , concurrentDbCurrentMeta :: TVar CurrentMetaPage - , concurrentDbMeta1 :: TVar (ConcurrentMeta root) - , concurrentDbMeta2 :: TVar (ConcurrentMeta root) - , concurrentDbReaders :: Map TxId Integer - } - --- | Lock the database. --- --- This needs to be called manually, if you want exclusive access, before --- calling either 'createConcurrentDb' or 'openConcurrentDb' --- --- Use 'unlockConcurrentDb' using the 'bracket' pattern to properly unlock the --- database. -lockConcurrentDb :: ConcurrentMetaStoreM m => ConcurrentHandles -> m () -lockConcurrentDb = lockHandle . concurrentHandlesRoot - --- | Unlock the database. -unlockConcurrentDb :: ConcurrentMetaStoreM m => ConcurrentHandles -> m () -unlockConcurrentDb = releaseHandle . concurrentHandlesRoot - --- | Open all concurrent handles. -openConcurrentHandles :: ConcurrentMetaStoreM m - => ConcurrentHandles -> m () -openConcurrentHandles ConcurrentHandles{..} = do - openHandle concurrentHandlesData - openHandle concurrentHandlesIndex - openHandle concurrentHandlesMetadata1 - openHandle concurrentHandlesMetadata2 - --- | Open a new concurrent database, with the given handles. -createConcurrentDb :: (Root root, MonadIO m, MonadMask m, ConcurrentMetaStoreM m) - => ConcurrentHandles - -> root - -> m (ConcurrentDb root) -createConcurrentDb hnds root = - bracket_ (openConcurrentHandles hnds) - (closeConcurrentHandles hnds) $ do - - db <- newConcurrentDb hnds meta0 - setCurrentMeta meta0 db - setCurrentMeta meta0 db - return db - where - meta0 = ConcurrentMeta { - concurrentMetaRevision = 0 - , concurrentMetaDataNumPages = DataState 0 - , concurrentMetaIndexNumPages = IndexState 0 - , concurrentMetaRoot = root - , concurrentMetaDataFreeTree = DataState $ Tree zeroHeight Nothing - , concurrentMetaIndexFreeTree = IndexState $ Tree zeroHeight Nothing - , concurrentMetaOverflowTree = Tree zeroHeight Nothing - , concurrentMetaDataCachedFreePages = DataState [] - , concurrentMetaIndexCachedFreePages = IndexState [] - } - --- | Open the an existing database, with the given handles. -openConcurrentDb :: (Root root, MonadIO m, MonadMask m, ConcurrentMetaStoreM m) - => ConcurrentHandles - -> m (Maybe (ConcurrentDb root)) -openConcurrentDb hnds@ConcurrentHandles{..} = - bracket_ (openConcurrentHandles hnds) - (closeConcurrentHandles hnds) $ do - - m1 <- readConcurrentMeta concurrentHandlesMetadata1 Proxy - m2 <- readConcurrentMeta concurrentHandlesMetadata2 Proxy - maybeDb <- case (m1, m2) of - (Nothing, Nothing) -> return Nothing - (Just m , Nothing) -> Just <$> newConcurrentDb hnds m - (Nothing, Just m ) -> Just <$> newConcurrentDb hnds m - (Just x , Just y ) -> if concurrentMetaRevision x > concurrentMetaRevision y - then Just <$> newConcurrentDb hnds x - else Just <$> newConcurrentDb hnds y - case maybeDb of - Nothing -> return Nothing - Just db -> do - meta <- liftIO . atomically $ getCurrentMeta db - cleanupAfterException hnds (concurrentMetaRevision meta + 1) - return (Just db) - --- | Close the handles of the database. -closeConcurrentHandles :: (MonadIO m, ConcurrentMetaStoreM m) - => ConcurrentHandles - -> m () -closeConcurrentHandles ConcurrentHandles{..} = do - closeHandle concurrentHandlesData - closeHandle concurrentHandlesIndex - closeHandle concurrentHandlesMetadata1 - closeHandle concurrentHandlesMetadata2 - --- | Create a new concurrent database with handles and metadata provided. -newConcurrentDb :: (Root root, MonadIO m) - => ConcurrentHandles - -> ConcurrentMeta root - -> m (ConcurrentDb root) -newConcurrentDb hnds meta0 = do - readers <- liftIO Map.newIO - meta <- liftIO $ newTVarIO Meta1 - lock <- liftIO newRLock - meta1 <- liftIO $ newTVarIO meta0 - meta2 <- liftIO $ newTVarIO meta0 - return $! ConcurrentDb - { concurrentDbHandles = hnds - , concurrentDbWriterLock = lock - , concurrentDbCurrentMeta = meta - , concurrentDbMeta1 = meta1 - , concurrentDbMeta2 = meta2 - , concurrentDbReaders = readers - } - --- | Get the current meta data. -getCurrentMeta :: Root root - => ConcurrentDb root - -> STM (ConcurrentMeta root) -getCurrentMeta db - | ConcurrentDb { concurrentDbCurrentMeta = v } <- db - = readTVar v >>= \case - Meta1 -> readTVar $ concurrentDbMeta1 db - Meta2 -> readTVar $ concurrentDbMeta2 db - --- | Write the new metadata, and switch the pointer to the current one. -setCurrentMeta :: (Root root, MonadIO m, ConcurrentMetaStoreM m) - => ConcurrentMeta root - -> ConcurrentDb root - -> m () -setCurrentMeta new db - | ConcurrentDb - { concurrentDbCurrentMeta = v - , concurrentDbHandles = hnds - } <- db - = liftIO (atomically $ readTVar v) >>= \case - Meta1 -> do - flushHandle (concurrentHandlesData hnds) - flushHandle (concurrentHandlesIndex hnds) - putConcurrentMeta (concurrentHandlesMetadata2 hnds) new - flushHandle (concurrentHandlesMetadata2 hnds) - liftIO . atomically $ do - writeTVar v Meta2 - writeTVar (concurrentDbMeta2 db) new - Meta2 -> do - flushHandle (concurrentHandlesData hnds) - flushHandle (concurrentHandlesIndex hnds) - putConcurrentMeta (concurrentHandlesMetadata1 hnds) new - flushHandle (concurrentHandlesMetadata1 hnds) - liftIO . atomically $ do - writeTVar v Meta1 - writeTVar (concurrentDbMeta1 db) new - --- | Execute a write transaction, with a result. -transact :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) - => (forall n. (AllocM n, MonadMask n) => root -> n (Transaction root a)) - -> ConcurrentDb root - -> m a -transact act db = withRLock (concurrentDbWriterLock db) $ do - cleanup - transactNow act db - where - cleanup :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) => m () - cleanup = actAndCommit db $ \meta -> do - v <- deleteOutdatedOverflowIds (concurrentMetaOverflowTree meta) - case v of - Nothing -> return (Nothing, ()) - Just tree -> do - let meta' = meta { concurrentMetaOverflowTree = tree } - return (Just meta', ()) - --- | Execute a write transaction, without cleaning up old overflow pages. -transactNow :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) - => (forall n. (AllocM n, MonadMask n) => root -> n (Transaction root a)) - -> ConcurrentDb root - -> m a -transactNow act db = withRLock (concurrentDbWriterLock db) $ - actAndCommit db $ \meta -> do - tx <- act (concurrentMetaRoot meta) - case tx of - Abort v -> return (Nothing, v) - Commit root v -> - let meta' = meta { concurrentMetaRoot = root } in - return (Just meta', v) - --- | Execute a write transaction, without a result. -transact_ :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) - => (forall n. (AllocM n, MonadMask n) => root -> n (Transaction root ())) - -> ConcurrentDb root - -> m () -transact_ act db = void $ transact act db - --- | Execute a read-only transaction. -transactReadOnly :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) - => (forall n. (AllocReaderM n, MonadMask n) => root -> n a) - -> ConcurrentDb root - -> m a -transactReadOnly act db = - bracket_ (openConcurrentHandles hnds) - (closeConcurrentHandles hnds) $ - - bracket acquireMeta - releaseMeta $ - \meta -> evalConcurrentT (act $ concurrentMetaRoot meta) - (ReaderEnv hnds) - where - hnds = concurrentDbHandles db - readers = concurrentDbReaders db - - addOne Nothing = Just 1 - addOne (Just x) = Just $! x + 1 - subOne Nothing = Nothing - subOne (Just 0) = Nothing - subOne (Just 1) = Nothing - subOne (Just x) = Just $! x - 1 - - acquireMeta = liftIO . atomically $ do - meta <- getCurrentMeta db - Map.alter (concurrentMetaRevision meta) addOne readers - return meta - - releaseMeta meta = - let rev = concurrentMetaRevision meta in - liftIO . atomically $ Map.alter rev subOne readers - --------------------------------------------------------------------------------- - --- | Run a write action that takes the current meta-data and returns new --- meta-data to be commited, or 'Nothing' if the write transaction should be --- aborted. -actAndCommit :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) - => ConcurrentDb root - -> (forall n. (MonadIO n, MonadMask n, ConcurrentMetaStoreM n) - => ConcurrentMeta root - -> ConcurrentT WriterEnv ConcurrentHandles n (Maybe (ConcurrentMeta root), a) - ) - -> m a -actAndCommit db act - | ConcurrentDb - { concurrentDbHandles = hnds - , concurrentDbWriterLock = lock - , concurrentDbReaders = readers - } <- db - = withRLock lock $ - bracket_ (openConcurrentHandles hnds) - (closeConcurrentHandles hnds) $ do - - meta <- liftIO . atomically $ getCurrentMeta db - let newRevision = concurrentMetaRevision meta + 1 - wrap hnds newRevision $ do - ((maybeMeta, v), env) <- runConcurrentT (act meta) $ - newWriter hnds - newRevision - readers - (concurrentMetaDataNumPages meta) - (concurrentMetaIndexNumPages meta) - (concurrentMetaDataCachedFreePages meta) - (concurrentMetaIndexCachedFreePages meta) - (concurrentMetaDataFreeTree meta) - (concurrentMetaIndexFreeTree meta) - - let maybeMeta' = updateMeta env <$> maybeMeta - - case maybeMeta' of - Nothing -> do - removeNewlyAllocatedOverflows env - return v - - Just meta' -> do - -- Bookkeeping - (newMeta, _) <- flip execStateT (meta', env) $ do - saveOverflowIds - saveFreePages' 0 DataState - writerDataFileState - (\e s -> e { writerDataFileState = s }) - saveFreePages' 0 IndexState - writerIndexFileState - (\e s -> e { writerIndexFileState = s }) - handleCachedFreePages - - -- Commit - setCurrentMeta (newMeta { concurrentMetaRevision = newRevision }) - db - return v - where - wrap :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) - => ConcurrentHandles - -> TxId - -> m a - -> m a - wrap hnds tx action = mask $ \restore -> - restore action `onException` cleanupAfterException hnds tx - --- | Cleanup after an exception occurs, or after a program crash. --- --- The 'TxId' of the aborted transaction should be passed. -cleanupAfterException :: (MonadIO m, MonadCatch m, ConcurrentMetaStoreM m) - => ConcurrentHandles - -> TxId - -> m () -cleanupAfterException hnds tx = do - let dir = getOverflowDir (concurrentHandlesOverflowDir hnds) tx - overflows <- filter filter' <$> listOverflows dir - mapM_ (\fp -> removeHandle fp `catch` ignore) overflows - where - filter' fp = fromMaybe False $ (== tx) . fst <$> readOverflowId fp - - ignore :: Monad m => SomeException -> m () - ignore _ = return () - --- | Remove all overflow pages that were written in the transaction. --- --- If the transaction is aborted, all written pages should be deleted. -removeNewlyAllocatedOverflows :: (MonadIO m, ConcurrentMetaStoreM m) - => WriterEnv ConcurrentHandles - -> m () -removeNewlyAllocatedOverflows env = do - let root = concurrentHandlesOverflowDir (writerHnds env) - sequence_ [ delete root (i - 1) | i <- [1..(writerOverflowCounter env)] ] - where - delete root c = do - let i = (writerTxId env, c) - removeHandle (getOverflowHandle root i) - --- | Update the meta-data from a writer environment -updateMeta :: WriterEnv ConcurrentHandles -> ConcurrentMeta root -> ConcurrentMeta root -updateMeta env m = m { - concurrentMetaDataFreeTree = fileStateFreeTree (writerDataFileState env) - , concurrentMetaIndexFreeTree = fileStateFreeTree (writerIndexFileState env) } - - --- | Save the newly free'd overflow pages, for deletion on the next tx. -saveOverflowIds :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) - => StateT (ConcurrentMeta root, WriterEnv ConcurrentHandles) m () -saveOverflowIds = do - (meta, env) <- get - case map (\(OldOverflow i) ->i) (writerRemovedOverflows env) of - [] -> return () - x:xs -> do - (tree', env') <- lift $ flip runConcurrentT env $ - insertOverflowIds (writerTxId env) - (x :| xs) - (concurrentMetaOverflowTree meta) - let meta' = (updateMeta env meta) - { concurrentMetaOverflowTree = tree' } - put (meta', env') - --- | Save the free'd pages to the free page database -saveFreePages' :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) - => Int - -> (forall a. a -> S t a) - -> (forall hnds. WriterEnv hnds -> FileState t) - -> (forall hnds. WriterEnv hnds -> FileState t -> WriterEnv hnds) - -> StateT (ConcurrentMeta root, WriterEnv ConcurrentHandles) m () -saveFreePages' paranoid cons getState setState - {- paranoid >= 100 = error "paranoid: looping!" - | otherwise-} - = do - - (meta, env) <- get - let tx = writerTxId env - (tree', envWithoutTree) <- lift $ - runConcurrentT (saveFreePages tx (getState env)) $ - env { writerQueryFreeTreeOn = False } - - let state' = (getState envWithoutTree) { fileStateFreeTree = cons tree' } - let env' = setState envWithoutTree state' - let meta' = updateMeta env' meta - put (meta', env') - - -- Did we free any new pages? We have to put them in the free tree! - unless (fileStateNewlyFreedPages state' == fileStateNewlyFreedPages (getState env)) $ - saveFreePages' (paranoid + 1) cons getState setState - --- | Handle the cached free pages. --- --- Save the cached free pages to the metadata for later use. --- --- Update the database size. -handleCachedFreePages :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) - => StateT (ConcurrentMeta root, WriterEnv ConcurrentHandles) m () -handleCachedFreePages = do - (meta, env) <- get - - let dataEnv = writerDataFileState env - let indexEnv = writerIndexFileState env - - let meta' = meta { concurrentMetaDataNumPages = - fileStateNewNumPages dataEnv - , concurrentMetaDataFreeTree = - fileStateFreeTree dataEnv - , concurrentMetaDataCachedFreePages = - fileStateCachedFreePages dataEnv - - , concurrentMetaIndexNumPages = - fileStateNewNumPages indexEnv - , concurrentMetaIndexFreeTree = - fileStateFreeTree indexEnv - , concurrentMetaIndexCachedFreePages = - fileStateCachedFreePages indexEnv - } - put (meta', env) - --------------------------------------------------------------------------------- diff --git a/src/Database/Haskey/Alloc/Concurrent/Environment.hs b/src/Database/Haskey/Alloc/Concurrent/Environment.hs deleted file mode 100644 --- a/src/Database/Haskey/Alloc/Concurrent/Environment.hs +++ /dev/null @@ -1,257 +0,0 @@ -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE MultiWayIf #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE StandaloneDeriving #-} --- | Environments of a read or write transaction. -module Database.Haskey.Alloc.Concurrent.Environment where - -import Control.Applicative ((<$>)) -import Control.Monad.State - -import Data.Binary (Binary) -import Data.Set (Set) -import Data.Typeable (Typeable) -import Data.Word (Word32) -import qualified Data.Binary as B -import qualified Data.Set as S - -import STMContainers.Map (Map) - -import Data.BTree.Primitives - -import Database.Haskey.Alloc.Concurrent.FreePages.Tree - -data StateType = TypeData - | TypeIndex - --- | Wrapper around a type to indicate it belongs to a file with either --- data/leaf nodes or index nodes. -data S (t :: StateType) a where - DataState :: a -> S 'TypeData a - IndexState :: a -> S 'TypeIndex a - deriving (Typeable) - -deriving instance Show a => Show (S t a) - -instance Binary a => Binary (S 'TypeData a) where - put (DataState a) = B.put a - get = DataState <$> B.get - -instance Binary a => Binary (S 'TypeIndex a) where - put (IndexState a) = B.put a - get = IndexState <$> B.get - -instance Functor (S t) where - f `fmap` (DataState v) = DataState (f v) - f `fmap` (IndexState v) = IndexState (f v) - -getSValue :: S t a -> a -getSValue (DataState a) = a -getSValue (IndexState a) = a - -newtype ReaderEnv hnds = ReaderEnv { readerHnds :: hnds } - -data FileState stateType = FileState { - fileStateNewlyFreedPages :: ![NewlyFreed] - -- ^ Pages free'd in this transaction, not ready for reuse until the - -- transaction is commited. - - , fileStateOriginalNumPages :: !(S stateType PageId) - -- ^ The original number of pages in the file, before the transaction - -- started. - - , fileStateNewNumPages :: !(S stateType PageId) - -- ^ The new uncommited number of pages in the file. - -- - -- All pages in the range 'fileStateOriginalNumPages' to - -- 'fileStateNewNumPages' (excluding) are freshly allocated in the - -- ongoing transaction. - - , fileStateDirtyPages :: !(Set PageId) - -- ^ Pages written to in this transaction. - - , fileStateFreeTree :: !(S stateType FreeTree) - -- ^ The root of the free tree, might change during a transaction. - - , fileStateCachedFreePages :: !(S stateType [FreePage]) - -- ^ All pages that are immediately ready for reuse in this and any - -- subsequent transactions. - } - -data WriterEnv hnds = WriterEnv - { writerHnds :: !hnds - , writerTxId :: !TxId - , writerReaders :: Map TxId Integer - - , writerIndexFileState :: FileState 'TypeIndex - -- ^ State of the file with index nodes. - - , writerDataFileState :: FileState 'TypeData - -- ^ State of the file with data/leaf nodes. - - , writerQueryFreeTreeOn :: !Bool - -- ^ Whether or not querying free pages from the free is enabled. - - , writerDirtyOverflows :: !(Set DirtyOverflow) - -- ^ Newly allocated overflow pages in this transaction. - - , writerOverflowCounter :: !Word32 - -- ^ Counts how many overflow pages were already allocated in this transaction. - - , writerRemovedOverflows :: ![OldOverflow] - -- ^ Old overflow pages that were removed in this transaction - -- and should be deleted when no longer in use. - } - --- | Create a new writer. -newWriter :: hnd -> TxId -> Map TxId Integer - -> S 'TypeData PageId -> S 'TypeIndex PageId - -> S 'TypeData [FreePage] -> S 'TypeIndex [FreePage] - -> S 'TypeData FreeTree -> S 'TypeIndex FreeTree - -> WriterEnv hnd -newWriter hnd tx readers - numDataPages numIndexPages - dataFreePages indexFreePages - dataFreeTree indexFreeTree = - WriterEnv { - writerHnds = hnd - , writerTxId = tx - , writerReaders = readers - - , writerIndexFileState = newFileState numIndexPages indexFreePages indexFreeTree - , writerDataFileState = newFileState numDataPages dataFreePages dataFreeTree - - , writerQueryFreeTreeOn = True - , writerDirtyOverflows = S.empty - , writerOverflowCounter = 0 - , writerRemovedOverflows = [] - } - where - newFileState numPages freePages freeTree = FileState { - fileStateNewlyFreedPages = [] - , fileStateOriginalNumPages = numPages - , fileStateNewNumPages = numPages - , fileStateDirtyPages = S.empty - , fileStateCachedFreePages = freePages - , fileStateFreeTree = freeTree - } - --- | Wrapper around 'PageId' indicating it is newly free'd and cannot be reused --- in the same transaction. -newtype NewlyFreed = NewlyFreed PageId deriving (Eq, Ord, Show) - --- | Wrapper around 'PageId' indicating it is free and can be reused in any --- transaction. -newtype FreePage = FreePage PageId deriving (Binary, Eq, Ord, Show) - --- | Wrapper around 'PageId' indicating that it is dirty, i.e. written to in --- this transaction. -newtype Dirty = Dirty PageId deriving (Eq, Ord, Show) - --- | Try to free a page, given a set of dirty pages. --- --- If the page was dirty, a 'FreePage' page is added to the environment, if --- not a 'NewlyFreed' page is added to the environment. --- --- Btw, give me lenses... -freePage :: (Functor m, MonadState (WriterEnv hnd) m) => S stateType PageId -> m () -freePage pid@(DataState pid') = do - dirty' <- dirty pid - modify' $ \e -> - e { writerDataFileState = - updateFileState (writerDataFileState e) DataState - dirty' pid' - } - -freePage pid@(IndexState pid') = do - dirty' <- dirty pid - modify' $ \e -> - e { writerIndexFileState = - updateFileState (writerIndexFileState e) IndexState - dirty' pid' - } - -updateFileState :: FileState t - -> (forall a. a -> S t a) - -> Maybe Dirty - -> PageId - -> FileState t -updateFileState e cons dirty' pid' = - if | Just (Dirty p) <- dirty' -> - e { fileStateCachedFreePages = - cons $ FreePage p : getSValue (fileStateCachedFreePages e) } - - | p <- pid' -> - e { fileStateNewlyFreedPages = - NewlyFreed p : fileStateNewlyFreedPages e } - --- | Get a 'Dirty' page, by first proving it is in fact dirty. -dirty :: (Functor m, MonadState (WriterEnv hnd) m) => S stateType PageId -> m (Maybe Dirty) -dirty pid = case pid of - DataState p -> (page p . fileStateDirtyPages . writerDataFileState) <$> get - IndexState p -> (page p . fileStateDirtyPages . writerIndexFileState) <$> get - where - page p dirtyPages - | p `S.member` dirtyPages = Just (Dirty p) - | otherwise = Nothing - --- | Touch a fresh page, make it dirty. --- --- We really need lenses... -touchPage :: MonadState (WriterEnv hnd) m => S stateType PageId -> m () -touchPage (DataState pid) = do - modify' $ \e -> - let dirtyPages = fileStateDirtyPages (writerDataFileState e) in - e { writerDataFileState = (writerDataFileState e) { - fileStateDirtyPages = S.insert pid dirtyPages } - } - modify' $ \e -> - let oldNum = getSValue $ fileStateNewNumPages (writerDataFileState e) - newNum = max oldNum (pid + 1) - in e { writerDataFileState = (writerDataFileState e) { - fileStateNewNumPages = DataState newNum } - } - -touchPage (IndexState pid) = do - modify' $ \e -> - let dirtyPages = fileStateDirtyPages (writerIndexFileState e) in - e { writerIndexFileState = (writerIndexFileState e) { - fileStateDirtyPages = S.insert pid dirtyPages } - } - modify' $ \e -> - let oldNum = getSValue $ fileStateNewNumPages (writerIndexFileState e) - newNum = max oldNum (pid + 1) - in e { writerIndexFileState = (writerIndexFileState e) { - fileStateNewNumPages = IndexState newNum } - } - --- | Wrapper around 'OverflowId' indicating that it is dirty. -newtype DirtyOverflow = DirtyOverflow OverflowId deriving (Eq, Ord, Show) - --- | Wrapper around 'OverflowId' indicating that it is an overflow --- page from a previous transaction. -newtype OldOverflow = OldOverflow OverflowId deriving (Eq, Ord, Show) - --- | Touch a fresh overflow page, making it dirty. -touchOverflow :: MonadState (WriterEnv hnd) m => OverflowId -> m () -touchOverflow i = modify' $ - \e -> e { writerDirtyOverflows = - S.insert (DirtyOverflow i) (writerDirtyOverflows e) } - --- | Get the type of the overflow page. -overflowType :: MonadState (WriterEnv hnd) m => OverflowId -> m (Either DirtyOverflow OldOverflow) -overflowType i = do - dirty' <- gets $ \e -> S.member (DirtyOverflow i) (writerDirtyOverflows e) - if dirty' then return $ Left (DirtyOverflow i) - else return $ Right (OldOverflow i) - --- | Free an old overflow page. -removeOldOverflow :: MonadState (WriterEnv hdn) m => OldOverflow -> m () -removeOldOverflow i = - modify' $ \e -> e { writerRemovedOverflows = i : writerRemovedOverflows e } diff --git a/src/Database/Haskey/Alloc/Concurrent/FreePages/Query.hs b/src/Database/Haskey/Alloc/Concurrent/FreePages/Query.hs deleted file mode 100644 --- a/src/Database/Haskey/Alloc/Concurrent/FreePages/Query.hs +++ /dev/null @@ -1,152 +0,0 @@ -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE RankNTypes #-} -module Database.Haskey.Alloc.Concurrent.FreePages.Query where - -import Control.Applicative ((<|>), (<$>)) -import Control.Concurrent.STM -import Control.Monad.State -import Control.Monad.Trans.Maybe - -import Data.List.NonEmpty (NonEmpty((:|))) -import qualified Data.List.NonEmpty as NE - -import Data.BTree.Alloc.Class -import Data.BTree.Impure -import Data.BTree.Impure.NonEmpty -import Data.BTree.Primitives - -import Database.Haskey.Alloc.Concurrent.Environment -import Database.Haskey.Alloc.Concurrent.FreePages.Tree -import Database.Haskey.Utils.Monad (ifM) -import qualified Database.Haskey.Utils.STM.Map as Map - --- | Get a free page. --- --- First try to get one from the in-memory dirty pages. Then try to get one --- from the in-memory free page cache stored in 'fileStateCachedFreePages'. If --- that one is empty, actually query one from the free database. -getFreePageId :: (Functor m, AllocM m, MonadIO m, MonadState (WriterEnv hnd) m) - => S stateType () - -> m (Maybe PageId) -getFreePageId t = - runMaybeT $ MaybeT (getCachedFreePageId t) - <|> MaybeT (queryNewFreePageIds t) - --- | Get a cached free page. --- --- Get a free page from the free database cache stored in --- 'fileStateCachedFreePages'. -getCachedFreePageId :: (Functor m, MonadState (WriterEnv hnd) m) - => S stateType () - -> m (Maybe PageId) -getCachedFreePageId stateType = - case stateType of - DataState () -> do - s <- writerDataFileState <$> get - let (pid, s') = query DataState s - modify' $ \env -> env { writerDataFileState = s' } - return pid - IndexState () -> do - s <- writerIndexFileState <$> get - let (pid, s') = query IndexState s - modify' $ \env -> env { writerIndexFileState = s' } - return pid - where - query :: (forall a. a -> S t a) - -> FileState t - -> (Maybe PageId, FileState t) - query cons env = case getSValue $ fileStateCachedFreePages env of - [] -> (Nothing, env) - FreePage pid : pageIds -> - let env' = env { fileStateCachedFreePages = cons pageIds } in - (Just pid, env') - --- | Try to get a list of free pages from the free page database, return the --- first free one for immediate use, and store the rest in the environment. --- --- Immediately remove the queried free pages from the free tree. -queryNewFreePageIds :: (AllocM m, MonadIO m, MonadState (WriterEnv hnd) m) - => S stateType () - -> m (Maybe PageId) -queryNewFreePageIds stateType = ifM (not . writerQueryFreeTreeOn <$> get) (return Nothing) $ do - flag <- case stateType of - DataState () -> - query DataState - writerDataFileState - (\e s -> e { writerDataFileState = s }) - - IndexState () -> - query IndexState - writerIndexFileState - (\e s -> e { writerIndexFileState = s }) - - if flag then getFreePageId stateType - else return Nothing - where - query :: (AllocM m, MonadIO m, MonadState (WriterEnv hnd) m) - => (forall a. a -> S t a) - -> (forall h. WriterEnv h -> FileState t) - -> (forall h. WriterEnv h -> FileState t -> WriterEnv h) - -> m Bool - query cons getState setState = do - tree <- gets $ getSValue . fileStateFreeTree . getState - - -- Lookup the oldest free page - lookupValidFreePageIds tree >>= \case - Nothing -> return False - Just (txId, x :| xs) -> do - -- Save them for reuse - modify' $ \e -> - let s = getState e - pids = map FreePage (x:xs) - in setState e $ - s { fileStateCachedFreePages = - cons $ pids ++ getSValue (fileStateCachedFreePages s) } - - -- Remove the entry from the tree - modify' $ \e -> e { writerQueryFreeTreeOn = False } - tree' <- txId `deleteSubtree` tree - modify' $ \e -> e { writerQueryFreeTreeOn = True } - - -- Update the tree - modify' $ \e -> setState e $ - (getState e) { fileStateFreeTree = cons tree' } - - return True - --- | Lookup a list of free pages from the free page database, guaranteed to be old enough. -lookupValidFreePageIds :: (MonadIO m, AllocReaderM m, MonadState (WriterEnv hnd) m) - => FreeTree - -> m (Maybe (TxId, NonEmpty PageId)) -lookupValidFreePageIds tree = runMaybeT $ - MaybeT (lookupFreePageIds tree) >>= (MaybeT . checkFreePages) - --- | Lookup a list of free pages from the free page database. -lookupFreePageIds :: (Functor m, AllocReaderM m, MonadState (WriterEnv hnd) m) - => FreeTree - -> m (Maybe (Unchecked (TxId, NonEmpty PageId))) -lookupFreePageIds tree = lookupMinTree tree >>= \case - Nothing -> return Nothing - Just (tx, subtree) -> do - pids <- subtreeToList subtree - return . Just $ Unchecked (tx, pids) - where - subtreeToList subtree = NE.map fst <$> nonEmptyToList subtree - --- | Auxiliry type to ensure the transaction ID of free pages are checked. -newtype Unchecked a = Unchecked a - --- | Check the transaction ID of the free pages, if it's to old, return --- 'Nothing'. -checkFreePages :: (Functor m, MonadIO m, MonadState (WriterEnv hnd) m) - => Unchecked (TxId, NonEmpty PageId) - -> m (Maybe (TxId, NonEmpty PageId)) -checkFreePages (Unchecked v) = do - readers <- writerReaders <$> get - oldest <- liftIO . atomically $ Map.lookupMinKey readers - tx <- writerTxId <$> get - if maybe True (> fst v) oldest && fst v + 1 < tx - then return (Just v) - else return Nothing diff --git a/src/Database/Haskey/Alloc/Concurrent/FreePages/Save.hs b/src/Database/Haskey/Alloc/Concurrent/FreePages/Save.hs deleted file mode 100644 --- a/src/Database/Haskey/Alloc/Concurrent/FreePages/Save.hs +++ /dev/null @@ -1,33 +0,0 @@ -module Database.Haskey.Alloc.Concurrent.FreePages.Save where - -import Data.List.NonEmpty (NonEmpty((:|))) - -import Data.BTree.Alloc.Class -import Data.BTree.Primitives - -import Database.Haskey.Alloc.Concurrent.Environment -import Database.Haskey.Alloc.Concurrent.FreePages.Tree - --- | Save the free pages from the dirty page list and the free page --- cache. -saveFreePages :: AllocM m - => TxId - -> FileState t - -> m FreeTree -saveFreePages tx env = saveNewlyFreedPages tx env tree - where - tree = getSValue $ fileStateFreeTree env - --- | Save the newly free pages of the current transaction, as stored by --- 'fileStateNewlyFreedPages'. -saveNewlyFreedPages :: AllocM m - => TxId - -> FileState t - -> FreeTree - -> m FreeTree -saveNewlyFreedPages tx env tree = - case newlyFreed of - [] -> deleteSubtree tx tree - x:xs -> replaceSubtree tx (x :| xs) tree - where - newlyFreed = map (\(NewlyFreed pid) -> pid) $ fileStateNewlyFreedPages env diff --git a/src/Database/Haskey/Alloc/Concurrent/FreePages/Tree.hs b/src/Database/Haskey/Alloc/Concurrent/FreePages/Tree.hs deleted file mode 100644 --- a/src/Database/Haskey/Alloc/Concurrent/FreePages/Tree.hs +++ /dev/null @@ -1,67 +0,0 @@ -{-# LANGUAGE GADTs #-} -{-# LANGUAGE LambdaCase #-} --- | Module describing the tree structure of the free page database. -module Database.Haskey.Alloc.Concurrent.FreePages.Tree where - -import Control.Monad ((>=>)) - -import Data.Foldable (traverse_) -import Data.List.NonEmpty (NonEmpty) -import qualified Data.List.NonEmpty as NE - -import Data.BTree.Alloc.Class -import Data.BTree.Impure -import Data.BTree.Impure.NonEmpty -import Data.BTree.Primitives - --- | The main tree structure of the free page database. --- --- The main free page database tree maps a 'TxId' to a 'FreeSubtree'. -type FreeTree = Tree TxId FreeSubtree - --- | the subtree structure of the free page database. --- --- Just a collection of free 'PageId's. -type FreeSubtree = NonEmptyTree PageId () - --- | Replace the subtree of a certain 'TxId'. -replaceSubtree :: AllocM m - => TxId - -> NonEmpty PageId - -> FreeTree - -> m FreeTree -replaceSubtree tx pids = deleteSubtree tx >=> insertSubtree tx pids - --- | Delete the subtree of a certain 'TxId'. --- --- The 'TxId' will not be present anymore in the free tree after this call. -deleteSubtree :: AllocM m - => TxId - -> FreeTree - -> m FreeTree -deleteSubtree tx tree = lookupTree tx tree >>= \case - Nothing -> return tree - Just (NonEmptyTree h nid) -> do - freeAllNodes h nid - deleteTree tx tree - where - freeAllNodes :: (AllocM m, Key key, Value val) - => Height h - -> NodeId h key val - -> m () - freeAllNodes h nid = readNode h nid >>= \case - Leaf _ -> freeNode h nid - Idx idx -> do - let subHgt = decrHeight h - traverse_ (freeAllNodes subHgt) idx - freeNode h nid - --- | Insert a subtree for a certain 'TxId'. -insertSubtree :: AllocM m - => TxId - -> NonEmpty PageId - -> FreeTree - -> m FreeTree -insertSubtree tx pids tree = do - subtree <- fromNonEmptyList (NE.zip pids (NE.repeat ())) - insertTree tx subtree tree diff --git a/src/Database/Haskey/Alloc/Concurrent/Internal/Database.hs b/src/Database/Haskey/Alloc/Concurrent/Internal/Database.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Internal/Database.hs @@ -0,0 +1,441 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +-- | This module implements data structures and functions related to the database. +module Database.Haskey.Alloc.Concurrent.Internal.Database where + +import Control.Applicative ((<$>)) +import Control.Concurrent.STM +import Control.Monad (void, unless) +import Control.Monad.IO.Class +import Control.Monad.Catch (MonadCatch, MonadMask, SomeException, + catch, mask, onException, bracket, bracket_) +import Control.Monad.State +import Control.Monad.Trans (lift) + +import Data.Proxy (Proxy(..)) +import Data.List.NonEmpty (NonEmpty((:|))) +import Data.Maybe (fromMaybe) + +import STMContainers.Map (Map) +import qualified STMContainers.Map as Map + +import Data.BTree.Alloc.Class +import Data.BTree.Impure (Tree(..)) +import Data.BTree.Primitives + +import Database.Haskey.Alloc.Concurrent.Internal.Environment +import Database.Haskey.Alloc.Concurrent.Internal.FreePages.Save +import Database.Haskey.Alloc.Concurrent.Internal.Meta +import Database.Haskey.Alloc.Concurrent.Internal.Monad +import Database.Haskey.Alloc.Concurrent.Internal.Overflow +import Database.Haskey.Alloc.Transaction +import Database.Haskey.Store +import Database.Haskey.Utils.RLock +import qualified Database.Haskey.Utils.STM.Map as Map + +-- | An active concurrent database. +-- +-- This can be shared amongst threads. +data ConcurrentDb root = ConcurrentDb + { concurrentDbHandles :: ConcurrentHandles + , concurrentDbWriterLock :: RLock + , concurrentDbCurrentMeta :: TVar CurrentMetaPage + , concurrentDbMeta1 :: TVar (ConcurrentMeta root) + , concurrentDbMeta2 :: TVar (ConcurrentMeta root) + , concurrentDbReaders :: Map TxId Integer + } + +-- | Lock the database. +-- +-- This needs to be called manually, if you want exclusive access, before +-- calling either 'createConcurrentDb' or 'openConcurrentDb' +-- +-- Use 'unlockConcurrentDb' using the 'bracket' pattern to properly unlock the +-- database. +lockConcurrentDb :: ConcurrentMetaStoreM m => ConcurrentHandles -> m () +lockConcurrentDb = lockHandle . concurrentHandlesRoot + +-- | Unlock the database. +unlockConcurrentDb :: ConcurrentMetaStoreM m => ConcurrentHandles -> m () +unlockConcurrentDb = releaseHandle . concurrentHandlesRoot + +-- | Open all concurrent handles. +openConcurrentHandles :: ConcurrentMetaStoreM m + => ConcurrentHandles -> m () +openConcurrentHandles ConcurrentHandles{..} = do + openHandle concurrentHandlesData + openHandle concurrentHandlesIndex + openHandle concurrentHandlesMetadata1 + openHandle concurrentHandlesMetadata2 + +-- | Open a new concurrent database, with the given handles. +createConcurrentDb :: (Root root, MonadIO m, MonadMask m, ConcurrentMetaStoreM m) + => ConcurrentHandles + -> root + -> m (ConcurrentDb root) +createConcurrentDb hnds root = + bracket_ (openConcurrentHandles hnds) + (closeConcurrentHandles hnds) $ do + + db <- newConcurrentDb hnds meta0 + setCurrentMeta meta0 db + setCurrentMeta meta0 db + return db + where + meta0 = ConcurrentMeta { + concurrentMetaRevision = 0 + , concurrentMetaDataNumPages = DataState 0 + , concurrentMetaIndexNumPages = IndexState 0 + , concurrentMetaRoot = root + , concurrentMetaDataFreeTree = DataState $ Tree zeroHeight Nothing + , concurrentMetaIndexFreeTree = IndexState $ Tree zeroHeight Nothing + , concurrentMetaOverflowTree = Tree zeroHeight Nothing + , concurrentMetaDataCachedFreePages = DataState [] + , concurrentMetaIndexCachedFreePages = IndexState [] + } + +-- | Open the an existing database, with the given handles. +openConcurrentDb :: (Root root, MonadIO m, MonadMask m, ConcurrentMetaStoreM m) + => ConcurrentHandles + -> m (Maybe (ConcurrentDb root)) +openConcurrentDb hnds@ConcurrentHandles{..} = + bracket_ (openConcurrentHandles hnds) + (closeConcurrentHandles hnds) $ do + + m1 <- readConcurrentMeta concurrentHandlesMetadata1 Proxy + m2 <- readConcurrentMeta concurrentHandlesMetadata2 Proxy + maybeDb <- case (m1, m2) of + (Nothing, Nothing) -> return Nothing + (Just m , Nothing) -> Just <$> newConcurrentDb hnds m + (Nothing, Just m ) -> Just <$> newConcurrentDb hnds m + (Just x , Just y ) -> if concurrentMetaRevision x > concurrentMetaRevision y + then Just <$> newConcurrentDb hnds x + else Just <$> newConcurrentDb hnds y + case maybeDb of + Nothing -> return Nothing + Just db -> do + meta <- liftIO . atomically $ getCurrentMeta db + cleanupAfterException hnds (concurrentMetaRevision meta + 1) + return (Just db) + +-- | Close the handles of the database. +closeConcurrentHandles :: (MonadIO m, ConcurrentMetaStoreM m) + => ConcurrentHandles + -> m () +closeConcurrentHandles ConcurrentHandles{..} = do + closeHandle concurrentHandlesData + closeHandle concurrentHandlesIndex + closeHandle concurrentHandlesMetadata1 + closeHandle concurrentHandlesMetadata2 + +-- | Create a new concurrent database with handles and metadata provided. +newConcurrentDb :: (Root root, MonadIO m) + => ConcurrentHandles + -> ConcurrentMeta root + -> m (ConcurrentDb root) +newConcurrentDb hnds meta0 = do + readers <- liftIO Map.newIO + meta <- liftIO $ newTVarIO Meta1 + lock <- liftIO newRLock + meta1 <- liftIO $ newTVarIO meta0 + meta2 <- liftIO $ newTVarIO meta0 + return $! ConcurrentDb + { concurrentDbHandles = hnds + , concurrentDbWriterLock = lock + , concurrentDbCurrentMeta = meta + , concurrentDbMeta1 = meta1 + , concurrentDbMeta2 = meta2 + , concurrentDbReaders = readers + } + +-- | Get the current meta data. +getCurrentMeta :: Root root + => ConcurrentDb root + -> STM (ConcurrentMeta root) +getCurrentMeta db + | ConcurrentDb { concurrentDbCurrentMeta = v } <- db + = readTVar v >>= \case + Meta1 -> readTVar $ concurrentDbMeta1 db + Meta2 -> readTVar $ concurrentDbMeta2 db + +-- | Write the new metadata, and switch the pointer to the current one. +setCurrentMeta :: (Root root, MonadIO m, ConcurrentMetaStoreM m) + => ConcurrentMeta root + -> ConcurrentDb root + -> m () +setCurrentMeta new db + | ConcurrentDb + { concurrentDbCurrentMeta = v + , concurrentDbHandles = hnds + } <- db + = liftIO (atomically $ readTVar v) >>= \case + Meta1 -> do + flushHandle (concurrentHandlesData hnds) + flushHandle (concurrentHandlesIndex hnds) + putConcurrentMeta (concurrentHandlesMetadata2 hnds) new + flushHandle (concurrentHandlesMetadata2 hnds) + liftIO . atomically $ do + writeTVar v Meta2 + writeTVar (concurrentDbMeta2 db) new + Meta2 -> do + flushHandle (concurrentHandlesData hnds) + flushHandle (concurrentHandlesIndex hnds) + putConcurrentMeta (concurrentHandlesMetadata1 hnds) new + flushHandle (concurrentHandlesMetadata1 hnds) + liftIO . atomically $ do + writeTVar v Meta1 + writeTVar (concurrentDbMeta1 db) new + +-- | Execute a write transaction, with a result. +transact :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) + => (forall n. (AllocM n, MonadMask n) => root -> n (Transaction root a)) + -> ConcurrentDb root + -> m a +transact act db = withRLock (concurrentDbWriterLock db) $ do + cleanup + transactNow act db + where + cleanup :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) => m () + cleanup = actAndCommit db $ \meta -> do + v <- deleteOutdatedOverflowIds (concurrentMetaOverflowTree meta) + case v of + Nothing -> return (Nothing, ()) + Just tree -> do + let meta' = meta { concurrentMetaOverflowTree = tree } + return (Just meta', ()) + +-- | Execute a write transaction, without cleaning up old overflow pages. +transactNow :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) + => (forall n. (AllocM n, MonadMask n) => root -> n (Transaction root a)) + -> ConcurrentDb root + -> m a +transactNow act db = withRLock (concurrentDbWriterLock db) $ + actAndCommit db $ \meta -> do + tx <- act (concurrentMetaRoot meta) + case tx of + Abort v -> return (Nothing, v) + Commit root v -> + let meta' = meta { concurrentMetaRoot = root } in + return (Just meta', v) + +-- | Execute a write transaction, without a result. +transact_ :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) + => (forall n. (AllocM n, MonadMask n) => root -> n (Transaction root ())) + -> ConcurrentDb root + -> m () +transact_ act db = void $ transact act db + +-- | Execute a read-only transaction. +transactReadOnly :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) + => (forall n. (AllocReaderM n, MonadMask n) => root -> n a) + -> ConcurrentDb root + -> m a +transactReadOnly act db = + bracket_ (openConcurrentHandles hnds) + (closeConcurrentHandles hnds) $ + + bracket acquireMeta + releaseMeta $ + \meta -> evalConcurrentT (act $ concurrentMetaRoot meta) + (ReaderEnv hnds) + where + hnds = concurrentDbHandles db + readers = concurrentDbReaders db + + addOne Nothing = Just 1 + addOne (Just x) = Just $! x + 1 + subOne Nothing = Nothing + subOne (Just 0) = Nothing + subOne (Just 1) = Nothing + subOne (Just x) = Just $! x - 1 + + acquireMeta = liftIO . atomically $ do + meta <- getCurrentMeta db + Map.alter (concurrentMetaRevision meta) addOne readers + return meta + + releaseMeta meta = + let rev = concurrentMetaRevision meta in + liftIO . atomically $ Map.alter rev subOne readers + +-------------------------------------------------------------------------------- + +-- | Run a write action that takes the current meta-data and returns new +-- meta-data to be commited, or 'Nothing' if the write transaction should be +-- aborted. +actAndCommit :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Root root) + => ConcurrentDb root + -> (forall n. (MonadIO n, MonadMask n, ConcurrentMetaStoreM n) + => ConcurrentMeta root + -> ConcurrentT WriterEnv ConcurrentHandles n (Maybe (ConcurrentMeta root), a) + ) + -> m a +actAndCommit db act + | ConcurrentDb + { concurrentDbHandles = hnds + , concurrentDbWriterLock = lock + , concurrentDbReaders = readers + } <- db + = withRLock lock $ + bracket_ (openConcurrentHandles hnds) + (closeConcurrentHandles hnds) $ do + + meta <- liftIO . atomically $ getCurrentMeta db + let newRevision = concurrentMetaRevision meta + 1 + wrap hnds newRevision $ do + ((maybeMeta, v), env) <- runConcurrentT (act meta) $ + newWriter hnds + newRevision + readers + (concurrentMetaDataNumPages meta) + (concurrentMetaIndexNumPages meta) + (concurrentMetaDataCachedFreePages meta) + (concurrentMetaIndexCachedFreePages meta) + (concurrentMetaDataFreeTree meta) + (concurrentMetaIndexFreeTree meta) + + let maybeMeta' = updateMeta env <$> maybeMeta + + case maybeMeta' of + Nothing -> do + removeNewlyAllocatedOverflows env + return v + + Just meta' -> do + -- Bookkeeping + (newMeta, _) <- flip execStateT (meta', env) $ do + saveOverflowIds + saveFreePages' 0 DataState + writerDataFileState + (\e s -> e { writerDataFileState = s }) + saveFreePages' 0 IndexState + writerIndexFileState + (\e s -> e { writerIndexFileState = s }) + handleCachedFreePages + + -- Commit + setCurrentMeta (newMeta { concurrentMetaRevision = newRevision }) + db + return v + where + wrap :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) + => ConcurrentHandles + -> TxId + -> m a + -> m a + wrap hnds tx action = mask $ \restore -> + restore action `onException` cleanupAfterException hnds tx + +-- | Cleanup after an exception occurs, or after a program crash. +-- +-- The 'TxId' of the aborted transaction should be passed. +cleanupAfterException :: (MonadIO m, MonadCatch m, ConcurrentMetaStoreM m) + => ConcurrentHandles + -> TxId + -> m () +cleanupAfterException hnds tx = do + let dir = getOverflowDir (concurrentHandlesOverflowDir hnds) tx + overflows <- filter filter' <$> listOverflows dir + mapM_ (\fp -> removeHandle fp `catch` ignore) overflows + where + filter' fp = fromMaybe False $ (== tx) . fst <$> readOverflowId fp + + ignore :: Monad m => SomeException -> m () + ignore _ = return () + +-- | Remove all overflow pages that were written in the transaction. +-- +-- If the transaction is aborted, all written pages should be deleted. +removeNewlyAllocatedOverflows :: (MonadIO m, ConcurrentMetaStoreM m) + => WriterEnv ConcurrentHandles + -> m () +removeNewlyAllocatedOverflows env = do + let root = concurrentHandlesOverflowDir (writerHnds env) + sequence_ [ delete root (i - 1) | i <- [1..(writerOverflowCounter env)] ] + where + delete root c = do + let i = (writerTxId env, c) + removeHandle (getOverflowHandle root i) + +-- | Update the meta-data from a writer environment +updateMeta :: WriterEnv ConcurrentHandles -> ConcurrentMeta root -> ConcurrentMeta root +updateMeta env m = m { + concurrentMetaDataFreeTree = fileStateFreeTree (writerDataFileState env) + , concurrentMetaIndexFreeTree = fileStateFreeTree (writerIndexFileState env) } + + +-- | Save the newly free'd overflow pages, for deletion on the next tx. +saveOverflowIds :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) + => StateT (ConcurrentMeta root, WriterEnv ConcurrentHandles) m () +saveOverflowIds = do + (meta, env) <- get + case map (\(OldOverflow i) ->i) (writerRemovedOverflows env) of + [] -> return () + x:xs -> do + (tree', env') <- lift $ flip runConcurrentT env $ + insertOverflowIds (writerTxId env) + (x :| xs) + (concurrentMetaOverflowTree meta) + let meta' = (updateMeta env meta) + { concurrentMetaOverflowTree = tree' } + put (meta', env') + +-- | Save the free'd pages to the free page database +saveFreePages' :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) + => Int + -> (forall a. a -> S t a) + -> (forall hnds. WriterEnv hnds -> FileState t) + -> (forall hnds. WriterEnv hnds -> FileState t -> WriterEnv hnds) + -> StateT (ConcurrentMeta root, WriterEnv ConcurrentHandles) m () +saveFreePages' paranoid cons getState setState + {- paranoid >= 100 = error "paranoid: looping!" + | otherwise-} + = do + + (meta, env) <- get + let tx = writerTxId env + (tree', envWithoutTree) <- lift $ + runConcurrentT (saveFreePages tx (getState env)) $ + env { writerQueryFreeTreeOn = False } + + let state' = (getState envWithoutTree) { fileStateFreeTree = cons tree' } + let env' = setState envWithoutTree state' + let meta' = updateMeta env' meta + put (meta', env') + + -- Did we free any new pages? We have to put them in the free tree! + unless (fileStateNewlyFreedPages state' == fileStateNewlyFreedPages (getState env)) $ + saveFreePages' (paranoid + 1) cons getState setState + +-- | Handle the cached free pages. +-- +-- Save the cached free pages to the metadata for later use. +-- +-- Update the database size. +handleCachedFreePages :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) + => StateT (ConcurrentMeta root, WriterEnv ConcurrentHandles) m () +handleCachedFreePages = do + (meta, env) <- get + + let dataEnv = writerDataFileState env + let indexEnv = writerIndexFileState env + + let meta' = meta { concurrentMetaDataNumPages = + fileStateNewNumPages dataEnv + , concurrentMetaDataFreeTree = + fileStateFreeTree dataEnv + , concurrentMetaDataCachedFreePages = + fileStateCachedFreePages dataEnv + + , concurrentMetaIndexNumPages = + fileStateNewNumPages indexEnv + , concurrentMetaIndexFreeTree = + fileStateFreeTree indexEnv + , concurrentMetaIndexCachedFreePages = + fileStateCachedFreePages indexEnv + } + put (meta', env) + +-------------------------------------------------------------------------------- diff --git a/src/Database/Haskey/Alloc/Concurrent/Internal/Environment.hs b/src/Database/Haskey/Alloc/Concurrent/Internal/Environment.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Internal/Environment.hs @@ -0,0 +1,257 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE MultiWayIf #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE StandaloneDeriving #-} +-- | Environments of a read or write transaction. +module Database.Haskey.Alloc.Concurrent.Internal.Environment where + +import Control.Applicative ((<$>)) +import Control.Monad.State + +import Data.Binary (Binary) +import Data.Set (Set) +import Data.Typeable (Typeable) +import Data.Word (Word32) +import qualified Data.Binary as B +import qualified Data.Set as S + +import STMContainers.Map (Map) + +import Data.BTree.Primitives + +import Database.Haskey.Alloc.Concurrent.Internal.FreePages.Tree + +data StateType = TypeData + | TypeIndex + +-- | Wrapper around a type to indicate it belongs to a file with either +-- data/leaf nodes or index nodes. +data S (t :: StateType) a where + DataState :: a -> S 'TypeData a + IndexState :: a -> S 'TypeIndex a + deriving (Typeable) + +deriving instance Show a => Show (S t a) + +instance Binary a => Binary (S 'TypeData a) where + put (DataState a) = B.put a + get = DataState <$> B.get + +instance Binary a => Binary (S 'TypeIndex a) where + put (IndexState a) = B.put a + get = IndexState <$> B.get + +instance Functor (S t) where + f `fmap` (DataState v) = DataState (f v) + f `fmap` (IndexState v) = IndexState (f v) + +getSValue :: S t a -> a +getSValue (DataState a) = a +getSValue (IndexState a) = a + +newtype ReaderEnv hnds = ReaderEnv { readerHnds :: hnds } + +data FileState stateType = FileState { + fileStateNewlyFreedPages :: ![NewlyFreed] + -- ^ Pages free'd in this transaction, not ready for reuse until the + -- transaction is commited. + + , fileStateOriginalNumPages :: !(S stateType PageId) + -- ^ The original number of pages in the file, before the transaction + -- started. + + , fileStateNewNumPages :: !(S stateType PageId) + -- ^ The new uncommited number of pages in the file. + -- + -- All pages in the range 'fileStateOriginalNumPages' to + -- 'fileStateNewNumPages' (excluding) are freshly allocated in the + -- ongoing transaction. + + , fileStateDirtyPages :: !(Set PageId) + -- ^ Pages written to in this transaction. + + , fileStateFreeTree :: !(S stateType FreeTree) + -- ^ The root of the free tree, might change during a transaction. + + , fileStateCachedFreePages :: !(S stateType [FreePage]) + -- ^ All pages that are immediately ready for reuse in this and any + -- subsequent transactions. + } + +data WriterEnv hnds = WriterEnv + { writerHnds :: !hnds + , writerTxId :: !TxId + , writerReaders :: Map TxId Integer + + , writerIndexFileState :: FileState 'TypeIndex + -- ^ State of the file with index nodes. + + , writerDataFileState :: FileState 'TypeData + -- ^ State of the file with data/leaf nodes. + + , writerQueryFreeTreeOn :: !Bool + -- ^ Whether or not querying free pages from the free is enabled. + + , writerDirtyOverflows :: !(Set DirtyOverflow) + -- ^ Newly allocated overflow pages in this transaction. + + , writerOverflowCounter :: !Word32 + -- ^ Counts how many overflow pages were already allocated in this transaction. + + , writerRemovedOverflows :: ![OldOverflow] + -- ^ Old overflow pages that were removed in this transaction + -- and should be deleted when no longer in use. + } + +-- | Create a new writer. +newWriter :: hnd -> TxId -> Map TxId Integer + -> S 'TypeData PageId -> S 'TypeIndex PageId + -> S 'TypeData [FreePage] -> S 'TypeIndex [FreePage] + -> S 'TypeData FreeTree -> S 'TypeIndex FreeTree + -> WriterEnv hnd +newWriter hnd tx readers + numDataPages numIndexPages + dataFreePages indexFreePages + dataFreeTree indexFreeTree = + WriterEnv { + writerHnds = hnd + , writerTxId = tx + , writerReaders = readers + + , writerIndexFileState = newFileState numIndexPages indexFreePages indexFreeTree + , writerDataFileState = newFileState numDataPages dataFreePages dataFreeTree + + , writerQueryFreeTreeOn = True + , writerDirtyOverflows = S.empty + , writerOverflowCounter = 0 + , writerRemovedOverflows = [] + } + where + newFileState numPages freePages freeTree = FileState { + fileStateNewlyFreedPages = [] + , fileStateOriginalNumPages = numPages + , fileStateNewNumPages = numPages + , fileStateDirtyPages = S.empty + , fileStateCachedFreePages = freePages + , fileStateFreeTree = freeTree + } + +-- | Wrapper around 'PageId' indicating it is newly free'd and cannot be reused +-- in the same transaction. +newtype NewlyFreed = NewlyFreed PageId deriving (Eq, Ord, Show) + +-- | Wrapper around 'PageId' indicating it is free and can be reused in any +-- transaction. +newtype FreePage = FreePage PageId deriving (Binary, Eq, Ord, Show) + +-- | Wrapper around 'PageId' indicating that it is dirty, i.e. written to in +-- this transaction. +newtype Dirty = Dirty PageId deriving (Eq, Ord, Show) + +-- | Try to free a page, given a set of dirty pages. +-- +-- If the page was dirty, a 'FreePage' page is added to the environment, if +-- not a 'NewlyFreed' page is added to the environment. +-- +-- Btw, give me lenses... +freePage :: (Functor m, MonadState (WriterEnv hnd) m) => S stateType PageId -> m () +freePage pid@(DataState pid') = do + dirty' <- dirty pid + modify' $ \e -> + e { writerDataFileState = + updateFileState (writerDataFileState e) DataState + dirty' pid' + } + +freePage pid@(IndexState pid') = do + dirty' <- dirty pid + modify' $ \e -> + e { writerIndexFileState = + updateFileState (writerIndexFileState e) IndexState + dirty' pid' + } + +updateFileState :: FileState t + -> (forall a. a -> S t a) + -> Maybe Dirty + -> PageId + -> FileState t +updateFileState e cons dirty' pid' = + if | Just (Dirty p) <- dirty' -> + e { fileStateCachedFreePages = + cons $ FreePage p : getSValue (fileStateCachedFreePages e) } + + | p <- pid' -> + e { fileStateNewlyFreedPages = + NewlyFreed p : fileStateNewlyFreedPages e } + +-- | Get a 'Dirty' page, by first proving it is in fact dirty. +dirty :: (Functor m, MonadState (WriterEnv hnd) m) => S stateType PageId -> m (Maybe Dirty) +dirty pid = case pid of + DataState p -> (page p . fileStateDirtyPages . writerDataFileState) <$> get + IndexState p -> (page p . fileStateDirtyPages . writerIndexFileState) <$> get + where + page p dirtyPages + | p `S.member` dirtyPages = Just (Dirty p) + | otherwise = Nothing + +-- | Touch a fresh page, make it dirty. +-- +-- We really need lenses... +touchPage :: MonadState (WriterEnv hnd) m => S stateType PageId -> m () +touchPage (DataState pid) = do + modify' $ \e -> + let dirtyPages = fileStateDirtyPages (writerDataFileState e) in + e { writerDataFileState = (writerDataFileState e) { + fileStateDirtyPages = S.insert pid dirtyPages } + } + modify' $ \e -> + let oldNum = getSValue $ fileStateNewNumPages (writerDataFileState e) + newNum = max oldNum (pid + 1) + in e { writerDataFileState = (writerDataFileState e) { + fileStateNewNumPages = DataState newNum } + } + +touchPage (IndexState pid) = do + modify' $ \e -> + let dirtyPages = fileStateDirtyPages (writerIndexFileState e) in + e { writerIndexFileState = (writerIndexFileState e) { + fileStateDirtyPages = S.insert pid dirtyPages } + } + modify' $ \e -> + let oldNum = getSValue $ fileStateNewNumPages (writerIndexFileState e) + newNum = max oldNum (pid + 1) + in e { writerIndexFileState = (writerIndexFileState e) { + fileStateNewNumPages = IndexState newNum } + } + +-- | Wrapper around 'OverflowId' indicating that it is dirty. +newtype DirtyOverflow = DirtyOverflow OverflowId deriving (Eq, Ord, Show) + +-- | Wrapper around 'OverflowId' indicating that it is an overflow +-- page from a previous transaction. +newtype OldOverflow = OldOverflow OverflowId deriving (Eq, Ord, Show) + +-- | Touch a fresh overflow page, making it dirty. +touchOverflow :: MonadState (WriterEnv hnd) m => OverflowId -> m () +touchOverflow i = modify' $ + \e -> e { writerDirtyOverflows = + S.insert (DirtyOverflow i) (writerDirtyOverflows e) } + +-- | Get the type of the overflow page. +overflowType :: MonadState (WriterEnv hnd) m => OverflowId -> m (Either DirtyOverflow OldOverflow) +overflowType i = do + dirty' <- gets $ \e -> S.member (DirtyOverflow i) (writerDirtyOverflows e) + if dirty' then return $ Left (DirtyOverflow i) + else return $ Right (OldOverflow i) + +-- | Free an old overflow page. +removeOldOverflow :: MonadState (WriterEnv hdn) m => OldOverflow -> m () +removeOldOverflow i = + modify' $ \e -> e { writerRemovedOverflows = i : writerRemovedOverflows e } diff --git a/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Query.hs b/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Query.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Query.hs @@ -0,0 +1,152 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE RankNTypes #-} +module Database.Haskey.Alloc.Concurrent.Internal.FreePages.Query where + +import Control.Applicative ((<|>), (<$>)) +import Control.Concurrent.STM +import Control.Monad.State +import Control.Monad.Trans.Maybe + +import Data.List.NonEmpty (NonEmpty((:|))) +import qualified Data.List.NonEmpty as NE + +import Data.BTree.Alloc.Class +import Data.BTree.Primitives +import qualified Data.BTree.Impure as B +import qualified Data.BTree.Impure.NonEmpty as NEB + +import Database.Haskey.Alloc.Concurrent.Internal.Environment +import Database.Haskey.Alloc.Concurrent.Internal.FreePages.Tree +import Database.Haskey.Utils.Monad (ifM) +import qualified Database.Haskey.Utils.STM.Map as Map + +-- | Get a free page. +-- +-- First try to get one from the in-memory dirty pages. Then try to get one +-- from the in-memory free page cache stored in 'fileStateCachedFreePages'. If +-- that one is empty, actually query one from the free database. +getFreePageId :: (Functor m, AllocM m, MonadIO m, MonadState (WriterEnv hnd) m) + => S stateType () + -> m (Maybe PageId) +getFreePageId t = + runMaybeT $ MaybeT (getCachedFreePageId t) + <|> MaybeT (queryNewFreePageIds t) + +-- | Get a cached free page. +-- +-- Get a free page from the free database cache stored in +-- 'fileStateCachedFreePages'. +getCachedFreePageId :: (Functor m, MonadState (WriterEnv hnd) m) + => S stateType () + -> m (Maybe PageId) +getCachedFreePageId stateType = + case stateType of + DataState () -> do + s <- writerDataFileState <$> get + let (pid, s') = query DataState s + modify' $ \env -> env { writerDataFileState = s' } + return pid + IndexState () -> do + s <- writerIndexFileState <$> get + let (pid, s') = query IndexState s + modify' $ \env -> env { writerIndexFileState = s' } + return pid + where + query :: (forall a. a -> S t a) + -> FileState t + -> (Maybe PageId, FileState t) + query cons env = case getSValue $ fileStateCachedFreePages env of + [] -> (Nothing, env) + FreePage pid : pageIds -> + let env' = env { fileStateCachedFreePages = cons pageIds } in + (Just pid, env') + +-- | Try to get a list of free pages from the free page database, return the +-- first free one for immediate use, and store the rest in the environment. +-- +-- Immediately remove the queried free pages from the free tree. +queryNewFreePageIds :: (AllocM m, MonadIO m, MonadState (WriterEnv hnd) m) + => S stateType () + -> m (Maybe PageId) +queryNewFreePageIds stateType = ifM (not . writerQueryFreeTreeOn <$> get) (return Nothing) $ do + flag <- case stateType of + DataState () -> + query DataState + writerDataFileState + (\e s -> e { writerDataFileState = s }) + + IndexState () -> + query IndexState + writerIndexFileState + (\e s -> e { writerIndexFileState = s }) + + if flag then getFreePageId stateType + else return Nothing + where + query :: (AllocM m, MonadIO m, MonadState (WriterEnv hnd) m) + => (forall a. a -> S t a) + -> (forall h. WriterEnv h -> FileState t) + -> (forall h. WriterEnv h -> FileState t -> WriterEnv h) + -> m Bool + query cons getState setState = do + tree <- gets $ getSValue . fileStateFreeTree . getState + + -- Lookup the oldest free page + lookupValidFreePageIds tree >>= \case + Nothing -> return False + Just (txId, x :| xs) -> do + -- Save them for reuse + modify' $ \e -> + let s = getState e + pids = map FreePage (x:xs) + in setState e $ + s { fileStateCachedFreePages = + cons $ pids ++ getSValue (fileStateCachedFreePages s) } + + -- Remove the entry from the tree + modify' $ \e -> e { writerQueryFreeTreeOn = False } + tree' <- txId `deleteSubtree` tree + modify' $ \e -> e { writerQueryFreeTreeOn = True } + + -- Update the tree + modify' $ \e -> setState e $ + (getState e) { fileStateFreeTree = cons tree' } + + return True + +-- | Lookup a list of free pages from the free page database, guaranteed to be old enough. +lookupValidFreePageIds :: (MonadIO m, AllocReaderM m, MonadState (WriterEnv hnd) m) + => FreeTree + -> m (Maybe (TxId, NonEmpty PageId)) +lookupValidFreePageIds tree = runMaybeT $ + MaybeT (lookupFreePageIds tree) >>= (MaybeT . checkFreePages) + +-- | Lookup a list of free pages from the free page database. +lookupFreePageIds :: (Functor m, AllocReaderM m, MonadState (WriterEnv hnd) m) + => FreeTree + -> m (Maybe (Unchecked (TxId, NonEmpty PageId))) +lookupFreePageIds tree = B.lookupMin tree >>= \case + Nothing -> return Nothing + Just (tx, subtree) -> do + pids <- subtreeToList subtree + return . Just $ Unchecked (tx, pids) + where + subtreeToList subtree = NE.map fst <$> NEB.toList subtree + +-- | Auxiliry type to ensure the transaction ID of free pages are checked. +newtype Unchecked a = Unchecked a + +-- | Check the transaction ID of the free pages, if it's to old, return +-- 'Nothing'. +checkFreePages :: (Functor m, MonadIO m, MonadState (WriterEnv hnd) m) + => Unchecked (TxId, NonEmpty PageId) + -> m (Maybe (TxId, NonEmpty PageId)) +checkFreePages (Unchecked v) = do + readers <- writerReaders <$> get + oldest <- liftIO . atomically $ Map.lookupMinKey readers + tx <- writerTxId <$> get + if maybe True (> fst v) oldest && fst v + 1 < tx + then return (Just v) + else return Nothing diff --git a/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Save.hs b/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Save.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Save.hs @@ -0,0 +1,33 @@ +module Database.Haskey.Alloc.Concurrent.Internal.FreePages.Save where + +import Data.List.NonEmpty (NonEmpty((:|))) + +import Data.BTree.Alloc.Class +import Data.BTree.Primitives + +import Database.Haskey.Alloc.Concurrent.Internal.Environment +import Database.Haskey.Alloc.Concurrent.Internal.FreePages.Tree + +-- | Save the free pages from the dirty page list and the free page +-- cache. +saveFreePages :: AllocM m + => TxId + -> FileState t + -> m FreeTree +saveFreePages tx env = saveNewlyFreedPages tx env tree + where + tree = getSValue $ fileStateFreeTree env + +-- | Save the newly free pages of the current transaction, as stored by +-- 'fileStateNewlyFreedPages'. +saveNewlyFreedPages :: AllocM m + => TxId + -> FileState t + -> FreeTree + -> m FreeTree +saveNewlyFreedPages tx env tree = + case newlyFreed of + [] -> deleteSubtree tx tree + x:xs -> replaceSubtree tx (x :| xs) tree + where + newlyFreed = map (\(NewlyFreed pid) -> pid) $ fileStateNewlyFreedPages env diff --git a/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Tree.hs b/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Tree.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Internal/FreePages/Tree.hs @@ -0,0 +1,69 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} +-- | Module describing the tree structure of the free page database. +module Database.Haskey.Alloc.Concurrent.Internal.FreePages.Tree where + +import Control.Monad ((>=>)) + +import Data.Foldable (traverse_) +import Data.List.NonEmpty (NonEmpty) +import qualified Data.List.NonEmpty as NE + +import Data.BTree.Alloc.Class +import Data.BTree.Impure (Tree) +import Data.BTree.Impure.NonEmpty (NonEmptyTree(..)) +import Data.BTree.Primitives +import qualified Data.BTree.Impure as B +import qualified Data.BTree.Impure.NonEmpty as NET + +-- | The main tree structure of the free page database. +-- +-- The main free page database tree maps a 'TxId' to a 'FreeSubtree'. +type FreeTree = Tree TxId FreeSubtree + +-- | the subtree structure of the free page database. +-- +-- Just a collection of free 'PageId's. +type FreeSubtree = NonEmptyTree PageId () + +-- | Replace the subtree of a certain 'TxId'. +replaceSubtree :: AllocM m + => TxId + -> NonEmpty PageId + -> FreeTree + -> m FreeTree +replaceSubtree tx pids = deleteSubtree tx >=> insertSubtree tx pids + +-- | Delete the subtree of a certain 'TxId'. +-- +-- The 'TxId' will not be present anymore in the free tree after this call. +deleteSubtree :: AllocM m + => TxId + -> FreeTree + -> m FreeTree +deleteSubtree tx tree = B.lookup tx tree >>= \case + Nothing -> return tree + Just (NonEmptyTree h nid) -> do + freeAllNodes h nid + B.delete tx tree + where + freeAllNodes :: (AllocM m, Key key, Value val) + => Height h + -> NodeId h key val + -> m () + freeAllNodes h nid = readNode h nid >>= \case + NET.Leaf _ -> freeNode h nid + NET.Idx idx -> do + let subHgt = decrHeight h + traverse_ (freeAllNodes subHgt) idx + freeNode h nid + +-- | Insert a subtree for a certain 'TxId'. +insertSubtree :: AllocM m + => TxId + -> NonEmpty PageId + -> FreeTree + -> m FreeTree +insertSubtree tx pids tree = do + subtree <- NET.fromList (NE.zip pids (NE.repeat ())) + B.insert tx subtree tree diff --git a/src/Database/Haskey/Alloc/Concurrent/Internal/Meta.hs b/src/Database/Haskey/Alloc/Concurrent/Internal/Meta.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Internal/Meta.hs @@ -0,0 +1,73 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE StandaloneDeriving #-} +-- | This module implements data structures and function related to the +-- metadata of the concurrent page allocator. +module Database.Haskey.Alloc.Concurrent.Internal.Meta where + +import Data.Binary (Binary) +import Data.Proxy (Proxy) +import Data.Typeable (Typeable) + +import GHC.Generics (Generic) + +import Data.BTree.Impure.Internal.Structures +import Data.BTree.Primitives + +import Database.Haskey.Alloc.Concurrent.Internal.Environment +import Database.Haskey.Alloc.Concurrent.Internal.FreePages.Tree +import Database.Haskey.Alloc.Concurrent.Internal.Overflow +import Database.Haskey.Store + +-- | User-defined data root stored inside 'ConcurrentMeta'. +-- +-- This can be a user-defined collection of 'Tree' roots. +class Value root => Root root where + +instance (Key k, Value v) => Root (Tree k v) where + +-- | Data type used to point to the most recent version of the meta data. +data CurrentMetaPage = Meta1 | Meta2 + +-- | Meta data of the page allocator. +-- +-- The @root@ type parameter should be a user-defined collection of 'Tree' +-- roots, instantiating the 'Root' type class. +-- +-- To store store a single tree, use @ConcurrentMeta (Tree k v)@. +data ConcurrentMeta root = ConcurrentMeta { + concurrentMetaRevision :: TxId + , concurrentMetaDataNumPages :: S 'TypeData PageId + , concurrentMetaIndexNumPages :: S 'TypeIndex PageId + , concurrentMetaRoot :: root + , concurrentMetaDataFreeTree :: S 'TypeData FreeTree + , concurrentMetaIndexFreeTree :: S 'TypeIndex FreeTree + , concurrentMetaOverflowTree :: OverflowTree + , concurrentMetaDataCachedFreePages :: S 'TypeData [FreePage] + , concurrentMetaIndexCachedFreePages :: S 'TypeIndex [FreePage] + } deriving (Generic, Typeable) + +deriving instance (Show root) => Show (ConcurrentMeta root) + +instance (Binary root) => Binary (ConcurrentMeta root) where + +-- | A class representing the storage requirements of the page allocator. +-- +-- A store supporting the page allocator should be an instance of this class. +class StoreM FilePath m => ConcurrentMetaStoreM m where + -- | Write the meta-data structure to a certain page. + putConcurrentMeta :: Root root + => FilePath + -> ConcurrentMeta root + -> m () + + -- | Try to read the meta-data structure from a handle, or return 'Nothing' + -- if the handle doesn't contain a meta page. + readConcurrentMeta :: Root root + => FilePath + -> Proxy root + -> m (Maybe (ConcurrentMeta root)) + diff --git a/src/Database/Haskey/Alloc/Concurrent/Internal/Monad.hs b/src/Database/Haskey/Alloc/Concurrent/Internal/Monad.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Internal/Monad.hs @@ -0,0 +1,193 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiWayIf #-} +{-# LANGUAGE RankNTypes #-} +-- | This module implements the 'ConcurrentT' monad. +-- +-- The 'ConcurrentT' monad is used to implement a page allocator with +-- concurrent readers and serialized writers. +module Database.Haskey.Alloc.Concurrent.Internal.Monad where + +import Control.Applicative (Applicative, (<$>)) +import Control.Monad.Catch +import Control.Monad.State + +import Data.Proxy (Proxy(..)) + +import System.FilePath (()) + +import Data.BTree.Alloc.Class +import Data.BTree.Primitives + +import Database.Haskey.Alloc.Concurrent.Internal.Environment +import Database.Haskey.Alloc.Concurrent.Internal.FreePages.Query +import Database.Haskey.Alloc.Concurrent.Internal.Meta +import Database.Haskey.Alloc.Concurrent.Internal.Overflow +import Database.Haskey.Store +import qualified Database.Haskey.Store.Class as Store + +-- | All necessary database handles. +data ConcurrentHandles = ConcurrentHandles { + concurrentHandlesRoot :: FilePath + , concurrentHandlesData :: FilePath + , concurrentHandlesIndex :: FilePath + , concurrentHandlesMetadata1 :: FilePath + , concurrentHandlesMetadata2 :: FilePath + , concurrentHandlesOverflowDir :: FilePath + } deriving (Show) + +-- | Construct a set of 'ConcurrentHandles' from a root directory. +concurrentHandles :: FilePath -> ConcurrentHandles +concurrentHandles fp = ConcurrentHandles { + concurrentHandlesRoot = fp + , concurrentHandlesData = fp "data" "data" + , concurrentHandlesIndex = fp "index" "index" + , concurrentHandlesMetadata1 = fp "meta" "1" + , concurrentHandlesMetadata2 = fp "meta" "2" + , concurrentHandlesOverflowDir = fp "overflow" + } + +-- | Monad in which page allocations can take place. +-- +-- The monad has access to a 'ConcurrentMetaStoreM' back-end which manages can +-- store and retreive the corresponding metadata. +newtype ConcurrentT env hnd m a = ConcurrentT { fromConcurrentT :: StateT (env hnd) m a } + deriving (Functor, Applicative, Monad, + MonadIO, MonadThrow, MonadCatch, MonadMask, + MonadState (env hnd)) + +instance MonadTrans (ConcurrentT env hnd) where + lift = ConcurrentT . lift + +-- | Run the actions in an 'ConcurrentT' monad, given a reader or writer +-- environment. +runConcurrentT :: ConcurrentMetaStoreM m + => ConcurrentT env ConcurrentHandles m a + -> env ConcurrentHandles + -> m (a, env ConcurrentHandles) +runConcurrentT m = runStateT (fromConcurrentT m) + +-- | Evaluate the actions in an 'ConcurrentT' monad, given a reader or writer +-- environment. +evalConcurrentT :: ConcurrentMetaStoreM m + => ConcurrentT env ConcurrentHandles m a + -> env ConcurrentHandles -> + m a +evalConcurrentT m env = fst <$> runConcurrentT m env + +instance + (ConcurrentMetaStoreM m, MonadIO m) + => AllocM (ConcurrentT WriterEnv ConcurrentHandles m) + where + nodePageSize = ConcurrentT Store.nodePageSize + maxPageSize = ConcurrentT Store.maxPageSize + maxKeySize = ConcurrentT Store.maxKeySize + maxValueSize = ConcurrentT Store.maxValueSize + + allocNode height n = do + hnd <- getWriterHnd height + pid <- getAndTouchPid + + let nid = pageIdToNodeId pid + lift $ putNodePage hnd height nid n + return nid + where + getAndTouchPid = getAndTouchFreePageId >>= \case + Just pid -> return pid + Nothing -> newTouchedPid + + getAndTouchFreePageId = case viewHeight height of + UZero -> getFreePageId (DataState ()) >>= \case + Nothing -> return Nothing + Just pid -> do + touchPage (DataState pid) + return (Just pid) + USucc _ -> getFreePageId (IndexState ()) >>= \case + Nothing -> return Nothing + Just pid -> do + touchPage (IndexState pid) + return (Just pid) + + newTouchedPid = case viewHeight height of + UZero -> do + pid <- fileStateNewNumPages . writerDataFileState <$> get + touchPage pid + return $ getSValue pid + USucc _ -> do + pid <- fileStateNewNumPages . writerIndexFileState <$> get + touchPage pid + return $ getSValue pid + + + freeNode height nid = case viewHeight height of + UZero -> freePage (DataState $ nodeIdToPageId nid) + USucc _ -> freePage (IndexState $ nodeIdToPageId nid) + + allocOverflow v = do + root <- concurrentHandlesOverflowDir . writerHnds <$> get + oid <- getNewOverflowId + touchOverflow oid + + let hnd = getOverflowHandle root oid + lift $ openHandle hnd + lift $ putOverflow hnd v + lift $ closeHandle hnd + return oid + + freeOverflow oid = overflowType oid >>= \case + Right i -> removeOldOverflow i + Left (DirtyOverflow i) -> deleteOverflowData i + + deleteOverflowData i = do + root <- concurrentHandlesOverflowDir . writerHnds <$> get + lift $ removeHandle (getOverflowHandle root i) + +instance + ConcurrentMetaStoreM m + => AllocReaderM (ConcurrentT WriterEnv ConcurrentHandles m) + where + readNode height nid = do + hnd <- getWriterHnd height + lift $ getNodePage hnd height Proxy Proxy nid + + readOverflow i = do + root <- concurrentHandlesOverflowDir . writerHnds <$> get + readOverflow' root i + +instance + ConcurrentMetaStoreM m + => AllocReaderM (ConcurrentT ReaderEnv ConcurrentHandles m) + where + readNode height nid = do + hnd <- getReaderHnd height + lift $ getNodePage hnd height Proxy Proxy nid + + readOverflow i = do + root <- concurrentHandlesOverflowDir . readerHnds <$> get + readOverflow' root i + +readOverflow' :: (ConcurrentMetaStoreM m, Value v) + => FilePath -> OverflowId -> ConcurrentT env hnd m v +readOverflow' root oid = do + let hnd = getOverflowHandle root oid + lift $ openHandle hnd + v <- lift $ getOverflow hnd Proxy + lift $ closeHandle hnd + return v + +getWriterHnd :: MonadState (WriterEnv ConcurrentHandles) m + => Height height + -> m FilePath +getWriterHnd h = case viewHeight h of + UZero -> gets $ concurrentHandlesData . writerHnds + USucc _ -> gets $ concurrentHandlesIndex . writerHnds + +getReaderHnd :: MonadState (ReaderEnv ConcurrentHandles) m + => Height height + -> m FilePath +getReaderHnd h = case viewHeight h of + UZero -> gets $ concurrentHandlesData . readerHnds + USucc _ -> gets $ concurrentHandlesIndex . readerHnds diff --git a/src/Database/Haskey/Alloc/Concurrent/Internal/Overflow.hs b/src/Database/Haskey/Alloc/Concurrent/Internal/Overflow.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Internal/Overflow.hs @@ -0,0 +1,140 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} +-- | Data structures and functions related to handling overflow pages. +module Database.Haskey.Alloc.Concurrent.Internal.Overflow where + +import Control.Applicative ((<$>)) +import Control.Concurrent.STM +import Control.Monad.State + +import Data.Bits (shiftR) +import Data.Foldable (traverse_) +import Data.List.NonEmpty (NonEmpty) +import Data.Maybe (fromMaybe, listToMaybe) +import Data.Word (Word8) +import qualified Data.List.NonEmpty as NE +import qualified Data.Map as M + +import Numeric (showHex, readHex) + +import System.FilePath ((), (<.>), dropExtension, takeFileName) + +import Data.BTree.Alloc.Class +import Data.BTree.Impure (Tree) +import Data.BTree.Impure.NonEmpty (NonEmptyTree(..)) +import Data.BTree.Primitives +import qualified Data.BTree.Impure as B +import qualified Data.BTree.Impure.NonEmpty as NEB + +import Database.Haskey.Alloc.Concurrent.Internal.Environment +import qualified Database.Haskey.Utils.STM.Map as Map + +getNewOverflowId :: (Functor m, MonadState (WriterEnv hnd) m) + => m OverflowId +getNewOverflowId = do + tx <- writerTxId <$> get + c <- writerOverflowCounter <$> get + modify' $ \e -> e { writerOverflowCounter = 1 + writerOverflowCounter e } + return (tx, c) + +getOverflowHandle :: FilePath -> OverflowId -> FilePath +getOverflowHandle root (TxId tx, c) = + getOverflowDir root (TxId tx) showHex' tx <.> showHex' c <.> "overflow" + +getOverflowDir :: FilePath -> TxId -> FilePath +getOverflowDir root (TxId tx) = + root lsb1 lsb2 + where + lsb1 = showHex' (fromIntegral tx :: Word8) + lsb2 = showHex' (fromIntegral (tx `shiftR` 8) :: Word8) + +readOverflowId :: FilePath -> Maybe OverflowId +readOverflowId fp = parse (dropExtension $ takeFileName fp) + where + parse s = do + (tx, s') <- readHex' s + s'' <- case s' of '.':xs -> return xs + _ -> Nothing + (c, _) <- readHex' s'' + return (tx, c) + +showHex' :: (Integral a, Show a) => a -> String +showHex' = flip showHex "" + +readHex' :: (Eq a, Num a) => String -> Maybe (a, String) +readHex' s = listToMaybe $ readHex s + +-------------------------------------------------------------------------------- + +-- | The main tree structure of the freed overflow page tree +type OverflowTree = Tree TxId OverflowSubtree + +-- | The subtree structure of the freed overflow page tree +type OverflowSubtree = NonEmptyTree OverflowId () + +-- | Save a set of overflow ids that were free'd in the transaction. +insertOverflowIds :: AllocM m + => TxId + -> NonEmpty OverflowId + -> OverflowTree + -> m OverflowTree +insertOverflowIds tx oids tree = do + subtree <- NEB.fromList (NE.zip oids (NE.repeat ())) + B.insert tx subtree tree + +-- | Delete the set of overflow ids that were free'd in the transaction. +deleteOverflowIds :: AllocM m + => TxId + -> OverflowTree + -> m OverflowTree +deleteOverflowIds tx tree = B.lookup tx tree >>= \case + Nothing -> return tree + Just (NonEmptyTree h nid) -> do + freeAllNodes h nid + B.delete tx tree + where + freeAllNodes :: (AllocM m) + => Height h + -> NodeId h OverflowId () + -> m () + freeAllNodes h nid = readNode h nid >>= \case + l@(NEB.Leaf _) -> freeOverflowInLeaf l >> freeNode h nid + NEB.Idx idx -> do + let subHgt = decrHeight h + traverse_ (freeAllNodes subHgt) idx + freeNode h nid + + freeOverflowInLeaf :: (AllocM m) + => NEB.Node 'Z OverflowId () + -> m () + freeOverflowInLeaf (NEB.Leaf items) = mapM_ deleteOverflowData $ M.keys items + +-------------------------------------------------------------------------------- + +deleteOutdatedOverflowIds :: (Functor m, AllocM m, MonadIO m, + MonadState (WriterEnv hnd) m) + => OverflowTree + -> m (Maybe OverflowTree) +deleteOutdatedOverflowIds tree = do + defaultTx <- writerTxId <$> get + readers <- writerReaders <$> get + oldest <- liftIO . atomically $ + fromMaybe defaultTx <$> Map.lookupMinKey readers + + B.lookupMin tree >>= \case + Nothing -> return Nothing + Just (tx, _) -> if tx >= oldest + then return Nothing + else Just <$> go oldest tx tree + where + go oldest tx t = do + t' <- deleteOverflowIds tx t + B.lookupMin t' >>= \case + Nothing -> return t' + Just (tx', _) -> if tx' >= oldest + then return t' + else go oldest tx' t' + +-------------------------------------------------------------------------------- diff --git a/src/Database/Haskey/Alloc/Concurrent/Meta.hs b/src/Database/Haskey/Alloc/Concurrent/Meta.hs deleted file mode 100644 --- a/src/Database/Haskey/Alloc/Concurrent/Meta.hs +++ /dev/null @@ -1,73 +0,0 @@ -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE StandaloneDeriving #-} --- | This module implements data structures and function related to the --- metadata of the concurrent page allocator. -module Database.Haskey.Alloc.Concurrent.Meta where - -import Data.Binary (Binary) -import Data.Proxy (Proxy) -import Data.Typeable (Typeable) - -import GHC.Generics (Generic) - -import Data.BTree.Impure.Structures -import Data.BTree.Primitives - -import Database.Haskey.Alloc.Concurrent.Environment -import Database.Haskey.Alloc.Concurrent.FreePages.Tree -import Database.Haskey.Alloc.Concurrent.Overflow -import Database.Haskey.Store - --- | User-defined data root stored inside 'ConcurrentMeta'. --- --- This can be a user-defined collection of 'Tree' roots. -class Value root => Root root where - -instance (Key k, Value v) => Root (Tree k v) where - --- | Data type used to point to the most recent version of the meta data. -data CurrentMetaPage = Meta1 | Meta2 - --- | Meta data of the page allocator. --- --- The @root@ type parameter should be a user-defined collection of 'Tree' --- roots, instantiating the 'Root' type class. --- --- To store store a single tree, use @ConcurrentMeta (Tree k v)@. -data ConcurrentMeta root = ConcurrentMeta { - concurrentMetaRevision :: TxId - , concurrentMetaDataNumPages :: S 'TypeData PageId - , concurrentMetaIndexNumPages :: S 'TypeIndex PageId - , concurrentMetaRoot :: root - , concurrentMetaDataFreeTree :: S 'TypeData FreeTree - , concurrentMetaIndexFreeTree :: S 'TypeIndex FreeTree - , concurrentMetaOverflowTree :: OverflowTree - , concurrentMetaDataCachedFreePages :: S 'TypeData [FreePage] - , concurrentMetaIndexCachedFreePages :: S 'TypeIndex [FreePage] - } deriving (Generic, Typeable) - -deriving instance (Show root) => Show (ConcurrentMeta root) - -instance (Binary root) => Binary (ConcurrentMeta root) where - --- | A class representing the storage requirements of the page allocator. --- --- A store supporting the page allocator should be an instance of this class. -class StoreM FilePath m => ConcurrentMetaStoreM m where - -- | Write the meta-data structure to a certain page. - putConcurrentMeta :: Root root - => FilePath - -> ConcurrentMeta root - -> m () - - -- | Try to read the meta-data structure from a handle, or return 'Nothing' - -- if the handle doesn't contain a meta page. - readConcurrentMeta :: Root root - => FilePath - -> Proxy root - -> m (Maybe (ConcurrentMeta root)) - diff --git a/src/Database/Haskey/Alloc/Concurrent/Monad.hs b/src/Database/Haskey/Alloc/Concurrent/Monad.hs deleted file mode 100644 --- a/src/Database/Haskey/Alloc/Concurrent/Monad.hs +++ /dev/null @@ -1,193 +0,0 @@ -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE MultiWayIf #-} -{-# LANGUAGE RankNTypes #-} --- | This module implements the 'ConcurrentT' monad. --- --- The 'ConcurrentT' monad is used to implement a page allocator with --- concurrent readers and serialized writers. -module Database.Haskey.Alloc.Concurrent.Monad where - -import Control.Applicative (Applicative, (<$>)) -import Control.Monad.Catch -import Control.Monad.State - -import Data.Proxy (Proxy(..)) - -import System.FilePath (()) - -import Data.BTree.Alloc.Class -import Data.BTree.Primitives - -import Database.Haskey.Alloc.Concurrent.Environment -import Database.Haskey.Alloc.Concurrent.FreePages.Query -import Database.Haskey.Alloc.Concurrent.Meta -import Database.Haskey.Alloc.Concurrent.Overflow -import Database.Haskey.Store -import qualified Database.Haskey.Store.Class as Store - --- | All necessary database handles. -data ConcurrentHandles = ConcurrentHandles { - concurrentHandlesRoot :: FilePath - , concurrentHandlesData :: FilePath - , concurrentHandlesIndex :: FilePath - , concurrentHandlesMetadata1 :: FilePath - , concurrentHandlesMetadata2 :: FilePath - , concurrentHandlesOverflowDir :: FilePath - } deriving (Show) - --- | Construct a set of 'ConcurrentHandles' from a root directory. -concurrentHandles :: FilePath -> ConcurrentHandles -concurrentHandles fp = ConcurrentHandles { - concurrentHandlesRoot = fp - , concurrentHandlesData = fp "data" "data" - , concurrentHandlesIndex = fp "index" "index" - , concurrentHandlesMetadata1 = fp "meta" "1" - , concurrentHandlesMetadata2 = fp "meta" "2" - , concurrentHandlesOverflowDir = fp "overflow" - } - --- | Monad in which page allocations can take place. --- --- The monad has access to a 'ConcurrentMetaStoreM' back-end which manages can --- store and retreive the corresponding metadata. -newtype ConcurrentT env hnd m a = ConcurrentT { fromConcurrentT :: StateT (env hnd) m a } - deriving (Functor, Applicative, Monad, - MonadIO, MonadThrow, MonadCatch, MonadMask, - MonadState (env hnd)) - -instance MonadTrans (ConcurrentT env hnd) where - lift = ConcurrentT . lift - --- | Run the actions in an 'ConcurrentT' monad, given a reader or writer --- environment. -runConcurrentT :: ConcurrentMetaStoreM m - => ConcurrentT env ConcurrentHandles m a - -> env ConcurrentHandles - -> m (a, env ConcurrentHandles) -runConcurrentT m = runStateT (fromConcurrentT m) - --- | Evaluate the actions in an 'ConcurrentT' monad, given a reader or writer --- environment. -evalConcurrentT :: ConcurrentMetaStoreM m - => ConcurrentT env ConcurrentHandles m a - -> env ConcurrentHandles -> - m a -evalConcurrentT m env = fst <$> runConcurrentT m env - -instance - (ConcurrentMetaStoreM m, MonadIO m) - => AllocM (ConcurrentT WriterEnv ConcurrentHandles m) - where - nodePageSize = ConcurrentT Store.nodePageSize - maxPageSize = ConcurrentT Store.maxPageSize - maxKeySize = ConcurrentT Store.maxKeySize - maxValueSize = ConcurrentT Store.maxValueSize - - allocNode height n = do - hnd <- getWriterHnd height - pid <- getAndTouchPid - - let nid = pageIdToNodeId pid - lift $ putNodePage hnd height nid n - return nid - where - getAndTouchPid = getAndTouchFreePageId >>= \case - Just pid -> return pid - Nothing -> newTouchedPid - - getAndTouchFreePageId = case viewHeight height of - UZero -> getFreePageId (DataState ()) >>= \case - Nothing -> return Nothing - Just pid -> do - touchPage (DataState pid) - return (Just pid) - USucc _ -> getFreePageId (IndexState ()) >>= \case - Nothing -> return Nothing - Just pid -> do - touchPage (IndexState pid) - return (Just pid) - - newTouchedPid = case viewHeight height of - UZero -> do - pid <- fileStateNewNumPages . writerDataFileState <$> get - touchPage pid - return $ getSValue pid - USucc _ -> do - pid <- fileStateNewNumPages . writerIndexFileState <$> get - touchPage pid - return $ getSValue pid - - - freeNode height nid = case viewHeight height of - UZero -> freePage (DataState $ nodeIdToPageId nid) - USucc _ -> freePage (IndexState $ nodeIdToPageId nid) - - allocOverflow v = do - root <- concurrentHandlesOverflowDir . writerHnds <$> get - oid <- getNewOverflowId - touchOverflow oid - - let hnd = getOverflowHandle root oid - lift $ openHandle hnd - lift $ putOverflow hnd v - lift $ closeHandle hnd - return oid - - freeOverflow oid = overflowType oid >>= \case - Right i -> removeOldOverflow i - Left (DirtyOverflow i) -> deleteOverflowData i - - deleteOverflowData i = do - root <- concurrentHandlesOverflowDir . writerHnds <$> get - lift $ removeHandle (getOverflowHandle root i) - -instance - ConcurrentMetaStoreM m - => AllocReaderM (ConcurrentT WriterEnv ConcurrentHandles m) - where - readNode height nid = do - hnd <- getWriterHnd height - lift $ getNodePage hnd height Proxy Proxy nid - - readOverflow i = do - root <- concurrentHandlesOverflowDir . writerHnds <$> get - readOverflow' root i - -instance - ConcurrentMetaStoreM m - => AllocReaderM (ConcurrentT ReaderEnv ConcurrentHandles m) - where - readNode height nid = do - hnd <- getReaderHnd height - lift $ getNodePage hnd height Proxy Proxy nid - - readOverflow i = do - root <- concurrentHandlesOverflowDir . readerHnds <$> get - readOverflow' root i - -readOverflow' :: (ConcurrentMetaStoreM m, Value v) - => FilePath -> OverflowId -> ConcurrentT env hnd m v -readOverflow' root oid = do - let hnd = getOverflowHandle root oid - lift $ openHandle hnd - v <- lift $ getOverflow hnd Proxy - lift $ closeHandle hnd - return v - -getWriterHnd :: MonadState (WriterEnv ConcurrentHandles) m - => Height height - -> m FilePath -getWriterHnd h = case viewHeight h of - UZero -> gets $ concurrentHandlesData . writerHnds - USucc _ -> gets $ concurrentHandlesIndex . writerHnds - -getReaderHnd :: MonadState (ReaderEnv ConcurrentHandles) m - => Height height - -> m FilePath -getReaderHnd h = case viewHeight h of - UZero -> gets $ concurrentHandlesData . readerHnds - USucc _ -> gets $ concurrentHandlesIndex . readerHnds diff --git a/src/Database/Haskey/Alloc/Concurrent/Overflow.hs b/src/Database/Haskey/Alloc/Concurrent/Overflow.hs deleted file mode 100644 --- a/src/Database/Haskey/Alloc/Concurrent/Overflow.hs +++ /dev/null @@ -1,138 +0,0 @@ -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE LambdaCase #-} --- | Data structures and functions related to handling overflow pages. -module Database.Haskey.Alloc.Concurrent.Overflow where - -import Control.Applicative ((<$>)) -import Control.Concurrent.STM -import Control.Monad.State - -import Data.Bits (shiftR) -import Data.Foldable (traverse_) -import Data.List.NonEmpty (NonEmpty) -import Data.Maybe (fromMaybe, listToMaybe) -import Data.Word (Word8) -import qualified Data.List.NonEmpty as NE -import qualified Data.Map as M - -import Numeric (showHex, readHex) - -import System.FilePath ((), (<.>), dropExtension, takeFileName) - -import Data.BTree.Alloc.Class -import Data.BTree.Impure -import Data.BTree.Impure.NonEmpty -import Data.BTree.Primitives - -import Database.Haskey.Alloc.Concurrent.Environment -import qualified Database.Haskey.Utils.STM.Map as Map - -getNewOverflowId :: (Functor m, MonadState (WriterEnv hnd) m) - => m OverflowId -getNewOverflowId = do - tx <- writerTxId <$> get - c <- writerOverflowCounter <$> get - modify' $ \e -> e { writerOverflowCounter = 1 + writerOverflowCounter e } - return (tx, c) - -getOverflowHandle :: FilePath -> OverflowId -> FilePath -getOverflowHandle root (TxId tx, c) = - getOverflowDir root (TxId tx) showHex' tx <.> showHex' c <.> "overflow" - -getOverflowDir :: FilePath -> TxId -> FilePath -getOverflowDir root (TxId tx) = - root lsb1 lsb2 - where - lsb1 = showHex' (fromIntegral tx :: Word8) - lsb2 = showHex' (fromIntegral (tx `shiftR` 8) :: Word8) - -readOverflowId :: FilePath -> Maybe OverflowId -readOverflowId fp = parse (dropExtension $ takeFileName fp) - where - parse s = do - (tx, s') <- readHex' s - s'' <- case s' of '.':xs -> return xs - _ -> Nothing - (c, _) <- readHex' s'' - return (tx, c) - -showHex' :: (Integral a, Show a) => a -> String -showHex' = flip showHex "" - -readHex' :: (Eq a, Num a) => String -> Maybe (a, String) -readHex' s = listToMaybe $ readHex s - --------------------------------------------------------------------------------- - --- | The main tree structure of the freed overflow page tree -type OverflowTree = Tree TxId OverflowSubtree - --- | The subtree structure of the freed overflow page tree -type OverflowSubtree = NonEmptyTree OverflowId () - --- | Save a set of overflow ids that were free'd in the transaction. -insertOverflowIds :: AllocM m - => TxId - -> NonEmpty OverflowId - -> OverflowTree - -> m OverflowTree -insertOverflowIds tx oids tree = do - subtree <- fromNonEmptyList (NE.zip oids (NE.repeat ())) - insertTree tx subtree tree - --- | Delete the set of overflow ids that were free'd in the transaction. -deleteOverflowIds :: AllocM m - => TxId - -> OverflowTree - -> m OverflowTree -deleteOverflowIds tx tree = lookupTree tx tree >>= \case - Nothing -> return tree - Just (NonEmptyTree h nid) -> do - freeAllNodes h nid - deleteTree tx tree - where - freeAllNodes :: (AllocM m) - => Height h - -> NodeId h OverflowId () - -> m () - freeAllNodes h nid = readNode h nid >>= \case - l@(Leaf _) -> freeOverflowInLeaf l >> freeNode h nid - Idx idx -> do - let subHgt = decrHeight h - traverse_ (freeAllNodes subHgt) idx - freeNode h nid - - freeOverflowInLeaf :: (AllocM m) - => Node 'Z OverflowId () - -> m () - freeOverflowInLeaf (Leaf items) = mapM_ deleteOverflowData $ M.keys items - --------------------------------------------------------------------------------- - -deleteOutdatedOverflowIds :: (Functor m, AllocM m, MonadIO m, - MonadState (WriterEnv hnd) m) - => OverflowTree - -> m (Maybe OverflowTree) -deleteOutdatedOverflowIds tree = do - defaultTx <- writerTxId <$> get - readers <- writerReaders <$> get - oldest <- liftIO . atomically $ - fromMaybe defaultTx <$> Map.lookupMinKey readers - - lookupMinTree tree >>= \case - Nothing -> return Nothing - Just (tx, _) -> if tx >= oldest - then return Nothing - else Just <$> go oldest tx tree - where - go oldest tx t = do - t' <- deleteOverflowIds tx t - lookupMinTree t' >>= \case - Nothing -> return t' - Just (tx', _) -> if tx' >= oldest - then return t' - else go oldest tx' t' - --------------------------------------------------------------------------------- diff --git a/src/Database/Haskey/Store/Class.hs b/src/Database/Haskey/Store/Class.hs --- a/src/Database/Haskey/Store/Class.hs +++ b/src/Database/Haskey/Store/Class.hs @@ -30,7 +30,7 @@ import qualified Data.Map as M import Data.BTree.Impure -import Data.BTree.Impure.Structures +import Data.BTree.Impure.Internal.Structures import Data.BTree.Primitives -------------------------------------------------------------------------------- diff --git a/src/Database/Haskey/Store/File.hs b/src/Database/Haskey/Store/File.hs --- a/src/Database/Haskey/Store/File.hs +++ b/src/Database/Haskey/Store/File.hs @@ -51,7 +51,7 @@ import System.FilePath (takeDirectory) import System.IO.Error (ioError, isDoesNotExistError) -import Data.BTree.Impure.Structures +import Data.BTree.Impure.Internal.Structures import Data.BTree.Primitives import Database.Haskey.Alloc.Concurrent diff --git a/src/Database/Haskey/Store/InMemory.hs b/src/Database/Haskey/Store/InMemory.hs --- a/src/Database/Haskey/Store/InMemory.hs +++ b/src/Database/Haskey/Store/InMemory.hs @@ -44,7 +44,7 @@ import Data.Word (Word64) import qualified Data.Map as M -import Data.BTree.Impure.Structures +import Data.BTree.Impure.Internal.Structures import Data.BTree.Primitives import Database.Haskey.Alloc.Concurrent diff --git a/src/Database/Haskey/Store/Page.hs b/src/Database/Haskey/Store/Page.hs --- a/src/Database/Haskey/Store/Page.hs +++ b/src/Database/Haskey/Store/Page.hs @@ -33,7 +33,7 @@ import Numeric (showHex) import Data.BTree.Impure -import Data.BTree.Impure.Structures (putLeafNode, getLeafNode, putIndexNode, getIndexNode) +import Data.BTree.Impure.Internal.Structures (putLeafNode, getLeafNode, putIndexNode, getIndexNode) import Data.BTree.Primitives import Database.Haskey.Alloc.Concurrent diff --git a/tests/Integration/WriteOpenRead/Concurrent.hs b/tests/Integration/WriteOpenRead/Concurrent.hs --- a/tests/Integration/WriteOpenRead/Concurrent.hs +++ b/tests/Integration/WriteOpenRead/Concurrent.hs @@ -36,7 +36,7 @@ import Data.BTree.Alloc.Class import Data.BTree.Impure import Data.BTree.Primitives -import qualified Data.BTree.Impure as Tree +import qualified Data.BTree.Impure as B import Database.Haskey.Alloc.Concurrent import Database.Haskey.Store.File @@ -90,7 +90,7 @@ ++ "\n got: " ++ show read' create :: MemoryFiles String -> IO (ConcurrentDb Root') - create = runMemoryStoreT (createConcurrentDb hnds Tree.empty) config + create = runMemoryStoreT (createConcurrentDb hnds B.empty) config where hnds = concurrentHandles "" @@ -140,7 +140,7 @@ create :: ConcurrentHandles -> IO (ConcurrentDb Root') - create hnds = runFileStoreT (createConcurrentDb hnds Tree.empty) config + create hnds = runFileStoreT (createConcurrentDb hnds B.empty) config openAndRead :: ConcurrentDb Root' @@ -164,9 +164,9 @@ writeTransaction (TestTransaction txType actions) = transaction where - writeAction (Insert k v) = insertTree k v - writeAction (Replace k v) = insertTree k v - writeAction (Delete k) = deleteTree k + writeAction (Insert k v) = B.insert k v + writeAction (Replace k v) = B.insert k v + writeAction (Delete k) = B.delete k writeAction ThrowException = const (throwM TestException) transaction = transact_ $ @@ -181,7 +181,7 @@ readAll :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Key k, Value v) => ConcurrentDb (Tree k v) -> m [(k, v)] -readAll = transactReadOnly Tree.toList +readAll = transactReadOnly B.toList -------------------------------------------------------------------------------- diff --git a/tests/Properties/Store/Page.hs b/tests/Properties/Store/Page.hs --- a/tests/Properties/Store/Page.hs +++ b/tests/Properties/Store/Page.hs @@ -23,7 +23,7 @@ import qualified Data.Map as M import qualified Data.Vector as V -import Data.BTree.Impure.Structures +import Data.BTree.Impure.Internal.Structures import Data.BTree.Primitives import Database.Haskey.Store.Page