DEV Community: Yi Zhang

How Tendermint Consensus Timeout Ticker works

Yi Zhang — Sat, 13 Jan 2024 03:16:03 +0000

In file "tendermint/consensus/ticker.go"



type TimeoutTicker interface {
        Start() error
        Stop() error
        Chan() <-chan timeoutInfo       // on which to receive a timeout
        ScheduleTimeout(ti timeoutInfo) // reset the timer

        SetLogger(log.Logger)
}

type timeoutTicker struct {
        service.BaseService

        timer    *time.Timer
        tickChan chan timeoutInfo // for scheduling timeouts
        tockChan chan timeoutInfo // for notifying about them
}

func NewTimeoutTicker() TimeoutTicker {
        tt := &timeoutTicker{
                timer:    time.NewTimer(0),
                tickChan: make(chan timeoutInfo, tickTockBufferSize),
                tockChan: make(chan timeoutInfo, tickTockBufferSize),
        }
        tt.BaseService = *service.NewBaseService(nil, "TimeoutTicker", tt)
        tt.stopTimer() // don't want to fire until the first scheduled timeout
        return tt
}

func (t *timeoutTicker) OnStart() error {
        go t.timeoutRoutine()
        return nil
}

func (t *timeoutTicker) OnStop() {
        t.BaseService.OnStop()
        t.stopTimer()
}

func (t *timeoutTicker) Chan() <-chan timeoutInfo {
        return t.tockChan
}

func (t *timeoutTicker) ScheduleTimeout(ti timeoutInfo) {
        t.tickChan <- ti
}

func (t *timeoutTicker) stopTimer() {
        // Stop() returns false if it was already fired or was stopped
        if !t.timer.Stop() {
                select {
                case <-t.timer.C:
                default:
                        t.Logger.Debug("Timer already stopped")
                }
        }
}

func (t *timeoutTicker) timeoutRoutine() {
        t.Logger.Debug("Starting timeout routine")
        var ti timeoutInfo
        for {
                select {
                case newti := <-t.tickChan:
                        t.Logger.Debug("Received tick", "old_ti", ti, "new_ti", newti)

                        // ignore tickers for old height/round/step
                        if newti.Height < ti.Height {
                                continue
                        } else if newti.Height == ti.Height {
                                if newti.Round < ti.Round {
                                        continue
                                } else if newti.Round == ti.Round {
                                        if ti.Step > 0 && newti.Step <= ti.Step {
                                                continue
                                        }
                                }
                        }

                        // stop the last timer
                        t.stopTimer()

                        ti = newti
                        t.timer.Reset(ti.Duration)
                        t.Logger.Debug("Scheduled timeout", "dur", ti.Duration, "height", ti.Height, "round", ti.Round, "step", ti.Step)
                case <-t.timer.C:
                        t.Logger.Info("Timed out", "dur", ti.Duration, "height", ti.Height, "round", ti.Round, "step", ti.Step)
                        go func(toi timeoutInfo) { t.tockChan <- toi }(ti)
                case <-t.Quit():
                        return
                }
        }
}

Conclusion:

timeouTicker is a service which we can start and stop. When started it runs a for loop to continuously and blockingly receive from 2 channels: "t.tickerChan" and "t.timer.C".

Basically the timeoutTiker allows caller to call "ScheduleTimeout(ti timeoutInfo)". The timeoutTicker will setup a timer to fire. When firing, the "timeoutInfo ti" will be available at the tockChan channel which caller can get by calling "timoutTicker.Chan()".

Tendermint P2P Reactor Types

Yi Zhang — Sat, 13 Jan 2024 03:03:36 +0000

By looking at the struct Node in File "node/node.go", we can see there are 5 kinds of reactors. Here I give them a quick summary.

Reactor	Package	Purpose
pex.Reactor	/p2p/pex	for exchanging peer addresses
bcReactor	/blockchain	for fast-syncing
mempl.Reactor	/mempool	for gossipping transactions
cs.Reactor	/consensus	for participating in the consensus
statesync.Reactor	/statesync	for hosting and restoring state sync snapshots

How Tendermint p2p Nodes Receive Messages

