haskoin-store 0.40.8 → 0.40.9
raw patch · 3 files changed
+70/−40 lines, 3 filesPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
API changes (from Hackage documentation)
- Haskoin.Store.Cache: cachePing :: MonadIO m => CacheWriter -> m ()
Files
- haskoin-store.cabal +1/−1
- src/Haskoin/Store/Cache.hs +66/−31
- src/Haskoin/Store/Manager.hs +3/−8
haskoin-store.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: haskoin-store-version: 0.40.8+version: 0.40.9 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
src/Haskoin/Store/Cache.hs view
@@ -18,7 +18,6 @@ , CacheWriter , CacheWriterInbox , cacheNewBlock- , cachePing , cacheWriter , isInCache , evictFromCache@@ -72,10 +71,14 @@ chainGetBlock) import Haskoin.Store.Common import Haskoin.Store.Data-import NQE (Inbox, Mailbox, receive, send)-import System.Random (randomIO)+import NQE (Inbox, Listen, Mailbox,+ inboxToMailbox, query, receive,+ send)+import System.Random (randomIO, randomRIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO,- bracket, liftIO, throwIO)+ atomically, bracket, liftIO, link,+ throwIO, withAsync)+import UnliftIO.Concurrent (threadDelay) runRedis :: MonadLoggerIO m => Redis (Either Reply a) -> CacheX m a runRedis action =@@ -405,7 +408,7 @@ data CacheWriterMessage = CacheNewBlock- | CachePing+ | CachePing (Listen ()) type CacheWriterInbox = Inbox CacheWriterMessage type CacheWriter = Mailbox CacheWriterMessage@@ -436,8 +439,13 @@ cacheWriter :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheConfig -> CacheWriterInbox -> m ()-cacheWriter cfg inbox = runReaderT go cfg+cacheWriter cfg inbox =+ withAsync ping $ \a ->+ link a >> runReaderT go cfg where+ ping = forever $ do+ threadDelay (cacheRefresh cfg * 1000)+ cachePing (inboxToMailbox inbox) go = do newBlockC forever $ do@@ -507,6 +515,28 @@ Just _ -> Just <$> f Nothing -> return Nothing +smallDelay :: MonadUnliftIO m => m ()+smallDelay = threadDelay =<< liftIO (randomRIO (50 * 1000, 250 * 1000))++withLockForever+ :: (MonadLoggerIO m, MonadUnliftIO m)+ => CacheX m a+ -> CacheX m a+withLockForever go = withLock go >>= \case+ Nothing -> smallDelay >> go+ Just x -> return x++withLockRetry+ :: (MonadLoggerIO m, MonadUnliftIO m)+ => Integer+ -> CacheX m a+ -> CacheX m (Maybe a)+withLockRetry i f+ | i <= 0 = return Nothing+ | otherwise = withLock f >>= \case+ Nothing -> smallDelay >> withLockRetry (i - 1) f+ Just x -> return (Just x)+ pruneDB :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) => CacheX m Integer pruneDB = do@@ -546,12 +576,13 @@ when s $ do newBlockC syncMempoolC-cacheWriterReact CachePing =+cacheWriterReact (CachePing respond) = inSync >>= \s -> when s $ do pruneDB newBlockC syncMempoolC+ atomically $ respond () lenNotNull :: [XPubBal] -> Int lenNotNull bals = length $ filter (not . nullBalance . xPubBal) bals@@ -613,7 +644,7 @@ => CacheX m () newBlockC = lift getBestBlock >>= \m ->- forM_ m $ \bb -> void $ withLock $ f bb+ forM_ m $ \bb -> withLock $ f bb where f bb = cacheGetHead >>= go bb go bb Nothing =@@ -911,31 +942,35 @@ Nothing -> [] Just bals -> addrsToAdd gap bals a -syncMempoolC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m)- => CacheX m ()-syncMempoolC =- void . withLock $- when_cool ck s m $ do- nodepool <- HashSet.fromList . map snd <$> lift getMempool- cachepool <- HashSet.fromList . map snd <$> cacheGetMempool- getem (HashSet.difference nodepool cachepool)- when_cool pk ps pm $- getem (HashSet.difference cachepool nodepool)+withCool :: (MonadLoggerIO m)+ => ByteString+ -> Integer+ -> CacheX m a+ -> CacheX m (Maybe a)+withCool key milliseconds run =+ is_cool >>= \c -> if c+ then run >>= \x -> cooldown >> return (Just x)+ else return Nothing where- pk = "prune"- ck = "cool"- s = Nothing- m = Just 500- ps = Just 10- pm = Nothing- when_cool k m s f = is_cool k >>= \c -> when c $ f >> cooldown k m s- is_cool k = isNothing <$> runRedis (Redis.get k)- cooldown k sec ms =+ is_cool = isNothing <$> runRedis (Redis.get key)+ cooldown = let opts = Redis.SetOpts- { Redis.setSeconds = sec- , Redis.setMilliseconds = ms+ { Redis.setSeconds = Nothing+ , Redis.setMilliseconds = Just milliseconds , Redis.setCondition = Just Redis.Nx }- in void . runRedis $ Redis.setOpts k "0" opts+ in void . runRedis $ Redis.setOpts key "0" opts++syncMempoolC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m)+ => CacheX m ()+syncMempoolC = do+ refresh <- toInteger <$> asks cacheRefresh+ void . withLock . withCool "cool" (refresh * 10 `div` 9) $ do+ nodepool <- HashSet.fromList . map snd <$> lift getMempool+ cachepool <- HashSet.fromList . map snd <$> cacheGetMempool+ getem (HashSet.difference nodepool cachepool)+ withCool "prune" (refresh * 100 `div` 9) $+ getem (HashSet.difference cachepool nodepool)+ where getem tset = do let tids = HashSet.toList tset txs <- catMaybes <$> mapM (lift . getTxData) tids@@ -1146,4 +1181,4 @@ cacheNewBlock = send CacheNewBlock cachePing :: MonadIO m => CacheWriter -> m ()-cachePing = send CachePing+cachePing = query CachePing
src/Haskoin/Store/Manager.hs view
@@ -35,8 +35,8 @@ blockStoreTxHashSTM, blockStoreTxSTM, withBlockStore) import Haskoin.Store.Cache (CacheConfig (..), CacheWriter,- cacheNewBlock, cachePing,- cacheWriter, connectRedis)+ cacheNewBlock, cacheWriter,+ connectRedis) import Haskoin.Store.Common (StoreEvent (..)) import Haskoin.Store.Database.Reader (DatabaseReader (..), DatabaseReaderT,@@ -203,14 +203,9 @@ -> m a -> m a cacheWriterProcesses crefresh evts cwm action =- withAsync events $ \a1 ->- withAsync ping $ \a2 ->- link a1 >> link a2 >> action+ withAsync events $ \a1 -> link a1 >> action where events = cacheWriterEvents evts cwm- ping = forever $ do- threadDelay (crefresh * 1000)- cachePing cwm cacheWriterEvents :: MonadIO m => Inbox StoreEvent -> CacheWriter -> m () cacheWriterEvents evts cwm =