save

Author: AskAlexSharov

db/kv/mdbx/kv_mdbx.go

@@ -36,10 +36,8 @@ import (
 	"github.com/c2h5oh/datasize"
 	"github.com/erigontech/mdbx-go/mdbx"
 	stack2 "github.com/go-stack/stack"
-	"golang.org/x/sync/errgroup"
 	"golang.org/x/sync/semaphore"
 
-	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/common/dbg"
 	"github.com/erigontech/erigon/common/dir"
 	"github.com/erigontech/erigon/common/estimate"
@@ -883,90 +881,7 @@ func (db *MdbxKV) WarmupTable(ctx context.Context, bucket string) {
 	if workers <= 0 {
 		return
 	}
-	const lvl = log.LvlInfo
-
-	var total, size uint64
-	var bounds [][]byte
-	if err := db.View(ctx, func(tx kv.Tx) error {
-		var err error
-		if total, err = tx.Count(bucket); err != nil {
-			return err
-		}
-		if size, err = tx.BucketSize(bucket); err != nil {
-			return err
-		}
-		if total == 0 {
-			return nil
-		}
-		b, err := tx.(*MdbxTx).SplitBucketByCount(bucket, nil, workers*4) // count-balanced via b-tree, even when keys cluster
-		if err != nil {
-			return err
-		}
-		bounds = make([][]byte, len(b)) // b's interior keys are zero-copy, valid only until this tx ends
-		for i, k := range b {
-			bounds[i] = bytes.Clone(k)
-		}
-		return nil
-	}); err != nil {
-		if !errors.Is(err, context.Canceled) {
-			db.log.Warn("[warmup] inspect failed", "table", bucket, "err", err)
-		}
-		return
-	}
-	if len(bounds) < 2 {
-		return
-	}
-
-	started := time.Now()
-	var progress, alive atomic.Int64
-	logEvery := time.NewTicker(20 * time.Second)
-	defer logEvery.Stop()
-
-	g, ctx := errgroup.WithContext(ctx)
-	g.SetLimit(workers)
-	for i := 0; i+1 < len(bounds); i++ {
-		from, to := bounds[i], bounds[i+1]
-		g.Go(func() error {
-			alive.Add(1)
-			defer alive.Add(-1)
-			return db.View(ctx, func(tx kv.Tx) error {
-				it, err := tx.Range(bucket, from, to, order.Asc, -1)
-				if err != nil {
-					return err
-				}
-				defer it.Close()
-				for it.HasNext() {
-					k, v, err := it.Next()
-					if err != nil {
-						return err
-					}
-					if len(v) > 0 {
-						_, _ = v[0], v[len(v)-1]
-					}
-					if len(k) > 0 {
-						_, _ = k[0], k[len(k)-1]
-					}
-					if p := progress.Add(1); p%128 == 0 {
-						select {
-						case <-ctx.Done():
-							return ctx.Err()
-						case <-logEvery.C:
-							pct := min(100, 100*float64(progress.Load())/float64(total))
-							db.log.Log(lvl, fmt.Sprintf("[warmup] Progress: %s %s %.2f%%, workers: %d/%d", bucket, common.ByteCount(size), pct, alive.Load(), workers))
-						default:
-						}
-					}
-				}
-				return nil
-			})
-		})
-	}
-	if err := g.Wait(); err != nil && !errors.Is(err, context.Canceled) {
-		db.log.Warn("[warmup] failed", "table", bucket, "err", err)
-	}
-	if took := time.Since(started); took > 30*time.Second {
-		db.log.Log(lvl, "[warmup] done", "table", bucket, "size", common.ByteCount(size), "keys", common.PrettyCounter(total), "took", took.Round(time.Second), "workers", workers)
-	}
+	kv.WarmupTable(ctx, db, bucket, workers)
 }
 
 // SplitBucketByCount partitions bucket into n approximately equal-COUNT key

db/kv/readahead.go

@@ -47,16 +47,31 @@ type ReadAheader struct {
 	consumerChunk atomic.Int64
 	cancel        context.CancelFunc
 	done          chan struct{}
+	label         string
+	full          bool // warm every chunk ignoring consumer position (whole-table warmup)
 }
 
 // NewReadAheader starts `workers` background prefetchers over `table` from `from`
 // (nil = table start). Call SetPos as the consumer advances, and Close when done.
 func NewReadAheader(ctx context.Context, db RoDB, table string, from []byte, workers int) *ReadAheader {
+	return newReadAheader(ctx, db, table, from, workers, "read-ahead", false)
+}
+
+// WarmupTable pulls the whole table into the OS page cache with `workers`
+// parallel readers and blocks until done or ctx is cancelled — a ReadAheader
+// with no consumer to throttle to, so every chunk gets warmed.
+func WarmupTable(ctx context.Context, db RoDB, table string, workers int) {
+	r := newReadAheader(ctx, db, table, nil, workers, "warmup", true)
+	<-r.done
+	r.cancel()
+}
+
+func newReadAheader(ctx context.Context, db RoDB, table string, from []byte, workers int, label string, full bool) *ReadAheader {
 	if workers < 1 {
 		workers = 1
 	}
 	ctx, cancel := context.WithCancel(ctx)
-	r := &ReadAheader{cancel: cancel, done: make(chan struct{})}
+	r := &ReadAheader{cancel: cancel, done: make(chan struct{}), label: label, full: full}
 	go r.run(ctx, db, table, from, workers)
 	return r
 }
@@ -68,6 +83,9 @@ func (r *ReadAheader) run(ctx context.Context, db RoDB, table string, from []byt
 	if len(bounds) < 2 {
 		return
 	}
+	if r.full {
+		ahead = int64(len(bounds)) // no consumer to follow: keep every chunk in the window
+	}
 	r.bounds.Store(&bounds)
 
 	var doneChunks, alive atomic.Int64
@@ -87,7 +105,7 @@ func (r *ReadAheader) run(ctx context.Context, db RoDB, table string, from []byt
 			doneChunks.Add(1) // count only chunks actually warmed, not the skipped ones
 			select {
 			case <-logEvery.C:
-				log.Info("[read-ahead]", "table", table, "consumer-chunk", r.consumerChunk.Load(), "done", doneChunks.Load(), "chunks", len(bounds)-1, "alive", alive.Load(), "workers", workers)
+				log.Info("["+r.label+"]", "table", table, "consumer-chunk", r.consumerChunk.Load(), "done", doneChunks.Load(), "chunks", len(bounds)-1, "alive", alive.Load(), "workers", workers)
 			default:
 			}
 			return nil