想寫點東西記錄一下最近看的一些Fabric源碼,本文使用的是fabric1.4的版本，所以對於其他版本的fabric，內容可能會有所不同。
本文想針對Fabric中鏈碼容器的啟動過程進行源碼的解析。這里的鏈碼指的是用戶鏈碼不是系統鏈碼,順便回顧一下系統鏈碼:
lscc(Life Cycle System ChainCode)生命周期系統鏈碼
cscc(Configuration System ChainCode)配置系統鏈碼
escc(Endorser System ChainCode)背書系統鏈碼
qscc(Query System ChainCode)查詢系統鏈碼
vscc(Verification System ChainCode)驗證系統鏈碼
本文主要解析的是用戶鏈碼的啟動過程。
1 起點
#這是用戶端鏈碼的main方法，也是整個流程的入口點，調用了shim包中的Start(cc Chaincode)方法.
func main(){
err :=shim.Start(new(Chaincode))
if err != nil {
fmt.Printf("Error starting Chaincode: %s",err)
}
}
首先定位到fabric/core/chaincode/shim/chaincode.go這個文件中的Start方法，這里是鏈碼啟動的起點。
可以看到傳的參數就是chaincode,接下來分析一下啟動過程
#方法中第一行代碼，根據名字可以看出是對鏈碼的Log進行設置
SetupChaincodeLogging()
#從輸入中獲取用戶定義的鏈碼的名稱
chaincodename := viper.GetString("chaincode.id.name")
#如果沒有輸入鏈碼名稱，直接返回沒有提供鏈碼id的錯誤，下面則不再執行
if chaincodename == "" {
return errors.New("error chaincode id not provided")
}
#看名字是一個工廠方法，點進行看一下
err := factory.InitFactories(factory.GetDefaultOpts())
首先進入到factory.GetDefaultOpts()方法中：
func GetDefaultOpts() *FactoryOpts {
return &FactoryOpts{
ProviderName: "SW",
SwOpts: &SwOpts{
HashFamily: "SHA2", #HASH類型
SecLevel: 256, #HASH級別
Ephemeral: true,
},
}
}
#可以猜到這個方法是獲取默認的加密操作，使用SHA256進行數據加密
不難猜到factory.InitFactories這個方法就是為當前鏈碼設置加密操作的一系列內容。回到Start()方法中接着往下看.
#這一部分就是將鏈碼數據以流的方式讀取進來，userChaincodeStreamGetter是一個方法，點進去看一下
if streamGetter == nil {
streamGetter = userChaincodeStreamGetter
}
stream, err := streamGetter(chaincodename)
if err != nil {
return err
}
userChaincodeStreamGetter還是在這個文件中第82行:
#這里的name是鏈碼名稱，讀取到鏈碼數據后以PeerChainCodeStream的方式返回
func userChaincodeStreamGetter(name string) (PeerChaincodeStream, error) {
#獲取peer.address
flag.StringVar(&peerAddress, "peer.address", "", "peer address")
//判斷是否使能TLS
if viper.GetBool("peer.tls.enabled") {
#獲取tls密鑰地址，在用戶安裝鏈碼的時候指定
keyPath := viper.GetString("tls.client.key.path")
#獲取tls證書地址
certPath := viper.GetString("tls.client.cert.path")
#從文件中讀取密鑰數據
data, err1 := ioutil.ReadFile(keyPath)
if err1 != nil {
err1 = errors.Wrap(err1, fmt.Sprintf("error trying to read file content %s", keyPath))
chaincodeLogger.Errorf("%+v", err1)
return nil, err1
}
key = string(data)
#從文件中讀取證書數據
data, err1 = ioutil.ReadFile(certPath)
if err1 != nil {
err1 = errors.Wrap(err1, fmt.Sprintf("error trying to read file content %s", certPath))
chaincodeLogger.Errorf("%+v", err1)
return nil, err1
}
cert = string(data)
}
#解析命令行參數到定義的flag
flag.Parse()
#日志輸出
chaincodeLogger.Debugf("Peer address: %s", getPeerAddress())
//與peer節點建立連接
clientConn, err := newPeerClientConnection()
看一下這個方法里面的內容，還是這個文件第317行：
func newPeerClientConnection() (*grpc.ClientConn, error) {
#首先獲取到peer節點的地址
var peerAddress = getPeerAddress()
#看名字就知道了，設置與鏈碼之間的心中信息
kaOpts := &comm.KeepaliveOptions{
ClientInterval: time.Duration(1) * time.Minute,
ClientTimeout: time.Duration(20) * time.Second,
}
判斷是否使能了TLS，然后根據結果建立鏈接,如何建立鏈接就不再細看了，我們回到之前的部分
if viper.GetBool("peer.tls.enabled") {
return comm.NewClientConnectionWithAddress(peerAddress, true, true,
comm.InitTLSForShim(key, cert), kaOpts)
}
return comm.NewClientConnectionWithAddress(peerAddress, true, false, nil, kaOpts)
}
還是之前的userChaincodeStreamGetter方法
clientConn, err := newPeerClientConnection()
if err != nil {
err = errors.Wrap(err, "error trying to connect to local peer")
chaincodeLogger.Errorf("%+v", err)
return nil, err
}
chaincodeLogger.Debugf("os.Args returns: %s", os.Args)
#接下來是這個方法，返回一個ChaincodeSupportClient實例,對應着鏈碼容器
chaincodeSupportClient := pb.NewChaincodeSupportClient(clientConn)
//這一步是與peer節點建立gRPC連接
stream, err := chaincodeSupportClient.Register(context.Background())
if err != nil {
return nil, errors.WithMessage(err, fmt.Sprintf("error chatting with leader at address=%s", getPeerAddress()))
}
return stream, nil
}
這個方法結束之后，鏈碼容器與Peer節點已經建立起了連接，接下來鏈碼容器與Peer節點開始互相發送消息了。
返回到Start()方法中，還剩最后的一個方法chatWithPeer()：
err = chatWithPeer(chaincodename, stream, cc)
return err
}
看一下鏈碼容器與Peer節點是如何互相通信的。這個方法是鏈碼容器啟動的過程中最重要的方法，包含所有的通信流程。chatWithPeer()在331行:
func chatWithPeer(chaincodename string, stream PeerChaincodeStream, cc Chaincode)
#傳入的參數有鏈碼名稱，流(這個是之前鏈碼容器與Peer節點建立gRPC連接所返回的)，鏈碼
首先第一步是新建一個ChaincodeHandler對象：是非常重要的一個對象。看一下該對象的內容,在core/chaincode/shim/handler.go文件中第166行:
func newChaincodeHandler(peerChatStream PeerChaincodeStream, chaincode Chaincode) *Handler {
v := &Handler{
ChatStream: peerChatStream, #與Peer節點通信的流
cc: chaincode, #鏈碼
}
v.responseChannel = make(map[string]chan pb.ChaincodeMessage) #鏈碼信息響應通道
v.state = created #表示將鏈碼容器的狀態更改為created
return v 將handler返回
}
這個ChaincodeHandler對象是鏈碼側完成鏈碼與Peer節點之前所有的消息的控制邏輯。
繼續往下看：
#在方法執行結束的時候關閉gRPC連接
defer stream.CloseSend()
#獲取鏈碼名稱
chaincodeID := &pb.ChaincodeID{Name: chaincodename}
#將獲取的鏈碼名稱序列化為有效載荷.
payload, err := proto.Marshal(chaincodeID)
if err != nil {
return errors.Wrap(err, "error marshalling chaincodeID during chaincode registration")
}
#日志輸出,這個日志信息在安裝鏈碼的時候應該有看到過吧
chaincodeLogger.Debugf("Registering.. sending %s", pb.ChaincodeMessage_REGISTER)
#鏈碼容器通過handler開始通過gRPC連接向Peer節點發送第一個消息了，鏈碼容器向Peer節點發送REGISTER消息，並附上鏈碼的名稱
if err = handler.serialSend(&pb.ChaincodeMessage{Type: pb.ChaincodeMessage_REGISTER, Payload: payload}); err != nil {
return errors.WithMessage(err, "error sending chaincode REGISTER")
}
#定義一個接收消息的結構體
type recvMsg struct {
msg *pb.ChaincodeMessage
err error
}
msgAvail := make(chan *recvMsg, 1)
errc := make(chan error)
receiveMessage := func() {
in, err := stream.Recv()
msgAvail <- &recvMsg{in, err}
}
#接收由Peer節點返回的響應消息
go receiveMessage()
接下來的部分就是鏈碼容器與Peer節點詳細的通信過程了：
2鏈碼側向Peer節點發送REGISTER消息
#前面的部分都是接收到錯誤消息的各種輸出邏輯，不再細看，我們看default這一部分，這一部分是正常情況下消息的處理情況：
for {
select {
case rmsg := <-msgAvail:
switch {
case rmsg.err == io.EOF:
err = errors.Wrapf(rmsg.err, "received EOF, ending chaincode stream")
chaincodeLogger.Debugf("%+v", err)
return err
case rmsg.err != nil:
err := errors.Wrap(rmsg.err, "receive failed")
chaincodeLogger.Errorf("Received error from server, ending chaincode stream: %+v", err)
return err
case rmsg.msg == nil:
err := errors.New("received nil message, ending chaincode stream")
chaincodeLogger.Debugf("%+v", err)
return err
default:
#這一句日志輸出應該看到過好多次吧。
chaincodeLogger.Debugf("[%s]Received message %s from peer", shorttxid(rmsg.msg.Txid), rmsg.msg.Type)
#重要的一個方法，在鏈碼容器與Peer節點建立起了聯系后，主要通過該方法對消息邏輯進行處理，我們點進行看一下。
err := handler.handleMessage(rmsg.msg, errc)
if err != nil {
err = errors.WithMessage(err, "error handling message")
return err
}
#當消息處理完成后，再次接收消息。
go receiveMessage()
}
#最后是發送失敗的處理
case sendErr := <-errc:
if sendErr != nil {
err := errors.Wrap(sendErr, "error sending")
return err
}
}
}
一個重要的方法：handleMessage在core/chaincode/shim/handler.go文件第801行：
func (handler *Handler) handleMessage(msg *pb.ChaincodeMessage, errc chan error) error {
#如果鏈碼容器接收到Peer節點發送的心跳消息后，直接將心跳消息返回，雙方就一直保持聯系。
if msg.Type == pb.ChaincodeMessage_KEEPALIVE {
chaincodeLogger.Debug("Sending KEEPALIVE response")
handler.serialSendAsync(msg, nil) // ignore errors, maybe next KEEPALIVE will work
return nil
}
#我們先看到這里，如果再往下看的話可能會亂掉，所以還是按照邏輯順序進行說明。
先說一下鏈碼側所做的工作：
- 首先進行各項基本配置，然后建立起與Peer節點的gRPC連接。
- 創建
Handler,並更改Handler狀態為created。 - 發送
REGISTER消息到Peer節點。 - 等待Peer節點返回的信息
3Peer節點接收到REGISTER消息后
之前講的都是鏈碼側的一系列流程，我們之前提到鏈碼側與Peer節點之間的第一個消息內容是由鏈碼側發送至Peer節點的REGISTER消息。接下來我們看一下Peer節點在接收到該消息后是如果進行處理的。
代碼在core/chaincode/handler.go文件中第174行，這里不是處理消息的開始，但是對於我們要說的鏈碼容器啟動過程中消息的處理剛好銜接上，所以就直接從這里開始了。另外很重要的一點，這里已經轉換到Peer節點側了，不是之前說的鏈碼側，我們看一下代碼：
func (h *Handler) handleMessage(msg *pb.ChaincodeMessage) error {
chaincodeLogger.Debugf("[%s] Fabric side handling ChaincodeMessage of type: %s in state %s", shorttxid(msg.Txid), msg.Type, h.state)
#這邊也是首先判斷是不是心跳信息，如果是心跳信息的話就什么也不做，與之前不同的是鏈碼側在收到心跳信息后會返回Peer節點一個心跳信息。
if msg.Type == pb.ChaincodeMessage_KEEPALIVE {
return nil
}
#之前我們提到，創建handler時，更改狀態為created,所以這里進入到handleMessageCreatedState這個方法內.
switch h.state {
case Created:
return h.handleMessageCreatedState(msg)
case Ready:
return h.handleMessageReadyState(msg)
default:
return errors.Errorf("handle message: invalid state %s for transaction %s", h.state, msg.Txid)
}
}
handleMessageCreatedState這個方法在第191行,方法內容很簡單，判斷消息類型是不是REGISTER，如果是則進入HandlerRegister(msg)方法內，如果不是則返回錯誤信息。
func (h *Handler) handleMessageCreatedState(msg *pb.ChaincodeMessage) error {
switch msg.Type {
case pb.ChaincodeMessage_REGISTER:
h.HandleRegister(msg)
default:
return fmt.Errorf("[%s] Fabric side handler cannot handle message (%s) while in created state", msg.Txid, msg.Type)
}
return nil
}
接下來我們看一下HandleRegister這個方法,在第495行：
func (h *Handler) HandleRegister(msg *pb.ChaincodeMessage) {
chaincodeLogger.Debugf("Received %s in state %s", msg.Type, h.state)
#獲取鏈碼ID
chaincodeID := &pb.ChaincodeID{}
#反序列化
err := proto.Unmarshal(msg.Payload, chaincodeID)
if err != nil {
chaincodeLogger.Errorf("Error in received %s, could NOT unmarshal registration info: %s", pb.ChaincodeMessage_REGISTER, err)
return
}
h.chaincodeID = chaincodeID
#這一行就是將鏈碼注冊到當前Peer節點上
err = h.Registry.Register(h)
if err != nil {
h.notifyRegistry(err)
return
}
從Peer節點側的handler獲取鏈碼名稱
h.ccInstance = ParseName(h.chaincodeID.Name)
chaincodeLogger.Debugf("Got %s for chaincodeID = %s, sending back %s", pb.ChaincodeMessage_REGISTER, chaincodeID, pb.ChaincodeMessage_REGISTERED)
#然后將REGISTERED消息返回給鏈碼側
if err := h.serialSend(&pb.ChaincodeMessage{Type: pb.ChaincodeMessage_REGISTERED}); err != nil {
chaincodeLogger.Errorf("error sending %s: %s", pb.ChaincodeMessage_REGISTERED, err)
h.notifyRegistry(err)
return
}
//更新handler狀態為Established
h.state = Established
chaincodeLogger.Debugf("Changed state to established for %+v", h.chaincodeID)
#還有這個方法也要看一下
h.notifyRegistry(nil)
}
簡單來說HandleRegister的功能就是將鏈碼注冊到Peer節點上，並發送RESIGSERED到鏈碼側，最后更新handler狀態為Established，我們看一下notifyRegistry方法,在478行：
func (h *Handler) notifyRegistry(err error) {
if err == nil {
//再往里面看,方法在459行
err = h.sendReady()
}
if err != nil {
h.Registry.Failed(h.chaincodeID.Name, err)
chaincodeLogger.Errorf("failed to start %s", h.chaincodeID)
return
}
h.Registry.Ready(h.chaincodeID.Name)
}
#sendReady()
func (h *Handler) sendReady() error {
chaincodeLogger.Debugf("sending READY for chaincode %+v", h.chaincodeID)
ccMsg := &pb.ChaincodeMessage{Type: pb.ChaincodeMessage_READY}
#Peer節點又向鏈碼容器發送了READY消息
if err := h.serialSend(ccMsg); err != nil {
chaincodeLogger.Errorf("error sending READY (%s) for chaincode %+v", err, h.chaincodeID)
return err
}
#同時更新handler狀態為Ready
h.state = Ready
chaincodeLogger.Debugf("Changed to state ready for chaincode %+v", h.chaincodeID)
return nil
}
到這里，Peer節點暫時分析完成，又到了鏈碼側對Peer節點發送的消息進行處理的流程.
我們先總結一下這一部分Peer節點做了哪些工作：
- 首先當Peer節點接收到鏈碼側發送的
REGISTER消息后，將鏈碼注冊到Peer端的Handler上，發送REGISTERED到鏈碼側，更新Handler的狀態為Established。 - 然后Peer節點向鏈碼側發送
READY消息，同時更新Handler的狀態為Ready。
4鏈碼側的回應
我們回到鏈碼側之前的這一部分core/chaincode/chaincode.go中第364行,這里是鏈碼鍘對接收到的Peer節點發送的消息進行處理的邏輯,至於發生錯誤的情況就不再說明，我們看handleMessage這個方法。
go receiveMessage()
for {
#相關代碼
...
err := handler.handleMessage(rmsg.msg, errc)
...
#相關代碼
go receiveMessage()
}
handleMessage這個方法在core/chaincode/shim/handler.go這個文件中，第801行。
#主要就是這一部分：
switch handler.state {
case ready:
err = handler.handleReady(msg, errc)
case established:
err = handler.handleEstablished(msg, errc)
case created:
err = handler.handleCreated(msg, errc)
default:
err = errors.Errorf("[%s] Chaincode handler cannot handle message (%s) with payload size (%d) while in state: %s", msg.Txid, msg.Type, len(msg.Payload), handler.state)
}
- 首先鏈碼側接收到Peer節點發送的
REGISTERED消息后，這里鏈碼側的handler與Peer節點側的handler並不是同一個，不要搞混了。判斷當前鏈碼側handler的狀態為created，進入到handleCreated方法中,在792行：
#將鏈碼側的handler的狀態更改為established
if msg.Type == pb.ChaincodeMessage_REGISTERED {
handler.state = established
return nil
}
- 當鏈碼側接收到Peer節點發送的
READY消息后，又一次進入上面的邏輯，由於鏈碼側的handler的狀態已經更改為established,所以這次進入到handleEstablished方法中。在783行：
#然后將鏈碼側的handler的狀態更改為ready
if msg.Type == pb.ChaincodeMessage_READY {
handler.state = ready
return nil
}
- 另外，當用戶對鏈碼進行實例化操作時，會通過Peer節點向鏈碼側發送
INIT消息，這里涉及到背書過程，之后再對背書過程進行討論，我們在這里只關注鏈碼側接收到INIT消息后的邏輯，還是handleMessage這個方法中：
#當判斷到消息類型為INIT時，會執行這個方法。
handler.handleInit(msg, errc)
handler.handleInit(msg, errc)方法在第177行：
func (handler *Handler) handleInit(msg *pb.ChaincodeMessage, errc chan error) {
go func() {
var nextStateMsg *pb.ChaincodeMessage
defer func() {
#這一名相當於更新鏈碼的狀態
handler.triggerNextState(nextStateMsg, errc)
}()
#判斷錯誤信息
errFunc := func(err error, payload []byte, ce *pb.ChaincodeEvent, errFmt string, args ...interface{}) *pb.ChaincodeMessage {
if err != nil {
// Send ERROR message to chaincode support and change state
if payload == nil {
payload = []byte(err.Error())
}
chaincodeLogger.Errorf(errFmt, args...)
return &pb.ChaincodeMessage{Type: pb.ChaincodeMessage_ERROR, Payload: payload, Txid: msg.Txid, ChaincodeEvent: ce, ChannelId: msg.ChannelId}
}
return nil
}
#獲取用戶輸入的參數
input := &pb.ChaincodeInput{}
#反序列化
unmarshalErr := proto.Unmarshal(msg.Payload, input)
if nextStateMsg = errFunc(unmarshalErr, nil, nil, "[%s] Incorrect payload format. Sending %s", shorttxid(msg.Txid), pb.ChaincodeMessage_ERROR.String()); nextStateMsg != nil {
return
}
#ChaincodeStub應該很熟悉了,很重要的一個對象，包含一項提案中所需要的內容。在``core/chaincode/shim/chaincode.go``文件中第53行，有興趣可以點進去看一下
stub := new(ChaincodeStub)
#這一行代碼的意思就是將提案中的信息抽取出來賦值到ChaincodeStub這個對象中
err := stub.init(handler, msg.ChannelId, msg.Txid, input, msg.Proposal)
if nextStateMsg = errFunc(err, nil, stub.chaincodeEvent, "[%s] Init get error response. Sending %s", shorttxid(msg.Txid), pb.ChaincodeMessage_ERROR.String()); nextStateMsg != nil {
return
}
#這里的Init方法就是鏈碼中所寫的Init()方法，就不再解釋了
res := handler.cc.Init(stub)
chaincodeLogger.Debugf("[%s] Init get response status: %d", shorttxid(msg.Txid), res.Status)
#ERROR的值為500,OK=200，ERRORTHRESHOLD = 400，大於等於400就代表錯誤信息或者被背書節點拒絕。
if res.Status >= ERROR {
err = errors.New(res.Message)
if nextStateMsg = errFunc(err, []byte(res.Message), stub.chaincodeEvent, "[%s] Init get error response. Sending %s", shorttxid(msg.Txid), pb.ChaincodeMessage_ERROR.String()); nextStateMsg != nil {
return
}
}
resBytes, err := proto.Marshal(&res)
if err != nil {
payload := []byte(err.Error())
chaincodeLogger.Errorf("[%s] Init marshal response error [%s]. Sending %s", shorttxid(msg.Txid), err, pb.ChaincodeMessage_ERROR)
nextStateMsg = &pb.ChaincodeMessage{Type: pb.ChaincodeMessage_ERROR, Payload: payload, Txid: msg.Txid, ChaincodeEvent: stub.chaincodeEvent}
return
}
// Send COMPLETED message to chaincode support and change state
nextStateMsg = &pb.ChaincodeMessage{Type: pb.ChaincodeMessage_COMPLETED, Payload: resBytes, Txid: msg.Txid, ChaincodeEvent: stub.chaincodeEvent, ChannelId: stub.ChannelId}
chaincodeLogger.Debugf("[%s] Init succeeded. Sending %s", shorttxid(msg.Txid), pb.ChaincodeMessage_COMPLETED)
#到這里就結束了，會調用上面的handler.triggerNextState(nextStateMsg, errc)方法，這個方法將初始化數據與COMPLETED狀態發送至Peer節點。
}()
}
這個方法還是比較簡單的，一共做了這些事情：
- 獲取用戶的輸入數據
- 新建一個
ChainCodeStub對象，然后將用戶輸入的數據賦值給該對象 - 調用用戶鏈碼中的
Init()方法 - 將所有數據封裝成
ChainCodeMessage，類型為COMPLETED,發送至Peer節點。
這個時候鏈碼已經初始化完成，已經進入了可被調用(invoke)的狀態.
之后的流程就差不多了，Peer節點發送TRANSACTION消息給鏈碼側，調用Invoke()方法，之后鏈碼側發送具體的調用方法到Peer節點，由Peer節點進行相應的處理，最后返回RESPONSE消息到鏈碼側，鏈碼側接收到RESPONSE消息后，返回COMPLETED消息到Peer節點。
5總結
到這里，Peer節點與鏈碼側的handler都處於READY狀態,鏈碼容器已經啟動完成。最后總結一下整體的流程：
- 通過用戶端鏈碼中的
main方法，調用了core/chaincode/shim/chaincode.go中的Start()方法，從而開始了鏈碼的啟動。 - 首先進行相關的配置比如基本的加密，證書的讀取。
- 創建與Peer節點之間的gRPC連接，創建
handler實例。 - 由鏈碼容器向Peer節點發送第一個消息:
REGISTER,然后等待接收由Peer節點發送的消息。如果接收到的是心跳消息，則向Peer節點返回心跳消息。 - Peer節點接收到鏈碼容器發送的
REGISTER消息后，將其注冊到Peer節點端的handler上。 - Peer節點發送
REGISTERED消息到鏈碼側，同時更新Peer節點端的handler狀態為Established。 - Peer節點發送
Ready消息到鏈碼側，同時更新Peer節點端的handler狀態為Ready。 - 鏈碼側接收到由Peer節點發送的
REGISTERED消息后，更新鏈碼側的handler狀態為Established。 - 鏈碼側接收到由Peer節點發送的
READY消息后，更新鏈碼側的handler狀態為ready。 - 當用戶執行實例化鏈碼時，通過Peer節點向鏈碼側發送
INIT消息。鏈碼側接收到INIT消息后，根據用戶輸入的參數進行實例化操作。實例化完成后，返回COMPLETED消息到Peer節點。 - 到這里鏈碼容器已經啟動，可以對鏈碼數據進行查詢調用等操作了。
另外，閱讀Fabric源碼中有一些沒有看明白或者分析有誤的地方，還望大家能夠批評指正。