Go是如何实现protobuf的编解码的(2): 源码

// main.go
package main

import (
    "fmt"

    "./types"
    "github.com/golang/protobuf/proto"
)

func main() {
    req := &types.Request{Data: "Hello Dabin"}

    // Marshal
    encoded, err := proto.Marshal(req)
    if err != nil {
        fmt.Printf("Encode to protobuf data error: %v", err)
    }
    ...
}

// Marshal takes a protocol buffer message
// and encodes it into the wire format, returning the data.
// This is the main entry point.
func Marshal(pb Message) ([]byte, error) {
    if m, ok := pb.(newMarshaler); ok {
        siz := m.XXX_Size()
        b := make([]byte, 0, siz)
        return m.XXX_Marshal(b, false)
    }
    if m, ok := pb.(Marshaler); ok {
        // If the message can marshal itself, let it do it, for compatibility.
        // NOTE: This is not efficient.
        return m.Marshal()
    }
    // in case somehow we didn't generate the wrapper
    if pb == nil {
        return nil, ErrNil
    }
    var info InternalMessageInfo
    siz := info.Size(pb)
    b := make([]byte, 0, siz)
    return info.Marshal(b, pb, false)
}

// newMarshaler is the interface representing objects that can marshal themselves.
//
// This exists to support protoc-gen-go generated messages.
// The proto package will stop type-asserting to this interface in the future.
//
// DO NOT DEPEND ON THIS.
type newMarshaler interface {
    XXX_Size() int
    XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
}

// Marshaler is the interface representing objects that can marshal themselves.
type Marshaler interface {
    Marshal() ([]byte, error)
}

// request.pb.go
func (m *Request) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
    print("Called xxx marshal\n")
    panic("I want see stack trace")
    return xxx_messageInfo_Request.Marshal(b, m, deterministic)
}

var xxx_messageInfo_Request proto.InternalMessageInfo

// InternalMessageInfo is a type used internally by generated .pb.go files.
// This type is not intended to be used by non-generated code.
// This type is not subject to any compatibility guarantee.
type InternalMessageInfo struct {
    marshal   *marshalInfo   // marshal信息
    unmarshal *unmarshalInfo // unmarshal信息
    merge     *mergeInfo
    discard   *discardInfo
}

// Marshal is the entry point from generated code,
// and should be ONLY called by generated code.
// It marshals msg to the end of b.
// a is a pointer to a place to store cached marshal info.
func (a *InternalMessageInfo) Marshal(b []byte, msg Message, deterministic bool) ([]byte, error) {
    // 获取该message类型的MarshalInfo，这些信息都缓存起来了
    // 大量并发时无需重复创建
    u := getMessageMarshalInfo(msg, a)
    // 入参校验
    ptr := toPointer(&msg)
    if ptr.isNil() {
        // We get here if msg is a typed nil ((*SomeMessage)(nil)),
        // so it satisfies the interface, and msg == nil wouldn't
        // catch it. We don't want crash in this case.
        return b, ErrNil
    }
    // 根据MarshalInfo对数据进行marshal
    return u.marshal(b, ptr, deterministic)
}

func getMessageMarshalInfo(msg interface{}, a *InternalMessageInfo) *marshalInfo {
    // u := a.marshal, but atomically.
    // We use an atomic here to ensure memory consistency.
    // 从InternalMessageInfo中读取
    u := atomicLoadMarshalInfo(&a.marshal)
    // 读取不到代表未保存过
    if u == nil {
        // Get marshal information from type of message.
        t := reflect.ValueOf(msg).Type()
        if t.Kind() != reflect.Ptr {
            panic(fmt.Sprintf("cannot handle non-pointer message type %v", t))
        }
        u = getMarshalInfo(t.Elem())
        // Store it in the cache for later users.
        // a.marshal = u, but atomically.
        atomicStoreMarshalInfo(&a.marshal, u)
    }
    return u
}

