以太坊源码阅读-数据结构篇-ethdb

  interface.go中定义了一些基本的KV数据库接口

package ethdb

// Code using batches should try to add this much data to the batch.
// The value was determined empirically.
const IdealBatchSize = 100 * 1024

// Putter wraps the database write operation supported by both batches and regular databases.
type Putter interface {
Put(key []byte, value []byte) error
}

// Deleter wraps the database delete operation supported by both batches and regular databases.
type Deleter interface {
Delete(key []byte) error
}

// Database wraps all database operations. All methods are safe for concurrent use.
type Database interface {
Putter
Deleter
Get(key []byte) ([]byte, error)
Has(key []byte) (bool, error)
Close()
NewBatch() Batch
}

// Batch is a write-only database that commits changes to its host database
// when Write is called. Batch cannot be used concurrently.
type Batch interface {
Putter
Deleter
ValueSize() int // amount of data in the batch
Write() error
// Reset resets the batch for reuse
Reset()
}

memory_database.go
这个基本上就是封装了一个内存的Map结构。然后使用了一把锁来对多线程进行资源的保护。

type MemDatabase struct {
db map[string][]byte
lock sync.RWMutex
}

func NewMemDatabase() (*MemDatabase, error) {
return &MemDatabase{
db: make(map[string][]byte),
}, nil
}

func (db *MemDatabase) Put(key []byte, value []byte) error {
db.lock.Lock()
defer db.lock.Unlock()
db.db[string(key)] = common.CopyBytes(value)
return nil
}
func (db *MemDatabase) Has(key []byte) (bool, error) {
db.lock.RLock()
defer db.lock.RUnlock()

_, ok := db.db[string(key)]
return ok, nil

}
然后是Batch的操作。也比较简单，一看便明白。

type kv struct{ k, v []byte }
type memBatch struct {
db *MemDatabase
writes []kv
size int
}
func (b *memBatch) Put(key, value []byte) error {
b.writes = append(b.writes, kv{common.CopyBytes(key), common.CopyBytes(value)})
b.size += len(value)
return nil
}
func (b *memBatch) Write() error {
b.db.lock.Lock()
defer b.db.lock.Unlock()

for _, kv := range b.writes {
    b.db.db[string(kv.k)] = kv.v
}
return nil

}

database.go就是ethereum客户端使用的代码，封装了levelDB的接口。
type LDBDatabase struct {
fn string // filename for reporting
db *leveldb.DB // LevelDB instance

compTimeMeter    metrics.Meter // Meter for measuring the total time spent in database compaction
compReadMeter    metrics.Meter // Meter for measuring the data read during compaction
compWriteMeter   metrics.Meter // Meter for measuring the data written during compaction
writeDelayNMeter metrics.Meter // Meter for measuring the write delay number due to database compaction
writeDelayMeter  metrics.Meter // Meter for measuring the write delay duration due to database compaction
diskReadMeter    metrics.Meter // Meter for measuring the effective amount of data read
diskWriteMeter   metrics.Meter // Meter for measuring the effective amount of data written

quitLock sync.Mutex      // Mutex protecting the quit channel access
quitChan chan chan error // Quit channel to stop the metrics collection before closing the database

log log.Logger // Contextual logger tracking the database path

}

// NewLDBDatabase returns a LevelDB wrapped object.
func NewLDBDatabase(file string, cache int, handles int) (*LDBDatabase, error) {
logger := log.New("database", file)

// Ensure we have some minimal caching and file guarantees
if cache < 16 {
    cache = 16
}
if handles < 16 {
    handles = 16
}
logger.Info("Allocated cache and file handles", "cache", cache, "handles", handles)

// Open the db and recover any potential corruptions
db, err := leveldb.OpenFile(file, &opt.Options{
    OpenFilesCacheCapacity: handles,
    BlockCacheCapacity:     cache / 2 * opt.MiB,
    WriteBuffer:            cache / 4 * opt.MiB, // Two of these are used internally
    Filter:                 filter.NewBloomFilter(10),
})
if _, corrupted := err.(*errors.ErrCorrupted); corrupted {
    db, err = leveldb.RecoverFile(file, nil)
}
// (Re)check for errors and abort if opening of the db failed
if err != nil {
    return nil, err
}
return &LDBDatabase{
    fn:  file,
    db:  db,
    log: logger,
}, nil

}

put/Has/Get/Delete方法等，非常简洁，这里先粗略的看一下。
// Put puts the given key / value to the queue
func (db *LDBDatabase) Put(key []byte, value []byte) error {
return db.db.Put(key, value, nil)
}

Metrics的处理，还包括quitChan的创建，并且在协程中运行meter.
// Meter configures the database metrics collectors and
func (db *LDBDatabase) Meter(prefix string) {
if metrics.Enabled {
// Initialize all the metrics collector at the requested prefix
db.compTimeMeter = metrics.NewRegisteredMeter(prefix+"compact/time", nil)
db.compReadMeter = metrics.NewRegisteredMeter(prefix+"compact/input", nil)
db.compWriteMeter = metrics.NewRegisteredMeter(prefix+"compact/output", nil)
db.diskReadMeter = metrics.NewRegisteredMeter(prefix+"disk/read", nil)
db.diskWriteMeter = metrics.NewRegisteredMeter(prefix+"disk/write", nil)
}
// Initialize write delay metrics no matter we are in metric mode or not.
db.writeDelayMeter = metrics.NewRegisteredMeter(prefix+"compact/writedelay/duration", nil)
db.writeDelayNMeter = metrics.NewRegisteredMeter(prefix+"compact/writedelay/counter", nil)

// Create a quit channel for the periodic collector and run it
db.quitLock.Lock()
db.quitChan = make(chan chan error)
db.quitLock.Unlock()

go db.meter(3 * time.Second)

}