Yi Zhang — Sat, 13 Jan 2024 02:04:52 +0000

In file "tendermint/p2p/peer.go", we have following code


// peer implements Peer.
//
// Before using a peer, you will need to perform a handshake on connection.
type peer struct {
        service.BaseService

        // raw peerConn and the multiplex connection
        peerConn
        mconn *tmconn.MConnection

        // peer's node info and the channel it knows about
        // channels = nodeInfo.Channels
        // cached to avoid copying nodeInfo in hasChannel
        nodeInfo NodeInfo
        channels []byte

        // User data
        Data *cmap.CMap

        metrics       *Metrics
        metricsTicker *time.Ticker
}

type PeerOption func(*peer)

func newPeer(
        pc peerConn,
        mConfig tmconn.MConnConfig,
        nodeInfo NodeInfo,
        reactorsByCh map[byte]Reactor,
        chDescs []*tmconn.ChannelDescriptor,
        onPeerError func(Peer, interface{}),
        options ...PeerOption,
) *peer {
        p := &peer{
                peerConn:      pc,
                nodeInfo:      nodeInfo,
                channels:      nodeInfo.(DefaultNodeInfo).Channels,
                Data:          cmap.NewCMap(),
                metricsTicker: time.NewTicker(metricsTickerDuration),
                metrics:       NopMetrics(),
        }

        p.mconn = createMConnection(
                pc.conn,
                p,
                reactorsByCh,
                chDescs,
                onPeerError,
                mConfig,
        )
        p.BaseService = *service.NewBaseService(nil, "Peer", p)
        for _, option := range options {
                option(p)
        }

        return p
}

........

func createMConnection(
        conn net.Conn,
        p *peer,
        reactorsByCh map[byte]Reactor,
        chDescs []*tmconn.ChannelDescriptor,
        onPeerError func(Peer, interface{}),
        config tmconn.MConnConfig,
) *tmconn.MConnection {

        onReceive := func(chID byte, msgBytes []byte) {
                reactor := reactorsByCh[chID]
                if reactor == nil {
                        // Note that its ok to panic here as it's caught in the conn._recover,
                        // which does onPeerError.
                        panic(fmt.Sprintf("Unknown channel %X", chID))
                }
                labels := []string{
                        "peer_id", string(p.ID()),
                        "chID", fmt.Sprintf("%#x", chID),
                }
                p.metrics.PeerReceiveBytesTotal.With(labels...).Add(float64(len(msgBytes)))
                reactor.Receive(chID, p, msgBytes)
        }

        onError := func(r interface{}) {
                onPeerError(p, r)
        }

        return tmconn.NewMConnectionWithConfig(
                conn,
                chDescs,
                onReceive,
                onError,
                config,
        )
}

In file "tendermint/p2p/conn/connection.go", we have code:


type MConnection struct {
        service.BaseService

        conn          net.Conn
        bufConnReader *bufio.Reader
        bufConnWriter *bufio.Writer
        sendMonitor   *flow.Monitor
        recvMonitor   *flow.Monitor
        send          chan struct{}
        pong          chan struct{}
        channels      []*Channel
        channelsIdx   map[byte]*Channel
        onReceive     receiveCbFunc
        onError       errorCbFunc
        errored       uint32
        config        MConnConfig
........
}