// getMarshalInfo returns the information to marshal a given type of message.
// The info it returns may not necessarily initialized.
// t is the type of the message (NOT the pointer to it).
// 获取MarshalInfo结构体，如果不存在则使用message类型t创建1个
func getMarshalInfo(t reflect.Type) *marshalInfo {
    marshalInfoLock.Lock()
    u, ok := marshalInfoMap[t]
    if !ok {
        u = &marshalInfo{typ: t}
        marshalInfoMap[t] = u
    }
    marshalInfoLock.Unlock()
    return u
}

// marshalInfo is the information used for marshaling a message.
type marshalInfo struct {
    typ          reflect.Type
    fields       []*marshalFieldInfo
    unrecognized field                      // offset of XXX_unrecognized
    extensions   field                      // offset of XXX_InternalExtensions
    v1extensions field                      // offset of XXX_extensions
    sizecache    field                      // offset of XXX_sizecache
    initialized  int32                      // 0 -- only typ is set, 1 -- fully initialized
    messageset   bool                       // uses message set wire format
    hasmarshaler bool                       // has custom marshaler
    sync.RWMutex                            // protect extElems map, also for initialization
    extElems     map[int32]*marshalElemInfo // info of extension elements
}

// marshal is the main function to marshal a message. It takes a byte slice and appends
// the encoded data to the end of the slice, returns the slice and error (if any).
// ptr is the pointer to the message.
// If deterministic is true, map is marshaled in deterministic order.
// 该函数是Marshal的主体函数，把消息编码为数据后，追加到b之后，最后返回b。
// deterministic为true代表map会以确定的顺序进行编码。
func (u *marshalInfo) marshal(b []byte, ptr pointer, deterministic bool) ([]byte, error) {
    // 初始化marshalInfo的基础信息
    // 主要是根据已有信息填充该结构体的一些字段
    if atomic.LoadInt32(&u.initialized) == 0 {
        u.computeMarshalInfo()
    }

    // If the message can marshal itself, let it do it, for compatibility.
    // NOTE: This is not efficient.
    // 如果该类型实现了Marshaler接口，即能够对自己Marshal，则自行Marshal
    // 结果追加到b
    if u.hasmarshaler {
        m := ptr.asPointerTo(u.typ).Interface().(Marshaler)
        b1, err := m.Marshal()
        b = append(b, b1...)
        return b, err
    }

    var err, errLater error
    // The old marshaler encodes extensions at beginning.
    // 检查扩展字段，把message的扩展字段追加到b
    if u.extensions.IsValid() {
        // offset函数用来根据指针偏移量获取message的指定字段
        e := ptr.offset(u.extensions).toExtensions()
        if u.messageset {
            b, err = u.appendMessageSet(b, e, deterministic)
        } else {
            b, err = u.appendExtensions(b, e, deterministic)
        }
        if err != nil {
            return b, err
        }
    }
    if u.v1extensions.IsValid() {
        m := *ptr.offset(u.v1extensions).toOldExtensions()
        b, err = u.appendV1Extensions(b, m, deterministic)
        if err != nil {
            return b, err
        }
    }

    // 遍历message的每一个字段，检查并做编码，然后追加到b
    for _, f := range u.fields {
        if f.required {
            // 如果required的字段未设置，则记录错误，所有的marshal工作完成后再处理
            if ptr.offset(f.field).getPointer().isNil() {
                // Required field is not set.
                // We record the error but keep going, to give a complete marshaling.
                if errLater == nil {
                    errLater = &RequiredNotSetError{f.name}
                }
                continue
            }
        }
        // 字段为指针类型，并且为nil，代表未设置，该字段无需编码
        if f.isPointer && ptr.offset(f.field).getPointer().isNil() {
            // nil pointer always marshals to nothing
            continue
        }
        // 利用这个字段的marshaler进行编码
        b, err = f.marshaler(b, ptr.offset(f.field), f.wiretag, deterministic)
        if err != nil {
            if err1, ok := err.(*RequiredNotSetError); ok {
                // required字段但未设置错误
                // Required field in submessage is not set.
                // We record the error but keep going, to give a complete marshaling.
                if errLater == nil {
                    errLater = &RequiredNotSetError{f.name + "." + err1.field}
                }
                continue
            }
            // “动态数组”中包含nil元素
            if err == errRepeatedHasNil {
                err = errors.New("proto: repeated field " + f.name + " has nil element")
            }
            if err == errInvalidUTF8 {
                if errLater == nil {
                    fullName := revProtoTypes[reflect.PtrTo(u.typ)] + "." + f.name
                    errLater = &invalidUTF8Error{fullName}
                }
                continue
            }
            return b, err
        }
    }
    // 为识别的类型字段，直接转为bytes，追加到b
    // computeMarshalInfo中已经收集这些字段
    if u.unrecognized.IsValid() {
        s := *ptr.offset(u.unrecognized).toBytes()
        b = append(b, s...)
    }
    return b, errLater
}

