Skip to content
This repository was archived by the owner on Aug 18, 2020. It is now read-only.

Commit 1a3b01b

Browse files
aviethavieth
committed
better index/db intersection computation
Now the index will not always rollback to origin when its tip is not in the database. It will try to find an intersection point within the security parameter, and roll back only to that point.
1 parent a71712a commit 1a3b01b

File tree

7 files changed

+161
-84
lines changed

7 files changed

+161
-84
lines changed

‎src/cddl-test/Main.hs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ dbArgs fp = do
102102
, immErr = EH.exceptions
103103
, immEpochInfo = epochInfo
104104
, immValidation = ValidateMostRecentEpoch
105-
, immIsEBB = isEBB
105+
, immIsEBB = fmap fst . isEBB
106106
, immAddHdrEnv = Byron.byronAddHeaderEnvelope
107107
, immCheckIntegrity = const True -- No validation
108108
, immHasFS = ioHasFS $ MountPoint (fp </> "immutable")

‎src/exec/Byron.hs‎

Lines changed: 4 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ import Data.List.NonEmpty (NonEmpty)
2222
import qualified Data.List.NonEmpty as NE
2323
import Data.Maybe (mapMaybe)
2424
import qualified Data.Text.Lazy.Builder as Text
25-
import Data.Word (Word64)
2625
import System.Random (StdGen, getStdGen, randomR)
2726

2827
import qualified Cardano.Binary as Binary
@@ -34,14 +33,14 @@ import qualified Pos.Chain.Block as CSL (Block, BlockHeader (..), GenesisBlock,
3433
MainBlockHeader, HeaderHash, headerHash)
3534
import qualified Pos.Infra.Diffusion.Types as CSL
3635

