>

字数 1.8k 阅读 12 评论 6

介绍

ABCI 是 Tendermint 中定义的一套 Core 和 App 交互的接口，它把共识引擎和应用层的逻辑进行解绑，可以帮助实现基于同一引擎上的高度定制化的 App 应用程序。

由于这种出色的架构设计把共识引擎独立出来，因此整个 tendermint 可以作为许多开源项目中一个可替代的底层共识机制来运作，大大的增加了其应用场景。

本文主要介绍 ABCI 的设计，以及 tendermint core 与 之的关系

下面是一段官方简介

Tendermint Core would be responsible for

  • Sharing blocks and transactions between nodes
  • Establishing a canonical/immutable order of transactions (the blockchain)

The application will be responsible for

  • Maintaining the UTXO database
  • Validating cryptographic signatures of transactions
  • Preventing transactions from spending non-existent transactions
  • Allowing clients to query the UTXO database.

翻译成中文就是

Core 所做的事

  • 节点之间共享 blocks 和 transactions 信息
  • 建立不可更改的交易顺序

Application 所做的事

  • 维护 UTXO database
  • 验证签名的有效性
  • 验证交易的合法性
  • 运行 client 去查询 UTXO database

总结来说

Core 是共识引擎，负责进行交易的共识
Application 则是状态机，记录交易开始的状态和最终 Commit 的状态

App 发出一笔交易，把状态（交易）信息发给共识引擎，共识引擎完成共识过程之后，把共识结果发回给 App，由 App 最终写入持久化。

ABCI 主要以下几个要点：

（1） 消息协议 （message protocol）

  • protobuf,
  • consensus makes requests, application responds

（2）Server/Client

  • consensus engine runs client
  • application runs server
  • two implementations: async raw bytes, grpc

（3） Blockchain protocol

Tendermint Core 主要维护 3 类连接：

  • mempool connection: CheckTx
  • consensus connnections: 只有 commited 的 transactions 才会被 executed。
  • query connections: only uses Query and Info

Note: The mempool and consensus logic act as clients, and each maintains an open ABCI connection with the application, which hosts an ABCI server.

ABCI Server

要在不同的语言中使用 ABCI，这个语言必须要实现一个 ABCI server。

简单来说，需要实现：

  • a socket server
  • a handler for ABCI Messages

具体如下：

####（1）ABCI server

1. Socket Server
1
2
3
4
5
6
7
8
9
10
func (s *SocketServer) OnStart() error {
    s.BaseService.OnStart()
    ln, err := net.Listen(s.proto, s.addr)
    if err != nil {
        return err
    }
    s.listener = ln
    go s.acceptConnectionsRoutine()
    return nil
}

注：这里我们可以既通过 socket RPC 来实现（效率高），也可以通过 gRPC 来实现（效率低）。

    1. Asynchronous raw byte server (比如 Tendermint Socket Protocol，Known as TMSP or Teaspoon)
    1. GRPC
2. ABCI Server Implementation

在 ABCI Server 中，源码在 abci/server/ 目录下：
server/
├── grpc_server.go
├── server.go
└── socket_server.go

比如，对于普通的 RPC 来说，在 socket_server.go 中的 handleRequest() 函数中，会根据 types.Request 的类型来调用 application 接口中的相关方法。

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
// Pull responses from 'responses' and write them to conn.
func (s *SocketServer) handleResponses(closeConn chan error, responses <-chan *types.Response, conn net.Conn) {
    var count int
    var bufWriter = bufio.NewWriter(conn)
    for {
        var res = <-responses
        err := types.WriteMessage(res, bufWriter)
        if err != nil {
            closeConn <- fmt.Errorf("Error writing message: %v", err.Error())
            return
        }
        if _, ok := res.Value.(*types.Response_Flush); ok {
            err = bufWriter.Flush()
            if err != nil {
                closeConn <- fmt.Errorf("Error flushing write buffer: %v", err.Error())
                return
            }
        }
        count++
    }
}
func (s *SocketServer) handleRequest(req *types.Request, responses chan<- *types.Response) {
    switch r := req.Value.(type) {
    case *types.Request_DeliverTx:
        res := s.app.DeliverTx(r.DeliverTx.Tx)
        responses <- types.ToResponseDeliverTx(res.Code, res.Data, res.Log)
    case *types.Request_CheckTx:
        res := s.app.CheckTx(r.CheckTx.Tx)
        responses <- types.ToResponseCheckTx(res.Code, res.Data, res.Log)
    case *types.Request_Commit:
        res := s.app.Commit()
        responses <- types.ToResponseCommit(res.Code, res.Data, res.Log)
    ...
    }
}

注意这里 handleRequest 的处理方式是写 channel，而不是直接 response，response 是在 handleResponse() 函数中根据 channel 中获得值进行处理的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// Pull responses from 'responses' and write them to conn.
func (s *SocketServer) handleResponses(closeConn chan error, responses <-chan *types.Response, conn net.Conn) {
    var count int
    var bufWriter = bufio.NewWriter(conn)
    for {
        var res = <-responses
        err := types.WriteMessage(res, bufWriter)
        if err != nil {
            closeConn <- fmt.Errorf("Error writing message: %v", err.Error())
            return
        }
        if _, ok := res.Value.(*types.Response_Flush); ok {
            err = bufWriter.Flush()
            if err != nil {
                closeConn <- fmt.Errorf("Error flushing write buffer: %v", err.Error())
                return
            }
        }
        count++
    }
}

