本篇文章主要是PBFT共识的简单实现,其中有许多地方都做了简化。PBFT的原理已在上篇文章中描述过,如果对PBFT的原理不太清晰的的可以进行查看。文章地址:共识算法学习总结。
代码实现的主要功能有:通过客户端添加区块,使用libp2p的mdns进行节点发现,动态的添加节点。
客户端
在启动客户端时,首先根据端口号创建一个客户端,然后启动客户端。
var clientCmd = &cobra.Command{Use: "client",Short: "client manage",Run: func(cmd *cobra.Command, args []string) {// 获取客户端的端口port, err := cmd.Flags().GetInt("port")if err != nil {log.Println("get param error: ", err)}// 客户端传递的数据data, err := cmd.Flags().GetString("data")if err != nil{log.Println("get param error: ", err)}client := NewClient(port)client.Start(data)},}
创建客户端
创建的客户端为libp2p节点,并设置节点的私钥。这里的加密算法使用的是Ed25519算法,不但效率更高并且可以在别的节点获取当前节点的公钥。
// 创建客户端func NewClient(listenPort int) *Client {// 生成密钥对r := rand.ReaderprvKey, _, err := crypto.GenerateKeyPairWithReader(crypto.Ed25519, 2048, r)if err != nil{log.Println(err)}pubKey := prvKey.GetPublic()sourceMultiAddr, _ := multiaddr.NewMultiaddr(fmt.Sprintf("/ip4/%s/tcp/%d", "127.0.0.1", listenPort))// 创建libp2p节点h, err := libp2p.New(libp2p.ListenAddrs(sourceMultiAddr),libp2p.Identity(prvKey),)if err != nil { log.Println("创建的客户端节点失败:", err)}h.SetStreamHandler(protocol.ID(protocolID), handleStream)fmt.Printf(">>> 创建客户端p2p节点成功,客户端多路地址是: /ip4/%s/tcp/%v/p2p/%s\n", "0.0.0.0", listenPort, h.ID().Pretty())keyPair := Keypair{privkey: prvKey,pubkey: pubKey,}// 创建客户端client := &Client{h,keyPair,[]KnownNode{},sync.Mutex{},make(map[string]*common.ReplyMsg),}fmt.Println(">>> 创建客户端成功...")return client}
启动客户端
客户端启动时,首先不断获取网络中的节点,然后发送request请求,并等待回应。
这里做了简化,在客户端启动时就直接发送request请求。
func (c *Client) Start (data string){fmt.Println(">>> 开始启动客户端...")ctx := context.Background()// 通过协程获取网络中的节点,使用libp2p的mdns节点发现go c.getAllKonwons(c.client)// 发送客户端请求c.sendRequest(ctx, data)// 处理响应go c.handleConnection()select {}}
客户端发送请求
客户端首先创建request消息,消息格式为。o: 请求的具体操作,t: 请求时客户端追加的时间戳,c:客户端标识。REQUEST: 包含消息内容m,以及消息摘要d(m)。客户端对请求进行签名。
客户端创建完request请求后就可以向主节点发送该请求。
func (c *Client) sendRequest(ctx context.Context, data string){fmt.Println(">>> 客户端准备request消息...")// 构建requestreq := common.Request{data,hex.EncodeToString(common.GenerateDigest(data)),}// 序列化pubKeymarshalPubkey, err := crypto.MarshalPublicKey(c.keypair.pubkey)sendClient := common.SendClient{c.client.ID().Pretty(),marshalPubkey,}// 构建request消息reqMsg := common.RequestMsg{"solve",int(time.Now().Unix()),sendClient,req,}// 对发送的消息进行签名sig, err := c.signMessage(reqMsg)if err != nil{fmt.Printf("%v\n", err)}// 组合并发送消息c.send(ctx, common.ComposeMsg(common.HRequest, reqMsg, sig), c.findPrimaryNode())fmt.Println(">>> 客户端发送消息完成...")}
数据发送
客户端发送数据时,首先连接到主节点,然后打开与主节点的stream。再打开数据发送的通道,最后序列化数据并发数据添加到发送数据的通道中。
func (c *Client) send(ctx context.Context, msg []byte, node KnownNode) {// 开始连接到主节点if err := c.client.Connect(ctx, node.h); err != nil{log.Println(">>> 连接到主节点失败")}// 打开streams, err := c.client.NewStream(context.Background(), node.h.ID, protocol.ID(protocolID))if err != nil {fmt.Println(">>> 打开stream失败", err)}fmt.Println(">>> 开始连接到: ", node.h)rw := bufio.NewReadWriter(bufio.NewReader(s), bufio.NewWriter(s))// 准备发送数据的通道go sendData(rw)// 序列化数据data, err := json.Marshal(msg)if err != nil{fmt.Println("序列化数据错误", err)}sendDataChan <- dataclose(sendDataChan)}
服务端命令行
服务端启动启动时,首先创建一个server,然后启动server。
var serverCmd = &cobra.Command{Use: "server",Short: "server manage",Run: func(cmd *cobra.Command, args []string) {port, err := cmd.Flags().GetInt("port")if err != nil {log.Println("get param error: ", err)}// 创建serverserver := NewServer(port)// 开始serverserver.start()},}
创建新的节点
创建服务端与创建客户端类似。
func NewNode(port int) *Node {// 生成密钥对r := rand.ReaderprvKey, _, err := crypto.GenerateKeyPairWithReader(crypto.Ed25519, 2048, r)if err != nil{log.Println(err)}pubKey := prvKey.GetPublic()sourceMultiAddr, _ := multiaddr.NewMultiaddr(fmt.Sprintf("/ip4/%s/tcp/%d", "127.0.0.1", port))// 创建libp2p节点h, err := libp2p.New(libp2p.ListenAddrs(sourceMultiAddr),libp2p.Identity(prvKey),)if err != nil {log.Println(err)}h.SetStreamHandler(protocol.ID(protocolID), handleStream)fmt.Printf(">>> 创建客户端p2p节点成功,客户端多路地址是: /ip4/%s/tcp/%v/p2p/%s\n", "0.0.0.0", port, h.ID().Pretty())keyPair := Keypair{privkey: prvKey,pubkey: pubKey,}// 创建nodereturn &Node{[]KnownNode{},ClientNode{},0,h,ViewID,make(chan []byte),keyPair,&MsgLog{make(map[string]map[string]bool),make(map[string]map[string]bool),make(map[string]map[string]bool),make(map[string]bool),},make(map[string]*common.RequestMsg),sync.Mutex{},}}
启动节点
服务端的启动仅仅开启一个消息处理协程。 通过消息处理协程,会把服务端接收到的消息分配到对应的逻辑中进行处理。
func (node *Node)Start(){// 处理消息go node.handleMsg()}func (node *Node) handleMsg() {fmt.Println(">>> 启动节点,等待接收消息...")for{//待改进 todorawData := <- receiveChan// 反序列得到的数据var data []byteerr := json.Unmarshal([]byte(rawData), &data)if err != nil {fmt.Println("反序列消息化失败:", err)return}// 分割消息,分别处理不同的消息header, payload, sign:= common.SplitMsg(data)switch header {case common.HRequest:node.handleRequest(payload, sign)case common.HPrePrepare:node.handlePrePrepare(payload, sign)case common.HPrepare:node.handlePrepare(payload, sign)case common.HCommit:node.handleCommit(payload, sign)default:fmt.Println("===============无法处理对应的消息============")}}}
request处理
节点接收到消息后,首先反序列化request消息,然后设置客户端。接着校验request消息的摘要及签名。通过验证后放入请求池,接着创建pre-prepare消息并进行签名。最组合消息进行发送。剩下的三个函数类似,这里就不再叙述。
func (node *Node) handleRequest(payload []byte, sig []byte){fmt.Println(">>> 主节点接收request消息...")var request common.RequestMsgvar prePrepareMsg common.PrePrepareMsg// 反序列化请求消息err := json.Unmarshal(payload, &request)if err != nil{log.Println("反序列化request错误: ", err)return}// 设置节点的客户端clientPubKey, err := crypto.UnmarshalPublicKey(request.Client.PubKey)if err != nil{fmt.Println(">>> 反序列化客户端公钥失败", err)}clientNode := ClientNode{request.Client.ID,clientPubKey,}node.clientNode = clientNode// 校验request的摘要vdig := common.VerifyDigest(request.CRequest.Message, request.CRequest.Digest)if vdig == false {fmt.Printf("验证摘要错误\n")return}// 校验request的签名_, err= common.VerifySignatrue(request, sig, clientPubKey)if err != nil{fmt.Printf("验证签名错误:%v\n", err)return}// 添加进请求池node.mutex.Lock()node.requestPool[request.CRequest.Digest] = &requestseqID := node.getSequenceID()node.mutex.Unlock()// 构建pre-Prepare消息prePrepareMsg = common.PrePrepareMsg{request,request.CRequest.Digest,ViewID,seqID,}// 消息签名msgSig, err:= node.signMessage(prePrepareMsg)if err != nil{fmt.Printf("%v\n", err)return}// 消息组合msg := common.ComposeMsg(common.HPrePrepare, prePrepareMsg, msgSig)// 日志处理node.mutex.Lock()if node.msgLog.preprepareLog[prePrepareMsg.Digest] == nil {node.msgLog.preprepareLog[prePrepareMsg.Digest] = make(map[string]bool)}node.msgLog.preprepareLog[prePrepareMsg.Digest][node.node.ID().String()] = truenode.mutex.Unlock()// 序列化消息data, err := json.Marshal(msg)if err != nil{fmt.Println("序列化request消息出错", err)return}fmt.Println(">>> 主节点广播prePrepare消息...")// 广播消息node.broadcast(data)}
pre-prepare消息处理
func (node *Node) handlePrePrepare(payload []byte, sig []byte) {fmt.Println(">>> 副节点开始接收prePrepare消息...")// 反序列化prePrepare消息var prePrepareMsg common.PrePrepareMsgerr := json.Unmarshal(payload,&prePrepareMsg)if err != nil {fmt.Printf("error happened:%v", err)return}// 找到主节点的公钥pnodeId := node.findPrimaryNode()pubKey, err := pnodeId.h.ID.ExtractPublicKey()if err != nil {fmt.Println("获取主节点的公钥失败", err)return}// 校验消息签名_, err = common.VerifySignatrue(prePrepareMsg, sig, pubKey)if err != nil{fmt.Printf("验证主节点签名错误:%v\n", err)return}// 校验消息的摘要if prePrepareMsg.Digest != prePrepareMsg.Request.CRequest.Digest {fmt.Printf("校验摘要错误\n")return}node.mutex.Lock()node.requestPool[prePrepareMsg.Request.CRequest.Digest] = &prePrepareMsg.Requestnode.mutex.Unlock()// 校验request的摘要err = node.verifyRequestDigest(prePrepareMsg.Digest)if err != nil{fmt.Printf("%v\n", err)return}node.mutex.Lock()node.requestPool[prePrepareMsg.Request.CRequest.Digest] = &prePrepareMsg.Requestnode.mutex.Unlock()err = node.verifyRequestDigest(prePrepareMsg.Digest)if err != nil{fmt.Printf("%v\n", err)return}node.mutex.Lock()if node.msgLog.preprepareLog[prePrepareMsg.Digest] == nil {node.msgLog.preprepareLog[prePrepareMsg.Digest] = make(map[string]bool)}node.msgLog.preprepareLog[prePrepareMsg.Digest][node.node.ID().String()] = truenode.mutex.Unlock()// 构建prePare消息prepareMsg := common.PrepareMsg{prePrepareMsg.Digest,ViewID,prePrepareMsg.SequenceID,node.node.ID(),}// 签名msgSig, err := common.SignMessage(prepareMsg, node.keypair.privkey)if err != nil{fmt.Printf("%v\n", err)return}// 消息组合sendMsg := common.ComposeMsg(common.HPrepare,prepareMsg,msgSig)// 序列化消息data, err := json.Marshal(sendMsg)if err != nil{fmt.Println("序列化prepare消息出错", err)return}fmt.Println(">>> 副节点广播prepare消息...")node.broadcast(data)}
prepare消息处理
func (node *Node) handlePrepare(payload []byte, sig []byte) {fmt.Println(">>> 副节点开始接收prepare消息...")// 反序列化prepare消息var prepareMsg common.PrepareMsgerr := json.Unmarshal(payload,&prepareMsg)if err != nil {fmt.Printf("error happened:%v", err)return}// 得到节点的公钥pnodeID := prepareMsg.NodeIDpubKey, err:= findNodePubkey(pnodeID)if err != nil {fmt.Println("获取主节点的公钥失败", err)return}_, err = common.VerifySignatrue(prepareMsg, sig, pubKey)if err != nil{fmt.Printf("校验签名prepare消息错误:%v\n", err)return}err = node.verifyRequestDigest(prepareMsg.Digest)if err != nil{fmt.Printf("%v\n", err)return}// 日记记录node.mutex.Lock()if node.msgLog.prepareLog[prepareMsg.Digest] == nil {node.msgLog.prepareLog[prepareMsg.Digest] = make(map[string]bool)}node.msgLog.prepareLog[prepareMsg.Digest][prepareMsg.NodeID.String()] = truenode.mutex.Unlock()// if receive prepare msg >= 2f +1, then broadcast commit msglimit := node.countNeedReceiveMsgAmount()sum, err:= node.findVerifiedPrepareMsgCount(prepareMsg.Digest)if err != nil {fmt.Printf("error happened:%v", err)return}if sum >= limit {//send commit msgcommitMsg := common.CommitMsg{prepareMsg.Digest,prepareMsg.ViewID,prepareMsg.SequenceID,node.node.ID(),}sig, err := node.signMessage(commitMsg)if err != nil{fmt.Printf("sign message happened error:%v\n", err)}sendMsg := common.ComposeMsg(common.HCommit,commitMsg,sig)data, err := json.Marshal(sendMsg)if err != nil{fmt.Println("序列化commit消息出错", err)}node.broadcast(data)fmt.Println(">>> 副节点广播commit消息成功")}}
commit 消息处理
func (node *Node) handleCommit(payload []byte, sig []byte) {fmt.Println(">>> 副节点开始接收commit消息")// 反序列化消息var commitMsg common.CommitMsgerr := json.Unmarshal(payload,&commitMsg)if err != nil {fmt.Printf("error happened:%v", err)}msgPubKey, err := findNodePubkey(commitMsg.NodeID)if err != nil{fmt.Println(err)return}verify, err := common.VerifySignatrue(commitMsg, sig, msgPubKey)if err != nil{fmt.Printf("verify signature failed:%v\n", err)return}if verify == false {fmt.Printf("verify signature failed\n")return}err = node.verifyRequestDigest(commitMsg.Digest)if err != nil{fmt.Printf("%v\n", err)return}node.mutex.Lock()if node.msgLog.commitLog[commitMsg.Digest] == nil {node.msgLog.commitLog[commitMsg.Digest] = make(map[string]bool)}node.msgLog.commitLog[commitMsg.Digest][commitMsg.NodeID.String()] = truenode.mutex.Unlock()// if receive commit msg >= 2f +1, then send reply msg to clientlimit := node.countNeedReceiveMsgAmount()sum, err := node.findVerifiedCommitMsgCount(commitMsg.Digest)if err != nil{fmt.Printf("error happened:%v", err)return}if sum >= limit {// if already send reply msg, then do nothingnode.mutex.Lock()exist := node.msgLog.replyLog[commitMsg.Digest]node.mutex.Unlock()if exist == true {return}// send reply msgnode.mutex.Lock()requestMsg := node.requestPool[commitMsg.Digest]node.mutex.Unlock()fmt.Printf("operstion:%smessage:%s executed... \n",requestMsg.Operation, requestMsg.CRequest.Message)done := fmt.Sprintf("operstion:%smessage:%s done ",requestMsg.Operation, requestMsg.CRequest.Message)replyMsg := common.ReplyMsg{node.View,int(time.Now().Unix()),requestMsg.Client.ID,node.node.ID().String(),done,}hostAddr, _ := ma.NewMultiaddr(fmt.Sprintf("/ipfs/%s", requestMsg.Client.ID))clientNode, err := peer.AddrInfoFromString(hostAddr.String())fmt.Println(">>> 客户端地址:", clientNode.ID.Pretty())if err != nil{fmt.Println(err)return}fmt.Println(">>> 开始向客户端回复数据...")sendMsg := common.ComposeMsg(common.HReply,replyMsg,[]byte{})data, err := json.Marshal(sendMsg)if err != nil{fmt.Println("序列化commit消息出错", err)}node.reply(context.Background(), data, *clientNode)node.mutex.Lock()node.msgLog.replyLog[commitMsg.Digest] = truenode.mutex.Unlock()fmt.Println(">>> 回复客户端成功...")}}
运行结果
最后
项目中做了很多简化并且有很多设计不合理的地方,以后会继续进行改进。源码:https://github.com/blockchainGuide/Consensus_Algorithm