// recvRoutine reads PacketMsgs and reconstructs the message using the channels' "recving" buffer.
// After a whole message has been assembled, it's pushed to onReceive().
// Blocks depending on how the connection is throttled.
// Otherwise, it never blocks.
func (c *MConnection) recvRoutine() {
        defer c._recover()

        protoReader := protoio.NewDelimitedReader(c.bufConnReader, c._maxPacketMsgSize)

FOR_LOOP:
        for {
........
                // Read packet type
                var packet tmp2p.Packet

                _n, err := protoReader.ReadMsg(&packet)
........
                // Read more depending on packet type.
                switch pkt := packet.Sum.(type) {
........
                case *tmp2p.Packet_PacketMsg:
                        channelID := byte(pkt.PacketMsg.ChannelID)
                        channel, ok := c.channelsIdx[channelID]
                        if pkt.PacketMsg.ChannelID < 0 || pkt.PacketMsg.ChannelID > math.MaxUint8 || !ok || channel == nil {
                                err := fmt.Errorf("unknown channel %X", pkt.PacketMsg.ChannelID)
                                    c.Logger.Debug("Connection failed @ recvRoutine", "conn", c, "err", err)
                                c.stopForError(err)
                                break FOR_LOOP
                        }

                        msgBytes, err := channel.recvPacketMsg(*pkt.PacketMsg)
                        if err != nil {
                                if c.IsRunning() {
                                        c.Logger.Debug("Connection failed @ recvRoutine", "conn", c, "err", err)
                                        c.stopForError(err)
                                }
                                break FOR_LOOP
                        }
                        if msgBytes != nil {
                                c.Logger.Debug("Received bytes", "chID", channelID, "msgBytes", msgBytes)
                                // NOTE: This means the reactor.Receive runs in the same thread as the p2p recv routine
                                c.onReceive(channelID, msgBytes)
                        }
                default:
........
}

Conclusion:

The MConnection relentlessly receives packets from network. When it receives a packet of type Packet_PacketMsg, it go ahead to receive the whole message with this function:

msgBytes, err := channel.recvPacketMsg(*pkt.PacketMsg)

Then it calls the

c.onReceive(channelID, msgBytes)

The onReceive is a callback defined in struc MConnection. It gets initialzied in "tendermint/p2p/peer.go". The callback first finds out the reactor which the channelID corresponds to. And then call the reactor's "Receive" function with the channelID, peer, and msgBytes as parameters:

reactor.Receive(chID, p, msgBytes)

This is how the reactor's Receive function gets message from peers.

Dig into the codebase of Tendermint and Cosmos SDK

Yi Zhang — Sat, 13 Jan 2024 01:50:24 +0000

Last year I worked on a project which required to convert Tendermint/Cosmos SDK into a network which could run multiple chains concurrently. We all know it's common for one blockchain to run on a set of nodes in one network. For cloud providers, each time when they create a new chain for a user (maybe a corporate or a small startup), they need at least 4 new nodes to form a basic Byzantine fault tolerance network. For some small startups, the cost is a bit significant. If multiple chains can share the same network/nodes before they start to have much traffic, it'll be a big win for both cloud providers and the users.

This project required a deep understanding and familiarity in the codebase of Tendermint and Cosmos SDK. To facilitate the process of learning the codebase, I've made lots of notes. Today I planed to share them in a series of posts here in dev dot to.

Notes:

The code samples of tendermint in the above posts are from version v0.34.19
The code samples of cosmos-sdk in the above posts are from version v0.45.3

How to create a 3-node kubernetes cluster and deploy an application on my ubuntu 22.04 minibox

Yi Zhang — Wed, 10 Jan 2024 12:38:00 +0000

Why did I want to do it?

I'm always interested in trying out some technologies which seem useful and I'm not familiar with yet. Kubernetes is sure one of them.

After reading the doc about it, especially after learning the architecture, I had an urge to try it out on my home lab - an ubuntu 22.04 desktop running on a minibox with AMD Ryzen 7840H, 16GB memory, and 1TB SSD.

How did I do it

I read many articles talking about running minikube, or microk8s before. I tried microk8s and it worked fine. But they all suppose you run them in one box for testing purpose. I'd like to try the production way of k8s. So I decided to run a cluster of 3 nodes, which I guess is the minumum configuration of a production k8s cluster.

Step1: Create 3 nodes

To make 3 nodes out of my home lab, it's pretty easy and quick with help of multipass:

$ multipass launch -c 2 -d 30G -m 3G -n node1
$ multipass launch -c 2 -d 30G -m 3G -n node2
$ multipass launch -c 2 -d 30G -m 3G -n node3

In a kubernetes cluster, there has to be at least one node for the control plane. So I just pick the node1 as the master node, to make a single node control plane. We may have multiple nodes for a control plane, so they can synchnize with each other to gain high availability. But today I just want to quickly make a working k8s custer, so one node is okay.

Step 2: Install kubernetes

To make a k8s cluster, first we need to install k8s onto the nodes. The following is how I installed k8s on node1:

$ multipass shell node1
$ sudo apt-get update
$ sudo apt-get install -y apt-transport-https ca-certificates curl gpg

$ curl -fsSL https://pkgs.k8s.io/core:/stable:/v1.29/deb/Release.key | sudo gpg --dearmor -o /etc/apt/keyrings/kubernetes-apt-keyring.gpg

$ echo 'deb [signed-by=/etc/apt/keyrings/kubernetes-apt-keyring.gpg] https://pkgs.k8s.io/core:/stable:/v1.29/deb/ /' | sudo tee /etc/apt/sources.list.d/kubernetes.list

$ sudo apt-get update
$ sudo apt-get install -y kubelet kubeadm kubectl
$ sudo apt-mark hold kubelet kubeadm kubectl

The second step is the install the container runtime. K8s itself is a set of container orchestration tools, which needs container runtime to handle containers. There are a few of container runtimes to choose from. I tried containerd, and cri-o. But the installation for them was not smooth for me. Since I was pretty familiar with docker engine, so I tried it and the installation was pretty smooth.

Step 3: Install Docker Engine

sudo apt-get update
sudo apt-get install -y tmux
sudo apt install -y docker.io
sudo systemctl start docker && sudo systemctl enable docker
sudo usermod -a -G docker $USER

For k8s to interact with containers, the container runtimes needs to be support CRI (Container Runtime Interface). To make Docker Engine CRI compatile, we need to install a small package, cri-dockerd.

Step 4: Install cri-dockerd

$ wget https://github.com/Mirantis/cri-dockerd/releases/download/v0.3.9/cri-dockerd-0.3.9.amd64.tgz
$ tar zxvf cri-dockerd-0.3.9.amd64.tgz
$ cd cri-dockerd
$ sudo mkdir -p /usr/local/bin
$ sudo install -o root -g root -m 0755 cri-dockerd/usr/local/bin/cri-dockerd
$ mkdir foo; cd foo
$ git clone git@github.com:Mirantis/cri-dockerd.git
$ cd cri-dockerd 
$ sudo install packaging/systemd/* /etc/systemd/system
$ sudo sed -i -e 's,/usr/bin/cri-dockerd,/usr/local/bin/cri-dockerd,' /etc/systemd/system/cri-docker.service
$ sudo systemctl daemon-reload
$ sudo systemctl enable --now cri-docker.socket

Step 5: Initialize the cluster

$ sudo kubeadm init --cri-socket=unix:///var/run/cri-dockerd.sock

The output of this command includes the command to add worker nodes into the cluster. So you may want to make a note.

To make kubectl work for your non-root user, run these commands, which are also part of the kubeadm init output:

$ mkdir -p $HOME/.kube
$ sudo cp -i /etc/kubernetes/admin.conf $HOME/.kube/config
$ sudo chown $(id -u):$(id -g) $HOME/.kube/config

Step 6: Install a Pod network add-on

Note: this step is only required on master nodes.

$ kubectl apply -f https://github.com/flannel-io/flannel/releases/latest/download/kube-flannel.yml

Step 7: Join the worker nodes

On node2 and node3, run the "kubeadm join" command shown in the output of "kubeadm init" command. For example:

$ kubeadm join 10.129.200.54:6443 --token b3orwg.1opdge6qz0pu3tg1 \
                --discovery-token-ca-cert-hash sha256:074cf7ccc39644c2f9b444feae1fe6d1e33bba557e796aa782e2c5d831e25b30

Step 8: Deploy application

To make it simple, we just try to deploy a pod consisting of a single nginx container. Run the following command on master node (node1):

kubectl apply -f https://k8s.io/examples/controllers/nginx-deployment.yaml

Conclusion

This small article quickly demonstrated how to setup environment and deploy application with Kubernetes in a home lab environment. You can do similar things on virutal machines in Cloud too. The official Kubernetes documents have much more content which may seem daunting in the beginning. This article can help you quickly build the concept about how Kubernetes works. The next step is for you to design the Kubernets architecture for your own projects.