NEO源码分析之网络通信

Posted 2021-04-05 NEL新经济实验室

0x00 前言

NEO被称为中国版的Ethereum，支持C#和java开发，并且在社区的努力下已经把SDK拓展到了js，python等编程环境，所以进行NEO开发的话是没有太大语言障碍的。 比特币在解决拜占庭错误这个问题时除了引入了区块链这个重要的概念之外，还引入了工作量证明（PoW）这个机智的解决方案，通过数学意义上的难题来保证每个区块创建都需要付出计算量。然而实践已经证明，通过计算来提供工作量证明，实在是太浪费：全世界所有的完全节点都进行同样的计算，然而只有一个节点计算出的结果会被添加到区块链中，其余节点计算消耗的电力就都白白浪费了。尤其，工作量证明存在一个51%的可能攻击方案，就是说只要有人掌握了世界上超过50%的算力，那么他就可以对比特币这个系统进行攻击，重置区块链。中本聪先生发明这个算力工作量证明方法的时候大概没有料到会有人专门为了挖矿开发出ASIC矿机。 NEO在解决这些问题的时候提出了一个新的共享机制DBFT 全称为 Delegated Byzantine Fault Tolerant。NEO将节点分为两种，一种为普通节点，不参与共识，也就是不进行认证交易签名区块的过程。另一种是则是共识节点。顾名思义，就是可以参与共识的节点，这部分基础概念可以参考官方文档。 接下来我将会以一系列的博客来从源码层面上对NEO进行分析。 而本文主要进行的是源码层级的NEO网络通信协议分析。

0x01 源码概览

本文分析的源码位于这里，通过git命令下载到本地：

    git clone https://github.com/neo-project/neo.git

我是用的编译器是VS2017社区版。打开neo项目之后可以看到项目根目录文件结构：

• Consensus 共识节点间共识协议

• Core neo核心

• Cryptography 加密方法

• Implementations 数据存储以及钱包的实现

• IO NEO的io类

• Network 用于p2p网络通信的方法

• SmartContract NEO智能合约的相关类

整个项目代码量不算很大，尤其是项目本身是C#高级语言编写，所以代码很容易读懂。

0x02 消息

在NEO网络中，所有的消息都以Message为单位进行传输，Message的定义在Message.cs文件中，其结构如下：

• Magic 字段用来确定当前节点是运行在正式网络还是在测试网络，如果是0x00746e41则为正式网，如果是0x74746e41则为测试网。

• _Command_命令的内容是直接使用的字符串，所以没有进行严格定义，在所有使用到的地方都是直接使用的字符串。这里给我的感觉是依赖特别严重，应该先定义好命令再在别的地方调用。虽然没有明说都有哪些命令，但是对消息路由的代码里我们可以找到所有使用到的命令：

源码位置：neo/Network/RemoteNode.cs/OnMessageReceived

            switch (message.Command)
            {                case "addr": 
                case "block": 
                case "consensus":                case "filteradd":                case "filterclear":                case "filterload":                case "getaddr":                case "getblocks":                case "getdata":                case "getheaders":                case "headers":                case "inv":                case "mempool":                case "tx":                case "verack":                case "version":                case "alert":                case "merkleblock":                case "notfound":                case "ping":                case "pong":                case "reject":
            }

以上源码中的对命令的处理部分我都删掉了，这个不是本小节讨论重点。通过分析代码可以知道，消息种类大致22种。 消息的具体内容在序列化之后存在在Message里的payload字段中。

在所有的消息类型中有一类消息非常特殊，这就是与账本相关的三种消息：账目消息（Block），共识消息（Consensus）以及交易消息（Transaction）。这三中消息分别对应系统中的三个类：

• neo/Core/Block

• neo/Core/Transaction

• neo/Network.Payloads/ConsensusPayload

这三个类都实现了接口IInventory，我把inventory翻译为账本，把实现了IInventory接口的类成为账本类，消息称为账本消息。IInventory接口定义了消息的哈希值Hash用来存放签名、账本消息类型InventoryType用来保存消息类型以及一个验证函数verify用来对消息进行验证，也就是说所有的账本消息都需要包含签名，并且需要验证。 账本消息的类型定义在InventoryType.cs文件中：

源码位置：neo/Network/InventoryType.cs

        /// 交易
        TX = 0x01,        /// 区块
        Block = 0x02,        /// 共识数据
        Consensus = 0xe0

对共识部分的消息感兴趣的可以查看我的另一篇博客NEO从源码分析看共识协议，本文仅仅关注于交易通信和普通节点的区块同步。

每个RemoteNode内部都有两个消息队列，一个高优先级队列和一个低优先级队列，高优先级队列主要负责：

• "alert"

• "consensus"

• "filteradd"

• "filterclear"

• "filterload"

• "getaddr"

• "mempool"

这几个命令，其余的命令都由低优先级队列负责。 发送命令的任务由StartSendLoop方法负责，在这个方法中有一个while循环，在每一轮循环中都会首先检测高优先级队列是否为空，如果不为空则先发送高优先命令，否则发送低优先级任务，循环中的核心源码如下：

