forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-30 00:23:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor: content storage

Browse source 
move ContentStorage interface to storage
This commit is contained in:
fearlseefe 2024-01-02 15:16:11 +08:00 committed by Chen Kai
parent 2b89219371
commit 7fb680502e
7 changed files with 401 additions and 398 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
go.sum
View file

@ -492,8 +492,6 @@ github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
github.com/optimism-java/utp-go v0.0.0-20231203033001-5a531e1e11a0 h1:fSjUuzS7gI3IXz5mo8opUlK+9UktElRy1MH5EweLg2k=
github.com/optimism-java/utp-go v0.0.0-20231203033001-5a531e1e11a0/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
github.com/optimism-java/utp-go v0.0.0-20231225095152-5a9690d82b58 h1:EZfd3NpJV+CL5vORquJ2O6eAUjkpI7+ge9a9n9HMyYE=
github.com/optimism-java/utp-go v0.0.0-20231225095152-5a9690d82b58/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=

10
p2p/discover/portal_protocol.go
View file

@ -68,12 +68,6 @@ var ErrNilContentKey = errors.New("content key cannot be nil")
var ContentNotFound = storage.ErrContentNotFound
type ContentStorage interface {
Get(contentId []byte) ([]byte, error)
Put(contentId []byte, content []byte) error
}
type ContentElement struct {
Node enode.ID
ContentKeys [][]byte
@ -144,7 +138,7 @@ type PortalProtocol struct {
radiusCache *fastcache.Cache
closeCtx context.Context
cancelCloseCtx context.CancelFunc
storage ContentStorage
storage storage.ContentStorage
toContentId func(contentKey []byte) []byte
contentQueue chan *ContentElement
@ -155,7 +149,7 @@ func defaultContentIdFunc(contentKey []byte) []byte {
return digest[:]
}
func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateKey *ecdsa.PrivateKey, storage ContentStorage, contentQueue chan *ContentElement, opts ...PortalProtocolOption) (*PortalProtocol, error) {
func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateKey *ecdsa.PrivateKey, storage storage.ContentStorage, contentQueue chan *ContentElement, opts ...PortalProtocolOption) (*PortalProtocol, error) {
nodeDB, err := enode.OpenDB(config.NodeDBPath)
if err != nil {
return nil, err

387
portalnetwork/storage/content_storage.go
View file

@ -1,388 +1,11 @@
package storage
import (
"bytes"
"database/sql"
"errors"
"fmt"
"math/big"
"path"
"strings"
import "fmt"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/portalnetwork/utils"
"github.com/holiman/uint256"
sqlite3 "github.com/mattn/go-sqlite3"
)
var ErrContentNotFound = fmt.Errorf("content not found")
const (
sqliteName = "shisui.sqlite"
contentDeletionFraction = 0.05 // 5% of the content will be deleted when the storage capacity is hit and radius gets adjusted.
// SQLite Statements
createSql = `CREATE TABLE IF NOT EXISTS kvstore (
key BLOB PRIMARY KEY,
value BLOB
);`
getSql = "SELECT value FROM kvstore WHERE key = (?1);"
putSql = "INSERT OR REPLACE INTO kvstore (key, value) VALUES (?1, ?2);"
deleteSql = "DELETE FROM kvstore WHERE key = (?1);"
containSql = "SELECT 1 FROM kvstore WHERE key = (?1);"
getAllOrderedByDistanceSql = "SELECT key, length(value), xor(key, (?1)) as distance FROM kvstore ORDER BY distance DESC;"
deleteOutOfRadiusStmt = "DELETE FROM kvstore WHERE greater(xor(key, (?1)), (?2)) = 1"
XOR_FIND_FARTHEST_QUERY = `SELECT
xor(key, (?1)) as distance
FROM kvstore
ORDER BY distance DESC`
)
type ContentStorage interface {
Get(contentId []byte) ([]byte, error)
var (
ErrContentNotFound = fmt.Errorf("content not found")
maxDistance = uint256.MustFromHex("0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff")
)
type ContentStorage struct {
nodeId enode.ID
nodeDataDir string
storageCapacityInBytes uint64
radius *uint256.Int
sqliteDB *sql.DB
getStmt *sql.Stmt
putStmt *sql.Stmt
delStmt *sql.Stmt
containStmt *sql.Stmt
log log.Logger
}
func xor(contentId, nodeId []byte) []byte {
// length of contentId maybe not 32bytes
padding := make([]byte, 32)
if len(contentId) != len(nodeId) {
copy(padding, contentId)
} else {
padding = contentId
}
res := make([]byte, len(padding))
for i := range padding {
res[i] = padding[i] ^ nodeId[i]
}
return res
}
// a > b return 1; a = b return 0; else return -1
func greater(a, b []byte) int {
return bytes.Compare(a, b)
}
func NewContentStorage(storageCapacityInBytes uint64, nodeId enode.ID, nodeDataDir string) (*ContentStorage, error) {
// avoid repeated register in tests
registered := false
drives := sql.Drivers()
for _, v := range drives {
if v == "sqlite3_custom" {
registered = true
}
}
if !registered {
sql.Register("sqlite3_custom", &sqlite3.SQLiteDriver{
ConnectHook: func(conn *sqlite3.SQLiteConn) error {
if err := conn.RegisterFunc("xor", xor, false); err != nil {
return err
}
if err := conn.RegisterFunc("greater", greater, false); err != nil {
return err
}
return nil
},
})
}
sqlDb, err := sql.Open("sqlite3_custom", path.Join(nodeDataDir, sqliteName))
if err != nil {
return nil, err
}
portalStorage := &ContentStorage{
nodeId: nodeId,
nodeDataDir: nodeDataDir,
storageCapacityInBytes: storageCapacityInBytes,
radius: maxDistance,
sqliteDB: sqlDb,
log: log.New("protocol_storage"),
}
err = portalStorage.createTable()
if err != nil {
return nil, err
}
err = portalStorage.initStmts()
// Check whether we already have data, and use it to set radius
return portalStorage, err
}
// Get the content according to the contentId
func (p *ContentStorage) Get(contentId []byte) ([]byte, error) {
var res []byte
err := p.getStmt.QueryRow(contentId).Scan(&res)
if err == sql.ErrNoRows {
return nil, ErrContentNotFound
}
return res, err
}
type PutResult struct {
err error
pruned bool
count int
}
func (p *PutResult) Err() error {
return p.err
}
func (p *PutResult) Pruned() bool {
return p.pruned
}
func (p *PutResult) PrunedCount() int {
return p.count
}
func newPutResultWithErr(err error) PutResult {
return PutResult{
err: err,
}
}
// Put saves the contentId and content
func (p *ContentStorage) Put(contentId []byte, content []byte) PutResult {
_, err := p.putStmt.Exec(contentId, content)
if err != nil {
return newPutResultWithErr(err)
}
dbSize, err := p.UsedSize()
if err != nil {
return newPutResultWithErr(err)
}
if dbSize > p.storageCapacityInBytes {
count, err := p.deleteContentFraction(contentDeletionFraction)
//
if err != nil {
log.Warn("failed to delete oversize item")
return newPutResultWithErr(err)
}
return PutResult{pruned: true, count: count}
}
return PutResult{}
}
func (p *ContentStorage) Close() error {
p.getStmt.Close()
p.putStmt.Close()
p.delStmt.Close()
p.containStmt.Close()
return p.sqliteDB.Close()
}
func (p *ContentStorage) createTable() error {
stat, err := p.sqliteDB.Prepare(createSql)
if err != nil {
return err
}
defer stat.Close()
_, err = stat.Exec()
return err
}
func (p *ContentStorage) initStmts() error {
var stat *sql.Stmt
var err error
if stat, err = p.sqliteDB.Prepare(getSql); err != nil {
return nil
}
p.getStmt = stat
if stat, err = p.sqliteDB.Prepare(putSql); err != nil {
return nil
}
p.putStmt = stat
if stat, err = p.sqliteDB.Prepare(deleteSql); err != nil {
return nil
}
p.delStmt = stat
if stat, err = p.sqliteDB.Prepare(containSql); err != nil {
return nil
}
p.containStmt = stat
return nil
}
// Size get database size, content size and similar
func (p *ContentStorage) Size() (uint64, error) {
sql := "SELECT page_count * page_size as size FROM pragma_page_count(), pragma_page_size();"
stmt, err := p.sqliteDB.Prepare(sql)
if err != nil {
return 0, err
}
var res uint64
err = stmt.QueryRow().Scan(&res)
return res, err
}
func (p *ContentStorage) UnusedSize() (uint64, error) {
sql := "SELECT freelist_count * page_size as size FROM pragma_freelist_count(), pragma_page_size();"
return p.queryRowUint64(sql)
}
// UsedSize = Size - UnusedSize
func (p *ContentStorage) UsedSize() (uint64, error) {
size, err := p.Size()
if err != nil {
return 0, err
}
unusedSize, err := p.UnusedSize()
if err != nil {
return 0, err
}
return size - unusedSize, err
}
// ContentCount return the total content count
func (p *ContentStorage) ContentCount() (uint64, error) {
sql := "SELECT COUNT(key) FROM kvstore;"
return p.queryRowUint64(sql)
}
func (p *ContentStorage) ContentSize() (uint64, error) {
sql := "SELECT SUM(length(value)) FROM kvstore"
return p.queryRowUint64(sql)
}
func (p *ContentStorage) queryRowUint64(sql string) (uint64, error) {
// sql := "SELECT SUM(length(value)) FROM kvstore"
stmt, err := p.sqliteDB.Prepare(sql)
if err != nil {
return 0, err
}
var res uint64
err = stmt.QueryRow().Scan(&res)
return res, err
}
// GetLargestDistance find the largest distance
func (p *ContentStorage) GetLargestDistance() (*uint256.Int, error) {
stmt, err := p.sqliteDB.Prepare(XOR_FIND_FARTHEST_QUERY)
if err != nil {
return nil, err
}
var distance []byte
err = stmt.QueryRow(p.nodeId[:]).Scan(&distance)
if err != nil {
return nil, err
}
// reverse the distance, because big.SetBytes is big-endian
utils.ReverseBytesInPlace(distance)
bigNum := new(big.Int).SetBytes(distance)
res := uint256.MustFromBig(bigNum)
return res, nil
}
// EstimateNewRadius calculates an estimated new radius based on the current radius, used size, and storage capacity.
// The method takes the currentRadius as input and returns the estimated new radius and an error (if any).
// It calculates the size ratio of usedSize to storageCapacityInBytes and adjusts the currentRadius accordingly.
// If the size ratio is greater than 0, it performs the adjustment; otherwise, it returns the currentRadius unchanged.
// The method returns an error if there is any issue in determining the used size.
func (p *ContentStorage) EstimateNewRadius(currentRadius *uint256.Int) (*uint256.Int, error) {
currrentSize, err := p.UsedSize()
if err != nil {
return nil, err
}
sizeRatio := currrentSize / p.storageCapacityInBytes
if sizeRatio > 0 {
bigFormat := new(big.Int).SetUint64(sizeRatio)
return new(uint256.Int).Div(currentRadius, uint256.MustFromBig(bigFormat)), nil
}
return currentRadius, nil
}
func (p *ContentStorage) deleteContentFraction(fraction float64) (deleteCount int, err error) {
if fraction <= 0 || fraction >= 1 {
return deleteCount, errors.New("fraction should be between 0 and 1")
}
totalContentSize, err := p.ContentSize()
if err != nil {
return deleteCount, err
}
bytesToDelete := uint64(fraction * float64(totalContentSize))
// deleteElements := 0
deleteBytes := 0
rows, err := p.sqliteDB.Query(getAllOrderedByDistanceSql, p.nodeId[:])
if err != nil {
return deleteCount, err
}
defer rows.Close()
idsToDelete := make([][]byte, 0)
for deleteBytes < int(bytesToDelete) && rows.Next() {
var contentId []byte
var payloadLen int
var distance []byte
err = rows.Scan(&contentId, &payloadLen, &distance)
if err != nil {
return deleteCount, err
}
idsToDelete = append(idsToDelete, contentId)
// err = p.del(contentId)
if err != nil {
return deleteCount, err
}
deleteBytes += payloadLen
deleteCount++
}
// row must close first, or database is locked
// rows.Close() can call multi times
rows.Close()
err = p.batchDel(idsToDelete)
return
}
func (p *ContentStorage) del(contentId []byte) error {
_, err := p.delStmt.Exec(contentId)
return err
}
func (p *ContentStorage) batchDel(ids [][]byte) error {
query := "DELETE FROM kvstore WHERE key IN (?" + strings.Repeat(", ?", len(ids)-1) + ")"
args := make([]interface{}, len(ids))
for i, id := range ids {
args[i] = id
}
// delete items
_, err := p.sqliteDB.Exec(query, args...)
return err
}
// ReclaimSpace reclaims space in the ContentStorage's SQLite database by performing a VACUUM operation.
// It returns an error if the VACUUM operation encounters any issues.
func (p *ContentStorage) ReclaimSpace() error {
_, err := p.sqliteDB.Exec("VACUUM;")
return err
}
func (p *ContentStorage) deleteContentOutOfRadius(radius *uint256.Int) error {
res, err := p.sqliteDB.Exec(deleteOutOfRadiusStmt, p.nodeId[:], radius.Bytes())
count, _ := res.RowsAffected()
p.log.Trace("delete %d items", count)
return err
}
// ForcePrune delete the content which distance is further than the given radius
func (p *ContentStorage) ForcePrune(radius *uint256.Int) error {
return p.deleteContentOutOfRadius(radius)
Put(contentId []byte, content []byte) error
}

387
portalnetwork/storage/sqlite/content_storage.go Normal file
View file

@ -0,0 +1,387 @@
package sqlite
import (
"bytes"
"database/sql"
"errors"
"math/big"
"path"
"strings"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/portalnetwork/storage"
"github.com/ethereum/go-ethereum/portalnetwork/utils"
"github.com/holiman/uint256"
sqlite3 "github.com/mattn/go-sqlite3"
)
const (
sqliteName = "shisui.sqlite"
contentDeletionFraction = 0.05 // 5% of the content will be deleted when the storage capacity is hit and radius gets adjusted.
// SQLite Statements
createSql = `CREATE TABLE IF NOT EXISTS kvstore (
key BLOB PRIMARY KEY,
value BLOB
);`
getSql = "SELECT value FROM kvstore WHERE key = (?1);"
putSql = "INSERT OR REPLACE INTO kvstore (key, value) VALUES (?1, ?2);"
deleteSql = "DELETE FROM kvstore WHERE key = (?1);"
containSql = "SELECT 1 FROM kvstore WHERE key = (?1);"
getAllOrderedByDistanceSql = "SELECT key, length(value), xor(key, (?1)) as distance FROM kvstore ORDER BY distance DESC;"
deleteOutOfRadiusStmt = "DELETE FROM kvstore WHERE greater(xor(key, (?1)), (?2)) = 1"
XOR_FIND_FARTHEST_QUERY = `SELECT
xor(key, (?1)) as distance
FROM kvstore
ORDER BY distance DESC`
)
var (
maxDistance = uint256.MustFromHex("0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff")
)
type ContentStorage struct {
nodeId enode.ID
nodeDataDir string
storageCapacityInBytes uint64
radius *uint256.Int
sqliteDB *sql.DB
getStmt *sql.Stmt
putStmt *sql.Stmt
delStmt *sql.Stmt
containStmt *sql.Stmt
log log.Logger
}
func xor(contentId, nodeId []byte) []byte {
// length of contentId maybe not 32bytes
padding := make([]byte, 32)
if len(contentId) != len(nodeId) {
copy(padding, contentId)
} else {
padding = contentId
}
res := make([]byte, len(padding))
for i := range padding {
res[i] = padding[i] ^ nodeId[i]
}
return res
}
// a > b return 1; a = b return 0; else return -1
func greater(a, b []byte) int {
return bytes.Compare(a, b)
}
func NewContentStorage(storageCapacityInBytes uint64, nodeId enode.ID, nodeDataDir string) (*ContentStorage, error) {
// avoid repeated register in tests
registered := false
drives := sql.Drivers()
for _, v := range drives {
if v == "sqlite3_custom" {
registered = true
}
}
if !registered {
sql.Register("sqlite3_custom", &sqlite3.SQLiteDriver{
ConnectHook: func(conn *sqlite3.SQLiteConn) error {
if err := conn.RegisterFunc("xor", xor, false); err != nil {
return err
}
if err := conn.RegisterFunc("greater", greater, false); err != nil {
return err
}
return nil
},
})
}
sqlDb, err := sql.Open("sqlite3_custom", path.Join(nodeDataDir, sqliteName))
if err != nil {
return nil, err
}
portalStorage := &ContentStorage{
nodeId: nodeId,
nodeDataDir: nodeDataDir,
storageCapacityInBytes: storageCapacityInBytes,
radius: maxDistance,
sqliteDB: sqlDb,
log: log.New("protocol_storage"),
}
err = portalStorage.createTable()
if err != nil {
return nil, err
}
err = portalStorage.initStmts()
// Check whether we already have data, and use it to set radius
return portalStorage, err
}
// Get the content according to the contentId
func (p *ContentStorage) Get(contentId []byte) ([]byte, error) {
var res []byte
err := p.getStmt.QueryRow(contentId).Scan(&res)
if err == sql.ErrNoRows {
return nil, storage.ErrContentNotFound
}
return res, err
}
type PutResult struct {
err error
pruned bool
count int
}
func (p *PutResult) Err() error {
return p.err
}
func (p *PutResult) Pruned() bool {
return p.pruned
}
func (p *PutResult) PrunedCount() int {
return p.count
}
func newPutResultWithErr(err error) PutResult {
return PutResult{
err: err,
}
}
// Put saves the contentId and content
func (p *ContentStorage) Put(contentId []byte, content []byte) PutResult {
_, err := p.putStmt.Exec(contentId, content)
if err != nil {
return newPutResultWithErr(err)
}
dbSize, err := p.UsedSize()
if err != nil {
return newPutResultWithErr(err)
}
if dbSize > p.storageCapacityInBytes {
count, err := p.deleteContentFraction(contentDeletionFraction)
//
if err != nil {
log.Warn("failed to delete oversize item")
return newPutResultWithErr(err)
}
return PutResult{pruned: true, count: count}
}
return PutResult{}
}
func (p *ContentStorage) Close() error {
p.getStmt.Close()
p.putStmt.Close()
p.delStmt.Close()
p.containStmt.Close()
return p.sqliteDB.Close()
}
func (p *ContentStorage) createTable() error {
stat, err := p.sqliteDB.Prepare(createSql)
if err != nil {
return err
}
defer stat.Close()
_, err = stat.Exec()
return err
}
func (p *ContentStorage) initStmts() error {
var stat *sql.Stmt
var err error
if stat, err = p.sqliteDB.Prepare(getSql); err != nil {
return nil
}
p.getStmt = stat
if stat, err = p.sqliteDB.Prepare(putSql); err != nil {
return nil
}
p.putStmt = stat
if stat, err = p.sqliteDB.Prepare(deleteSql); err != nil {
return nil
}
p.delStmt = stat
if stat, err = p.sqliteDB.Prepare(containSql); err != nil {
return nil
}
p.containStmt = stat
return nil
}
// Size get database size, content size and similar
func (p *ContentStorage) Size() (uint64, error) {
sql := "SELECT page_count * page_size as size FROM pragma_page_count(), pragma_page_size();"
stmt, err := p.sqliteDB.Prepare(sql)
if err != nil {
return 0, err
}
var res uint64
err = stmt.QueryRow().Scan(&res)
return res, err
}
func (p *ContentStorage) UnusedSize() (uint64, error) {
sql := "SELECT freelist_count * page_size as size FROM pragma_freelist_count(), pragma_page_size();"
return p.queryRowUint64(sql)
}
// UsedSize = Size - UnusedSize
func (p *ContentStorage) UsedSize() (uint64, error) {
size, err := p.Size()
if err != nil {
return 0, err
}
unusedSize, err := p.UnusedSize()
if err != nil {
return 0, err
}
return size - unusedSize, err
}
// ContentCount return the total content count
func (p *ContentStorage) ContentCount() (uint64, error) {
sql := "SELECT COUNT(key) FROM kvstore;"
return p.queryRowUint64(sql)
}
func (p *ContentStorage) ContentSize() (uint64, error) {
sql := "SELECT SUM(length(value)) FROM kvstore"
return p.queryRowUint64(sql)
}
func (p *ContentStorage) queryRowUint64(sql string) (uint64, error) {
// sql := "SELECT SUM(length(value)) FROM kvstore"
stmt, err := p.sqliteDB.Prepare(sql)
if err != nil {
return 0, err
}
var res uint64
err = stmt.QueryRow().Scan(&res)
return res, err
}
// GetLargestDistance find the largest distance
func (p *ContentStorage) GetLargestDistance() (*uint256.Int, error) {
stmt, err := p.sqliteDB.Prepare(XOR_FIND_FARTHEST_QUERY)
if err != nil {
return nil, err
}
var distance []byte
err = stmt.QueryRow(p.nodeId[:]).Scan(&distance)
if err != nil {
return nil, err
}
// reverse the distance, because big.SetBytes is big-endian
utils.ReverseBytesInPlace(distance)
bigNum := new(big.Int).SetBytes(distance)
res := uint256.MustFromBig(bigNum)
return res, nil
}
// EstimateNewRadius calculates an estimated new radius based on the current radius, used size, and storage capacity.
// The method takes the currentRadius as input and returns the estimated new radius and an error (if any).
// It calculates the size ratio of usedSize to storageCapacityInBytes and adjusts the currentRadius accordingly.
// If the size ratio is greater than 0, it performs the adjustment; otherwise, it returns the currentRadius unchanged.
// The method returns an error if there is any issue in determining the used size.
func (p *ContentStorage) EstimateNewRadius(currentRadius *uint256.Int) (*uint256.Int, error) {
currrentSize, err := p.UsedSize()
if err != nil {
return nil, err
}
sizeRatio := currrentSize / p.storageCapacityInBytes
if sizeRatio > 0 {
bigFormat := new(big.Int).SetUint64(sizeRatio)
return new(uint256.Int).Div(currentRadius, uint256.MustFromBig(bigFormat)), nil
}
return currentRadius, nil
}
func (p *ContentStorage) deleteContentFraction(fraction float64) (deleteCount int, err error) {
if fraction <= 0 || fraction >= 1 {
return deleteCount, errors.New("fraction should be between 0 and 1")
}
totalContentSize, err := p.ContentSize()
if err != nil {
return deleteCount, err
}
bytesToDelete := uint64(fraction * float64(totalContentSize))
// deleteElements := 0
deleteBytes := 0
rows, err := p.sqliteDB.Query(getAllOrderedByDistanceSql, p.nodeId[:])
if err != nil {
return deleteCount, err
}
defer rows.Close()
idsToDelete := make([][]byte, 0)
for deleteBytes < int(bytesToDelete) && rows.Next() {
var contentId []byte
var payloadLen int
var distance []byte
err = rows.Scan(&contentId, &payloadLen, &distance)
if err != nil {
return deleteCount, err
}
idsToDelete = append(idsToDelete, contentId)
// err = p.del(contentId)
if err != nil {
return deleteCount, err
}
deleteBytes += payloadLen
deleteCount++
}
// row must close first, or database is locked
// rows.Close() can call multi times
rows.Close()
err = p.batchDel(idsToDelete)
return
}
func (p *ContentStorage) del(contentId []byte) error {
_, err := p.delStmt.Exec(contentId)
return err
}
func (p *ContentStorage) batchDel(ids [][]byte) error {
query := "DELETE FROM kvstore WHERE key IN (?" + strings.Repeat(", ?", len(ids)-1) + ")"
args := make([]interface{}, len(ids))
for i, id := range ids {
args[i] = id
}
// delete items
_, err := p.sqliteDB.Exec(query, args...)
return err
}
// ReclaimSpace reclaims space in the ContentStorage's SQLite database by performing a VACUUM operation.
// It returns an error if the VACUUM operation encounters any issues.
func (p *ContentStorage) ReclaimSpace() error {
_, err := p.sqliteDB.Exec("VACUUM;")
return err
}
func (p *ContentStorage) deleteContentOutOfRadius(radius *uint256.Int) error {
res, err := p.sqliteDB.Exec(deleteOutOfRadiusStmt, p.nodeId[:], radius.Bytes())
count, _ := res.RowsAffected()
p.log.Trace("delete %d items", count)
return err
}
// ForcePrune delete the content which distance is further than the given radius
func (p *ContentStorage) ForcePrune(radius *uint256.Int) error {
return p.deleteContentOutOfRadius(radius)
}

13
portalnetwork/storage/content_storage_test.go → portalnetwork/storage/sqlite/content_storage_test.go
View file

@ -1,4 +1,4 @@
package storage
package sqlite
import (
"fmt"
@ -7,6 +7,7 @@ import (
"testing"
"github.com/ethereum/go-ethereum/p2p/enode"
contentStorage "github.com/ethereum/go-ethereum/portalnetwork/storage"
"github.com/holiman/uint256"
"github.com/stretchr/testify/assert"
)
@ -36,7 +37,7 @@ func TestBasicStorage(t *testing.T) {
content := []byte("value")
_, err = storage.Get(contentId)
assert.Equal(t, ErrContentNotFound, err)
assert.Equal(t, contentStorage.ErrContentNotFound, err)
pt := storage.Put(contentId, content)
assert.NoError(t, pt.Err())
@ -177,10 +178,10 @@ func TestDBPruning(t *testing.T) {
assert.True(t, usedSize < storage.storageCapacityInBytes)
_, err = storage.Get(furthestElement.Bytes())
assert.Equal(t, ErrContentNotFound, err)
assert.Equal(t, contentStorage.ErrContentNotFound, err)
_, err = storage.Get(secondFurthest.Bytes())
assert.Equal(t, ErrContentNotFound, err)
assert.Equal(t, contentStorage.ErrContentNotFound, err)
val, err := storage.Get(thirdFurthest.Bytes())
assert.NoError(t, err)
@ -241,10 +242,10 @@ func TestSimpleForcePruning(t *testing.T) {
assert.NoError(t, err)
_, err = storage.Get(furthestElement.Bytes())
assert.Equal(t, ErrContentNotFound, err)
assert.Equal(t, contentStorage.ErrContentNotFound, err)
_, err = storage.Get(secondFurthest.Bytes())
assert.Equal(t, ErrContentNotFound, err)
assert.Equal(t, contentStorage.ErrContentNotFound, err)
_, err = storage.Get(third.Bytes())
assert.NoError(t, err)

BIN
portalnetwork/storage/sqlite/shisui.sqlite Normal file
View file

Binary file not shown.

BIN
portalnetwork/storage/sqlite/shisui.sqlite-journal Normal file
View file

Binary file not shown.