Tendermint abci 项目主页

这篇文章以 ABCI 示例 KVStore 应用及默认的 socket 连接为例说明 ABCI 应用的启动及 abci-cli 客户端与其交互的过程，以加深开发 ABCI 应用的模式及源码组织方式的理解。

整体流程说明

ABCI 应用服务端：在命令行执行 abci-cli kvstore 启动应用后，它会在 46658 端口等待客户端的 TCP 连接。

abci-cli 客户端：这个客户端指的是 echo 、info 和 deliverTx 等子命令。执行这些命令会建立一条与 ABCI 应用服务端的 TCP 连接，并将子命令后面的参数当作请求消息发送给应用服务端进行处理(这里的 ABCI 应用与 Tendermint 节点绑定在一起)。

构建命令过程

程序入口在 abci/cmd/abci-cli/main.go 的 Execute 函数。

这里面做事情有：

构建 RootCmd 命令，即 abci-cli 命令及各子命令。

注册全局 Flags，主要包括：

ABCI 应用服务端监听地址 flagAddress ，默认为：tcp://0.0.0.0:46658

abci-cli 客户端与 ABCI 应用服务端的通信协议 flagAbci，默认为：socket

日志等级 flagLogLevel，默认为：debug

添加 ABCI 应用 kvstore 和 dummy 以及 echoCmd、infoCmd、deliverTxCmd 和 commitCmd 等客户端命令。

RootCmd、kvstore 命令的实现及启动

RootCmd 主命令逻辑

所有子命令都要添加到 RootCmd 主命令下面。

执行 abci-cli 命令只会列出其使用文档，在执行具体子命令时才会执行其定义的相应应用逻辑。

这里只需看这段代码：

// 执行主命令时如果 client 为空，会创建 client 并启动，会根据 flagAbci 参数来判断要创建 socket 客户端

// 还是 RPC 客户端

if client == nil {

var err error

client, err = abcicli.NewClient(flagAddress, flagAbci, false)

if err != nil {

return err

}

client.SetLogger(logger.With("module", "abci-client"))

// 启动客户端，实际执行的是 client.OnStart() 函数

if err := client.Start(); err != nil {

return err

}

}

abci-cli 客户端

在命令行执行 abci-cli echo hello，会与 ABCI 应用服务端建立一条 TCP 连接并将 “abc” 发送到服务端进行处理，收到应后断开连接。但这样比较麻烦，可以使用 abci-cli console 命令可以在交互式命令行中与应用服务端交互。

通过调用 abcicli.NewClient 函数来创建客户端。

返回的是接口 abci/client/Client，这个接口有 socket、grpc 和 local 三种实现，但 flag 只可以指定前两种，默认为 socket。

func NewClient(addr, transport string, mustConnect bool) (client Client, err error) {

switch transport {

case "socket":

client = NewSocketClient(addr, mustConnect)

case "grpc":

client = NewGRPCClient(addr, mustConnect)

default:

err = fmt.Errorf("Unknown abci transport %s", transport)

}

return

}

abci/client/Client 接口继承了 tmlibs/common/service/Service 接口，可以启动、停止和重置。客户端和服务端都需要这些功能，使用时可以通过把 BaseService 作为自定义结构的匿名字段来实现。

首先看 socketClient 函数的结构，它实现了 abci/client/Client 接口，由于 cmn.BaseService 结构是它的匿名字段，也间接实现了 tmlibs/common/service/Service 接口：

type socketClient struct {

// 这个结构实现了 Service 接口，内部包含 impl Service字段，即具体实现 Service 的结构

cmn.BaseService

// 用来传递请求及应答消息的通道，

reqQueue chan *ReqRes

flushTimer *cmn.ThrottleTimer

mustConnect bool

mtx sync.Mutex

addr string

conn net.Conn

err error

// 标准库中的双向链表，从 reqQueue 读取到的请求消息都会先放入这个链表的尾端

reqSent *list.List

resCb func(*types.Request, *types.Response) // listens to all callbacks

}

