rocksdb-haskell-jprupp 2.5.0 → 2.6.0

raw patch · 6 files changed

+206/−12 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +15/−0
• rocksdb-haskell-jprupp.cabal +1/−1
• src/Database/RocksDB/Base.hs +58/−3
• src/Database/RocksDB/C.hs +39/−0
• src/Database/RocksDB/Internal.hs +51/−8
• test/Spec.hs +42/−0

@@ -1,3 +1,18 @@+## 2.6.0++ * Add `multiGet` and `multiGetCF`, which read the values of many keys in+   one call to RocksDB's `MultiGet`.+ * Add `blockCacheSize` to `Config`: the bytes of a least-recently-used block+   cache shared by the database and all its column families. Without it+   RocksDB gives each column family its own default cache, of 32 MiB in+   RocksDB 8.+ * Add `cacheIndexAndFilterBlocks` to `Config`, to keep the index and filter+   blocks of a column family in the block cache, at high priority, with+   those of level 0 pinned, instead of in memory outside the cache for every+   open file.+ * Stop leaking the block-based table options of every database and column+   family opened with a Bloom filter.+ ## 2.5.0   * Add the `EstimateTableReadersMem`, `CurSizeAllMemTables` and

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           rocksdb-haskell-jprupp-version:        2.5.0+version:        2.6.0 synopsis:       Haskell bindings for RocksDB description:    See README at <https://github.com/jprupp/rocksdb-haskell#readme> category:       Database, FFI

@@ -34,6 +34,8 @@     , write     , get     , getCF+    , multiGet+    , multiGetCF     , withSnapshot     , createSnapshot     , releaseSnapshot@@ -62,6 +64,7 @@ import           Foreign.ForeignPtr import           Foreign.Marshal.Alloc import           Foreign.Marshal.Array+import           Foreign.Marshal.Utils     (withMany) import           Foreign.Ptr import           Foreign.Storable import           System.Directory@@ -94,7 +97,8 @@ -- when the function exits. withDB :: FilePath -> Config -> (DB -> IO a) -> IO a withDB path config f =-    withOptions config $ \opts_ptr ->+    withBlockCache config $ \cache ->+    withOptionsCache cache config $ \opts_ptr ->     withReadOpts Nothing $ \read_opts ->     withWriteOpts $ \write_opts ->     bracket (create_db opts_ptr read_opts write_opts) destroy_db f@@ -118,8 +122,9 @@          -> (DB -> IO a)          -> IO a withDBCF path config cf_cfgs f =-    withOptions config $ \opts_ptr ->-    withOptionsCF (map snd cf_cfgs) $ \cf_opts ->+    withBlockCache config $ \cache ->+    withOptionsCache cache config $ \opts_ptr ->+    withOptionsCFCache cache (map snd cf_cfgs) $ \cf_opts ->     withReadOpts Nothing $ \read_opts ->     withWriteOpts $ \write_opts ->     withStrings (map fst cf_cfgs) $ \cf_names ->@@ -324,6 +329,56 @@                 vlen <- peek vlen_ptr                 res <- BU.unsafePackMallocCStringLen (val_ptr, cSizeToInt vlen)                 return $ Just res++-- | Read the values of many keys in one call. RocksDB looks the keys up+-- together, sharing the work of finding the files and blocks that hold+-- them, which is faster than reading them one by one. The values are in the+-- order of the keys.+multiGet :: DB -> [ByteString] -> IO [Maybe ByteString]+multiGet db = multiGetCommon db Nothing++-- | 'multiGet' on a column family.+multiGetCF :: DB -> ColumnFamily -> [ByteString] -> IO [Maybe ByteString]+multiGetCF db cf = multiGetCommon db (Just cf)++multiGetCommon :: DB -> Maybe ColumnFamily -> [ByteString] -> IO [Maybe ByteString]+multiGetCommon _ _ [] = return []+multiGetCommon DB{rocksDB = db_ptr, readOpts = read_opts} mcf keys =+    withMany BU.unsafeUseAsCStringLen keys $ \kls ->+    withArrayLen (map fst kls) $ \n key_ptrs ->+    withArray (map (intToCSize . snd) kls) $ \key_lens ->+    allocaArray n $ \val_ptrs ->+    allocaArray n $ \val_lens ->+    allocaArray n $ \err_ptrs -> do+        case mcf of+            Just cf ->+                withArray (replicate n cf) $ \cfs ->+                c_rocksdb_multi_get_cf+                    db_ptr read_opts cfs (intToCSize n)+                    key_ptrs key_lens val_ptrs val_lens err_ptrs+            Nothing ->+                c_rocksdb_multi_get+                    db_ptr read_opts (intToCSize n)+                    key_ptrs key_lens val_ptrs val_lens err_ptrs+        vals <- peekArray n val_ptrs+        lens <- peekArray n val_lens+        errs <- peekArray n err_ptrs+        -- Own every value and free every error before throwing, so that+        -- none of them leaks.+        res <- forM (zip vals lens) $ \(v, l) ->+            if v == nullPtr+                then return Nothing+                else Just <$> BU.unsafePackMallocCStringLen (v, cSizeToInt l)+        msgs <- forM errs $ \e ->+            if e == nullPtr+                then return Nothing+                else do+                    m <- peekCString e+                    free e+                    return (Just m)+        case [m | Just m <- msgs] of+            m : _ -> throwIO $ userError $ "multi_get: " ++ m+            [] -> return res  delete :: DB -> ByteString -> IO () delete db = deleteCommon db Nothing