源码位置：neo/Netwotk/RemoteNode.cs/StartSendLoop

                Message message = null;
                lock (message_queue_high)
                {                    //高优先级消息队列不为空
                    if (message_queue_high.Count > 0)
                    {
                        message = message_queue_high.Dequeue();
                    }
                }                //若没有高优先级任务
                if (message == null)
                {
                    lock (message_queue_low)
                    {                        if (message_queue_low.Count > 0)
                        {                            //获取低优先级任务
                            message = message_queue_low.Dequeue();
                        }
                    }
                }

由于每个RemoteNode对象都只负责和一个相对应的远程节点通信，所以接收消息的地方没有设置消息缓存队列。接收消息的循环就在调用StartSendLoop位置的下面，由于StartSendLoop本身是个异步方法，所以不会阻塞代码的接收消息循环的执行，在每次收到消息后，都会触发OnMessageReceived方法，并将收到的message消息作为参数传递过去。在上文中也讲了，这个OnMessageReceived方法其实是个消息的路由器来着，会根据消息类型的不同调用响应的处理函数。

0x03 新节点组网

源码位置：neo/Network/LocalNode.cs/Start

Task.Run(async () =>
                {                    if ((port > 0 || ws_port > 0)
                        && UpnpEnabled
                        && LocalAddresses.All(p => !p.IsIPv4MappedToIPv6 || IsIntranetAddress(p))
                        && await UPnP.DiscoverAsync())
                    {                        try
                        {
                            LocalAddresses.Add(await UPnP.GetExternalIPAsync());   //添加获取到的网络中节点信息
                            if (port > 0)                                await UPnP.ForwardPortAsync(port, ProtocolType.Tcp, "NEO");  //向服务器注册本地节点
                            if (ws_port > 0)                                await UPnP.ForwardPortAsync(ws_port, ProtocolType.Tcp, "NEO WebSocket");
                        }                        catch { }
                    }
                    connectThread.Start();  //开启线程与网络中节点建立连接
                    poolThread?.Start();                    if (port > 0)
                    {
                        listener = new TcpListener(IPAddress.Any, port); //开启服务，监听网络中的广播信息
                        listener.Server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, 1);                        try
                        {
                            listener.Start();  //开启端口，监听连接请求
                            Port = (ushort)port;
                            AcceptPeers();  //处理p2p网络中的socket连接请求
                        }                        catch (SocketException) { }
                    }                    if (ws_port > 0)
                    {
                        ws_host = new WebHostBuilder().UseKestrel().UseUrls($"http://*:{ws_port}").Configure(app => app.UseWebSockets().Run(ProcessWebSocketAsync)).Build();
                        ws_host.Start();
                    }
                });

源码位置：neo/Network/LocalNode.cs/ConnectToPeerAsync

            //新建远程节点对象
            TcpRemoteNode remoteNode = new TcpRemoteNode(this, remoteEndpoint);            if (await remoteNode.ConnectAsync())
            {
                OnConnected(remoteNode);
            }

TcpRemoteNode类继承自RemoteNode，每个对象都代表着一个与自己建立连接的远程节点，RemoteNode和LocalNode的关系大致可以这样表示： 

TcpRemoteNode的构造函数在接收到远程节点信息之后会与远程节点建立socket连接并返回一个RemoteNode对象，所有的远程节点对象都被保存在LocalNode中的远程节点列表里。

获取网络节点的方式除了从NEO服务器获取之外还有一个主动获取的方式，那就是向所有的与本地节点建立连接的节点广播网络节点请求，通过获取这些与远程节点建立连接的节点列表来实时获取整个网络中的节点信息。这部分代码在与远程节点建立连接的线程中：

源码位置：neo/Network/LocalNode.cs/ConnectToPeersLoop

                        lock (connectedPeers)
                        {                            foreach (RemoteNode node in connectedPeers)
                                node.RequestPeers();
                        }

向远程节点请求节点列表的RequestPeers方法在RemoteNode类中，这个方法通过向远程节点发送指令“getaddr”来获取。 由于RemoteNode的责任是与其对应的远程节点进行通信，所以对“getaddr”这个远程命令的解析和路由也是在RemoteNode类中进行。在RemoteNode接收到远程节点信息后会触发OnMessageReceived方法对收到的信息进行解析和路由：

源码位置：neo/Network/RemoteNode.cs

        /// <summary>
        /// 对接收信息进行路由
        /// </summary>
        /// <param name="message"></param>
        private void OnMessageReceived(Message message)        {            switch (message.Command)
            {                case "getaddr":
                    OnGetAddrMessageReceived();                    break;                    //代码省略
            }
        }

switch中对于别的命令的解析我都删掉了，这里只关注“getaddr”命令。在收到“getaddr”命令后，会调用相应的处理函数OnGetAddrMessageReceived：