37-
import Ouroboros.Byron.Proxy.Block (ByronBlock,
36+
import Ouroboros.Byron.Proxy.Block (ByronBlock, checkpointOffsets,
3837
coerceHashToLegacy, headerHash)
3938
import Ouroboros.Byron.Proxy.Main
4039
import Ouroboros.Consensus.Block (Header)
4140
import Ouroboros.Consensus.Ledger.Byron (ByronHash(..),
4241
byronHeaderRaw, mkByronBlock)
4342
import Ouroboros.Consensus.Ledger.Byron.Auxiliary as Cardano
44-
import Ouroboros.Consensus.Protocol.Abstract (SecurityParam (maxRollbacks))
43+
import Ouroboros.Consensus.Protocol.Abstract (SecurityParam)
4544
import Ouroboros.Network.Block (ChainHash (..), Point, pointHash)
4645
import qualified Ouroboros.Network.AnchoredFragment as AF
4746
import qualified Ouroboros.Network.ChainFragment as CF
@@ -136,25 +135,14 @@ download tracer genesisBlock epochSlots securityParam db bp = do
136135
checkpoints
137136
:: AF.AnchoredFragment (Header ByronBlock)
138137
-> [CSL.HeaderHash]
139-
checkpoints = mapMaybe pointToHash . AF.selectPoints (fmap fromIntegral offsets)
138+
checkpoints = mapMaybe pointToHash .
139+
AF.selectPoints (fmap fromIntegral (checkpointOffsets securityParam))
140140

141141
pointToHash :: Point (Header ByronBlock) -> Maybe CSL.HeaderHash
142142
pointToHash pnt = case pointHash pnt of
143143
GenesisHash -> Nothing
144144
BlockHash (ByronHash hash) -> Just $ coerceHashToLegacy hash
145145

146-
-- Offsets for selectPoints. Defined in the same way as for the Shelley
147-
-- chain sync client: fibonacci numbers including 0 and k.
148-
offsets :: [Word64]
149-
offsets = 0 : foldr includeK ([] {- this is never forced -}) (tail fibs)
150-
151-
includeK :: Word64 -> [Word64] -> [Word64]
152-
includeK w ws | w >= k = [k]
153-
| otherwise = w : ws
154-
155-
fibs :: [Word64]
156-
fibs = 1 : 1 : zipWith (+) fibs (tail fibs)
157-
158146
streamer :: CSL.HeaderHash -> CSL.StreamBlocks CSL.Block IO CSL.HeaderHash
159147
streamer tipHash = CSL.StreamBlocks
160148
{ CSL.streamBlocksMore = \blocks -> do
@@ -188,9 +176,6 @@ download tracer genesisBlock epochSlots securityParam db bp = do
188176
let (idx, rndGen') = randomR (0, NE.length ne - 1) rndGen
189177
in (ne NE.!! idx, rndGen')
190178

191-
k :: Word64
192-
k = maxRollbacks securityParam
193-
194179
recodeBlockOrFail
195180
:: Cardano.EpochSlots
196181
-> (forall x . Binary.DecoderError -> IO x)

‎src/exec/DB.hs‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,10 @@ import Ouroboros.Consensus.BlockchainTime (BlockchainTime)
2323
import Ouroboros.Consensus.Ledger.Byron.Config (pbftEpochSlots)
2424
import Ouroboros.Consensus.Ledger.Extended (ExtLedgerState)
2525
import Ouroboros.Consensus.Node (withChainDB)
26+
import Ouroboros.Consensus.Util.ResourceRegistry (ResourceRegistry)
2627
import Ouroboros.Consensus.Protocol (NodeConfig,
2728
pbftExtConfig)
28-
import Ouroboros.Consensus.Util.ResourceRegistry (ResourceRegistry)
29-
import qualified Ouroboros.Consensus.Util.ResourceRegistry as ResourceRegistry
29+
import Ouroboros.Consensus.Protocol.Abstract (protocolSecurityParam)
3030
import Ouroboros.Storage.ChainDB.API (ChainDB)
3131
import qualified Ouroboros.Storage.ChainDB.Impl as ChainDB
3232
import Ouroboros.Storage.ChainDB.Impl.Args (ChainDbArgs (..))
@@ -61,8 +61,7 @@ withDB dbOptions dbTracer indexTracer rr btime nodeConfig extLedgerState k = do
6161
withChainDB dbTracer rr btime (dbFilePath dbOptions) nodeConfig extLedgerState customiseArgs
6262
$ \cdb ->
6363
Sqlite.withIndexAuto epochSlots indexTracer (indexFilePath dbOptions) $ \idx -> do
64-
_ <- ResourceRegistry.forkLinkedThread rr $ Index.trackChainDB rr idx cdb
65-
k idx cdb
64+
Index.trackChainDB rr idx cdb (protocolSecurityParam nodeConfig) (k idx cdb)
6665

6766
where
6867

‎src/lib/Ouroboros/Byron/Proxy/Block.hs‎

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,11 @@ module Ouroboros.Byron.Proxy.Block
1919
, coerceHashToLegacy
2020
, headerHash
2121
, isEBB
22+
, checkpointOffsets
2223
) where
2324

2425
import qualified Codec.CBOR.Write as CBOR (toStrictByteString)
26+
import Data.Word (Word64)
2527

2628
import qualified Pos.Chain.Block as CSL (HeaderHash)
2729
import qualified Pos.Crypto.Hashing as Legacy (AbstractHash (..))
@@ -31,8 +33,9 @@ import qualified Cardano.Chain.Block as Cardano
3133
import Cardano.Crypto.Hashing (AbstractHash (..))
3234

3335
import qualified Ouroboros.Consensus.Block as Consensus (GetHeader (..))
34-
import Ouroboros.Consensus.Ledger.Byron (ByronBlock (..), ByronHash (..),
35-
encodeByronBlock, byronHeaderHash)
36+
import Ouroboros.Consensus.Ledger.Byron (ByronBlock (..),
37+
ByronHash (..), encodeByronBlock, byronHeaderHash)
38+
import Ouroboros.Consensus.Protocol.Abstract (SecurityParam (..))
3639
import Ouroboros.Storage.Common (EpochNo(..))
3740

3841
-- For type instance HeaderHash (Header blk) = HeaderHash blk
@@ -60,11 +63,23 @@ headerHash = unByronHash . byronHeaderHash
6063

6164
-- | Return @Just@ the epoch number if the block is an EBB, @Nothing@ for
6265
-- regular blocks
63-
isEBB :: ByronBlock -> Maybe EpochNo
66+
isEBB :: ByronBlock -> Maybe (EpochNo, ByronHash)
6467
isEBB blk = case byronBlockRaw blk of
6568
Cardano.ABOBBlock _ -> Nothing
66-
Cardano.ABOBBoundary ebb -> Just
67-
. EpochNo
68-
. Cardano.boundaryEpoch
69-
. Cardano.boundaryHeader
70-
$ ebb
69+
Cardano.ABOBBoundary ebb -> Just (epochNo, byronHash)
70+
where
71+
epochNo = EpochNo . Cardano.boundaryEpoch . Cardano.boundaryHeader $ ebb
72+
byronHash = ByronHash . headerHash . Consensus.getHeader $ blk
73+
74+
-- | Compute the offsets for use by ChainFragment.selectPoints or
75+
-- AnchoredFragment.selectPoints for some security parameter k. It uses
76+
-- fibonacci numbers with 0 and k as endpoints, and the duplicate 1 at the
77+
-- start removed.
78+
checkpointOffsets :: SecurityParam -> [Word64]
79+
checkpointOffsets (SecurityParam k) = 0 : foldr includeK ([] {- this is never forced -}) (tail fibs)
80+
where
81+
includeK :: Word64 -> [Word64] -> [Word64]
82+
includeK w ws | w >= k = [k]
83+
| otherwise = w : ws
84+
fibs :: [Word64]
85+
fibs = 1 : 1 : zipWith (+) fibs (tail fibs)

‎src/lib/Ouroboros/Byron/Proxy/Index/ChainDB.hs‎

Lines changed: 58 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -5,14 +5,18 @@ module Ouroboros.Byron.Proxy.Index.ChainDB
55
( trackChainDB
66
) where
77

8+
import Control.Concurrent.Async (race)
89
import Control.Exception (bracket)
10+
import Data.Word (Word64)
911

12+
import Ouroboros.Byron.Proxy.Block (checkpointOffsets)
1013
import Ouroboros.Byron.Proxy.Index.Types (Index)
1114
import qualified Ouroboros.Byron.Proxy.Index.Types as Index
1215
import Ouroboros.Consensus.Block (GetHeader (Header))
1316
import Ouroboros.Consensus.Util.ResourceRegistry (ResourceRegistry)
17+
import Ouroboros.Consensus.Protocol.Abstract (SecurityParam)
1418
import Ouroboros.Network.Block (ChainUpdate (..), Point (..))
15-
import Ouroboros.Network.Point (WithOrigin (Origin))
19+
import Ouroboros.Network.Point (WithOrigin (Origin), block)
1620
import Ouroboros.Storage.ChainDB.API (ChainDB, Reader)
1721
import qualified Ouroboros.Storage.ChainDB.API as ChainDB
1822

@@ -21,7 +25,7 @@ trackReaderBlocking
2125
:: ( Monad m )
2226
=> Index m (Header blk)
2327
-> Reader m blk (Header blk)
24-
-> m x
28+
-> m void
2529
trackReaderBlocking idx reader = do
2630
instruction <- ChainDB.readerInstructionBlocking reader
2731
case instruction of
@@ -50,42 +54,69 @@ trackReader idx reader = do
5054
trackReader idx reader
5155
Nothing -> pure ()
5256

53-
-- | Have an Index track a ChainDB using its Reader API. You probably want to
54-
-- race this with some other thread that runs your application.
55-
--
56-
-- If the ChainDB does not contain the tip of the Index, then the whole index
57-
-- will be rebuilt.
57+
-- | Have an Index track a ChainDB using its Reader API for the duration of
58+
-- some monadic action.
5859
--
5960
-- It will spawn a thread to do the index updates. This must be the only
6061
-- index writer. It is run by `race` with the action, so exceptions in either
6162
-- the action or the writer thread will be re-thrown here.
6263
--
63-
-- If the tip of the index is not in the ChainDB, then the entire index will be
64-
-- rebuilt. This is not ideal: there may be an intersection. TODO would be
65-
-- better to check the newest slot older than `k` back from tip of index, and
66-
-- go from there.
64+
-- If the tip of the index is in the ChainDB, then no work must be done in the
65+
-- beginning. But if it's not in the ChainDB, there will have to be a rollback
66+
-- on the index. The SecurityParam k is used to decide how far back to try. If
67+
-- Only index entries at most k slots old will be checked against the
68+
-- ChainDB. If none are in it, then the entire index will be rebuild (rollback
69+
-- to Origin).
6770
trackChainDB
68-
:: forall blk void .
71+
:: forall blk t .
6972
ResourceRegistry IO
7073
-> Index IO (Header blk)
7174
-> ChainDB IO blk
72-
-> IO void
73-
trackChainDB rr idx cdb = bracket acquireReader releaseReader $ \rdr -> do
74-
tipPoint <- Index.tip idx
75-
mPoint <- ChainDB.readerForward rdr [Point tipPoint]
76-
-- `readerForward` docs say that if we get `Nothing`, the next reader
77-
-- instruction may not be a rollback, so we'll manually roll the index
78-
-- back. It's assumed the read pointer will be at origin (nothing else
79-
-- would make sense).
80-
case mPoint of
81-
Nothing -> Index.rollbackward idx Origin
82-
Just _ -> pure ()
83-
-- First, block until the index is caught up to the tip ...
84-
trackReader idx rdr
85-
-- ... then attempt to stay in sync.
86-
trackReaderBlocking idx rdr
75+
-> SecurityParam
76+
-> IO t
77+
-> IO t
78+
trackChainDB rr idx cdb k act = bracket acquireReader releaseReader $ \rdr -> do
79+
checkpoints <- Index.streamFromTip idx checkpointsFold
80+
mPoint <- ChainDB.readerForward rdr checkpoints
81+
case mPoint of
82+
-- `readerForward` docs say that the next instruction will be a rollback,
83+
-- so we don't have to do anything here; the call to `trackReader` will
84+
-- do what needs to be done.
85+
Just _ -> pure ()
86+
-- `readerForward` docs say that if we get `Nothing`, the next reader
87+
-- instruction may not be a rollback, so we'll manually roll the index
88+
-- back. It's assumed the read pointer will be at origin (nothing else
89+
-- would make sense).
90+
Nothing -> Index.rollbackward idx Origin
91+
-- First, block until the index is caught up to the tip ...
92+
trackReader idx rdr
93+
-- ... then attempt to stay in sync.
94+
outcome <- race (trackReaderBlocking idx rdr) act
95+
case outcome of
96+
Left impossible -> impossible
97+
Right t -> pure t
8798
where
8899
acquireReader :: IO (Reader IO blk (Header blk))
89100
acquireReader = ChainDB.deserialiseReader <$> ChainDB.newHeaderReader cdb rr
90101
releaseReader :: Reader IO blk (Header blk) -> IO ()
91102
releaseReader = ChainDB.readerClose
103+
104+
checkpointsFold :: Index.Fold (Header blk) [Point blk]
105+
checkpointsFold = checkpointsFoldN 0 (checkpointOffsets k)
106+
107+
-- Count up from 0 on the first parameter. Whenever it coincides with the
108+
-- head of the second parameter (an increasing list) include that point.
109+
-- Stop when the second list is empty.
110+
-- Since checkpointsFold always includes the paramater k, the k'th entry
111+
-- in the index will always be in here, unless the index is shorter
112+
-- than k. This block is _at least_ k slots behind the DB, so if it's not
113+
-- in the DB then the index is way out of date.
114+
checkpointsFoldN
115+
:: Word64
116+
-> [Word64]
117+
-> Index.Fold (Header blk) [Point blk]
118+
checkpointsFoldN _ [] = Index.Stop []
119+
checkpointsFoldN w (o : os) = Index.More [] $ \slotNo hash ->
120+
if w == o
121+
then fmap ((:) (Point (block slotNo hash))) (checkpointsFoldN (w+1) os)
122+
else checkpointsFoldN (w+1) (o : os)

‎src/lib/Ouroboros/Byron/Proxy/Index/Sqlite.hs‎

Lines changed: 53 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -55,10 +55,11 @@ index
5555
-> Sql.Statement -- for insert
5656
-> Index IO (Header ByronBlock)
5757
index epochSlots tracer conn insertStatement = Index
58-
{ Index.lookup = sqliteLookup epochSlots conn
59-
, tip = sqliteTip epochSlots conn
60-
, rollforward = sqliteRollforward epochSlots tracer insertStatement
61-
, rollbackward = sqliteRollbackward epochSlots tracer conn
58+
{ Index.lookup = sqliteLookup epochSlots conn
59+
, tip = sqliteTip epochSlots conn
60+
, streamFromTip = sqliteStreamFromTip epochSlots conn
61+
, rollforward = sqliteRollforward epochSlots tracer insertStatement
62+
, rollbackward = sqliteRollbackward epochSlots tracer conn
6263
}
6364

6465
-- | Open a new or existing SQLite database. If new, it will set up the schema.
@@ -157,6 +158,24 @@ createTable conn = Sql.execute_ conn sql_create_table
157158
createIndex :: Sql.Connection -> IO ()
158159
createIndex conn = Sql.execute_ conn sql_create_index
159160

161+
convertHashBlob :: ByteString -> IO HeaderHash
162+
convertHashBlob blob = case digestFromByteString blob of
163+
Just hh -> pure (AbstractHash hh)
164+
Nothing -> throwIO $ InvalidHash blob
165+
166+
-- | Convert the database encoding of relative slot to the offset in an
167+
-- epoch. The header hash is taken for error-reporting purposes.
168+
offsetInEpoch :: HeaderHash -> Int -> IO Word64
169+
offsetInEpoch hh i
170+
| i == -1 = pure 0
171+
| i >= 0 = pure $ fromIntegral i
172+
| otherwise = throwIO $ InvalidRelativeSlot hh i
173+
174+
toAbsoluteSlot :: EpochSlots -> HeaderHash -> Word64 -> Int -> IO SlotNo
175+
toAbsoluteSlot epochSlots hh epochNo slotInt = do
176+
offset <- offsetInEpoch hh slotInt
177+
pure $ SlotNo $ unEpochSlots epochSlots * epochNo + offset
178+
160179
-- | The tip is the entry with the highest epoch and slot pair.
161180
sql_get_tip :: Query
162181
sql_get_tip =
@@ -169,18 +188,36 @@ sqliteTip epochSlots conn = do
169188
case rows of
170189
[] -> pure Origin
171190
((hhBlob, epoch, slotInt) : _) -> do
172-
hh <- case digestFromByteString hhBlob of
173-
Just hh -> pure (AbstractHash hh)
174-
Nothing -> throwIO $ InvalidHash hhBlob
175-
offsetInEpoch :: Word64 <-
176-
if slotInt == -1
177-
then pure 0
178-
else if slotInt >= 0
179-
then pure $ fromIntegral slotInt
180-
else throwIO $ InvalidRelativeSlot hh slotInt
181-
let slotNo = SlotNo $ unEpochSlots epochSlots * epoch + offsetInEpoch
191+
hh <- convertHashBlob hhBlob
192+
slotNo <- toAbsoluteSlot epochSlots hh epoch slotInt
182193
pure $ At $ Point.Block slotNo (ByronHash hh)
183194

195+
sql_get_all :: Query
196+
sql_get_all =
197+
"SELECT header_hash, epoch, slot FROM block_index\
198+
\ ORDER BY epoch DESC, slot DESC;"
199+
200+
-- | Stream rows from the tip by using the prepared statement and nextRow
201+
-- API.
202+
sqliteStreamFromTip
203+
:: EpochSlots
204+
-> Sql.Connection
205+
-> Index.Fold (Header ByronBlock) t
206+
-> IO t
207+
sqliteStreamFromTip epochSlots conn fold = Sql.withStatement conn sql_get_all $ \stmt ->
208+
go stmt fold
209+
where
210+
go stmt step = case step of
211+
Index.Stop t -> pure t
212+
Index.More t k -> do
213+
next <- Sql.nextRow stmt
214+
case next of
215+
Nothing -> pure t
216+
Just (hhBlob, epoch, slotInt) -> do
217+
hh <- convertHashBlob hhBlob
218+
slotNo <- toAbsoluteSlot epochSlots hh epoch slotInt
219+
go stmt (k slotNo (ByronHash hh))
220+
184221
sql_get_hash :: Query
185222
sql_get_hash =
186223
"SELECT epoch, slot FROM block_index\
@@ -194,13 +231,8 @@ sqliteLookup epochSlots conn (ByronHash hh@(AbstractHash digest)) = do
194231
case rows of
195232
[] -> pure Nothing
196233
((epoch, slotInt) : _) -> do
197-
offsetInEpoch :: Word64 <-
198-
if slotInt == -1
199-
then pure 0
200-
else if slotInt >= 0
201-
then pure $ fromIntegral slotInt
202-
else throwIO $ InvalidRelativeSlot hh slotInt
203-
pure $ Just $ SlotNo $ unEpochSlots epochSlots * epoch + offsetInEpoch
234+
slotNo <- toAbsoluteSlot epochSlots hh epoch slotInt
235+
pure $ Just slotNo
204236

205237
-- | Note that there is a UNIQUE constraint. This will fail if a duplicate
206238
-- entry is inserted.

0 commit comments

Comments
 (0)