diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/rocksdb-haskell-jprupp.cabal b/rocksdb-haskell-jprupp.cabal --- a/rocksdb-haskell-jprupp.cabal +++ b/rocksdb-haskell-jprupp.cabal @@ -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 category: Database, FFI diff --git a/src/Database/RocksDB/Base.hs b/src/Database/RocksDB/Base.hs --- a/src/Database/RocksDB/Base.hs +++ b/src/Database/RocksDB/Base.hs @@ -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 diff --git a/src/Database/RocksDB/C.hs b/src/Database/RocksDB/C.hs --- a/src/Database/RocksDB/C.hs +++ b/src/Database/RocksDB/C.hs @@ -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 diff --git a/src/Database/RocksDB/Internal.hs b/src/Database/RocksDB/Internal.hs --- a/src/Database/RocksDB/Internal.hs +++ b/src/Database/RocksDB/Internal.hs @@ -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 = diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -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