From 5241866252ff4b41d095db5f50cafa704dcac37f Mon Sep 17 00:00:00 2001 From: "alex.sharov" Date: Mon, 15 Jul 2024 11:00:38 +0700 Subject: [PATCH 1/7] save --- .../freezeblocks/block_snapshots.go | 22 +++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/turbo/snapshotsync/freezeblocks/block_snapshots.go b/turbo/snapshotsync/freezeblocks/block_snapshots.go index b4d298236e6..26f697b9329 100644 --- a/turbo/snapshotsync/freezeblocks/block_snapshots.go +++ b/turbo/snapshotsync/freezeblocks/block_snapshots.go @@ -506,6 +506,20 @@ func (s *RoSnapshots) unlockSegments() { }) } +func (s *RoSnapshots) rLockSegments() { + s.segments.Scan(func(segtype snaptype.Enum, value *segments) bool { + value.lock.RLock() + return true + }) +} + +func (s *RoSnapshots) rUnlockSegments() { + s.segments.Scan(func(segtype snaptype.Enum, value *segments) bool { + value.lock.RUnlock() + return true + }) +} + func (s *RoSnapshots) rebuildSegments(fileNames []string, open bool, optimistic bool) error { s.lockSegments() defer s.unlockSegments() @@ -808,8 +822,8 @@ func (s *RoSnapshots) buildMissedIndices(logPrefix string, ctx context.Context, } func (s *RoSnapshots) PrintDebug() { - s.lockSegments() - defer s.unlockSegments() + view := s.View() + defer view.Close() s.segments.Scan(func(key snaptype.Enum, value *segments) bool { fmt.Println(" == [dbg] Snapshots,", key.String()) @@ -2042,7 +2056,7 @@ type View struct { func (s *RoSnapshots) View() *View { v := &View{s: s, baseSegType: coresnaptype.Headers} - s.lockSegments() + s.rLockSegments() return v } @@ -2051,7 +2065,7 @@ func (v *View) Close() { return } v.closed = true - v.s.unlockSegments() + v.s.rUnlockSegments() } func (v *View) Segments(t snaptype.Type) []*Segment { From 96ead46e45429e047a285b685e18a53769c7d915 Mon Sep 17 00:00:00 2001 From: "alex.sharov" Date: Mon, 15 Jul 2024 11:59:16 +0700 Subject: [PATCH 2/7] save --- .../freezeblocks/block_snapshots.go | 24 +++++++------------ 1 file changed, 8 insertions(+), 16 deletions(-) diff --git a/turbo/snapshotsync/freezeblocks/block_snapshots.go b/turbo/snapshotsync/freezeblocks/block_snapshots.go index 26f697b9329..bc0c4b77ea0 100644 --- a/turbo/snapshotsync/freezeblocks/block_snapshots.go +++ b/turbo/snapshotsync/freezeblocks/block_snapshots.go @@ -506,20 +506,6 @@ func (s *RoSnapshots) unlockSegments() { }) } -func (s *RoSnapshots) rLockSegments() { - s.segments.Scan(func(segtype snaptype.Enum, value *segments) bool { - value.lock.RLock() - return true - }) -} - -func (s *RoSnapshots) rUnlockSegments() { - s.segments.Scan(func(segtype snaptype.Enum, value *segments) bool { - value.lock.RUnlock() - return true - }) -} - func (s *RoSnapshots) rebuildSegments(fileNames []string, open bool, optimistic bool) error { s.lockSegments() defer s.unlockSegments() @@ -2056,7 +2042,10 @@ type View struct { func (s *RoSnapshots) View() *View { v := &View{s: s, baseSegType: coresnaptype.Headers} - s.rLockSegments() + s.segments.Scan(func(segtype snaptype.Enum, value *segments) bool { + value.lock.RLock() + return true + }) return v } @@ -2065,7 +2054,10 @@ func (v *View) Close() { return } v.closed = true - v.s.rUnlockSegments() + v.s.segments.Scan(func(segtype snaptype.Enum, value *segments) bool { + value.lock.RUnlock() + return true + }) } func (v *View) Segments(t snaptype.Type) []*Segment { From f8fc5251d3705924589a3bc25b875e5bdd8ff89b Mon Sep 17 00:00:00 2001 From: "alex.sharov" Date: Mon, 15 Jul 2024 12:04:17 +0700 Subject: [PATCH 3/7] save --- .../snapshotsync/freezeblocks/block_reader.go | 24 +++++++---- .../freezeblocks/block_snapshots.go | 43 +++++++++++++++++++ 2 files changed, 58 insertions(+), 9 deletions(-) diff --git a/turbo/snapshotsync/freezeblocks/block_reader.go b/turbo/snapshotsync/freezeblocks/block_reader.go index bc1b33033c6..9ef9d1b2709 100644 --- a/turbo/snapshotsync/freezeblocks/block_reader.go +++ b/turbo/snapshotsync/freezeblocks/block_reader.go @@ -656,15 +656,14 @@ func (r *BlockReader) blockWithSenders(ctx context.Context, tx kv.Getter, hash c return } - view := r.sn.View() - defer view.Close() - seg, ok := view.HeadersSegment(blockHeight) + seg, ok, release := r.sn.ViewSingleFile(coresnaptype.Headers, blockHeight) if !ok { if dbgLogs { log.Info(dbgPrefix + "no header files for this block num") } return } + defer release() var buf []byte h, buf, err := r.headerFromSnapshot(blockHeight, seg, buf) @@ -677,21 +676,26 @@ func (r *BlockReader) blockWithSenders(ctx context.Context, tx kv.Getter, hash c } return } + release() var b *types.Body var baseTxnId uint64 var txsAmount uint32 - bodySeg, ok := view.BodiesSegment(blockHeight) + bodySeg, ok, release := r.sn.ViewSingleFile(coresnaptype.Bodies, blockHeight) if !ok { if dbgLogs { log.Info(dbgPrefix + "no bodies file for this block num") } return } + defer release() + b, baseTxnId, txsAmount, buf, err = r.bodyFromSnapshot(blockHeight, bodySeg, buf) if err != nil { return nil, nil, err } + release() + if b == nil { if dbgLogs { log.Info(dbgPrefix + "got nil body from file") @@ -710,18 +714,21 @@ func (r *BlockReader) blockWithSenders(ctx context.Context, tx kv.Getter, hash c return block, senders, nil } - txnSeg, ok := view.TxsSegment(blockHeight) + txnSeg, ok, release := r.sn.ViewSingleFile(coresnaptype.Transactions, blockHeight) if !ok { if dbgLogs { log.Info(dbgPrefix+"no transactions file for this block num", "r.sn.BlocksAvailable()", r.sn.BlocksAvailable(), "r.sn.indicesReady", r.sn.indicesReady.Load()) } return } + defer release() var txs []types.Transaction txs, senders, err = r.txsFromSnapshot(baseTxnId, txsAmount, txnSeg, buf) if err != nil { return nil, nil, err } + release() + block = types.NewBlockFromStorage(hash, h, txs, b.Uncles, b.Withdrawals) if len(senders) != block.Transactions().Len() { if dbgLogs { @@ -1015,10 +1022,9 @@ func (r *BlockReader) TxnLookup(_ context.Context, tx kv.Getter, txnHash common. return *n, true, nil } - view := r.sn.View() - defer view.Close() - - _, blockNum, ok, err := r.txnByHash(txnHash, view.Txs(), nil) + txns, release := r.sn.ViewType(coresnaptype.Transactions) + defer release() + _, blockNum, ok, err := r.txnByHash(txnHash, txns, nil) if err != nil { return 0, false, err } diff --git a/turbo/snapshotsync/freezeblocks/block_snapshots.go b/turbo/snapshotsync/freezeblocks/block_snapshots.go index bc0c4b77ea0..501c2e3884c 100644 --- a/turbo/snapshotsync/freezeblocks/block_snapshots.go +++ b/turbo/snapshotsync/freezeblocks/block_snapshots.go @@ -2060,6 +2060,49 @@ func (v *View) Close() { }) } +var noop = func() {} + +func (s *RoSnapshots) ViewType(t snaptype.Type) (segments []*Segment, release func()) { + segs, ok := s.segments.Get(t.Enum()) + if !ok { + return nil, noop + } + + segs.lock.RLock() + var released = false + return segs.segments, func() { + if released { + return + } + segs.lock.Unlock() + released = true + } +} + +func (s *RoSnapshots) ViewSingleFile(t snaptype.Type, blockNum uint64) (segment *Segment, ok bool, release func()) { + segs, ok := s.segments.Get(t.Enum()) + if !ok { + return nil, false, noop + } + + segs.lock.RLock() + var released = false + for _, seg := range segs.segments { + if !(blockNum >= seg.from && blockNum < seg.to) { + continue + } + return seg, true, func() { + if released { + return + } + segs.lock.Unlock() + released = true + } + } + segs.lock.RUnlock() + return nil, false, noop +} + func (v *View) Segments(t snaptype.Type) []*Segment { if s, ok := v.s.segments.Get(t.Enum()); ok { return s.segments From e98c07531ba65be7759656dac0034a64d385352a Mon Sep 17 00:00:00 2001 From: "alex.sharov" Date: Mon, 15 Jul 2024 12:47:01 +0700 Subject: [PATCH 4/7] save --- .../snapshotsync/freezeblocks/block_reader.go | 69 ++++++++++--------- 1 file changed, 36 insertions(+), 33 deletions(-) diff --git a/turbo/snapshotsync/freezeblocks/block_reader.go b/turbo/snapshotsync/freezeblocks/block_reader.go index 9ef9d1b2709..51f0b6f4db6 100644 --- a/turbo/snapshotsync/freezeblocks/block_reader.go +++ b/turbo/snapshotsync/freezeblocks/block_reader.go @@ -386,12 +386,11 @@ func (r *BlockReader) HeaderByNumber(ctx context.Context, tx kv.Getter, blockHei } } - view := r.sn.View() - defer view.Close() - seg, ok := view.HeadersSegment(blockHeight) + seg, ok, release := r.sn.ViewSingleFile(coresnaptype.Headers, blockHeight) if !ok { return } + defer release() h, _, err = r.headerFromSnapshot(blockHeight, seg, nil) if err != nil { @@ -410,9 +409,8 @@ func (r *BlockReader) HeaderByHash(ctx context.Context, tx kv.Getter, hash commo return h, nil } - view := r.sn.View() - defer view.Close() - segments := view.Headers() + segments, release := r.sn.ViewType(coresnaptype.Headers) + defer release() buf := make([]byte, 128) for i := len(segments) - 1; i >= 0; i-- { @@ -442,12 +440,11 @@ func (r *BlockReader) CanonicalHash(ctx context.Context, tx kv.Getter, blockHeig return h, nil } - view := r.sn.View() - defer view.Close() - seg, ok := view.HeadersSegment(blockHeight) + seg, ok, release := r.sn.ViewSingleFile(coresnaptype.Headers, blockHeight) if !ok { return } + defer release() header, _, err := r.headerFromSnapshot(blockHeight, seg, nil) if err != nil { @@ -477,12 +474,12 @@ func (r *BlockReader) Header(ctx context.Context, tx kv.Getter, hash common.Hash } } - view := r.sn.View() - defer view.Close() - seg, ok := view.HeadersSegment(blockHeight) + seg, ok, release := r.sn.ViewSingleFile(coresnaptype.Headers, blockHeight) if !ok { return } + defer release() + h, _, err = r.headerFromSnapshot(blockHeight, seg, nil) if err != nil { return h, err @@ -517,40 +514,46 @@ func (r *BlockReader) BodyWithTransactions(ctx context.Context, tx kv.Getter, ha } } - view := r.sn.View() - defer view.Close() - - var baseTxnID uint64 - var txsAmount uint32 - var buf []byte - seg, ok := view.BodiesSegment(blockHeight) + seg, ok, release := r.sn.ViewSingleFile(coresnaptype.Bodies, blockHeight) if !ok { if dbgLogs { log.Info(dbgPrefix + "no bodies file for this block num") } return nil, nil } - body, baseTxnID, txsAmount, buf, err = r.bodyFromSnapshot(blockHeight, seg, buf) + defer release() + + var baseTxnID uint64 + var txCount uint32 + var buf []byte + body, baseTxnID, txCount, buf, err = r.bodyFromSnapshot(blockHeight, seg, buf) if err != nil { return nil, err } + release() + if body == nil { if dbgLogs { log.Info(dbgPrefix + "got nil body from file") } return nil, nil } - txnSeg, ok := view.TxsSegment(blockHeight) + + txnSeg, ok, release := r.sn.ViewSingleFile(coresnaptype.Transactions, blockHeight) if !ok { if dbgLogs { log.Info(dbgPrefix+"no transactions file for this block num", "r.sn.BlocksAvailable()", r.sn.BlocksAvailable(), "r.sn.idxMax", r.sn.idxMax.Load(), "r.sn.segmetntsMax", r.sn.segmentsMax.Load()) } return nil, nil } - txs, senders, err := r.txsFromSnapshot(baseTxnID, txsAmount, txnSeg, buf) + defer release() + + txs, senders, err := r.txsFromSnapshot(baseTxnID, txCount, txnSeg, buf) if err != nil { return nil, err } + release() + if txs == nil { if dbgLogs { log.Info(dbgPrefix + "got nil txs from file") @@ -586,13 +589,13 @@ func (r *BlockReader) Body(ctx context.Context, tx kv.Getter, hash common.Hash, body, _, txAmount = rawdb.ReadBody(tx, hash, blockHeight) return body, txAmount, nil } - view := r.sn.View() - defer view.Close() - seg, ok := view.BodiesSegment(blockHeight) + seg, ok, release := r.sn.ViewSingleFile(coresnaptype.Bodies, blockHeight) if !ok { return } + defer release() + body, _, txAmount, _, err = r.bodyFromSnapshot(blockHeight, seg, nil) if err != nil { return nil, 0, err @@ -983,18 +986,18 @@ func (r *BlockReader) TxnByIdxInBlock(ctx context.Context, tx kv.Getter, blockNu return rawdb.TxnByIdxInBlock(tx, canonicalHash, blockNum, txIdxInBlock) } - view := r.sn.View() - defer view.Close() - seg, ok := view.BodiesSegment(blockNum) + seg, ok, release := r.sn.ViewSingleFile(coresnaptype.Bodies, blockNum) if !ok { return } + defer release() var b *types.BodyForStorage b, _, err = r.bodyForStorageFromSnapshot(blockNum, seg, nil) if err != nil { return nil, err } + release() if b == nil { return } @@ -1004,10 +1007,12 @@ func (r *BlockReader) TxnByIdxInBlock(ctx context.Context, tx kv.Getter, blockNu return nil, nil } - txnSeg, ok := view.TxsSegment(blockNum) + txnSeg, ok, release := r.sn.ViewSingleFile(coresnaptype.Transactions, blockNum) if !ok { return } + defer release() + // +1 because block has system-txn in the beginning of block return r.txnByID(b.BaseTxId+1+uint64(txIdxInBlock), txnSeg, nil) } @@ -1033,13 +1038,11 @@ func (r *BlockReader) TxnLookup(_ context.Context, tx kv.Getter, txnHash common. } func (r *BlockReader) FirstTxnNumNotInSnapshots() uint64 { - view := r.sn.View() - defer view.Close() - - sn, ok := view.TxsSegment(r.sn.BlocksAvailable()) + sn, ok, release := r.sn.ViewSingleFile(coresnaptype.Transactions, r.sn.BlocksAvailable()) if !ok { return 0 } + defer release() lastTxnID := sn.Index(coresnaptype.Indexes.TxnHash).BaseDataID() + uint64(sn.Count()) return lastTxnID From 299c6dd11bdc8db0bf0b1cebd54b16fafedac5d0 Mon Sep 17 00:00:00 2001 From: "alex.sharov" Date: Mon, 15 Jul 2024 13:15:12 +0700 Subject: [PATCH 5/7] save --- turbo/snapshotsync/freezeblocks/block_snapshots.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/turbo/snapshotsync/freezeblocks/block_snapshots.go b/turbo/snapshotsync/freezeblocks/block_snapshots.go index 501c2e3884c..b0c3220ae64 100644 --- a/turbo/snapshotsync/freezeblocks/block_snapshots.go +++ b/turbo/snapshotsync/freezeblocks/block_snapshots.go @@ -2074,7 +2074,7 @@ func (s *RoSnapshots) ViewType(t snaptype.Type) (segments []*Segment, release fu if released { return } - segs.lock.Unlock() + segs.lock.RUnlock() released = true } } @@ -2095,7 +2095,7 @@ func (s *RoSnapshots) ViewSingleFile(t snaptype.Type, blockNum uint64) (segment if released { return } - segs.lock.Unlock() + segs.lock.RUnlock() released = true } } From 4688127b959581d86858060891c9ac7235fcddf6 Mon Sep 17 00:00:00 2001 From: "alex.sharov" Date: Mon, 15 Jul 2024 16:27:52 +0700 Subject: [PATCH 6/7] save --- .../snapshotsync/freezeblocks/block_reader.go | 48 +++++++++---------- 1 file changed, 24 insertions(+), 24 deletions(-) diff --git a/turbo/snapshotsync/freezeblocks/block_reader.go b/turbo/snapshotsync/freezeblocks/block_reader.go index 51f0b6f4db6..060ec99de9b 100644 --- a/turbo/snapshotsync/freezeblocks/block_reader.go +++ b/turbo/snapshotsync/freezeblocks/block_reader.go @@ -10,6 +10,7 @@ import ( "sort" "time" + borsnaptype "github.com/ledgerwatch/erigon/polygon/bor/snaptype" "github.com/ledgerwatch/log/v3" "github.com/ledgerwatch/erigon-lib/common/hexutility" @@ -1256,10 +1257,10 @@ func (r *BlockReader) BorStartEventID(ctx context.Context, tx kv.Tx, hash common } borTxHash := bortypes.ComputeBorTxHash(blockHeight, hash) - view := r.borSn.View() - defer view.Close() - segments := view.Events() + segments, release := r.borSn.ViewType(borsnaptype.BorEvents) + defer release() + for i := len(segments) - 1; i >= 0; i-- { sn := segments[i] if sn.from > blockHeight { @@ -1337,9 +1338,10 @@ func (r *BlockReader) EventsByBlock(ctx context.Context, tx kv.Tx, hash common.H return result, nil } borTxHash := bortypes.ComputeBorTxHash(blockHeight, hash) - view := r.borSn.View() - defer view.Close() - segments := view.Events() + + segments, release := r.borSn.ViewType(borsnaptype.BorEvents) + defer release() + var buf []byte result := []rlp.RawValue{} for i := len(segments) - 1; i >= 0; i-- { @@ -1377,10 +1379,9 @@ func (r *BlockReader) EventsByBlock(ctx context.Context, tx kv.Tx, hash common.H // EventsByIdFromSnapshot returns the list of records limited by time, or the number of records along with a bool value to signify if the records were limited by time func (r *BlockReader) EventsByIdFromSnapshot(from uint64, to time.Time, limit int) ([]*heimdall.EventRecordWithTime, bool, error) { - view := r.borSn.View() - defer view.Close() + segments, release := r.borSn.ViewType(borsnaptype.BorEvents) + defer release() - segments := view.Events() var buf []byte var result []*heimdall.EventRecordWithTime stateContract := bor.GenesisContractStateReceiverABI() @@ -1457,9 +1458,9 @@ func (r *BlockReader) LastFrozenEventId() uint64 { return 0 } - view := r.borSn.View() - defer view.Close() - segments := view.Events() + segments, release := r.borSn.ViewType(borsnaptype.BorEvents) + defer release() + if len(segments) == 0 { return 0 } @@ -1514,9 +1515,9 @@ func (r *BlockReader) LastFrozenSpanId() uint64 { return 0 } - view := r.borSn.View() - defer view.Close() - segments := view.Spans() + segments, release := r.borSn.ViewType(borsnaptype.BorSpans) + defer release() + if len(segments) == 0 { return 0 } @@ -1558,9 +1559,9 @@ func (r *BlockReader) Span(ctx context.Context, tx kv.Getter, spanId uint64) ([] } return common.Copy(v), nil } - view := r.borSn.View() - defer view.Close() - segments := view.Spans() + segments, release := r.borSn.ViewType(borsnaptype.BorSpans) + defer release() + for i := len(segments) - 1; i >= 0; i-- { sn := segments[i] idx := sn.Index() @@ -1662,9 +1663,9 @@ func (r *BlockReader) Checkpoint(ctx context.Context, tx kv.Getter, checkpointId return common.Copy(v), nil } - view := r.borSn.View() - defer view.Close() - segments := view.Checkpoints() + segments, release := r.borSn.ViewType(borsnaptype.BorCheckpoints) + defer release() + for i := len(segments) - 1; i >= 0; i-- { sn := segments[i] index := sn.Index() @@ -1688,9 +1689,8 @@ func (r *BlockReader) LastFrozenCheckpointId() uint64 { return 0 } - view := r.borSn.View() - defer view.Close() - segments := view.Checkpoints() + segments, release := r.borSn.ViewType(borsnaptype.BorCheckpoints) + defer release() if len(segments) == 0 { return 0 } From 8017d052debe15eb72101bc76f3d688098afd08a Mon Sep 17 00:00:00 2001 From: "alex.sharov" Date: Mon, 15 Jul 2024 16:29:46 +0700 Subject: [PATCH 7/7] save --- turbo/snapshotsync/freezeblocks/block_snapshots.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/turbo/snapshotsync/freezeblocks/block_snapshots.go b/turbo/snapshotsync/freezeblocks/block_snapshots.go index b0c3220ae64..2c203945194 100644 --- a/turbo/snapshotsync/freezeblocks/block_snapshots.go +++ b/turbo/snapshotsync/freezeblocks/block_snapshots.go @@ -2040,6 +2040,7 @@ type View struct { closed bool } +// ViewSingleFile - RLock files of all types func (s *RoSnapshots) View() *View { v := &View{s: s, baseSegType: coresnaptype.Headers} s.segments.Scan(func(segtype snaptype.Enum, value *segments) bool { @@ -2062,6 +2063,7 @@ func (v *View) Close() { var noop = func() {} +// ViewSingleFile - RLock all files of given type func (s *RoSnapshots) ViewType(t snaptype.Type) (segments []*Segment, release func()) { segs, ok := s.segments.Get(t.Enum()) if !ok { @@ -2079,6 +2081,7 @@ func (s *RoSnapshots) ViewType(t snaptype.Type) (segments []*Segment, release fu } } +// ViewSingleFile - RLock all files of given type if has file with `blockNum` func (s *RoSnapshots) ViewSingleFile(t snaptype.Type, blockNum uint64) (segment *Segment, ok bool, release func()) { segs, ok := s.segments.Get(t.Enum()) if !ok {