shed

Documentation

Overview

包棚提供了一个简单的抽象组件来组成 对按字段和索引组织的存储数据进行更复杂的操作。

唯一保存有关群存储块数据的逻辑信息的类型 元数据是项。这一部分主要不是为了 性能原因。

Example (Store)

示例存储使用shed包构造一个简单的存储实现。

//<developer>
//    <name>linapex 曹一峰</name>
//    <email>linapex@163.com</email>
//    <wx>superexc</wx>
//    <qqgroup>128148617</qqgroup>
//    <url>https://jsq.ink</url>
//    <role>pku engineer</role>
//    <date>2019-03-16 19:16:44</date>
//</624450117558079488>

package main

import (
	"bytes"
	"context"
	"encoding/binary"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"time"

	"github.com/ethereum/go-ethereum/swarm/shed"
	"github.com/ethereum/go-ethereum/swarm/storage"
	"github.com/syndtr/goleveldb/leveldb"
)

// 存储保留字段和索引（包括其编码功能）
// 并通过从中组合数据来定义对它们的操作。
// 它实现了storage.chunkstore接口。
// 这只是一个不支持并行操作的例子
// 或者现实世界的实现。
type Store struct {
	db *shed.DB

	//字段和索引
	schemaName     shed.StringField
	sizeCounter    shed.Uint64Field
	accessCounter  shed.Uint64Field
	retrievalIndex shed.Index
	accessIndex    shed.Index
	gcIndex        shed.Index
}

// 新的返回新的存储。所有字段和索引都已初始化
// 并检查与现有数据库中架构的可能冲突
// 自动地。
func New(path string) (s *Store, err error) {
	db, err := shed.NewDB(path, "")
	if err != nil {
		return nil, err
	}
	s = &Store{
		db: db,
	}
	//用任意名称标识当前存储架构。
	s.schemaName, err = db.NewStringField("schema-name")
	if err != nil {
		return nil, err
	}
	//块访问的全局不断递增索引。
	s.accessCounter, err = db.NewUint64Field("access-counter")
	if err != nil {
		return nil, err
	}
	//存储实际块地址、数据和存储时间戳的索引。
	s.retrievalIndex, err = db.NewIndex("Address->StoreTimestamp|Data", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			return fields.Address, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.Address = key
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			b := make([]byte, 8)
			binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
			value = append(b, fields.Data...)
			return value, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
			e.Data = value[8:]
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	//存储特定地址的访问时间戳的索引。
	//需要它来更新迭代顺序的GC索引键。
	s.accessIndex, err = db.NewIndex("Address->AccessTimestamp", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			return fields.Address, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.Address = key
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			b := make([]byte, 8)
			binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp))
			return b, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.AccessTimestamp = int64(binary.BigEndian.Uint64(value))
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	//索引键按访问时间戳排序，用于垃圾收集优先级。
	s.gcIndex, err = db.NewIndex("AccessTimestamp|StoredTimestamp|Address->nil", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			b := make([]byte, 16, 16+len(fields.Address))
			binary.BigEndian.PutUint64(b[:8], uint64(fields.AccessTimestamp))
			binary.BigEndian.PutUint64(b[8:16], uint64(fields.StoreTimestamp))
			key = append(b, fields.Address...)
			return key, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.AccessTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
			e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[8:16]))
			e.Address = key[16:]
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			return nil, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	return s, nil
}

// 放置存储块并设置它存储时间戳。
func (s *Store) Put(_ context.Context, ch storage.Chunk) (err error) {
	return s.retrievalIndex.Put(shed.Item{
		Address:        ch.Address(),
		Data:           ch.Data(),
		StoreTimestamp: time.Now().UTC().UnixNano(),
	})
}

