TinyKv Project1 Standalone KV-阿里云开发者社区

前言

project1还是比较简单的，project2的难道就陡增了。对于project1来说，本文更多的还是讲一些不了解的概念。在每一个project1开始之前，仔细阅读文档。

Project1 StandaloneKV 文档翻译

Project1 StandaloneKV

在这个项目中，我们将会在列族的支持下建立一个独立的 key/value 存储 gRPC 服务。Standalone 意味着只有一个节点，而不是一个分布式系统。列族（CF, Column family）是一个类似 key 命名空间的术语，即同一个 key 在不同列族中的值是不同的。你可以简单地将多个 CF 视为独立的小型数据库。CF 在 Project4 中被用来支持事务模型。

该服务支持四个基本操作：Put/Delete/Get/Scan，它维护了一个简单的 key/value Pairs 的数据库，其中 key and value 都是字符串。

Put：替换数据库中指定 CF 的某个 key 的 value。
Delete：删除指定 CF 的 key 的 value。
Get：获取指定 CF 的某个 key 的当前值。
Scan：获取指定 CF 的一系列 key 的 current value。

该项目可以分为2个步骤，包括:

实现一个独立的存储引擎。
实现原始的 key/value 服务处理程序。

代码

gRPC 服务在 kv/main.go 中被初始化，它包含一个 tinykv.Server，它提供了名为 TinyKv 的 gRPC 服务。它由 proto/proto/tinykvpb.proto 中的 protocol-buffer 定义，rpc 请求和响应的细节被定义在 proto/proto/kvrpcpb.proto 中。

一般来说，不需要改变 proto 文件，因为所有必要的字段都已经被定义了。但如果仍然需要改变，可以修改 proto 文件并运行 make proto 来更新 proto/pkg/xxx/xxx.pb.go 中生成的 go 相关代码。

此外，Server 依赖于一个 Storage，这是一个需要为独立存储引擎实现的接口，位于 kv/storage/standalone_storage/standalone_storage.go 中。一旦 StandaloneStorage 中实现了接口 Storage，就可以用它实现Server 的原始 key/value 服务。

实现独立的存储引擎

第一个任务是实现 badger key/value 的 API。gRPC 服务依赖于一个在 kv/storage/storage.go 中定义的 Storage。在这种情况下，独立的存储引擎只是由两个方法提供的 badger key/value API。

type Storage interface {
    // 省略其他的代码
    Write(ctx *kvrpcpb.Context, batch []Modify) error
    Reader(ctx *kvrpcpb.Context) (StorageReader, error)
}

Write：应该提供一种方式，将一系列的修改应用到内部状态，内部状态是一个 badger 的实例。
Reader：应该返回一个 StorageReader，支持 key/value 的单点读取和快照扫描的操作。
现在不需要考虑 kvrpcpb.Context，它在后面的项目中使用。

提示：

使用 badger.Txn 来实现 Reader 函数，因为 badger 提供的事务处理程序可以提供 keys 和 values 的一致快照。

Badger 没有给出对 CF 的支持。engine_util 包（kv/util/engine_util）通过给 keys 添加前缀来模拟 CF。例如，一个属于特定 CF 的 key 被存储为 $ { cf } _ $ { key } 。它封装了 badger 以提供对 CF 的操作，还提供了许多有用的辅助函数。所以应该通过 engine_util 提供的方法进行所有读写操作。阅读 util/engine_util/doc.go 可以了解更多。

TinyKV使用了 badger 原始版本的分支，并进行了一些修正，所以应该使用 github.com/Connor1996/badger 而不是 github.com/dgraph-io/badger。

不要忘记为 badger.Txn 调用 Discard()，并在销毁前关闭所有迭代器。

实现服务处理程序

这个项目的最后一步是用实现的存储引擎来构建原始的 key/value 服务处理程序，包括 RawGet / RawScan / RawPut / RawDelete。处理程序已经定义好了，只需要在 kv/server/raw_api.go 中补充实现。一旦完成，记得运行 make project1 以通过测试用例。

文档的重点内容

列族(CF Column family)

列族(CF Column family)：在文档的开始，所说的列族到底是什么意思？

