packages feed

streamly-lmdb (empty) → 0.0.1

raw patch · 11 files changed

+892/−0 lines, 11 filesdep +QuickCheckdep +asyncdep +basesetup-changed

Dependencies added: QuickCheck, async, base, bytestring, directory, streamly, streamly-lmdb, tasty, tasty-quickcheck, temporary

Files

+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for streamly-lmdb++## Unreleased changes
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Shlok Datye (c) 2020++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 Shlok Datye 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.
+ README.md view
@@ -0,0 +1,60 @@+# streamly-lmdb++Stream data to or from LMDB databases using the Haskell [streamly](https://hackage.haskell.org/package/streamly) library.++## Requirements++Install LMDB on your system:++* Debian Linux: `sudo apt-get install liblmdb-dev`.+* macOS: `brew install lmdb`.++## Quick start++```haskell+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Streamly.External.LMDB+    (Limits (mapSize), WriteOptions (chunkSize), defaultLimits,+    defaultWriteOptions, getDatabase, openEnvironment, readLMDB,+    tebibyte, writeLMDB)+import qualified Streamly.Prelude as S++main :: IO ()+main = do+    -- Open an environment. There should already exist a file or+    -- directory at the given path. (Empty for a new environment.)+    env <- openEnvironment "/path/to/lmdb-database" $+            defaultLimits { mapSize = tebibyte }++    -- Get the main database.+    -- Note: It is common practice with LMDB to create the database+    -- once and reuse it for the remainder of the program’s execution.+    db <- getDatabase env Nothing++    -- Stream key-value pairs into the database.+    let fold' = writeLMDB db defaultWriteOptions { chunkSize = 1 }+    let writeStream = S.fromList [("baz", "a"), ("foo", "b"), ("bar", "c")]+    _ <- S.fold fold' writeStream++    -- Stream key-value pairs out of the+    -- database, printing them along the way.+    -- Output:+    --     ("bar","c")+    --     ("baz","a")+    --     ("foo","b")+    let unfold' = readLMDB db+    let readStream = S.unfold unfold' undefined+    S.mapM_ print readStream+```++## Benchmarks++See `bench/README.md`. Summary (with rough figures from our machine<sup id="a1">[1](#fn1)</sup>):++* **Reading.** For reading a fully cached LMDB database, this library (when `unsafeReadLMDB` is used instead of `readLMDB`) has a 10 ns/pair overhead compared to plain Haskell `IO` code, which has another 10 ns/pair overhead compared to C. (The first two being similar fulfills the promise of [streamly](https://hackage.haskell.org/package/streamly) and stream fusion.) We deduce that if your total workload per pair takes longer than 20 ns, your bottleneck will not be your usage of this library as opposed to C.+* **Writing**. Writing with plain Haskell `IO` code and with this library is, respectively, 10% and 20% slower than writing with C. We have not dug further into these differences because this write performance is currently good enough for our purposes.++<sup id="fn1">1</sup> [Linode](https://linode.com); Debian 10, Dedicated 32GB: 16 CPU, 640GB Storage, 32GB RAM. [↩](#a1)
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ src/Streamly/External/LMDB.hs view
@@ -0,0 +1,262 @@+{-# LANGUAGE BangPatterns, ScopedTypeVariables #-}++-- | The functionality for the limits and getting the environment and database were mostly+-- obtained from the [lmdb-simple](https://hackage.haskell.org/package/lmdb-simple) library.+module Streamly.External.LMDB+    (+    -- ** Types+    Database,+    Environment,+    Limits(..),+    LMDB_Error(..),+    MDB_ErrCode(..),+    Mode,+    ReadWrite,+    ReadOnly,+    WriteOptions(..),++    -- ** Environment and database+    defaultLimits,+    openEnvironment,+    isReadOnlyEnvironment,+    getDatabase,++    -- ** Utility+    gibibyte,+    tebibyte,+    clearDatabase,++    -- ** Reading+    readLMDB,+    unsafeReadLMDB,++    -- ** Writing+    defaultWriteOptions,+    writeLMDB) where++import Control.Concurrent (isCurrentThreadBound)+import Control.Concurrent.Async (asyncBound, wait)+import Control.Exception (Exception, catch, tryJust, mask_, throw)+import Control.Monad (guard, when)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.ByteString (ByteString, packCStringLen)+import Data.ByteString.Unsafe (unsafeUseAsCStringLen)+import Data.Maybe (fromJust)+import Data.Void (Void)+import Foreign (Ptr, free, malloc, nullPtr, peek)+import Foreign.C (Errno (Errno), eNOTDIR)+import Foreign.C.String (CStringLen)+import Streamly.Internal.Data.Fold (Fold (Fold))+import Streamly.Internal.Data.Stream.StreamD (newFinalizedIORef, runIORefFinalizer)+import Streamly.Internal.Data.Stream.StreamD.Type (Step (Stop, Yield))+import Streamly.Internal.Data.Unfold (supply)+import Streamly.Internal.Data.Unfold.Types (Unfold (Unfold))++import Streamly.External.LMDB.Internal+import Streamly.External.LMDB.Internal.Foreign++newtype Environment mode = Environment (Ptr MDB_env)++isReadOnlyEnvironment :: Mode mode => Environment mode -> Bool+isReadOnlyEnvironment = isReadOnlyMode . mode+    where+        mode :: Environment mode -> mode+        mode = undefined++-- | LMDB environments have various limits on the size and number of databases and concurrent readers.+data Limits = Limits+    { mapSize      :: !Int  -- ^ Memory map size, in bytes (also the maximum size of all databases).+    , maxDatabases :: !Int  -- ^ Maximum number of named databases.+    , maxReaders   :: !Int  -- ^ Maximum number of concurrent 'ReadOnly' transactions+                            --   (also the number of slots in the lock table).+  }++-- | The default limits are 1 MiB map size, 0 named databases, and 126 concurrent readers. These can be adjusted+-- freely, and in particular the 'mapSize' may be set very large (limited only by available address space). However,+-- LMDB is not optimized for a large number of named databases so 'maxDatabases' should be kept to a minimum.+--+-- The default 'mapSize' is intentionally small, and should be changed to something appropriate for your application.+-- It ought to be a multiple of the OS page size, and should be chosen as large as possible to accommodate future+-- growth of the database(s). Once set for an environment, this limit cannot be reduced to a value smaller than+-- the space already consumed by the environment, however it can later be increased.+--+-- If you are going to use any named databases then you will need to change 'maxDatabases'+-- to the number of named databases you plan to use. However, you do not need to change+-- this field if you are only going to use the single main (unnamed) database.+defaultLimits :: Limits+defaultLimits = Limits+    { mapSize      = 1024 * 1024  -- 1 MiB.+    , maxDatabases = 0+    , maxReaders   = 126+    }++-- A convenience constant for obtaining a 1 GiB map size.+gibibyte :: Int+gibibyte = 1024 * 1024 * 1024++-- A convenience constant for obtaining a 1 TiB map size.+tebibyte :: Int+tebibyte = 1024 * 1024 * 1024 * 1024++-- | Open an LMDB environment in either 'ReadWrite' or 'ReadOnly' mode. The 'FilePath' argument+-- may be either a directory or a regular file, but it must already exist. If a regular file,+-- an additional file with "-lock" appended to the name is used for the reader lock table.+--+-- Note that an environment must have been opened in 'ReadWrite'+-- mode at least once before it can be opened in 'ReadOnly' mode.+--+-- An environment opened in 'ReadOnly' mode may still modify the reader lock table+-- (except when the filesystem is read-only, in which case no locks are used).+openEnvironment :: Mode mode => FilePath -> Limits -> IO (Environment mode)+openEnvironment path limits = do+    penv <- mdb_env_create++    mdb_env_set_mapsize penv (mapSize limits)+    let maxDbs = maxDatabases limits in when (maxDbs /= 0) $ mdb_env_set_maxdbs penv maxDbs+    mdb_env_set_maxreaders penv (maxReaders limits)++    -- Always use MDB_NOTLS.+    let env = Environment penv :: Mode mode => Environment mode+        flags = mdb_notls : [mdb_rdonly | isReadOnlyEnvironment env]++    let isNotDirectoryError :: LMDB_Error -> Bool+        isNotDirectoryError LMDB_Error { e_code = Left code }+            | Errno (fromIntegral code) == eNOTDIR = True+        isNotDirectoryError _                      = False++    r <- tryJust (guard . isNotDirectoryError) $ mdb_env_open penv path (combineOptions flags)+    case r of+        Left  _ -> mdb_env_open penv path (combineOptions $ mdb_nosubdir : flags)+        Right _ -> return ()++    return env++getDatabase :: (Mode mode) => Environment mode -> Maybe String -> IO (Database mode)+getDatabase env@(Environment penv) name = do+    ptxn <- mdb_txn_begin penv nullPtr (combineOptions $ [mdb_rdonly | isReadOnlyEnvironment env])+    dbi <- mdb_dbi_open ptxn name (combineOptions $ [mdb_create | not $ isReadOnlyEnvironment env])+    mdb_txn_commit ptxn+    return $ Database penv dbi++-- | Clears, i.e., removes all key-value pairs from, the given database.+clearDatabase :: (Mode mode) => Database mode -> IO ()+clearDatabase (Database penv dbi) = asyncBound (do+    ptxn <- mdb_txn_begin penv nullPtr 0+    mdb_clear ptxn dbi+    mdb_txn_commit ptxn) >>= wait++-- | Creates an unfold with which we can stream all key-value pairs from the given database in increasing key order.+--+-- A read transaction is kept open for the duration of the unfold; one should therefore+-- bear in mind LMDB's [caveats regarding long-lived transactions](https://git.io/JJZE6).+--+-- If you don’t want the overhead of intermediate 'ByteString's (on your+-- way to your eventual data structures), use 'unsafeReadLMDB' instead.+{-# INLINE readLMDB #-}+readLMDB :: (MonadIO m, Mode mode) => Database mode -> Unfold m Void (ByteString, ByteString)+readLMDB db = unsafeReadLMDB db packCStringLen packCStringLen++-- | Creates an unfold with which we can stream all key-value pairs from the given database in increasing key order.+--+-- A read transaction is kept open for the duration of the unfold; one should therefore+-- bear in mind LMDB's [caveats regarding long-lived transactions](https://git.io/JJZE6).+--+-- To ensure safety, make sure that the memory pointed to by the 'CStringLen' for each key/value mapping function+-- call is (a) only read (and not written to); and (b) not used after the mapping function has returned. One way to+-- transform the 'CStringLen's to your desired data structures is to use 'Data.ByteString.Unsafe.unsafePackCStringLen'.+{-# INLINE unsafeReadLMDB #-}+unsafeReadLMDB :: (MonadIO m, Mode mode)+               => Database mode -> (CStringLen -> IO k) -> (CStringLen -> IO v) -> Unfold m Void (k, v)+unsafeReadLMDB (Database penv dbi) kmap vmap =+    flip supply mdb_first $ Unfold+        (\(op, pcurs, pk, pv, ref) -> do+            found <- liftIO $ c_mdb_cursor_get pcurs pk pv op >>= \rc ->+                if rc /= 0 && rc /= mdb_notfound then do+                    runIORefFinalizer ref+                    throwLMDBErrNum "mdb_cursor_get" rc+                else+                    return $ rc /= mdb_notfound+            if found then do+                !k <- liftIO $ (\x -> kmap (mv_data x, fromIntegral $ mv_size x)) =<< peek pk+                !v <- liftIO $ (\x -> vmap (mv_data x, fromIntegral $ mv_size x)) =<< peek pv+                return $ Yield (k, v) (mdb_next, pcurs, pk, pv, ref)+            else do+                runIORefFinalizer ref+                return Stop)+        (\op -> do+            (pcurs, pk, pv, ref) <- liftIO $ mask_ $ do+                ptxn <- liftIO $ mdb_txn_begin penv nullPtr mdb_rdonly+                pcurs <- liftIO $ mdb_cursor_open ptxn dbi+                pk <- liftIO malloc+                pv <- liftIO malloc+                ref <- liftIO . newFinalizedIORef $ do+                    free pv >> free pk+                    c_mdb_cursor_close pcurs+                    -- No need to commit this read-only transaction.+                    c_mdb_txn_abort ptxn+                return (pcurs, pk, pv, ref)+            return (op, pcurs, pk, pv, ref))++data WriteOptions = WriteOptions+    { writeTransactionSize :: !Int+    , noOverwrite :: !Bool+    , writeAppend :: !Bool }++defaultWriteOptions :: WriteOptions+defaultWriteOptions = WriteOptions+    { writeTransactionSize = 1+    , noOverwrite = False+    , writeAppend = False }+++newtype ExceptionString = ExceptionString String deriving (Show)+instance Exception ExceptionString++-- | Creates a fold with which we can stream key-value pairs into the given database.+--+-- It is the responsibility of the user to execute the fold on a bound thread.+--+-- The fold currently cannot be used with a scan. (The plan is for this shortcoming to be+-- remedied with or after a future release of streamly that addresses the underlying issue.)+--+-- Please specify a suitable transaction size in the write options; the default of 1 (one write transaction for each+-- key-value pair) could yield suboptimal performance. One could try, e.g., 100 KB chunks and benchmark from there.+{-# INLINE writeLMDB #-}+writeLMDB :: (MonadIO m) => Database ReadWrite -> WriteOptions -> Fold m (ByteString, ByteString) ()+writeLMDB (Database penv dbi) options =+    let txnSize = max 1 (writeTransactionSize options)+        flags = combineOptions $ [mdb_nooverwrite | noOverwrite options] ++ [mdb_append | writeAppend options]+    in Fold (\(currChunkSz, mtxn) (k, v) -> do+        currChunkSz' <- liftIO $+            if currChunkSz >= txnSize then do+                let (_, ref) = fromJust mtxn+                runIORefFinalizer ref+                return 0+            else+                return currChunkSz++        (ptxn, ref) <-+            if currChunkSz' == 0 then+                liftIO $ mask_ $ do+                    ptxn <- mdb_txn_begin penv nullPtr 0+                    ref <- newFinalizedIORef $ mdb_txn_commit ptxn+                    return (ptxn, ref)+            else+                return $ fromJust mtxn++        liftIO $ unsafeUseAsCStringLen k $ \(kp, kl) -> unsafeUseAsCStringLen v $ \(vp, vl) ->+            catch (mdb_put_ ptxn dbi kp (fromIntegral kl) vp (fromIntegral vl) flags)+                (\(e :: LMDB_Error) -> runIORefFinalizer ref >> throw e)++        return (currChunkSz' + 1, Just (ptxn, ref)))+    (do+        isBound <- liftIO isCurrentThreadBound+        if isBound then+            return (0, Nothing)+        else+            throw $ ExceptionString "Error: writeLMDB should be executed on a bound thread")+    -- This final part is incompatible with scans.+    (\(_, mtxn) -> liftIO $+        case mtxn of+            Nothing -> return ()+            Just (_, ref) -> runIORefFinalizer ref)
+ src/Streamly/External/LMDB/Internal.hs view
@@ -0,0 +1,17 @@+module Streamly.External.LMDB.Internal where++import Foreign (Ptr)+import Streamly.External.LMDB.Internal.Foreign (MDB_dbi_t, MDB_env)++-- This is in a separate internal module because the tests make use of the Database constructor.++class Mode a where+    isReadOnlyMode :: a -> Bool++data ReadWrite+data ReadOnly++instance Mode ReadWrite where isReadOnlyMode _ = False+instance Mode ReadOnly  where isReadOnlyMode _ = True++data Database mode = Database (Ptr MDB_env) MDB_dbi_t
+ src/Streamly/External/LMDB/Internal/Foreign.hsc view
@@ -0,0 +1,259 @@+-- Much of the code below was obtained from the lmdb library:+-- https://hackage.haskell.org/package/lmdb (https://hackage.haskell.org/package/lmdb-0.2.5/src/LICENSE)+module Streamly.External.LMDB.Internal.Foreign where++#include <lmdb.h>++import Control.Exception (Exception, throwIO)+import Control.Monad (when)+import Foreign ((.|.), Ptr, Storable (alignment, peek, peekByteOff,+                poke, pokeByteOff, sizeOf), Word16, Word32, alloca, nullPtr)+import Foreign.C.String (CString, peekCString, withCString)+import Foreign.C.Types (CChar, CInt (CInt), CSize (CSize), CUInt (CUInt))++import qualified Data.List as L++type MDB_mode_t = #type mdb_mode_t+type MDB_dbi_t = #type MDB_dbi+type MDB_cursor_op_t = #type MDB_cursor_op++data MDB_env+data MDB_txn+data MDB_cursor++data MDB_val = MDB_val+    { mv_size :: {-# UNPACK #-} !CSize+    , mv_data :: {-# UNPACK #-} !(Ptr CChar) }++instance Storable MDB_val where+    alignment _ = #{alignment MDB_val}+    {-# INLINE alignment #-}+    sizeOf _ = #{size MDB_val}+    {-# INLINE sizeOf #-}+    peek ptr = do+        sz <- #{peek MDB_val, mv_size} ptr+        pd <- #{peek MDB_val, mv_data} ptr+        return $! MDB_val sz pd+    {-# INLINE peek #-}+    poke ptr (MDB_val sz pd) = do+        #{poke MDB_val, mv_size} ptr sz+        #{poke MDB_val, mv_data} ptr pd+    {-# INLINE poke #-}++foreign import ccall unsafe "lmdb.h mdb_strerror"+    c_mdb_strerror :: CInt -> IO CString++foreign import ccall unsafe "lmdb.h mdb_env_create"+    c_mdb_env_create :: Ptr (Ptr MDB_env) -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_env_set_mapsize"+    c_mdb_env_set_mapsize :: Ptr MDB_env -> CSize -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_env_set_maxreaders"+    c_mdb_env_set_maxreaders :: Ptr MDB_env -> CUInt -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_env_set_maxdbs"+    c_mdb_env_set_maxdbs :: Ptr MDB_env -> MDB_dbi_t -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_env_open"+    c_mdb_env_open :: Ptr MDB_env -> CString -> CUInt -> MDB_mode_t -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_txn_begin"+    c_mdb_txn_begin :: Ptr MDB_env -> Ptr MDB_txn -> CUInt -> Ptr (Ptr MDB_txn) -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_dbi_open"+    c_mdb_dbi_open :: Ptr MDB_txn -> CString -> CUInt -> Ptr MDB_dbi_t -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_txn_commit"+    c_mdb_txn_commit :: Ptr MDB_txn -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_txn_abort"+    c_mdb_txn_abort :: Ptr MDB_txn -> IO ()++foreign import ccall unsafe "lmdb.h mdb_cursor_open"+    c_mdb_cursor_open :: Ptr MDB_txn -> MDB_dbi_t -> Ptr (Ptr MDB_cursor) -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_cursor_get"+    c_mdb_cursor_get :: Ptr MDB_cursor -> Ptr MDB_val -> Ptr MDB_val -> MDB_cursor_op_t -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_cursor_close"+    c_mdb_cursor_close :: Ptr MDB_cursor -> IO ()++foreign import ccall unsafe "lmdb.h mdb_put"+    c_mdb_put :: Ptr MDB_txn -> MDB_dbi_t -> Ptr MDB_val -> Ptr MDB_val -> CUInt -> IO CInt++foreign import ccall unsafe "streamly_lmdb_foreign.h mdb_put_"+    c_mdb_put_ :: Ptr MDB_txn -> MDB_dbi_t -> Ptr CChar -> CSize -> Ptr CChar -> CSize -> CUInt -> IO CInt++foreign import ccall unsafe "lmdb.h mdb_drop"+    c_mdb_drop :: Ptr MDB_txn -> MDB_dbi_t -> CInt -> IO CInt++data LMDB_Error = LMDB_Error+    { e_context     :: String+    , e_description :: String+    , e_code        :: Either Int MDB_ErrCode+    } deriving (Show)+instance Exception LMDB_Error++data MDB_ErrCode+    = MDB_KEYEXIST+    | MDB_NOTFOUND+    | MDB_PAGE_NOTFOUND+    | MDB_CORRUPTED+    | MDB_PANIC+    | MDB_VERSION_MISMATCH+    | MDB_INVALID+    | MDB_MAP_FULL+    | MDB_DBS_FULL+    | MDB_READERS_FULL+    | MDB_TLS_FULL+    | MDB_TXN_FULL+    | MDB_CURSOR_FULL+    | MDB_PAGE_FULL+    | MDB_MAP_RESIZED+    | MDB_INCOMPATIBLE+    | MDB_BAD_RSLOT+    | MDB_BAD_TXN+    | MDB_BAD_VALSIZE+    | MDB_BAD_DBI+    deriving (Show)++{-# INLINE errCodes #-}+errCodes :: [(MDB_ErrCode, Int)]+errCodes =+    [ (MDB_KEYEXIST, #const MDB_KEYEXIST)+    , (MDB_NOTFOUND, #const MDB_NOTFOUND)+    , (MDB_PAGE_NOTFOUND, #const MDB_PAGE_NOTFOUND)+    , (MDB_CORRUPTED, #const MDB_CORRUPTED)+    , (MDB_PANIC, #const MDB_PANIC)+    , (MDB_VERSION_MISMATCH, #const MDB_VERSION_MISMATCH)+    , (MDB_INVALID, #const MDB_INVALID)+    , (MDB_MAP_FULL, #const MDB_MAP_FULL)+    , (MDB_DBS_FULL, #const MDB_DBS_FULL)+    , (MDB_READERS_FULL, #const MDB_READERS_FULL)+    , (MDB_TLS_FULL, #const MDB_TLS_FULL)+    , (MDB_TXN_FULL, #const MDB_TXN_FULL)+    , (MDB_CURSOR_FULL, #const MDB_CURSOR_FULL)+    , (MDB_PAGE_FULL, #const MDB_PAGE_FULL)+    , (MDB_MAP_RESIZED, #const MDB_MAP_RESIZED)+    , (MDB_INCOMPATIBLE, #const MDB_INCOMPATIBLE)+    , (MDB_BAD_RSLOT, #const MDB_BAD_RSLOT)+    , (MDB_BAD_TXN, #const MDB_BAD_TXN)+    , (MDB_BAD_VALSIZE, #const MDB_BAD_VALSIZE)+    , (MDB_BAD_DBI, #const MDB_BAD_DBI) ]++{-# INLINE numToErrVal #-}+numToErrVal :: Int -> Either Int MDB_ErrCode+numToErrVal code =+    case L.find ((== code) . snd) errCodes of+        Nothing -> Left code+        Just (ec,_) -> Right ec++{-# INLINE throwLMDBErrNum #-}+throwLMDBErrNum :: String -> CInt -> IO noReturn+throwLMDBErrNum context errNum = do+    desc <- peekCString =<< c_mdb_strerror errNum+    throwIO $! LMDB_Error+        { e_context = context+        , e_description = desc+        , e_code = numToErrVal (fromIntegral errNum) }++mdb_notfound :: CInt+mdb_notfound = #const MDB_NOTFOUND++mdb_rdonly :: CUInt+mdb_rdonly = #const MDB_RDONLY++mdb_notls :: CUInt+mdb_notls = #const MDB_NOTLS++mdb_nosubdir :: CUInt+mdb_nosubdir = #const MDB_NOSUBDIR++mdb_nooverwrite :: CUInt+mdb_nooverwrite = #const MDB_NOOVERWRITE++mdb_append :: CUInt+mdb_append = #const MDB_APPEND++mdb_create :: CUInt+mdb_create = #const MDB_CREATE++combineOptions :: [CUInt] -> CUInt+combineOptions = foldr (.|.) 0++mdb_first :: MDB_cursor_op_t+mdb_first = #const MDB_FIRST++mdb_next :: MDB_cursor_op_t+mdb_next = #const MDB_NEXT++mdb_env_create :: IO (Ptr MDB_env)+mdb_env_create = do+    alloca $ \ppenv -> c_mdb_env_create ppenv >>= \rc ->+        if rc /= 0 then throwLMDBErrNum "mdb_env_create" rc else peek ppenv++mdb_env_set_mapsize :: Ptr MDB_env -> Int -> IO ()+mdb_env_set_mapsize penv size =+    c_mdb_env_set_mapsize penv (fromIntegral size) >>= \rc ->+        when (rc /= 0) $ throwLMDBErrNum "mdb_env_set_mapsize" rc++mdb_env_set_maxdbs :: Ptr MDB_env -> Int -> IO ()+mdb_env_set_maxdbs penv num =+    c_mdb_env_set_maxdbs penv (fromIntegral num) >>= \rc ->+        when (rc /= 0) $ throwLMDBErrNum "mdb_env_set_maxdbs" rc++mdb_env_set_maxreaders :: Ptr MDB_env -> Int -> IO ()+mdb_env_set_maxreaders penv num =+    c_mdb_env_set_maxreaders penv (fromIntegral $ num) >>= \rc ->+        when (rc /= 0) $ throwLMDBErrNum "mdb_env_set_maxreaders" rc++mdb_env_open :: Ptr MDB_env -> FilePath -> CUInt -> IO ()+mdb_env_open penv path flags =+    withCString path $ \cpath ->+        c_mdb_env_open penv cpath flags 0o660 >>= \rc ->+            when (rc /= 0) $ throwLMDBErrNum "mdb_env_open" rc++mdb_txn_begin :: Ptr MDB_env -> Ptr MDB_txn -> CUInt -> IO (Ptr MDB_txn)+mdb_txn_begin penv parent flags =+    alloca $ \pptxn -> c_mdb_txn_begin penv parent flags pptxn >>= \rc ->+        if rc /= 0 then throwLMDBErrNum "mdb_txn_begin" rc else peek pptxn++-- If the commit fails, aborts the transaction.+mdb_txn_commit :: Ptr MDB_txn -> IO ()+mdb_txn_commit ptxn =+    c_mdb_txn_commit ptxn >>= \rc ->+        when (rc /= 0) $ c_mdb_txn_abort ptxn >> throwLMDBErrNum "mdb_txn_commit" rc++mdb_cursor_open :: Ptr MDB_txn -> MDB_dbi_t -> IO (Ptr MDB_cursor)+mdb_cursor_open ptxn dbi =+    alloca $ \ppcurs -> c_mdb_cursor_open ptxn dbi ppcurs >>= \rc ->+        if rc /= 0 then c_mdb_txn_abort ptxn >> throwLMDBErrNum "mdb_cursor_open" rc else peek ppcurs++mdb_dbi_open :: Ptr MDB_txn -> Maybe String -> CUInt -> IO MDB_dbi_t+mdb_dbi_open ptxn name flags = do+    withCStringMaybe name $ \cname ->+        alloca $ \pdbi -> c_mdb_dbi_open ptxn cname flags pdbi >>= \rc ->+            if rc /= 0 then c_mdb_txn_abort ptxn >> throwLMDBErrNum "mdb_dbi_open" rc else peek pdbi++{-# INLINE mdb_put #-}+mdb_put :: Ptr MDB_txn -> MDB_dbi_t -> Ptr MDB_val -> Ptr MDB_val -> CUInt -> IO ()+mdb_put ptxn dbi pk pv flags =+    c_mdb_put ptxn dbi pk pv flags >>= \rc ->+        when (rc /= 0) $ throwLMDBErrNum "mdb_put" rc++{-# INLINE mdb_put_ #-}+mdb_put_ :: Ptr MDB_txn -> MDB_dbi_t -> Ptr CChar -> CSize -> Ptr CChar -> CSize -> CUInt -> IO ()+mdb_put_ ptxn dbi pk ks pv vs flags =+    c_mdb_put_ ptxn dbi pk ks pv vs flags >>= \rc ->+        when (rc /= 0) $ throwLMDBErrNum "mdb_put_" rc++mdb_clear :: Ptr MDB_txn -> MDB_dbi_t -> IO ()+mdb_clear ptxn dbi =+    c_mdb_drop ptxn dbi 0 >>= \rc ->+        when (rc /= 0) $ throwLMDBErrNum "mdb_clear" rc++-- | Use a nullable CString.+withCStringMaybe :: Maybe String -> (CString -> IO a) -> IO a+withCStringMaybe Nothing f = f nullPtr+withCStringMaybe (Just s) f = withCString s f
+ src/Streamly/External/LMDB/Internal/streamly_lmdb_foreign.c view
@@ -0,0 +1,11 @@+#include "streamly_lmdb_foreign.h"++int mdb_put_(MDB_txn *txn, MDB_dbi dbi, char *key, size_t key_size, char *val, size_t val_size, unsigned int flags) {+    MDB_val k;+    k.mv_data = key;+    k.mv_size = key_size;+    MDB_val v;+    v.mv_data = val;+    v.mv_size = val_size;+    return mdb_put(txn, dbi, &k, &v, flags);+}
+ streamly-lmdb.cabal view
@@ -0,0 +1,73 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: 842ffd0abdf7b09639ffae5af90237140b800743e7e8c53744189c32edb85f39++name:           streamly-lmdb+version:        0.0.1+synopsis:       Stream data to or from LMDB databases using the streamly library.+description:    Please see the README on GitHub at <https://github.com/shlok/streamly-lmdb#readme>+category:       Database, Streaming, Streamly+homepage:       https://github.com/shlok/streamly-lmdb#readme+bug-reports:    https://github.com/shlok/streamly-lmdb/issues+author:         Shlok Datye+maintainer:     sd-haskell@quant.is+copyright:      2020 Shlok Datye+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    ChangeLog.md++source-repository head+  type: git+  location: https://github.com/shlok/streamly-lmdb++library+  exposed-modules:+      Streamly.External.LMDB+      Streamly.External.LMDB.Internal+      Streamly.External.LMDB.Internal.Foreign+  other-modules:+      Paths_streamly_lmdb+  hs-source-dirs:+      src+  ghc-options: -Wall+  c-sources:+      src/Streamly/External/LMDB/Internal/streamly_lmdb_foreign.c+  extra-libraries:+      lmdb+  build-depends:+      async >=2.2.2 && <2.3+    , base >=4.7 && <5+    , bytestring >=0.10.10.0 && <0.11+    , streamly >=0.7.2 && <0.8+  default-language: Haskell2010++test-suite streamly-lmdb-test+  type: exitcode-stdio-1.0+  main-is: TestSuite.hs+  other-modules:+      Streamly.External.LMDB.Tests+      Paths_streamly_lmdb+  hs-source-dirs:+      test+  ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N+  extra-libraries:+      lmdb+  build-depends:+      QuickCheck >=2.13.2 && <2.14+    , async >=2.2.2 && <2.3+    , base >=4.7 && <5+    , bytestring >=0.10.10.0 && <0.11+    , directory >=1.3.6.0 && <1.4+    , streamly >=0.7.2 && <0.8+    , streamly-lmdb ==0.0.1+    , tasty >=1.2.3 && <1.3+    , tasty-quickcheck >=0.10.1.1 && <0.11+    , temporary >=1.3 && <1.4+  default-language: Haskell2010
+ test/Streamly/External/LMDB/Tests.hs view
@@ -0,0 +1,148 @@+{-# LANGUAGE TypeApplications #-}++module Streamly.External.LMDB.Tests (tests) where++import Control.Concurrent.Async (asyncBound, wait)+import Control.Exception (SomeException, onException, try)+import Control.Monad (forM_)+import Data.ByteString (ByteString, pack)+import Data.ByteString.Unsafe (unsafeUseAsCStringLen)+import Data.List (find, foldl', nubBy, sort)+import Foreign (castPtr, nullPtr, with)+import Streamly.Prelude (fromList, toList, unfold)+import Test.QuickCheck.Monadic (PropertyM, monadicIO, pick, run)+import Test.Tasty (TestTree)+import Test.Tasty.QuickCheck (arbitrary, testProperty)++import qualified Data.ByteString as B+import qualified Streamly.Prelude as S++import Streamly.External.LMDB (Mode, ReadWrite, WriteOptions(..),+    clearDatabase, defaultWriteOptions, readLMDB, unsafeReadLMDB, writeLMDB)+import Streamly.External.LMDB.Internal (Database (..))+import Streamly.External.LMDB.Internal.Foreign (MDB_val (..), combineOptions,+    mdb_nooverwrite, mdb_put, mdb_txn_begin, mdb_txn_commit)++tests :: IO (Database ReadWrite) -> [TestTree]+tests db = [ testReadLMDB db, testUnsafeReadLMDB db, testWriteLMDB db, testWriteLMDB_2 db ]++-- | Clear the database, write key-value pairs to it in a normal manner, read+-- them back using our library, and make sure the result is what we wrote.+testReadLMDB :: (Mode mode) => IO (Database mode) -> TestTree+testReadLMDB res = testProperty "readLMDB" . monadicIO $ do+    db <- run res+    keyValuePairs <- arbitraryKeyValuePairs+    run $ clearDatabase db++    run $ writeChunk db False keyValuePairs+    let keyValuePairsInDb = sort . removeDuplicateKeys $ keyValuePairs++    let stream = unfold (readLMDB db) undefined+    readPairsAll <- run . toList $ stream+    let allAsExpected = readPairsAll == keyValuePairsInDb++    -- Also make sure the stream’s 'take' functionality works.+    readPairsFirstFew <- run . toList $ S.take 3 stream+    let firstFewAsExpected = readPairsFirstFew == take 3 keyValuePairsInDb++    return $ allAsExpected && firstFewAsExpected++-- | Similar to 'testReadLMDB', except that it tests the unsafe function in a different manner.+testUnsafeReadLMDB :: (Mode mode) => IO (Database mode) -> TestTree+testUnsafeReadLMDB res = testProperty "unsafeReadLMDB" . monadicIO $ do+    db <- run res+    keyValuePairs <- arbitraryKeyValuePairs+    run $ clearDatabase db++    run $ writeChunk db False keyValuePairs+    let lengthsInDb = map (\(k, v) -> (B.length k, B.length v)) . sort . removeDuplicateKeys $ keyValuePairs++    let stream = unfold (unsafeReadLMDB db (return . snd) (return . snd)) undefined+    readLengthsAll <- run . toList $ stream++    return $ readLengthsAll == lengthsInDb++-- | Clear the database, write key-value pairs to it using our library with key overwriting allowed, read+-- them back using our library (already covered by 'testReadLMDB'), and make sure the result is what we wrote.+testWriteLMDB :: IO (Database ReadWrite) -> TestTree+testWriteLMDB res = testProperty "writeLMDB" . monadicIO $ do+    db <- run res+    keyValuePairs <- arbitraryKeyValuePairs+    run $ clearDatabase db++    chunkSz <- pick arbitrary++    let fol' = writeLMDB db $ defaultWriteOptions { writeTransactionSize = chunkSz, noOverwrite = False }++    -- TODO: Run with new "bound" functionality in streamly.+    run $ asyncBound (S.fold fol' (fromList keyValuePairs)) >>= wait+    let keyValuePairsInDb = sort . removeDuplicateKeys $ keyValuePairs++    readPairsAll <- run . toList $ unfold (readLMDB db) undefined++    return $ keyValuePairsInDb == readPairsAll++-- | Clear the database, write key-value pairs to it using our library with key overwriting+-- disallowed, and make sure an exception occurs iff we had a duplicate key in our pairs.+-- Furthermore make sure that key-value pairs prior to a duplicate key are actually in the database.+testWriteLMDB_2 :: IO (Database ReadWrite) -> TestTree+testWriteLMDB_2 res = testProperty "writeLMDB_2" . monadicIO $ do+    db <- run res+    keyValuePairs <- arbitraryKeyValuePairs+    run $ clearDatabase db++    chunkSz <- pick arbitrary++    -- TODO: Run with new "bound" functionality in streamly.+    let fol' = writeLMDB db $ defaultWriteOptions { writeTransactionSize = chunkSz, noOverwrite = True }+    e <- run $ try @SomeException $ (asyncBound (S.fold fol' (fromList keyValuePairs)) >>= wait)+    exceptionAsExpected <-+        case e of+            Left _ ->+                -- This is somewhat rare but will be encountered for enough number of tests (e.g., 1000 instead of 100).+                return $ hasDuplicateKeys keyValuePairs+            Right _ -> return . not $ hasDuplicateKeys keyValuePairs++    let keyValuePairsInDb = sort . prefixBeforeDuplicate  $ keyValuePairs+    readPairsAll <- run . toList $ unfold (readLMDB db) undefined+    let pairsAsExpected = keyValuePairsInDb == readPairsAll++    return $ exceptionAsExpected && pairsAsExpected++arbitraryKeyValuePairs :: PropertyM IO [(ByteString, ByteString)]+arbitraryKeyValuePairs =+    map (\(ws1, ws2) -> (pack ws1, pack ws2))+    . filter (\(ws1, _) -> not (null ws1)) -- LMDB does not allow empty keys.+   <$> pick arbitrary++-- | Note that this function retains the last value for each key.+removeDuplicateKeys :: (Eq a) => [(a, b)] -> [(a, b)]+removeDuplicateKeys = foldl' (\acc (a, b) -> if any ((== a) . fst) acc then acc else (a, b) : acc) [] . reverse++hasDuplicateKeys :: (Eq a) => [(a, b)] -> Bool+hasDuplicateKeys l =+    let l2 = nubBy (\(a1, _) (a2, _) -> a1 == a2) l+     in length l /= length l2++prefixBeforeDuplicate :: (Eq a) => [(a, b)] -> [(a, b)]+prefixBeforeDuplicate xs =+    let fstDup = snd <$> find (\((a, _), i) -> a `elem` map fst (take i xs)) (zip xs [0..])+    in case fstDup of+        Nothing -> xs+        Just i -> take i xs++-- Writes the given key-value pairs to the given database.+writeChunk :: (Foldable t, Mode mode) => Database mode -> Bool -> t (ByteString, ByteString) -> IO ()+writeChunk (Database penv dbi) noOverwrite' keyValuePairs =+    let flags = combineOptions $ [mdb_nooverwrite | noOverwrite']+    in asyncBound (do+        ptxn <- mdb_txn_begin penv nullPtr 0+        onException (forM_ keyValuePairs $ \(k, v) ->+            marshalOut k $ \k' -> marshalOut v $ \v' -> with k' $ \k'' -> with v' $ \v'' ->+                mdb_put ptxn dbi k'' v'' flags)+            (mdb_txn_commit ptxn) -- Make sure the key-value pairs we have so far are committed.+        mdb_txn_commit ptxn) >>= wait++{-# INLINE marshalOut #-}+marshalOut :: ByteString -> (MDB_val -> IO ()) -> IO ()+marshalOut bs f = unsafeUseAsCStringLen bs $ \(ptr, len) -> f $ MDB_val (fromIntegral len) (castPtr ptr)
+ test/TestSuite.hs view
@@ -0,0 +1,27 @@+module Main where++import System.Directory (removeDirectoryRecursive)+import System.Environment (setEnv)+import System.IO.Temp (createTempDirectory, getCanonicalTemporaryDirectory)+import Test.Tasty (TestTree, defaultMain, testGroup, withResource)++import Streamly.External.LMDB (Database, ReadWrite, defaultLimits, getDatabase, mapSize, openEnvironment, tebibyte)+import qualified Streamly.External.LMDB.Tests (tests)++main :: IO ()+main = do+    setEnv "TASTY_NUM_THREADS" "1" -- Multiple tests use the same LMDB database.+    defaultMain $ withResource+        (do tmpParent <- getCanonicalTemporaryDirectory+            tmpDir <- createTempDirectory tmpParent "streamly-lmdb-tests"+            env <- openEnvironment tmpDir $ defaultLimits { mapSize = tebibyte }+            db <- getDatabase env Nothing+            return (tmpDir, db))+        (\(tmpDir, _) -> removeDirectoryRecursive tmpDir)+        (\io -> tests $ snd <$> io)++tests :: IO (Database ReadWrite) -> TestTree+tests res =+    testGroup "Tests"+        [ testGroup "Streamly.External.LMDB.Tests" $+                     Streamly.External.LMDB.Tests.tests res ]