源码位置:neo/Network/RemoteNode.cs/OnGetAddrMessageReceived

            AddrPayload payload;
            lock (localNode.connectedPeers)
            {                const int MaxCountToSend = 200;                //  获取本地连接节点
                IEnumerable<RemoteNode> peers = localNode.connectedPeers.Where(p => p.ListenerEndpoint != null && p.Version != null);                if (localNode.connectedPeers.Count > MaxCountToSend)
                {
                    Random rand = new Random();
                    peers = peers.OrderBy(p => rand.Next());
                }
                peers = peers.Take(MaxCountToSend);
                payload = AddrPayload.Create(peers.Select(p => NetworkAddressWithTime.Create(p.ListenerEndpoint, p.Version.Services, p.Version.Timestamp)).ToArray());
            }
            EnqueueMessage("addr", payload);

由于直接与远程节点进行通信的是与其对应的本地的RemoteNode对象，而这些对象有需要获取LocalNode中保存的信息，NEO源码的处理方式是直接在创建RemoteNode对象的时候传入LocalNode的引用，这里我感觉很不舒服，因为明显有循环引用，尽管在这里功能上不会有什么问题。 因为每个节点既做为客户端，又作为服务端，与本节点建立的网络连接里，即存在自己主动发起的socket连接，也存在远程节点将本节点作为服务端而建立的socket连接。 监听socket连接的任务在线程中不断的执行，每当接收到一个新的socket连接，当前节点会根据这个socket来创建一个新的TcpRemoteNode对象并保存在LocalNode的远程节点列表中：

源码位置：neo/Network/LocalNode.cs/AcceptPeers

 TcpRemoteNode remoteNode = new TcpRemoteNode(this, socket);
 OnConnected(remoteNode);

最后以三个节点的网络拓扑为例： 

0x04 区块同步

新区快的生成与同步主要依靠共识完成后的广播，但是对于新组网的节点应该如何获取完整的区块链呢？本小节将针对这个问题进行源码的分析。

当一个新的RemoteNode对象创建之后，会开启这个对象的protocal： 源码位置：neo/Network/LocalNode.cs

 private void OnConnected(RemoteNode remoteNode)        {            lock (connectedPeers)
            {
                connectedPeers.Add(remoteNode);
            }
            remoteNode.Disconnected += RemoteNode_Disconnected;//断开连接通知
            remoteNode.InventoryReceived += RemoteNode_InventoryReceived;//账单消息通知
            remoteNode.PeersReceived += RemoteNode_PeersReceived;//节点列表信息通知
            remoteNode.StartProtocol();//开启通信协议
        }

在协议开始执行后，会向远程节点发送一个 "version" 命令。在查询这个 "version" 命令的响应方法的时候简直把我吓了一大跳，居然调用的是Disconnect而且传的参数是true。本着“新连接建立之后的第一件事肯定不会是断开连接”这个唯物主义价值观，我又对代码进行了一番研究，终于发现这个发送 “version” 的命令是直接由ReceiveMessageAsync方法获取的，也就是不经过那个消息路由。由于在两个节点建立连接后。两者做的第一件事都是发送 “version” 命令和自己的VersionPayload过去，所以在这个socket连接中节点接收到的第一条消息也都是“version”类型的消息。

源码位置：neo/Network/RemoteNode.cs/StartProtocol

if (!await SendMessageAsync(Message.Create("version", VersionPayload.Create(localNode.Port, localNode.Nonce, localNode.UserAgent))))
                return;Message message = await ReceiveMessageAsync(HalfMinute);

这里需要对这个VersionPayload进行下讲解，这个VersionPayload里包含当前节点的状态信息： 

也就是说在连接建立后，当前节点就可以知道远程节点当前的区块链高度，如果自己当前的区块链高度低于远程节点，就会向远程节点发送 "getblocks" 命令请求区块链同步： 源码位置：neo/Network/RemoteNode.cs/StartProtocol

if (missions.Count == 0 && Blockchain.Default.Height < Version.StartHeight)
{
        EnqueueMessage("getblocks", GetBlocksPayload.Create(Blockchain.Default.CurrentBlockHash));
}

因为区块链有非常大的数据量，区块链同步不可能直接一次完成，每次收到 “getblocks”的命令之后，每次发送500个区块的哈希值：

源码位置：neo/Network/RemoteNode.cs/OnGetBlocksMessageReceived

            List<UInt256> hashes = new List<UInt256>();            do
            {                hash = Blockchain.Default.GetNextBlockHash(hash);                if (hash == null) break;
                hashes.Add(hash);
            } while (hash != payload.HashStop && hashes.Count < 500);
            EnqueueMessage("inv", InvPayload.Create(InventoryType.Block, hashes.ToArray()));

之后在每次接收到远程节点的消息之后，如果当前节点区块高度依然小于远程节点，本地节点会继续发送区块链同步请求，直到与远程节点的区块链同步。

本文由NEL内容奖励计划支持

往期链接

---