@@ -21,6 +21,7 @@     , Iterator     , WriteBatch     , FilterPolicy+    , Cache     , ErrPtr     , DBName     , CFName@@ -82,6 +83,12 @@     , c_rocksdb_block_based_options_create     , c_rocksdb_block_based_options_set_filter_policy     , c_rocksdb_block_based_options_destroy+    , c_rocksdb_block_based_options_set_block_cache+    , c_rocksdb_block_based_options_set_cache_index_and_filter_blocks+    , c_rocksdb_block_based_options_set_cache_index_and_filter_blocks_with_high_priority+    , c_rocksdb_block_based_options_set_pin_l0_filter_and_index_blocks_in_cache+    , c_rocksdb_cache_create_lru+    , c_rocksdb_cache_destroy     , c_rocksdb_slicetransform_create_fixed_prefix     , c_rocksdb_slicetransform_destroy     , c_rocksdb_filterpolicy_destroy@@ -109,6 +116,7 @@ data LWriteOpts data LFilterPolicy data LSliceTransform+data LCache  type RocksDB            = Ptr LRocksDB type ColumnFamily       = Ptr LColumnFamily@@ -121,6 +129,7 @@ type Iterator           = Ptr LIterator type FilterPolicy       = Ptr LFilterPolicy type BlockBasedOptions  = Ptr LBlockBasedOptions+type Cache              = Ptr LCache  type ErrPtr           = Ptr CString type DBName           = CString@@ -412,10 +421,40 @@                                                   -> FilterPolicy                                                   -> IO () +foreign import ccall safe "rocksdb/c.h rocksdb_block_based_options_set_block_cache"+  c_rocksdb_block_based_options_set_block_cache :: BlockBasedOptions+                                                -> Cache+                                                -> IO ()++foreign import ccall safe "rocksdb/c.h rocksdb_block_based_options_set_cache_index_and_filter_blocks"+  c_rocksdb_block_based_options_set_cache_index_and_filter_blocks :: BlockBasedOptions+                                                                  -> CUChar+                                                                  -> IO ()++foreign import ccall safe "rocksdb/c.h rocksdb_block_based_options_set_cache_index_and_filter_blocks_with_high_priority"+  c_rocksdb_block_based_options_set_cache_index_and_filter_blocks_with_high_priority :: BlockBasedOptions+                                                                                     -> CUChar+                                                                                     -> IO ()++foreign import ccall safe "rocksdb/c.h rocksdb_block_based_options_set_pin_l0_filter_and_index_blocks_in_cache"+  c_rocksdb_block_based_options_set_pin_l0_filter_and_index_blocks_in_cache :: BlockBasedOptions+                                                                            -> CUChar+                                                                            -> IO ()+ foreign import ccall safe "rocksdb/c.h rocksdb_options_set_block_based_table_factory"   c_rocksdb_options_set_block_based_table_factory :: Options                                                   -> BlockBasedOptions                                                   -> IO ()++--+-- Block Cache+--++foreign import ccall safe "rocksdb/c.h rocksdb_cache_create_lru"+  c_rocksdb_cache_create_lru :: CSize -> IO Cache++foreign import ccall safe "rocksdb/c.h rocksdb_cache_destroy"+  c_rocksdb_cache_destroy :: Cache -> IO ()  -- -- Prefix Extractor