// computeMarshalInfo initializes the marshal info.
func (u *marshalInfo) computeMarshalInfo() {
    // 加锁，代表了不能同时计算marshal信息
    u.Lock()
    defer u.Unlock()
    // 计算1次即可
    if u.initialized != 0 { // non-atomic read is ok as it is protected by the lock
        return
    }

    // 获取要marshal的message类型
    t := u.typ
    u.unrecognized = invalidField
    u.extensions = invalidField
    u.v1extensions = invalidField
    u.sizecache = invalidField

    // If the message can marshal itself, let it do it, for compatibility.
    // 判断当前类型是否实现了Marshal接口，如果实现标记为类型自有marshaler
    // 没用类型断言是因为t是Type类型，不是保存在某个接口的变量
    // NOTE: This is not efficient.
    if reflect.PtrTo(t).Implements(marshalerType) {
        u.hasmarshaler = true
        atomic.StoreInt32(&u.initialized, 1)
        // 可以直接返回了，后面使用自有的marshaler编码
        return
    }

    // get oneof implementers
    // 看*t实现了以下哪个接口，oneof特性
    var oneofImplementers []interface{}
    switch m := reflect.Zero(reflect.PtrTo(t)).Interface().(type) {
    case oneofFuncsIface:
        _, _, _, oneofImplementers = m.XXX_OneofFuncs()
    case oneofWrappersIface:
        oneofImplementers = m.XXX_OneofWrappers()
    }

    n := t.NumField()

    // deal with XXX fields first
    // 遍历t的每一个XXX字段
    for i := 0; i < t.NumField(); i++ {
        f := t.Field(i)
        // 跳过非XXX开头的字段
        if !strings.HasPrefix(f.Name, "XXX_") {
            continue
        }
        // 处理以下几个protobuf自带的字段
        switch f.Name {
        case "XXX_sizecache":
            u.sizecache = toField(&f)
        case "XXX_unrecognized":
            u.unrecognized = toField(&f)
        case "XXX_InternalExtensions":
            u.extensions = toField(&f)
            u.messageset = f.Tag.Get("protobuf_messageset") == "1"
        case "XXX_extensions":
            u.v1extensions = toField(&f)
        case "XXX_NoUnkeyedLiteral":
            // nothing to do
        default:
            panic("unknown XXX field: " + f.Name)
        }
        n--
    }

    // normal fields
    // 处理message的普通字段
    fields := make([]marshalFieldInfo, n) // batch allocation
    u.fields = make([]*marshalFieldInfo, 0, n)
    for i, j := 0, 0; i < t.NumField(); i++ {
        f := t.Field(i)

        // 跳过XXX字段
        if strings.HasPrefix(f.Name, "XXX_") {
            continue
        }

        // 取fields的下一个有效字段，指针类型
        // j代表了fields有效字段数量，n是包含了XXX字段的总字段数量
        field := &fields[j]
        j++
        field.name = f.Name
        // 填充到u.fields
        u.fields = append(u.fields, field)
        // 字段的tag里包含“protobuf_oneof”特殊处理
        if f.Tag.Get("protobuf_oneof") != "" {
            field.computeOneofFieldInfo(&f, oneofImplementers)
            continue
        }
        // 字段里不包含“protobuf”，代表不是protoc自动生成的字段
        if f.Tag.Get("protobuf") == "" {
            // field has no tag (not in generated message), ignore it
            // 删除刚刚保存的字段信息
            u.fields = u.fields[:len(u.fields)-1]
            j--
            continue
        }
        // 填充字段的marshal信息
        field.computeMarshalFieldInfo(&f)
    }

    // fields are marshaled in tag order on the wire.
    // 字段排序
    sort.Sort(byTag(u.fields))

    // 初始化完成
    atomic.StoreInt32(&u.initialized, 1)
}

