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.49.0 +### Changed +- Improve conduit merging algorithm. + ## 0.48.0 ### Added - Timeouts and token bucket for web requests. 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: 386ac449dae5732a7ab6b6f5a3aa3f54779b3acc90fda516a2560304d449327b +-- hash: a206136ca02c8f998e0567b78de9de9f7b2d2a99ab278d1d62f5c1b193b9e2e2 name: haskoin-store -version: 0.48.0 +version: 0.49.0 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network @@ -61,7 +61,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 - , haskoin-store-data ==0.48.0 + , haskoin-store-data ==0.49.0 , hedis >=0.12.13 , http-types >=0.12.3 , lens >=4.18.1 @@ -113,7 +113,7 @@ , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 , haskoin-store - , haskoin-store-data ==0.48.0 + , haskoin-store-data ==0.49.0 , hedis >=0.12.13 , http-types >=0.12.3 , lens >=4.18.1 @@ -170,7 +170,7 @@ , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 , haskoin-store - , haskoin-store-data ==0.48.0 + , haskoin-store-data ==0.49.0 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3 diff --git a/src/Haskoin/Store/Database/Reader.hs b/src/Haskoin/Store/Database/Reader.hs --- a/src/Haskoin/Store/Database/Reader.hs +++ b/src/Haskoin/Store/Database/Reader.hs @@ -40,7 +40,7 @@ import Haskoin.Store.Common import Haskoin.Store.Data import Haskoin.Store.Database.Types -import Haskoin.Store.Logic (joinStreams) +import Haskoin.Store.Logic (joinDescStreams) import UnliftIO (MonadIO, MonadUnliftIO, liftIO) type DatabaseReaderT = ReaderT DatabaseReader @@ -178,7 +178,7 @@ getAddressesTxsDB addrs limits bdb@DatabaseReader{databaseHandle = db} = liftIO $ iters addrs [] $ \cs -> runConduit $ - joinStreams (flip compare) cs .| applyLimitsC limits .| sinkList + joinDescStreams cs .| applyLimitsC limits .| sinkList where iters [] acc f = f acc iters (a : as) acc f = diff --git a/src/Haskoin/Store/Logic.hs b/src/Haskoin/Store/Logic.hs --- a/src/Haskoin/Store/Logic.hs +++ b/src/Haskoin/Store/Logic.hs @@ -14,40 +14,39 @@ , newMempoolTx , deleteUnconfirmedTx , streamThings - , joinStreams + , joinDescStreams ) where -import Conduit (ConduitT, await, lift, sealConduitT, - yield, ($$++)) -import Control.Monad (forM, forM_, guard, unless, void, when, - zipWithM_) -import Control.Monad.Except (MonadError, throwError) -import Control.Monad.Logger (MonadLoggerIO (..), logDebugS, - logErrorS) -import qualified Data.ByteString as B -import Data.Either (rights) -import Data.Function (on) -import qualified Data.IntMap.Strict as I -import Data.List (nub, sortBy) -import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, - mapMaybe) -import Data.Serialize (encode) -import Data.Word (Word32, Word64) -import Haskoin (Address, Block (..), BlockHash, - BlockHeader (..), BlockNode (..), - Network (..), OutPoint (..), Tx (..), - TxHash, TxIn (..), TxOut (..), - blockHashToHex, computeSubsidy, - eitherToMaybe, genesisBlock, - genesisNode, headerHash, isGenesis, - nullOutPoint, scriptToAddressBS, txHash, - txHashToHex) +import Conduit (ConduitT, await, lift, sealConduitT, + yield, ($$++)) +import Control.Monad (forM, forM_, guard, unless, void, when, + zipWithM_) +import Control.Monad.Except (MonadError, throwError) +import Control.Monad.Logger (MonadLoggerIO (..), logDebugS, logErrorS) +import qualified Data.ByteString as B +import Data.Either (rights) +import Data.Function (on) +import qualified Data.IntMap.Strict as I +import Data.List (nub, sortBy) +import qualified Data.Map.Strict as Map +import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, + mapMaybe) +import Data.Serialize (encode) +import Data.Word (Word32, Word64) +import Haskoin (Address, Block (..), BlockHash, + BlockHeader (..), BlockNode (..), + Network (..), OutPoint (..), Tx (..), + TxHash, TxIn (..), TxOut (..), + blockHashToHex, computeSubsidy, + eitherToMaybe, genesisBlock, genesisNode, + headerHash, isGenesis, nullOutPoint, + scriptToAddressBS, txHash, txHashToHex) import Haskoin.Store.Common -import Haskoin.Store.Data (Balance (..), BlockData (..), - BlockRef (..), Prev (..), Spender (..), - TxData (..), TxRef (..), UnixTime, - Unspent (..), confirmed) -import UnliftIO (Exception) +import Haskoin.Store.Data (Balance (..), BlockData (..), + BlockRef (..), Prev (..), Spender (..), + TxData (..), TxRef (..), UnixTime, + Unspent (..), confirmed) +import UnliftIO (Exception) type MonadImport m = ( MonadError ImportException m @@ -694,26 +693,26 @@ mapM yield ls go (last ls) -joinStreams :: Monad m - => (a -> a -> Ordering) - -> [ConduitT () a m ()] - -> ConduitT () a m () -joinStreams c xs = do +joinDescStreams :: (Monad m, Ord a) + => [ConduitT () a m ()] + -> ConduitT () a m () +joinDescStreams xs = do let ss = map sealConduitT xs - ys <- mapMaybe j <$> - lift (traverse ($$++ await) ss) - go Nothing ys + go Nothing =<< g ss where - j (x, y) = (,) x <$> y - go m ys = - case sortBy (c `on` snd) ys of - [] -> return () - (i,x):ys' -> do + g ss = let f (x, y) = (, [x]) <$> y + l = mapMaybe f <$> lift (traverse ($$++ await) ss) + in Map.fromListWith (++) <$> l + j (x, y) = (, [x]) <$> y + go m mp = case Map.lookupMax mp of + Nothing -> return () + Just (x, ss) -> do case m of Nothing -> yield x Just x' - | c x x' == EQ -> return () + | x == x' -> return () | otherwise -> yield x - j <$> lift (i $$++ await) >>= \case - Nothing -> go (Just x) ys' - Just y -> go (Just x) (y:ys') + mp1 <- g ss + let mp2 = Map.deleteMax mp + mp' = Map.unionWith (++) mp1 mp2 + go (Just x) mp' diff --git a/src/Haskoin/Store/Web.hs b/src/Haskoin/Store/Web.hs --- a/src/Haskoin/Store/Web.hs +++ b/src/Haskoin/Store/Web.hs @@ -1321,7 +1321,7 @@ -> HashSet Address -> ConduitT () BinfoUnspent m () getBinfoUnspents numtxid height xspecs addrs = - joinStreams (flip compare `on` fst) conduits .| mapC (uncurry binfo) + joinDescStreams conduits .| mapC (uncurry binfo) where binfo Unspent{..} xp = let conf = case unspentBlock of @@ -1370,7 +1370,7 @@ -> Int64 -- starting balance -> ConduitT () BinfoTx m () getBinfoTxs abook sxspecs saddrs baddrs bfilter numtxid prune bal = - joinStreams (flip compare) conduits .| go bal + joinDescStreams conduits .| go bal where sxspecs_ls = HashSet.toList sxspecs saddrs_ls = HashSet.toList saddrs