diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,32 @@ +Copyright (c) 2017, Henri Verroken, Steven Keuchel + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Henri Verroken or Steven Keuchel nor the + names of other contributors may be used to endorse or promote + products derived from this software without specific prior + written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,20 @@ +haskey +====== + +[![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) + +Haskey is a transactional, ACID compliant, embeddable, scalable key-value +store written entirely in Haskell. It was developed as part of the [Summer of Haskell 2017][soh2017] project. + + [soh2017]: https://summer.haskell.org/news/2017-05-24-accepted-projects.html + +An introductory blog post on Haskey can be found [here][introduction]. + + [introduction]: https://deliquus.com/posts/2017-08-24-introducing-haskey.html + +_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._ diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/haskey.cabal b/haskey.cabal new file mode 100644 --- /dev/null +++ b/haskey.cabal @@ -0,0 +1,141 @@ +name: haskey +version: 0.1.0.0 +synopsis: A transcatoinal, ACID compliant, embeddable key-value store. +description: + Haskey is a transactional, ACID compliant, embeddable, scalable key-value + store written entirely in Haskell. + . + For more information on how to use this package, visit + +homepage: https://github.com/haskell-haskey +license: BSD3 +license-file: LICENSE +author: Henri Verroken, Steven Keuchel +maintainer: steven.keuchel@gmail.com +copyright: Copyright (c) 2017, Henri Verroken, Steven Keuchel +category: Database +build-type: Simple +cabal-version: >=1.10 + +extra-source-files: README.md + +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.Transaction + Database.Haskey.Store + Database.Haskey.Store.Class + Database.Haskey.Store.File + Database.Haskey.Store.InMemory + Database.Haskey.Store.Page + + other-modules: + Database.Haskey.Utils.IO + Database.Haskey.Utils.Monad + Database.Haskey.Utils.Monad.Catch + Database.Haskey.Utils.RLock + Database.Haskey.Utils.STM.Map + FileIO + + other-extensions: + DataKinds + DeriveFoldable + DeriveFunctor + DeriveTraversable + GADTs + KindSignatures + MultiWayIf + ScopedTypeVariables + StandaloneDeriving + + build-depends: + base >=4.7 && <5, + binary >=0.6 && <0.9 || >0.9 && <1, + bytestring >=0.10 && <1, + containers >=0.5 && <1, + directory >=1.2 && <2, + exceptions >=0.8.3 && <0.9, + filepath >=1.4 && <2, + focus >=0.1.2 && <0.2, + haskey-btree >=0.1 && <1, + list-t >=0.2 && <2, + lz4 >=0.2 && <1, + mtl >=2.1 && <3, + semigroups >=0.12 && <1, + stm >=2.1 && <3, + stm-containers >=0.2 && <1, + transformers >=0.3 && <1, + unix >=2.7.1.0 && <3, + xxhash-ffi >=0.1.0.1 && <1 + + default-language: Haskell2010 + ghc-options: -Wall + hs-source-dirs: src + hs-source-dirs: src-unix + +test-suite haskey-properties + main-is: Properties.hs + type: exitcode-stdio-1.0 + other-modules: + Properties.Store.Page + Properties.Utils + + build-depends: + base >=4.7 && <5, + binary >=0.6 && <0.9 || >0.9 && <1, + bytestring >=0.10 && <1, + containers >=0.5 && <1, + vector >=0.10 && <1, + + HUnit >=1.3 && <2, + QuickCheck >=2 && <3, + test-framework >=0.8 && <1, + test-framework-hunit >=0.3 && <1, + test-framework-quickcheck2 >=0.3 && <1, + haskey, + haskey-btree + + default-language: Haskell2010 + ghc-options: -Wall + hs-source-dirs: tests + +test-suite haskey-integration + main-is: Integration.hs + type: exitcode-stdio-1.0 + other-modules: + Integration.WriteOpenRead.Concurrent + Integration.WriteOpenRead.Transactions + + build-depends: + base >=4.7 && <5, + binary >=0.6 && <0.9 || >0.9 && <1, + bytestring >=0.10 && <1, + containers >=0.5 && <1, + directory >=1.2 && <2, + exceptions >=0.8.3 && <0.9, + mtl >=2.1 && <3, + transformers >=0.3 && <1, + temporary >=1.2 && <1.3, + vector >=0.10 && <1, + + QuickCheck >=2 && <3, + test-framework >=0.8 && <1, + test-framework-quickcheck2 >=0.3 && <1, + haskey, + haskey-btree + + default-language: Haskell2010 + ghc-options: -Wall + hs-source-dirs: tests + +source-repository head + type: git + location: https://github.com/haskell-haskey/haskey diff --git a/src-unix/FileIO.hs b/src-unix/FileIO.hs new file mode 100644 --- /dev/null +++ b/src-unix/FileIO.hs @@ -0,0 +1,201 @@ +-- | Module exporting some low level file primitives for Unix. +-- +-- This source file was taken from acid-state-0.14.3, and slightly modified. +module FileIO ( + FHandle + , openReadWrite + , write + , read + , flush + , close + , seek + , setFileSize + , getFileSize + , obtainPrefixLock + , releasePrefixLock + , PrefixLock +) where + +import Prelude hiding (read) +import qualified Prelude as P + +import Control.Applicative ((<$>)) +import Control.Exception (SomeException(..), throw, try) +import Control.Monad (void) + +import Data.Maybe (listToMaybe) +import Data.Word (Word8, Word64) + +import Foreign (Ptr) + +import System.Directory (createDirectoryIfMissing, removeFile) +import System.FilePath +import System.IO +import System.Posix (Fd, + openFd, + fdReadBuf, + fdWriteBuf, + fdToHandle, + fdSeek, + setFdSize, + fileSynchronise, + closeFd, + OpenMode(ReadWrite), + exclusive, trunc, + defaultFileFlags, + stdFileMode) +import System.Posix.Process (getProcessID) +import System.Posix.Signals (nullSignal, signalProcess) +import System.Posix.Types (ProcessID) +import qualified System.IO.Error as SE + + +newtype PrefixLock = PrefixLock FilePath + +newtype FHandle = FHandle Fd + +-- | Open the specified file in read-write mode. +openReadWrite :: FilePath -> IO FHandle +openReadWrite filename = + FHandle <$> openFd filename ReadWrite (Just stdFileMode) defaultFileFlags + +-- | Write **at most** the specified amount of bytes to a handle. +write :: FHandle -> Ptr Word8 -> Word64 -> IO Word64 +write (FHandle fd) data' length' = + fmap fromIntegral $ fdWriteBuf fd data' $ fromIntegral length' + +-- | Read **at most** the specified amount of bytes from a handle. +-- +-- Return the amount of bytes actually read, or throw an IO error (including +-- EOF). +read :: FHandle -> Ptr Word8 -> Word64 -> IO Word64 +read (FHandle fd) buf len = + fromIntegral <$> fdReadBuf fd buf (fromIntegral len) + +-- | Synchronize the file contents to disk. +flush :: FHandle -> IO () +flush (FHandle fd) = fileSynchronise fd + +-- | Seek to an absolute position. +seek :: FHandle -> Word64 -> IO () +seek (FHandle fd) offset = void $ fdSeek fd AbsoluteSeek (fromIntegral offset) + +-- | Set the filesize to a certain length. +setFileSize :: FHandle -> Word64 -> IO () +setFileSize (FHandle fd) size = setFdSize fd (fromIntegral size) + +-- | Get the filesize. This **edits the file pointer**. +getFileSize :: FHandle -> IO Word64 +getFileSize (FHandle fd) = fromIntegral <$> fdSeek fd SeekFromEnd 0 + +-- | Close a file. +close :: FHandle -> IO () +close (FHandle fd) = closeFd fd + +-- Unix needs to use a special open call to open files for exclusive writing +--openExclusively :: FilePath -> IO Handle +--openExclusively fp = +-- fdToHandle =<< openFd fp ReadWrite (Just 0o600) flags +-- where flags = defaultFileFlags {exclusive = True, trunc = True} + + + + +obtainPrefixLock :: FilePath -> IO PrefixLock +obtainPrefixLock prefix = checkLock fp >> takeLock fp + where fp = prefix ++ ".lock" + +-- |Read the lock and break it if the process is dead. +checkLock :: FilePath -> IO () +checkLock fp = readLock fp >>= maybeBreakLock fp + +-- |Read the lock and return the process id if possible. +readLock :: FilePath -> IO (Maybe ProcessID) +readLock fp = do + pid <- try (readFile fp) + return $ either (checkReadFileError fp) + (fmap (fromInteger . P.read) . listToMaybe . lines) + pid + +-- |Is this a permission error? If so we don't have permission to +-- remove the lock file, abort. +checkReadFileError :: String -> IOError -> Maybe ProcessID +checkReadFileError fp e | SE.isPermissionError e = throw (userError ("Could not read lock file: " ++ show fp)) + | SE.isDoesNotExistError e = Nothing + | otherwise = throw e + +maybeBreakLock :: FilePath -> Maybe ProcessID -> IO () +maybeBreakLock fp Nothing = + -- The lock file exists, but there's no PID in it. At this point, + -- we will break the lock, because the other process either died + -- or will give up when it failed to read its pid back from this + -- file. + breakLock fp +maybeBreakLock fp (Just pid) = do + -- The lock file exists and there is a PID in it. We can break the + -- lock if that process has died. + -- getProcessStatus only works on the children of the calling process. + -- exists <- try (getProcessStatus False True pid) >>= either checkException (return . isJust) + exists <- doesProcessExist pid + if exists + then throw (lockedBy fp pid) + else breakLock fp + +doesProcessExist :: ProcessID -> IO Bool +doesProcessExist pid = do + -- Implementation 1 + -- doesDirectoryExist ("/proc/" ++ show pid) + -- Implementation 2 + v <- try (signalProcess nullSignal pid) + return $ either checkException (const True) v + where checkException e | SE.isDoesNotExistError e = False + | otherwise = throw e + +-- |We have determined the locking process is gone, try to remove the +-- lock. +breakLock :: FilePath -> IO () +breakLock fp = try (removeFile fp) >>= either checkBreakError (const (return ())) + +-- |An exception when we tried to break a lock, if it says the lock +-- file has already disappeared we are still good to go. +checkBreakError :: IOError -> IO () +checkBreakError e | SE.isDoesNotExistError e = return () + | otherwise = throw e + +-- |Try to create lock by opening the file with the O_EXCL flag and +-- writing our PID into it. Verify by reading the pid back out and +-- matching, maybe some other process slipped in before we were done +-- and broke our lock. +takeLock :: FilePath -> IO PrefixLock +takeLock fp = do + createDirectoryIfMissing True (takeDirectory fp) + h <- openFd fp ReadWrite (Just 0o600) (defaultFileFlags {exclusive = True, trunc = True}) >>= fdToHandle + pid <- getProcessID + hPrint h pid >> hClose h + -- Read back our own lock and make sure its still ours + readLock fp >>= maybe (throw (cantLock fp pid)) + (\ pid' -> if pid /= pid' + then throw (stolenLock fp pid pid') + else return (PrefixLock fp)) + +-- |An exception saying the data is locked by another process. +lockedBy :: (Show a) => FilePath -> a -> SomeException +lockedBy fp pid = SomeException (SE.mkIOError SE.alreadyInUseErrorType ("Locked by " ++ show pid) Nothing (Just fp)) + +-- |An exception saying we don't have permission to create lock. +cantLock :: FilePath -> ProcessID -> SomeException +cantLock fp pid = SomeException (SE.mkIOError SE.alreadyInUseErrorType ("Process " ++ show pid ++ " could not create a lock") Nothing (Just fp)) + +-- |An exception saying another process broke our lock before we +-- finished creating it. +stolenLock :: FilePath -> ProcessID -> ProcessID -> SomeException +stolenLock fp pid pid' = SomeException (SE.mkIOError SE.alreadyInUseErrorType ("Process " ++ show pid ++ "'s lock was stolen by process " ++ show pid') Nothing (Just fp)) + +-- |Relinquish the lock by removing it and then verifying the removal. +releasePrefixLock :: PrefixLock -> IO () +releasePrefixLock (PrefixLock fp) = + dropLock >>= either checkDrop return + where + dropLock = try (removeFile fp) + checkDrop e | SE.isDoesNotExistError e = return () + | otherwise = throw e diff --git a/src/Database/Haskey/Alloc/Concurrent.hs b/src/Database/Haskey/Alloc/Concurrent.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent.hs @@ -0,0 +1,28 @@ +-- | The module implements an page allocator with page reuse and support for +-- multiple readers and serialized writers. +module Database.Haskey.Alloc.Concurrent ( + -- * Allocator + ConcurrentDb(..) + + -- * Open, close and create databases +, ConcurrentHandles(..) +, concurrentHandles +, createConcurrentDb +, openConcurrentDb +, closeConcurrentHandles + + -- * Manipulation and transactions +, module Database.Haskey.Alloc.Transaction +, transact +, transact_ +, transactReadOnly + + -- * Storage requirements +, ConcurrentMeta(..) +, ConcurrentMetaStoreM(..) +) where + +import Database.Haskey.Alloc.Concurrent.Database +import Database.Haskey.Alloc.Concurrent.Meta +import Database.Haskey.Alloc.Concurrent.Monad +import Database.Haskey.Alloc.Transaction diff --git a/src/Database/Haskey/Alloc/Concurrent/Database.hs b/src/Database/Haskey/Alloc/Concurrent/Database.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Database.hs @@ -0,0 +1,434 @@ +{-# 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) +import Control.Monad.State +import Control.Monad.Trans (lift) + +import Data.Proxy (Proxy(..)) +import Data.List.NonEmpty (NonEmpty((:|))) +import Data.Maybe (fromMaybe) +import qualified Data.Set as S + +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 k v = ConcurrentDb + { concurrentDbHandles :: ConcurrentHandles + , concurrentDbWriterLock :: RLock + , concurrentDbCurrentMeta :: TVar CurrentMetaPage + , concurrentDbMeta1 :: TVar (ConcurrentMeta k v) + , concurrentDbMeta2 :: TVar (ConcurrentMeta k v) + , concurrentDbReaders :: Map TxId Integer + } + +-- | 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 :: (Key k, Value v, MonadIO m, ConcurrentMetaStoreM m) + => ConcurrentHandles -> m (ConcurrentDb k v) +createConcurrentDb hnds = do + openConcurrentHandles hnds + db <- newConcurrentDb hnds meta0 + setCurrentMeta meta0 db + setCurrentMeta meta0 db + return db + where + meta0 = ConcurrentMeta { + concurrentMetaRevision = 0 + , concurrentMetaDataNumPages = DataState 0 + , concurrentMetaIndexNumPages = IndexState 0 + , concurrentMetaTree = Tree zeroHeight Nothing + , concurrentMetaDataFreeTree = DataState $ Tree zeroHeight Nothing + , concurrentMetaIndexFreeTree = IndexState $ Tree zeroHeight Nothing + , concurrentMetaOverflowTree = Tree zeroHeight Nothing + , concurrentMetaDataFreshUnusedPages = DataState S.empty + , concurrentMetaIndexFreshUnusedPages = IndexState S.empty + } + +-- | Open the an existing database, with the given handles. +openConcurrentDb :: (Key k, Value v, MonadIO m, MonadMask m, ConcurrentMetaStoreM m) + => ConcurrentHandles -> m (Maybe (ConcurrentDb k v)) +openConcurrentDb hnds@ConcurrentHandles{..} = do + openConcurrentHandles hnds + m1 <- readConcurrentMeta concurrentHandlesMetadata1 Proxy Proxy + m2 <- readConcurrentMeta concurrentHandlesMetadata2 Proxy 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 :: (Key k, Value v, MonadIO m) + => ConcurrentHandles + -> ConcurrentMeta k v + -> m (ConcurrentDb k v) +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 :: (Key k, Value v) + => ConcurrentDb k v -> STM (ConcurrentMeta k v) +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 :: (MonadIO m, ConcurrentMetaStoreM m, Key k, Value v) + => ConcurrentMeta k v -> ConcurrentDb k v -> 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, Key key, Value val) + => (forall n. (AllocM n, MonadMask n) => Tree key val -> n (Transaction key val a)) + -> ConcurrentDb key val -> 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, Key k, Value v) + => (forall n. (AllocM n, MonadMask n) => Tree k v -> n (Transaction k v a)) + -> ConcurrentDb k v -> m a +transactNow act db = withRLock (concurrentDbWriterLock db) $ + actAndCommit db $ \meta -> do + tx <- act (concurrentMetaTree meta) + case tx of + Abort v -> return (Nothing, v) + Commit tree v -> + let meta' = meta { concurrentMetaTree = tree } in + return (Just meta', v) + +-- | Execute a write transaction, without a result. +transact_ :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Key k, Value v) + => (forall n. (AllocM n, MonadMask n) => Tree k v -> n (Transaction k v ())) + -> ConcurrentDb k v -> m () +transact_ act db = void $ transact act db + +-- | Execute a read-only transaction. +transactReadOnly :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Key key, Value val) + => (forall n. (AllocReaderM n, MonadMask m) => Tree key val -> n a) + -> ConcurrentDb key val -> m a +transactReadOnly act db = + bracket acquireMeta + releaseMeta $ + \meta -> evalConcurrentT (act $ concurrentMetaTree 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 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, Key k, Value v) + => ConcurrentDb k v + -> (forall n. (MonadIO n, MonadMask n, ConcurrentMetaStoreM n) + => ConcurrentMeta k v + -> ConcurrentT WriterEnv ConcurrentHandles n (Maybe (ConcurrentMeta k v), a) + ) + -> m a +actAndCommit db act + | ConcurrentDb + { concurrentDbHandles = hnds + , concurrentDbWriterLock = lock + , concurrentDbReaders = readers + } <- db + = withRLock lock $ 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) + (concurrentMetaDataFreshUnusedPages meta) + (concurrentMetaIndexFreshUnusedPages 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 }) + handleFreedDirtyPages + + -- 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 k v -> ConcurrentMeta k v +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 k v, 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 k v, WriterEnv ConcurrentHandles) m () +saveFreePages' paranoid cons getState setState + {- paranoid >= 100 = error "paranoid: looping!" + | otherwise-} + = do + + -- Saving the free pages + -- ===================== + -- + -- Saving free pages to the free database is a complicated task. At the + -- end of a transaction we have 3 types of free pages: + -- + -- 1. 'DirtyFree': Pages that were freshly allocated from the end of + -- the dabase file, but are no longer used. These are free'd + -- by saving them in the metadata. They can freely be used + -- during this routine. + -- + -- 2. 'NewlyFreed': Pages that were written by a previous transaction, + -- but free'd in this transaction. They might still be in use + -- by an older reader, and can thus not be used anyways. + -- + -- Note that this list **may grow during this routine**, as + -- new pages can be free'd. + -- + -- 3. 'OldFree': Pages that were fetched from the free database while + -- executing the transaction. Technically, they can be used + -- during this routine, BUT that would mean the list of + -- 'OldFree' pages can grow and shrink during the call, which + -- would complicate the convergence/termination conditions of + -- this routine. So currently, **we disable the use of these + -- pages in this routine.** + + (meta, env) <- get + let tx = writerTxId env + (tree', envWithoutTree) <- lift $ + runConcurrentT (saveFreePages tx (getState env)) $ + env { writerReusablePagesOn = 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 dirty pages. +-- +-- Save the newly created free dirty pages to the metadata for later use. +-- +-- Update the database size. +handleFreedDirtyPages :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m) + => StateT (ConcurrentMeta k v, WriterEnv ConcurrentHandles) m () +handleFreedDirtyPages = do + (meta, env) <- get + + let dataEnv = writerDataFileState env + let indexEnv = writerIndexFileState env + + let meta' = meta { concurrentMetaDataNumPages = + fileStateNewNumPages dataEnv + , concurrentMetaDataFreeTree = + fileStateFreeTree dataEnv + , concurrentMetaDataFreshUnusedPages = + fileStateFreedDirtyPages dataEnv + + , concurrentMetaIndexNumPages = + fileStateNewNumPages indexEnv + , concurrentMetaIndexFreeTree = + fileStateFreeTree indexEnv + , concurrentMetaIndexFreshUnusedPages = + fileStateFreedDirtyPages indexEnv + } + put (meta', env) + +-------------------------------------------------------------------------------- diff --git a/src/Database/Haskey/Alloc/Concurrent/Environment.hs b/src/Database/Haskey/Alloc/Concurrent/Environment.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Environment.hs @@ -0,0 +1,310 @@ +{-# LANGUAGE DataKinds #-} +{-# 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.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 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. + + , fileStateFreedDirtyPages :: !(S stateType (Set DirtyFree)) + -- ^ Pages freshly allocated AND free'd in this transaction. Immediately + -- ready for reuse. + + , fileStateFreeTree :: !(S stateType FreeTree) + -- ^ The root of the free tree, might change during a transaction. + + , fileStateDirtyReusablePages :: !(Set DirtyOldFree) + -- ^ All pages queried from the free page database for + -- 'fileStateReusablePagesTxId', and actually used once already. + + , fileStateReusablePages :: ![OldFree] + -- ^ Pages queried from the free pages database and ready for immediate + -- reuse. + + , fileStateReusablePagesTxId :: !(Maybe TxId) + -- ^ The 'TxId' of the pages in 'fileStateReusablePages', or 'Nothing' if no + -- pages were queried yet from the free database. + + } + +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. + + , writerReusablePagesOn :: !Bool + -- ^ Used to turn of querying the free page database for free pages. + + , 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 (Set DirtyFree) -> S 'TypeIndex (Set DirtyFree) + -> S 'TypeData FreeTree -> S 'TypeIndex FreeTree + -> WriterEnv hnd +newWriter hnd tx readers + numDataPages numIndexPages + dataDirtyFree indexDirtyFree + dataFreeTree indexFreeTree = + WriterEnv { + writerHnds = hnd + , writerTxId = tx + , writerReaders = readers + + , writerIndexFileState = newFileState numIndexPages indexDirtyFree indexFreeTree + , writerDataFileState = newFileState numDataPages dataDirtyFree dataFreeTree + + , writerReusablePagesOn = True + , writerDirtyOverflows = S.empty + , writerOverflowCounter = 0 + , writerRemovedOverflows = [] + } + where + newFileState numPages dirtyFree freeTree = FileState { + fileStateNewlyFreedPages = [] + , fileStateOriginalNumPages = numPages + , fileStateNewNumPages = numPages + , fileStateFreedDirtyPages = dirtyFree + , fileStateFreeTree = freeTree + , fileStateDirtyReusablePages = S.empty + , fileStateReusablePages = [] + , fileStateReusablePagesTxId = Nothing + } + +-- | Wrapper around 'PageId' indicating it is a fresh page, allocated at the +-- end of the database. +newtype Fresh = Fresh PageId deriving (Eq, Ord, Show) + +-- | 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 a dirty page. +newtype Dirty = Dirty PageId deriving (Eq, Ord, Show) + +-- | Wrapper around 'PageId' indicating the page is dirty and free for reuse. +newtype DirtyFree = DirtyFree PageId deriving (Binary, Eq, Ord, Show) + +-- | Wrapper around 'PageId' inidcating it was fetched from the free database +-- and is ready for reuse. +newtype OldFree = OldFree PageId deriving (Eq, Ord, Show) + +-- | Wrapper around 'PageId' indicating it wa fetched from the free database +-- and is actually dirty. +newtype DirtyOldFree = DirtyOldFree PageId deriving (Eq, Ord, Show) + +-- | A sum type repesenting any type of free page, that can immediately be used +-- to write something to. +data SomeFreePage = FreshFreePage Fresh + | DirtyFreePage DirtyFree + | OldFreePage OldFree + +getSomeFreePageId :: SomeFreePage -> PageId +getSomeFreePageId (FreshFreePage (Fresh pid)) = pid +getSomeFreePageId (DirtyFreePage (DirtyFree pid)) = pid +getSomeFreePageId (OldFreePage (OldFree pid)) = pid + +-- | Try to free a page, given a set of dirty pages. +-- +-- If the page was dirty, a 'DirtyFree' 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 + dirtyOldFree' <- dirtyOldFree pid + modify' $ \e -> + e { writerDataFileState = + updateFileState (writerDataFileState e) DataState + dirty' dirtyOldFree' pid' + } + +freePage pid@(IndexState pid') = do + dirty' <- dirty pid + dirtyOldFree' <- dirtyOldFree pid + modify' $ \e -> + e { writerIndexFileState = + updateFileState (writerIndexFileState e) IndexState + dirty' dirtyOldFree' pid' + } + +updateFileState :: FileState t + -> (forall a. a -> S t a) + -> Maybe Dirty + -> Maybe DirtyOldFree + -> PageId + -> FileState t +updateFileState e cons dirty' dirtyOldFree' pid' = + if | Just (Dirty p) <- dirty' -> + e { fileStateFreedDirtyPages = + cons $ S.insert (DirtyFree p) (getSValue $ fileStateFreedDirtyPages e) } + + | Just (DirtyOldFree p) <- dirtyOldFree' -> + e { fileStateReusablePages = + OldFree p : fileStateReusablePages 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 . fileStateOriginalNumPages . writerDataFileState) <$> get + IndexState p -> (page p . fileStateOriginalNumPages . writerIndexFileState) <$> get + where + page p origNumPages + | p >= getSValue origNumPages = Just (Dirty p) + | otherwise = Nothing + +-- | Get a 'DirtyOldFree' page, by first proving it is in fact a dirty old free page. +dirtyOldFree :: (Functor m, MonadState (WriterEnv hnd) m) => S stateType PageId -> m (Maybe DirtyOldFree) +dirtyOldFree pid = case pid of + DataState p -> (page p . fileStateDirtyReusablePages . writerDataFileState) <$> get + IndexState p -> (page p . fileStateDirtyReusablePages . writerIndexFileState) <$> get + where + page p dirty' + | S.member (DirtyOldFree p) dirty' = Just (DirtyOldFree p) + | otherwise = Nothing + + +-- | Touch a fresh page, make it dirty. +-- +-- We really need lenses... +touchPage :: MonadState (WriterEnv hnd) m => S stateType SomeFreePage -> m () +touchPage (DataState (DirtyFreePage _)) = return() +touchPage (IndexState (DirtyFreePage _)) = return () + +touchPage (DataState (FreshFreePage (Fresh pid))) = modify' $ \e -> + case fileStateNewNumPages (writerDataFileState e) of + DataState numPages -> + if numPages < pid + 1 + then e { writerDataFileState = (writerDataFileState e) { + fileStateNewNumPages = DataState (pid + 1) } + } + else e +touchPage (IndexState (FreshFreePage (Fresh pid))) = modify' $ \e -> + case fileStateNewNumPages (writerIndexFileState e) of + IndexState numPages -> + if numPages < pid + 1 + then e { writerIndexFileState = (writerIndexFileState e) { + fileStateNewNumPages = IndexState (pid + 1) } + } + else e + +touchPage (DataState (OldFreePage (OldFree pid))) = modify' $ \e -> + let s = fileStateDirtyReusablePages (writerDataFileState e) in + e { writerDataFileState = (writerDataFileState e) { + fileStateDirtyReusablePages = S.insert (DirtyOldFree pid) s } + } +touchPage (IndexState (OldFreePage (OldFree pid))) = modify' $ \e -> + let s = fileStateDirtyReusablePages (writerIndexFileState e) in + e { writerIndexFileState = (writerIndexFileState e) { + fileStateDirtyReusablePages = S.insert (DirtyOldFree pid) s } + } + +-- | 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 new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/FreePages/Query.hs @@ -0,0 +1,185 @@ +{-# 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 qualified Data.Set as S + +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 'writerReusablePages'. 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 SomeFreePage) +getFreePageId t = + runMaybeT $ (DirtyFreePage <$> MaybeT (getFreedDirtyPageId t)) + <|> (OldFreePage <$> MaybeT (getCachedFreePageId t)) + <|> (OldFreePage <$> MaybeT (queryNewFreePageIds t)) + +-- | Get a free'd dirty page. +-- +-- Get a free'd dirty page, that is immediately suitable for reuse in the +-- current transaction. +getFreedDirtyPageId :: (Functor m, MonadState (WriterEnv hnd) m) + => S stateType () + -> m (Maybe DirtyFree) +getFreedDirtyPageId stateType = + case stateType of + DataState () -> do + s <- writerDataFileState <$> get + let (pid, s') = query s DataState + modify' $ \env -> env { writerDataFileState = s' } + return pid + IndexState () -> do + s <- writerIndexFileState <$> get + let (pid, s') = query s IndexState + modify' $ \env -> env { writerIndexFileState = s' } + return pid + where + query :: FileState t + -> (forall a. a -> S t a) + -> (Maybe DirtyFree, FileState t) + query env cons = + case S.minView (getSValue $ fileStateFreedDirtyPages env) of + Nothing -> (Nothing, env) + Just (pid, s') -> + let env' = env { fileStateFreedDirtyPages = cons s' } in + (Just pid, env') + +-- | Get a cached free page. +-- +-- Get a free page from the free database cache stored in 'writerReusablePages'. +getCachedFreePageId :: (Functor m, MonadState (WriterEnv hnd) m) + => S stateType () + -> m (Maybe OldFree) +getCachedFreePageId stateType = + ifM (not . writerReusablePagesOn <$> get) (return Nothing) $ + case stateType of + DataState () -> do + s <- writerDataFileState <$> get + let (pid, s') = query s + modify' $ \env -> env { writerDataFileState = s' } + return pid + IndexState () -> do + s <- writerIndexFileState <$> get + let (pid, s') = query s + modify' $ \env -> env { writerIndexFileState = s' } + return pid + where + query :: FileState t -> (Maybe OldFree, FileState t) + query env = case fileStateReusablePages env of + [] -> (Nothing, env) + pid : pageIds -> + let env' = env { fileStateReusablePages = 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. +-- +-- This function will delete the lastly used entry from the free database, +-- query a new one, and then update the free page cache in the state. +-- +-- This function only works when 'writerReusablePagesOn' is 'True'. +-- +-- This function expects 'writerReusablePages' to be empty. +queryNewFreePageIds :: (AllocM m, MonadIO m, MonadState (WriterEnv hnd) m) + => S stateType () + -> m (Maybe OldFree) +queryNewFreePageIds stateType = ifM (not . writerReusablePagesOn <$> get) (return Nothing) $ + case stateType of + DataState () -> + query DataState + writerDataFileState + (\e s -> e { writerDataFileState = s }) + + IndexState () -> + query IndexState + writerIndexFileState + (\e s -> e { writerIndexFileState = s }) + 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 (Maybe OldFree) + query cons getState setState = do + tree <- gets $ getSValue . fileStateFreeTree . getState + oldTxId <- gets $ fileStateReusablePagesTxId . getState + + -- Delete the previous used 'TxId' from the tree. + modify' $ \e -> e { writerReusablePagesOn = False } + tree' <- maybe (return tree) (`deleteSubtree` tree) oldTxId + modify' $ \e -> e { writerReusablePagesOn = True } + + -- Set the new free tree + modify' $ \e -> setState e $ + (getState e) { fileStateFreeTree = cons tree' } + + -- Lookup the oldest free page + lookupValidFreePageIds tree' >>= \case + Nothing -> do + modify' $ \e -> setState e $ + (getState e) { fileStateDirtyReusablePages = S.empty + , fileStateReusablePages = [] + , fileStateReusablePagesTxId = Nothing } + return Nothing + Just (txId, pid :| pageIds) -> do + modify' $ \e -> setState e $ + (getState e) { fileStateDirtyReusablePages = S.empty + , fileStateReusablePages = map OldFree pageIds + , fileStateReusablePagesTxId = Just txId } + return (Just $ OldFree pid) + +-- | 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 + if maybe True (> fst v) oldest + 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 new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/FreePages/Save.hs @@ -0,0 +1,49 @@ +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 + >>= saveCachedFreePages env + where + tree = getSValue $ fileStateFreeTree env + +-- | Save the newly free pages of the current transaction, as stored by +-- 'writerNewlyFreedPages'. +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 + +-- | Save the free apges from the free page cache in +-- 'writerReusablePages' using 'writerReuseablePagesTxId'. +saveCachedFreePages :: AllocM m + => FileState t + -> FreeTree + -> m FreeTree +saveCachedFreePages env tree = case fileStateReusablePagesTxId env of + Nothing -> return tree + Just k -> + case freePages of + [] -> deleteSubtree k tree + x:xs -> replaceSubtree k (x :| xs) tree + where + freePages = map (\(OldFree pid) -> pid) $ fileStateReusablePages env diff --git a/src/Database/Haskey/Alloc/Concurrent/FreePages/Tree.hs b/src/Database/Haskey/Alloc/Concurrent/FreePages/Tree.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/FreePages/Tree.hs @@ -0,0 +1,67 @@ +{-# 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/Meta.hs b/src/Database/Haskey/Alloc/Concurrent/Meta.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Meta.hs @@ -0,0 +1,61 @@ +{-# LANGUAGE DataKinds #-} +{-# 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.Set as Set + +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 + +-- | Data type used to point to the most recent version of the meta data. +data CurrentMetaPage = Meta1 | Meta2 + +-- | Meta data of the page allocator. +data ConcurrentMeta k v = ConcurrentMeta { + concurrentMetaRevision :: TxId + , concurrentMetaDataNumPages :: S 'TypeData PageId + , concurrentMetaIndexNumPages :: S 'TypeIndex PageId + , concurrentMetaTree :: Tree k v + , concurrentMetaDataFreeTree :: S 'TypeData FreeTree + , concurrentMetaIndexFreeTree :: S 'TypeIndex FreeTree + , concurrentMetaOverflowTree :: OverflowTree + , concurrentMetaDataFreshUnusedPages :: S 'TypeData (Set DirtyFree) + , concurrentMetaIndexFreshUnusedPages :: S 'TypeIndex (Set DirtyFree) + } deriving (Generic) + +deriving instance (Show k, Show v) => Show (ConcurrentMeta k v) + +instance (Binary k, Binary v) => Binary (ConcurrentMeta k v) 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 :: (Key k, Value v) + => FilePath + -> ConcurrentMeta k v + -> m () + + -- | Try to read the meta-data structure from a handle, or return 'Nothing' + -- if the handle doesn't contain a meta page. + readConcurrentMeta :: (Key k, Value v) + => FilePath + -> Proxy k + -> Proxy v + -> m (Maybe (ConcurrentMeta k v)) + diff --git a/src/Database/Haskey/Alloc/Concurrent/Monad.hs b/src/Database/Haskey/Alloc/Concurrent/Monad.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Monad.hs @@ -0,0 +1,191 @@ +{-# 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 { + 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 { + 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 (getSomeFreePageId 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 + let pid' = FreshFreePage . Fresh <$> pid + touchPage pid' + return $ getSValue pid' + USucc _ -> do + pid <- fileStateNewNumPages . writerIndexFileState <$> get + let pid'' = FreshFreePage . Fresh <$> pid + 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) -> 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 new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Concurrent/Overflow.hs @@ -0,0 +1,131 @@ +{-# 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 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, 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 + +-------------------------------------------------------------------------------- + +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/Alloc/Transaction.hs b/src/Database/Haskey/Alloc/Transaction.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Alloc/Transaction.hs @@ -0,0 +1,26 @@ +-- | This module implements mechanisms to work with transactions. +module Database.Haskey.Alloc.Transaction where + +import Data.BTree.Alloc.Class +import Data.BTree.Impure.Structures + +-- | A committed or aborted transaction, with a return value of type @a@. +data Transaction key val a = + Commit (Tree key val) a + | Abort a + +-- | Commit the new tree and return a computed value. +commit :: AllocM n => a -> Tree key val -> n (Transaction key val a) +commit v t = return $ Commit t v + +-- | Commit the new tree, without return a computed value. +commit_ :: AllocM n => Tree key val -> n (Transaction key val ()) +commit_ = commit () + +-- | Abort the transaction and return a computed value. +abort :: AllocM n => a -> n (Transaction key val a) +abort = return . Abort + +-- | Abort the transaction, without returning a computed value. +abort_ :: AllocM n => n (Transaction key val ()) +abort_ = return $ Abort () diff --git a/src/Database/Haskey/Store.hs b/src/Database/Haskey/Store.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Store.hs @@ -0,0 +1,6 @@ +-- | Storage back-ends that manage physical storage of pages. +module Database.Haskey.Store ( + module Database.Haskey.Store.Class +) where + +import Database.Haskey.Store.Class diff --git a/src/Database/Haskey/Store/Class.hs b/src/Database/Haskey/Store/Class.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Store/Class.hs @@ -0,0 +1,238 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE FunctionalDependencies #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE UndecidableInstances #-} +-- | A storage back-end manages physical storage of pages. +module Database.Haskey.Store.Class ( + -- * Class + StoreM(..) + + -- * Helpers +, arbitrarySearch +, calculateMaxKeySize +, calculateMaxValueSize +, ZeroEncoded(..) +) where + +import Prelude hiding (max, min, pred) + +import Control.Applicative (Applicative) +import Control.Monad.Trans +import Control.Monad.Trans.Reader (ReaderT) +import Control.Monad.Trans.State (StateT) + +import Data.Binary (Binary(..), Get) +import Data.Proxy +import Data.Typeable (Typeable) +import Data.Word (Word8, Word64) +import qualified Data.Map as M + +import Data.BTree.Impure +import Data.BTree.Impure.Structures +import Data.BTree.Primitives + +-------------------------------------------------------------------------------- + +-- | A storage back-end that can store and fetch physical pages. +class (Applicative m, Monad m) => StoreM hnd m | m -> hnd where + -- | Open a database handle for reading and writing. + openHandle :: hnd -> m () + + -- | Flush the contents of a handle to disk (or other storage). + flushHandle :: hnd -> m () + + -- | Close a database handle. + closeHandle :: hnd -> m () + + -- | Remove a handle from the storage back-end. + removeHandle :: hnd -> m () + + -- | A function that calculates the hypothetical size of a node, if it were + -- to be written to a page (regardless of the maximum page size). + nodePageSize :: (Key key, Value val) + => m (Height height -> Node height key val -> PageSize) + + -- | The maximum page size the allocator can handle. + maxPageSize :: m PageSize + + -- | Get the maximum key size + -- + -- The default implementation will repeatedly call 'calculateMaxKeySize'. + -- You might want to cache this value in your own implementation. + maxKeySize :: m Word64 + maxKeySize = do + f <- nodePageSize + fmax <- maxPageSize + return $ calculateMaxKeySize fmax (f zeroHeight) + + -- | Get the maximum value size + -- + -- The default implementation will repeatedly call 'calculateMaxValueSize'. + -- You might want to cache this value in your own implementation. + maxValueSize :: m Word64 + maxValueSize = do + f <- nodePageSize + key <- maxKeySize + fmax <- maxPageSize + return $ calculateMaxValueSize fmax key (f zeroHeight) + + -- | Read a page and return the actual node and the transaction id when the + -- node was written. + getNodePage :: (Key key, Value val) + => hnd + -> Height height + -> Proxy key + -> Proxy val + -> NodeId height key val + -> m (Node height key val) + + -- | Write a node to a physical page. + putNodePage :: (Key key, Value val) + => hnd + -> Height height + -> NodeId height key val + -> Node height key val + -> m () + + -- | Read a value from an overflow page + getOverflow :: (Value val) + => hnd + -> Proxy val + -> m val + + -- | Write a value to an overflow page + putOverflow :: (Value val) + => hnd + -> val + -> m () + + -- | List overflow pages in the specific overflow directory. + -- + -- The result should include **AT LEAST** the handles in the specified + -- directory, but it may contain more handles, even handles that do not + -- belong to an overflow page. + listOverflows :: hnd -> m [hnd] + + +instance StoreM hnd m => StoreM hnd (StateT s m) where + openHandle = lift. openHandle + flushHandle = lift. flushHandle + closeHandle = lift. closeHandle + removeHandle = lift. closeHandle + nodePageSize = lift nodePageSize + maxPageSize = lift maxPageSize + maxKeySize = lift maxKeySize + maxValueSize = lift maxValueSize + getNodePage = ((((lift.).).).). getNodePage + putNodePage = (((lift.).).). putNodePage + getOverflow = (lift.). getOverflow + putOverflow = (lift.). putOverflow + listOverflows = lift. listOverflows + +instance StoreM hnd m => StoreM hnd (ReaderT s m) where + openHandle = lift. openHandle + flushHandle = lift. flushHandle + closeHandle = lift. closeHandle + removeHandle = lift. closeHandle + nodePageSize = lift nodePageSize + maxPageSize = lift maxPageSize + maxKeySize = lift maxKeySize + maxValueSize = lift maxValueSize + getNodePage = ((((lift.).).).). getNodePage + putNodePage = (((lift.).).). putNodePage + getOverflow = (lift.). getOverflow + putOverflow = (lift.). putOverflow + listOverflows = lift. listOverflows + +-------------------------------------------------------------------------------- + +-- | Calculate the maximum key size. +-- +-- Return the size for which at least 4 key-value pairs with keys and values of +-- that size can fit in a leaf node. +calculateMaxKeySize :: PageSize + -- ^ Maximum pages size + -> (Node 'Z ZeroEncoded ZeroEncoded -> PageSize) + -- ^ Function that calculates the page size of a node + -> Word64 + -- ^ Maximum key size +calculateMaxKeySize fmax f = arbitrarySearch 2 pred fmax + where + pred n = f (Leaf $ kvs n) + kvs n = M.fromList + [(ZeroEncoded n i, RawValue $ ZeroEncoded n i) | i <- [1..4]] + +-- | Calculate the maximum value size. +-- +-- Return the size for which at least 4 key-value pairs of the specified +-- maximum key size and values of the returned size can fit in a leaf node. +-- that size can fit in a leaf node. +calculateMaxValueSize :: PageSize + -- ^ Maximum page size + -> Word64 + -- ^ Maximum key size + -> (Node 'Z ZeroEncoded ZeroEncoded -> PageSize) + -- ^ Function that calculates the page size of a node + -> Word64 + -- ^ Maximum value size +calculateMaxValueSize fmax keySize f = arbitrarySearch 2 pred fmax + where + pred n = f (Leaf $ kvs n) + kvs n = M.fromList + [(ZeroEncoded keySize i, RawValue $ ZeroEncoded n i) | i <- [1..4]] + +-- | Search an arbitrary number, less than a limit, greater than a starting +-- value. +arbitrarySearch :: (Ord v, Integral n) => n -> (n -> v) -> v -> n +arbitrarySearch start f fmax = go start + where + go n = + let s = f n in + if s == fmax + then n + else if s > fmax + then search (n `quot` 2) n + else go (n*2) + + search min max + | max - min == 1 = min + | otherwise + = + let c = min + ((max - min) `quot` 2) + s = f c in + if s == fmax + then c + else if s > fmax + then search min c + else search c max + +-- | Data type which encodes the integer using a variable amount of NULL or ONE +-- bytes. +data ZeroEncoded = ZeroEncoded { getZeroEncoded :: Word64 + , getZeroEncodedValue :: Word64 } + deriving (Eq, Ord, Show, Typeable) + + +instance Binary ZeroEncoded where + put (ZeroEncoded 0 _) = error "must be >0" + put (ZeroEncoded 1 0) = put (255 :: Word8) + put (ZeroEncoded 1 _) = error "value too large" + put (ZeroEncoded n v) = put byte >> put (ZeroEncoded (n-1) v') + where + byte = fromIntegral $ v `rem` 255 :: Word8 + v' = v `quot` 255 + + get = do + byte <- get :: Get Word8 + case byte of + 0 -> return (ZeroEncoded 1 0) + _ -> do + next <- get + return $ ZeroEncoded (getZeroEncoded next + 1) 0 + +instance Key ZeroEncoded where +instance Value ZeroEncoded where + +-------------------------------------------------------------------------------- diff --git a/src/Database/Haskey/Store/File.hs b/src/Database/Haskey/Store/File.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Store/File.hs @@ -0,0 +1,304 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE ScopedTypeVariables #-} +-- | On-disk storage back-end. Can be used as a storage back-end for the +-- append-only page allocator (see "Data.BTree.Alloc"). +module Database.Haskey.Store.File ( + -- * Storage + Page(..) +, Files +, FileStoreConfig(..) +, defFileStoreConfig +, fileStoreConfigWithPageSize +, FileStoreT +, runFileStoreT +, newFileStore + + -- * Binary encoding +, encodeAndPad + + -- * Exceptions +, FileNotFoundError(..) +, PageOverflowError(..) +, WrongNodeTypeError(..) +, WrongOverflowValueError(..) +) where + +import Control.Applicative (Applicative, (<$>)) +import Control.Monad +import Control.Monad.Catch +import Control.Monad.IO.Class +import Control.Monad.Reader + +import Data.Coerce (coerce) +import Data.Map (Map) +import Data.Maybe (fromJust) +import Data.Monoid ((<>)) +import Data.IORef +import Data.Typeable (Typeable) +import Data.Word (Word64) +import qualified Data.ByteString.Lazy as BL +import qualified Data.Map as M + +import qualified FileIO as IO + +import System.Directory (createDirectoryIfMissing, removeFile, getDirectoryContents) +import System.FilePath (takeDirectory) +import System.IO.Error (ioError, isDoesNotExistError) + +import Data.BTree.Impure.Structures +import Data.BTree.Primitives + +import Database.Haskey.Alloc.Concurrent +import Database.Haskey.Store.Class +import Database.Haskey.Store.Page +import Database.Haskey.Utils.IO (readByteString, writeLazyByteString) +import Database.Haskey.Utils.Monad.Catch (justErrM) + +-------------------------------------------------------------------------------- + +-- | Encode a page padding it to the maxim page size. +-- +-- Return 'Nothing' of the page is too large to fit into one page size. +encodeAndPad :: PageSize -> Page t -> Maybe BL.ByteString +encodeAndPad size page + | Just n <- padding = Just . prependChecksum $ + enc <> BL.replicate n 0 + | otherwise = Nothing + where + enc = encodeNoChecksum page + + -- Reserve 8 bytes for the checksum + padding | n <- fromIntegral size - BL.length enc - 8, n >= 0 = Just n + | otherwise = Nothing + +-------------------------------------------------------------------------------- + +-- | A collection of files, each associated with a certain @fp@ handle. +-- +-- Each file is a 'Handle' opened in 'System.IO.ReadWriteMode' and contains a +-- collection of physical pages. +-- +-- These files can be safely shared between threads. +type Files fp = IORef (Map fp IO.FHandle) + +-- | Access the files. +get :: MonadIO m => FileStoreT fp m (Map fp IO.FHandle) +get = FileStoreT . lift $ ask >>= liftIO . readIORef + +-- | Modify the files. +modify' :: MonadIO m + => (Map fp IO.FHandle -> Map fp IO.FHandle) + -> FileStoreT fp m () +modify' f = FileStoreT . lift $ ask >>= liftIO . flip modifyIORef' f + +lookupHandle :: (Functor m, MonadThrow m, Ord fp, Show fp, Typeable fp) + => fp -> Map fp IO.FHandle -> m IO.FHandle +lookupHandle fp m = justErrM (FileNotFoundError fp) $ M.lookup fp m + +-- | Monad in which on-disk storage operations can take place. +-- +-- Two important instances are 'StoreM' making it a storage back-end, and +-- 'ConcurrentMetaStoreM' making it a storage back-end compatible with the +-- concurrent page allocator. +newtype FileStoreT fp m a = FileStoreT + { fromFileStoreT :: ReaderT FileStoreConfig (ReaderT (Files fp) m) a + } deriving (Applicative, Functor, Monad, + MonadIO, MonadThrow, MonadCatch, MonadMask, + MonadReader FileStoreConfig) + +-- | File store configuration. +-- +-- The default configuration can be obtained by using 'defFileStoreConfig' +-- +-- A configuration with a specific page size can be obtained by using +-- 'fileStoreConfigWithPageSize'. +data FileStoreConfig = FileStoreConfig { + fileStoreConfigPageSize :: !PageSize + , fileStoreConfigMaxKeySize :: !Word64 + , fileStoreConfigMaxValueSize :: !Word64 + } deriving (Show) + +-- | The default configuration +-- +-- This is an unwrapped 'fileStoreConfigWithPageSize' with a page size of 4096 +-- bytes. +defFileStoreConfig :: FileStoreConfig +defFileStoreConfig = fromJust (fileStoreConfigWithPageSize 4096) + +-- | Create a configuration with a specific page size. +-- +-- The maximum key and value sizes are calculated using 'calculateMaxKeySize' +-- and 'calculateMaxValueSize'. +-- +-- If the page size is too small, 'Nothing' is returned. +fileStoreConfigWithPageSize :: PageSize -> Maybe FileStoreConfig +fileStoreConfigWithPageSize pageSize + | keySize < 8 && valueSize < 8 = Nothing + | otherwise = Just FileStoreConfig { + fileStoreConfigPageSize = pageSize + , fileStoreConfigMaxKeySize = keySize + , fileStoreConfigMaxValueSize = valueSize } + where + keySize = calculateMaxKeySize pageSize (encodedPageSize zeroHeight) + valueSize = calculateMaxValueSize pageSize keySize (encodedPageSize zeroHeight) + +-- | Run the storage operations in the 'FileStoreT' monad, given a collection of +-- open files. +runFileStoreT :: FileStoreT fp m a -- ^ Action + -> FileStoreConfig -- ^ Configuration + -> Files fp -- ^ Open files + -> m a +runFileStoreT m config = runReaderT (runReaderT (fromFileStoreT m) config) + +-- | An empty file store, with no open files. +newFileStore :: IO (Files fp) +newFileStore = newIORef M.empty + +-------------------------------------------------------------------------------- + +instance (Applicative m, Monad m, MonadIO m, MonadThrow m) => + StoreM FilePath (FileStoreT FilePath m) + where + openHandle fp = do + alreadyOpen <- M.member fp <$> get + unless alreadyOpen $ do + liftIO $ createDirectoryIfMissing True (takeDirectory fp) + fh <- liftIO $ IO.openReadWrite fp + modify' $ M.insert fp fh + + flushHandle fp = do + fh <- get >>= lookupHandle fp + liftIO $ IO.flush fh + + closeHandle fp = do + fh <- get >>= lookupHandle fp + liftIO $ IO.flush fh + liftIO $ IO.close fh + modify' (M.delete fp) + + removeHandle fp = + liftIO $ removeFile fp `catchIOError` \e -> + unless (isDoesNotExistError e) (ioError e) + + + nodePageSize = return encodedPageSize + maxPageSize = asks fileStoreConfigPageSize + maxKeySize = asks fileStoreConfigMaxKeySize + maxValueSize = asks fileStoreConfigMaxValueSize + + getNodePage fp height key val nid = do + h <- get >>= lookupHandle fp + size <- maxPageSize + + let PageId pid = nodeIdToPageId nid + offset = fromIntegral $ pid * fromIntegral size + + liftIO $ IO.seek h offset + bs <- liftIO $ readByteString h (fromIntegral size) + + case viewHeight height of + UZero -> decodeM (leafNodePage height key val) bs >>= \case + LeafNodePage hgtSrc tree -> + justErrM WrongNodeTypeError $ castNode hgtSrc height tree + USucc _ -> decodeM (indexNodePage height key val) bs >>= \case + IndexNodePage hgtSrc tree -> + justErrM WrongNodeTypeError $ castNode hgtSrc height tree + + putNodePage fp hgt nid node = do + h <- get >>= lookupHandle fp + size <- maxPageSize + + let PageId pid = nodeIdToPageId nid + offset = fromIntegral $ pid * fromIntegral size + + liftIO $ IO.seek h offset + bs <- justErrM PageOverflowError $ pg size + liftIO $ writeLazyByteString h bs + where + pg size = case viewHeight hgt of + UZero -> encodeAndPad size $ LeafNodePage hgt node + USucc _ -> encodeAndPad size $ IndexNodePage hgt node + + getOverflow fp val = do + h <- get >>= lookupHandle fp + + len <- liftIO $ IO.getFileSize h + liftIO $ IO.seek h 0 + bs <- liftIO $ readByteString h (fromIntegral len) + n <- decodeM (overflowPage val) bs + case n of + OverflowPage v -> justErrM WrongOverflowValueError $ castValue v + + + putOverflow fp val = do + fh <- get >>= lookupHandle fp + liftIO $ IO.setFileSize fh (fromIntegral $ BL.length bs) + liftIO $ IO.seek fh 0 + liftIO $ writeLazyByteString fh bs + where + bs = encode $ OverflowPage val + + listOverflows dir = liftIO $ getDirectoryContents dir `catch` catch' + where catch' e | isDoesNotExistError e = return [] + | otherwise = ioError e + +-------------------------------------------------------------------------------- + +instance (Applicative m, Monad m, MonadIO m, MonadThrow m) => + ConcurrentMetaStoreM (FileStoreT FilePath m) + where + putConcurrentMeta fp meta = do + h <- get >>= lookupHandle fp + + let page = ConcurrentMetaPage meta + bs = encode page + liftIO $ IO.setFileSize h (fromIntegral $ BL.length bs) + liftIO $ IO.seek h 0 + liftIO $ writeLazyByteString h bs + + readConcurrentMeta fp k v = do + fh <- get >>= lookupHandle fp + + len <- liftIO $ IO.getFileSize fh + liftIO $ IO.seek fh 0 + bs <- liftIO $ readByteString fh (fromIntegral len) + decodeM (concurrentMetaPage k v) bs >>= \case + ConcurrentMetaPage meta -> return $ Just (coerce meta) + +-------------------------------------------------------------------------------- + +-- | Exception thrown when a file is accessed that doesn't exist. +newtype FileNotFoundError hnd = FileNotFoundError hnd deriving (Show, Typeable) + +instance (Typeable hnd, Show hnd) => Exception (FileNotFoundError hnd) where + +-- | Exception thrown when a page that is too large is written. +-- +-- As used in 'putNodePage'. +data PageOverflowError = PageOverflowError deriving (Show, Typeable) + +instance Exception PageOverflowError where + +-- | Exception thrown when a node cannot be cast to the right type. +-- +-- As used in 'getNodePage'. +data WrongNodeTypeError = WrongNodeTypeError deriving (Show, Typeable) + +instance Exception WrongNodeTypeError where + +-- | Exception thrown when a value from an overflow page cannot be cast. +-- +-- As used in 'getOverflow'. +data WrongOverflowValueError = WrongOverflowValueError deriving (Show, Typeable) + +instance Exception WrongOverflowValueError where + +-------------------------------------------------------------------------------- diff --git a/src/Database/Haskey/Store/InMemory.hs b/src/Database/Haskey/Store/InMemory.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Store/InMemory.hs @@ -0,0 +1,242 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE InstanceSigs #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE ScopedTypeVariables #-} +-- | Binary in-memory storage back-end. Can be used as a storage back-end for +-- the append-only page allocator (see "Data.BTree.Alloc"). +module Database.Haskey.Store.InMemory ( + -- * Storage + Page(..) +, MemoryFile +, MemoryFiles +, MemoryStoreConfig(..) +, defMemoryStoreConfig +, memoryStoreConfigWithPageSize +, MemoryStoreT +, runMemoryStoreT +, newEmptyMemoryStore + + -- * Exceptions +, FileNotFoundError(..) +, PageNotFoundError(..) +, WrongNodeTypeError(..) +, WrongOverflowValueError(..) +) where + +import Control.Applicative (Applicative, (<$>)) +import Control.Monad +import Control.Monad.Catch +import Control.Monad.IO.Class +import Control.Monad.Reader + +import Data.ByteString (ByteString) +import Data.ByteString.Lazy (toStrict) +import Data.Coerce +import Data.IORef +import Data.Map (Map) +import Data.Maybe (fromJust) +import Data.Typeable (Typeable) +import Data.Word (Word64) +import qualified Data.Map as M + +import Data.BTree.Impure.Structures +import Data.BTree.Primitives + +import Database.Haskey.Alloc.Concurrent +import Database.Haskey.Store.Class +import Database.Haskey.Store.Page +import Database.Haskey.Utils.Monad.Catch (justErrM) + +-------------------------------------------------------------------------------- + +-- | A file containing a collection of pages. +type MemoryFile = Map PageId ByteString + +-- | A collection of 'File's, each associated with a certain @fp@ handle. +-- +-- This is shareable amongst multiple threads. +type MemoryFiles fp = IORef (Map fp MemoryFile) + +-- | Access the files. +get :: MonadIO m => MemoryStoreT fp m (Map fp MemoryFile) +get = MemoryStoreT . lift $ ask >>= liftIO . readIORef + +-- | Access the files. +gets :: (Functor m, MonadIO m) + => (Map fp MemoryFile -> a) + -> MemoryStoreT fp m a +gets f = f <$> get + +-- | Modify the files. +modify' :: MonadIO m => + (Map fp MemoryFile -> Map fp MemoryFile) + -> MemoryStoreT fp m () +modify' f = MemoryStoreT . lift $ ask >>= liftIO . flip modifyIORef' f + +lookupFile :: (MonadThrow m, Ord fp, Show fp, Typeable fp) + => fp -> Map fp MemoryFile -> m MemoryFile +lookupFile fp m = justErrM (FileNotFoundError fp) $ M.lookup fp m + +lookupPage :: (Functor m, MonadThrow m, Ord fp, Show fp, Typeable fp) + => fp -> PageId -> Map fp MemoryFile -> m ByteString +lookupPage fp pid m = M.lookup pid <$> lookupFile fp m + >>= justErrM (PageNotFoundError fp pid) + +-- | Monad in which binary storage operations can take place. +-- +-- Two important instances are 'StoreM' making it a storage back-end, and +-- 'ConcurrentMetaStoreM' making it a storage back-end compatible with the +-- concurrent page allocator. +newtype MemoryStoreT fp m a = MemoryStoreT + { fromMemoryStoreT :: ReaderT MemoryStoreConfig (ReaderT (MemoryFiles fp) m) a + } deriving (Applicative, Functor, Monad, + MonadIO, MonadThrow, MonadCatch, MonadMask, + MonadReader MemoryStoreConfig) + +-- | Memory store configuration. +-- +-- The default configuration can be obtained by using 'defMemoryStoreConfig'. +-- +-- A configuration with a specific page size can be obtained by using +-- 'memoryStoreConfigWithPageSize'. +data MemoryStoreConfig = MemoryStoreConfig { + memoryStoreConfigPageSize :: !PageSize + , memoryStoreConfigMaxKeySize :: !Word64 + , memoryStoreConfigMaxValueSize :: !Word64 + } deriving (Show) + +-- | The default configuration. +-- +-- This is an unwrapped 'memoryStoreConfigWithPageSize' with a page size of +-- 4096. +defMemoryStoreConfig :: MemoryStoreConfig +defMemoryStoreConfig = fromJust (memoryStoreConfigWithPageSize 4096) + +-- | Create a configuration with a specific page size. +-- +-- The maximum key and value sizes are calculated using 'calculateMaxKeySize' +-- and 'calculateMaxValueSize'. +-- +-- If the page size is too small, 'Nothing' is returned. +memoryStoreConfigWithPageSize :: PageSize -> Maybe MemoryStoreConfig +memoryStoreConfigWithPageSize pageSize + | keySize < 8 && valueSize < 8 = Nothing + | otherwise = Just MemoryStoreConfig { + memoryStoreConfigPageSize = pageSize + , memoryStoreConfigMaxKeySize = keySize + , memoryStoreConfigMaxValueSize = valueSize } + where + keySize = calculateMaxKeySize pageSize (encodedPageSize zeroHeight) + valueSize = calculateMaxValueSize pageSize keySize (encodedPageSize zeroHeight) + +-- | Run the storage operations in the 'MemoryStoreT' monad, given a collection of +-- 'File's. +runMemoryStoreT :: MemoryStoreT fp m a -- ^ Action to run + -> MemoryStoreConfig -- ^ Configuration + -> MemoryFiles fp -- ^ Data + -> m a +runMemoryStoreT m config = runReaderT (runReaderT (fromMemoryStoreT m) config) + +-- | Construct a store with an empty database with name of type @hnd@. +newEmptyMemoryStore :: IO (MemoryFiles hnd) +newEmptyMemoryStore = newIORef M.empty + +-------------------------------------------------------------------------------- + +instance (Applicative m, Monad m, MonadIO m, MonadThrow m, + Ord fp, Show fp, Typeable fp) => + StoreM fp (MemoryStoreT fp m) + where + openHandle fp = + modify' $ M.insertWith (flip const) fp M.empty + + flushHandle _ = return () + + closeHandle _ = return () + + removeHandle fp = + modify' $ M.delete fp + + nodePageSize = return encodedPageSize + maxPageSize = asks memoryStoreConfigPageSize + maxKeySize = asks memoryStoreConfigMaxKeySize + maxValueSize = asks memoryStoreConfigMaxValueSize + + getNodePage hnd h key val nid = do + bs <- get >>= lookupPage hnd (nodeIdToPageId nid) + case viewHeight h of + UZero -> decodeM (leafNodePage h key val) bs >>= \case + LeafNodePage heightSrc n -> + justErrM WrongNodeTypeError $ castNode heightSrc h n + USucc _ -> decodeM (indexNodePage h key val) bs >>= \case + IndexNodePage heightSrc n -> + justErrM WrongNodeTypeError $ castNode heightSrc h n + + putNodePage hnd height nid node = + modify' $ M.update (Just . M.insert (nodeIdToPageId nid) pg) hnd + where + pg = case viewHeight height of + UZero -> toStrict . encode $ LeafNodePage height node + USucc _ -> toStrict . encode $ IndexNodePage height node + + getOverflow hnd val = do + bs <- get >>= lookupPage hnd 0 + decodeM (overflowPage val) bs >>= \case + OverflowPage v -> justErrM WrongOverflowValueError $ castValue v + + putOverflow hnd val = + modify' $ M.update (Just . M.insert 0 pg) hnd + where + pg = toStrict . encode $ OverflowPage val + + listOverflows _ = gets M.keys + +-------------------------------------------------------------------------------- + +instance (Applicative m, Monad m, MonadIO m, MonadThrow m) => + ConcurrentMetaStoreM (MemoryStoreT FilePath m) + where + putConcurrentMeta h meta = + modify' $ M.update (Just . M.insert 0 pg) h + where + pg = toStrict . encode $ ConcurrentMetaPage meta + + readConcurrentMeta hnd k v = do + Just bs <- gets (M.lookup hnd >=> M.lookup 0) + decodeM (concurrentMetaPage k v) bs >>= \case + ConcurrentMetaPage meta -> return . Just $! coerce meta + +-------------------------------------------------------------------------------- + +-- | Exception thrown when a file is accessed that doesn't exist. +newtype FileNotFoundError hnd = FileNotFoundError hnd deriving (Show, Typeable) + +instance (Typeable hnd, Show hnd) => Exception (FileNotFoundError hnd) where + +-- | Exception thrown when a page that is accessed doesn't exist. +data PageNotFoundError hnd = PageNotFoundError hnd PageId deriving (Show, Typeable) + +instance (Typeable hnd, Show hnd) => Exception (PageNotFoundError hnd) where + +-- | Exception thrown when a node cannot be cast to the right type. +-- +-- As used in 'getNodePage'. +data WrongNodeTypeError = WrongNodeTypeError deriving (Show, Typeable) + +instance Exception WrongNodeTypeError where + +-- | Exception thrown when a value from an overflow page cannot be cast. +-- +-- As used in 'getOverflow'. +data WrongOverflowValueError = WrongOverflowValueError deriving (Show, Typeable) + +instance Exception WrongOverflowValueError where + +-------------------------------------------------------------------------------- + diff --git a/src/Database/Haskey/Store/Page.hs b/src/Database/Haskey/Store/Page.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Store/Page.hs @@ -0,0 +1,250 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} +-- | This module contains structures and functions to decode and encode +-- pages. +module Database.Haskey.Store.Page where + +import Codec.Compression.LZ4 + +import Control.Applicative ((<$>)) +import Control.Monad.Catch + +import Data.Binary (Binary(..), Put, Get) +import Data.Binary.Get (runGetOrFail) +import Data.Binary.Put (runPut) +import Data.Bits ((.&.), (.|.)) +import Data.ByteString (ByteString) +import Data.ByteString.Lazy (fromStrict, toStrict) +import Data.Digest.XXHash.FFI (xxh64) +import Data.Maybe (fromMaybe) +import Data.Proxy +import Data.Typeable (Typeable) +import Data.Word (Word8, Word64) +import qualified Data.Binary as B +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy as BL + +import Numeric (showHex) + +import Data.BTree.Impure +import Data.BTree.Impure.Structures (putLeafNode, getLeafNode, putIndexNode, getIndexNode) +import Data.BTree.Primitives + +import Database.Haskey.Alloc.Concurrent + +-- | The type of a page. +data PageType = TypeEmpty + | TypeConcurrentMeta + | TypeOverflow + | TypeLeafNode + | TypeIndexNode + deriving (Eq, Show) + +data SPageType t where + STypeEmpty :: SPageType 'TypeEmpty + STypeConcurrentMeta :: SPageType 'TypeConcurrentMeta + STypeOverflow :: SPageType 'TypeOverflow + STypeLeafNode :: SPageType 'TypeLeafNode + STypeIndexNode :: SPageType 'TypeIndexNode + +instance Binary PageType where + put TypeEmpty = put (0x00 :: Word8) + put TypeConcurrentMeta = put (0x20 :: Word8) + put TypeOverflow = put (0x40 :: Word8) + put TypeLeafNode = put (0x60 :: Word8) + put TypeIndexNode = put (0x80 :: Word8) + get = (get :: Get Word8) >>= \case + 0x00 -> return TypeEmpty + 0x20 -> return TypeConcurrentMeta + 0x40 -> return TypeOverflow + 0x60 -> return TypeLeafNode + 0x80 -> return TypeIndexNode + t -> fail $ "unknown page type: " ++ showHex t "" + +-- | A decoded page, of a certain type @t@ of kind 'PageType'. +data Page (t :: PageType) where + EmptyPage :: Page 'TypeEmpty + ConcurrentMetaPage :: (Key k, Value v) + => ConcurrentMeta k v + -> Page 'TypeConcurrentMeta + OverflowPage :: (Value v) + => v + -> Page 'TypeOverflow + LeafNodePage :: (Key k, Value v) + => Height 'Z + -> Node 'Z k v + -> Page 'TypeLeafNode + IndexNodePage :: (Key k, Value v) + => Height ('S h) + -> Node ('S h) k v + -> Page 'TypeIndexNode + +-- | A decoder with its type. +data SGet t = SGet (SPageType t) (Get (Page t)) + +-- | Get the type of a 'Page'. +pageType :: SPageType t -> PageType +pageType STypeEmpty = TypeEmpty +pageType STypeConcurrentMeta = TypeConcurrentMeta +pageType STypeOverflow = TypeOverflow +pageType STypeLeafNode = TypeLeafNode +pageType STypeIndexNode = TypeIndexNode + +-- | Encode a page to a lazy byte string, but with the checksum set to zero. +encodeZeroChecksum :: Page t -> BL.ByteString +encodeZeroChecksum p = zero `BL.append` encodeNoChecksum p + where zero = "\NUL\NUL\NUL\NUL\NUL\NUL\NUL\NUL" + +-- | Encode a page to a lazy byte string, and prepend the calculated checksum. +encode :: Page t -> BL.ByteString +encode = prependChecksum . encodeNoChecksum + +-- | Prepend the xxHash 64-bit checksum of a bytestring to itself. +prependChecksum :: BL.ByteString -> BL.ByteString +prependChecksum bs = B.encode (xxh64 bs checksumSeed :: Word64) `BL.append` bs + +-- | Encode a page to a lazy byte string, without prepending the checksum. +encodeNoChecksum :: Page t -> BL.ByteString +encodeNoChecksum = runPut . putPage + -- let bs = runPut (putPage p) in fromMaybe bs (tryCompress bs) + where + _tryCompress bs = do + (t, body) <- BL.uncons bs + c <- compress (toStrict body) + if BS.length c < fromIntegral (BL.length bs) + then Just $ maskCompressed t `BL.cons` fromStrict c + else Nothing + + maskCompressed t = t .|. 0x01 + +-- | Size of a node, if it were to be encoded. +encodedPageSize :: (Key k, Value v) => Height h -> Node h k v -> PageSize +encodedPageSize h = case viewHeight h of + UZero -> fromIntegral . BL.length . encodeZeroChecksum . LeafNodePage h + USucc _ -> fromIntegral . BL.length . encodeZeroChecksum . IndexNodePage h + +-- | Decode a page, and verify the checksum. +decode :: SGet t -> ByteString -> Either String (Page t) +decode g@(SGet t _) bs = do + let (cksumBs, body) = BS.splitAt 8 bs + cksum <- if BS.length cksumBs < 8 + then Left $ "could not decode " ++ show (pageType t) ++ ": " + ++ "not enough checksum bytes" + else Right $ B.decode (fromStrict cksumBs) + let cksum' = xxh64 body checksumSeed + if cksum' /= cksum + then Left $ "could not decode " ++ show (pageType t) ++ ": " + ++ "expected checksum " ++ show cksum' ++ " but checksum " + ++ "field contains " ++ show cksum + else decodeNoChecksum g body + + +-- | Decode a page with a specific decoder, or return the error. +decodeNoChecksum :: SGet t -> ByteString -> Either String (Page t) +decodeNoChecksum (SGet t g) bs = case runGetOrFail g (fromStrict bs) of + Left err -> Left $ err' err + Right (_, _, v) -> Right v + where + err' (bs', offset, err) = + "could not decode " ++ show (pageType t) ++ ": " ++ err ++ + "at pos " ++ show offset ++ ", remaining bytes: " ++ show bs' ++ + ", full body: " ++ show bs + + _decompressed = fromMaybe (fromStrict bs) $ do + (tb, body) <- BS.uncons bs + if isCompressed tb + then do + c <- decompress body + Just $ unmaskCompressed tb `BL.cons` fromStrict c + else Nothing + + isCompressed b = b .&. 0x01 == 0x01 + unmaskCompressed b = b .&. 0xFE + + +-- | Monadic wrapper around 'decode' +decodeM :: MonadThrow m => SGet t -> ByteString -> m (Page t) +decodeM g bs = case decode g bs of + Left err -> throwM $ DecodeError err + Right v -> return v + +-- | The encoder of a 'Page'. +putPage :: Page t -> Put +putPage EmptyPage = put TypeEmpty +putPage (ConcurrentMetaPage m) = put TypeConcurrentMeta >> put m +putPage (OverflowPage v) = put TypeOverflow >> put v +putPage (LeafNodePage _ n) = put TypeLeafNode >> putLeafNode n +putPage (IndexNodePage h n) = put TypeIndexNode >> put h >> putIndexNode n + +-- | Decoder for an empty page. +emptyPage :: SGet 'TypeEmpty +emptyPage = SGet STypeEmpty $ get >>= \case + TypeEmpty -> return EmptyPage + x -> fail $ "unexpected " ++ show x ++ " while decoding TypeEmpty" + +-- | Decoder for a leaf node page. +leafNodePage :: (Key k, Value v) + => Height 'Z + -> Proxy k + -> Proxy v + -> SGet 'TypeLeafNode +leafNodePage h k v = SGet STypeLeafNode $ get >>= \case + TypeLeafNode -> LeafNodePage h <$> get' h k v + x -> fail $ "unexpected " ++ show x ++ " while decoding TypeLeafNode" + where + get' :: (Key k, Value v) + => Height 'Z -> Proxy k -> Proxy v -> Get (Node 'Z k v) + get' h' _ _ = getLeafNode h' + +-- | Decoder for a leaf node page. +indexNodePage :: (Key k, Value v) + => Height ('S n) + -> Proxy k + -> Proxy v + -> SGet 'TypeIndexNode +indexNodePage h k v = SGet STypeIndexNode $ get >>= \case + TypeIndexNode -> do + h' <- get + if fromHeight h == fromHeight h' + then IndexNodePage h <$> get' h k v + else fail $ "expected height " ++ show h ++ " but got " + ++ show h' ++ " while decoding TypeNode" + x -> fail $ "unexpected " ++ show x ++ " while decoding TypeIndexNode" + where + get' :: (Key k, Value v) + => Height ('S n) -> Proxy k -> Proxy v -> Get (Node ('S n) k v) + get' h' _ _ = getIndexNode h' + +-- | Decoder for an overflow page. +overflowPage :: (Value v) => Proxy v -> SGet 'TypeOverflow +overflowPage v = SGet STypeOverflow $ get >>= \case + TypeOverflow -> OverflowPage <$> get' v + x -> fail $ "unexpected " ++ show x ++ " while decoding TypeOverflow" + where + get' :: (Value v) => Proxy v -> Get v + get' _ = get + +concurrentMetaPage :: (Key k, Value v) + => Proxy k + -> Proxy v + -> SGet 'TypeConcurrentMeta +concurrentMetaPage k v = SGet STypeConcurrentMeta $ get >>= \ case + TypeConcurrentMeta -> ConcurrentMetaPage <$> get' k v + x -> fail $ "unexpected " ++ show x ++ " while decoding TypeConcurrentMeta" + where + get' :: (Key k, Value v) => Proxy k -> Proxy v -> Get (ConcurrentMeta k v) + get' _ _ = get + +-- | Exception thrown when decoding of a page fails. +newtype DecodeError = DecodeError String deriving (Show, Typeable) + +instance Exception DecodeError where + +checksumSeed :: Word64 +checksumSeed = 0 diff --git a/src/Database/Haskey/Utils/IO.hs b/src/Database/Haskey/Utils/IO.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Utils/IO.hs @@ -0,0 +1,38 @@ +module Database.Haskey.Utils.IO where + +import Data.ByteString (ByteString, packCStringLen) +import Data.ByteString.Unsafe (unsafeUseAsCString) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy as BL + +import Foreign (allocaBytes, castPtr, plusPtr) + +import qualified FileIO as IO + +import System.IO.Error (mkIOError, eofErrorType, ioeSetErrorString) + +readByteString :: IO.FHandle -> Int -> IO ByteString +readByteString fd n = allocaBytes n $ \buf -> do + go 0 buf + packCStringLen (castPtr buf, fromIntegral n) + where + go c buf + | c == n = return () + | otherwise = do + rc <- IO.read fd buf (fromIntegral (n - c)) + case rc of + 0 -> ioError (ioeSetErrorString (mkIOError eofErrorType "getByteString" Nothing Nothing) "EOF") + n' -> go (c + fromIntegral n') (buf `plusPtr` fromIntegral n') + +writeByteString :: IO.FHandle -> ByteString -> IO () +writeByteString fd bs = unsafeUseAsCString bs $ \buf -> go 0 buf + where + n = BS.length bs + go c buf + | c == n = return () + | otherwise = do + n' <- IO.write fd (castPtr buf) (fromIntegral (n - c)) + go (c + fromIntegral n') (buf `plusPtr` fromIntegral n') + +writeLazyByteString :: IO.FHandle -> BL.ByteString -> IO () +writeLazyByteString fh bs = mapM_ (writeByteString fh) (BL.toChunks bs) diff --git a/src/Database/Haskey/Utils/Monad.hs b/src/Database/Haskey/Utils/Monad.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Utils/Monad.hs @@ -0,0 +1,4 @@ +module Database.Haskey.Utils.Monad where + +ifM :: Monad m => m Bool -> m a -> m a -> m a +ifM f y n = do f' <- f; if f' then y else n diff --git a/src/Database/Haskey/Utils/Monad/Catch.hs b/src/Database/Haskey/Utils/Monad/Catch.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Utils/Monad/Catch.hs @@ -0,0 +1,7 @@ +module Database.Haskey.Utils.Monad.Catch where + +import Control.Monad.Catch + +justErrM :: (MonadThrow m, Exception e) => e -> Maybe a -> m a +justErrM _ (Just v) = return v +justErrM e Nothing = throwM e diff --git a/src/Database/Haskey/Utils/RLock.hs b/src/Database/Haskey/Utils/RLock.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Utils/RLock.hs @@ -0,0 +1,59 @@ +{-# LANGUAGE DeriveDataTypeable #-} +-- | Simple implementations of reentrant locks using 'MVar' +module Database.Haskey.Utils.RLock where + +import Control.Concurrent (ThreadId, myThreadId) +import Control.Concurrent.MVar +import Control.Exception (Exception, throwIO) +import Control.Monad (unless, when) +import Control.Monad.Catch (MonadMask, bracket_) +import Control.Monad.IO.Class (MonadIO, liftIO) + +import Data.Typeable (Typeable) + +-- | A reentrant lock. +type RLock = (MVar (Maybe (ThreadId, Integer)), MVar ()) + +-- | Create a new reentrant lock. +newRLock :: IO RLock +newRLock = do { a <- newMVar Nothing; b <- newMVar (); return (a, b) } + +-- | Acquire a reentrant lock, blocks. +acquireRLock :: RLock -> IO () +acquireRLock (r, l) = do + myId <- myThreadId + ok <- modifyMVar r $ \state -> case state of + Nothing -> return (state, False) + Just (tId, x) -> if tId == myId + then return (Just (myId, x + 1), True) + else return (state, False) + + unless ok $ do + () <- takeMVar l + modifyMVar_ r $ const (return $ Just (myId, 1)) + +-- | Release a reentrant lock. +releaseRLock :: RLock -> IO () +releaseRLock (r, l) = do + myId <- myThreadId + done <- modifyMVar r $ \state -> case state of + Nothing -> throwIO $ RLockError "the lock has no inhabitant" + Just (_, 0) -> throwIO $ RLockError "the lock is already released" + Just (tId, n) -> if tId == myId + then if n == 1 + then return (Nothing, True) + else return (Just (myId, n-1), False) + else throwIO $ RLockError "lock not held by releaser" + + when done $ + putMVar l () + +-- | Execute an action with the lock, bracketed, exception-safe +withRLock :: (MonadIO m, MonadMask m) => RLock -> m a -> m a +withRLock l = bracket_ (liftIO $ acquireRLock l) + (liftIO $ releaseRLock l) + +-- | Exception raised when 'RLock' is improperly used. +newtype RLockError = RLockError String deriving (Show, Typeable) + +instance Exception RLockError where diff --git a/src/Database/Haskey/Utils/STM/Map.hs b/src/Database/Haskey/Utils/STM/Map.hs new file mode 100644 --- /dev/null +++ b/src/Database/Haskey/Utils/STM/Map.hs @@ -0,0 +1,20 @@ +{-# LANGUAGE ConstraintKinds #-} +module Database.Haskey.Utils.STM.Map where + +import Control.Applicative ((<$>)) +import Control.Concurrent.STM (STM) + +import qualified Focus as F +import qualified ListT as L + +import STMContainers.Map (Map, Key) +import qualified STMContainers.Map as M + +lookupMin :: Map k v -> STM (Maybe (k, v)) +lookupMin = L.head . M.stream + +lookupMinKey :: Map k v -> STM (Maybe k) +lookupMinKey = ((fst <$>) <$>) . lookupMin + +alter :: Key k => k -> (Maybe v -> Maybe v) -> Map k v -> STM () +alter k f = M.focus (F.alterM (return . f)) k diff --git a/tests/Integration.hs b/tests/Integration.hs new file mode 100644 --- /dev/null +++ b/tests/Integration.hs @@ -0,0 +1,15 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE RankNTypes #-} +module Main (main) where + +import Test.Framework (Test, defaultMain) + +import qualified Integration.WriteOpenRead.Concurrent + +tests :: [Test] +tests = + [ Integration.WriteOpenRead.Concurrent.tests + ] + +main :: IO () +main = defaultMain tests diff --git a/tests/Integration/WriteOpenRead/Concurrent.hs b/tests/Integration/WriteOpenRead/Concurrent.hs new file mode 100644 --- /dev/null +++ b/tests/Integration/WriteOpenRead/Concurrent.hs @@ -0,0 +1,213 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE RankNTypes #-} +module Integration.WriteOpenRead.Concurrent where + +import Test.Framework (Test, testGroup) +import Test.Framework.Providers.QuickCheck2 (testProperty) +import Test.QuickCheck +import Test.QuickCheck.Monadic +import Test.QuickCheck.Random + +import Control.Applicative ((<$>)) +import Control.Monad +import Control.Monad.Catch (MonadMask, Exception, throwM, catch) +import Control.Monad.IO.Class +import Control.Monad.Trans (lift) +import Control.Monad.Trans.Maybe + +import Data.Binary (Binary(..)) +import Data.Foldable (foldlM) +import Data.Map (Map) +import Data.Maybe (isJust, fromJust, fromMaybe) +import Data.Typeable (Typeable) +import Data.Word (Word8) +import qualified Data.Map as M + +import GHC.Generics (Generic) + +import System.Directory (removeDirectoryRecursive, + getTemporaryDirectory, doesDirectoryExist, + writable, getPermissions) +import System.IO.Temp (createTempDirectory) + +import Data.BTree.Alloc.Class +import Data.BTree.Impure +import Data.BTree.Primitives +import qualified Data.BTree.Impure as Tree + +import Database.Haskey.Alloc.Concurrent +import Database.Haskey.Store.File +import Database.Haskey.Store.InMemory + +import Integration.WriteOpenRead.Transactions + +tests :: Test +tests = testGroup "WriteOpenRead.Concurrent" + [ testProperty "memory backend" (monadicIO prop_memory_backend) + , testProperty "file backend" (monadicIO prop_file_backend) + ] + +case_bad_seed :: IO () +case_bad_seed = do + putStrLn "Testing bad case..." + quickCheckWith args (monadicIO prop_memory_backend) + putStrLn " done" + where + -- This seed results in out of memory!! + seed = 1576280407925194075 + gen = (mkQCGen seed, seed) + args = stdArgs { replay = Just gen } + +prop_memory_backend :: PropertyM IO () +prop_memory_backend = forAllM (genTestSequence False) $ \(TestSequence txs) -> do + files <- run newEmptyMemoryStore + db <- run $ create files + _ <- run $ foldlM (writeReadTest db files) + M.empty + txs + return () + where + + writeReadTest :: ConcurrentDb Integer TestValue + -> MemoryFiles String + -> Map Integer TestValue + -> TestTransaction Integer TestValue + -> IO (Map Integer TestValue) + writeReadTest db files m tx = do + openAndWrite db files tx + read' <- openAndRead db files + let expected = fromMaybe m $ testTransactionResult m tx + if read' == M.toList expected + then return expected + else error $ "error:" + ++ "\n after: " ++ show tx + ++ "\n expectd: " ++ show (M.toList expected) + ++ "\n got: " ++ show read' + + create :: MemoryFiles String -> IO (ConcurrentDb Integer TestValue) + create = runMemoryStoreT (createConcurrentDb hnds) config + where + hnds = concurrentHandles "" + + openAndRead db = runMemoryStoreT (readAll db) config + + openAndWrite db files tx = + runMemoryStoreT (writeTransaction tx db) config files + + config = fromJust $ memoryStoreConfigWithPageSize 256 + +-------------------------------------------------------------------------------- + +prop_file_backend :: PropertyM IO () +prop_file_backend = forAllM (genTestSequence True) $ \(TestSequence txs) -> do + exists <- run $ doesDirectoryExist "/var/run/shm" + w <- if exists then run $ writable <$> getPermissions "/var/run/shm" + else return False + tmpDir <- if w then return "/var/run/shm" + else run getTemporaryDirectory + fp <- run $ createTempDirectory tmpDir "db.haskey" + + let hnds = concurrentHandles fp + files <- run newFileStore + db <- run $ create files hnds + result <- run . runMaybeT $ foldM (writeReadTest db files) + M.empty + txs + + _ <- run $ runFileStoreT (closeConcurrentHandles hnds) config files + + run $ removeDirectoryRecursive fp + + assert $ isJust result + where + writeReadTest :: ConcurrentDb Integer TestValue + -> Files FilePath + -> Map Integer TestValue + -> TestTransaction Integer TestValue + -> MaybeT IO (Map Integer TestValue) + writeReadTest db files m tx = do + _ <- lift $ void (openAndWrite db files tx) `catch` + \TestException -> return () + read' <- lift $ openAndRead db files + let expected = fromMaybe m $ testTransactionResult m tx + if read' == M.toList expected + then return expected + else error $ "error:" + ++ "\n after: " ++ show tx + ++ "\n expectd: " ++ show (M.toList expected) + ++ "\n got: " ++ show read' + + create :: Files FilePath + -> ConcurrentHandles + -> IO (ConcurrentDb Integer TestValue) + create files hnds = runFileStoreT (createConcurrentDb hnds) config files + + openAndRead :: ConcurrentDb Integer TestValue + -> Files FilePath + -> IO [(Integer, TestValue)] + openAndRead db = runFileStoreT (readAll db) config + + openAndWrite :: ConcurrentDb Integer TestValue + -> Files FilePath + -> TestTransaction Integer TestValue + -> IO () + openAndWrite db files tx = + runFileStoreT (void $ writeTransaction tx db) config files + + config = fromJust $ fileStoreConfigWithPageSize 256 + +-------------------------------------------------------------------------------- + +writeTransaction :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Key k, Value v) + => TestTransaction k v + -> ConcurrentDb k v + -> m () +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 ThrowException = const (throwM TestException) + + transaction = transact_ $ + foldl (>=>) return (map writeAction actions) + >=> commitOrAbort + + commitOrAbort :: (AllocM n, MonadMask n) => Tree key val -> n (Transaction key val ()) + commitOrAbort + | TxAbort <- txType = const abort_ + | TxCommit <- txType = commit_ + +readAll :: (MonadIO m, MonadMask m, ConcurrentMetaStoreM m, Key k, Value v) + => ConcurrentDb k v + -> m [(k, v)] +readAll = transactReadOnly Tree.toList + +-------------------------------------------------------------------------------- + +-- | Value used for testing. +-- +-- This value will overflow 20% of the time. +newtype TestValue = TestValue (Either Integer [Word8]) + deriving (Eq, Generic, Show, Typeable) + +instance Binary TestValue where +instance Value TestValue where + +instance Arbitrary TestValue where + arbitrary = + TestValue <$> frequency [(80, Left <$> small), (20, Right <$> big)] + where + small = arbitrary + big = arbitrary + +-- | Exception used for testing +data TestException = TestException deriving (Show, Typeable) + +instance Exception TestException where + +-------------------------------------------------------------------------------- diff --git a/tests/Integration/WriteOpenRead/Transactions.hs b/tests/Integration/WriteOpenRead/Transactions.hs new file mode 100644 --- /dev/null +++ b/tests/Integration/WriteOpenRead/Transactions.hs @@ -0,0 +1,130 @@ +{-# LANGUAGE RecordWildCards #-} +module Integration.WriteOpenRead.Transactions where + +import Test.QuickCheck + +import Control.Applicative ((<$>), (<*>), pure) +import Control.Monad.State + +import Data.Foldable (foldlM) +import Data.List (inits) +import Data.Map (Map) +import Data.Maybe (fromMaybe) +import qualified Data.Map as M + +-------------------------------------------------------------------------------- + +newtype TestSequence k v = TestSequence [TestTransaction k v] + deriving (Show) + +data TransactionSetup = TransactionSetup { sequenceInsertFrequency :: !Int + , sequenceReplaceFrequency :: !Int + , sequenceDeleteFrequency :: !Int + , sequenceExceptionFrequency :: !Int } + deriving (Show) + +deleteHeavySetup :: TransactionSetup +deleteHeavySetup = TransactionSetup { sequenceInsertFrequency = 35 + , sequenceReplaceFrequency = 20 + , sequenceDeleteFrequency = 45 + , sequenceExceptionFrequency = 0 } + +insertHeavySetup :: TransactionSetup +insertHeavySetup = TransactionSetup { sequenceInsertFrequency = 12 + , sequenceReplaceFrequency = 4 + , sequenceDeleteFrequency = 4 + , sequenceExceptionFrequency = 0} + +withExceptionSetup :: TransactionSetup +withExceptionSetup = insertHeavySetup { sequenceExceptionFrequency = 5 } + +genTransactionSetup :: Bool -> Gen TransactionSetup +genTransactionSetup withExc = + frequency [(45, return deleteHeavySetup), + (45, return insertHeavySetup), + (f, return withExceptionSetup)] + where + f | withExc = 10 + | otherwise = 0 + +data TxType = TxAbort | TxCommit + deriving (Show) + +genTxType :: Gen TxType +genTxType = elements [TxAbort, TxCommit] + +data TestTransaction k v = TestTransaction TxType [TestAction k v] + deriving (Show) + +testTransactionResult :: Ord k => Map k v -> TestTransaction k v -> Maybe (Map k v) +testTransactionResult m (TestTransaction TxAbort _) = Just m +testTransactionResult m (TestTransaction TxCommit actions) + = foldlM (flip doAction) m actions + +data TestAction k v = Insert k v + | Replace k v + | Delete k + | ThrowException + deriving (Show) + +doAction :: Ord k => TestAction k v -> Map k v -> Maybe (Map k v) +doAction action m + | Insert k v <- action = Just $ M.insert k v m + | Replace k v <- action = Just $ M.insert k v m + | Delete k <- action = Just $ M.delete k m + | ThrowException <- action = Nothing + +genTestTransaction :: (Ord k, Arbitrary k, Arbitrary v) => Map k v -> TransactionSetup -> Gen (TestTransaction k v, Maybe (Map k v)) +genTestTransaction db TransactionSetup{..} = sized $ \n -> do + k <- choose (0, n) + (m, actions) <- execStateT (replicateM k next) (Just db, []) + tx <- TestTransaction <$> genTxType <*> pure (reverse actions) + return (tx, m) + where + genAction :: (Ord k, Arbitrary k, Arbitrary v) + => Maybe (Map k v) + -> Gen (TestAction k v) + genAction Nothing = genException + genAction (Just m) + | M.null m = genInsert + | otherwise = frequency [(sequenceInsertFrequency, genInsert ), + (sequenceReplaceFrequency, genReplace m), + (sequenceDeleteFrequency, genDelete m ), + (sequenceExceptionFrequency, genException)] + + genInsert :: (Arbitrary k, Arbitrary v) => Gen (TestAction k v) + genInsert = Insert <$> arbitrary <*> arbitrary + genReplace m = Replace <$> elements (M.keys m) <*> arbitrary + genDelete m = Delete <$> elements (M.keys m) + genException = return ThrowException + + next :: (Ord k, Arbitrary k, Arbitrary v) + => StateT (Maybe (Map k v), [TestAction k v]) Gen () + next = do + (m, actions) <- get + action <- lift $ genAction m + put (m >>= doAction action, action:actions) + +shrinkTestTransaction :: (Ord k, Arbitrary k, Arbitrary v) + => TestTransaction k v + -> [TestTransaction k v] +shrinkTestTransaction (TestTransaction _ []) = [] +shrinkTestTransaction (TestTransaction t actions) = map (TestTransaction t) (init (inits actions)) + +genTestSequence :: (Ord k, Arbitrary k, Arbitrary v) => Bool -> Gen (TestSequence k v) +genTestSequence withExc = sized $ \n -> do + k <- choose (0, n) + (_, txs) <- execStateT (replicateM k next) (M.empty, []) + return $ TestSequence (reverse txs) + where + next :: (Ord k, Arbitrary k, Arbitrary v) + => StateT (Map k v, [TestTransaction k v]) Gen () + next = do + (m, txs) <- get + (tx, m') <- lift $ genTransactionSetup withExc >>= genTestTransaction m + put (fromMaybe m m', tx:txs) + +shrinkTestSequence :: (Ord k, Arbitrary k, Arbitrary v) + => TestSequence k v + -> [TestSequence k v] +shrinkTestSequence (TestSequence txs) = map TestSequence (shrinkList shrinkTestTransaction txs) diff --git a/tests/Properties.hs b/tests/Properties.hs new file mode 100644 --- /dev/null +++ b/tests/Properties.hs @@ -0,0 +1,17 @@ +module Main (main) where + +import Test.Framework (Test, defaultMain) + +import qualified Properties.Store.Page + +-------------------------------------------------------------------------------- + +tests :: [Test] +tests = + [ Properties.Store.Page.tests + ] + +main :: IO () +main = defaultMain tests + +-------------------------------------------------------------------------------- diff --git a/tests/Properties/Store/Page.hs b/tests/Properties/Store/Page.hs new file mode 100644 --- /dev/null +++ b/tests/Properties/Store/Page.hs @@ -0,0 +1,132 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# OPTIONS_GHC -fno-warn-orphans #-} +module Properties.Store.Page where + +import Test.Framework (Test, testGroup) +import Test.Framework.Providers.HUnit (testCase) +import Test.Framework.Providers.QuickCheck2 (testProperty) +import Test.HUnit hiding (Test, Node) +import Test.QuickCheck + +import Control.Applicative ((<$>)) + +import Data.Int +import Data.List (nub) +import Data.Monoid ((<>)) +import Data.Proxy +import qualified Data.Binary as B +import qualified Data.ByteString.Lazy as BL +import qualified Data.Map as M +import qualified Data.Vector as V + +import Data.BTree.Impure.Structures +import Data.BTree.Primitives + +import Database.Haskey.Store.Page + +-------------------------------------------------------------------------------- + +tests :: Test +tests = testGroup "Store.Page" + [ testProperty "binary pageType" prop_binary_pageType + , testProperty "binary emptyPage" prop_binary_emptyPage + , testProperty "binary nodePage leaf" prop_binary_leafNodePage + , testProperty "binary nodePage idx" prop_binary_indexNodePage + , testCase "zero checksum length" case_zero_checksum_length + ] + +prop_binary_pageType :: Property +prop_binary_pageType = forAll types $ \t -> + let bs = B.encode t in BL.length bs == 1 && B.decode bs == t + where + types = elements [TypeEmpty, + TypeConcurrentMeta, + TypeOverflow, + TypeLeafNode, + TypeIndexNode] + +prop_binary_emptyPage :: Bool +prop_binary_emptyPage = case decode' emptyPage (encode EmptyPage) of + Right EmptyPage -> True + Left _ -> False + +prop_binary_leafNodePage :: Property +prop_binary_leafNodePage = forAll genLeafNode $ \leaf -> + case decode' (leafNodePage zeroHeight key val) + (encode (LeafNodePage zeroHeight leaf)) of + Right (LeafNodePage h n) -> maybe False (== leaf) $ castNode h zeroHeight n + Left _ -> False + where + key = Proxy :: Proxy Int64 + val = Proxy :: Proxy Bool + +prop_binary_indexNodePage :: Property +prop_binary_indexNodePage = forAll genIndexNode $ \(srcHgt, idx) -> + case decode' (indexNodePage srcHgt key val) + (encode (IndexNodePage srcHgt idx)) of + Right (IndexNodePage h n) -> maybe False (== idx) $ castNode h srcHgt n + Left _ -> False + where + key = Proxy :: Proxy Int64 + val = Proxy :: Proxy Bool + +case_zero_checksum_length :: Assertion +case_zero_checksum_length = do + assertEqual "zero checksum should prepend 8 bytes" 8 $ + BL.length withZero' - BL.length without' + assertEqual "zero checksum length should equal regular checksum lenth" + (BL.length withZero') + (BL.length with') + where + withZero' = encodeZeroChecksum pg + without' = encodeNoChecksum pg + with' = encode pg + + pg = LeafNodePage zeroHeight (Leaf M.empty :: Node 'Z Int64 Int64) + +decode' :: SGet t -> BL.ByteString -> Either String (Page t) +decode' x = decode x . BL.toStrict + +-------------------------------------------------------------------------------- + +genIndexNode :: Gen (Height ('S h), Node ('S h) Int64 Bool) +genIndexNode = do + h <- genNonZeroHeight + n <- Idx <$> arbitrary + return (h, n) + +genLeafNode :: Gen (Node 'Z Int64 Bool) +genLeafNode = Leaf <$> arbitrary + +instance Arbitrary v => Arbitrary (LeafValue v) where + arbitrary = oneof [RawValue <$> arbitrary, OverflowValue <$> arbitrary] + +instance Arbitrary TxId where + arbitrary = TxId <$> arbitrary + +deriving instance Arbitrary (Height h) + +genNonZeroHeight :: Gen (Height h) +genNonZeroHeight = suchThat arbitrary $ \h -> case viewHeight h of + UZero -> False + USucc _ -> True + +instance (Key k, Arbitrary k, Arbitrary v) => Arbitrary (Index k v) where + arbitrary = do + keys <- V.fromList . nub <$> orderedList + vals <- V.fromList <$> vector (V.length keys + 1) + return (Index keys vals) + shrink (Index keys vals) = + [ Index newKeys newVals + | k <- [0..V.length keys - 1] + , let (preKeys,sufKeys) = V.splitAt k keys + newKeys = preKeys <> V.drop 1 sufKeys + (preVals,sufVals) = V.splitAt k vals + newVals = preVals <> V.drop 1 sufVals + ] + +deriving instance Arbitrary (NodeId height key val) diff --git a/tests/Properties/Utils.hs b/tests/Properties/Utils.hs new file mode 100644 --- /dev/null +++ b/tests/Properties/Utils.hs @@ -0,0 +1,6 @@ +module Properties.Utils where + +import qualified Data.Binary as B + +testBinary :: (Eq a, B.Binary a) => a -> Bool +testBinary x = B.decode (B.encode x) == x