// request.pb.go
type Request struct{
    Data                 string   `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"`
    ...
}

// computeMarshalFieldInfo fills up the information to marshal a field.
func (fi *marshalFieldInfo) computeMarshalFieldInfo(f *reflect.StructField) {
    // parse protobuf tag of the field.
    // tag has format of "bytes,49,opt,name=foo,def=hello!"
    // 获取"protobuf"的完整tag，然后使用，分割，得到上面的格式
    tags := strings.Split(f.Tag.Get("protobuf"), ",")
    if tags[0] == "" {
        return
    }
    // tag的编号，即message中设置的string name = x，则x就是这个字段的tag id
    tag, err := strconv.Atoi(tags[1])
    if err != nil {
        panic("tag is not an integer")
    }
    // 要转换成的类型，bytes，varint等等
    wt := wiretype(tags[0])
    // 设置字段是required还是opt
    if tags[2] == "req" {
        fi.required = true
    }
    // 设置field和tag信息到marshalFieldInfo
    fi.setTag(f, tag, wt)
    // 根据当前的tag信息（类型等），选择marshaler函数
    fi.setMarshaler(f, tags)
}

// setMarshaler fills up the sizer and marshaler in the info of a field.
func (fi *marshalFieldInfo) setMarshaler(f *reflect.StructField, tags []string) {
    // map类型字段特殊处理
    switch f.Type.Kind() {
    case reflect.Map:
        // map field
        fi.isPointer = true
        fi.sizer, fi.marshaler = makeMapMarshaler(f)
        return
    case reflect.Ptr, reflect.Slice:
        // 指针字段和切片字段标记指针类型
        fi.isPointer = true
    }

    // 根据字段类型和tag选择marshaler
    fi.sizer, fi.marshaler = typeMarshaler(f.Type, tags, true, false)
}

