e3: option to discard all history writes. useful to test execution performance. (#752)

state/aggregator22.go

@@ -561,6 +561,18 @@ func (a *Aggregator22) Warmup(txFrom, limit uint64) {
 	}()
 }
 
+// StartWrites - pattern: `defer agg.StartWrites().FinishWrites()`
+func (a *Aggregator22) DiscardHistory() *Aggregator22 {
+	a.accounts.DiscardHistory(a.tmpdir)
+	a.storage.DiscardHistory(a.tmpdir)
+	a.code.DiscardHistory(a.tmpdir)
+	a.logAddrs.DiscardHistory(a.tmpdir)
+	a.logTopics.DiscardHistory(a.tmpdir)
+	a.tracesFrom.DiscardHistory(a.tmpdir)
+	a.tracesTo.DiscardHistory(a.tmpdir)
+	return a
+}
+
 // StartWrites - pattern: `defer agg.StartWrites().FinishWrites()`
 func (a *Aggregator22) StartWrites() *Aggregator22 {
 	a.accounts.StartWrites(a.tmpdir)
@@ -617,7 +629,7 @@ func (a *Aggregator22) CanPruneFrom(tx kv.Tx) uint64 {
 	return math2.MaxUint64
 }
 func (a *Aggregator22) Prune(ctx context.Context, limit uint64) error {
-	a.Warmup(0, cmp.Max(a.aggregationStep, limit)) // warmup is asyn and moving faster than data deletion
+	//a.Warmup(0, cmp.Max(a.aggregationStep, limit)) // warmup is asyn and moving faster than data deletion
 	defer func(t time.Time) {
 		if time.Since(t) > time.Second {
 			log.Debug(fmt.Sprintf("prune took: %s\n", time.Since(t)))

state/history.go

@@ -444,11 +444,18 @@ func (h *History) AddPrevValue(key1, key2, original []byte) (err error) {
 	h.walLock.RUnlock()
 	return err
 }
+
+func (h *History) DiscardHistory(tmpdir string) {
+	h.InvertedIndex.StartWrites(tmpdir)
+	h.walLock.Lock()
+	defer h.walLock.Unlock()
+	h.wal = h.newWriter(tmpdir, false, true)
+}
 func (h *History) StartWrites(tmpdir string) {
 	h.InvertedIndex.StartWrites(tmpdir)
 	h.walLock.Lock()
 	defer h.walLock.Unlock()
-	h.wal = h.newWriter(tmpdir)
+	h.wal = h.newWriter(tmpdir, true, false)
 }
 func (h *History) FinishWrites() {
 	h.InvertedIndex.FinishWrites()
@@ -462,7 +469,7 @@ func (h *History) Rotate() historyFlusher {
 	h.walLock.Lock()
 	defer h.walLock.Unlock()
 	w := h.wal
-	h.wal = h.newWriter(h.wal.tmpdir)
+	h.wal = h.newWriter(h.wal.tmpdir, h.wal.buffered, h.wal.discard)
 	h.wal.autoIncrement = w.autoIncrement
 	return historyFlusher{w, h.InvertedIndex.Rotate()}
 }
@@ -490,6 +497,7 @@ type historyWAL struct {
 	historyKey       []byte
 	autoIncrement    uint64
 	buffered         bool
+	discard          bool
 }
 
 func (h *historyWAL) close() {
@@ -499,14 +507,17 @@ func (h *historyWAL) close() {
 	h.historyVals.Close()
 }
 
-func (h *History) newWriter(tmpdir string) *historyWAL {
+func (h *History) newWriter(tmpdir string, buffered, discard bool) *historyWAL {
 	w := &historyWAL{h: h,
 		tmpdir:   tmpdir,
+		buffered: buffered,
+		discard:  discard,
+
 		autoIncrementBuf: make([]byte, 8),
 		historyKey:       make([]byte, 0, 128),
-
+	}
-		buffered:    true,
+	if buffered {
-		historyVals: etl.NewCollector(h.historyValsTable, tmpdir, etl.NewSortableBuffer(WALCollectorRam)),
+		w.historyVals = etl.NewCollector(h.historyValsTable, tmpdir, etl.NewSortableBuffer(WALCollectorRam))
 	}
 	w.historyVals.LogLvl(log.LvlTrace)
 
@@ -536,6 +547,10 @@ func (h *historyWAL) flush(tx kv.RwTx) error {
 }
 
 func (h *historyWAL) addPrevValue(key1, key2, original []byte) error {
+	if h.discard {
+		return nil
+	}
+
 	/*
 		lk := len(key1) + len(key2)
 		historyKey := make([]byte, lk+8)

state/inverted_index.go

@@ -321,10 +321,15 @@ func (ii *InvertedIndex) Add(key []byte) error {
 	return ii.add(key, key)
 }
 
+func (ii *InvertedIndex) DiscardHistory(tmpdir string) {
+	ii.walLock.Lock()
+	defer ii.walLock.Unlock()
+	ii.wal = ii.newWriter(tmpdir, false, true)
+}
 func (ii *InvertedIndex) StartWrites(tmpdir string) {
 	ii.walLock.Lock()
 	defer ii.walLock.Unlock()
-	ii.wal = ii.newWriter(tmpdir)
+	ii.wal = ii.newWriter(tmpdir, true, false)
 }
 func (ii *InvertedIndex) FinishWrites() {
 	ii.walLock.Lock()
@@ -338,7 +343,7 @@ func (ii *InvertedIndex) Rotate() *invertedIndexWAL {
 	defer ii.walLock.Unlock()
 	wal := ii.wal
 	if wal != nil {
-		ii.wal = ii.newWriter(ii.wal.tmpdir)
+		ii.wal = ii.newWriter(ii.wal.tmpdir, ii.wal.buffered, ii.wal.discard)
 	}
 	return wal
 }
@@ -349,6 +354,7 @@ type invertedIndexWAL struct {
 	indexKeys *etl.Collector
 	tmpdir    string
 	buffered  bool
+	discard   bool
 }
 
 // loadFunc - is analog of etl.Identity, but it signaling to etl - use .Put instead of .AppendDup - to allow duplicates
@@ -389,14 +395,17 @@ func init() {
 	}
 }
 
-func (ii *InvertedIndex) newWriter(tmpdir string) *invertedIndexWAL {
+func (ii *InvertedIndex) newWriter(tmpdir string, buffered, discard bool) *invertedIndexWAL {
 	w := &invertedIndexWAL{ii: ii,
-		buffered: true,
+		buffered: buffered,
+		discard:  discard,
 		tmpdir:   tmpdir,
+	}
+	if buffered {
 		// 3 history + 4 indices = 10 etl collectors, 10*256Mb/16 = 256mb - for all indices buffers
 		// etl collector doesn't fsync: means if have enough ram, all files produced by all collectors will be in ram
-		index:     etl.NewCollector(ii.indexTable, tmpdir, etl.NewSortableBuffer(WALCollectorRam)),
+		w.index = etl.NewCollector(ii.indexTable, tmpdir, etl.NewSortableBuffer(WALCollectorRam))
-		indexKeys: etl.NewCollector(ii.indexKeysTable, tmpdir, etl.NewSortableBuffer(WALCollectorRam)),
+		w.indexKeys = etl.NewCollector(ii.indexKeysTable, tmpdir, etl.NewSortableBuffer(WALCollectorRam))
 	}
 	w.index.LogLvl(log.LvlTrace)
 	w.indexKeys.LogLvl(log.LvlTrace)
@@ -404,6 +413,10 @@ func (ii *InvertedIndex) newWriter(tmpdir string) *invertedIndexWAL {
 }
 
 func (ii *invertedIndexWAL) add(key, indexKey []byte) error {
+	if ii.discard {
+		return nil
+	}
+
 	if ii.buffered {
 		if err := ii.indexKeys.Collect(ii.ii.txNumBytes[:], key); err != nil {
 			return err