####（2）Application

1. App Interface

作为一个 ABCI server，其必须实现 Application 接口，这个接口的声明在 tpes/application.go 中，如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
type Application interface {
    // Info/Query Connection
    Info() ResponseInfo                              // Return application info
    SetOption(key string, value string) (log string) // Set application option
    Query(reqQuery RequestQuery) ResponseQuery       // Query for state

    // Mempool Connection
    CheckTx(tx []byte) Result // Validate a tx for the mempool

    // Consensus Connection
    InitChain(validators []*Validator)       // Initialize blockchain with validators from TendermintCore
    BeginBlock(hash []byte, header *Header)  // Signals the beginning of a block
    DeliverTx(tx []byte) Result              // Deliver a tx for full processing
    EndBlock(height uint64) ResponseEndBlock // Signals the end of a block, returns changes to the validator set
    Commit() Result                          // Commit the state and return the application Merkle root hash
}

ABCI Messages 所使用的 消息类型都是通过 Protobuf 协议定义的，在 types/types.proto 文件中声明（包括 Validator，Request, Response 等消息格式）。

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
enum CodeType {
    OK                    = 0;
    InternalError         = 1;
    EncodingError         = 2;
    ...
}

message Request {
    oneof value{
        RequestEcho echo = 1;
        RequestFlush flush = 2;
        RequestInfo info = 3;
        RequestSetOption set_option = 4;
        RequestDeliverTx deliver_tx = 5;
        RequestCheckTx check_tx = 6;
        RequestCommit commit = 7;
        ...
    }
}

message Response {
    oneof value{
        ResponseException exception = 1;
        ResponseEcho echo = 2;
        ResponseFlush flush = 3;
        ResponseInfo info = 4;
        ResponseSetOption set_option = 5;
        ...
    }
}

message Validator {
    bytes pubKey = 1;
    uint64 power = 2;
}
2. App Implementation

Application 中声明的接口方法，既可以在具体的 application 代码中实现，tendermint 提供了 abci server 的两个简单实现，dummy 和 counter，侦听端口是 46658。
dummy 在 abci/example/dummy/dummy.go 中实现，而 counter 则在 abci/example/counter/counter.go 中实现。

ABCI Message 中最重要的就是 deliver_tx, check_tx, and commit 三类消息。

tendermint engine 的角色是作为 ABCI Client，因此任何一个 Validator 节点（也即 Tendermint Node）都是作为 ABCI Client，启动 Tendermint Node 之后就可以连接到 ABCI Server 了。

Tendermint 还提供了一个简易的 ABCI 客户端，即 abci-cli，通过 go get -u github.com/tendermint/abci/cmd/… 获得。
这样，就可以通过这个 abci-cli 给 ABCI Server 发消息了。
比如：

1
2
$ abci-cli echo hello
$ abci-cli info

abci/abci-cli/abci-cli.go 提供的是 command，而 client 代码的具体实现全部在 abci/client/ 目录下
client/
├── client.go
├── grpc_client.go
├── local_client.go
└── socket_client.go

1
2
3
4
5
if client == nil {
    var err error
    client, err = abcicli.NewClient(c.GlobalString("address"), c.GlobalString("abci"), false)
    ...
}

具体的 client 类型是通过 transport 类型来指定的（默认是 socket）

1
2
3
4
5
6
7
8
9
10
11
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
}

默认的 transport 类型是 socket

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
app.Flags = []cli.Flag{
    cli.StringFlag{
        Name:  "address",
        Value: "tcp://127.0.0.1:46658",
        Usage: "address of application socket",
    },
    cli.StringFlag{
        Name:  "abci",
        Value: "socket",
        Usage: "socket or grpc",
    },
    cli.BoolFlag{
        Name:  "verbose",
        Usage: "print the command and results as if it were a console session",
    },
}
启动 Server
1
2
3
4
srv, _ := server.NewServer("tcp://0.0.0.0:46658", "socket", app)
if _, err := srv.Start(); err != nil {
    os.Exit(1)
}

解释：
这里通过调用 server.NewServer() 来创建一个 Server，三个参数分别是，侦听地址，transport 类型（socket/grpc），app 实现。返回类型是 cmn.Service。 具体的启动逻辑在 srv.Start() 里面。 SocketServer 内置了 BaseService 结构体成员变量，其实现了 Service 声明的方法。

1
2
3
4
5
6
7
8
type Service interface {
    Start() (bool, error)
    OnStart() error
    Stop() bool
    OnStop()
    Reset() (bool, error)
    ...
}

可见，最终是通过调用 OnStart() 来启动的，其过程很简单，简单来说就是调用 net.Listen() 和 go s.acceptConnectionsRoutine() 来启动侦听和接受连接。

1
2
3
4
5
6
7
8
9
10
func (s *SocketServer) OnStart() error {
    s.BaseService.OnStart()
    ln, err := net.Listen(s.proto, s.addr)
    if err != nil {
        return err
    }
    s.listener = ln
    go s.acceptConnectionsRoutine()
    return nil
}

全文完!


如果你对我的文章感兴趣，欢迎留言或者关注我的专栏。

微信公众号：“知辉”

搜索“deliverit”或

扫描二维码