@@ -15,7 +15,10 @@      -- * Smart constructors & extractors     , withOptions+    , withOptionsCache     , withOptionsCF+    , withOptionsCFCache+    , withBlockCache     , withReadOpts     , createReadOpts     , destroyReadOpts@@ -34,6 +37,7 @@ import           Control.Exception import           Control.Monad import           Data.Default+import           Data.Maybe        (isJust) import           Database.RocksDB.C import           Foreign import           Foreign.C.String@@ -61,6 +65,17 @@                        -- ^ bytes for all memtables across all column                        -- families (RocksDB default unlimited); database-wide,                        -- ignored in column family configs+                     , blockCacheSize :: !(Maybe Int)+                       -- ^ bytes of a least-recently-used block cache shared+                       -- by the database and all its column families+                       -- ('Nothing': RocksDB gives each column family its+                       -- own default cache, of 32 MiB in RocksDB 8);+                       -- database-wide, ignored in column family configs+                     , cacheIndexAndFilterBlocks :: !Bool+                       -- ^ keep index and filter blocks in the block cache,+                       -- at high priority, with those of level 0 pinned,+                       -- instead of in memory outside it for every open+                       -- file; per column family                      , maxTotalWalSize :: !(Maybe Int)                        -- ^ bytes of write-ahead log after which the column                        -- families holding the oldest log are flushed, so it@@ -79,16 +94,31 @@                  , bloomFilter      = False                  , writeBufferSize  = Nothing                  , dbWriteBufferSize = Nothing+                 , blockCacheSize   = Nothing+                 , cacheIndexAndFilterBlocks = False                  , maxTotalWalSize  = Nothing                  }  withOptions :: Config -> (Options -> IO a) -> IO a-withOptions Config {..} f = with_opts $ \opts -> do-    when bloomFilter $ do-        fp <- c_rocksdb_filterpolicy_create_bloom_full 10-        bo <- c_rocksdb_block_based_options_create-        c_rocksdb_block_based_options_set_filter_policy bo fp-        c_rocksdb_options_set_block_based_table_factory opts bo+withOptions = withOptionsCache Nothing++-- | Options of a database or column family that use this block cache, if+-- any.+withOptionsCache :: Maybe Cache -> Config -> (Options -> IO a) -> IO a+withOptionsCache cache Config {..} f = with_opts $ \opts -> do+    when (bloomFilter || isJust cache || cacheIndexAndFilterBlocks) $+        bracket c_rocksdb_block_based_options_create+                c_rocksdb_block_based_options_destroy $ \bo -> do+            when bloomFilter $ do+                fp <- c_rocksdb_filterpolicy_create_bloom_full 10+                c_rocksdb_block_based_options_set_filter_policy bo fp+            forM_ cache $ c_rocksdb_block_based_options_set_block_cache bo+            when cacheIndexAndFilterBlocks $ do+                c_rocksdb_block_based_options_set_cache_index_and_filter_blocks bo 1+                c_rocksdb_block_based_options_set_cache_index_and_filter_blocks_with_high_priority bo 1+                c_rocksdb_block_based_options_set_pin_l0_filter_and_index_blocks_in_cache bo 1+            -- the table factory copies the options, filter policy and cache+            c_rocksdb_options_set_block_based_table_factory opts bo     forM_ prefixLength $ \l -> do         t <- c_rocksdb_slicetransform_create_fixed_prefix (intToCSize l)         c_rocksdb_options_set_prefix_extractor opts t@@ -117,11 +147,24 @@   withOptionsCF :: [Config] -> ([Options] -> IO a) -> IO a-withOptionsCF cfgs f =+withOptionsCF = withOptionsCFCache Nothing++-- | Options of column families that use this block cache, if any.+withOptionsCFCache :: Maybe Cache -> [Config] -> ([Options] -> IO a) -> IO a+withOptionsCFCache cache cfgs f =     go [] cfgs   where     go acc [] = f (reverse acc)-    go acc (c:cs) = withOptions c $ \o -> go (o:acc) cs+    go acc (c:cs) = withOptionsCache cache c $ \o -> go (o:acc) cs++-- | The block cache of 'blockCacheSize', if it is set. The database and+-- column families that use it keep it alive after this returns.+withBlockCache :: Config -> (Maybe Cache -> IO a) -> IO a+withBlockCache Config {blockCacheSize = Nothing} f = f Nothing+withBlockCache Config {blockCacheSize = Just n} f =+    bracket (c_rocksdb_cache_create_lru (intToCSize n))+            c_rocksdb_cache_destroy+            (f . Just)  withReadOpts :: Maybe Snapshot -> (ReadOpts -> IO a) -> IO a withReadOpts maybe_snap_ptr =