创建 socketClient 的函数：

func NewSocketClient(addr string, mustConnect bool) *socketClient {

cli := &socketClient{

reqQueue: make(chan *ReqRes, reqQueueSize),

flushTimer: cmn.NewThrottleTimer("socketClient", flushThrottleMS),

mustConnect: mustConnect,

addr: addr,

reqSent: list.New(),

resCb: nil,

}

// 这里传入了具体实现 Service 的结构 cli

cli.BaseService = *cmn.NewBaseService(nil, "socketClient", cli)

return cli

}

在命令行执行 abci-cli kvstore 命令时会执行 client.Start() 启动服务，这里用默认的 socket 连接及 kvstore 应用举例说明。这里执行的是 socketClient 结构中匿名字段 cmn.BaseService 的方法：

func (bs *BaseService) Start() error {

if atomic.CompareAndSwapUint32(&bs.started, 0, 1) {

if atomic.LoadUint32(&bs.stopped) == 1 {

bs.Logger.Error(Fmt("Not starting %v -- already stopped", bs.name), "impl", bs.impl)

return ErrAlreadyStopped

}

bs.Logger.Info(Fmt("Starting %v", bs.name), "impl", bs.impl)

// 这里实际执行上面传入的 cli(即 socketClient 结构) 的 OnStart 函数来启动服务

err := bs.impl.OnStart()

if err != nil {

// revert flag

atomic.StoreUint32(&bs.started, 0)

return err

}

return nil

}

bs.Logger.Debug(Fmt("Not starting %v -- already started", bs.name), "impl", bs.impl)

return ErrAlreadyStarted

}

以上就是客户端的启动过程。

ABCI 应用服务端

现在看一下执行 abci-cli kvstore 命令都做了什么。

执行此命令时，实际执行的是 cmdKVStore 函数，启动了应用服务端，在 tcp://0.0.0.0:46658 监听连接。

启动服务端：

// 默认创建 socket 服务端

srv, err := server.NewServer(flagAddrD, flagAbci, app)

if err != nil {

return err

}

srv.SetLogger(logger.With("module", "abci-server"))

if err := srv.Start(); err != nil {

return err

}

NewSocketServer 创建服务端：

func NewSocketServer(protoAddr string, app types.Application) cmn.Service {

proto, addr := cmn.ProtocolAndAddress(protoAddr)

s := &SocketServer{

proto: proto,

addr: addr,

listener: nil,

app: app,

conns: make(map[int]net.Conn),

}

// 这里使用的模式与 Client 端一致

s.BaseService = *cmn.NewBaseService(nil, "ABCIServer", s)

return s

}

执行 srv.Start() 函数时，实际执行的是 SocketServer 的实现，通过 BaseService 结构的 Start 方法调用：

func (s *SocketServer) OnStart() error {

if err := s.BaseService.OnStart(); err != nil {

return err

}

ln, err := net.Listen(s.proto, s.addr)

if err != nil {

return err

}

s.listener = ln

// 启动一个协程来监听连接

go s.acceptConnectionsRoutine()

return nil

}

至此已经把 ABCI 应用服务端是如何启动的说明了，下面的部分会详细说明请求及应答处理的细节。

请求及应答处理

这部分以 deliver_tx 命令为例来进行说明。

abci-cli 客户端

为了方便，这里再看一下 socketClient 的数据结构：

type socketClient struct {

cmn.BaseService

reqQueue chan *ReqRes

flushTimer *cmn.ThrottleTimer

mustConnect bool

mtx sync.Mutex

addr string

conn net.Conn

err error

// 这里会把请求写入双向链表 reqSent 的尾端，在 recvResponseRoutine 函数中接收到应答时会从此链表取出

// 第一个请求进行类型比较，如果与应答类型一样则返回给前端

reqSent *list.List

resCb func(*types.Request, *types.Response) // listens to all callbacks

}

