net.h

   1 // Copyright (c) 2009-2010 Satoshi Nakamoto
   2 // Distributed under the MIT/X11 software license, see the accompanying
   3 // file license.txt or http://www.opensource.org/licenses/mit-license.php.
   4
   5 class CMessageHeader;
   6 class CAddress;
   7 class CInv;
   8 class CRequestTracker;
   9 class CNode;
  10 class CBlockIndex;
  11 extern int nBestHeight;
  12
  13
  14
  15 static const unsigned short DEFAULT_PORT = 0x8d20; // htons(8333)
  16 static const unsigned int PUBLISH_HOPS = 5;
  17 enum
  18 {
  19     NODE_NETWORK = (1 << 0),
  20 };
  21
  22
  23
  24
  25 bool ConnectSocket(const CAddress& addrConnect, SOCKET& hSocketRet);
  26 bool GetMyExternalIP(unsigned int& ipRet);
  27 bool AddAddress(CAddress addr);
  28 void AddressCurrentlyConnected(const CAddress& addr);
  29 CNode* FindNode(unsigned int ip);
  30 CNode* ConnectNode(CAddress addrConnect, int64 nTimeout=0);
  31 void AbandonRequests(void (*fn)(void*, CDataStream&), void* param1);
  32 bool AnySubscribed(unsigned int nChannel);
  33 bool BindListenPort(string& strError=REF(string()));
  34 void StartNode(void* parg);
  35 bool StopNode();
  36
  37
  38
  39
  40
  41
  42
  43
  44 //
  45 // Message header
  46 //  (4) message start
  47 //  (12) command
  48 //  (4) size
  49 //  (4) checksum
  50
  51 // The message start string is designed to be unlikely to occur in normal data.
  52 // The characters are rarely used upper ascii, not valid as UTF-8, and produce
  53 // a large 4-byte int at any alignment.
  54 static const char pchMessageStart[4] = { 0xf9, 0xbe, 0xb4, 0xd9 };
  55
  56 class CMessageHeader
  57 {
  58 public:
  59     enum { COMMAND_SIZE=12 };
  60     char pchMessageStart[sizeof(::pchMessageStart)];
  61     char pchCommand[COMMAND_SIZE];
  62     unsigned int nMessageSize;
  63     unsigned int nChecksum;
  64
  65     CMessageHeader()
  66     {
  67         memcpy(pchMessageStart, ::pchMessageStart, sizeof(pchMessageStart));
  68         memset(pchCommand, 0, sizeof(pchCommand));
  69         pchCommand[1] = 1;
  70         nMessageSize = -1;
  71         nChecksum = 0;
  72     }
  73
  74     CMessageHeader(const char* pszCommand, unsigned int nMessageSizeIn)
  75     {
  76         memcpy(pchMessageStart, ::pchMessageStart, sizeof(pchMessageStart));
  77         strncpy(pchCommand, pszCommand, COMMAND_SIZE);
  78         nMessageSize = nMessageSizeIn;
  79         nChecksum = 0;
  80     }
  81
  82     IMPLEMENT_SERIALIZE
  83     (
  84         READWRITE(FLATDATA(pchMessageStart));
  85         READWRITE(FLATDATA(pchCommand));
  86         READWRITE(nMessageSize);
  87         if (nVersion >= 209)
  88             READWRITE(nChecksum);
  89     )
  90
  91     string GetCommand()
  92     {
  93         if (pchCommand[COMMAND_SIZE-1] == 0)
  94             return string(pchCommand, pchCommand + strlen(pchCommand));
  95         else
  96             return string(pchCommand, pchCommand + COMMAND_SIZE);
  97     }
  98
  99     bool IsValid()
 100     {
 101         // Check start string
 102         if (memcmp(pchMessageStart, ::pchMessageStart, sizeof(pchMessageStart)) != 0)
 103             return false;
 104
 105         // Check the command string for errors
 106         for (char* p1 = pchCommand; p1 < pchCommand + COMMAND_SIZE; p1++)
 107         {
 108             if (*p1 == 0)
 109             {
 110                 // Must be all zeros after the first zero
 111                 for (; p1 < pchCommand + COMMAND_SIZE; p1++)
 112                     if (*p1 != 0)
 113                         return false;
 114             }
 115             else if (*p1 < ' ' || *p1 > 0x7E)
 116                 return false;
 117         }
 118
 119         // Message size
 120         if (nMessageSize > 0x10000000)
 121         {
 122             printf("CMessageHeader::IsValid() : nMessageSize too large %u\n", nMessageSize);
 123             return false;
 124         }
 125
 126         return true;
 127     }
 128 };
 129
 130
 131
 132
 133
 134
 135 static const unsigned char pchIPv4[12] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff };
 136
 137 class CAddress
 138 {
 139 public:
 140     uint64 nServices;
 141     unsigned char pchReserved[12];
 142     unsigned int ip;
 143     unsigned short port;
 144
 145     // disk only
 146     unsigned int nTime;
 147
 148     // memory only
 149     unsigned int nLastTry;
 150
 151     CAddress()
 152     {
 153         Init();
 154     }
 155
 156     CAddress(unsigned int ipIn, unsigned short portIn=DEFAULT_PORT, uint64 nServicesIn=NODE_NETWORK)
 157     {
 158         Init();
 159         ip = ipIn;
 160         port = portIn;
 161         nServices = nServicesIn;
 162     }
 163
 164     explicit CAddress(const struct sockaddr_in& sockaddr, uint64 nServicesIn=NODE_NETWORK)
 165     {
 166         Init();
 167         ip = sockaddr.sin_addr.s_addr;
 168         port = sockaddr.sin_port;
 169         nServices = nServicesIn;
 170     }
 171
 172     explicit CAddress(const char* pszIn, uint64 nServicesIn=NODE_NETWORK)
 173     {
 174         Init();
 175         SetAddress(pszIn);
 176         nServices = nServicesIn;
 177     }
 178
 179     explicit CAddress(string strIn, uint64 nServicesIn=NODE_NETWORK)
 180     {
 181         Init();
 182         SetAddress(strIn.c_str());
 183         nServices = nServicesIn;
 184     }
 185
 186     void Init()
 187     {
 188         nServices = NODE_NETWORK;
 189         memcpy(pchReserved, pchIPv4, sizeof(pchReserved));
 190         ip = INADDR_NONE;
 191         port = DEFAULT_PORT;
 192         nTime = GetAdjustedTime();
 193         nLastTry = 0;
 194     }
 195
 196     bool SetAddress(const char* pszIn)
 197     {
 198         ip = INADDR_NONE;
 199         port = DEFAULT_PORT;
 200         char psz[100];
 201         strlcpy(psz, pszIn, sizeof(psz));
 202         unsigned int a=0, b=0, c=0, d=0, e=0;
 203         if (sscanf(psz, "%u.%u.%u.%u:%u", &a, &b, &c, &d, &e) < 4)
 204             return false;
 205         char* pszPort = strchr(psz, ':');
 206         if (pszPort)
 207         {
 208             *pszPort++ = '\0';
 209             port = htons(atoi(pszPort));
 210             if (atoi(pszPort) < 0 || atoi(pszPort) > USHRT_MAX)
 211                 port = htons(USHRT_MAX);
 212         }
 213         ip = inet_addr(psz);
 214         return IsValid();
 215     }
 216
 217     bool SetAddress(string strIn)
 218     {
 219         return SetAddress(strIn.c_str());
 220     }
 221
 222     IMPLEMENT_SERIALIZE
 223     (
 224         if (nType & SER_DISK)
 225         {
 226             READWRITE(nVersion);
 227             READWRITE(nTime);
 228         }
 229         READWRITE(nServices);
 230         READWRITE(FLATDATA(pchReserved)); // for IPv6
 231         READWRITE(ip);
 232         READWRITE(port);
 233     )
 234
 235     friend inline bool operator==(const CAddress& a, const CAddress& b)
 236     {
 237         return (memcmp(a.pchReserved, b.pchReserved, sizeof(a.pchReserved)) == 0 &&
 238                 a.ip   == b.ip &&
 239                 a.port == b.port);
 240     }
 241
 242     friend inline bool operator!=(const CAddress& a, const CAddress& b)
 243     {
 244         return (!(a == b));
 245     }
 246
 247     friend inline bool operator<(const CAddress& a, const CAddress& b)
 248     {
 249         int ret = memcmp(a.pchReserved, b.pchReserved, sizeof(a.pchReserved));
 250         if (ret < 0)
 251             return true;
 252         else if (ret == 0)
 253         {
 254             if (ntohl(a.ip) < ntohl(b.ip))
 255                 return true;
 256             else if (a.ip == b.ip)
 257                 return ntohs(a.port) < ntohs(b.port);
 258         }
 259         return false;
 260     }
 261
 262     vector<unsigned char> GetKey() const
 263     {
 264         CDataStream ss;
 265         ss.reserve(18);
 266         ss << FLATDATA(pchReserved) << ip << port;
 267
 268         #if defined(_MSC_VER) && _MSC_VER < 1300
 269         return vector<unsigned char>((unsigned char*)&ss.begin()[0], (unsigned char*)&ss.end()[0]);
 270         #else
 271         return vector<unsigned char>(ss.begin(), ss.end());
 272         #endif
 273     }
 274
 275     struct sockaddr_in GetSockAddr() const
 276     {
 277         struct sockaddr_in sockaddr;
 278         memset(&sockaddr, 0, sizeof(sockaddr));
 279         sockaddr.sin_family = AF_INET;
 280         sockaddr.sin_addr.s_addr = ip;
 281         sockaddr.sin_port = port;
 282         return sockaddr;
 283     }
 284
 285     bool IsIPv4() const
 286     {
 287         return (memcmp(pchReserved, pchIPv4, sizeof(pchIPv4)) == 0);
 288     }
 289
 290     bool IsRoutable() const
 291     {
 292         return IsValid() &&
 293             !(GetByte(3) == 10 ||
 294               (GetByte(3) == 192 && GetByte(2) == 168) ||
 295               GetByte(3) == 127 ||
 296               GetByte(3) == 0);
 297     }
 298
 299     bool IsValid() const
 300     {
 301         // Clean up 3-byte shifted addresses caused by garbage in size field
 302         // of addr messages from versions before 0.2.9 checksum.
 303         // Two consecutive addr messages look like this:
 304         // header20 vectorlen3 addr26 addr26 addr26 header20 vectorlen3 addr26 addr26 addr26...
 305         // so if the first length field is garbled, it reads the second batch
 306         // of addr misaligned by 3 bytes.
 307         if (memcmp(pchReserved, pchIPv4+3, sizeof(pchIPv4)-3) == 0)
 308             return false;
 309
 310         return (ip != 0 && ip != INADDR_NONE && port != htons(USHRT_MAX));
 311     }
 312
 313     unsigned char GetByte(int n) const
 314     {
 315         return ((unsigned char*)&ip)[3-n];
 316     }
 317
 318     string ToStringIPPort() const
 319     {
 320         return strprintf("%u.%u.%u.%u:%u", GetByte(3), GetByte(2), GetByte(1), GetByte(0), ntohs(port));
 321     }
 322
 323     string ToStringIP() const
 324     {
 325         return strprintf("%u.%u.%u.%u", GetByte(3), GetByte(2), GetByte(1), GetByte(0));
 326     }
 327
 328     string ToStringPort() const
 329     {
 330         return strprintf("%u", ntohs(port));
 331     }
 332
 333     string ToStringLog() const
 334     {
 335         return "";
 336     }
 337
 338     string ToString() const
 339     {
 340         return strprintf("%u.%u.%u.%u:%u", GetByte(3), GetByte(2), GetByte(1), GetByte(0), ntohs(port));
 341     }
 342
 343     void print() const
 344     {
 345         printf("CAddress(%s)\n", ToString().c_str());
 346     }
 347 };
 348
 349
 350
 351
 352
 353
 354
 355 enum
 356 {
 357     MSG_TX = 1,
 358     MSG_BLOCK,
 359 };
 360
 361 static const char* ppszTypeName[] =
 362 {
 363     "ERROR",
 364     "tx",
 365     "block",
 366 };
 367
 368 class CInv
 369 {
 370 public:
 371     int type;
 372     uint256 hash;
 373
 374     CInv()
 375     {
 376         type = 0;
 377         hash = 0;
 378     }
 379
 380     CInv(int typeIn, const uint256& hashIn)
 381     {
 382         type = typeIn;
 383         hash = hashIn;
 384     }
 385
 386     CInv(const string& strType, const uint256& hashIn)
 387     {
 388         int i;
 389         for (i = 1; i < ARRAYLEN(ppszTypeName); i++)
 390         {
 391             if (strType == ppszTypeName[i])
 392             {
 393                 type = i;
 394                 break;
 395             }
 396         }
 397         if (i == ARRAYLEN(ppszTypeName))
 398             throw std::out_of_range(strprintf("CInv::CInv(string, uint256) : unknown type '%s'", strType.c_str()));
 399         hash = hashIn;
 400     }
 401
 402     IMPLEMENT_SERIALIZE
 403     (
 404         READWRITE(type);
 405         READWRITE(hash);
 406     )
 407
 408     friend inline bool operator<(const CInv& a, const CInv& b)
 409     {
 410         return (a.type < b.type || (a.type == b.type && a.hash < b.hash));
 411     }
 412
 413     bool IsKnownType() const
 414     {
 415         return (type >= 1 && type < ARRAYLEN(ppszTypeName));
 416     }
 417
 418     const char* GetCommand() const
 419     {
 420         if (!IsKnownType())
 421             throw std::out_of_range(strprintf("CInv::GetCommand() : type=% unknown type", type));
 422         return ppszTypeName[type];
 423     }
 424
 425     string ToString() const
 426     {
 427         return strprintf("%s %s", GetCommand(), hash.ToString().substr(0,16).c_str());
 428     }
 429
 430     void print() const
 431     {
 432         printf("CInv(%s)\n", ToString().c_str());
 433     }
 434 };
 435
 436
 437
 438
 439
 440 class CRequestTracker
 441 {
 442 public:
 443     void (*fn)(void*, CDataStream&);
 444     void* param1;
 445
 446     explicit CRequestTracker(void (*fnIn)(void*, CDataStream&)=NULL, void* param1In=NULL)
 447     {
 448         fn = fnIn;
 449         param1 = param1In;
 450     }
 451
 452     bool IsNull()
 453     {
 454         return fn == NULL;
 455     }
 456 };
 457
 458
 459
 460
 461
 462 extern bool fClient;
 463 extern uint64 nLocalServices;
 464 extern CAddress addrLocalHost;
 465 extern CNode* pnodeLocalHost;
 466 extern uint64 nLocalHostNonce;
 467 extern array<int, 10> vnThreadsRunning;
 468 extern SOCKET hListenSocket;
 469 extern int64 nThreadSocketHandlerHeartbeat;
 470
 471 extern vector<CNode*> vNodes;
 472 extern CCriticalSection cs_vNodes;
 473 extern map<vector<unsigned char>, CAddress> mapAddresses;
 474 extern CCriticalSection cs_mapAddresses;
 475 extern map<CInv, CDataStream> mapRelay;
 476 extern deque<pair<int64, CInv> > vRelayExpiration;
 477 extern CCriticalSection cs_mapRelay;
 478 extern map<CInv, int64> mapAlreadyAskedFor;
 479
 480 // Settings
 481 extern int fUseProxy;
 482 extern CAddress addrProxy;
 483
 484
 485
 486
 487
 488
 489 class CNode
 490 {
 491 public:
 492     // socket
 493     uint64 nServices;
 494     SOCKET hSocket;
 495     CDataStream vSend;
 496     CDataStream vRecv;
 497     CCriticalSection cs_vSend;
 498     CCriticalSection cs_vRecv;
 499     int64 nLastSend;
 500     int64 nLastRecv;
 501     int64 nLastSendEmpty;
 502     int64 nTimeConnected;
 503     unsigned int nHeaderStart;
 504     unsigned int nMessageStart;
 505     CAddress addr;
 506     int nVersion;
 507     bool fClient;
 508     bool fInbound;
 509     bool fNetworkNode;
 510     bool fSuccessfullyConnected;
 511     bool fDisconnect;
 512 protected:
 513     int nRefCount;
 514 public:
 515     int64 nReleaseTime;
 516     map<uint256, CRequestTracker> mapRequests;
 517     CCriticalSection cs_mapRequests;
 518     uint256 hashContinue;
 519     CBlockIndex* pindexLastGetBlocksBegin;
 520     uint256 hashLastGetBlocksEnd;
 521     int nStartingHeight;
 522
 523     // flood
 524     vector<CAddress> vAddrToSend;
 525     set<CAddress> setAddrKnown;
 526     bool fGetAddr;
 527
 528     // inventory based relay
 529     set<CInv> setInventoryKnown;
 530     vector<CInv> vInventoryToSend;
 531     CCriticalSection cs_inventory;
 532     multimap<int64, CInv> mapAskFor;
 533
 534     // publish and subscription
 535     vector<char> vfSubscribe;
 536
 537
 538     CNode(SOCKET hSocketIn, CAddress addrIn, bool fInboundIn=false)
 539     {
 540         nServices = 0;
 541         hSocket = hSocketIn;
 542         vSend.SetType(SER_NETWORK);
 543         vSend.SetVersion(0);
 544         vRecv.SetType(SER_NETWORK);
 545         vRecv.SetVersion(0);
 546         // Version 0.2 obsoletes 20 Feb 2012
 547         if (GetTime() > 1329696000)
 548         {
 549             vSend.SetVersion(209);
 550             vRecv.SetVersion(209);
 551         }
 552         nLastSend = 0;
 553         nLastRecv = 0;
 554         nLastSendEmpty = GetTime();
 555         nTimeConnected = GetTime();
 556         nHeaderStart = -1;
 557         nMessageStart = -1;
 558         addr = addrIn;
 559         nVersion = 0;
 560         fClient = false; // set by version message
 561         fInbound = fInboundIn;
 562         fNetworkNode = false;
 563         fSuccessfullyConnected = false;
 564         fDisconnect = false;
 565         nRefCount = 0;
 566         nReleaseTime = 0;
 567         hashContinue = 0;
 568         pindexLastGetBlocksBegin = 0;
 569         hashLastGetBlocksEnd = 0;
 570         nStartingHeight = -1;
 571         fGetAddr = false;
 572         vfSubscribe.assign(256, false);
 573
 574         // Push a version message
 575         /// when NTP implemented, change to just nTime = GetAdjustedTime()
 576         int64 nTime = (fInbound ? GetAdjustedTime() : GetTime());
 577         CAddress addrYou = (fUseProxy ? CAddress("0.0.0.0") : addr);
 578         CAddress addrMe = (fUseProxy ? CAddress("0.0.0.0") : addrLocalHost);
 579         RAND_bytes((unsigned char*)&nLocalHostNonce, sizeof(nLocalHostNonce));
 580         PushMessage("version", VERSION, nLocalServices, nTime, addrYou, addrMe,
 581                     nLocalHostNonce, string(pszSubVer), nBestHeight);
 582     }
 583
 584     ~CNode()
 585     {
 586         if (hSocket != INVALID_SOCKET)
 587         {
 588             closesocket(hSocket);
 589             hSocket = INVALID_SOCKET;
 590         }
 591     }
 592
 593 private:
 594     CNode(const CNode&);
 595     void operator=(const CNode&);
 596 public:
 597
 598
 599     int GetRefCount()
 600     {
 601         return max(nRefCount, 0) + (GetTime() < nReleaseTime ? 1 : 0);
 602     }
 603
 604     CNode* AddRef(int64 nTimeout=0)
 605     {
 606         if (nTimeout != 0)
 607             nReleaseTime = max(nReleaseTime, GetTime() + nTimeout);
 608         else
 609             nRefCount++;
 610         return this;
 611     }
 612
 613     void Release()
 614     {
 615         nRefCount--;
 616     }
 617
 618
 619
 620     void AddAddressKnown(const CAddress& addr)
 621     {
 622         setAddrKnown.insert(addr);
 623     }
 624
 625     void PushAddress(const CAddress& addr)
 626     {
 627         // Known checking here is only to save space from duplicates.
 628         // SendMessages will filter it again for knowns that were added
 629         // after addresses were pushed.
 630         if (addr.IsValid() && !setAddrKnown.count(addr))
 631             vAddrToSend.push_back(addr);
 632     }
 633
 634
 635     void AddInventoryKnown(const CInv& inv)
 636     {
 637         CRITICAL_BLOCK(cs_inventory)
 638             setInventoryKnown.insert(inv);
 639     }
 640
 641     void PushInventory(const CInv& inv)
 642     {
 643         CRITICAL_BLOCK(cs_inventory)
 644             if (!setInventoryKnown.count(inv))
 645                 vInventoryToSend.push_back(inv);
 646     }
 647
 648     void AskFor(const CInv& inv)
 649     {
 650         // We're using mapAskFor as a priority queue,
 651         // the key is the earliest time the request can be sent
 652         int64& nRequestTime = mapAlreadyAskedFor[inv];
 653         printf("askfor %s   %"PRI64d"\n", inv.ToString().c_str(), nRequestTime);
 654
 655         // Make sure not to reuse time indexes to keep things in the same order
 656         int64 nNow = (GetTime() - 1) * 1000000;
 657         static int64 nLastTime;
 658         nLastTime = nNow = max(nNow, ++nLastTime);
 659
 660         // Each retry is 2 minutes after the last
 661         nRequestTime = max(nRequestTime + 2 * 60 * 1000000, nNow);
 662         mapAskFor.insert(make_pair(nRequestTime, inv));
 663     }
 664
 665
 666
 667     void BeginMessage(const char* pszCommand)
 668     {
 669         cs_vSend.Enter();
 670         if (nHeaderStart != -1)
 671             AbortMessage();
 672         nHeaderStart = vSend.size();
 673         vSend << CMessageHeader(pszCommand, 0);
 674         nMessageStart = vSend.size();
 675         if (fDebug)
 676             printf("%s ", DateTimeStrFormat("%x %H:%M:%S", GetTime()).c_str());
 677         printf("sending: %s ", pszCommand);
 678     }
 679
 680     void AbortMessage()
 681     {
 682         if (nHeaderStart == -1)
 683             return;
 684         vSend.resize(nHeaderStart);
 685         nHeaderStart = -1;
 686         nMessageStart = -1;
 687         cs_vSend.Leave();
 688         printf("(aborted)\n");
 689     }
 690
 691     void EndMessage()
 692     {
 693         if (mapArgs.count("-dropmessagestest") && GetRand(atoi(mapArgs["-dropmessagestest"])) == 0)
 694         {
 695             printf("dropmessages DROPPING SEND MESSAGE\n");
 696             AbortMessage();
 697             return;
 698         }
 699
 700         if (nHeaderStart == -1)
 701             return;
 702
 703         // Set the size
 704         unsigned int nSize = vSend.size() - nMessageStart;
 705         memcpy((char*)&vSend[nHeaderStart] + offsetof(CMessageHeader, nMessageSize), &nSize, sizeof(nSize));
 706
 707         // Set the checksum
 708         if (vSend.GetVersion() >= 209)
 709         {
 710             uint256 hash = Hash(vSend.begin() + nMessageStart, vSend.end());
 711             unsigned int nChecksum = 0;
 712             memcpy(&nChecksum, &hash, sizeof(nChecksum));
 713             assert(nMessageStart - nHeaderStart >= offsetof(CMessageHeader, nChecksum) + sizeof(nChecksum));
 714             memcpy((char*)&vSend[nHeaderStart] + offsetof(CMessageHeader, nChecksum), &nChecksum, sizeof(nChecksum));
 715         }
 716
 717         printf("(%d bytes) ", nSize);
 718         printf("\n");
 719
 720         nHeaderStart = -1;
 721         nMessageStart = -1;
 722         cs_vSend.Leave();
 723     }
 724
 725     void EndMessageAbortIfEmpty()
 726     {
 727         if (nHeaderStart == -1)
 728             return;
 729         int nSize = vSend.size() - nMessageStart;
 730         if (nSize > 0)
 731             EndMessage();
 732         else
 733             AbortMessage();
 734     }
 735
 736     const char* GetMessageCommand() const
 737     {
 738         if (nHeaderStart == -1)
 739             return "";
 740         return &vSend[nHeaderStart] + offsetof(CMessageHeader, pchCommand);
 741     }
 742
 743
 744
 745
 746     void PushMessage(const char* pszCommand)
 747     {
 748         try
 749         {
 750             BeginMessage(pszCommand);
 751             EndMessage();
 752         }
 753         catch (...)
 754         {
 755             AbortMessage();
 756             throw;
 757         }
 758     }
 759
 760     template<typename T1>
 761     void PushMessage(const char* pszCommand, const T1& a1)
 762     {
 763         try
 764         {
 765             BeginMessage(pszCommand);
 766             vSend << a1;
 767             EndMessage();
 768         }
 769         catch (...)
 770         {
 771             AbortMessage();
 772             throw;
 773         }
 774     }
 775
 776     template<typename T1, typename T2>
 777     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2)
 778     {
 779         try
 780         {
 781             BeginMessage(pszCommand);
 782             vSend << a1 << a2;
 783             EndMessage();
 784         }
 785         catch (...)
 786         {
 787             AbortMessage();
 788             throw;
 789         }
 790     }
 791
 792     template<typename T1, typename T2, typename T3>
 793     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3)
 794     {
 795         try
 796         {
 797             BeginMessage(pszCommand);
 798             vSend << a1 << a2 << a3;
 799             EndMessage();
 800         }
 801         catch (...)
 802         {
 803             AbortMessage();
 804             throw;
 805         }
 806     }
 807
 808     template<typename T1, typename T2, typename T3, typename T4>
 809     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4)
 810     {
 811         try
 812         {
 813             BeginMessage(pszCommand);
 814             vSend << a1 << a2 << a3 << a4;
 815             EndMessage();
 816         }
 817         catch (...)
 818         {
 819             AbortMessage();
 820             throw;
 821         }
 822     }
 823
 824     template<typename T1, typename T2, typename T3, typename T4, typename T5>
 825     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4, const T5& a5)
 826     {
 827         try
 828         {
 829             BeginMessage(pszCommand);
 830             vSend << a1 << a2 << a3 << a4 << a5;
 831             EndMessage();
 832         }
 833         catch (...)
 834         {
 835             AbortMessage();
 836             throw;
 837         }
 838     }
 839
 840     template<typename T1, typename T2, typename T3, typename T4, typename T5, typename T6>
 841     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4, const T5& a5, const T6& a6)
 842     {
 843         try
 844         {
 845             BeginMessage(pszCommand);
 846             vSend << a1 << a2 << a3 << a4 << a5 << a6;
 847             EndMessage();
 848         }
 849         catch (...)
 850         {
 851             AbortMessage();
 852             throw;
 853         }
 854     }
 855
 856     template<typename T1, typename T2, typename T3, typename T4, typename T5, typename T6, typename T7>
 857     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4, const T5& a5, const T6& a6, const T7& a7)
 858     {
 859         try
 860         {
 861             BeginMessage(pszCommand);
 862             vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7;
 863             EndMessage();
 864         }
 865         catch (...)
 866         {
 867             AbortMessage();
 868             throw;
 869         }
 870     }
 871
 872     template<typename T1, typename T2, typename T3, typename T4, typename T5, typename T6, typename T7, typename T8>
 873     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4, const T5& a5, const T6& a6, const T7& a7, const T8& a8)
 874     {
 875         try
 876         {
 877             BeginMessage(pszCommand);
 878             vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7 << a8;
 879             EndMessage();
 880         }
 881         catch (...)
 882         {
 883             AbortMessage();
 884             throw;
 885         }
 886     }
 887
 888     template<typename T1, typename T2, typename T3, typename T4, typename T5, typename T6, typename T7, typename T8, typename T9>
 889     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4, const T5& a5, const T6& a6, const T7& a7, const T8& a8, const T9& a9)
 890     {
 891         try
 892         {
 893             BeginMessage(pszCommand);
 894             vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7 << a8 << a9;
 895             EndMessage();
 896         }
 897         catch (...)
 898         {
 899             AbortMessage();
 900             throw;
 901         }
 902     }
 903
 904
 905     void PushRequest(const char* pszCommand,
 906                      void (*fn)(void*, CDataStream&), void* param1)
 907     {
 908         uint256 hashReply;
 909         RAND_bytes((unsigned char*)&hashReply, sizeof(hashReply));
 910
 911         CRITICAL_BLOCK(cs_mapRequests)
 912             mapRequests[hashReply] = CRequestTracker(fn, param1);
 913
 914         PushMessage(pszCommand, hashReply);
 915     }
 916
 917     template<typename T1>
 918     void PushRequest(const char* pszCommand, const T1& a1,
 919                      void (*fn)(void*, CDataStream&), void* param1)
 920     {
 921         uint256 hashReply;
 922         RAND_bytes((unsigned char*)&hashReply, sizeof(hashReply));
 923
 924         CRITICAL_BLOCK(cs_mapRequests)
 925             mapRequests[hashReply] = CRequestTracker(fn, param1);
 926
 927         PushMessage(pszCommand, hashReply, a1);
 928     }
 929
 930     template<typename T1, typename T2>
 931     void PushRequest(const char* pszCommand, const T1& a1, const T2& a2,
 932                      void (*fn)(void*, CDataStream&), void* param1)
 933     {
 934         uint256 hashReply;
 935         RAND_bytes((unsigned char*)&hashReply, sizeof(hashReply));
 936
 937         CRITICAL_BLOCK(cs_mapRequests)
 938             mapRequests[hashReply] = CRequestTracker(fn, param1);
 939
 940         PushMessage(pszCommand, hashReply, a1, a2);
 941     }
 942
 943
 944
 945     void PushGetBlocks(CBlockIndex* pindexBegin, uint256 hashEnd);
 946     bool IsSubscribed(unsigned int nChannel);
 947     void Subscribe(unsigned int nChannel, unsigned int nHops=0);
 948     void CancelSubscribe(unsigned int nChannel);
 949     void CloseSocketDisconnect();
 950     void Cleanup();
 951 };
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962 inline void RelayInventory(const CInv& inv)
 963 {
 964     // Put on lists to offer to the other nodes
 965     CRITICAL_BLOCK(cs_vNodes)
 966         foreach(CNode* pnode, vNodes)
 967             pnode->PushInventory(inv);
 968 }
 969
 970 template<typename T>
 971 void RelayMessage(const CInv& inv, const T& a)
 972 {
 973     CDataStream ss(SER_NETWORK);
 974     ss.reserve(10000);
 975     ss << a;
 976     RelayMessage(inv, ss);
 977 }
 978
 979 template<>
 980 inline void RelayMessage<>(const CInv& inv, const CDataStream& ss)
 981 {
 982     CRITICAL_BLOCK(cs_mapRelay)
 983     {
 984         // Expire old relay messages
 985         while (!vRelayExpiration.empty() && vRelayExpiration.front().first < GetTime())
 986         {
 987             mapRelay.erase(vRelayExpiration.front().second);
 988             vRelayExpiration.pop_front();
 989         }
 990
 991         // Save original serialized message so newer versions are preserved
 992         mapRelay[inv] = ss;
 993         vRelayExpiration.push_back(make_pair(GetTime() + 15 * 60, inv));
 994     }
 995
 996     RelayInventory(inv);
 997 }
 998
 999
