diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -1,5 +1,24 @@ # Changelog +- 0.2.0 (2026-11-04) + * Breaking: 'openDbi' now takes an 'Env' rather than a 'Txn', and + opens the database in a transaction of its own. The handle is + valid until the 'Env' is closed. + * Fixes a crash when a write transaction's commit fails, and a + double abort when an async exception arrives just after a commit + or abort. + * Fixes database names being passed to LMDB without a NUL + terminator; names containing a NUL are now rejected. + * Transactions may now be used from any Haskell thread, under either + RTS: environments use MDB_NOTLS, so read transactions may nest and + run concurrently, and write transactions are serialized and run on + a single OS thread. + * Fixes the MDB_val layout on 32-bit targets. + * Adds 'sync'. 'open' rejects a negative 'envMapSize' or an + out-of-range 'envMaxDbs'. + * Key-value and cursor operations use unsafe foreign calls, for + speed. + - 0.1.0 (2026-05-24) * Initial release, consisting of support for LMDB environments, transactions, basic row operations (get, put, del), and range scans. diff --git a/bench/Main.hs b/bench/Main.hs --- a/bench/Main.hs +++ b/bench/Main.hs @@ -3,8 +3,10 @@ module Main where +import Control.DeepSeq (force) +import Control.Exception (evaluate) import Control.Monad (forM_) -import Criterion.Main +import Criterion.Main hiding (env) import qualified Data.ByteString as BS import qualified Data.ByteString.Builder as BB import qualified Data.ByteString.Lazy as BSL @@ -28,22 +30,26 @@ { L.envMapSize = 128 * 1024 * 1024 , L.envMaxDbs = 4 } + -- build keys and values outside the timed region + kv1 <- evaluate (force (kvs 1000)) + kv10 <- evaluate (force (kvs 10000)) + ks1 <- evaluate (force (scattered 1000)) + ks10 <- evaluate (force (scattered 10000)) L.withEnv dir flags $ \env -> do - -- pre-populate - L.withWriteTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing True - forM_ (kvs 10000) $ \(k, v) -> L.put txn dbi k v + dbi <- L.openDbi env Nothing True + -- pre-populate; the put benchmarks then overwrite these keys + insert env dbi kv10 defaultMain [ bgroup "put" - [ bench "1k inserts (one txn)" $ whnfIO (insert env 1000) - , bench "10k inserts (one txn)" $ whnfIO (insert env 10000) + [ bench "1k overwrites (one txn)" $ whnfIO (insert env dbi kv1) + , bench "10k overwrites (one txn)" $ whnfIO (insert env dbi kv10) ] , bgroup "get" - [ bench "1k random hits" $ whnfIO (readN env 1000) - , bench "10k random hits" $ whnfIO (readN env 10000) + [ bench "1k random hits" $ whnfIO (readN env dbi ks1) + , bench "10k random hits" $ whnfIO (readN env dbi ks10) ] , bgroup "cursor" - [ bench "scan 10k entries" $ whnfIO (scanAll env) + [ bench "scan 10k entries" $ whnfIO (scanAll env dbi) ] ] removeDirectoryRecursive dir @@ -53,6 +59,11 @@ kvs :: Int -> [(BS.ByteString, BS.ByteString)] kvs n = [ (mkKey i, mkVal i) | i <- [0 .. n - 1] ] +-- every key in [0, n), in a scattered order: 7919 is prime, so +-- coprime with n, and multiplying by it mod n permutes [0, n) +scattered :: Int -> [BS.ByteString] +scattered n = [ mkKey ((i * 7919) `mod` n) | i <- [0 .. n - 1] ] + mkKey :: Int -> BS.ByteString mkKey i = BSL.toStrict (BB.toLazyByteString (BB.string7 "k" <> BB.intDec i)) @@ -61,24 +72,20 @@ -- workloads ------------------------------------------------------------------ -insert :: L.Env -> Int -> IO () -insert env n = L.withWriteTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False - forM_ (kvs n) $ \(k, v) -> L.put txn dbi k v +insert :: L.Env -> L.Dbi -> [(BS.ByteString, BS.ByteString)] -> IO () +insert env dbi kv = L.withWriteTxn env $ \txn -> + forM_ kv $ \(k, v) -> L.put txn dbi k v -readN :: L.Env -> Int -> IO () -readN env n = L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False - forM_ [0 .. n - 1] $ \i -> do - _ <- L.get txn dbi (mkKey i) +readN :: L.Env -> L.Dbi -> [BS.ByteString] -> IO () +readN env dbi ks = L.withReadTxn env $ \txn -> + forM_ ks $ \k -> do + _ <- L.get txn dbi k pure () -scanAll :: L.Env -> IO Int -scanAll env = L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False +scanAll :: L.Env -> L.Dbi -> IO Int +scanAll env dbi = L.withReadTxn env $ \txn -> L.withCursor txn dbi $ \cur -> do let go !acc Nothing = pure acc go !acc (Just _) = L.cursorNext cur >>= go (acc + 1) mfirst <- L.cursorFirst cur go (0 :: Int) mfirst - diff --git a/bench/Weight.hs b/bench/Weight.hs --- a/bench/Weight.hs +++ b/bench/Weight.hs @@ -3,6 +3,8 @@ module Main where +import Control.DeepSeq (force) +import Control.Exception (evaluate) import Control.Monad (forM_) import qualified Data.ByteString as BS import qualified Data.ByteString.Builder as BB @@ -28,14 +30,17 @@ { L.envMapSize = 64 * 1024 * 1024 , L.envMaxDbs = 4 } + -- build keys and values outside the weighed region + kv1 <- evaluate (force (kvs 1000)) + ks1 <- evaluate (force (map fst kv1)) L.withEnv dir envFlags $ \env -> do - L.withWriteTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing True - forM_ (kvs 1000) $ \(k, v) -> L.put txn dbi k v + dbi <- L.openDbi env Nothing True + -- pre-populate; the put case then overwrites these keys + insert env dbi kv1 mainWith $ do - io "1k put (one txn)" (insert env) 1000 - io "1k get (one txn)" (readN env) 1000 - io "scan 1k entries" scanAll env + io "1k overwrites (one txn)" (insert env dbi) kv1 + io "1k gets (one txn)" (readN env dbi) ks1 + io "scan 1k entries" (scanAll env) dbi existsAfter <- doesDirectoryExist dir if existsAfter then removeDirectoryRecursive dir else pure () @@ -48,21 +53,18 @@ mkVal :: Int -> BS.ByteString mkVal i = BSL.toStrict (BB.toLazyByteString (BB.string7 "v" <> BB.intDec i)) -insert :: L.Env -> Int -> IO () -insert env n = L.withWriteTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False - forM_ (kvs n) $ \(k, v) -> L.put txn dbi k v +insert :: L.Env -> L.Dbi -> [(BS.ByteString, BS.ByteString)] -> IO () +insert env dbi kv = L.withWriteTxn env $ \txn -> + forM_ kv $ \(k, v) -> L.put txn dbi k v -readN :: L.Env -> Int -> IO () -readN env n = L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False - forM_ [0 .. n - 1] $ \i -> do - _ <- L.get txn dbi (mkKey i) +readN :: L.Env -> L.Dbi -> [BS.ByteString] -> IO () +readN env dbi ks = L.withReadTxn env $ \txn -> + forM_ ks $ \k -> do + _ <- L.get txn dbi k pure () -scanAll :: L.Env -> IO Int -scanAll env = L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False +scanAll :: L.Env -> L.Dbi -> IO Int +scanAll env dbi = L.withReadTxn env $ \txn -> L.withCursor txn dbi $ \cur -> do mfirst <- L.cursorFirst cur let go !acc Nothing = pure acc diff --git a/lib/Database/LMDB.hs b/lib/Database/LMDB.hs --- a/lib/Database/LMDB.hs +++ b/lib/Database/LMDB.hs @@ -21,9 +21,25 @@ -- level via a phantom parameter on 'Txn', so using a write primitive -- (e.g. 'put') on a read transaction is a compile-time error. -- --- Most operations throw 'LMDBException' on error. 'get' and 'del' are --- the exceptions: they map a missing key onto 'Nothing' and 'False' --- respectively, since absence is normal control flow. +-- Most operations throw 'LMDBException' on error. Lookups ('get' and +-- the cursor functions) and 'del' are the exceptions: they report a +-- missing key as 'Nothing' or 'False', since absence is normal control +-- flow. +-- +-- Transactions may be used from any thread, under either RTS; see +-- 'withReadTxn' and 'withWriteTxn' for the details. +-- +-- Lookups, writes and cursor moves are unsafe foreign calls, for +-- speed. On a database much larger than memory, a page fault in one +-- stalls garbage collection under the threaded RTS until the read +-- completes. +-- +-- Handles are not checked for validity. A 'Txn' or 'Cursor' is valid +-- only within the bracket that supplied it, and a 'Dbi' only until its +-- 'Env' is closed; an 'Env' must be closed at most once, after its +-- transactions have ended. Using a handle outside those bounds is +-- undefined behaviour. As LMDB requires, a process must not have the +-- same environment open twice at once. module Database.LMDB ( -- * Environments @@ -33,6 +49,7 @@ , open , close , withEnv + , sync -- * Transactions , Txn @@ -63,22 +80,27 @@ , LMDBException(..) ) where +import Control.Concurrent (rtsSupportsBoundThreads, runInBoundThread) +import Control.Concurrent.MVar (MVar, newMVar, withMVar) import Control.Exception - ( Exception, bracket, bracketOnError, mask, onException, throwIO) + (Exception, bracket, mask, onException, throwIO) import qualified Data.ByteString as BS import qualified Data.ByteString.Unsafe as BU -import Foreign.C.String (peekCString, withCString) -import Foreign.C.Types (CInt, CSize, CUInt) +import Foreign.C.Error (Errno(..), eINVAL) +import Foreign.C.String (CString, peekCString, withCString) +import Foreign.C.Types (CChar, CInt, CSize, CUInt) import Foreign.Marshal.Alloc (alloca, allocaBytes) import Foreign.Ptr (Ptr, nullPtr, plusPtr) -import Foreign.Storable (peek, peekByteOff, pokeByteOff) +import Foreign.Storable (peek, peekByteOff, pokeByteOff, sizeOf) import Database.LMDB.Internal -- handles -------------------------------------------------------------------- -- | An LMDB environment. Roughly, an open database file (or, with -- 'envNoSubdir' off, a directory containing one). -newtype Env = Env (Ptr MDB_env) +data Env = Env + {-# UNPACK #-} !(Ptr MDB_env) + {-# UNPACK #-} !(MVar ()) -- held for the duration of a write txn -- | Phantom tag for read-only transactions. data RO @@ -88,13 +110,15 @@ -- | A transaction. The phantom @s@ distinguishes 'RO' from 'RW' so -- that write operations refuse to type-check against read-only --- transactions. +-- transactions. Valid only within the 'withReadTxn' or +-- 'withWriteTxn' call that supplied it. newtype Txn s = Txn (Ptr MDB_txn) --- | A database identifier within an 'Env'. +-- | A database handle, valid until its 'Env' is closed. newtype Dbi = Dbi MDB_dbi --- | A cursor for range-scanning a 'Dbi'. +-- | A cursor for range-scanning a 'Dbi'. Valid only within the +-- 'withCursor' call that supplied it. newtype Cursor s = Cursor (Ptr MDB_cursor) -- environment flags ---------------------------------------------------------- @@ -113,9 +137,10 @@ , envReadOnly :: !Bool -- ^ open the environment read-only. , envNoSync :: !Bool - -- ^ skip @fsync@ at commit time. Faster, but a crash may lose - -- the last transaction. The database remains internally - -- consistent. + -- ^ skip @fsync@ at commit time. Faster, but an OS crash or power + -- loss may undo the most recent commits, and corrupt the + -- database unless the filesystem preserves write order. A + -- process crash loses nothing. Use 'sync' to flush explicitly. } deriving Show -- | Default 'EnvFlags': 10 MiB map, one database, directory layout, @@ -132,9 +157,12 @@ , envNoSync = False } +-- MDB_NOTLS is always set: GHC multiplexes Haskell threads over OS +-- threads, so reader slots must belong to txns rather than threads. envFlagBits :: EnvFlags -> CUInt envFlagBits EnvFlags { envNoSubdir = nsd, envReadOnly = ro, envNoSync = ns } = - (if nsd then _MDB_NOSUBDIR else 0) + _MDB_NOTLS + + (if nsd then _MDB_NOSUBDIR else 0) + (if ro then _MDB_RDONLY else 0) + (if ns then _MDB_NOSYNC else 0) @@ -144,9 +172,10 @@ -- raw error code plus the message from @mdb_strerror@. data LMDBException = LMDBKeyExist - | LMDBNotFound -- ^ Only seen via "Database.LMDB.Internal"; - -- the safe layer maps this to 'Nothing' - -- or 'False' instead of throwing. + | LMDBNotFound -- ^ Thrown by 'openDbi' for a missing named + -- database. Lookups, cursor moves and + -- 'del' report absence as 'Nothing' or + -- 'False' instead. | LMDBMapFull | LMDBCorrupted | LMDBPanic @@ -173,34 +202,53 @@ check c = throwLmdb c {-# INLINE check #-} +-- Reject an argument before it reaches LMDB, with the EINVAL LMDB +-- itself would use. +invalid :: String -> IO a +invalid msg = throwIO (LMDBOther (fromIntegral e) msg) + where + Errno e = eINVAL + -- environment lifecycle ------------------------------------------------------ -- | Open an LMDB environment at the given path. The path must already -- exist (as a directory unless 'envNoSubdir' is set, in which case -- the file's parent directory must exist). -- --- Throws 'LMDBException' on failure. +-- Throws 'LMDBException' on failure, including for a negative +-- 'envMapSize' or an out-of-range 'envMaxDbs'. -- -- >>> env <- open "/tmp/mydb" defaultEnvFlags { envNoSubdir = True } open :: FilePath -> EnvFlags -> IO Env -open path flags = mask $ \restore -> do - envp <- alloca $ \pp -> do - check =<< mdb_env_create pp - peek pp - let env = Env envp - flip onException (close env) . restore $ do - check =<< mdb_env_set_mapsize envp - (fromIntegral (envMapSize flags)) - check =<< mdb_env_set_maxdbs envp - (fromIntegral (envMaxDbs flags)) - withCString path $ \cpath -> - check =<< mdb_env_open envp cpath (envFlagBits flags) 0o644 - pure env +open path flags = do + checkFlags flags + lock <- newMVar () + mask $ \restore -> do + envp <- alloca $ \pp -> do + check =<< mdb_env_create pp + peek pp + let env = Env envp lock + flip onException (close env) . restore $ do + check =<< mdb_env_set_mapsize envp + (fromIntegral (envMapSize flags)) + check =<< mdb_env_set_maxdbs envp + (fromIntegral (envMaxDbs flags)) + withCString path $ \cpath -> + check =<< mdb_env_open envp cpath (envFlagBits flags) 0o644 + pure env --- | Close an environment. Must not be called while transactions are --- live. +-- Reject values that would wrap when converted to LMDB's types. +checkFlags :: EnvFlags -> IO () +checkFlags EnvFlags { envMapSize = sz, envMaxDbs = dbs } + | sz < 0 = invalid "open: negative envMapSize" + | dbs < 0 || toInteger dbs > toInteger (maxBound :: MDB_dbi) = + invalid "open: envMaxDbs out of range" + | otherwise = pure () + +-- | Close an environment. Must be called at most once, after all of +-- its transactions have ended. close :: Env -> IO () -close (Env envp) = mdb_env_close envp +close (Env envp _) = mdb_env_close envp -- | Bracketed environment lifecycle: 'open' the environment, run the -- action, 'close' on exit (including async exceptions). @@ -209,81 +257,136 @@ withEnv :: FilePath -> EnvFlags -> (Env -> IO a) -> IO a withEnv path flags = bracket (open path flags) close +-- | Flush the environment's buffers to disk. Commits already do so +-- unless 'envNoSync' is set. +-- +-- >>> sync env +sync :: Env -> IO () +sync (Env envp _) = check =<< mdb_env_sync envp 1 + -- transactions --------------------------------------------------------------- beginTxn :: Env -> CUInt -> IO (Ptr MDB_txn) -beginTxn (Env envp) flags = alloca $ \pp -> do +beginTxn (Env envp _) flags = alloca $ \pp -> do check =<< mdb_txn_begin envp nullPtr flags pp peek pp -- | Run an action in a read-only transaction. The transaction is -- aborted on exit; read transactions have no work to commit. -- +-- Read transactions don't wait on writers. They may be nested, or +-- run from any number of threads at once, up to LMDB's limit of +-- 126 concurrent readers. +-- -- >>> withReadTxn env $ \txn -> ... withReadTxn :: Env -> (Txn RO -> IO a) -> IO a -withReadTxn env act = bracketOnError - (beginTxn env _MDB_RDONLY) - mdb_txn_abort - (\txnp -> do - r <- act (Txn txnp) - mdb_txn_abort txnp - pure r) +withReadTxn env act = + bracket (beginTxn env _MDB_RDONLY) mdb_txn_abort (act . Txn) -- | Run an action in a read-write transaction. Commits if the action -- returns normally; aborts on exception. -- +-- Write transactions are serialized: one waits for any other in +-- progress to finish. They must not be nested. +-- +-- LMDB requires a write transaction to begin and end on the same +-- OS thread. Under the threaded RTS, if the calling thread is +-- unbound, the action therefore runs in a temporary bound thread +-- (see 'Control.Concurrent.runInBoundThread'), and asynchronous +-- exceptions thrown to the caller are deferred until the +-- transaction has ended. +-- -- >>> withWriteTxn env $ \txn -> ... withWriteTxn :: Env -> (Txn RW -> IO a) -> IO a -withWriteTxn env act = bracketOnError - (beginTxn env 0) - mdb_txn_abort - (\txnp -> do - r <- act (Txn txnp) - check =<< mdb_txn_commit txnp - pure r) +withWriteTxn env act = locked env 0 (act . Txn) +-- Run a txn under the env lock, committing if the action returns +-- normally and aborting if it throws. +locked :: Env -> CUInt -> (Ptr MDB_txn -> IO a) -> IO a +locked env@(Env _ lock) flags act = + -- writers queue on the lock, not on LMDB's own mutex, which would + -- block an OS thread: under the non-threaded RTS, the only one + withMVar lock $ \_ -> pinned $ mask $ \restore -> do + txnp <- beginTxn env flags + r <- restore (act txnp) `onException` mdb_txn_abort txnp + -- the commit runs masked, and must not be followed by an abort: + -- mdb_txn_commit ends the txn whether or not it succeeds + check =<< mdb_txn_commit txnp + pure r + +-- Run an action on a single OS thread. Under the non-threaded RTS, +-- all Haskell code already does. +pinned :: IO a -> IO a +pinned act + | rtsSupportsBoundThreads = runInBoundThread act + | otherwise = act + -- databases ------------------------------------------------------------------ --- | Open a database within a transaction. Pass 'Nothing' for the --- unnamed default database. If the @create@ flag is 'True', the --- database is created if it does not already exist — and the --- transaction must be a read-write transaction. +-- | Open a database, in a transaction of its own. Pass 'Nothing' for +-- the unnamed default database. If @create@ is 'True', the database +-- is created if it does not already exist, which requires a +-- writable environment; otherwise a missing named database throws +-- 'LMDBNotFound'. -- --- >>> dbi <- openDbi txn Nothing True +-- The handle is valid until the 'Env' is closed. Open each database +-- once, up front, and reuse the handle: a transaction that began +-- before 'openDbi' returned can't use it. Like a write transaction, +-- 'openDbi' must not be called within 'withWriteTxn'. +-- +-- >>> dbi <- openDbi env Nothing True openDbi - :: Txn s + :: Env -> Maybe BS.ByteString -- ^ database name; 'Nothing' for the default -> Bool -- ^ create if not present -> IO Dbi -openDbi (Txn txnp) mname create = alloca $ \pp -> do - let flags = if create then _MDB_CREATE else 0 - go cstr = do - check =<< mdb_dbi_open txnp cstr flags pp - Dbi <$> peek pp - case mname of - Nothing -> go nullPtr - Just nm -> BU.unsafeUseAsCString nm go +openDbi env mname create = withName mname $ \cname -> + -- LMDB forbids concurrent mdb_dbi_open calls, and keeps a handle + -- only once the txn that opened it commits; 'locked' ensures both + locked env txnFlags $ \txnp -> alloca $ \pp -> do + check =<< mdb_dbi_open txnp cname dbiFlags pp + Dbi <$> peek pp + where + (txnFlags, dbiFlags) + | create = (0, _MDB_CREATE) + | otherwise = (_MDB_RDONLY, 0) +withName :: Maybe BS.ByteString -> (CString -> IO a) -> IO a +withName Nothing k = k nullPtr +withName (Just nm) k + | BS.elem 0 nm = invalid "openDbi: database name contains NUL" + | otherwise = BS.useAsCString nm k + -- bytestring <-> MDB_val helpers -------------------------------------------- --- An MDB_val is two words: a CSize length at offset 0, a payload Ptr --- at offset 8 (on 64-bit). We poke fields directly rather than going --- through the Storable instance so hot paths don't depend on it being --- inlined. +-- An MDB_val is a size_t length followed by a data pointer, each one +-- word wide on every GHC target. We poke fields directly rather than +-- going through the Storable instance so hot paths don't depend on it +-- being inlined. +-- Offset of an MDB_val's data pointer, and the size of an MDB_val. +valData, valSize :: Int +valData = sizeOf (nullPtr :: Ptr ()) +valSize = 2 * valData + +pokeVal :: Ptr MDB_val -> Ptr CChar -> Int -> IO () +pokeVal vp p n = do + pokeByteOff vp 0 (fromIntegral n :: CSize) + pokeByteOff vp valData p +{-# INLINE pokeVal #-} + withBSAsVal :: BS.ByteString -> (Ptr MDB_val -> IO a) -> IO a withBSAsVal bs k = BU.unsafeUseAsCStringLen bs $ \(p, n) -> - allocaBytes 16 $ \vp -> do - pokeByteOff vp 0 (fromIntegral n :: CSize) - pokeByteOff vp 8 p + allocaBytes valSize $ \vp -> do + pokeVal vp p n k vp {-# INLINE withBSAsVal #-} peekVal :: Ptr MDB_val -> IO BS.ByteString peekVal vp = do sz <- peekByteOff vp 0 :: IO CSize - pd <- peekByteOff vp 8 + pd <- peekByteOff vp valData BS.packCStringLen (pd, fromIntegral sz) {-# INLINE peekVal #-} @@ -304,10 +407,9 @@ -> IO (Maybe BS.ByteString) -- ^ value, if present get (Txn txnp) (Dbi dbi) k = BU.unsafeUseAsCStringLen k $ \(kdata, klen) -> - allocaBytes 32 $ \kp -> do - let !vp = kp `plusPtr` 16 - pokeByteOff kp 0 (fromIntegral klen :: CSize) - pokeByteOff kp 8 kdata + allocaBytes (2 * valSize) $ \kp -> do + let !vp = kp `plusPtr` valSize + pokeVal kp kdata klen code <- mdb_get txnp dbi kp vp if code == _MDB_NOTFOUND then pure Nothing @@ -328,12 +430,10 @@ put (Txn txnp) (Dbi dbi) k v = BU.unsafeUseAsCStringLen k $ \(kdata, klen) -> BU.unsafeUseAsCStringLen v $ \(vdata, vlen) -> - allocaBytes 32 $ \kp -> do - let !vp = kp `plusPtr` 16 - pokeByteOff kp 0 (fromIntegral klen :: CSize) - pokeByteOff kp 8 kdata - pokeByteOff vp 0 (fromIntegral vlen :: CSize) - pokeByteOff vp 8 vdata + allocaBytes (2 * valSize) $ \kp -> do + let !vp = kp `plusPtr` valSize + pokeVal kp kdata klen + pokeVal vp vdata vlen check =<< mdb_put txnp dbi kp vp 0 {-# INLINE put #-} @@ -371,15 +471,15 @@ peek pp use' p = act (Cursor p) --- Scan-position helper (no input key): one 32-byte buffer for both --- output MDB_vals. +-- Scan-position helper (no input key): one buffer for both output +-- MDB_vals. cursorMove :: Cursor s -> CInt -- MDB_cursor_op -> IO (Maybe (BS.ByteString, BS.ByteString)) cursorMove (Cursor cp) op = - allocaBytes 32 $ \kp -> do - let !vp = kp `plusPtr` 16 + allocaBytes (2 * valSize) $ \kp -> do + let !vp = kp `plusPtr` valSize code <- mdb_cursor_get cp kp vp op if code == _MDB_NOTFOUND then pure Nothing @@ -428,10 +528,9 @@ -> IO (Maybe (BS.ByteString, BS.ByteString)) cursorSeek (Cursor cp) k = BU.unsafeUseAsCStringLen k $ \(kdata, klen) -> - allocaBytes 32 $ \kp -> do - let !vp = kp `plusPtr` 16 - pokeByteOff kp 0 (fromIntegral klen :: CSize) - pokeByteOff kp 8 kdata + allocaBytes (2 * valSize) $ \kp -> do + let !vp = kp `plusPtr` valSize + pokeVal kp kdata klen code <- mdb_cursor_get cp kp vp _MDB_SET_RANGE if code == _MDB_NOTFOUND then pure Nothing diff --git a/lib/Database/LMDB/Internal.hs b/lib/Database/LMDB/Internal.hs --- a/lib/Database/LMDB/Internal.hs +++ b/lib/Database/LMDB/Internal.hs @@ -184,6 +184,14 @@ -- foreign imports ------------------------------------------------------------ +-- The key-value and cursor calls are unsafe: they are short and don't +-- block, so skipping the safe-call overhead (roughly 100ns a call) is +-- worthwhile. The cost is that an unsafe call can't be descheduled: a +-- page fault on data not yet in memory stalls the calling capability, +-- and under the threaded RTS any GC waiting on it, until the read +-- completes. Calls that may block for long (txn begin and commit, +-- environment setup) remain safe. + -- environment lifecycle foreign import capi "lmdb.h mdb_env_create" @@ -236,11 +244,11 @@ -- key-value ops -foreign import capi "lmdb.h mdb_get" +foreign import capi unsafe "lmdb.h mdb_get" mdb_get :: Ptr MDB_txn -> MDB_dbi -> Ptr MDB_val -> Ptr MDB_val -> IO CInt -foreign import capi "lmdb.h mdb_put" +foreign import capi unsafe "lmdb.h mdb_put" mdb_put :: Ptr MDB_txn -> MDB_dbi @@ -249,7 +257,7 @@ -> CUInt -- flags -> IO CInt -foreign import capi "lmdb.h mdb_del" +foreign import capi unsafe "lmdb.h mdb_del" mdb_del :: Ptr MDB_txn -> MDB_dbi @@ -259,14 +267,14 @@ -- cursors -foreign import capi "lmdb.h mdb_cursor_open" +foreign import capi unsafe "lmdb.h mdb_cursor_open" mdb_cursor_open :: Ptr MDB_txn -> MDB_dbi -> Ptr (Ptr MDB_cursor) -> IO CInt -foreign import capi "lmdb.h mdb_cursor_close" +foreign import capi unsafe "lmdb.h mdb_cursor_close" mdb_cursor_close :: Ptr MDB_cursor -> IO () -foreign import capi "lmdb.h mdb_cursor_get" +foreign import capi unsafe "lmdb.h mdb_cursor_get" mdb_cursor_get :: Ptr MDB_cursor -> Ptr MDB_val -- key (in/out) diff --git a/ppad-lmdb.cabal b/ppad-lmdb.cabal --- a/ppad-lmdb.cabal +++ b/ppad-lmdb.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: ppad-lmdb -version: 0.1.0 +version: 0.2.0 synopsis: Minimal bindings to LMDB. license: MIT license-file: LICENSE @@ -52,11 +52,9 @@ base >= 4.9 && < 5 , bytestring >= 0.9 && < 0.13 -test-suite lmdb-tests - type: exitcode-stdio-1.0 +common tests default-language: Haskell2010 hs-source-dirs: test - main-is: Main.hs ghc-options: -rtsopts -Wall -O2 build-depends: @@ -71,6 +69,19 @@ , tasty-hunit , tasty-quickcheck +-- the same suite, under each RTS +test-suite lmdb-tests + import: tests + type: exitcode-stdio-1.0 + main-is: Main.hs + +test-suite lmdb-tests-threaded + import: tests + type: exitcode-stdio-1.0 + main-is: Main.hs + ghc-options: + -threaded -with-rtsopts=-N4 + benchmark lmdb-bench type: exitcode-stdio-1.0 default-language: Haskell2010 @@ -78,10 +89,13 @@ main-is: Main.hs ghc-options: -rtsopts -O2 -Wall + if flag(llvm) + ghc-options: -fllvm build-depends: base , bytestring , criterion + , deepseq , directory , filepath , ppad-lmdb @@ -98,6 +112,7 @@ build-depends: base , bytestring + , deepseq , directory , filepath , ppad-lmdb diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -2,19 +2,28 @@ module Main where +import Control.Concurrent + ( forkIO + , isCurrentThreadBound + , newEmptyMVar + , putMVar + , rtsSupportsBoundThreads + , takeMVar + , yield + ) import qualified Control.Exception as E -import Control.Monad (forM_) +import Control.Monad (forM, forM_, replicateM_) import qualified Data.ByteString as BS import qualified Data.ByteString.Char8 as BSC import qualified Data.Map.Strict as Map import qualified Database.LMDB as L import System.Directory ( createDirectory - , doesDirectoryExist , getTemporaryDirectory , removeDirectoryRecursive ) import System.FilePath (()) +import System.IO.Error (isAlreadyExistsError) import Test.Tasty import Test.Tasty.HUnit import Test.Tasty.QuickCheck as QC @@ -27,19 +36,31 @@ -- helpers -------------------------------------------------------------------- --- | Run an action against a fresh temporary db directory. The --- directory is created (clean) before the action and removed after. -withTmpEnv :: L.EnvFlags -> (L.Env -> IO a) -> IO a -withTmpEnv envFlags k = do +-- | Run an action against a fresh temporary directory. The directory +-- is unique to the call, so tests may run concurrently, and is +-- removed afterwards even if the action throws. +withTmpDir :: (FilePath -> IO a) -> IO a +withTmpDir k = do tmpRoot <- getTemporaryDirectory - let dir = tmpRoot "ppad-lmdb-test" - exists <- doesDirectoryExist dir - if exists then removeDirectoryRecursive dir else pure () - createDirectory dir - result <- L.withEnv dir envFlags k - removeDirectoryRecursive dir - pure result + E.bracket (fresh tmpRoot (0 :: Int)) removeDirectoryRecursive k + where + fresh root n = do + let dir = root ("ppad-lmdb-test-" ++ show n) + r <- E.try (createDirectory dir) + case r of + Right () -> pure dir + Left e + | isAlreadyExistsError e -> fresh root (n + 1) + | otherwise -> E.throwIO e +withTmpEnv :: L.EnvFlags -> (L.Env -> IO a) -> IO a +withTmpEnv envFlags k = withTmpDir $ \dir -> L.withEnv dir envFlags k + +-- | 'withTmpEnv', with the default database opened. +withTmpDb :: L.EnvFlags -> (L.Env -> L.Dbi -> IO a) -> IO a +withTmpDb envFlags k = withTmpEnv envFlags $ \env -> + L.openDbi env Nothing True >>= k env + -- shared bigger map for tests flags :: L.EnvFlags flags = L.defaultEnvFlags @@ -48,36 +69,47 @@ } writeKV - :: L.Env -> [(BS.ByteString, BS.ByteString)] -> IO () -writeKV env kvs = L.withWriteTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing True + :: L.Env -> L.Dbi -> [(BS.ByteString, BS.ByteString)] -> IO () +writeKV env dbi kvs = L.withWriteTxn env $ \txn -> forM_ kvs $ \(k, v) -> L.put txn dbi k v +readKey :: L.Env -> L.Dbi -> BS.ByteString -> IO (Maybe BS.ByteString) +readKey env dbi k = L.withReadTxn env $ \txn -> L.get txn dbi k + +-- | Run each action in its own forkIO thread and wait for all of +-- them, rethrowing the first failure. +forkAll :: [IO ()] -> IO () +forkAll acts = do + dones <- forM acts $ \act -> do + done <- newEmptyMVar + _ <- forkIO $ + (E.try act :: IO (Either E.SomeException ())) >>= putMVar done + pure done + rs <- mapM takeMVar dones + either E.throwIO pure (sequence_ rs) + -- unit tests ----------------------------------------------------------------- unitTests :: TestTree unitTests = testGroup "unit" [ testCase "put then get roundtrip" $ - withTmpEnv flags $ \env -> do - writeKV env [("hello", "world")] - v <- L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False + withTmpDb flags $ \env dbi -> do + writeKV env dbi [("hello", "world")] + v <- L.withReadTxn env $ \txn -> L.get txn dbi "hello" v @?= Just "world" , testCase "get on missing key returns Nothing" $ - withTmpEnv flags $ \env -> do - writeKV env [("a", "1")] - v <- L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False + withTmpDb flags $ \env dbi -> do + writeKV env dbi [("a", "1")] + v <- L.withReadTxn env $ \txn -> L.get txn dbi "absent" v @?= Nothing , testCase "del returns True only when key existed" $ - withTmpEnv flags $ \env -> do - writeKV env [("a", "1"), ("b", "2")] + withTmpDb flags $ \env dbi -> do + writeKV env dbi [("a", "1"), ("b", "2")] (existed, repeated, missing) <- L.withWriteTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False e <- L.del txn dbi "a" e2 <- L.del txn dbi "a" m <- L.del txn dbi "never-existed" @@ -87,10 +119,9 @@ missing @?= False , testCase "write txn aborted on exception" $ - withTmpEnv flags $ \env -> do + withTmpDb flags $ \env dbi -> do -- attempt that throws after a write let attempt = L.withWriteTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing True L.put txn dbi "k" "v" E.throwIO (userError "boom") r <- E.try attempt :: IO (Either IOError ()) @@ -98,29 +129,192 @@ Left _ -> pure () Right _ -> assertFailure "expected the user error" -- key must not be visible to a fresh read txn - v <- L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing True + v <- L.withReadTxn env $ \txn -> L.get txn dbi "k" v @?= Nothing - , testCase "cursor walks keys in ascending order" $ + , testCase "failed commit leaves env usable" $ + withTmpDb flags { L.envMapSize = 1024 * 1024 } $ \env dbi -> do + -- a caught MapFull poisons the txn, so its commit must fail + r <- E.try $ L.withWriteTxn env $ \txn -> do + full <- E.try $ forM_ [0 :: Int ..] $ \i -> + L.put txn dbi (BSC.pack (show i)) (BS.replicate 512 0) + case full of + Left L.LMDBMapFull -> pure () + _ -> assertFailure "expected LMDBMapFull" + case r of + Left (L.LMDBOther (-30782) _) -> pure () -- MDB_BAD_TXN + _ -> assertFailure ("expected MDB_BAD_TXN, got " ++ show r) + forM_ [1 :: Int .. 3] $ \_ -> writeKV env dbi [("k", "v")] + v <- L.withReadTxn env $ \txn -> + L.get txn dbi "k" + v @?= Just "v" + + , testCase "dbi name is used exactly as given" $ withTmpEnv flags $ \env -> do + -- a slice: the bytes after it are not a NUL terminator + _ <- L.openDbi env (Just (BS.take 3 "foobar")) True + r <- E.try (L.openDbi env (Just "foobar") False) + case r of + Left L.LMDBNotFound -> pure () + _ -> assertFailure "opened a database that was never created" + _ <- L.openDbi env (Just "foo") False + pure () + + , testCase "dbi name containing NUL is rejected" $ + withTmpEnv flags $ \env -> do + r <- E.try (L.openDbi env (Just "foo\0bar") True) + case r of + Left (L.LMDBOther 22 _) -> pure () -- EINVAL + _ -> assertFailure "expected EINVAL" + + , testCase "named databases are independent" $ + withTmpEnv flags $ \env -> do + a <- L.openDbi env (Just "a") True + b <- L.openDbi env (Just "b") True + L.withWriteTxn env $ \txn -> do + L.put txn a "k" "1" + L.put txn b "k" "2" + vs <- L.withReadTxn env $ \txn -> + (,) <$> L.get txn a "k" <*> L.get txn b "k" + vs @?= (Just "1", Just "2") + + , testCase "concurrent openDbi calls" $ + withTmpEnv flags $ \env -> + forkAll + [ do dbi <- L.openDbi env (Just name) True + writeKV env dbi [("k", name)] + v <- readKey env dbi "k" + v @?= Just name + | name <- ["w", "x", "y", "z"] ] + + , testCase "read-only env serves reads, refuses writes" $ + withTmpDir $ \dir -> do + L.withEnv dir flags $ \env -> do + dbi <- L.openDbi env Nothing True + writeKV env dbi [("k", "v")] + L.withEnv dir flags { L.envReadOnly = True } $ \env -> do + dbi <- L.openDbi env Nothing False + v <- readKey env dbi "k" + v @?= Just "v" + w <- E.try (L.withWriteTxn env (\_ -> pure ())) + case w of + Left (L.LMDBOther 13 _) -> pure () -- EACCES + _ -> assertFailure "expected EACCES from withWriteTxn" + c <- E.try (L.openDbi env (Just "new") True) + case c of + Left (L.LMDBOther 13 _) -> pure () + _ -> assertFailure "expected EACCES from openDbi" + + , testCase "read txns nest" $ + withTmpDb flags $ \env dbi -> do + writeKV env dbi [("k", "v")] + v <- L.withReadTxn env $ \_ -> readKey env dbi "k" + v @?= Just "v" + + , testCase "concurrent readers see their own snapshots" $ + withTmpDb flags $ \env dbi -> do + writeKV env dbi [("k", "old")] + entered <- newEmptyMVar + written <- newEmptyMVar + seen <- newEmptyMVar + forkAll + [ L.withReadTxn env $ \txn -> do + putMVar entered () + takeMVar written + L.get txn dbi "k" >>= putMVar seen + , do takeMVar entered + -- a second reader, while the first is live + old <- readKey env dbi "k" + old @?= Just "old" + writeKV env dbi [("k", "new")] + putMVar written () + ] + stale <- takeMVar seen + stale @?= Just "old" + fresh <- readKey env dbi "k" + fresh @?= Just "new" + + , testCase "concurrent writers serialize" $ + withTmpDb flags { L.envNoSync = True } $ \env dbi -> do + let bump = L.withWriteTxn env $ \txn -> do + n <- maybe 0 (maybe 0 fst . BSC.readInt) + <$> L.get txn dbi "n" + yield -- invite another writer in mid-txn + L.put txn dbi "n" (BSC.pack (show (n + 1))) + forkAll (replicate 8 (replicateM_ 50 bump)) + n <- readKey env dbi "n" + n @?= Just "400" + + , testCase "write txns run bound where the RTS allows" $ + withTmpEnv flags $ \env -> do + bound <- newEmptyMVar + forkAll + [ L.withWriteTxn env (\_ -> isCurrentThreadBound) + >>= putMVar bound ] + b <- takeMVar bound + b @?= rtsSupportsBoundThreads + + , testCase "open rejects out-of-range flags" $ + forM_ [ flags { L.envMapSize = -1 } + , flags { L.envMaxDbs = -1 } + , flags { L.envMaxDbs = 2 ^ (32 :: Int) } + ] $ \bad -> do + r <- E.try (withTmpEnv bad (\_ -> pure ())) + case r of + Left (L.LMDBOther 22 _) -> pure () -- EINVAL + _ -> assertFailure ("expected EINVAL for " ++ show bad) + + , testCase "sync flushes a no-sync env" $ + withTmpDb flags { L.envNoSync = True } $ \env dbi -> do + writeKV env dbi [("k", "v")] + L.sync env + + , testCase "cursor walks keys in ascending order" $ + withTmpDb flags $ \env dbi -> do let kvs = [ (BSC.pack [c], BSC.pack [c]) | c <- "abcdefghij" ] - writeKV env kvs - seen <- L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False + writeKV env dbi kvs + seen <- L.withReadTxn env $ \txn -> L.withCursor txn dbi (cursorWalk []) seen @?= kvs , testCase "cursorSeek lands on first >= key" $ - withTmpEnv flags $ \env -> do - writeKV env [ ("a","1"), ("c","3"), ("e","5") ] - r <- L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False + withTmpDb flags $ \env dbi -> do + writeKV env dbi [ ("a","1"), ("c","3"), ("e","5") ] + r <- L.withReadTxn env $ \txn -> L.withCursor txn dbi $ \cur -> L.cursorSeek cur "b" r @?= Just ("c", "3") + + , testCase "cursorSeek past the last key finds nothing" $ + withTmpDb flags $ \env dbi -> do + writeKV env dbi [ ("a","1"), ("c","3") ] + r <- L.withReadTxn env $ \txn -> + L.withCursor txn dbi $ \cur -> + L.cursorSeek cur "d" + r @?= Nothing + + , testCase "cursor walks keys in descending order" $ + withTmpDb flags $ \env dbi -> do + let kvs = [ (BSC.pack [c], BSC.pack [c]) + | c <- "abcdefghij" ] + writeKV env dbi kvs + seen <- L.withReadTxn env $ \txn -> + L.withCursor txn dbi $ \cur -> do + let go acc Nothing = pure (reverse acc) + go acc (Just kv) = L.cursorPrev cur >>= go (kv : acc) + L.cursorLast cur >>= go [] + seen @?= reverse kvs + + , testCase "cursors on an empty database find nothing" $ + withTmpDb flags $ \env dbi -> do + r <- L.withReadTxn env $ \txn -> + L.withCursor txn dbi $ \cur -> + (,,) <$> L.cursorFirst cur + <*> L.cursorLast cur + <*> L.cursorSeek cur "a" + r @?= (Nothing, Nothing, Nothing) ] -- walk a cursor, accumulating pairs from current position to end @@ -156,19 +350,17 @@ propTests :: TestTree propTests = testGroup "props" [ QC.testProperty "put / get roundtrip" $ \(Key k) (Val v) -> - QC.ioProperty $ withTmpEnv flags $ \env -> do - writeKV env [(k, v)] - got <- L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False + QC.ioProperty $ withTmpDb flags $ \env dbi -> do + writeKV env dbi [(k, v)] + got <- L.withReadTxn env $ \txn -> L.get txn dbi k pure (got == Just v) , QC.testProperty "cursor enumerates Map.toAscList" $ \kvs0 -> let kvs = uniqueKeys [(k, v) | (Key k, Val v) <- kvs0] - in QC.ioProperty $ withTmpEnv flags $ \env -> do - writeKV env kvs - seen <- L.withReadTxn env $ \txn -> do - dbi <- L.openDbi txn Nothing False + in QC.ioProperty $ withTmpDb flags $ \env dbi -> do + writeKV env dbi kvs + seen <- L.withReadTxn env $ \txn -> L.withCursor txn dbi (cursorWalk []) pure (seen == Map.toAscList (Map.fromList kvs)) ]