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