diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html) +## 0.50.0 +### Added +- Limit number of simultaneous xpubs being cached. + ## 0.49.0 ### Changed - Improve conduit merging algorithm. diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -82,6 +82,7 @@ , configWebRequests :: !Int , configWebReqTimeout :: !Int , configWebPriceGet :: !Int + , configCacheThreads :: !Int } instance Default Config where @@ -116,6 +117,7 @@ , configWebRequests = defWebRequests , configWebReqTimeout = defWebReqTimeout , configWebPriceGet = defWebPriceGet + , configCacheThreads = defCacheThreads } defEnv :: MonadIO m => String -> a -> (String -> Maybe a) -> m a @@ -275,6 +277,11 @@ defEnv "WEB_REQUESTS" 32 readMaybe {-# NOINLINE defWebRequests #-} +defCacheThreads :: Int +defCacheThreads = unsafePerformIO $ + defEnv "CACHE_THREADS" 8 readMaybe +{-# NOINLINE defCacheThreads #-} + defWebReqTimeout :: Int defWebReqTimeout = unsafePerformIO $ defEnv "WEB_REQ_TIMEOUT" (45 * 1000 * 1000) readMaybe @@ -557,6 +564,13 @@ <> help "How often to retrieve price information" <> showDefault <> value (configWebPriceGet def) + configCacheThreads <- + option auto $ + metavar "INT" + <> long "cache-threads" + <> help "Number of simultaneous xpub caching threads" + <> showDefault + <> value (configCacheThreads def) pure Config { configWebLimits = WebLimits {..} @@ -636,6 +650,7 @@ , configWebRequests = wreqs , configWebReqTimeout = wtimeout , configWebPriceGet = wpget + , configCacheThreads = cth } = runStderrLoggingT . filterLogger l . with_stats $ \stats -> do $(logInfoS) "Main" $ @@ -665,6 +680,7 @@ , storeConfCacheRefresh = crefresh , storeConfCacheRetryDelay = cretrydelay , storeConfStats = stats + , storeConfCacheThreads = cth } withStore scfg $ \st -> runWeb diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: a206136ca02c8f998e0567b78de9de9f7b2d2a99ab278d1d62f5c1b193b9e2e2 +-- hash: 5b36767c6735c36f43dfb4f75547a52955c10cde6bd527c536bdc387d03d6482 name: haskoin-store -version: 0.49.0 +version: 0.50.0 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network diff --git a/src/Haskoin/Store/Cache.hs b/src/Haskoin/Store/Cache.hs --- a/src/Haskoin/Store/Cache.hs +++ b/src/Haskoin/Store/Cache.hs @@ -15,6 +15,7 @@ , CacheT , CacheError(..) , newCacheMetrics + , newCacheThreads , withCache , connectRedis , blockRefScore @@ -93,8 +94,9 @@ import System.Random (randomIO, randomRIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO, TQueue, TVar, async, atomically, - bracket, liftIO, link, readTQueue, - readTVar, throwIO, withAsync, + bracket, liftIO, link, modifyTVar, + newTVarIO, readTQueue, readTVar, + throwIO, wait, withAsync, writeTQueue, writeTVar) import UnliftIO.Concurrent (threadDelay) @@ -115,8 +117,14 @@ , cacheRefresh :: !Int -- millisenconds , cacheRetryDelay :: !Int -- microseconds , cacheMetrics :: !(Maybe CacheMetrics) + , cacheThreads :: !CacheThreads } +type CacheThreads = TVar Int + +newCacheThreads :: MonadIO m => Int -> m CacheThreads +newCacheThreads = newTVarIO + data CacheMetrics = CacheMetrics { cacheHits :: !Metrics.Counter , cacheMisses :: !Metrics.Counter @@ -226,7 +234,7 @@ Just cfg -> lift (runReaderT (getXPubTxs xpub limits) cfg) xPubTxCount xpub = ask >>= \case - Nothing -> lift (xPubTxCount xpub) + Nothing -> lift (xPubTxCount xpub) Just cfg -> lift (runReaderT (getXPubTxCount xpub) cfg) getMaxGap = lift getMaxGap getInitialGap = lift getInitialGap @@ -244,6 +252,9 @@ utxoPfx :: ByteString utxoPfx = "u" +idxPfx :: ByteString +idxPfx = "i" + getXPubTxs :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => XPubSpec -> Limits -> CacheX m [TxRef] @@ -679,21 +690,21 @@ newXPubC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => XPubSpec -> [XPubBal] -> CacheX m () -newXPubC xpub bals = - should_index >>= \i -> +newXPubC xpub bals = should_index >>= \i -> if i then do - incrementCounter cacheMisses - withMetrics cacheIndexTime index + a <- async $ do + incrementCounter cacheMisses + withMetrics cacheIndexTime index + wait a else incrementCounter cacheIgnore where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) - should_index = do - x <- asks cacheMin + should_index = asks cacheMin >>= \x -> if x <= lenNotNull bals - then inSync - else return False - index = do + then is_indexing >>= \y -> if y then return False else inSync + else return False + index = bracket set_index unset_index $ \(x, y) -> when (x && y) $ do xpubtxt <- xpubText xpub $(logDebugS) "Cache" $ "Caching " <> xpubtxt <> ": " <> cs (show (length bals)) <> @@ -715,6 +726,35 @@ e <- redisAddXPubTxs xpub xtxs return $ b >> c >> d >> e >> return () $(logDebugS) "Cache" $ "Cached " <> xpubtxt + is_indexing = runRedis (Redis.exists key) + key = idxPfx <> encode xpub + get_thread = do + v <- asks cacheThreads + atomically $ do + x <- readTVar v + if x <= 0 + then return False + else modifyTVar v (subtract 1) >> return True + put_thread = do + v <- asks cacheThreads + atomically $ modifyTVar v (+1) + unset_index (x, y) = do + when x put_thread + when y . void . runRedis $ Redis.del [key] + set_index = + let opts = Redis.SetOpts + { Redis.setSeconds = Just 600 + , Redis.setMilliseconds = Nothing + , Redis.setCondition = Just Redis.Nx + } + red = Redis.setOpts key "1" opts + in get_thread >>= \case + True -> do + conn <- asks cacheConn + liftIO (Redis.runRedis conn red) >>= \case + Right _ -> return (True, True) + Left _ -> return (True, False) + False -> return (False, False) inSync :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m Bool @@ -1282,7 +1322,7 @@ net <- lift getNetwork let suffix = case xPubDeriveType xpub of DeriveNormal -> "" - DeriveP2SH -> "/p2sh" + DeriveP2SH -> "/p2sh" DeriveP2WPKH -> "/p2wpkh" return . cs $ suffix <> xPubExport net (xPubSpecKey xpub) diff --git a/src/Haskoin/Store/Manager.hs b/src/Haskoin/Store/Manager.hs --- a/src/Haskoin/Store/Manager.hs +++ b/src/Haskoin/Store/Manager.hs @@ -36,16 +36,17 @@ blockStoreTxSTM, withBlockStore) import Haskoin.Store.Cache (CacheConfig (..), CacheWriter, cacheNewBlock, cacheWriter, - connectRedis, newCacheMetrics) + connectRedis, newCacheMetrics, + newCacheThreads) import Haskoin.Store.Common (StoreEvent (..)) import Haskoin.Store.Database.Reader (DatabaseReader (..), DatabaseReaderT, withDatabaseReader) -import Network.Socket (SockAddr (..)) import NQE (Inbox, Process (..), Publisher, publishSTM, receive, withProcess, withPublisher, withSubscription) +import Network.Socket (SockAddr (..)) import qualified System.Metrics as Metrics (Store) import UnliftIO (MonadIO, MonadUnliftIO, STM, atomically, link, withAsync) @@ -102,6 +103,8 @@ -- ^ delay in microseconds to retry getting cache lock , storeConfStats :: !(Maybe Metrics.Store) -- ^ stats store + , storeConfCacheThreads :: !Int + -- ^ how many caching threads } withStore :: (MonadLoggerIO m, MonadUnliftIO m) @@ -187,14 +190,15 @@ mapM newCacheMetrics (storeConfStats cfg) >>= \metrics -> connectRedis redisurl >>= \conn -> withSubscription pub $ \evts -> - withProcess (f conn metrics) $ \p -> - cacheWriterProcesses crefresh evts (getProcessMailbox p) $ - action (Just (c conn metrics)) + newCacheThreads (storeConfCacheThreads cfg) >>= \cth -> + let conf = c conn metrics cth + in withProcess (f conf) $ \p -> + cacheWriterProcesses crefresh evts (getProcessMailbox p) $ do + action (Just conf) where crefresh = storeConfCacheRefresh cfg - f conn metrics cwinbox = - runReaderT (cacheWriter (c conn metrics) cwinbox) db - c conn metrics = + f conf cwinbox = runReaderT (cacheWriter conf cwinbox) db + c conn metrics cth = CacheConfig { cacheConn = conn , cacheMin = storeConfCacheMin cfg @@ -203,6 +207,7 @@ , cacheRefresh = storeConfCacheRefresh cfg , cacheRetryDelay = storeConfCacheRetryDelay cfg , cacheMetrics = metrics + , cacheThreads = cth } cacheWriterProcesses :: MonadUnliftIO m diff --git a/test/Haskoin/StoreSpec.hs b/test/Haskoin/StoreSpec.hs --- a/test/Haskoin/StoreSpec.hs +++ b/test/Haskoin/StoreSpec.hs @@ -99,6 +99,7 @@ , storeConfCacheRefresh = 750 , storeConfCacheRetryDelay = 100000 , storeConfStats = Nothing + , storeConfCacheThreads = 8 } withStore cfg $ \Store {..} -> withSubscription storePublisher $ \sub ->