// get使用提供的地址检索块。
// 它通过删除以前的索引来更新访问和GC索引
// 并添加新项作为索引项的键
// 被改变了。
func (s *Store) Get(_ context.Context, addr storage.Address) (c storage.Chunk, err error) {
	batch := new(leveldb.Batch)

	//获取块数据和存储时间戳。
	item, err := s.retrievalIndex.Get(shed.Item{
		Address: addr,
	})
	if err != nil {
		if err == leveldb.ErrNotFound {
			return nil, storage.ErrChunkNotFound
		}
		return nil, err
	}

	//获取区块访问时间戳。
	accessItem, err := s.accessIndex.Get(shed.Item{
		Address: addr,
	})
	switch err {
	case nil:
		//如果找到访问时间戳，则删除gc索引项。
		err = s.gcIndex.DeleteInBatch(batch, shed.Item{
			Address:         item.Address,
			StoreTimestamp:  accessItem.AccessTimestamp,
			AccessTimestamp: item.StoreTimestamp,
		})
		if err != nil {
			return nil, err
		}
	case leveldb.ErrNotFound:
		//找不到访问时间戳。不要做任何事。
		//这是firs get请求。
	default:
		return nil, err
	}

	//指定新的访问时间戳
	accessTimestamp := time.Now().UTC().UnixNano()

	//在访问索引中放置新的访问时间戳。
	err = s.accessIndex.PutInBatch(batch, shed.Item{
		Address:         addr,
		AccessTimestamp: accessTimestamp,
	})
	if err != nil {
		return nil, err
	}

	//在gc索引中放置新的访问时间戳。
	err = s.gcIndex.PutInBatch(batch, shed.Item{
		Address:         item.Address,
		AccessTimestamp: accessTimestamp,
		StoreTimestamp:  item.StoreTimestamp,
	})
	if err != nil {
		return nil, err
	}

	//递增访问计数器。
	//目前此信息不在任何地方使用。
	_, err = s.accessCounter.IncInBatch(batch)
	if err != nil {
		return nil, err
	}

	//编写批处理。
	err = s.db.WriteBatch(batch)
	if err != nil {
		return nil, err
	}

	//返回块。
	return storage.NewChunk(item.Address, item.Data), nil
}

// Collect垃圾是索引迭代的一个例子。
// 它不提供可靠的垃圾收集功能。
func (s *Store) CollectGarbage() (err error) {
	const maxTrashSize = 100
	maxRounds := 10 //任意数，需要计算

	//进行几轮GC测试。
	for roundCount := 0; roundCount < maxRounds; roundCount++ {
		var garbageCount int
		//新一轮CG的新批次。
		trash := new(leveldb.Batch)
		//遍历所有索引项，并在需要时中断。
		err = s.gcIndex.Iterate(func(item shed.Item) (stop bool, err error) {
			//移除区块。
			err = s.retrievalIndex.DeleteInBatch(trash, item)
			if err != nil {
				return false, err
			}
			//删除gc索引中的元素。
			err = s.gcIndex.DeleteInBatch(trash, item)
			if err != nil {
				return false, err
			}
			//删除访问索引中的关系。
			err = s.accessIndex.DeleteInBatch(trash, item)
			if err != nil {
				return false, err
			}
			garbageCount++
			if garbageCount >= maxTrashSize {
				return true, nil
			}
			return false, nil
		}, nil)
		if err != nil {
			return err
		}
		if garbageCount == 0 {
			return nil
		}
		err = s.db.WriteBatch(trash)
		if err != nil {
			return err
		}
	}
	return nil
}

// getSchema是检索最简单的
// 数据库字段中的字符串。
func (s *Store) GetSchema() (name string, err error) {
	name, err = s.schemaName.Get()
	if err == leveldb.ErrNotFound {
		return "", nil
	}
	return name, err
}

// getSchema是存储最简单的
// 数据库字段中的字符串。
func (s *Store) PutSchema(name string) (err error) {
	return s.schemaName.Put(name)
}

// 关闭关闭基础数据库。
func (s *Store) Close() error {
	return s.db.Close()
}