OnStart 函数所做的就是与服务端建立连接，启动两个协程来处理请求与应答。

先看处理请求的函数 sendRequestsRoutine：

func (cli *socketClient) sendRequestsRoutine(conn net.Conn) {

w := bufio.NewWriter(conn)

for {

select {

// 发送 flush 类型请求的定时器

case

select {

case cli.reqQueue

default:

// Probably will fill the buffer, or retry later.

}

case

return

case reqres :=

// 这里会把请求写入双向链表 reqSent 的尾端

cli.willSendReq(reqres)

// 把请求消息写入连接缓冲，这时还没有发送给服务端

err := types.WriteMessage(reqres.Request, w)

if err != nil {

cli.StopForError(fmt.Errorf("Error writing msg: %v", err))

return

}

// 如果请求是 flush 类型，会把缓冲的请求消息 (包括此 flush 请求) 写入连接，

// 由 kvstore 服务端接收并处理。

// 有两种方式发送 flush 类型的请求：1) 定时器触发；2) DeliverTxSync 函数中主动发送

if _, ok := reqres.Request.Value.(*types.Request_Flush); ok {

err = w.Flush()

if err != nil {

cli.StopForError(fmt.Errorf("Error flushing writer: %v", err))

return

}

}

}

}

}

现在看处理应答的 recvResponseRoutine 函数：

func (cli *socketClient) recvResponseRoutine(conn net.Conn) {

r := bufio.NewReader(conn) // Buffer reads

for {

var res = &types.Response{}

// 从连接中读取应答，出错时会关闭连接并执行 flushQueue() 释放 wg.WaitGroup

err := types.ReadMessage(r, res)

if err != nil {

cli.StopForError(err)

return

}

switch r := res.Value.(type) {

case *types.Response_Exception:

cli.StopForError(errors.New(r.Exception.Error))

return

default:

// 应答处理逻辑在这里

err := cli.didRecvResponse(res)

if err != nil {

cli.StopForError(err)

return

}

}

}

}

func (cli *socketClient) didRecvResponse(res *types.Response) error {

cli.mtx.Lock()

defer cli.mtx.Unlock()

// 从双向链表 reqSent 中取出第一个请求

next := cli.reqSent.Front()

if next == nil {

return fmt.Errorf("Unexpected result type %v when nothing expected", reflect.TypeOf(res.Value))

}

reqres := next.Value.(*ReqRes)

// 检查请求与应答的类型是否匹配

if !resMatchesReq(reqres.Request, res) {

return fmt.Errorf("Unexpected result type %v when response to %v expected",

reflect.TypeOf(res.Value), reflect.TypeOf(reqres.Request.Value))

}

reqres.Response = res // Set response

reqres.Done() // 释放此请求创建时执行的 wg.Add(1)

cli.reqSent.Remove(next) // 从链表中删除第一个请求

// Notify reqRes listener if set

if cb := reqres.GetCallback(); cb != nil {

cb(res)

}

// Notify client listener if set

if cli.resCb != nil {

cli.resCb(reqres.Request, res)

}

return nil

}

ABCI 应用服务端

创建应用细节：

先看应用的数据结构 KVStoreApplication：

// 这个结构只实现了 Info、DeliverTx、CheckTx、Commit 和 Query 方法

type KVStoreApplication struct {

// 这个基础结构实现了 "tendermint/abci/types/Application" 接口(此项目中基本都是用的这种模式)。

// 这个结构实现的接口的方法中没有具体应用逻辑，以供开发者在自己的应用结构中继承此结构后，可以只实现

// 必须的方法，而无需实现接口的全部方法

types.BaseApplication

state State

}

创建应用：

func NewKVStoreApplication() *KVStoreApplication {

state := loadState(dbm.NewMemDB())

return &KVStoreApplication{state: state}

}

