好了，前面做了那么多的铺垫，到了这里，关键部分终于闪亮登场了！开始分析Bitcoin Core初始化中的启动节点过程吧，对应的源码是src/init.cpp的AppInitMain()方法的Step 11: start node这部分，详见：https://github.com/bitcoin/bitcoin/blob/v0.16.1/src/init.cpp
分析源码前先参考《精通比特币》梳理下比特币节点的网络发现过程：
(1). 比特币新节点启动后与对等节点建立一条TCP连接
(2). TCP连接建立后，新节点会发送一个包含基本认证信息的version消息给对等节点，开始“握手”通信过程，然后等待对等节点对version消息进行检验，检验完成后会发送verack应答消息。同时对等节点也会发送version消息给新节点，新节点也要进行检验和发送verack应答消息。
(3). 若上述“握手”通信过程成功完成说明此时节点之间的连接成功建立了，连接建立成功后，新节点将一条包含自身IP地址的addr消息发送给其对等节点，对等节点再将此条addr消息依次转发给它们各自的对等节点，从而保证新节点被广为所知并保证连接更稳定。另外，新接入的节点可以向它的对等节点发送getaddr消息，要求它们返回其已知对等节点的IP地址列表。通过这种方式，节点可以找到能连接到的对等节点，并向网络宣告它的存在，以便其他节点能够找到它。
下面开始分析源码：
寻找本地IP地址并保存
代码读到了Discover(）函数这里，该函数主要功能是寻找所有的本地IP地址。
Discover(threadGroup);分析下该函数源码：
void Discover(boost::thread_group& threadGroup)
{
if (!fDiscover)
return;
#ifdef WIN32
// Get local host IP
char pszHostName[256] = "";
if (gethostname(pszHostName, sizeof(pszHostName)) != SOCKET_ERROR)
{
std::vector<CNetAddr> vaddr;
if (LookupHost(pszHostName, vaddr, 0, true))
{
for (const CNetAddr &addr : vaddr)
{
if (AddLocal(addr, LOCAL_IF))
LogPrintf("%s: %s - %s\n", __func__, pszHostName, addr.ToString());
}
}
}
#else
// Get local host ip
struct ifaddrs* myaddrs;
if (getifaddrs(&myaddrs) == 0)
{
for (struct ifaddrs* ifa = myaddrs; ifa != nullptr; ifa = ifa->ifa_next)
{
if (ifa->ifa_addr == nullptr) continue;
if ((ifa->ifa_flags & IFF_UP) == 0) continue;
if (strcmp(ifa->ifa_name, "lo") == 0) continue;
if (strcmp(ifa->ifa_name, "lo0") == 0) continue;
if (ifa->ifa_addr->sa_family == AF_INET)
{
struct sockaddr_in* s4 = (struct sockaddr_in*)(ifa->ifa_addr);
CNetAddr addr(s4->sin_addr);
if (AddLocal(addr, LOCAL_IF))
LogPrintf("%s: IPv4 %s: %s\n", __func__, ifa->ifa_name, addr.ToString());
}
else if (ifa->ifa_addr->sa_family == AF_INET6)
{
struct sockaddr_in6* s6 = (struct sockaddr_in6*)(ifa->ifa_addr);
CNetAddr addr(s6->sin6_addr);
if (AddLocal(addr, LOCAL_IF))
LogPrintf("%s: IPv6 %s: %s\n", __func__, ifa->ifa_name, addr.ToString());
}
}
freeifaddrs(myaddrs);
}
#endif
}可以看到windows系统和Linux系统的发现本地IP地址的方法是不同的，windows系统是先调用gethostname()系统函数获取本机的主机名，再调用LookupHost()函数根据主机名获取IP地址，最后调用AddLocal()函数将本机IP地址存储到全局变量mapLocalHost中，mapLocalHos定义在net.h中。 而Linux系统是先调用getifaddrs()系统函数获取本机的ip地址，然后调用AddLocal()函数将本机IP地址存储到全局变量mapLocalHost中。
upnp端口映射
接下去看代码：
// Map ports with UPnP
MapPort(gArgs.GetBoolArg("-upnp", DEFAULT_UPNP));MapPort()这个方法啥也没干….估计是预留着以后会加一些逻辑功能。
网络连接参数Options对象初始化
接着初始化一个Options对象，该对象封装了网络连接所需的各种参数，看下源码：
CConnman::Options connOptions;
connOptions.nLocalServices = nLocalServices;
connOptions.nMaxConnections = nMaxConnections;
connOptions.nMaxOutbound = std::min(MAX_OUTBOUND_CONNECTIONS, connOptions.nMaxConnections);
connOptions.nMaxAddnode = MAX_ADDNODE_CONNECTIONS;
connOptions.nMaxFeeler = 1;
connOptions.nBestHeight = chain_active_height;
connOptions.uiInterface = &uiInterface;
connOptions.m_msgproc = peerLogic.get();
connOptions.nSendBufferMaxSize = 1000*gArgs.GetArg("-maxsendbuffer", DEFAULT_MAXSENDBUFFER);
connOptions.nReceiveFloodSize = 1000*gArgs.GetArg("-maxreceivebuffer", DEFAULT_MAXRECEIVEBUFFER);
connOptions.m_added_nodes = gArgs.GetArgs("-addnode");
connOptions.nMaxOutboundTimeframe = nMaxOutboundTimeframe;
connOptions.nMaxOutboundLimit = nMaxOutboundLimit;
for (const std::string& strBind : gArgs.GetArgs("-bind")) {
CService addrBind;
if (!Lookup(strBind.c_str(), addrBind, GetListenPort(), false)) {
return InitError(ResolveErrMsg("bind", strBind));
}
connOptions.vBinds.push_back(addrBind);
}
for (const std::string& strBind : gArgs.GetArgs("-whitebind")) {
CService addrBind;
if (!Lookup(strBind.c_str(), addrBind, 0, false)) {
return InitError(ResolveErrMsg("whitebind", strBind));
}
if (addrBind.GetPort() == 0) {
return InitError(strprintf(_("Need to specify a port with -whitebind: '%s'"), strBind));
}
connOptions.vWhiteBinds.push_back(addrBind);
}
for (const auto& net : gArgs.GetArgs("-whitelist")) {
CSubNet subnet;
LookupSubNet(net.c_str(), subnet);
if (!subnet.IsValid())
return InitError(strprintf(_("Invalid netmask specified in -whitelist: '%s'"), net));
connOptions.vWhitelistedRange.push_back(subnet);
}
connOptions.vSeedNodes = gArgs.GetArgs("-seednode");
// Initiate outbound connections unless connect=0
connOptions.m_use_addrman_outgoing = !gArgs.IsArgSet("-connect");
if (!connOptions.m_use_addrman_outgoing) {
const auto connect = gArgs.GetArgs("-connect");
if (connect.size() != 1 || connect[0] != "0") {
connOptions.m_specified_outgoing = connect;
}
}可以看到Options对象初始化了很多参数，包括最大连接数，最大能向外连接的节点数等等，通过命令行指定的”-addnode”和”-seednode”选项的参数也会保存在相应的参数中。
启动节点
好了，到了这里就是执行真正的启动节点操作了
if (!connman.Start(scheduler, connOptions)) {
return false;
}接下来就详细分析下Start()方法源码
Init(connOptions);该方法就是讲Options对象中封装的参数保存到CConnman的成员变量中。
然后初始化一些其他成员变量：
{
LOCK(cs_totalBytesRecv);
nTotalBytesRecv = 0;
}
{
LOCK(cs_totalBytesSent);
nTotalBytesSent = 0;
nMaxOutboundTotalBytesSentInCycle = 0;
nMaxOutboundCycleStartTime = 0;
}
if (fListen && !InitBinds(connOptions.vBinds, connOptions.vWhiteBinds)) {
if (clientInterface) {
clientInterface->ThreadSafeMessageBox(
_("Failed to listen on any port. Use -listen=0 if you want this."),
"", CClientUIInterface::MSG_ERROR);
}
return false;
}
for (const auto& strDest : connOptions.vSeedNodes) {
AddOneShot(strDest);
}可以看到，通过AddOneShot()方法循环遍历了vSeedNodes(即”-seednode”选项指定的参数)种子节点数据，备份了一份在vOneShots成员变量中。
接下去开始加载peers.dat中保存着的之前连接过的P2P对等节点信息：
if (clientInterface) {
clientInterface->InitMessage(_("Loading P2P addresses..."));
}
// Load addresses from peers.dat
int64_t nStart = GetTimeMillis();
{
CAddrDB adb;
if (adb.Read(addrman))
LogPrintf("Loaded %i addresses from peers.dat %dms\n", addrman.size(), GetTimeMillis() - nStart);
else {
addrman.Clear(); // Addrman can be in an inconsistent state after failure, reset it
LogPrintf("Invalid or missing peers.dat; recreating\n");
DumpAddresses();
}
}然后加载banlist.dat中保存着的被禁用的节点ip：
if (clientInterface)
clientInterface->InitMessage(_("Loading banlist..."));
// Load addresses from banlist.dat
nStart = GetTimeMillis();
CBanDB bandb;
banmap_t banmap;
if (bandb.Read(banmap)) {
SetBanned(banmap); // thread save setter
SetBannedSetDirty(false); // no need to write down, just read data
SweepBanned(); // sweep out unused entries
LogPrint(BCLog::NET, "Loaded %d banned node ips/subnets from banlist.dat %dms\n",
banmap.size(), GetTimeMillis() - nStart);
} else {
LogPrintf("Invalid or missing banlist.dat; recreating\n");
SetBannedSetDirty(true); // force write
DumpBanlist();
}然后，启动5个网络工作相关的线程：
uiInterface.InitMessage(_("Starting network threads..."));
fAddressesInitialized = true;
if (semOutbound == nullptr) {
// initialize semaphore
semOutbound = MakeUnique<CSemaphore>(std::min((nMaxOutbound + nMaxFeeler), nMaxConnections));
}
if (semAddnode == nullptr) {
// initialize semaphore
semAddnode = MakeUnique<CSemaphore>(nMaxAddnode);
}
//
// Start threads
//
assert(m_msgproc);
InterruptSocks5(false);
interruptNet.reset();
flagInterruptMsgProc = false;
{
std::unique_lock<std::mutex> lock(mutexMsgProc);
fMsgProcWake = false;
}
// Send and receive from sockets, accept connections
threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));
if (!gArgs.GetBoolArg("-dnsseed", true))
LogPrintf("DNS seeding disabled\n");
else
threadDNSAddressSeed = std::thread(&TraceThread<std::function<void()> >, "dnsseed", std::function<void()>(std::bind(&CConnman::ThreadDNSAddressSeed, this)));
// Initiate outbound connections from -addnode
threadOpenAddedConnections = std::thread(&TraceThread<std::function<void()> >, "addcon", std::function<void()>(std::bind(&CConnman::ThreadOpenAddedConnections, this)));
if (connOptions.m_use_addrman_outgoing && !connOptions.m_specified_outgoing.empty()) {
if (clientInterface) {
clientInterface->ThreadSafeMessageBox(
_("Cannot provide specific connections and have addrman find outgoing connections at the same."),
"", CClientUIInterface::MSG_ERROR);
}
return false;
}
if (connOptions.m_use_addrman_outgoing || !connOptions.m_specified_outgoing.empty())
threadOpenConnections = std::thread(&TraceThread<std::function<void()> >, "opencon", std::function<void()>(std::bind(&CConnman::ThreadOpenConnections, this, connOptions.m_specified_outgoing)));
// Process messages
threadMessageHandler = std::thread(&TraceThread<std::function<void()> >, "msghand", std::function<void()>(std::bind(&CConnman::ThreadMessageHandler, this)));
// Dump network addresses
scheduler.scheduleEvery(std::bind(&CConnman::DumpData, this), DUMP_ADDRESSES_INTERVAL * 1000);这部分源码中我们关键就是来看这5个线程：
(1) “net”线程，该线程的任务是通过socket收发消息，并接受其它节点发起的连接。
// Send and receive from sockets, accept connections
threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));该线程的函数是 void CConnman::ThreadSocketHandler()，来看下该方法：
void CConnman::ThreadSocketHandler()
{
unsigned int nPrevNodeCount = 0;
while (!interruptNet)
{首先是一个while循环，说明循环体里代码一直不断在执行，看下主要代码：
{
LOCK(cs_vNodes);
for (CNode* pnode : vNodes)
{
// Implement the following logic:
// * If there is data to send, select() for sending data. As this only
// happens when optimistic write failed, we choose to first drain the
// write buffer in this case before receiving more. This avoids
// needlessly queueing received data, if the remote peer is not themselves
// receiving data. This means properly utilizing TCP flow control signalling.
// * Otherwise, if there is space left in the receive buffer, select() for
// receiving data.
// * Hand off all complete messages to the processor, to be handled without
// blocking here.
bool select_recv = !pnode->fPauseRecv;
bool select_send;
{
LOCK(pnode->cs_vSend);
select_send = !pnode->vSendMsg.empty();
}
LOCK(pnode->cs_hSocket);
if (pnode->hSocket == INVALID_SOCKET)
continue;
FD_SET(pnode->hSocket, &fdsetError);
hSocketMax = std::max(hSocketMax, pnode->hSocket);
have_fds = true;
if (select_send) {
FD_SET(pnode->hSocket, &fdsetSend);
continue;
}
if (select_recv) {
FD_SET(pnode->hSocket, &fdsetRecv);
}
}
}
int nSelect = select(have_fds ? hSocketMax + 1 : 0,
&fdsetRecv, &fdsetSend, &fdsetError, &timeout);这里用到了系统提供的select()函数的多路复用机制，让程序监视多个文件描述符的状态变化。可以看到，上述代码遍历了所有节点，根据判断逻辑将其socket放入相应的发送文件描述符集合和接收文件描述符集合中，然后调用select()方法阻塞等待相应的文件描述符的读写事件的发生。
接下去这部分代码就是接受其他节点的连接，简单看下：
//
// Accept new connections
//
for (const ListenSocket& hListenSocket : vhListenSocket)
{
if (hListenSocket.socket != INVALID_SOCKET && FD_ISSET(hListenSocket.socket, &fdsetRecv))
{
AcceptConnection(hListenSocket);
}
}接着就是处理收发数据的逻辑了，遍历所有节点，判断其socket是否有I/O操作的发生：
for (CNode* pnode : vNodesCopy)
{
if (interruptNet)
return;
//
// Receive
//
bool recvSet = false;
bool sendSet = false;
bool errorSet = false;
{
LOCK(pnode->cs_hSocket);
if (pnode->hSocket == INVALID_SOCKET)
continue;
recvSet = FD_ISSET(pnode->hSocket, &fdsetRecv);
sendSet = FD_ISSET(pnode->hSocket, &fdsetSend);
errorSet = FD_ISSET(pnode->hSocket, &fdsetError);
}若某个节点的socket可读时将通过socket读取数据：
if (recvSet || errorSet)
{
// typical socket buffer is 8K-64K
char pchBuf[0x10000];
int nBytes = 0;
{
LOCK(pnode->cs_hSocket);
if (pnode->hSocket == INVALID_SOCKET)
continue;
nBytes = recv(pnode->hSocket, pchBuf, sizeof(pchBuf), MSG_DONTWAIT);
}
if (nBytes > 0)
{
bool notify = false;
if (!pnode->ReceiveMsgBytes(pchBuf, nBytes, notify))
pnode->CloseSocketDisconnect();
RecordBytesRecv(nBytes);
if (notify) {
size_t nSizeAdded = 0;
auto it(pnode->vRecvMsg.begin());
for (; it != pnode->vRecvMsg.end(); ++it) {
if (!it->complete())
break;
nSizeAdded += it->vRecv.size() + CMessageHeader::HEADER_SIZE;
}
{
LOCK(pnode->cs_vProcessMsg);
pnode->vProcessMsg.splice(pnode->vProcessMsg.end(), pnode->vRecvMsg, pnode->vRecvMsg.begin(), it);
pnode->nProcessQueueSize += nSizeAdded;
pnode->fPauseRecv = pnode->nProcessQueueSize > nReceiveFloodSize;
}
WakeMessageHandler();
}
}通过recv()方法读取数据放在临时的pchBuf缓冲数组中，然后通过pnode->ReceiveMsgBytes(pchBuf, nBytes, notify)方法将数据组装成CNetMessage消息对象，放入Cnode的vRecvMsg消息队列中。
然后通过 WakeMessageHandler()方法唤醒”msghand”消息处理线程，告诉他有新消息需要处理了：
void CConnman::WakeMessageHandler()
{
{
std::lock_guard<std::mutex> lock(mutexMsgProc);
fMsgProcWake = true;
}
condMsgProc.notify_one();
}继续看代码，若某个节点的socket可写时将数据通过socket发送出去，
//
// Send
//
if (sendSet)
{
LOCK(pnode->cs_vSend);
size_t nBytes = SocketSendData(pnode);
if (nBytes) {
RecordBytesSent(nBytes);
}
}(2) “dnsseed”线程，解析DNS seeds种子节点
threadDNSAddressSeed = std::thread(&TraceThread<std::function<void()> >, "dnsseed", std::function<void()>(std::bind(&CConnman::ThreadDNSAddressSeed, this)));《精通比特币》里讲过，利用种子节点发现对等节点有2中方法：
方法(1). 当-dnsseed开启时(即设置为1，默认就是1)，利用内置的硬编码的种子节点，这些种子节点由Bicoin Core核心开发者维护。
方法(2). 也可以通过-seednode 选项指定连接到某个种子节点。
这里的”dnsseed”线程实现的就是上述的方法(1)。
该线程的函数是 void CConnman::ThreadDNSAddressSeed()，来看下该函数：
const std::vector<std::string> &vSeeds = Params().DNSSeeds();
int found = 0;
LogPrintf("Loading addresses from DNS seeds (could take a while)\n");
for (const std::string &seed : vSeeds) {
if (interruptNet) {
return;
}
if (HaveNameProxy()) {
AddOneShot(seed);
} else {
std::vector<CNetAddr> vIPs;
std::vector<CAddress> vAdd;
ServiceFlags requiredServiceBits = GetDesirableServiceFlags(NODE_NONE);
std::string host = strprintf("x%x.%s", requiredServiceBits, seed);
CNetAddr resolveSource;
if (!resolveSource.SetInternal(host)) {
continue;
}
unsigned int nMaxIPs = 256; // Limits number of IPs learned from a DNS seed
if (LookupHost(host.c_str(), vIPs, nMaxIPs, true))
{
for (const CNetAddr& ip : vIPs)
{
int nOneDay = 24*3600;
CAddress addr = CAddress(CService(ip, Params().GetDefaultPort()), requiredServiceBits);
addr.nTime = GetTime() - 3*nOneDay - GetRand(4*nOneDay); // use a random age between 3 and 7 days old
vAdd.push_back(addr);
found++;
}
addrman.Add(vAdd, resolveSource);
} else {
// We now avoid directly using results from DNS Seeds which do not support service bit filtering,
// instead using them as a oneshot to get nodes with our desired service bits.
AddOneShot(seed);
}
}
}先通过Params().DNSSeeds()获取所有的DNS种子，然后遍历所有的DNS种子，通过LookupHost()方法解析出每个DNS种子的所有ip地址，存储在CAddrMan的mapAddr成员变量中。
(3) “addcon”线程，连接到由 “-addnode” 选项指定的节点
// Initiate outbound connections from -addnode
threadOpenAddedConnections = std::thread(&TraceThread<std::function<void()> >, "addcon", std::function<void()>(std::bind(&CConnman::ThreadOpenAddedConnections, this)));该线程的函数是 void CConnman::ThreadOpenAddedConnections()，来看下该函数：
void CConnman::ThreadOpenAddedConnections()
{
while (true)
{
CSemaphoreGrant grant(*semAddnode);
std::vector<AddedNodeInfo> vInfo = GetAddedNodeInfo();
bool tried = false;
for (const AddedNodeInfo& info : vInfo) {
if (!info.fConnected) {
if (!grant.TryAcquire()) {
// If we've used up our semaphore and need a new one, lets not wait here since while we are waiting
// the addednodeinfo state might change.
break;
}
tried = true;
CAddress addr(CService(), NODE_NONE);
OpenNetworkConnection(addr, false, &grant, info.strAddedNode.c_str(), false, false, true);
if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
return;
}
}
// Retry every 60 seconds if a connection was attempted, otherwise two seconds
if (!interruptNet.sleep_for(std::chrono::seconds(tried ? 60 : 2)))
return;
}
}代码比较少，过程也比较清晰，首先通过 vInfo = GetAddedNodeInfo() 获取由 “-addnode” 选项指定的节点，然后遍历这些节点，通过OpenNetworkConnection()方法建立与这些节点的连接。
(4) “opencon”线程，连接到”-seednode”选项指定的种子节点 和 内置的DNS种子节点，也会连接到由 “-connect” 选项指定的节点
threadOpenConnections = std::thread(&TraceThread<std::function<void()> >, "opencon", std::function<void()>(std::bind(&CConnman::ThreadOpenConnections, this, connOptions.m_specified_outgoing)));该线程的函数是 void CConnman::ThreadOpenConnections()，这个函数比较长，来看下该函数：
void CConnman::ThreadOpenConnections(const std::vector<std::string> connect)
{
// Connect to specific addresses
if (!connect.empty())
{
for (int64_t nLoop = 0;; nLoop++)
{
ProcessOneShot();首先执行ProcessOneShot()方法，看下该方法源码：
void CConnman::ProcessOneShot()
{
std::string strDest;
{
LOCK(cs_vOneShots);
if (vOneShots.empty())
return;
strDest = vOneShots.front();
vOneShots.pop_front();
}
CAddress addr;
CSemaphoreGrant grant(*semOutbound, true);
if (grant) {
OpenNetworkConnection(addr, false, &grant, strDest.c_str(), true);
}
}
通过 strDest = vOneShots.front()取得一个要连接的目的地址，而vOneShots队列里的元素是通过AddOneShot()方法添加的：
void CConnman::AddOneShot(const std::string& strDest)
{
LOCK(cs_vOneShots);
vOneShots.push_back(strDest);
}查找该方法使用的地方可以知道2种情况会调用该方法从而添加元素：
1). CConnman::Start()方法会添加”-seednode”选项指定的种子节点
2). ThreadDNSAddressSeed()方法在获取内置的DNS种子节点时会添加(当然关闭”-dnsseed”选项后，就不会进行ThreadDNSAddressSeed()相关操作了)
所以也就是说ProcessOneShot()方法会连接到 “-seednode”选项指定的种子节点 和 内置的DNS种子节点 这两部分种子节点。
好了，搞清楚ProcessOneShot()方法做的事情后，继续回到ThreadOpenConnections()方法分析源码：
for (const std::string& strAddr : connect)
{
CAddress addr(CService(), NODE_NONE);
OpenNetworkConnection(addr, false, nullptr, strAddr.c_str(), false, false, true);
for (int i = 0; i < 10 && i < nLoop; i++)
{
if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
return;
}
}
if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
return;即循环遍历由”-connect”选项指定的节点，并与之建立连接。
继续看代码，进入一个循环内：
// Minimum time before next feeler connection (in microseconds).
int64_t nNextFeeler = PoissonNextSend(nStart*1000*1000, FEELER_INTERVAL);
while (!interruptNet)
{
ProcessOneShot();
if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
return;首先又执行了一遍ProcessOneShot()方法，然后接下去是按规则从addrman中取出一个合适的地址：
//
// Choose an address to connect to based on most recently seen
//
CAddress addrConnect;
// Only connect out to one peer per network group (/16 for IPv4).
// Do this here so we don't have to critsect vNodes inside mapAddresses critsect.
int nOutbound = 0;
std::set<std::vector<unsigned char> > setConnected;
{
LOCK(cs_vNodes);
for (CNode* pnode : vNodes) {
if (!pnode->fInbound && !pnode->m_manual_connection) {
// Netgroups for inbound and addnode peers are not excluded because our goal here
// is to not use multiple of our limited outbound slots on a single netgroup
// but inbound and addnode peers do not use our outbound slots. Inbound peers
// also have the added issue that they're attacker controlled and could be used
// to prevent us from connecting to particular hosts if we used them here.
setConnected.insert(pnode->addr.GetGroup());
nOutbound++;
}
}
}
// Feeler Connections
//
// Design goals:
// * Increase the number of connectable addresses in the tried table.
//
// Method:
// * Choose a random address from new and attempt to connect to it if we can connect
// successfully it is added to tried.
// * Start attempting feeler connections only after node finishes making outbound
// connections.
// * Only make a feeler connection once every few minutes.
//
bool fFeeler = false;
if (nOutbound >= nMaxOutbound && !GetTryNewOutboundPeer()) {
int64_t nTime = GetTimeMicros(); // The current time right now (in microseconds).
if (nTime > nNextFeeler) {
nNextFeeler = PoissonNextSend(nTime, FEELER_INTERVAL);
fFeeler = true;
} else {
continue;
}
}
int64_t nANow = GetAdjustedTime();
int nTries = 0;
while (!interruptNet)
{
CAddrInfo addr = addrman.Select(fFeeler);
// if we selected an invalid address, restart
if (!addr.IsValid() || setConnected.count(addr.GetGroup()) || IsLocal(addr))
break;
// If we didn't find an appropriate destination after trying 100 addresses fetched from addrman,
// stop this loop, and let the outer loop run again (which sleeps, adds seed nodes, recalculates
// already-connected network ranges, ...) before trying new addrman addresses.
nTries++;
if (nTries > 100)
break;
if (IsLimited(addr))
continue;
// only consider very recently tried nodes after 30 failed attempts
if (nANow - addr.nLastTry < 600 && nTries < 30)
continue;
// for non-feelers, require all the services we'll want,
// for feelers, only require they be a full node (only because most
// SPV clients don't have a good address DB available)
if (!fFeeler && !HasAllDesirableServiceFlags(addr.nServices)) {
continue;
} else if (fFeeler && !MayHaveUsefulAddressDB(addr.nServices)) {
continue;
}
// do not allow non-default ports, unless after 50 invalid addresses selected already
if (addr.GetPort() != Params().GetDefaultPort() && nTries < 50)
continue;
addrConnect = addr;
break;
}上述循环执行完后，若找到一个有效的地址，则建立连接：
if (addrConnect.IsValid()) {
if (fFeeler) {
// Add small amount of random noise before connection to avoid synchronization.
int randsleep = GetRandInt(FEELER_SLEEP_WINDOW * 1000);
if (!interruptNet.sleep_for(std::chrono::milliseconds(randsleep)))
return;
LogPrint(BCLog::NET, "Making feeler connection to %s\n", addrConnect.ToString());
}
OpenNetworkConnection(addrConnect, (int)setConnected.size() >= std::min(nMaxConnections - 1, 2), &grant, nullptr, false, fFeeler);
}(5) “msghand”线程
// Process messages
threadMessageHandler = std::thread(&TraceThread<std::function<void()> >, "msghand", std::function<void()>(std::bind(&CConnman::ThreadMessageHandler, this)));该线程的函数是 void CConnman::ThreadMessageHandler()，来看下该函数：
void CConnman::ThreadMessageHandler()
{
while (!flagInterruptMsgProc)
{
std::vector<CNode*> vNodesCopy;
{
LOCK(cs_vNodes);
vNodesCopy = vNodes;
for (CNode* pnode : vNodesCopy) {
pnode->AddRef();
}
}
bool fMoreWork = false;
for (CNode* pnode : vNodesCopy)
{
if (pnode->fDisconnect)
continue;
// Receive messages
bool fMoreNodeWork = m_msgproc->ProcessMessages(pnode, flagInterruptMsgProc);
fMoreWork |= (fMoreNodeWork && !pnode->fPauseSend);
if (flagInterruptMsgProc)
return;
// Send messages
{
LOCK(pnode->cs_sendProcessing);
m_msgproc->SendMessages(pnode, flagInterruptMsgProc);
}
if (flagInterruptMsgProc)
return;
}
{
LOCK(cs_vNodes);
for (CNode* pnode : vNodesCopy)
pnode->Release();
}
std::unique_lock<std::mutex> lock(mutexMsgProc);
if (!fMoreWork) {
condMsgProc.wait_until(lock, std::chrono::steady_clock::now() + std::chrono::milliseconds(100), [this] { return fMsgProcWake; });
}
fMsgProcWake = false;
}
}
该函数比较简单，循环遍历所有节点，调用PeerLogicValidation::ProcessMessages()从节点的消息缓冲池中取出消息进行处理，PeerLogicValidation::ProcessMessages()这个方法最终会调用net_processing.cpp中的bool static ProcessMessage()静态方法处理消息，这个方法非常庞大，大概有1300多行：
bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStream& vRecv, int64_t nTimeReceived, const CChainParams& chainparams, CConnman* connman, const std::atomic<bool>& interruptMsgProc)
{
LogPrint(BCLog::NET, "received: %s (%u bytes) peer=%d\n", SanitizeString(strCommand), vRecv.size(), pfrom->GetId());
if (gArgs.IsArgSet("-dropmessagestest") && GetRand(gArgs.GetArg("-dropmessagestest", 0)) == 0)
{
LogPrintf("dropmessagestest DROPPING RECV MESSAGE\n");
return true;
}
if (!(pfrom->GetLocalServices() & NODE_BLOOM) &&
(strCommand == NetMsgType::FILTERLOAD ||
strCommand == NetMsgType::FILTERADD))
{
if (pfrom->nVersion >= NO_BLOOM_VERSION) {
LOCK(cs_main);
Misbehaving(pfrom->GetId(), 100);
return false;
} else {
pfrom->fDisconnect = true;
return false;
}
}
if (strCommand == NetMsgType::REJECT)
{
if (LogAcceptCategory(BCLog::NET)) {
try {
std::string strMsg; unsigned char ccode; std::string strReason;
vRecv >> LIMITED_STRING(strMsg, CMessageHeader::COMMAND_SIZE) >> ccode >> LIMITED_STRING(strReason, MAX_REJECT_MESSAGE_LENGTH);
std::ostringstream ss;
ss << strMsg << " code " << itostr(ccode) << ": " << strReason;
if (strMsg == NetMsgType::BLOCK || strMsg == NetMsgType::TX)
{
uint256 hash;
vRecv >> hash;
ss << ": hash " << hash.ToString();
}
LogPrint(BCLog::NET, "Reject %s\n", SanitizeString(ss.str()));
} catch (const std::ios_base::failure&) {
// Avoid feedback loops by preventing reject messages from triggering a new reject message.
LogPrint(BCLog::NET, "Unparseable reject message received\n");
}
}
}
else if (strCommand == NetMsgType::VERSION)
{
// Each connection can only send one version message
if (pfrom->nVersion != 0)
{所有类型的消息都在这个方法里处理，包括”version”，”verack”，”addr”，”inv”，”getdata”等所有类型，详见的src/protocol.h的NetMsgType命名空间里。
好了，回到ThreadMessageHandler()方法，继续分析，然后调用SendMessages()发送消息。最后该线程调用 condMsgProc.wait_until() 方法进入阻塞状态，当收到新的消息时由”net”线程调用WakeMessageHandler()方法唤醒该线程：
void CConnman::WakeMessageHandler()
{
{
std::lock_guard<std::mutex> lock(mutexMsgProc);
fMsgProcWake = true;
}
condMsgProc.notify_one();
}这里也可以看到收发消息是通过”net”线程和”msghand”线程结合condMsgProc条件变量协调工作的。
网络发现
上述讲了5个线程的相关工作，接下去从源码分析启动过程中的网络发现。
上述在讲”addcon”线程的ThreadOpenAddedConnections()方法和”opencon”线程的ThreadOpenConnections()方法时都讲到调用OpenNetworkConnection()方法与对等节点建立连接，网络发现的过程正是由这个OpenNetworkConnection()方法发起的:
// if successful, this moves the passed grant to the constructed node
void CConnman::OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant *grantOutbound, const char *pszDest, bool fOneShot, bool fFeeler, bool manual_connection)
{
//
// Initiate outbound network connection
//
if (interruptNet) {
return;
}
if (!fNetworkActive) {
return;
}
if (!pszDest) {
if (IsLocal(addrConnect) ||
FindNode((CNetAddr)addrConnect) || IsBanned(addrConnect) ||
FindNode(addrConnect.ToStringIPPort()))
return;
} else if (FindNode(std::string(pszDest)))
return;
CNode* pnode = ConnectNode(addrConnect, pszDest, fCountFailure);
if (!pnode)
return;
if (grantOutbound)
grantOutbound->MoveTo(pnode->grantOutbound);
if (fOneShot)
pnode->fOneShot = true;
if (fFeeler)
pnode->fFeeler = true;
if (manual_connection)
pnode->m_manual_connection = true;
m_msgproc->InitializeNode(pnode);
{
LOCK(cs_vNodes);
vNodes.push_back(pnode);
}
}上述源码看到，先由ConnectNode(addrConnect, pszDest, fCountFailure)方法与对等节点建立连接，然后调用InitializeNode(pnode)方法初始化节点，看下该方法：
void PeerLogicValidation::InitializeNode(CNode *pnode) {
CAddress addr = pnode->addr;
std::string addrName = pnode->GetAddrName();
NodeId nodeid = pnode->GetId();
{
LOCK(cs_main);
mapNodeState.emplace_hint(mapNodeState.end(), std::piecewise_construct, std::forward_as_tuple(nodeid), std::forward_as_tuple(addr, std::move(addrName)));
}
if(!pnode->fInbound)
PushNodeVersion(pnode, connman, GetTime());
}PushNodeVersion(pnode, connman, GetTime());就是将version消息发送给对等节点：
void PushNodeVersion(CNode *pnode, CConnman* connman, int64_t nTime)
{
ServiceFlags nLocalNodeServices = pnode->GetLocalServices();
uint64_t nonce = pnode->GetLocalNonce();
int nNodeStartingHeight = pnode->GetMyStartingHeight();
NodeId nodeid = pnode->GetId();
CAddress addr = pnode->addr;
CAddress addrYou = (addr.IsRoutable() && !IsProxy(addr) ? addr : CAddress(CService(), addr.nServices));
CAddress addrMe = CAddress(CService(), nLocalNodeServices);
connman->PushMessage(pnode, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::VERSION, PROTOCOL_VERSION, (uint64_t)nLocalNodeServices, nTime, addrYou, addrMe,
nonce, strSubVersion, nNodeStartingHeight, ::fRelayTxes));
if (fLogIPs) {
LogPrint(BCLog::NET, "send version message: version %d, blocks=%d, us=%s, them=%s, peer=%d\n", PROTOCOL_VERSION, nNodeStartingHeight, addrMe.ToString(), addrYou.ToString(), nodeid);
} else {
LogPrint(BCLog::NET, "send version message: version %d, blocks=%d, us=%s, peer=%d\n", PROTOCOL_VERSION, nNodeStartingHeight, addrMe.ToString(), nodeid);
}
}version消息发送给对等节点后就是等待对等节点返回verack应答消息了。前面分析 “msghand”线程时讲过，节点接收到消息后最终会调用net_processing.cpp中的bool static ProcessMessage()静态方法处理消息，来看下是如何处理”version”类型的消息的：
else if (strCommand == NetMsgType::VERSION)
{
// Each connection can only send one version message
if (pfrom->nVersion != 0)
{
connman->PushMessage(pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::REJECT, strCommand, REJECT_DUPLICATE, std::string("Duplicate version message")));
LOCK(cs_main);
Misbehaving(pfrom->GetId(), 1);
return false;
}
int64_t nTime;
CAddress addrMe;
CAddress addrFrom;
uint64_t nNonce = 1;
uint64_t nServiceInt;
ServiceFlags nServices;
int nVersion;
int nSendVersion;
std::string strSubVer;
std::string cleanSubVer;
int nStartingHeight = -1;
bool fRelay = true;
vRecv >> nVersion >> nServiceInt >> nTime >> addrMe;
nSendVersion = std::min(nVersion, PROTOCOL_VERSION);
nServices = ServiceFlags(nServiceInt);
if (!pfrom->fInbound)
{
connman->SetServices(pfrom->addr, nServices);
}
if (!pfrom->fInbound && !pfrom->fFeeler && !pfrom->m_manual_connection && !HasAllDesirableServiceFlags(nServices))
{
LogPrint(BCLog::NET, "peer=%d does not offer the expected services (%08x offered, %08x expected); disconnecting\n", pfrom->GetId(), nServices, GetDesirableServiceFlags(nServices));
connman->PushMessage(pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::REJECT, strCommand, REJECT_NONSTANDARD,
strprintf("Expected to offer services %08x", GetDesirableServiceFlags(nServices))));
pfrom->fDisconnect = true;
return false;
}先是进行一些检验，然后调用 connman->PushMessage(pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::VERACK));发送”verack”应答消息。