// 示例存储使用shed包构造一个简单的存储实现。
func main() {
	dir, err := ioutil.TempDir("", "ephemeral")
	if err != nil {
		log.Fatal(err)
	}
	defer os.RemoveAll(dir)

	s, err := New(dir)
	if err != nil {
		log.Fatal(err)
	}
	defer s.Close()

	ch := storage.GenerateRandomChunk(1024)
	err = s.Put(context.Background(), ch)
	if err != nil {
		log.Fatal(err)
	}

	got, err := s.Get(context.Background(), ch.Address())
	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(bytes.Equal(got.Data(), ch.Data()))

	// 输出：真
}

Index

• type DB
  • func NewDB(path string, metricsPrefix string) (db *DB, err error)
  • func (db *DB) Close() (err error)
  • func (db *DB) Delete(key []byte) (err error)
  • func (db *DB) Get(key []byte) (value []byte, err error)
  • func (db *DB) NewIndex(name string, funcs IndexFuncs) (f Index, err error)
  • func (db *DB) NewIterator() iterator.Iterator
  • func (db *DB) NewStringField(name string) (f StringField, err error)
  • func (db *DB) NewStructField(name string) (f StructField, err error)
  • func (db *DB) NewUint64Field(name string) (f Uint64Field, err error)
  • func (db *DB) Put(key []byte, value []byte) (err error)
  • func (db *DB) WriteBatch(batch *leveldb.Batch) (err error)
• type Index
  • func (f Index) Count() (count int, err error)
  • func (f Index) CountFrom(start Item) (count int, err error)
  • func (f Index) Delete(keyFields Item) (err error)
  • func (f Index) DeleteInBatch(batch *leveldb.Batch, keyFields Item) (err error)
  • func (f Index) Get(keyFields Item) (out Item, err error)
  • func (f Index) Iterate(fn IndexIterFunc, options *IterateOptions) (err error)
  • func (f Index) Put(i Item) (err error)
  • func (f Index) PutInBatch(batch *leveldb.Batch, i Item) (err error)
• type IndexFuncs
• type IndexIterFunc
• type Item
  • func (i Item) Merge(i2 Item) (new Item)
• type IterateOptions
• type StringField
  • func (f StringField) Get() (val string, err error)
  • func (f StringField) Put(val string) (err error)
  • func (f StringField) PutInBatch(batch *leveldb.Batch, val string)
• type StructField
  • func (f StructField) Get(val interface{}) (err error)
  • func (f StructField) Put(val interface{}) (err error)
  • func (f StructField) PutInBatch(batch *leveldb.Batch, val interface{}) (err error)
• type Uint64Field
  • func (f Uint64Field) Dec() (val uint64, err error)
  • func (f Uint64Field) DecInBatch(batch *leveldb.Batch) (val uint64, err error)
  • func (f Uint64Field) Get() (val uint64, err error)
  • func (f Uint64Field) Inc() (val uint64, err error)
  • func (f Uint64Field) IncInBatch(batch *leveldb.Batch) (val uint64, err error)
  • func (f Uint64Field) Put(val uint64) (err error)
  • func (f Uint64Field) PutInBatch(batch *leveldb.Batch, val uint64)

Examples

• Package (Store)

Types

type DB

type DB struct {
	// contains filtered or unexported fields
}

DB在级别数据库上提供抽象，以便 使用字段和有序索引实现复杂结构。 它提供了一个模式功能来存储字段和索引 有关命名和类型的信息。

func NewDB

func NewDB(path string, metricsPrefix string) (db *DB, err error)

new db构造新的db并验证架构 如果它存在于给定路径上的数据库中。 metricsPrefix用于为给定数据库收集度量。

func (*DB) Close

func (db *DB) Close() (err error)

关闭关闭级别数据库。

func (*DB) Delete

func (db *DB) Delete(key []byte) (err error)

DELETE包装级别DB DELETE方法以增加度量计数器。

func (*DB) Get

func (db *DB) Get(key []byte) (value []byte, err error)

get wrapps leveldb get方法以增加度量计数器。

