mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-19 18:32:23 +00:00
369 lines
10 KiB
Go
369 lines
10 KiB
Go
// Copyright 2018 The go-ethereum Authors
|
|
// This file is part of the go-ethereum library.
|
|
//
|
|
// The go-ethereum library is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU Lesser General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// The go-ethereum library is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU Lesser General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU Lesser General Public License
|
|
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
package localstore
|
|
|
|
import (
|
|
"encoding/hex"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/swarm/shed"
|
|
"github.com/syndtr/goleveldb/leveldb"
|
|
)
|
|
|
|
// Mode enumerates different modes of access and update
|
|
// operations on a database.
|
|
type Mode int
|
|
|
|
// Modes of access and update.
|
|
const (
|
|
ModeSyncing Mode = iota
|
|
ModeUpload
|
|
ModeRequest
|
|
ModeSynced
|
|
// this modes are internal only
|
|
// they can be removed completely
|
|
// if accessors are not used internally
|
|
modeAccess
|
|
modeRemoval
|
|
)
|
|
|
|
// ModeName returns a descriptive name of a Mode.
|
|
// If the Mode is not known, a blank string is returned.
|
|
func ModeName(m Mode) (name string) {
|
|
switch m {
|
|
case ModeSyncing:
|
|
return "syncing"
|
|
case ModeUpload:
|
|
return "upload"
|
|
case ModeRequest:
|
|
return "request"
|
|
case ModeSynced:
|
|
return "synced"
|
|
case modeAccess:
|
|
return "access"
|
|
case modeRemoval:
|
|
return "removal"
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// access is called by an Accessor with a specific Mode.
|
|
// This function utilizes different indexes depending on
|
|
// the Mode.
|
|
func (db *DB) access(mode Mode, item shed.IndexItem) (out shed.IndexItem, err error) {
|
|
if db.useRetrievalCompositeIndex {
|
|
out, err = db.retrievalCompositeIndex.Get(item)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
} else {
|
|
// No need to get access timestamp here as it is used
|
|
// only for some of Modes in update and access time
|
|
// is not property of the chunk returned by the Accessor.Get.
|
|
out, err = db.retrievalDataIndex.Get(item)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
}
|
|
switch mode {
|
|
case ModeRequest, modeAccess:
|
|
// update the access timestamp and fc index
|
|
return out, db.updateOnAccess(mode, out)
|
|
default:
|
|
// all other modes are not updating the index
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
var (
|
|
updateLockTimeout = 3 * time.Second
|
|
updateLockCheckDelay = 30 * time.Microsecond
|
|
)
|
|
|
|
// update performs different operations on fields and indexes
|
|
// depending on the provided Mode. It is called in accessor
|
|
// put function.
|
|
// It protects parallel updates of items with the same address
|
|
// with updateLocks map and waiting using a simple for loop.
|
|
func (db *DB) update(mode Mode, item shed.IndexItem) (err error) {
|
|
// protect parallel updates
|
|
start := time.Now()
|
|
lockKey := hex.EncodeToString(item.Address)
|
|
for {
|
|
_, loaded := db.updateLocks.LoadOrStore(lockKey, struct{}{})
|
|
if !loaded {
|
|
break
|
|
}
|
|
time.Sleep(updateLockCheckDelay)
|
|
if time.Since(start) > updateLockTimeout {
|
|
return ErrUpdateLockTimeout
|
|
}
|
|
}
|
|
defer db.updateLocks.Delete(lockKey)
|
|
|
|
batch := new(leveldb.Batch)
|
|
|
|
switch mode {
|
|
case ModeSyncing:
|
|
// put to indexes: retrieve, pull
|
|
item.StoreTimestamp = now()
|
|
if db.useRetrievalCompositeIndex {
|
|
db.retrievalCompositeIndex.PutInBatch(batch, item)
|
|
} else {
|
|
db.retrievalDataIndex.PutInBatch(batch, item)
|
|
}
|
|
db.pullIndex.PutInBatch(batch, item)
|
|
|
|
case ModeUpload:
|
|
// put to indexes: retrieve, push, pull
|
|
item.StoreTimestamp = now()
|
|
if db.useRetrievalCompositeIndex {
|
|
db.retrievalCompositeIndex.PutInBatch(batch, item)
|
|
} else {
|
|
db.retrievalDataIndex.PutInBatch(batch, item)
|
|
}
|
|
db.pullIndex.PutInBatch(batch, item)
|
|
db.pushIndex.PutInBatch(batch, item)
|
|
|
|
case ModeRequest:
|
|
// putting a chunk on mode request does not do anything
|
|
return nil
|
|
|
|
case ModeSynced:
|
|
// delete from push, insert to gc
|
|
|
|
// need to get access timestamp here as it is not
|
|
// provided by the access function, and it is not
|
|
// a property of a chunk provided to Accessor.Put.
|
|
if db.useRetrievalCompositeIndex {
|
|
i, err := db.retrievalCompositeIndex.Get(item)
|
|
if err != nil {
|
|
if err == leveldb.ErrNotFound {
|
|
// chunk is not found,
|
|
// no need to update gc index
|
|
// just delete from the push index
|
|
// if it is there
|
|
db.pushIndex.DeleteInBatch(batch, item)
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
item.AccessTimestamp = i.AccessTimestamp
|
|
item.StoreTimestamp = i.StoreTimestamp
|
|
if item.AccessTimestamp == 0 {
|
|
// the chunk is not accessed before
|
|
// set access time for gc index
|
|
item.AccessTimestamp = now()
|
|
db.retrievalCompositeIndex.PutInBatch(batch, item)
|
|
} else {
|
|
// the chunk is accessed before
|
|
// remove the current gc index item
|
|
db.gcIndex.DeleteInBatch(batch, item)
|
|
atomic.AddInt64(&db.gcSize, -1)
|
|
}
|
|
} else {
|
|
i, err := db.retrievalDataIndex.Get(item)
|
|
if err != nil {
|
|
if err == leveldb.ErrNotFound {
|
|
// chunk is not found,
|
|
// no need to update gc index
|
|
// just delete from the push index
|
|
// if it is there
|
|
db.pushIndex.DeleteInBatch(batch, item)
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
item.StoreTimestamp = i.StoreTimestamp
|
|
|
|
i, err = db.retrievalAccessIndex.Get(item)
|
|
switch err {
|
|
case nil:
|
|
item.AccessTimestamp = i.AccessTimestamp
|
|
db.gcIndex.DeleteInBatch(batch, item)
|
|
atomic.AddInt64(&db.gcSize, -1)
|
|
case leveldb.ErrNotFound:
|
|
// the chunk is not accessed before
|
|
default:
|
|
return err
|
|
}
|
|
item.AccessTimestamp = now()
|
|
db.retrievalAccessIndex.PutInBatch(batch, item)
|
|
}
|
|
db.pushIndex.DeleteInBatch(batch, item)
|
|
db.gcIndex.PutInBatch(batch, item)
|
|
atomic.AddInt64(&db.gcSize, 1)
|
|
|
|
case modeAccess:
|
|
// putting a chunk on mode access does not do anything
|
|
return nil
|
|
|
|
case modeRemoval:
|
|
// delete from retrieve, pull, gc
|
|
|
|
// need to get access timestamp here as it is not
|
|
// provided by the access function, and it is not
|
|
// a property of a chunk provided to Accessor.Put.
|
|
if db.useRetrievalCompositeIndex {
|
|
i, err := db.retrievalCompositeIndex.Get(item)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
item.StoreTimestamp = i.StoreTimestamp
|
|
item.AccessTimestamp = i.AccessTimestamp
|
|
} else {
|
|
i, err := db.retrievalAccessIndex.Get(item)
|
|
switch err {
|
|
case nil:
|
|
item.AccessTimestamp = i.AccessTimestamp
|
|
case leveldb.ErrNotFound:
|
|
default:
|
|
return err
|
|
}
|
|
i, err = db.retrievalDataIndex.Get(item)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
item.StoreTimestamp = i.StoreTimestamp
|
|
}
|
|
if db.useRetrievalCompositeIndex {
|
|
db.retrievalCompositeIndex.DeleteInBatch(batch, item)
|
|
} else {
|
|
db.retrievalDataIndex.DeleteInBatch(batch, item)
|
|
db.retrievalAccessIndex.DeleteInBatch(batch, item)
|
|
}
|
|
db.pullIndex.DeleteInBatch(batch, item)
|
|
db.gcIndex.DeleteInBatch(batch, item)
|
|
// TODO: optimize in garbage collection
|
|
// get is too expensive operation
|
|
if _, err := db.gcIndex.Get(item); err == nil {
|
|
atomic.AddInt64(&db.gcSize, -1)
|
|
}
|
|
|
|
default:
|
|
return ErrInvalidMode
|
|
}
|
|
|
|
return db.shed.WriteBatch(batch)
|
|
}
|
|
|
|
// updateOnAccess is called in access function and performs
|
|
// different operations on fields and indexes depending on
|
|
// the provided Mode.
|
|
// This function is separated from the update function to prevent
|
|
// changes on calling accessor put function in access and request modes.
|
|
// It protects parallel updates of items with the same address
|
|
// with updateLocks map and waiting using a simple for loop.
|
|
func (db *DB) updateOnAccess(mode Mode, item shed.IndexItem) (err error) {
|
|
// protect parallel updates
|
|
start := time.Now()
|
|
lockKey := hex.EncodeToString(item.Address)
|
|
for {
|
|
_, loaded := db.updateLocks.LoadOrStore(lockKey, struct{}{})
|
|
if !loaded {
|
|
break
|
|
}
|
|
time.Sleep(updateLockCheckDelay)
|
|
if time.Since(start) > updateLockTimeout {
|
|
return ErrUpdateLockTimeout
|
|
}
|
|
}
|
|
defer db.updateLocks.Delete(lockKey)
|
|
|
|
batch := new(leveldb.Batch)
|
|
|
|
switch mode {
|
|
case ModeRequest:
|
|
// update accessTimeStamp in retrieve, gc
|
|
|
|
if db.useRetrievalCompositeIndex {
|
|
// access timestamp is already populated
|
|
// in the provided item, passed from access function.
|
|
} else {
|
|
i, err := db.retrievalAccessIndex.Get(item)
|
|
switch err {
|
|
case nil:
|
|
item.AccessTimestamp = i.AccessTimestamp
|
|
case leveldb.ErrNotFound:
|
|
// no chunk accesses
|
|
default:
|
|
return err
|
|
}
|
|
}
|
|
if item.AccessTimestamp == 0 {
|
|
// chunk is not yes synced
|
|
// do not add it to the gc index
|
|
return nil
|
|
}
|
|
// delete current entry from the gc index
|
|
db.gcIndex.DeleteInBatch(batch, item)
|
|
// update access timestamp
|
|
item.AccessTimestamp = now()
|
|
// update retrieve access index
|
|
if db.useRetrievalCompositeIndex {
|
|
db.retrievalCompositeIndex.PutInBatch(batch, item)
|
|
} else {
|
|
db.retrievalAccessIndex.PutInBatch(batch, item)
|
|
}
|
|
// add new entry to gc index
|
|
db.gcIndex.PutInBatch(batch, item)
|
|
|
|
// Q: modeAccess and ModeRequest are very similar, why do we need both?
|
|
case modeAccess:
|
|
// update accessTimeStamp in retrieve, pull, gc
|
|
|
|
if db.useRetrievalCompositeIndex {
|
|
// access timestamp is already populated
|
|
// in the provided item, passed from access function.
|
|
} else {
|
|
i, err := db.retrievalAccessIndex.Get(item)
|
|
switch err {
|
|
case nil:
|
|
item.AccessTimestamp = i.AccessTimestamp
|
|
case leveldb.ErrNotFound:
|
|
// no chunk accesses
|
|
default:
|
|
return err
|
|
}
|
|
}
|
|
// Q: why do we need to update this index?
|
|
db.pullIndex.PutInBatch(batch, item)
|
|
if item.AccessTimestamp == 0 {
|
|
// chunk is not yes synced
|
|
// do not add it to the gc index
|
|
return nil
|
|
}
|
|
// delete current entry from the gc index
|
|
db.gcIndex.DeleteInBatch(batch, item)
|
|
// update access timestamp
|
|
item.AccessTimestamp = now()
|
|
// update retrieve access index
|
|
if db.useRetrievalCompositeIndex {
|
|
db.retrievalCompositeIndex.PutInBatch(batch, item)
|
|
} else {
|
|
db.retrievalAccessIndex.PutInBatch(batch, item)
|
|
}
|
|
// add new entry to gc index
|
|
db.gcIndex.PutInBatch(batch, item)
|
|
|
|
default:
|
|
return ErrInvalidMode
|
|
}
|
|
|
|
return db.shed.WriteBatch(batch)
|
|
}
|