mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2024-12-28 14:47:16 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Implemented ForAmount and MDBX sequencing (#497)

Browse Source
This commit is contained in:
Giulio rebuffo 2022-06-18 21:24:30 +02:00 committed by GitHub
parent 988a4b5510
commit d57a16c7f6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 24 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
kv/memdb/memory_mutation.go
View File

@ -15,7 +15,6 @@ package memdb
import (
"context"
"encoding/binary"
"fmt"
"github.com/ledgerwatch/log/v3"
@ -86,43 +85,32 @@ func (m *MemoryMutation) getMem(table string, key []byte) ([]byte, bool) {
func (m *MemoryMutation) DBSize() (uint64, error) { panic("not implemented") }
func (m *MemoryMutation) PageSize() uint64 { panic("not implemented") }
func (m *MemoryMutation) IncrementSequence(bucket string, amount uint64) (res uint64, err error) {
v, ok := m.getMem(kv.Sequence, []byte(bucket))
if !ok && m.db != nil {
v, err = m.db.GetOne(kv.Sequence, []byte(bucket))
if err != nil {
return 0, err
}
}
var currentV uint64 = 0
if len(v) > 0 {
currentV = binary.BigEndian.Uint64(v)
}
newVBytes := make([]byte, 8)
binary.BigEndian.PutUint64(newVBytes, currentV+amount)
if err = m.Put(kv.Sequence, []byte(bucket), newVBytes); err != nil {
func (m *MemoryMutation) IncrementSequence(bucket string, amount uint64) (uint64, error) {
memRes, err := m.memTx.ReadSequence(bucket)
if err != nil {
return 0, err
}
return currentV, nil
}
func (m *MemoryMutation) ReadSequence(bucket string) (res uint64, err error) {
v, ok := m.getMem(kv.Sequence, []byte(bucket))
if !ok && m.db != nil {
v, err = m.db.GetOne(kv.Sequence, []byte(bucket))
if memRes == 0 {
base, err := m.db.ReadSequence(bucket)
if err != nil {
return 0, err
}
if _, err := m.memTx.IncrementSequence(bucket, base); err != nil {
return 0, err
}
}
var currentV uint64 = 0
if len(v) > 0 {
currentV = binary.BigEndian.Uint64(v)
}
return m.memTx.IncrementSequence(bucket, amount)
}
return currentV, nil
func (m *MemoryMutation) ReadSequence(bucket string) (uint64, error) {
memRes, err := m.memTx.ReadSequence(bucket)
if err != nil {
return 0, err
}
if memRes == 0 {
return m.db.ReadSequence(bucket)
}
return memRes, nil
}
// Can only be called from the worker thread
@ -179,7 +167,7 @@ func (m *MemoryMutation) Put(table string, key []byte, value []byte) error {
}
func (m *MemoryMutation) Append(table string, key []byte, value []byte) error {
return m.Put(table, key, value)
return m.memTx.Append(table, key, value)
}
func (m *MemoryMutation) AppendDup(table string, key []byte, value []byte) error {
@ -202,7 +190,10 @@ func (m *MemoryMutation) ForPrefix(bucket string, prefix []byte, walker func(k,
func (m *MemoryMutation) ForAmount(bucket string, prefix []byte, amount uint32, walker func(k, v []byte) error) error {
m.panicOnEmptyDB()
return m.db.ForAmount(bucket, prefix, amount, walker)
if err := m.db.ForAmount(bucket, prefix, amount, walker); err != nil {
return err
}
return m.memTx.ForAmount(bucket, prefix, amount, walker)
}
func (m *MemoryMutation) Delete(table string, k, v []byte) error {