Variable pipelining to optimize memory usage in the wallet restoration phase

.gitignore

@@ -37,10 +37,12 @@ hie.yaml
 ### Direnv ###
 .envrc-local
 .envrc-override
+.direnv/flake-profile
+.direnv/flake-profile-1-link
 
 ### auto-generated faulty JSON golden tests ###
 *.faulty.json
-*.faulty.json
+*.faulty.reencoded.json
 lib/shelley/test/data/balanceTx/**/actual
 
 ### Release scripts output
@@ -49,3 +51,8 @@ lib/shelley/test/data/balanceTx/**/actual
 ### Docs build
 /_build/
 .vscode/settings.json
+
+## local ignored space
+ignore-me
+
+

lib/core/src/Cardano/Wallet/Primitive/Model.hs

@@ -41,6 +41,7 @@ module Cardano.Wallet.Primitive.Model
 
     , BlockData (..)
     , firstHeader
+    , lastHeader
 
     -- * Accessors
     , currentTip
@@ -388,6 +389,8 @@ firstHeader :: BlockData m addr txs s -> BlockHeader
 firstHeader (List xs) = header $ NE.head xs
 firstHeader (Summary _ BlockSummary{from}) = from
 
+-- | Last 'BlockHeader' of the blocks represented
+-- by 'BlockData'.
 lastHeader :: BlockData m addr txs s -> BlockHeader
 lastHeader (List xs) = header $ NE.last xs
 lastHeader (Summary _ BlockSummary{to}) = to

lib/core/src/Ouroboros/Network/Client/Wallet.hs

@@ -4,10 +4,14 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE Rank2Types #-}
+{-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE ViewPatterns #-}
 
 -- |
 -- Copyright: © 2020 IOHK
@@ -24,6 +28,9 @@ module Ouroboros.Network.Client.Wallet
 
       -- * ChainSyncWithBlocks
     , chainSyncWithBlocks
+    , PipeliningStrategy(..)
+    , thousandPipeliningStrategy
+    , tunedForMainnetPipeliningStrategy
 
       -- * LocalTxSubmission
     , LocalTxSubmissionCmd (..)
@@ -74,6 +81,8 @@ import Data.List.NonEmpty
     ( NonEmpty (..) )
 import Data.Ord
     ( comparing )
+import Data.Text
+    ( Text )
 import Data.Void
     ( Void )
 import Network.TypedProtocol.Pipelined
@@ -114,6 +123,7 @@ import Ouroboros.Network.Protocol.LocalTxSubmission.Type
     ( SubmitResult (..) )
 
 import qualified Data.List.NonEmpty as NE
+import qualified Data.Text as T
 import qualified Ouroboros.Network.Protocol.ChainSync.ClientPipelined as P
 import qualified Ouroboros.Network.Protocol.LocalStateQuery.Client as LSQ
 
@@ -225,6 +235,32 @@ chainSyncFollowTip toCardanoEra onTipUpdate =
 type RequestNextStrategy m n block
     = P.ClientPipelinedStIdle n block (Point block) (Tip block) m Void
 
+-- | How to drive pipelining size from the block height
+data PipeliningStrategy block = PipeliningStrategy
+    { getPipeliningSize :: block -> Natural
+    , pipeliningStrategyName :: Text
+    }
+
+instance Show  (PipeliningStrategy block) where 
+    show PipeliningStrategy{pipeliningStrategyName} 
+        = T.unpack pipeliningStrategyName 
+
+thousandPipeliningStrategy :: PipeliningStrategy block
+thousandPipeliningStrategy = PipeliningStrategy {..}
+    where 
+        getPipeliningSize _ = 1_000
+        pipeliningStrategyName = "Constant pipelining of 1000 blocks"
+
+tunedForMainnetPipeliningStrategy :: HasHeader block => PipeliningStrategy block 
+tunedForMainnetPipeliningStrategy =  PipeliningStrategy {..}
+    where 
+        getPipeliningSize (blockNo -> n) 
+            | n <= 5_200_000 = 1000
+            | n <= 6_100_000 = 200
+            | n <= 6_500_000 = 125
+            | otherwise      = 100
+        pipeliningStrategyName = "Variable pipelining suited for mainnet blockchain"
+
 -- | Helper type for the different ways we handle rollbacks.
 --
 -- Helps remove some boilerplate.
@@ -278,9 +314,10 @@ data LocalRollbackResult block
 chainSyncWithBlocks
     :: forall m block. (Monad m, MonadSTM m, MonadThrow m, HasHeader block)
     => Tracer m (ChainSyncLog block (Point block))
+    -> PipeliningStrategy block 
     -> ChainFollower m (Point block) (Tip block) (NonEmpty block)
     -> ChainSyncClientPipelined block (Point block) (Tip block) m Void
-chainSyncWithBlocks tr chainFollower =
+chainSyncWithBlocks tr pipeliningStrategy chainFollower =
     ChainSyncClientPipelined clientStNegotiateIntersection
   where
     -- Return the _number of slots between two tips.
@@ -381,12 +418,17 @@ chainSyncWithBlocks tr chainFollower =
             let blocks' = NE.reverse (block :| blocks)
             traceWith tr $ MsgChainRollForward blocks' (getTipPoint tip)
             handleRollforward blocks' tip
-
             let distance = tipDistance (blockNo block) tip
             traceWith tr $ MsgTipDistance distance
             let strategy = if distance <= 1
                     then oneByOne
-                    else pipeline (fromIntegral $ min distance 1000) Zero
+                    else pipeline 
+                        (fromIntegral 
+                            . min distance 
+                            . getPipeliningSize pipeliningStrategy 
+                            $ block
+                        ) 
+                        Zero
             clientStIdle strategy
 
         , P.recvMsgRollBackward = \point tip -> do

lib/shelley/bench/latency-bench.hs

@@ -102,6 +102,8 @@ import Network.Wai.Middleware.Logging
     ( ApiLog (..) )
 import Numeric.Natural
     ( Natural )
+import Ouroboros.Network.Client.Wallet
+    ( tunedForMainnetPipeliningStrategy )
 import System.Directory
     ( createDirectory )
 import System.FilePath
@@ -474,6 +476,7 @@ withShelleyServer tracers action = do
         serveWallet
             (NodeSource conn vData)
             np
+            tunedForMainnetPipeliningStrategy
             (SomeNetworkDiscriminant $ Proxy @'Mainnet)
             tracers
             (SyncTolerance 10)