net.cpp

   1 // Copyright (c) 2009 Satoshi Nakamoto\r
   2 // Distributed under the MIT/X11 software license, see the accompanying\r
   3 // file license.txt or http://www.opensource.org/licenses/mit-license.php.\r
   4 \r
   5 #include "headers.h"\r
   6 \r
   7 void ThreadMessageHandler2(void* parg);\r
   8 void ThreadSocketHandler2(void* parg);\r
   9 void ThreadOpenConnections2(void* parg);\r
  10 bool OpenNetworkConnection(const CAddress& addrConnect);\r
  11 \r
  12 \r
  13 \r
  14 \r
  15 \r
  16 \r
  17 //\r
  18 // Global state variables\r
  19 //\r
  20 bool fClient = false;\r
  21 uint64 nLocalServices = (fClient ? 0 : NODE_NETWORK);\r
  22 CAddress addrLocalHost(0, DEFAULT_PORT, nLocalServices);\r
  23 CNode* pnodeLocalHost = NULL;\r
  24 uint64 nLocalHostNonce = 0;\r
  25 bool fShutdown = false;\r
  26 array<int, 10> vnThreadsRunning;\r
  27 SOCKET hListenSocket = INVALID_SOCKET;\r
  28 \r
  29 vector<CNode*> vNodes;\r
  30 CCriticalSection cs_vNodes;\r
  31 map<vector<unsigned char>, CAddress> mapAddresses;\r
  32 CCriticalSection cs_mapAddresses;\r
  33 map<CInv, CDataStream> mapRelay;\r
  34 deque<pair<int64, CInv> > vRelayExpiration;\r
  35 CCriticalSection cs_mapRelay;\r
  36 map<CInv, int64> mapAlreadyAskedFor;\r
  37 \r
  38 // Settings\r
  39 int fUseProxy = false;\r
  40 CAddress addrProxy("127.0.0.1:9050");\r
  41 \r
  42 \r
  43 \r
  44 bool ConnectSocket(const CAddress& addrConnect, SOCKET& hSocketRet)\r
  45 {\r
  46     hSocketRet = INVALID_SOCKET;\r
  47 \r
  48     SOCKET hSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);\r
  49     if (hSocket == INVALID_SOCKET)\r
  50         return false;\r
  51 #if defined(__BSD__) || defined(__WXOSX__)\r
  52     int set = 1;\r
  53     setsockopt(hSocket, SOL_SOCKET, SO_NOSIGPIPE, (void*)&set, sizeof(int));\r
  54 #endif\r
  55 \r
  56     bool fRoutable = !(addrConnect.GetByte(3) == 10 || (addrConnect.GetByte(3) == 192 && addrConnect.GetByte(2) == 168));\r
  57     bool fProxy = (fUseProxy && fRoutable);\r
  58     struct sockaddr_in sockaddr = (fProxy ? addrProxy.GetSockAddr() : addrConnect.GetSockAddr());\r
  59 \r
  60     if (connect(hSocket, (struct sockaddr*)&sockaddr, sizeof(sockaddr)) == SOCKET_ERROR)\r
  61     {\r
  62         closesocket(hSocket);\r
  63         return false;\r
  64     }\r
  65 \r
  66     if (fProxy)\r
  67     {\r
  68         printf("Proxy connecting %s\n", addrConnect.ToStringLog().c_str());\r
  69         char pszSocks4IP[] = "\4\1\0\0\0\0\0\0user";\r
  70         memcpy(pszSocks4IP + 2, &addrConnect.port, 2);\r
  71         memcpy(pszSocks4IP + 4, &addrConnect.ip, 4);\r
  72         char* pszSocks4 = pszSocks4IP;\r
  73         int nSize = sizeof(pszSocks4IP);\r
  74 \r
  75         int ret = send(hSocket, pszSocks4, nSize, MSG_NOSIGNAL);\r
  76         if (ret != nSize)\r
  77         {\r
  78             closesocket(hSocket);\r
  79             return error("Error sending to proxy");\r
  80         }\r
  81         char pchRet[8];\r
  82         if (recv(hSocket, pchRet, 8, 0) != 8)\r
  83         {\r
  84             closesocket(hSocket);\r
  85             return error("Error reading proxy response");\r
  86         }\r
  87         if (pchRet[1] != 0x5a)\r
  88         {\r
  89             closesocket(hSocket);\r
  90             return error("Proxy returned error %d", pchRet[1]);\r
  91         }\r
  92         printf("Proxy connection established %s\n", addrConnect.ToStringLog().c_str());\r
  93     }\r
  94 \r
  95     hSocketRet = hSocket;\r
  96     return true;\r
  97 }\r
  98 \r
  99 \r
 100 \r
 101 bool GetMyExternalIP2(const CAddress& addrConnect, const char* pszGet, const char* pszKeyword, unsigned int& ipRet)\r
 102 {\r
 103     SOCKET hSocket;\r
 104     if (!ConnectSocket(addrConnect, hSocket))\r
 105         return error("GetMyExternalIP() : connection to %s failed", addrConnect.ToString().c_str());\r
 106 \r
 107     send(hSocket, pszGet, strlen(pszGet), MSG_NOSIGNAL);\r
 108 \r
 109     string strLine;\r
 110     while (RecvLine(hSocket, strLine))\r
 111     {\r
 112         if (strLine.empty())\r
 113         {\r
 114             loop\r
 115             {\r
 116                 if (!RecvLine(hSocket, strLine))\r
 117                 {\r
 118                     closesocket(hSocket);\r
 119                     return false;\r
 120                 }\r
 121                 if (strLine.find(pszKeyword) != -1)\r
 122                 {\r
 123                     strLine = strLine.substr(strLine.find(pszKeyword) + strlen(pszKeyword));\r
 124                     break;\r
 125                 }\r
 126             }\r
 127             closesocket(hSocket);\r
 128             if (strLine.find("<"))\r
 129                 strLine = strLine.substr(0, strLine.find("<"));\r
 130             strLine = strLine.substr(strspn(strLine.c_str(), " \t\n\r"));\r
 131             while (strLine.size() > 0 && isspace(strLine[strLine.size()-1]))\r
 132                 strLine.resize(strLine.size()-1);\r
 133             CAddress addr(strLine.c_str());\r
 134             printf("GetMyExternalIP() received [%s] %s\n", strLine.c_str(), addr.ToString().c_str());\r
 135             if (addr.ip == 0 || addr.ip == INADDR_NONE || !addr.IsRoutable())\r
 136                 return false;\r
 137             ipRet = addr.ip;\r
 138             return true;\r
 139         }\r
 140     }\r
 141     closesocket(hSocket);\r
 142     return error("GetMyExternalIP() : connection closed");\r
 143 }\r
 144 \r
 145 \r
 146 bool GetMyExternalIP(unsigned int& ipRet)\r
 147 {\r
 148     CAddress addrConnect;\r
 149     char* pszGet;\r
 150     char* pszKeyword;\r
 151 \r
 152     if (fUseProxy)\r
 153         return false;\r
 154 \r
 155     for (int nLookup = 0; nLookup <= 1; nLookup++)\r
 156     for (int nHost = 1; nHost <= 2; nHost++)\r
 157     {\r
 158         if (nHost == 1)\r
 159         {\r
 160             addrConnect = CAddress("70.86.96.218:80"); // www.ipaddressworld.com\r
 161 \r
 162             if (nLookup == 1)\r
 163             {\r
 164                 struct hostent* phostent = gethostbyname("www.ipaddressworld.com");\r
 165                 if (phostent && phostent->h_addr_list && phostent->h_addr_list[0])\r
 166                     addrConnect = CAddress(*(u_long*)phostent->h_addr_list[0], htons(80));\r
 167             }\r
 168 \r
 169             pszGet = "GET /ip.php HTTP/1.1\r\n"\r
 170                      "Host: www.ipaddressworld.com\r\n"\r
 171                      "User-Agent: Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)\r\n"\r
 172                      "Connection: close\r\n"\r
 173                      "\r\n";\r
 174 \r
 175             pszKeyword = "IP:";\r
 176         }\r
 177         else if (nHost == 2)\r
 178         {\r
 179             addrConnect = CAddress("208.78.68.70:80"); // checkip.dyndns.org\r
 180 \r
 181             if (nLookup == 1)\r
 182             {\r
 183                 struct hostent* phostent = gethostbyname("checkip.dyndns.org");\r
 184                 if (phostent && phostent->h_addr_list && phostent->h_addr_list[0])\r
 185                     addrConnect = CAddress(*(u_long*)phostent->h_addr_list[0], htons(80));\r
 186             }\r
 187 \r
 188             pszGet = "GET / HTTP/1.1\r\n"\r
 189                      "Host: checkip.dyndns.org\r\n"\r
 190                      "User-Agent: Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)\r\n"\r
 191                      "Connection: close\r\n"\r
 192                      "\r\n";\r
 193 \r
 194             pszKeyword = "Address:";\r
 195         }\r
 196 \r
 197         if (GetMyExternalIP2(addrConnect, pszGet, pszKeyword, ipRet))\r
 198             return true;\r
 199     }\r
 200 \r
 201     return false;\r
 202 }\r
 203 \r
 204 \r
 205 \r
 206 \r
 207 \r
 208 bool AddAddress(CAddrDB& addrdb, CAddress addr, bool fCurrentlyOnline)\r
 209 {\r
 210     if (!addr.IsRoutable())\r
 211         return false;\r
 212     if (addr.ip == addrLocalHost.ip)\r
 213         return false;\r
 214     if (fCurrentlyOnline)\r
 215         addr.nTime = GetAdjustedTime();\r
 216     CRITICAL_BLOCK(cs_mapAddresses)\r
 217     {\r
 218         map<vector<unsigned char>, CAddress>::iterator it = mapAddresses.find(addr.GetKey());\r
 219         if (it == mapAddresses.end())\r
 220         {\r
 221             // New address\r
 222             mapAddresses.insert(make_pair(addr.GetKey(), addr));\r
 223             addrdb.WriteAddress(addr);\r
 224             return true;\r
 225         }\r
 226         else\r
 227         {\r
 228             bool fUpdated = false;\r
 229             CAddress& addrFound = (*it).second;\r
 230             if ((addrFound.nServices | addr.nServices) != addrFound.nServices)\r
 231             {\r
 232                 // Services have been added\r
 233                 addrFound.nServices |= addr.nServices;\r
 234                 fUpdated = true;\r
 235             }\r
 236             int64 nUpdateInterval = (fCurrentlyOnline ? 60 * 60 : 24 * 60 * 60);\r
 237             if (addrFound.nTime < addr.nTime - nUpdateInterval)\r
 238             {\r
 239                 // Periodically update most recently seen time\r
 240                 addrFound.nTime = addr.nTime;\r
 241                 fUpdated = true;\r
 242             }\r
 243             if (fUpdated)\r
 244                 addrdb.WriteAddress(addrFound);\r
 245         }\r
 246     }\r
 247     return false;\r
 248 }\r
 249 \r
 250 void AddressCurrentlyConnected(const CAddress& addr)\r
 251 {\r
 252     CRITICAL_BLOCK(cs_mapAddresses)\r
 253     {\r
 254         // Only if it's been published already\r
 255         map<vector<unsigned char>, CAddress>::iterator it = mapAddresses.find(addr.GetKey());\r
 256         if (it != mapAddresses.end())\r
 257         {\r
 258             CAddress& addrFound = (*it).second;\r
 259             int64 nUpdateInterval = 60 * 60;\r
 260             if (addrFound.nTime < GetAdjustedTime() - nUpdateInterval)\r
 261             {\r
 262                 // Periodically update most recently seen time\r
 263                 addrFound.nTime = GetAdjustedTime();\r
 264                 CAddrDB addrdb;\r
 265                 addrdb.WriteAddress(addrFound);\r
 266             }\r
 267         }\r
 268     }\r
 269 }\r
 270 \r
 271 \r
 272 \r
 273 \r
 274 \r
 275 void AbandonRequests(void (*fn)(void*, CDataStream&), void* param1)\r
 276 {\r
 277     // If the dialog might get closed before the reply comes back,\r
 278     // call this in the destructor so it doesn't get called after it's deleted.\r
 279     CRITICAL_BLOCK(cs_vNodes)\r
 280     {\r
 281         foreach(CNode* pnode, vNodes)\r
 282         {\r
 283             CRITICAL_BLOCK(pnode->cs_mapRequests)\r
 284             {\r
 285                 for (map<uint256, CRequestTracker>::iterator mi = pnode->mapRequests.begin(); mi != pnode->mapRequests.end();)\r
 286                 {\r
 287                     CRequestTracker& tracker = (*mi).second;\r
 288                     if (tracker.fn == fn && tracker.param1 == param1)\r
 289                         pnode->mapRequests.erase(mi++);\r
 290                     else\r
 291                         mi++;\r
 292                 }\r
 293             }\r
 294         }\r
 295     }\r
 296 }\r
 297 \r
 298 \r
 299 \r
 300 \r
 301 \r
 302 \r
 303 \r
 304 //\r
 305 // Subscription methods for the broadcast and subscription system.\r
 306 // Channel numbers are message numbers, i.e. MSG_TABLE and MSG_PRODUCT.\r
 307 //\r
 308 // The subscription system uses a meet-in-the-middle strategy.\r
 309 // With 100,000 nodes, if senders broadcast to 1000 random nodes and receivers\r
 310 // subscribe to 1000 random nodes, 99.995% (1 - 0.99^1000) of messages will get through.\r
 311 //\r
 312 \r
 313 bool AnySubscribed(unsigned int nChannel)\r
 314 {\r
 315     if (pnodeLocalHost->IsSubscribed(nChannel))\r
 316         return true;\r
 317     CRITICAL_BLOCK(cs_vNodes)\r
 318         foreach(CNode* pnode, vNodes)\r
 319             if (pnode->IsSubscribed(nChannel))\r
 320                 return true;\r
 321     return false;\r
 322 }\r
 323 \r
 324 bool CNode::IsSubscribed(unsigned int nChannel)\r
 325 {\r
 326     if (nChannel >= vfSubscribe.size())\r
 327         return false;\r
 328     return vfSubscribe[nChannel];\r
 329 }\r
 330 \r
 331 void CNode::Subscribe(unsigned int nChannel, unsigned int nHops)\r
 332 {\r
 333     if (nChannel >= vfSubscribe.size())\r
 334         return;\r
 335 \r
 336     if (!AnySubscribed(nChannel))\r
 337     {\r
 338         // Relay subscribe\r
 339         CRITICAL_BLOCK(cs_vNodes)\r
 340             foreach(CNode* pnode, vNodes)\r
 341                 if (pnode != this)\r
 342                     pnode->PushMessage("subscribe", nChannel, nHops);\r
 343     }\r
 344 \r
 345     vfSubscribe[nChannel] = true;\r
 346 }\r
 347 \r
 348 void CNode::CancelSubscribe(unsigned int nChannel)\r
 349 {\r
 350     if (nChannel >= vfSubscribe.size())\r
 351         return;\r
 352 \r
 353     // Prevent from relaying cancel if wasn't subscribed\r
 354     if (!vfSubscribe[nChannel])\r
 355         return;\r
 356     vfSubscribe[nChannel] = false;\r
 357 \r
 358     if (!AnySubscribed(nChannel))\r
 359     {\r
 360         // Relay subscription cancel\r
 361         CRITICAL_BLOCK(cs_vNodes)\r
 362             foreach(CNode* pnode, vNodes)\r
 363                 if (pnode != this)\r
 364                     pnode->PushMessage("sub-cancel", nChannel);\r
 365 \r
 366         // Clear memory, no longer subscribed\r
 367         if (nChannel == MSG_PRODUCT)\r
 368             CRITICAL_BLOCK(cs_mapProducts)\r
 369                 mapProducts.clear();\r
 370     }\r
 371 }\r
 372 \r
 373 \r
 374 \r
 375 \r
 376 \r
 377 \r
 378 \r
 379 \r
 380 \r
 381 CNode* FindNode(unsigned int ip)\r
 382 {\r
 383     CRITICAL_BLOCK(cs_vNodes)\r
 384     {\r
 385         foreach(CNode* pnode, vNodes)\r
 386             if (pnode->addr.ip == ip)\r
 387                 return (pnode);\r
 388     }\r
 389     return NULL;\r
 390 }\r
 391 \r
 392 CNode* FindNode(CAddress addr)\r
 393 {\r
 394     CRITICAL_BLOCK(cs_vNodes)\r
 395     {\r
 396         foreach(CNode* pnode, vNodes)\r
 397             if (pnode->addr == addr)\r
 398                 return (pnode);\r
 399     }\r
 400     return NULL;\r
 401 }\r
 402 \r
 403 CNode* ConnectNode(CAddress addrConnect, int64 nTimeout)\r
 404 {\r
 405     if (addrConnect.ip == addrLocalHost.ip)\r
 406         return NULL;\r
 407 \r
 408     // Look for an existing connection\r
 409     CNode* pnode = FindNode(addrConnect.ip);\r
 410     if (pnode)\r
 411     {\r
 412         if (nTimeout != 0)\r
 413             pnode->AddRef(nTimeout);\r
 414         else\r
 415             pnode->AddRef();\r
 416         return pnode;\r
 417     }\r
 418 \r
 419     /// debug print\r
 420     printf("trying connection %s\n", addrConnect.ToStringLog().c_str());\r
 421 \r
 422     // Connect\r
 423     SOCKET hSocket;\r
 424     if (ConnectSocket(addrConnect, hSocket))\r
 425     {\r
 426         /// debug print\r
 427         printf("connected %s\n", addrConnect.ToStringLog().c_str());\r
 428 \r
 429         // Set to nonblocking\r
 430 #ifdef __WXMSW__\r
 431                 u_long nOne = 1;\r
 432         if (ioctlsocket(hSocket, FIONBIO, &nOne) == SOCKET_ERROR)\r
 433             printf("ConnectSocket() : ioctlsocket nonblocking setting failed, error %d\n", WSAGetLastError());\r
 434 #else\r
 435         if (fcntl(hSocket, F_SETFL, O_NONBLOCK) == SOCKET_ERROR)\r
 436             printf("ConnectSocket() : fcntl nonblocking setting failed, error %d\n", errno);\r
 437 #endif\r
 438 \r
 439         // Add node\r
 440         CNode* pnode = new CNode(hSocket, addrConnect, false);\r
 441         if (nTimeout != 0)\r
 442             pnode->AddRef(nTimeout);\r
 443         else\r
 444             pnode->AddRef();\r
 445         CRITICAL_BLOCK(cs_vNodes)\r
 446             vNodes.push_back(pnode);\r
 447 \r
 448         CRITICAL_BLOCK(cs_mapAddresses)\r
 449             mapAddresses[addrConnect.GetKey()].nLastFailed = 0;\r
 450         return pnode;\r
 451     }\r
 452     else\r
 453     {\r
 454         CRITICAL_BLOCK(cs_mapAddresses)\r
 455             mapAddresses[addrConnect.GetKey()].nLastFailed = GetAdjustedTime();\r
 456         return NULL;\r
 457     }\r
 458 }\r
 459 \r
 460 void CNode::DoDisconnect()\r
 461 {\r
 462     printf("disconnecting node %s\n", addr.ToStringLog().c_str());\r
 463 \r
 464     closesocket(hSocket);\r
 465 \r
 466     // If outbound and never got version message, mark address as failed\r
 467     if (!fInbound && !fSuccessfullyConnected)\r
 468         CRITICAL_BLOCK(cs_mapAddresses)\r
 469             mapAddresses[addr.GetKey()].nLastFailed = GetAdjustedTime();\r
 470 \r
 471     // All of a nodes broadcasts and subscriptions are automatically torn down\r
 472     // when it goes down, so a node has to stay up to keep its broadcast going.\r
 473 \r
 474     CRITICAL_BLOCK(cs_mapProducts)\r
 475         for (map<uint256, CProduct>::iterator mi = mapProducts.begin(); mi != mapProducts.end();)\r
 476             AdvertRemoveSource(this, MSG_PRODUCT, 0, (*(mi++)).second);\r
 477 \r
 478     // Cancel subscriptions\r
 479     for (unsigned int nChannel = 0; nChannel < vfSubscribe.size(); nChannel++)\r
 480         if (vfSubscribe[nChannel])\r
 481             CancelSubscribe(nChannel);\r
 482 }\r
 483 \r
 484 \r
 485 \r
 486 \r
 487 \r
 488 \r
 489 \r
 490 \r
 491 \r
 492 \r
 493 \r
 494 \r
 495 \r
 496 void ThreadSocketHandler(void* parg)\r
 497 {\r
 498     IMPLEMENT_RANDOMIZE_STACK(ThreadSocketHandler(parg));\r
 499 \r
 500     try\r
 501     {\r
 502         vnThreadsRunning[0]++;\r
 503         ThreadSocketHandler2(parg);\r
 504         vnThreadsRunning[0]--;\r
 505     }\r
 506     catch (std::exception& e) {\r
 507         vnThreadsRunning[0]--;\r
 508         PrintException(&e, "ThreadSocketHandler()");\r
 509     } catch (...) {\r
 510         vnThreadsRunning[0]--;\r
 511         PrintException(NULL, "ThreadSocketHandler()");\r
 512     }\r
 513 \r
 514     printf("ThreadSocketHandler exiting\n");\r
 515 }\r
 516 \r
 517 void ThreadSocketHandler2(void* parg)\r
 518 {\r
 519     printf("ThreadSocketHandler started\n");\r
 520     list<CNode*> vNodesDisconnected;\r
 521     int nPrevNodeCount = 0;\r
 522 \r
 523     loop\r
 524     {\r
 525         //\r
 526         // Disconnect nodes\r
 527         //\r
 528         CRITICAL_BLOCK(cs_vNodes)\r
 529         {\r
 530             // Disconnect unused nodes\r
 531             vector<CNode*> vNodesCopy = vNodes;\r
 532             foreach(CNode* pnode, vNodesCopy)\r
 533             {\r
 534                 if (pnode->ReadyToDisconnect() && pnode->vRecv.empty() && pnode->vSend.empty())\r
 535                 {\r
 536                     // remove from vNodes\r
 537                     vNodes.erase(remove(vNodes.begin(), vNodes.end(), pnode), vNodes.end());\r
 538                     pnode->DoDisconnect();\r
 539 \r
 540                     // hold in disconnected pool until all refs are released\r
 541                     pnode->nReleaseTime = max(pnode->nReleaseTime, GetTime() + 5 * 60);\r
 542                     if (pnode->fNetworkNode)\r
 543                         pnode->Release();\r
 544                     vNodesDisconnected.push_back(pnode);\r
 545                 }\r
 546             }\r
 547 \r
 548             // Delete disconnected nodes\r
 549             list<CNode*> vNodesDisconnectedCopy = vNodesDisconnected;\r
 550             foreach(CNode* pnode, vNodesDisconnectedCopy)\r
 551             {\r
 552                 // wait until threads are done using it\r
 553                 if (pnode->GetRefCount() <= 0)\r
 554                 {\r
 555                     bool fDelete = false;\r
 556                     TRY_CRITICAL_BLOCK(pnode->cs_vSend)\r
 557                      TRY_CRITICAL_BLOCK(pnode->cs_vRecv)\r
 558                       TRY_CRITICAL_BLOCK(pnode->cs_mapRequests)\r
 559                        TRY_CRITICAL_BLOCK(pnode->cs_inventory)\r
 560                         fDelete = true;\r
 561                     if (fDelete)\r
 562                     {\r
 563                         vNodesDisconnected.remove(pnode);\r
 564                         delete pnode;\r
 565                     }\r
 566                 }\r
 567             }\r
 568         }\r
 569         if (vNodes.size() != nPrevNodeCount)\r
 570         {\r
 571             nPrevNodeCount = vNodes.size();\r
 572             MainFrameRepaint();\r
 573         }\r
 574 \r
 575 \r
 576         //\r
 577         // Find which sockets have data to receive\r
 578         //\r
 579         struct timeval timeout;\r
 580         timeout.tv_sec  = 0;\r
 581         timeout.tv_usec = 50000; // frequency to poll pnode->vSend\r
 582 \r
 583         fd_set fdsetRecv;\r
 584         fd_set fdsetSend;\r
 585         FD_ZERO(&fdsetRecv);\r
 586         FD_ZERO(&fdsetSend);\r
 587         SOCKET hSocketMax = 0;\r
 588         FD_SET(hListenSocket, &fdsetRecv);\r
 589         hSocketMax = max(hSocketMax, hListenSocket);\r
 590         CRITICAL_BLOCK(cs_vNodes)\r
 591         {\r
 592             foreach(CNode* pnode, vNodes)\r
 593             {\r
 594                 FD_SET(pnode->hSocket, &fdsetRecv);\r
 595                 hSocketMax = max(hSocketMax, pnode->hSocket);\r
 596                 TRY_CRITICAL_BLOCK(pnode->cs_vSend)\r
 597                     if (!pnode->vSend.empty())\r
 598                         FD_SET(pnode->hSocket, &fdsetSend);\r
 599             }\r
 600         }\r
 601 \r
 602         vnThreadsRunning[0]--;\r
 603         int nSelect = select(hSocketMax + 1, &fdsetRecv, &fdsetSend, NULL, &timeout);\r
 604         vnThreadsRunning[0]++;\r
 605         if (fShutdown)\r
 606             return;\r
 607         if (nSelect == SOCKET_ERROR)\r
 608         {\r
 609             int nErr = WSAGetLastError();\r
 610             printf("select failed: %d\n", nErr);\r
 611             for (int i = 0; i <= hSocketMax; i++)\r
 612             {\r
 613                 FD_SET(i, &fdsetRecv);\r
 614                 FD_SET(i, &fdsetSend);\r
 615             }\r
 616             Sleep(timeout.tv_usec/1000);\r
 617         }\r
 618 \r
 619         //// debug print\r
 620         //foreach(CNode* pnode, vNodes)\r
 621         //{\r
 622         //    printf("vRecv = %-5d ", pnode->vRecv.size());\r
 623         //    printf("vSend = %-5d    ", pnode->vSend.size());\r
 624         //}\r
 625         //printf("\n");\r
 626 \r
 627 \r
 628         //\r
 629         // Accept new connections\r
 630         //\r
 631         if (FD_ISSET(hListenSocket, &fdsetRecv))\r
 632         {\r
 633             struct sockaddr_in sockaddr;\r
 634 #ifdef __WXMSW__\r
 635             int len = sizeof(sockaddr);\r
 636 #else\r
 637             socklen_t len = sizeof(sockaddr);\r
 638 #endif\r
 639             SOCKET hSocket = accept(hListenSocket, (struct sockaddr*)&sockaddr, &len);\r
 640             CAddress addr(sockaddr);\r
 641             if (hSocket == INVALID_SOCKET)\r
 642             {\r
 643                 if (WSAGetLastError() != WSAEWOULDBLOCK)\r
 644                     printf("ERROR ThreadSocketHandler accept failed: %d\n", WSAGetLastError());\r
 645             }\r
 646             else\r
 647             {\r
 648                 printf("accepted connection %s\n", addr.ToStringLog().c_str());\r
 649                 CNode* pnode = new CNode(hSocket, addr, true);\r
 650                 pnode->AddRef();\r
 651                 CRITICAL_BLOCK(cs_vNodes)\r
 652                     vNodes.push_back(pnode);\r
 653             }\r
 654         }\r
 655 \r
 656 \r
 657         //\r
 658         // Service each socket\r
 659         //\r
 660         vector<CNode*> vNodesCopy;\r
 661         CRITICAL_BLOCK(cs_vNodes)\r
 662             vNodesCopy = vNodes;\r
 663         foreach(CNode* pnode, vNodesCopy)\r
 664         {\r
 665             if (fShutdown)\r
 666                 return;\r
 667             SOCKET hSocket = pnode->hSocket;\r
 668 \r
 669             //\r
 670             // Receive\r
 671             //\r
 672             if (FD_ISSET(hSocket, &fdsetRecv))\r
 673             {\r
 674                 TRY_CRITICAL_BLOCK(pnode->cs_vRecv)\r
 675                 {\r
 676                     CDataStream& vRecv = pnode->vRecv;\r
 677                     unsigned int nPos = vRecv.size();\r
 678 \r
 679                     // typical socket buffer is 8K-64K\r
 680                     const unsigned int nBufSize = 0x10000;\r
 681                     vRecv.resize(nPos + nBufSize);\r
 682                     int nBytes = recv(hSocket, &vRecv[nPos], nBufSize, 0);\r
 683                     vRecv.resize(nPos + max(nBytes, 0));\r
 684                     if (nBytes == 0)\r
 685                     {\r
 686                         // socket closed gracefully\r
 687                         if (!pnode->fDisconnect)\r
 688                             printf("recv: socket closed\n");\r
 689                         pnode->fDisconnect = true;\r
 690                     }\r
 691                     else if (nBytes < 0)\r
 692                     {\r
 693                         // socket error\r
 694                         int nErr = WSAGetLastError();\r
 695                         if (nErr != WSAEWOULDBLOCK && nErr != WSAEMSGSIZE && nErr != WSAEINTR && nErr != WSAEINPROGRESS)\r
 696                         {\r
 697                             if (!pnode->fDisconnect)\r
 698                                 printf("recv failed: %d\n", nErr);\r
 699                             pnode->fDisconnect = true;\r
 700                         }\r
 701                     }\r
 702                 }\r
 703             }\r
 704 \r
 705             //\r
 706             // Send\r
 707             //\r
 708             if (FD_ISSET(hSocket, &fdsetSend))\r
 709             {\r
 710                 TRY_CRITICAL_BLOCK(pnode->cs_vSend)\r
 711                 {\r
 712                     CDataStream& vSend = pnode->vSend;\r
 713                     if (!vSend.empty())\r
 714                     {\r
 715                         int nBytes = send(hSocket, &vSend[0], vSend.size(), MSG_NOSIGNAL);\r
 716                         if (nBytes > 0)\r
 717                         {\r
 718                             vSend.erase(vSend.begin(), vSend.begin() + nBytes);\r
 719                         }\r
 720                         else if (nBytes == 0)\r
 721                         {\r
 722                             if (pnode->ReadyToDisconnect())\r
 723                                 pnode->vSend.clear();\r
 724                         }\r
 725                         else\r
 726                         {\r
 727                             printf("send error %d\n", nBytes);\r
 728                             if (pnode->ReadyToDisconnect())\r
 729                                 pnode->vSend.clear();\r
 730                         }\r
 731                     }\r
 732                 }\r
 733             }\r
 734         }\r
 735 \r
 736 \r
 737         Sleep(10);\r
 738     }\r
 739 }\r
 740 \r
 741 \r
 742 \r
 743 \r
 744 \r
 745 \r
 746 \r
 747 \r
 748 \r
 749 \r
 750 void ThreadOpenConnections(void* parg)\r
 751 {\r
 752     IMPLEMENT_RANDOMIZE_STACK(ThreadOpenConnections(parg));\r
 753 \r
 754     try\r
 755     {\r
 756         vnThreadsRunning[1]++;\r
 757         ThreadOpenConnections2(parg);\r
 758         vnThreadsRunning[1]--;\r
 759     }\r
 760     catch (std::exception& e) {\r
 761         vnThreadsRunning[1]--;\r
 762         PrintException(&e, "ThreadOpenConnections()");\r
 763     } catch (...) {\r
 764         vnThreadsRunning[1]--;\r
 765         PrintException(NULL, "ThreadOpenConnections()");\r
 766     }\r
 767 \r
 768     printf("ThreadOpenConnections exiting\n");\r
 769 }\r
 770 \r
 771 void ThreadOpenConnections2(void* parg)\r
 772 {\r
 773     printf("ThreadOpenConnections started\n");\r
 774 \r
 775     // Connect to one specified address\r
 776     while (mapArgs.count("-connect"))\r
 777     {\r
 778         OpenNetworkConnection(CAddress(mapArgs["-connect"]));\r
 779         for (int i = 0; i < 10; i++)\r
 780         {\r
 781             Sleep(1000);\r
 782             if (fShutdown)\r
 783                 return;\r
 784         }\r
 785     }\r
 786 \r
 787     // Connect to manually added nodes first\r
 788     if (mapArgs.count("-addnode"))\r
 789     {\r
 790         foreach(string strAddr, mapMultiArgs["-addnode"])\r
 791         {\r
 792             CAddress addr(strAddr, NODE_NETWORK);\r
 793             if (addr.IsValid())\r
 794             {\r
 795                 OpenNetworkConnection(addr);\r
 796                 Sleep(1000);\r
 797                 if (fShutdown)\r
 798                     return;\r
 799             }\r
 800         }\r
 801     }\r
 802 \r
 803     // Initiate network connections\r
 804     loop\r
 805     {\r
 806         // Wait\r
 807         vnThreadsRunning[1]--;\r
 808         Sleep(500);\r
 809         const int nMaxConnections = 15;\r
 810         while (vNodes.size() >= nMaxConnections || vNodes.size() >= mapAddresses.size())\r
 811         {\r
 812             if (fShutdown)\r
 813                 return;\r
 814             Sleep(2000);\r
 815         }\r
 816         vnThreadsRunning[1]++;\r
 817         if (fShutdown)\r
 818             return;\r
 819 \r
 820         //\r
 821         // Choose an address to connect to based on most recently seen\r
 822         //\r
 823         CAddress addrConnect;\r
 824         int64 nBestTime = 0;\r
 825         int64 nDelay = ((60 * 60) << vNodes.size());\r
 826         if (vNodes.size() >= 3)\r
 827             nDelay *= 4;\r
 828         if (nGotIRCAddresses > 0)\r
 829             nDelay *= 100;\r
 830 \r
 831         // Do this here so we don't have to critsect vNodes inside mapAddresses critsect\r
 832         set<unsigned int> setConnected;\r
 833         CRITICAL_BLOCK(cs_vNodes)\r
 834             foreach(CNode* pnode, vNodes)\r
 835                 setConnected.insert(pnode->addr.ip);\r
 836 \r
 837         CRITICAL_BLOCK(cs_mapAddresses)\r
 838         {\r
 839             foreach(const PAIRTYPE(vector<unsigned char>, CAddress)& item, mapAddresses)\r
 840             {\r
 841                 const CAddress& addr = item.second;\r
 842                 if (!addr.IsIPv4() || !addr.IsValid() || setConnected.count(addr.ip))\r
 843                     continue;\r
 844 \r
 845                 // Limit retry frequency\r
 846                 if (GetAdjustedTime() < addr.nLastFailed + nDelay)\r
 847                     continue;\r
 848 \r
 849                 // Try again only after all addresses had a first attempt\r
 850                 int64 nTime = addr.nTime;\r
 851                 if (addr.nLastFailed > addr.nTime)\r
 852                     nTime -= 365 * 24 * 60 * 60;\r
 853 \r
 854                 // Randomize the order a little, putting the standard port first\r
 855                 nTime += GetRand(1 * 60 * 60);\r
 856                 if (addr.port != DEFAULT_PORT)\r
 857                     nTime -= 1 * 60 * 60;\r
 858 \r
 859                 if (nTime > nBestTime)\r
 860                 {\r
 861                     nBestTime = nTime;\r
 862                     addrConnect = addr;\r
 863                 }\r
 864             }\r
 865         }\r
 866 \r
 867         if (addrConnect.IsValid())\r
 868             OpenNetworkConnection(addrConnect);\r
 869     }\r
 870 }\r
 871 \r
 872 bool OpenNetworkConnection(const CAddress& addrConnect)\r
 873 {\r
 874     //\r
 875     // Initiate outbound network connection\r
 876     //\r
 877     if (fShutdown)\r
 878         return false;\r
 879     if (addrConnect.ip == addrLocalHost.ip || !addrConnect.IsIPv4() || FindNode(addrConnect.ip))\r
 880         return false;\r
 881 \r
 882     vnThreadsRunning[1]--;\r
 883     CNode* pnode = ConnectNode(addrConnect);\r
 884     vnThreadsRunning[1]++;\r
 885     if (fShutdown)\r
 886         return false;\r
 887     if (!pnode)\r
 888         return false;\r
 889     pnode->fNetworkNode = true;\r
 890 \r
 891     if (addrLocalHost.IsRoutable() && !fUseProxy)\r
 892     {\r
 893         // Advertise our address\r
 894         vector<CAddress> vAddrToSend;\r
 895         vAddrToSend.push_back(addrLocalHost);\r
 896         pnode->PushMessage("addr", vAddrToSend);\r
 897     }\r
 898 \r
 899     // Get as many addresses as we can\r
 900     pnode->PushMessage("getaddr");\r
 901     pnode->fGetAddr = true; // don't relay the results of the getaddr\r
 902 \r
 903     ////// should the one on the receiving end do this too?\r
 904     // Subscribe our local subscription list\r
 905     const unsigned int nHops = 0;\r
 906     for (unsigned int nChannel = 0; nChannel < pnodeLocalHost->vfSubscribe.size(); nChannel++)\r
 907         if (pnodeLocalHost->vfSubscribe[nChannel])\r
 908             pnode->PushMessage("subscribe", nChannel, nHops);\r
 909 \r
 910     return true;\r
 911 }\r
 912 \r
 913 \r
 914 \r
 915 \r
 916 \r
 917 \r
 918 \r
 919 \r
 920 void ThreadMessageHandler(void* parg)\r
 921 {\r
 922     IMPLEMENT_RANDOMIZE_STACK(ThreadMessageHandler(parg));\r
 923 \r
 924     try\r
 925     {\r
 926         vnThreadsRunning[2]++;\r
 927         ThreadMessageHandler2(parg);\r
 928         vnThreadsRunning[2]--;\r
 929     }\r
 930     catch (std::exception& e) {\r
 931         vnThreadsRunning[2]--;\r
 932         PrintException(&e, "ThreadMessageHandler()");\r
 933     } catch (...) {\r
 934         vnThreadsRunning[2]--;\r
 935         PrintException(NULL, "ThreadMessageHandler()");\r
 936     }\r
 937 \r
 938     printf("ThreadMessageHandler exiting\n");\r
 939 }\r
 940 \r
 941 void ThreadMessageHandler2(void* parg)\r
 942 {\r
 943     printf("ThreadMessageHandler started\n");\r
 944     SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_BELOW_NORMAL);\r
 945     loop\r
 946     {\r
 947         // Poll the connected nodes for messages\r
 948         vector<CNode*> vNodesCopy;\r
 949         CRITICAL_BLOCK(cs_vNodes)\r
 950             vNodesCopy = vNodes;\r
 951         foreach(CNode* pnode, vNodesCopy)\r
 952         {\r
 953             pnode->AddRef();\r
 954 \r
 955             // Receive messages\r
 956             TRY_CRITICAL_BLOCK(pnode->cs_vRecv)\r
 957                 ProcessMessages(pnode);\r
 958             if (fShutdown)\r
 959                 return;\r
 960 \r
 961             // Send messages\r
 962             TRY_CRITICAL_BLOCK(pnode->cs_vSend)\r
 963                 SendMessages(pnode);\r
 964             if (fShutdown)\r
 965                 return;\r
 966 \r
 967             pnode->Release();\r
 968         }\r
 969 \r
 970         // Wait and allow messages to bunch up\r
 971         vnThreadsRunning[2]--;\r
 972         Sleep(100);\r
 973         vnThreadsRunning[2]++;\r
 974         if (fShutdown)\r
 975             return;\r
 976     }\r
 977 }\r
 978 \r
 979 \r
 980 \r
 981 \r
 982 \r
 983 \r
 984 \r
 985 \r
 986 \r
 987 bool BindListenPort(string& strError)\r
 988 {\r
 989     strError = "";\r
 990     int nOne = 1;\r
 991 \r
 992 #ifdef __WXMSW__\r
 993     // Initialize Windows Sockets\r
 994     WSADATA wsadata;\r
 995     int ret = WSAStartup(MAKEWORD(2,2), &wsadata);\r
 996     if (ret != NO_ERROR)\r
 997     {\r
 998         strError = strprintf("Error: TCP/IP socket library failed to start (WSAStartup returned error %d)", ret);\r
 999         printf("%s\n", strError.c_str());\r
1000         return false;\r
1001     }\r
1002 #endif\r
1003 \r
1004     // Create socket for listening for incoming connections\r
1005     hListenSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);\r
1006     if (hListenSocket == INVALID_SOCKET)\r
1007     {\r
1008         strError = strprintf("Error: Couldn't open socket for incoming connections (socket returned error %d)", WSAGetLastError());\r
1009         printf("%s\n", strError.c_str());\r
1010         return false;\r
1011     }\r
1012 \r
1013 #if defined(__BSD__) || defined(__WXOSX__)\r
1014     // Different way of disabling SIGPIPE on BSD\r
1015     setsockopt(hListenSocket, SOL_SOCKET, SO_NOSIGPIPE, (void*)&nOne, sizeof(int));\r
1016 #endif\r
1017 \r
1018 #ifndef __WXMSW__\r
1019     // Allow binding if the port is still in TIME_WAIT state after\r
1020     // the program was closed and restarted.  Not an issue on windows.\r
1021     setsockopt(hListenSocket, SOL_SOCKET, SO_REUSEADDR, (void*)&nOne, sizeof(int));\r
1022 #endif\r
1023 \r
1024 #ifdef __WXMSW__\r
1025     // Set to nonblocking, incoming connections will also inherit this\r
1026     if (ioctlsocket(hListenSocket, FIONBIO, (u_long*)&nOne) == SOCKET_ERROR)\r
1027 #else\r
1028     if (fcntl(hListenSocket, F_SETFL, O_NONBLOCK) == SOCKET_ERROR)\r
1029 #endif\r
1030     {\r
1031         strError = strprintf("Error: Couldn't set properties on socket for incoming connections (error %d)", WSAGetLastError());\r
1032         printf("%s\n", strError.c_str());\r
1033         return false;\r
1034     }\r
1035 \r
1036     // The sockaddr_in structure specifies the address family,\r
1037     // IP address, and port for the socket that is being bound\r
1038     struct sockaddr_in sockaddr;\r
1039     memset(&sockaddr, 0, sizeof(sockaddr));\r
1040     sockaddr.sin_family = AF_INET;\r
1041     sockaddr.sin_addr.s_addr = INADDR_ANY; // bind to all IPs on this computer\r
1042     sockaddr.sin_port = DEFAULT_PORT;\r
1043     if (::bind(hListenSocket, (struct sockaddr*)&sockaddr, sizeof(sockaddr)) == SOCKET_ERROR)\r
1044     {\r
1045         int nErr = WSAGetLastError();\r
1046         if (nErr == WSAEADDRINUSE)\r
1047             strError = strprintf("Unable to bind to port %d on this computer.  Bitcoin is probably already running.", ntohs(sockaddr.sin_port));\r
1048         else\r
1049             strError = strprintf("Error: Unable to bind to port %d on this computer (bind returned error %d)", ntohs(sockaddr.sin_port), nErr);\r
1050         printf("%s\n", strError.c_str());\r
1051         return false;\r
1052     }\r
1053     printf("Bound to port %d\n", ntohs(sockaddr.sin_port));\r
1054 \r
1055     // Listen for incoming connections\r
1056     if (listen(hListenSocket, SOMAXCONN) == SOCKET_ERROR)\r
1057     {\r
1058         strError = strprintf("Error: Listening for incoming connections failed (listen returned error %d)", WSAGetLastError());\r
1059         printf("%s\n", strError.c_str());\r
1060         return false;\r
1061     }\r
1062 \r
1063     return true;\r
1064 }\r
1065 \r
1066 bool StartNode(string& strError)\r
1067 {\r
1068     strError = "";\r
1069     if (pnodeLocalHost == NULL)\r
1070         pnodeLocalHost = new CNode(INVALID_SOCKET, CAddress("127.0.0.1", nLocalServices));\r
1071 \r
1072     // Get local host ip\r
1073     char pszHostName[255];\r
1074     if (gethostname(pszHostName, sizeof(pszHostName)) == SOCKET_ERROR)\r
1075     {\r
1076         strError = strprintf("Error: Unable to get IP address of this computer (gethostname returned error %d)", WSAGetLastError());\r
1077         printf("%s\n", strError.c_str());\r
1078         return false;\r
1079     }\r
1080     struct hostent* phostent = gethostbyname(pszHostName);\r
1081     if (!phostent)\r
1082     {\r
1083         strError = strprintf("Error: Unable to get IP address of this computer (gethostbyname returned error %d)", WSAGetLastError());\r
1084         printf("%s\n", strError.c_str());\r
1085         return false;\r
1086     }\r
1087 \r
1088     // Take the first IP that isn't loopback 127.x.x.x\r
1089     for (int i = 0; phostent->h_addr_list[i] != NULL; i++)\r
1090         printf("host ip %d: %s\n", i, CAddress(*(unsigned int*)phostent->h_addr_list[i]).ToStringIP().c_str());\r
1091     for (int i = 0; phostent->h_addr_list[i] != NULL; i++)\r
1092     {\r
1093         addrLocalHost = CAddress(*(unsigned int*)phostent->h_addr_list[i], DEFAULT_PORT, nLocalServices);\r
1094         if (addrLocalHost.IsValid() && addrLocalHost.GetByte(3) != 127)\r
1095             break;\r
1096     }\r
1097     printf("addrLocalHost = %s\n", addrLocalHost.ToString().c_str());\r
1098 \r
1099     // Get our external IP address for incoming connections\r
1100     if (fUseProxy)\r
1101     {\r
1102         // Proxies can't take incoming connections\r
1103         addrLocalHost.ip = CAddress("0.0.0.0").ip;\r
1104         printf("addrLocalHost = %s\n", addrLocalHost.ToString().c_str());\r
1105     }\r
1106     else\r
1107     {\r
1108         if (addrIncoming.ip)\r
1109             addrLocalHost.ip = addrIncoming.ip;\r
1110 \r
1111         if (GetMyExternalIP(addrLocalHost.ip))\r
1112         {\r
1113             addrIncoming = addrLocalHost;\r
1114             CWalletDB().WriteSetting("addrIncoming", addrIncoming);\r
1115         }\r
1116     }\r
1117 \r
1118     // Get addresses from IRC and advertise ours\r
1119     if (_beginthread(ThreadIRCSeed, 0, NULL) == -1)\r
1120         printf("Error: _beginthread(ThreadIRCSeed) failed\n");\r
1121 \r
1122     //\r
1123     // Start threads\r
1124     //\r
1125     if (_beginthread(ThreadSocketHandler, 0, NULL) == -1)\r
1126     {\r
1127         strError = "Error: _beginthread(ThreadSocketHandler) failed";\r
1128         printf("%s\n", strError.c_str());\r
1129         return false;\r
1130     }\r
1131 \r
1132     if (_beginthread(ThreadOpenConnections, 0, NULL) == -1)\r
1133     {\r
1134         strError = "Error: _beginthread(ThreadOpenConnections) failed";\r
1135         printf("%s\n", strError.c_str());\r
1136         return false;\r
1137     }\r
1138 \r
1139     if (_beginthread(ThreadMessageHandler, 0, NULL) == -1)\r
1140     {\r
1141         strError = "Error: _beginthread(ThreadMessageHandler) failed";\r
1142         printf("%s\n", strError.c_str());\r
1143         return false;\r
1144     }\r
1145 \r
1146     return true;\r
1147 }\r
1148 \r
1149 bool StopNode()\r
1150 {\r
1151     printf("StopNode()\n");\r
1152     fShutdown = true;\r
1153     nTransactionsUpdated++;\r
1154     int64 nStart = GetTime();\r
1155     while (vnThreadsRunning[0] > 0 || vnThreadsRunning[2] > 0 || vnThreadsRunning[3] > 0)\r
1156     {\r
1157         if (GetTime() - nStart > 15)\r
1158             break;\r
1159         Sleep(20);\r
1160     }\r
1161     if (vnThreadsRunning[0] > 0) printf("ThreadSocketHandler still running\n");\r
1162     if (vnThreadsRunning[1] > 0) printf("ThreadOpenConnections still running\n");\r
1163     if (vnThreadsRunning[2] > 0) printf("ThreadMessageHandler still running\n");\r
1164     if (vnThreadsRunning[3] > 0) printf("ThreadBitcoinMiner still running\n");\r
1165     while (vnThreadsRunning[2] > 0)\r
1166         Sleep(20);\r
1167     Sleep(50);\r
1168 \r
1169     return true;\r
1170 }\r