@@ -43,6 +43,20 @@         putCF db two "two_key" "two_value"         getCF db two "two_key" `shouldReturn` Just "two_value"         get db "two_key" `shouldReturn` Nothing+    describe "Multi-get" $ do+      it "reads many keys at once, in order, with missing ones" $ \db -> do+        let one = head (columnFamilies db)+        forM_ [1 .. 100 :: Int] $ \i -> do+          put db (C.pack (show i)) (C.pack ("v" <> show i))+          putCF db one (C.pack (show i)) (C.pack ("one" <> show i))+        let keys = map (C.pack . show) [50, 0, 1, 100, 101, 50 :: Int]+        multiGet db keys+          `shouldReturn` [Just "v50", Nothing, Just "v1", Just "v100", Nothing, Just "v50"]+        multiGetCF db one keys+          `shouldReturn` [Just "one50", Nothing, Just "one1", Just "one100", Nothing, Just "one50"]+        multiGet db [] `shouldReturn` []+        ks <- forM [1 .. 2000 :: Int] $ \i -> return (C.pack (show (i `mod` 150)))+        multiGet db ks >>= (`shouldBe` map (\k -> if read (C.unpack k) `elem` [1 .. 100 :: Int] then Just ("v" <> k) else Nothing) ks)     describe "Multithreading" $ do       it "stores and retrieve items from multiple threads" $ \db -> do         let key i = C.pack $ printf "key_%04d" i@@ -264,6 +278,34 @@           >>= (`shouldSatisfy` isNumber)         getProperty db (PropertyName "rocksdb.no-such-property")           `shouldReturn` Nothing+      it "shares one block cache, holding index and filter blocks" $ \_ -> do+        let size = 8 * 1024 * 1024+            cached =+              conf+                { errorIfExists = False,+                  blockCacheSize = Just size,+                  cacheIndexAndFilterBlocks = True,+                  writeBufferSize = Just (64 * 1024)+                }+            key :: Int -> C.ByteString+            key i = C.pack (printf "k%07d" i)+            value = C.replicate 100 'v'+            capacity d cf =+              maybe (getProperty d p) (\c -> getPropertyCF d c p) cf+              where+                p = PropertyName "rocksdb.block-cache-capacity"+        withSystemTempDirectory "rocksdb-tests-cache" $ \path -> do+          withDBCF path cached [("one", cached)] $ \db -> do+            let one = head (columnFamilies db)+            forM_ [1 .. 20000] $ \i -> putCF db one (key i) value+          withDBCF path cached [("one", cached)] $ \db -> do+            let one = head (columnFamilies db)+            capacity db Nothing `shouldReturn` Just (C.pack (show size))+            capacity db (Just one) `shouldReturn` Just (C.pack (show size))+            multiGetCF db one (map key [1, 10000, 20000])+              `shouldReturn` replicate 3 (Just value)+            usage <- getProperty db BlockCacheUsage+            usage `shouldSatisfy` maybe False ((> 0) . (read :: String -> Int) . C.unpack)       it "opens with a bound on the write-ahead log" $ \_ ->         withSystemTempDirectory "rocksdb-tests-wal" $ \path ->           withDBCF path conf {maxTotalWalSize = Just (1024 * 1024)} [("one", conf)] $ \db -> do