Column Family，也叫 CF，这个概念从 HBase 中来，就是将多个列合并为一个CF进行管理。这样读取一行数据时，你可以按照 CF 加载列，不需要加载所有列（通常同一个CF的列会保存在同一个文件中，所以这样有很高的效率）。此外因为同一列的数据格式相同，你可以针对某种格式采用高效的压缩算法。

default_wxf
default_www
default_xxx
write_wxf
write_www
write_xxx
lock_wxf
lock_www
lock_xxx

从上面的样例可以看出，CF的本质，就是key的前缀，就是一个字符串，起命名空间的作用。

func KeyWithCF(cf string, key []byte) []byte {
  return append([]byte(cf+"_"), key...)
}

独立存储引擎

Write：应该提供一种方式，将一系列的修改应用到内部状态，内部状态是一个 badger 的实例。
Reader：应该返回一个 StorageReader，支持 key/value 的单点读取和快照扫描的操作。

看上面两句话的描述，其实核心就是，让我们使用基于badger实现write的操作，至于badger是怎么write的，我们不用管，我们只需要对badger进行一层封装即可。

对于Reader来说，核心就是让我们返回一个StorageReader接口，我们应该去实现这个接口然后返回，而这个接口如何实现，提示里也说了，使用 badger.Txn去实现该接口。

实现了这两个接口，那么文档中所说的独立存储引擎就实现了。

原始的 key/value 服务处理程序

原始的 key/value 服务处理程序(Put/Delete/Get/Scan)：其实就是对存储引擎的一层上层封装，有了这层封装，我们可以随意改变底层的存储引擎。使用则无需关注底层细节，只要遵循上层接口使用规范即可。

StandAloneStorage

通过文档中的提示2我们知道，engine_util对badger进行了封装，StandAloneStorage就是要在engine_util的基础上再封装一层。

// Engines keeps references to and data for the engines used by unistore.
// All engines are badger key/value databases.
// the Path fields are the filesystem path to where the data is stored.
type Engines struct {
  // Data, including data which is committed (i.e., committed across other nodes) and un-committed (i.e., only present
  // locally).
  Kv     *badger.DB
  KvPath string
  // Metadata used by Raft.
  Raft     *badger.DB
  RaftPath string
}

那么我们就再封装一层即可，为什么这里要加一个conf？因为我们创建的时候需要用到conf里面的参数，因为不知道后续会不会用到这个config，所以先保存起来。

type StandAloneStorage struct {
  // Your Data Here (1).
  engine *engine_util.Engines
  conf   *config.Config
}
func NewStandAloneStorage(conf *config.Config) *StandAloneStorage {
  // Your Code Here (1).
  dbPath := conf.DBPath
  kvPath := path.Join(dbPath, "kv")
  raftPath := path.Join(dbPath, "raft")
  kvDB := engine_util.CreateDB(kvPath, false)
  //project1没用到raft
  raftDB := engine_util.CreateDB(raftPath, true)
  //func CreateDB(path string, raft bool) *badger.DB
  //func NewEngines(kvEngine, raftEngine *badger.DB, kvPath, raftPath string) *Engines
  return &StandAloneStorage{
    engine: engine_util.NewEngines(kvDB, raftDB, kvPath, raftPath),
    conf:   conf,
  }
}

StandAloneStorage是一个实例，它应该去实现接口，即Storage接口。那么我们下面重点关注Write和Reader。

type Storage interface {
  Start() error
  Stop() error
  Write(ctx *kvrpcpb.Context, batch []Modify) error
  Reader(ctx *kvrpcpb.Context) (StorageReader, error)
}

Write

去看看Modify类型是什么

Write(ctx *kvrpcpb.Context, batch []Modify) error

Modify 本质上代表着Put和Delete两种操作，通过下面的函数可以看到，它是通过断言来区分Put还是Delete的。一个Modify对应一个kv，所以可以看到上面Write接口中，Modify是一个切片，那么对于Write，我们需要遍历range这个切片进行操作。

// Modify is a single modification to TinyKV's underlying storage.
type Modify struct {
  Data interface{}
}
type Put struct {
  Key   []byte
  Value []byte
  Cf    string
}
type Delete struct {
  Key []byte
  Cf  string
}
func (m *Modify) Key() []byte {
  switch m.Data.(type) {
  case Put:
    return m.Data.(Put).Key
  case Delete:
    return m.Data.(Delete).Key
  }
  return nil
}
func (m *Modify) Value() []byte {
  if putData, ok := m.Data.(Put); ok {
    return putData.Value
  }
  return nil
}
func (m *Modify) Cf() string {
  switch m.Data.(type) {
  case Put:
    return m.Data.(Put).Cf
  case Delete:
    return m.Data.(Delete).Cf
  }
  return ""
}

提示2中说了，engine_util包中提供了方法进行所有的读写操作，所以现在去看看提供了什么api供我们使用。

可以看到提供了两个函数给我们使用，看一下源码其实也能发现，在真实存储的时候，key被加了前缀：cf_key.

func PutCF(engine *badger.DB, cf string, key []byte, val []byte) error {
  return engine.Update(func(txn *badger.Txn) error {
    return txn.Set(KeyWithCF(cf, key), val)
  })
}
func DeleteCF(engine *badger.DB, cf string, key []byte) error {
  return engine.Update(func(txn *badger.Txn) error {
    return txn.Delete(KeyWithCF(cf, key))
  })
}

在了解了engine_util提供的方法，以及Modify的含义后，Write怎么实现其实就迎刃而解了。

func (s *StandAloneStorage) Write(ctx *kvrpcpb.Context, batch []storage.Modify) error {
  // Your Code Here (1).
  //func PutCF(engine *badger.DB, cf string, key []byte, val []byte)
  var err error
  for _, m := range batch {
    key, val, cf := m.Key(), m.Value(), m.Cf()
    if _, ok := m.Data.(storage.Put); ok {
      err = engine_util.PutCF(s.engine.Kv, cf, key, val)
    } else {
      err = engine_util.DeleteCF(s.engine.Kv, cf, key)
    }
    if err != nil {
      return err
    }
  }
  return nil
}

Reader

Reader函数需要我们返回一个StorageReader接口，那么我们就需要去实现一个StorageReader接口的结构体，来看看实现它需要哪些函数

Reader(ctx *kvrpcpb.Context) (StorageReader, error)
type StorageReader interface {
  // When the key doesn't exist, return nil for the value
  GetCF(cf string, key []byte) ([]byte, error)
  IterCF(cf string) engine_util.DBIterator
  Close()
}

看一下engine_util包给我们提供了什么函数，可以看到从txn中读取或者遍历的方法不用自己写，直接使用api即可，那么上面的GetCF和IterCF就是对下面两个函数的封装。

// get value
val, err := engine_util.GetCFFromTxn(txn, cf, key)
func GetCFFromTxn(txn *badger.Txn, cf string, key []byte) (val []byte, err error)
// get iterator
iter := engine_util.NewCFIterator(cf, txn)
func NewCFIterator(cf string, txn *badger.Txn) *BadgerIterator

可以看到engine_util提供的GetCFFromTxn和NewCFIterator两个函数都需要badger.Txn，那么这个Txn是什么呢？其实是一个事务。获取badger.Txn的函数engine_util并未给出，需要直接调用badger.DB.NewTransaction函数。

txn *badger.Txn
//update为真表示Put/Delete两个写操作，为假表示Get/Scan两个读操作。
//NewTransaction creates a new transaction.
func (db *DB) NewTransaction(update bool) *Txn

所以不难发现，想要实现StorageReader接口，那么结构体中就要包含badger.Txn，继而去调用engine_util提供的api。在这里，可以把这个字段放在StandAloneStorage中；不过我还是把它拆到新的结构体里面了。

为什么呢？在上面的Write函数中，我们仅仅传入的是*badger.DB，其内部也是有*badger.Txn的，但是Write对事务进行了屏蔽。在StorageReader接口的两个函数中，也没有事务的身影，但是我们因为是读，所以需要用到新的读事务，只能去创建一个。

综合来看，这些接口的封装，很明显就是不想让我们去过多的使用事务的，所以干脆把其放在一个新结构体里面return掉好了。

func (s *StandAloneStorage) Reader(ctx *kvrpcpb.Context) (storage.StorageReader, error) {
  // Your Code Here (1).
  //func (db *DB) NewTransaction(update bool) *Txn
  //For read-only transactions, set update to false.
  txn := s.engine.Kv.NewTransaction(false)
  return NewStandAloneStorageReader(txn), nil
}
//实现 type StorageReader interface 接口
type StandAloneStorageReader struct {
  KvTxn *badger.Txn
}
func NewStandAloneStorageReader(txn *badger.Txn) *StandAloneStorageReader {
  return &StandAloneStorageReader{
    KvTxn: txn,
  }
}
func (s *StandAloneStorageReader) GetCF(cf string, key []byte) ([]byte, error) {
  val, err := engine_util.GetCFFromTxn(s.KvTxn, cf, key)
  //StandAloneStorage是底层的一个系统,这里认为not found不是err
  //将not found屏蔽掉
  //上层不认为not found是err
  if err == badger.ErrKeyNotFound {
    return nil, nil
  }
  return val, err
}
func (s *StandAloneStorageReader) IterCF(cf string) engine_util.DBIterator {
  //func NewCFIterator(cf string, KvTxn *badger.Txn) *BadgerIterator
  return engine_util.NewCFIterator(cf, s.KvTxn)
}
func (s *StandAloneStorageReader) Close() {
  s.KvTxn.Discard()
}

Server

前面以及说过了，原始的 key/value 服务处理程序(Put/Delete/Get/Scan)：其实就是对存储引擎的一层上层封装，有了这层封装，我们可以随意改变底层的存储引擎。使用则无需关注底层细节，只要遵循上层接口使用规范即可。

// Server is a TinyKV server, it 'faces outwards', sending and receiving messages from clients such as TinySQL.
type Server struct {
  storage storage.Storage
  //...
}

这样做我觉得最大的好处就是，底层的存储引擎无论怎么换，对上层都是无感的。这让我想到了mysql的innodb和myisam，都是无感的。在project1中，我们实现的是单机的存储引擎，所以在后续的project中，我们就可以直接换掉它。

func (server *Server) RawGet(_ context.Context, req *kvrpcpb.RawGetRequest) (*kvrpcpb.RawGetResponse, error)
func (server *Server) RawPut(_ context.Context, req *kvrpcpb.RawPutRequest) (*kvrpcpb.RawPutResponse, error)
func (server *Server) RawDelete(_ context.Context, req *kvrpcpb.RawDeleteRequest) (*kvrpcpb.RawDeleteResponse, error)
func (server *Server) RawScan(_ context.Context, req *kvrpcpb.RawScanRequest) (*kvrpcpb.RawScanResponse, error)

这4个接口都是对上面的StandAloneStorage中写的接口的再次封装。

RawGet：如果获取不到，返回时要标记 reply.NotFound=true。
RawPut：把单一的 put 请求用 storage.Modify 包装成一个数组，一次性写入。
RawDelete：和 RawPut 同理。
RawScan：通过 reader 获取 iter，从 StartKey 开始，同时注意 limit。

需要特别注意的一点是：RawGet：如果获取不到，返回时要标记 reply.NotFound=true

我所理解的Scan，是线性扫描。那么转入请求中的Limit，就是限制多少个。首先根据Seek定位到第一个key，然后向后读Limit个，同时要注意下一个的合法性，具体见代码。