func (*DB) NewIndex

func (db *DB) NewIndex(name string, funcs IndexFuncs) (f Index, err error)

new index返回一个新的索引实例，该实例具有定义的名称和 编码函数。名称必须唯一并将被验证 对于键前缀字节的数据库架构。

func (*DB) NewIterator

func (db *DB) NewIterator() iterator.Iterator

NewIterator将LevelDB NewIterator方法包装为增量度量计数器。

func (*DB) NewStringField

func (db *DB) NewStringField(name string) (f StringField, err error)

newStringField重新运行StringField的新实例。 它根据数据库模式验证其名称和类型。

func (*DB) NewStructField

func (db *DB) NewStructField(name string) (f StructField, err error)

NewstructField返回新的structField。 它根据数据库模式验证其名称和类型。

func (*DB) NewUint64Field

func (db *DB) NewUint64Field(name string) (f Uint64Field, err error)

newuint64字段返回新的uint64字段。 它根据数据库模式验证其名称和类型。

func (*DB) Put

func (db *DB) Put(key []byte, value []byte) (err error)

将Wrapps-LevelDB-Put方法添加到增量度量计数器。

func (*DB) WriteBatch

func (db *DB) WriteBatch(batch *leveldb.Batch) (err error)

WRITEBATCH将LEVELDB WRITE方法包装为增量度量计数器。

type Index

type Index struct {
	// contains filtered or unexported fields
}

索引表示一组具有公共 前缀。它具有编码和解码密钥和值的功能 要对包含以下内容的已保存数据提供透明操作： -获取特定项目 -保存特定项目 -在已排序的级别数据库键上迭代 它实现了indexIteratorInterface接口。

func (Index) Count

func (f Index) Count() (count int, err error)

count返回索引中的项数。

func (Index) CountFrom

func (f Index) CountFrom(start Item) (count int, err error)

CountFrom返回索引键中的项数 从从提供的项编码的密钥开始。

func (Index) Delete

func (f Index) Delete(keyFields Item) (err error)

删除接受项以删除键/值对 基于其字段的数据库。

func (Index) DeleteInBatch

func (f Index) DeleteInBatch(batch *leveldb.Batch, keyFields Item) (err error)

deleteinbatch与只删除操作相同 在批处理上而不是在数据库上执行。

func (Index) Get

func (f Index) Get(keyFields Item) (out Item, err error)

get接受表示为要检索的项的键字段 索引中的值并返回最大可用信息 从另一项表示的索引。

func (Index) Iterate

func (f Index) Iterate(fn IndexIterFunc, options *IterateOptions) (err error)

迭代函数迭代索引的键。 如果迭代次数为零，则迭代次数将覆盖所有键。

func (Index) Put

func (f Index) Put(i Item) (err error)

放置接受项以对来自它的信息进行编码 并保存到数据库。

func (Index) PutInBatch

func (f Index) PutInBatch(batch *leveldb.Batch, i Item) (err error)

putinbatch与put方法相同，但它只是 将键/值对保存到批处理中 直接到数据库。

type IndexFuncs

type IndexFuncs struct {
	EncodeKey   func(fields Item) (key []byte, err error)
	DecodeKey   func(key []byte) (e Item, err error)
	EncodeValue func(fields Item) (value []byte, err error)
	DecodeValue func(keyFields Item, value []byte) (e Item, err error)
}

indexfuncs结构定义了用于编码和解码的函数 特定索引的LevelDB键和值。

type IndexIterFunc

type IndexIterFunc func(item Item) (stop bool, err error)

indexiterfunc是对解码的每个项的回调 通过迭代索引键。 通过返回一个true for stop变量，迭代将 停止，返回错误，该错误将 传播到索引上被调用的迭代器方法。

type Item

type Item struct {
	Address         []byte
	Data            []byte
	AccessTimestamp int64
	StoreTimestamp  int64
	//useMockStore是用于标识
	//join函数中字段的未设置状态。
	UseMockStore *bool
}

