mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-27 15:16:43 +00:00
feat: store size in each Put
This commit is contained in:
parent
7fd095c4a6
commit
1ac486bb08
1 changed files with 32 additions and 46 deletions
|
|
@ -4,8 +4,8 @@ import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"encoding/binary"
|
"encoding/binary"
|
||||||
"runtime"
|
"runtime"
|
||||||
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/cockroachdb/pebble"
|
"github.com/cockroachdb/pebble"
|
||||||
"github.com/cockroachdb/pebble/bloom"
|
"github.com/cockroachdb/pebble/bloom"
|
||||||
|
|
@ -123,10 +123,9 @@ type ContentStorage struct {
|
||||||
radius atomic.Value
|
radius atomic.Value
|
||||||
log log.Logger
|
log log.Logger
|
||||||
db *pebble.DB
|
db *pebble.DB
|
||||||
size uint64
|
size atomic.Uint64
|
||||||
sizeChan chan struct{}
|
|
||||||
capacityChan chan uint64
|
|
||||||
writeOptions *pebble.WriteOptions
|
writeOptions *pebble.WriteOptions
|
||||||
|
bytePool sync.Pool
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewPeppleStorage(config PeppleStorageConfig) (storage.ContentStorage, error) {
|
func NewPeppleStorage(config PeppleStorageConfig) (storage.ContentStorage, error) {
|
||||||
|
|
@ -135,9 +134,13 @@ func NewPeppleStorage(config PeppleStorageConfig) (storage.ContentStorage, error
|
||||||
db: config.DB,
|
db: config.DB,
|
||||||
storageCapacityInBytes: config.StorageCapacityMB * 1000_000,
|
storageCapacityInBytes: config.StorageCapacityMB * 1000_000,
|
||||||
log: log.New("storage", config.NetworkName),
|
log: log.New("storage", config.NetworkName),
|
||||||
sizeChan: make(chan struct{}, 1),
|
|
||||||
capacityChan: make(chan uint64, 100),
|
|
||||||
writeOptions: &pebble.WriteOptions{Sync: false},
|
writeOptions: &pebble.WriteOptions{Sync: false},
|
||||||
|
bytePool: sync.Pool{
|
||||||
|
New: func() interface{} {
|
||||||
|
out := make([]byte, 8)
|
||||||
|
return &out
|
||||||
|
},
|
||||||
|
},
|
||||||
}
|
}
|
||||||
cs.radius.Store(storage.MaxDistance)
|
cs.radius.Store(storage.MaxDistance)
|
||||||
|
|
||||||
|
|
@ -148,7 +151,7 @@ func NewPeppleStorage(config PeppleStorageConfig) (storage.ContentStorage, error
|
||||||
if err == nil {
|
if err == nil {
|
||||||
size := binary.BigEndian.Uint64(val)
|
size := binary.BigEndian.Uint64(val)
|
||||||
// init stage, no need to use lock
|
// init stage, no need to use lock
|
||||||
cs.size = size
|
cs.size.Store(size)
|
||||||
}
|
}
|
||||||
|
|
||||||
iter, err := cs.db.NewIter(nil)
|
iter, err := cs.db.NewIter(nil)
|
||||||
|
|
@ -165,9 +168,6 @@ func NewPeppleStorage(config PeppleStorageConfig) (storage.ContentStorage, error
|
||||||
}
|
}
|
||||||
cs.radius.Store(dis)
|
cs.radius.Store(dis)
|
||||||
}
|
}
|
||||||
|
|
||||||
cs.sizeChan <- struct{}{}
|
|
||||||
go cs.saveCapacity()
|
|
||||||
return cs, nil
|
return cs, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -195,25 +195,33 @@ func (c *ContentStorage) Put(contentKey []byte, contentId []byte, content []byte
|
||||||
if !valid {
|
if !valid {
|
||||||
return storage.ErrInsufficientRadius
|
return storage.ErrInsufficientRadius
|
||||||
}
|
}
|
||||||
|
length := uint64(len(contentId)) + uint64(len(content))
|
||||||
|
newSize := c.size.Add(length)
|
||||||
|
|
||||||
err = c.db.Set(distance, content, c.writeOptions)
|
buf := c.bytePool.Get().(*[]byte)
|
||||||
|
defer c.bytePool.Put(buf)
|
||||||
|
binary.BigEndian.PutUint64(*buf, newSize)
|
||||||
|
batch := c.db.NewBatch()
|
||||||
|
|
||||||
|
err = batch.Set(storage.SizeKey, *buf, c.writeOptions)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
err = batch.Set(distance, content, c.writeOptions)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
err = batch.Commit(c.writeOptions)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
length := uint64(len(contentId)) + uint64(len(content))
|
if newSize > c.storageCapacityInBytes {
|
||||||
<-c.sizeChan
|
|
||||||
c.size += length
|
|
||||||
if c.size > c.storageCapacityInBytes {
|
|
||||||
err := c.prune()
|
err := c.prune()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.sizeChan <- struct{}{}
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
} else {
|
|
||||||
c.capacityChan <- c.size
|
|
||||||
}
|
}
|
||||||
c.sizeChan <- struct{}{}
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -224,31 +232,6 @@ func (c *ContentStorage) Radius() *uint256.Int {
|
||||||
return val
|
return val
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *ContentStorage) saveCapacity() {
|
|
||||||
ticker := time.NewTicker(5 * time.Second)
|
|
||||||
defer ticker.Stop()
|
|
||||||
capacityChanged := false
|
|
||||||
var currentCapacity uint64 = 0
|
|
||||||
buf := make([]byte, 8) // uint64
|
|
||||||
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-ticker.C:
|
|
||||||
if capacityChanged {
|
|
||||||
binary.BigEndian.PutUint64(buf, currentCapacity)
|
|
||||||
err := c.db.Set(storage.SizeKey, buf, c.writeOptions)
|
|
||||||
if err != nil {
|
|
||||||
c.log.Error("save capacity failed", "error", err)
|
|
||||||
}
|
|
||||||
capacityChanged = false
|
|
||||||
}
|
|
||||||
case capacity := <-c.capacityChan:
|
|
||||||
capacityChanged = true
|
|
||||||
currentCapacity = capacity
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *ContentStorage) prune() error {
|
func (c *ContentStorage) prune() error {
|
||||||
expectSize := uint64(float64(c.storageCapacityInBytes) * contentDeletionFraction)
|
expectSize := uint64(float64(c.storageCapacityInBytes) * contentDeletionFraction)
|
||||||
var curentSize uint64 = 0
|
var curentSize uint64 = 0
|
||||||
|
|
@ -278,12 +261,15 @@ func (c *ContentStorage) prune() error {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
newSize := c.size.Add(-curentSize)
|
||||||
|
buf := c.bytePool.Get().(*[]byte)
|
||||||
|
defer c.bytePool.Put(buf)
|
||||||
|
binary.BigEndian.PutUint64(*buf, newSize)
|
||||||
|
batch.Set(storage.SizeKey, *buf, c.writeOptions)
|
||||||
err = batch.Commit(&pebble.WriteOptions{Sync: true})
|
err = batch.Commit(&pebble.WriteOptions{Sync: true})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
c.size -= curentSize
|
|
||||||
c.capacityChan <- c.size - curentSize
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue