go-ethereum 源码笔记(rlp 模块-序列化与反序列化)

RLP 的定义,作用,用法在 wiki 文档中有文字说明,这部分我们只做简单描述。

黄皮书里有 RLP 的形式化说明,这里我会总结一下,然后侧重于分析源码,讲解实现原理。

定义

RLP(Recursive Length Prefix)递归长度前缀是一种编码算法,用于编码任意嵌套结构的二进制数据,它是以太坊中数据序列化和反序列化的主要方法,区块、交易等数据结构在持久化时会先经过 RLP 编码再存储到数据库,p2p 网络中节点之间的数据传输也需要 RLP 的编码。

RLP 有两个特点:一个是递归,被编码的数据是递归的结构,编码算法也是递归进行处理的;二是长度前缀,RLP 编码都带有一个前缀,这个前缀与被编码数据长度相关。

根据定义,RLP 编码只处理两类数据:一类是字符串(例如字节数组),一类是列表。字符串指的是一串二进制数据,列表是一个嵌套递归的结构,里面可以包含字符串和列表,例如["cat",["puppy","cow"],"horse",[[]],"pig",[""],"sheep"]就是一个复杂的列表。其他类型的数据需要转成以上的两类。

规则和例子

wiki 中有详细定义,这里不赘述。

黄皮书里的形式化方法

黄皮书中有 RLP 的形式化定义,形式化的描述比大白话更直观,这里我们做一下简单总结:

首先定义集合 T:

rlp-structure-t

$\mathbb{T}$ 表示所有的字节数组和树形结构的组合,$\mathbb{L}$ 表示不止单一节点的树形结构,如结构体,树节点的分支节点,非叶子节点。 $\mathbb{O}$ 表示所有 byte 的集合,$\mathbb{B}$ 表示所有可能的字节数组,

$\mathbb{T}$ 和 $\mathbb{L}$ 是 RLP 需要处理的两种结构。

RLP 可以用两个子函数描述:

rlp-structure-rlp

其中:

rlp-structure-rbx

rlp-structure-rlx

对于 $\mathbb{B}$ 类型的字节数组,处理规则是:

  • 如果字节数组只有一个字节,并且字节大小小于 128,编码结果就是它本身。
  • 如果字节数组(字符串)长度小于56,编码结果是128加上字节数组长度的和作为前缀,再加上原始数据。由于被编码的字符串长度是55=0x37,因此单字节前缀的最大值是 0x80+0x37=0xb7,即编码的第一个字节的取值范围是 [0x80, 0xb7]
  • 如果字节数组长度大于等于56,编码结果是: 以183加上原始数据的长度的大端表示的长度作为前缀,加上原始数据长度的大端表示,再加上原始数据。

其中 $\mathtt{\tiny BE}$ 实际上表示去掉前导0的大端模式。

对于 $\mathtt{T} $类型(树形结构),处理规则是:

  • 首先对树形结构里面的每一个元素使用 RLP 处理,然后将结果连接起来,记为 s。
  • 如果连接后的字节长度小于56,编码结果是:以192加上连接后的长度作为前缀,加上连接后的结果,即 0xc0 加上列表的总长度,编码的第一个字节的取值范围是 [0xc0, 0xf7]
  • 如果连接后的字节长度大于56字节,编码结果是:以247加上连接后的长度的大端模式的长度的结果作为前缀,加上连接后的长度的大端模式,再加上列表中各元素项 RLP 编码的结果。即 0xf7 加上列表总长度的二进制形式的字节长度。编码的第一个字节的取值范围是 [0xf8, 0xff]

源码解析

RLP 包的源码在 rlp 目录下:

1
2
3
4
5
6
7
8
9
10
├── decode.go # 解码器,反序列化的过程,将 RLP 数据解码为 Golang 数据结构
├── decode_tail_test.go
├── decode_test.go
├── doc.go # 包的文档
├── encode.go # 编码器,序列化的过程,将 Golang 数据编码为 RLP 结构数据
├── encode_test.go
├── encoder_example_test.go
├── raw.go # 没有解码的原生数据
├── raw_test.go
└── typecache.go # 类型缓存,记录了类型 ->(编码器|解码器)的内容

测试代码也是很好的参考文档,有时候甚至比注释还好懂,可以好好看看。这里我们开始讲解序列化和反序列化的代码实现。

typeinfo

上面提到,对于不同的类型,有不同的编码方式,在 C++ 这样的语言中,可以通过相同函数名,不同类型,也就是重载来实现不同类型的编码器的分派,也可以通过泛型实现分派。

Go 不支持这些语言级别的特性,为了实现函数分派,引入了 typeinfo 结构,可以根据类型快速找到编码器和解码器函数,这部分内容在 rlp/typecache.go 中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
var (
typeCacheMutex sync.RWMutex
typeCache = make(map[typekey]*typeinfo)
)
type typeinfo struct {
decoder
writer
}
type tags struct {
tail bool
ignored bool
}
type typekey struct {
reflect.Type
tags
}

其中 typeCacheMutex 是读写锁,在多线程的使用过程中保护 typeCachetypeCache 的作用是记录类型和编码器函数的对应关系。

获取编码器,解码器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func cachedTypeInfo(typ reflect.Type, tags tags) (*typeinfo, error) {
typeCacheMutex.RLock()
info := typeCache[typekey{typ, tags}]
typeCacheMutex.RUnlock()
if info != nil {
return info, nil
}
typeCacheMutex.Lock()
defer typeCacheMutex.Unlock()
return cachedTypeInfo1(typ, tags)
}
func cachedTypeInfo1(typ reflect.Type, tags tags) (*typeinfo, error) {
key := typekey{typ, tags}
info := typeCache[key]
if info != nil {
return info, nil
}
typeCache[key] = new(typeinfo)
info, err := genTypeInfo(typ, tags)
if err != nil {
delete(typeCache, key)
return nil, err
}
*typeCache[key] = *info
return typeCache[key], err
}

没能明白 cachedTypeInfocachedTypeInfo1 的真正区别是什么,有熟悉这块代码的朋友可以赐教一下。

获取编码器,解码器时,首先会从缓存中请求,如果没有命中的话再调用 genTypeInfo

1
2
3
4
5
6
7
8
9
10
func genTypeInfo(typ reflect.Type, tags tags) (info *typeinfo, err error) {
info = new(typeinfo)
if info.decoder, err = makeDecoder(typ, tags); err != nil {
return nil, err
}
if info.writer, err = makeWriter(typ, tags); err != nil {
return nil, err
}
return info, nil
}

编码器和解码器大同小异,只是方向不同,这里我们以编码器为例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
func makeWriter(typ reflect.Type, ts tags) (writer, error) {
kind := typ.Kind()
switch {
case typ == rawValueType:
return writeRawValue, nil
case typ.Implements(encoderInterface):
return writeEncoder, nil
case kind != reflect.Ptr && reflect.PtrTo(typ).Implements(encoderInterface):
return writeEncoderNoPtr, nil
case kind == reflect.Interface:
return writeInterface, nil
case typ.AssignableTo(reflect.PtrTo(bigInt)):
return writeBigIntPtr, nil
case typ.AssignableTo(bigInt):
return writeBigIntNoPtr, nil
case isUint(kind):
return writeUint, nil
case kind == reflect.Bool:
return writeBool, nil
case kind == reflect.String:
return writeString, nil
case kind == reflect.Slice && isByte(typ.Elem()):
return writeBytes, nil
case kind == reflect.Array && isByte(typ.Elem()):
return writeByteArray, nil
case kind == reflect.Slice || kind == reflect.Array:
return makeSliceWriter(typ, ts)
case kind == reflect.Struct:
return makeStructWriter(typ)
case kind == reflect.Ptr:
return makePtrWriter(typ)
default:
return nil, fmt.Errorf("rlp: type %v is not RLP-serializable", typ)
}
}

可以看到 makeWriter 实际上就是一堆 switch 分支,根据不同的类型,调用不同的编码方法。接下来我们会挑取几个典型方法来看。

序列化

首先看字符串的编码方法。

1
2
3
4
5
6
7
8
9
10
func writeString(val reflect.Value, w *encbuf) error {
s := val.String()
if len(s) == 1 && s[0] <= 0x7f {
w.str = append(w.str, s[0])
} else {
w.encodeStringHeader(len(s))
w.str = append(w.str, s...)
}
return nil
}

对于字符串,如果只有一个字节,并且字节大小小于128,编码结果是其本身,这对应规则一的第1点。对于第2,第3种情况,调用 encodeStringHeader

1
2
3
4
5
6
7
8
9
func (w *encbuf) encodeStringHeader(size int) {
if size < 56 {
w.str = append(w.str, 0x80+byte(size))
} else {
sizesize := putint(w.sizebuf[1:], uint64(size))
w.sizebuf[0] = 0xB7 + byte(sizesize)
w.str = append(w.str, w.sizebuf[:sizesize+1]...)
}
}

如果长度小于56,编码结果是128加上字节数组长度的和作为前缀再加上原始数据,这对应的是第2种情况,如果长度大于等于56,则编码结果是0x87加上原始数据长度的大端表示的长度作为前缀,再加上原始数据长度的大端表示,再加上原始数据,这对应的是第3种情况。

再看树形结构的编码方法。

实际上这里所指的树形结构就是结构体数据。对于普通的类型,例如字符串,整形,布尔型,我们可以直接调用对应的编码方法,往 encbuf 里填充数据,对于结构体类型,我们没法确定其具体结构,但每种类型都实现了 writer 这一类型,即 type writer func(reflect.Value, *encbuf) error,注意到 writer 类型与 func writeString(val reflect.Value, w *encbuf) error 这些编码方法的类型是一致的。因此我们可以用递归的方式解决结构体类型的编码。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
func makeStructWriter(typ reflect.Type) (writer, error) {
fields, err := structFields(typ)
if err != nil {
return nil, err
}
writer := func(val reflect.Value, w *encbuf) error {
lh := w.list()
for _, f := range fields {
if err := f.info.writer(val.Field(f.index), w); err != nil {
return err
}
}
w.listEnd(lh)
return nil
}
return writer, nil
}

首先拿到该结构体的所有字段,然后定义一个 writer 的匿名方法,即 writer func(reflect.Value, *encbuf) error 类型,这个方法可能会被递归调用,在这个方法中,遍历之前通过 structFields 拿到的所有字段,调用对应的 writer 方法,这些结果都会加入到 w(encbuf 类型)中,根据 RLP 编码的定义,需要根据树形结构的长度确定前缀的大小,这是通过 list, listEnd, size, toBytes 等方法实现的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
func (w *encbuf) list() *listhead {
lh := &listhead{offset: len(w.str), size: w.lhsize}
w.lheads = append(w.lheads, lh)
return lh
}
func (w *encbuf) listEnd(lh *listhead) {
lh.size = w.size() - lh.offset - lh.size
if lh.size < 56 {
w.lhsize += 1
} else {
w.lhsize += 1 + intsize(uint64(lh.size))
}
}
func (w *encbuf) size() int {
return len(w.str) + w.lhsize
}
func (w *encbuf) toBytes() []byte {
out := make([]byte, w.size())
strpos := 0
pos := 0
for _, head := range w.lheads {
n := copy(out[pos:], w.str[strpos:head.offset])
pos += n
strpos += n
enc := head.encode(out[pos:])
pos += len(enc)
}
copy(out[pos:], w.str[strpos:])
return out
}

其中 toBytes()encbuf 最后的处理逻辑,返回最终的 RLP 数据。

编码即序列化的整个流程就是这样,解码的过程就不赘述了,相反流程而已,最后看一下 encode 模块暴露的外部方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
func Encode(w io.Writer, val interface{}) error {
if outer, ok := w.(*encbuf); ok {
return outer.encode(val)
}
eb := encbufPool.Get().(*encbuf)
defer encbufPool.Put(eb)
eb.reset()
if err := eb.encode(val); err != nil {
return err
}
return eb.toWriter(w)
}
func EncodeToBytes(val interface{}) ([]byte, error) {
eb := encbufPool.Get().(*encbuf)
defer encbufPool.Put(eb)
eb.reset()
if err := eb.encode(val); err != nil {
return nil, err
}
return eb.toBytes(), nil
}
func EncodeToReader(val interface{}) (size int, r io.Reader, err error) {
eb := encbufPool.Get().(*encbuf)
eb.reset()
if err := eb.encode(val); err != nil {
return 0, nil, err
}
return eb.size(), &encReader{buf: eb}, nil
}

EncodeEncodeToBytesEncodeToReader 等方法将值编码到 io.Writer, Bytes, reader 等等,它们有一个共同点,都调用了 encode 这个内部方法。

1
2
3
4
5
6
7
8
func (w *encbuf) encode(val interface{}) error {
rval := reflect.ValueOf(val)
ti, err := cachedTypeInfo(rval.Type(), tags{})
if err != nil {
return err
}
return ti.writer(rval, w)
}

encode 方法会从 cachedTypeInfo 中获得某类型对应的编码器,然后调用 writer 方法,结果就能写到 encbuf 类型中。EncodeToBytesEncodeToReader 等只是最后调用的方法不一致,如果是调用 toBytes,则返回经过 RLP 编码的字节码数据。

参考资料