// typeMarshaler returns the sizer and marshaler of a given field.
// t is the type of the field.
// tags is the generated "protobuf" tag of the field.
// If nozero is true, zero value is not marshaled to the wire.
// If oneof is true, it is a oneof field.
// 函数非常长，省略内容
func typeMarshaler(t reflect.Type, tags []string, nozero, oneof bool) (sizer, marshaler) {
    ...
    switch t.Kind() {
    case reflect.Bool:
        if pointer {
            return sizeBoolPtr, appendBoolPtr
        }
        if slice {
            if packed {
                return sizeBoolPackedSlice, appendBoolPackedSlice
            }
            return sizeBoolSlice, appendBoolSlice
        }
        if nozero {
            return sizeBoolValueNoZero, appendBoolValueNoZero
        }
        return sizeBoolValue, appendBoolValue
    case reflect.Uint32:
    ...
    case reflect.Int32:
    ....
    case reflect.Struct:
    ...
}

func appendBoolValue(b []byte, ptr pointer, wiretag uint64, _ bool) ([]byte, error) {
    v := *ptr.toBool()
    b = appendVarint(b, wiretag)
    if v {
        b = append(b, 1)
    } else {
        b = append(b, 0)
    }
    return b, nil
}

func appendStringValue(b []byte, ptr pointer, wiretag uint64, _ bool) ([]byte, error) {
    v := *ptr.toString()
    b = appendVarint(b, wiretag)
    b = appendVarint(b, uint64(len(v)))
    b = append(b, v...)
    return b, nil
}

| wiretag | data | wiretag | data | ... | data |

// Unmarshaler is the interface representing objects that can
// unmarshal themselves.  The argument points to data that may be
// overwritten, so implementations should not keep references to the
// buffer.
// Unmarshal implementations should not clear the receiver.
// Any unmarshaled data should be merged into the receiver.
// Callers of Unmarshal that do not want to retain existing data
// should Reset the receiver before calling Unmarshal.
type Unmarshaler interface {
    Unmarshal([]byte) error
}

// newUnmarshaler is the interface representing objects that can
// unmarshal themselves. The semantics are identical to Unmarshaler.
//
// This exists to support protoc-gen-go generated messages.
// The proto package will stop type-asserting to this interface in the future.
//
// DO NOT DEPEND ON THIS.
type newUnmarshaler interface {
    // 实现了XXX_Unmarshal
    XXX_Unmarshal([]byte) error
}

// Unmarshal parses the protocol buffer representation in buf and places the
// decoded result in pb.  If the struct underlying pb does not match
// the data in buf, the results can be unpredictable.
//
// Unmarshal resets pb before starting to unmarshal, so any
// existing data in pb is always removed. Use UnmarshalMerge
// to preserve and append to existing data.
func Unmarshal(buf []byte, pb Message) error {
    pb.Reset()
    // pb自己有unmarshal函数，实现了newUnmarshaler接口
    if u, ok := pb.(newUnmarshaler); ok {
        return u.XXX_Unmarshal(buf)
    }
    // pb自己有unmarshal函数，实现了Unmarshaler接口
    if u, ok := pb.(Unmarshaler); ok {
        return u.Unmarshal(buf)
    }
    // 使用默认的Unmarshal
    return NewBuffer(buf).Unmarshal(pb)
}

// request.pb.go
func (m *Request) XXX_Unmarshal(b []byte) error {
    return xxx_messageInfo_Request.Unmarshal(m, b)
}

// Unmarshal is the entry point from the generated .pb.go files.
// This function is not intended to be used by non-generated code.
// This function is not subject to any compatibility guarantee.
// msg contains a pointer to a protocol buffer struct.
// b is the data to be unmarshaled into the protocol buffer.
// a is a pointer to a place to store cached unmarshal information.
func (a *InternalMessageInfo) Unmarshal(msg Message, b []byte) error {
    // Load the unmarshal information for this message type.
    // The atomic load ensures memory consistency.
    // 获取保存在a中的unmarshal信息
    u := atomicLoadUnmarshalInfo(&a.unmarshal)
    if u == nil {
        // Slow path: find unmarshal info for msg, update a with it.
        u = getUnmarshalInfo(reflect.TypeOf(msg).Elem())
        atomicStoreUnmarshalInfo(&a.unmarshal, u)
    }
    // Then do the unmarshaling.
    // 执行unmarshal
    err := u.unmarshal(toPointer(&msg), b)
    return err
}

// unmarshal does the main work of unmarshaling a message.
// u provides type information used to unmarshal the message.
// m is a pointer to a protocol buffer message.
// b is a byte stream to unmarshal into m.
// This is top routine used when recursively unmarshaling submessages.
func (u *unmarshalInfo) unmarshal(m pointer, b []byte) error {
    if atomic.LoadInt32(&u.initialized) == 0 {
        // 为u填充unmarshal信息，以及设置每个字段类型的unmarshaler函数
        u.computeUnmarshalInfo()
    }
    if u.isMessageSet {
        return unmarshalMessageSet(b, m.offset(u.extensions).toExtensions())
    }
    var reqMask uint64 // bitmask of required fields we've seen.
    var errLater error
    for len(b) > 0 {
        // Read tag and wire type.
        // Special case 1 and 2 byte varints.
        var x uint64
        if b[0] < 128 {
            x = uint64(b[0])
            b = b[1:]
        } else if len(b) >= 2 && b[1] < 128 {
            x = uint64(b[0]&0x7f) + uint64(b[1])<<7
            b = b[2:]
        } else {
            var n int
            x, n = decodeVarint(b)
            if n == 0 {
                return io.ErrUnexpectedEOF
            }
            b = b[n:]
        }
        // 获取tag和wire标记
        tag := x >> 3
        wire := int(x) & 7

        // Dispatch on the tag to one of the unmarshal* functions below.
        // 根据tag选择该类型的unmarshalFieldInfo：f
        var f unmarshalFieldInfo
        if tag < uint64(len(u.dense)) {
            f = u.dense[tag]
        } else {
            f = u.sparse[tag]
        }
        // 如果该类型有unmarshaler函数，则执行解码和错误处理
        if fn := f.unmarshal; fn != nil {
            var err error
            // 从b解析，然后填充到f的对应字段
            b, err = fn(b, m.offset(f.field), wire)
            if err == nil {
                reqMask |= f.reqMask
                continue
            }
            if r, ok := err.(*RequiredNotSetError); ok {
                // Remember this error, but keep parsing. We need to produce
                // a full parse even if a required field is missing.
                if errLater == nil {
                    errLater = r
                }
                reqMask |= f.reqMask
                continue
            }
            if err != errInternalBadWireType {
                if err == errInvalidUTF8 {
                    if errLater == nil {
                        fullName := revProtoTypes[reflect.PtrTo(u.typ)] + "." + f.name
                        errLater = &invalidUTF8Error{fullName}
                    }
                    continue
                }
                return err
            }
            // Fragments with bad wire type are treated as unknown fields.
        }

        // Unknown tag.
        // 跳过未知tag，可能是proto中的message定义升级了，增加了一些字段，使用老版本的，就不识别新的字段
        if !u.unrecognized.IsValid() {
            // Don't keep unrecognized data; just skip it.
            var err error
            b, err = skipField(b, wire)
            if err != nil {
                return err
            }
            continue
        }
        // 检查未识别字段是不是extension
        // Keep unrecognized data around.
        // maybe in extensions, maybe in the unrecognized field.
        z := m.offset(u.unrecognized).toBytes()
        var emap map[int32]Extension
        var e Extension
        for _, r := range u.extensionRanges {
            if uint64(r.Start) <= tag && tag <= uint64(r.End) {
                if u.extensions.IsValid() {
                    mp := m.offset(u.extensions).toExtensions()
                    emap = mp.extensionsWrite()
                    e = emap[int32(tag)]
                    z = &e.enc
                    break
                }
                if u.oldExtensions.IsValid() {
                    p := m.offset(u.oldExtensions).toOldExtensions()
                    emap = *p
                    if emap == nil {
                        emap = map[int32]Extension{}
                        *p = emap
                    }
                    e = emap[int32(tag)]
                    z = &e.enc
                    break
                }
                panic("no extensions field available")
            }
        }

        // Use wire type to skip data.
        var err error
        b0 := b
        b, err = skipField(b, wire)
        if err != nil {
            return err
        }
        *z = encodeVarint(*z, tag<<3|uint64(wire))
        *z = append(*z, b0[:len(b0)-len(b)]...)

        if emap != nil {
            emap[int32(tag)] = e
        }
    }
    // 校验解析到的required字段的数量，如果与u中记录的不匹配，则报错
    if reqMask != u.reqMask && errLater == nil {
        // A required field of this message is missing.
        for _, n := range u.reqFields {
            if reqMask&1 == 0 {
                errLater = &RequiredNotSetError{n}
            }
            reqMask >>= 1
        }
    }
    return errLater
}

// typeUnmarshaler returns an unmarshaler for the given field type / field tag pair.
func typeUnmarshaler(t reflect.Type, tags string) unmarshaler {
    ...
    // Figure out packaging (pointer, slice, or both)
    slice := false
    pointer := false
    if t.Kind() == reflect.Slice && t.Elem().Kind() != reflect.Uint8 {
        slice = true
        t = t.Elem()
    }
    if t.Kind() == reflect.Ptr {
        pointer = true
        t = t.Elem()
    }
    ...
    switch t.Kind() {
    case reflect.Bool:
        if pointer {
            return unmarshalBoolPtr
        }
        if slice {
            return unmarshalBoolSlice
        }
        return unmarshalBoolValue
    }
}

func unmarshalBoolValue(b []byte, f pointer, w int) ([]byte, error) {
    if w != WireVarint {
        return b, errInternalBadWireType
    }
    // Note: any length varint is allowed, even though any sane
    // encoder will use one byte.
    // See https://github.com/golang/protobuf/issues/76
    x, n := decodeVarint(b)
    if n == 0 {
        return nil, io.ErrUnexpectedEOF
    }
    // TODO: check if x>1? Tests seem to indicate no.
    // toBool是返回bool类型的指针
    // 完成对字段f的赋值
    v := x != 0
    *f.toBool() = v
    return b[n:], nil
}