项目包含与群块数据和元数据相关的字段。 群存储和操作所需的所有信息 必须在此处定义该存储。 这种结构在逻辑上与群存储相连， 这个包中唯一没有被概括的部分， 主要是出于性能原因。

项是用于检索、存储和编码的类型 块数据和元数据。它作为参数传递给索引编码 函数、获取函数和Put函数。 但是，它还返回来自get函数调用的附加数据 作为迭代器函数定义中的参数。

func (Item) Merge

func (i Item) Merge(i2 Item) (new Item)

合并是一个帮助器方法，用于构造新的 用默认值填充字段 一个特定项的值来自另一个项。

type IterateOptions

type IterateOptions struct {
	//StartFrom是从中开始迭代的项。
	StartFrom *Item
	//如果skipStartFromItem为true，则StartFrom项不会
	//重复进行。
	SkipStartFromItem bool
	//迭代键具有公共前缀的项。
	Prefix []byte
}

迭代器定义迭代器函数的可选参数。

type StringField

type StringField struct {
	// contains filtered or unexported fields
}

StringField是最简单的字段实现 它在特定的leveldb键下存储一个任意字符串。

func (StringField) Get

func (f StringField) Get() (val string, err error)

get返回数据库中的字符串值。 如果找不到该值，则返回空字符串 没有错误。

func (StringField) Put

func (f StringField) Put(val string) (err error)

将存储字符串放入数据库。

func (StringField) PutInBatch

func (f StringField) PutInBatch(batch *leveldb.Batch, val string)

putinbatch将字符串存储在可以 稍后保存在数据库中。

type StructField

type StructField struct {
	// contains filtered or unexported fields
}

structField是用于存储复杂结构的帮助程序 以rlp格式对其进行编码。

func (StructField) Get

func (f StructField) Get(val interface{}) (err error)

将数据库中的数据解包到提供的VAL。 如果找不到数据，则返回leveldb.errnotfound。

func (StructField) Put

func (f StructField) Put(val interface{}) (err error)

放入marshals提供的val并将其保存到数据库中。

func (StructField) PutInBatch

func (f StructField) PutInBatch(batch *leveldb.Batch, val interface{}) (err error)

Putinbatch Marshals提供了VAL并将其放入批处理中。

type Uint64Field

type Uint64Field struct {
	// contains filtered or unexported fields
}

uint64字段提供了一种在数据库中使用简单计数器的方法。 它透明地将uint64类型值编码为字节。

func (Uint64Field) Dec

func (f Uint64Field) Dec() (val uint64, err error)

DEC减少数据库中的uint64值。 此操作不是goroutine save。 该字段受到保护，不会溢出为负值。

func (Uint64Field) DecInBatch

func (f Uint64Field) DecInBatch(batch *leveldb.Batch) (val uint64, err error)

decinbatch在批处理中递减uint64值 通过从数据库中检索值，而不是同一批。 此操作不是goroutine save。 该字段受到保护，不会溢出为负值。

func (Uint64Field) Get

func (f Uint64Field) Get() (val uint64, err error)

get从数据库中检索uint64值。 如果在数据库中找不到该值，则为0 返回，没有错误。

func (Uint64Field) Inc

func (f Uint64Field) Inc() (val uint64, err error)

inc在数据库中增加uint64值。 此操作不是goroutine save。

func (Uint64Field) IncInBatch

func (f Uint64Field) IncInBatch(batch *leveldb.Batch) (val uint64, err error)

incinbatch在批中增加uint64值 通过从数据库中检索值，而不是同一批。 此操作不是goroutine save。

func (Uint64Field) Put

func (f Uint64Field) Put(val uint64) (err error)

放入编码uin64值并将其存储在数据库中。

func (Uint64Field) PutInBatch

func (f Uint64Field) PutInBatch(batch *leveldb.Batch, val uint64)

Putinbatch在批中存储uint64值 以后可以保存到数据库中。