// The functions below are Server's Raw API. (implements TinyKvServer).
// Some helper methods can be found in sever.go in the current directory
// 上层允许err时返回nil结构体
// RawGet return the corresponding Get response based on RawGetRequest's CF and Key fields
func (server *Server) RawGet(_ context.Context, req *kvrpcpb.RawGetRequest) (*kvrpcpb.RawGetResponse, error) {
  // Your Code Here (1).
  reader, err := server.storage.Reader(req.Context)
  if err != nil {
    return nil, err
  }
  val, err := reader.GetCF(req.Cf, req.Key)
  if err != nil {
    return nil, err
  }
  resp := &kvrpcpb.RawGetResponse{
    Value:    val,
    NotFound: false,
  }
  if val == nil {
    resp.NotFound = true
  }
  return resp, nil
}
// RawPut puts the target data into storage and returns the corresponding response
func (server *Server) RawPut(_ context.Context, req *kvrpcpb.RawPutRequest) (*kvrpcpb.RawPutResponse, error) {
  // Your Code Here (1).
  // Hint: Consider using Storage.Modify to store data to be modified
  put := storage.Put{
    Key:   req.Key,
    Value: req.Value,
    Cf:    req.Cf,
  }
  batch := storage.Modify{Data: put}
  err := server.storage.Write(req.Context, []storage.Modify{batch})
  if err != nil {
    return nil, err
  }
  return &kvrpcpb.RawPutResponse{}, nil
}
// RawDelete delete the target data from storage and returns the corresponding response
func (server *Server) RawDelete(_ context.Context, req *kvrpcpb.RawDeleteRequest) (*kvrpcpb.RawDeleteResponse, error) {
  // Your Code Here (1).
  // Hint: Consider using Storage.Modify to store data to be deleted
  del := storage.Delete{
    Key: req.Key,
    Cf:  req.Cf,
  }
  batch := storage.Modify{Data: del}
  err := server.storage.Write(req.Context, []storage.Modify{batch})
  if err != nil {
    return nil, err
  }
  return &kvrpcpb.RawDeleteResponse{}, nil
}
// RawScan scan the data starting from the start key up to limit. and return the corresponding result
func (server *Server) RawScan(_ context.Context, req *kvrpcpb.RawScanRequest) (*kvrpcpb.RawScanResponse, error) {
  // Your Code Here (1).
  // Hint: Consider using reader.IterCF
  reader, err := server.storage.Reader(req.Context)
  if err != nil {
    return nil, err
  }
  var Kvs []*kvrpcpb.KvPair
  limit := req.Limit
  iter := reader.IterCF(req.Cf)
  defer iter.Close()
  for iter.Seek(req.StartKey); iter.Valid(); iter.Next() {
    item := iter.Item()
    key := item.Key()
    val, _ := item.Value()
    Kvs = append(Kvs, &kvrpcpb.KvPair{
      Key:   key,
      Value: val,
    })
    limit--
    if limit == 0 {
      break
    }
  }
  resp := &kvrpcpb.RawScanResponse{
    Kvs: Kvs,
  }
  return resp, nil
}

单元测试

pass咯~~~

root@wxf:/tinykv# make project1
GO111MODULE=on go test -v --count=1 --parallel=1 -p=1 ./kv/server -run 1
=== RUN   TestRawGet1
--- PASS: TestRawGet1 (0.85s)
=== RUN   TestRawGetNotFound1
--- PASS: TestRawGetNotFound1 (1.05s)
=== RUN   TestRawPut1
--- PASS: TestRawPut1 (2.91s)
=== RUN   TestRawGetAfterRawPut1
--- PASS: TestRawGetAfterRawPut1 (3.94s)
=== RUN   TestRawGetAfterRawDelete1
--- PASS: TestRawGetAfterRawDelete1 (3.69s)
=== RUN   TestRawDelete1
--- PASS: TestRawDelete1 (1.54s)
=== RUN   TestRawScan1
--- PASS: TestRawScan1 (1.04s)
=== RUN   TestRawScanAfterRawPut1
--- PASS: TestRawScanAfterRawPut1 (1.03s)
=== RUN   TestRawScanAfterRawDelete1
--- PASS: TestRawScanAfterRawDelete1 (0.88s)
=== RUN   TestIterWithRawDelete1
--- PASS: TestIterWithRawDelete1 (1.08s)
PASS
ok    github.com/pingcap-incubator/tinykv/kv/server 18.027s
root@wxf:/tinykv#

TinyKv Project1 Standalone KV

前言

Project1 StandaloneKV 文档翻译

文档的重点内容

StandAloneStorage

Write

Reader

Server

单元测试

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

TinyKv Project1 Standalone KV

前言

Project1 StandaloneKV 文档翻译

文档的重点内容

StandAloneStorage

Write

Reader

Server

单元测试

热门文章

最新文章

相关电子书