主要看 loadState 函数，它根据键 stateKey 从内存存储 MemDB 结构中获取对应状态，因为是初始化，肯定没有对应值，返回的是一个带有新建 MemDB (就是一个带锁的 map)的 State：

func loadState(db dbm.DB) State {

stateBytes := db.Get(stateKey)

var state State

if len(stateBytes) != 0 {

err := json.Unmarshal(stateBytes, &state)

if err != nil {

panic(err)

}

}

state.db = db

return state

}

服务端处理请求及应答细节：

重点看接受连接的函数：

func (s *SocketServer) acceptConnectionsRoutine() {

for {

// 接受连接，下面这些日志就是命令行启动 kvstore 后看到的信息

s.Logger.Info("Waiting for new connection...")

conn, err := s.listener.Accept()

if err != nil {

if !s.IsRunning() {

return // Ignore error from listener closing.

}

s.Logger.Error("Failed to accept connection: " + err.Error())

continue

}

s.Logger.Info("Accepted a new connection")

// 可以接受多条连接并记录

connID := s.addConn(conn)

closeConn := make(chan error, 2) // Push to signal connection closed

responses := make(chan *types.Response, 1000) // A channel to buffer responses

// 从连接读取请求并处理

go s.handleRequests(closeConn, conn, responses)

// 从 'responses' 获取应答并写到连接中

go s.handleResponses(closeConn, conn, responses)

// 等待信号来关闭连接

go s.waitForClose(closeConn, connID)

}

}

先看处理请求的 handleRequests 函数：

func (s *SocketServer) handleRequests(closeConn chan error, conn net.Conn, responses chan

var count int

var bufReader = bufio.NewReader(conn)

for {

var req = &types.Request{}

// 从连接上读取请求消息，读取完毕或出错后要通知 waitForClose 协程来关闭连接

err := types.ReadMessage(bufReader, req)

if err != nil {

if err == io.EOF {

closeConn

} else {

closeConn

}

return

}

s.appMtx.Lock()

count++

// 处理请求时要加锁，这个函数会根据请求的类型调用具体函数来处理，

// 比如 types.Request_DeliverTx 类型时就会调用 KVStoreApplication.DeliverTx 函数来处理，

// 应答会写入 responses 通道，以便 handleResponses 函数处理

s.handleRequest(req, responses)

s.appMtx.Unlock()

}

}

现在看 handleResponses 函数：

func (s *SocketServer) handleResponses(closeConn chan error, conn net.Conn, responses

var count int

var bufWriter = bufio.NewWriter(conn)

for {

// 从 responses 通道读取应答并写入连接。同样，出错时要通知 waitForClose 来关闭连接

var res =

err := types.WriteMessage(res, bufWriter)

if err != nil {

closeConn

return

}

// flush 类型的应答是哪里来的？

// 与客户端处理类似，如果是此类型要进行 Flush 处理，把缓冲的数据写入连接

if _, ok := res.Value.(*types.Response_Flush); ok {

err = bufWriter.Flush()

if err != nil {

closeConn

return

}

}

count++

}

}

deliver_tx 子命令执行过程

现在以此命令为例说明发起请求及收到应答的整体过程。

服务端启动，启动两个协程，一个处理请求，一个处理应答。

在命令行输入 abci-cli console 进入交互模式，创建客户端并与服务端建立了持久连接。服务端和客户端各启动两个协程，一个处理请求，一个处理应答。

输入 deliver_tx "abc"，客户端会识别子命令，调用 cmdDeliverTx 函数，此函数在解析到 tx 后进行编码，随后会调用 cli.DeliverTxSync(txBytes) (同步的) 函数。

请求消息写入连接后，服务端读取到此请求。

根据请求类型，调用 KVStore 应用的 DeliverTx 函数进行处理。

服务端处理完毕后的应答写入连接，客户端接收到应答，前端打印显示到命令行。