1000
1001
1002
1003
1004
1005
1006 //
1007 // Templates for the publish and subscription system.
1008 // The object being published as T& obj needs to have:
1009 //   a set<unsigned int> setSources member
1010 //   specializations of AdvertInsert and AdvertErase
1011 // Currently implemented for CTable and CProduct.
1012 //
1013
1014 template<typename T>
1015 void AdvertStartPublish(CNode* pfrom, unsigned int nChannel, unsigned int nHops, T& obj)
1016 {
1017     // Add to sources
1018     obj.setSources.insert(pfrom->addr.ip);
1019
1020     if (!AdvertInsert(obj))
1021         return;
1022
1023     // Relay
1024     CRITICAL_BLOCK(cs_vNodes)
1025         foreach(CNode* pnode, vNodes)
1026             if (pnode != pfrom && (nHops < PUBLISH_HOPS || pnode->IsSubscribed(nChannel)))
1027                 pnode->PushMessage("publish", nChannel, nHops, obj);
1028 }
1029
1030 template<typename T>
1031 void AdvertStopPublish(CNode* pfrom, unsigned int nChannel, unsigned int nHops, T& obj)
1032 {
1033     uint256 hash = obj.GetHash();
1034
1035     CRITICAL_BLOCK(cs_vNodes)
1036         foreach(CNode* pnode, vNodes)
1037             if (pnode != pfrom && (nHops < PUBLISH_HOPS || pnode->IsSubscribed(nChannel)))
1038                 pnode->PushMessage("pub-cancel", nChannel, nHops, hash);
1039
1040     AdvertErase(obj);
1041 }
1042
1043 template<typename T>
1044 void AdvertRemoveSource(CNode* pfrom, unsigned int nChannel, unsigned int nHops, T& obj)
1045 {
1046     // Remove a source
1047     obj.setSources.erase(pfrom->addr.ip);
1048
1049     // If no longer supported by any sources, cancel it
1050     if (obj.setSources.empty())
1051         AdvertStopPublish(pfrom, nChannel, nHops, obj);
1052 }