haskoin-store 0.49.0 → 0.50.0

raw patch · 6 files changed

+89/−23 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +4/−0
• app/Main.hs +16/−0
• haskoin-store.cabal +2/−2
• src/Haskoin/Store/Cache.hs +53/−13
• src/Haskoin/Store/Manager.hs +13/−8
• test/Haskoin/StoreSpec.hs +1/−0

@@ -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.

@@ -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

@@ -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 <https://github.com/haskoin/haskoin-store#readme> category:       Bitcoin, Finance, Network

@@ -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) 

@@ -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

@@ -99,6 +99,7 @@               , storeConfCacheRefresh = 750               , storeConfCacheRetryDelay = 100000               , storeConfStats = Nothing+              , storeConfCacheThreads = 8               }     withStore cfg $ \Store {..} ->         withSubscription storePublisher $ \sub ->