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,20).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     string strSubVer;
 508     bool fClient;
 509     bool fInbound;
 510     bool fNetworkNode;
 511     bool fSuccessfullyConnected;
 512     bool fDisconnect;
 513 protected:
 514     int nRefCount;
 515 public:
 516     int64 nReleaseTime;
 517     map<uint256, CRequestTracker> mapRequests;
 518     CCriticalSection cs_mapRequests;
 519     uint256 hashContinue;
 520     CBlockIndex* pindexLastGetBlocksBegin;
 521     uint256 hashLastGetBlocksEnd;
 522     int nStartingHeight;
 523
 524     // flood relay
 525     vector<CAddress> vAddrToSend;
 526     set<CAddress> setAddrKnown;
 527     bool fGetAddr;
 528     set<uint256> setKnown;
 529
 530     // inventory based relay
 531     set<CInv> setInventoryKnown;
 532     vector<CInv> vInventoryToSend;
 533     CCriticalSection cs_inventory;
 534     multimap<int64, CInv> mapAskFor;
 535
 536     // publish and subscription
 537     vector<char> vfSubscribe;
 538
 539
 540     CNode(SOCKET hSocketIn, CAddress addrIn, bool fInboundIn=false)
 541     {
 542         nServices = 0;
 543         hSocket = hSocketIn;
 544         vSend.SetType(SER_NETWORK);
 545         vSend.SetVersion(0);
 546         vRecv.SetType(SER_NETWORK);
 547         vRecv.SetVersion(0);
 548         // Version 0.2 obsoletes 20 Feb 2012
 549         if (GetTime() > 1329696000)
 550         {
 551             vSend.SetVersion(209);
 552             vRecv.SetVersion(209);
 553         }
 554         nLastSend = 0;
 555         nLastRecv = 0;
 556         nLastSendEmpty = GetTime();
 557         nTimeConnected = GetTime();
 558         nHeaderStart = -1;
 559         nMessageStart = -1;
 560         addr = addrIn;
 561         nVersion = 0;
 562         strSubVer = "";
 563         fClient = false; // set by version message
 564         fInbound = fInboundIn;
 565         fNetworkNode = false;
 566         fSuccessfullyConnected = false;
 567         fDisconnect = false;
 568         nRefCount = 0;
 569         nReleaseTime = 0;
 570         hashContinue = 0;
 571         pindexLastGetBlocksBegin = 0;
 572         hashLastGetBlocksEnd = 0;
 573         nStartingHeight = -1;
 574         fGetAddr = false;
 575         vfSubscribe.assign(256, false);
 576
 577         // Push a version message
 578         /// when NTP implemented, change to just nTime = GetAdjustedTime()
 579         int64 nTime = (fInbound ? GetAdjustedTime() : GetTime());
 580         CAddress addrYou = (fUseProxy ? CAddress("0.0.0.0") : addr);
 581         CAddress addrMe = (fUseProxy ? CAddress("0.0.0.0") : addrLocalHost);
 582         RAND_bytes((unsigned char*)&nLocalHostNonce, sizeof(nLocalHostNonce));
 583         PushMessage("version", VERSION, nLocalServices, nTime, addrYou, addrMe,
 584                     nLocalHostNonce, string(pszSubVer), nBestHeight);
 585     }
 586
 587     ~CNode()
 588     {
 589         if (hSocket != INVALID_SOCKET)
 590         {
 591             closesocket(hSocket);
 592             hSocket = INVALID_SOCKET;
 593         }
 594     }
 595
 596 private:
 597     CNode(const CNode&);
 598     void operator=(const CNode&);
 599 public:
 600
 601
 602     int GetRefCount()
 603     {
 604         return max(nRefCount, 0) + (GetTime() < nReleaseTime ? 1 : 0);
 605     }
 606
 607     CNode* AddRef(int64 nTimeout=0)
 608     {
 609         if (nTimeout != 0)
 610             nReleaseTime = max(nReleaseTime, GetTime() + nTimeout);
 611         else
 612             nRefCount++;
 613         return this;
 614     }
 615
 616     void Release()
 617     {
 618         nRefCount--;
 619     }
 620
 621
 622
 623     void AddAddressKnown(const CAddress& addr)
 624     {
 625         setAddrKnown.insert(addr);
 626     }
 627
 628     void PushAddress(const CAddress& addr)
 629     {
 630         // Known checking here is only to save space from duplicates.
 631         // SendMessages will filter it again for knowns that were added
 632         // after addresses were pushed.
 633         if (addr.IsValid() && !setAddrKnown.count(addr))
 634             vAddrToSend.push_back(addr);
 635     }
 636
 637
 638     void AddInventoryKnown(const CInv& inv)
 639     {
 640         CRITICAL_BLOCK(cs_inventory)
 641             setInventoryKnown.insert(inv);
 642     }
 643
 644     void PushInventory(const CInv& inv)
 645     {
 646         CRITICAL_BLOCK(cs_inventory)
 647             if (!setInventoryKnown.count(inv))
 648                 vInventoryToSend.push_back(inv);
 649     }
 650
 651     void AskFor(const CInv& inv)
 652     {
 653         // We're using mapAskFor as a priority queue,
 654         // the key is the earliest time the request can be sent
 655         int64& nRequestTime = mapAlreadyAskedFor[inv];
 656         printf("askfor %s   %"PRI64d"\n", inv.ToString().c_str(), nRequestTime);
 657
 658         // Make sure not to reuse time indexes to keep things in the same order
 659         int64 nNow = (GetTime() - 1) * 1000000;
 660         static int64 nLastTime;
 661         nLastTime = nNow = max(nNow, ++nLastTime);
 662
 663         // Each retry is 2 minutes after the last
 664         nRequestTime = max(nRequestTime + 2 * 60 * 1000000, nNow);
 665         mapAskFor.insert(make_pair(nRequestTime, inv));
 666     }
 667
 668
 669
 670     void BeginMessage(const char* pszCommand)
 671     {
 672         cs_vSend.Enter();
 673         if (nHeaderStart != -1)
 674             AbortMessage();
 675         nHeaderStart = vSend.size();
 676         vSend << CMessageHeader(pszCommand, 0);
 677         nMessageStart = vSend.size();
 678         if (fDebug)
 679             printf("%s ", DateTimeStrFormat("%x %H:%M:%S", GetTime()).c_str());
 680         printf("sending: %s ", pszCommand);
 681     }
 682
 683     void AbortMessage()
 684     {
 685         if (nHeaderStart == -1)
 686             return;
 687         vSend.resize(nHeaderStart);
 688         nHeaderStart = -1;
 689         nMessageStart = -1;
 690         cs_vSend.Leave();
 691         printf("(aborted)\n");
 692     }
 693
 694     void EndMessage()
 695     {
 696         if (mapArgs.count("-dropmessagestest") && GetRand(atoi(mapArgs["-dropmessagestest"])) == 0)
 697         {
 698             printf("dropmessages DROPPING SEND MESSAGE\n");
 699             AbortMessage();
 700             return;
 701         }
 702
 703         if (nHeaderStart == -1)
 704             return;
 705
 706         // Set the size
 707         unsigned int nSize = vSend.size() - nMessageStart;
 708         memcpy((char*)&vSend[nHeaderStart] + offsetof(CMessageHeader, nMessageSize), &nSize, sizeof(nSize));
 709
 710         // Set the checksum
 711         if (vSend.GetVersion() >= 209)
 712         {
 713             uint256 hash = Hash(vSend.begin() + nMessageStart, vSend.end());
 714             unsigned int nChecksum = 0;
 715             memcpy(&nChecksum, &hash, sizeof(nChecksum));
 716             assert(nMessageStart - nHeaderStart >= offsetof(CMessageHeader, nChecksum) + sizeof(nChecksum));
 717             memcpy((char*)&vSend[nHeaderStart] + offsetof(CMessageHeader, nChecksum), &nChecksum, sizeof(nChecksum));
 718         }
 719
 720         printf("(%d bytes) ", nSize);
 721         printf("\n");
 722
 723         nHeaderStart = -1;
 724         nMessageStart = -1;
 725         cs_vSend.Leave();
 726     }
 727
 728     void EndMessageAbortIfEmpty()
 729     {
 730         if (nHeaderStart == -1)
 731             return;
 732         int nSize = vSend.size() - nMessageStart;
 733         if (nSize > 0)
 734             EndMessage();
 735         else
 736             AbortMessage();
 737     }
 738
 739     const char* GetMessageCommand() const
 740     {
 741         if (nHeaderStart == -1)
 742             return "";
 743         return &vSend[nHeaderStart] + offsetof(CMessageHeader, pchCommand);
 744     }
 745
 746
 747
 748
 749     void PushMessage(const char* pszCommand)
 750     {
 751         try
 752         {
 753             BeginMessage(pszCommand);
 754             EndMessage();
 755         }
 756         catch (...)
 757         {
 758             AbortMessage();
 759             throw;
 760         }
 761     }
 762
 763     template<typename T1>
 764     void PushMessage(const char* pszCommand, const T1& a1)
 765     {
 766         try
 767         {
 768             BeginMessage(pszCommand);
 769             vSend << a1;
 770             EndMessage();
 771         }
 772         catch (...)
 773         {
 774             AbortMessage();
 775             throw;
 776         }
 777     }
 778
 779     template<typename T1, typename T2>
 780     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2)
 781     {
 782         try
 783         {
 784             BeginMessage(pszCommand);
 785             vSend << a1 << a2;
 786             EndMessage();
 787         }
 788         catch (...)
 789         {
 790             AbortMessage();
 791             throw;
 792         }
 793     }
 794
 795     template<typename T1, typename T2, typename T3>
 796     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3)
 797     {
 798         try
 799         {
 800             BeginMessage(pszCommand);
 801             vSend << a1 << a2 << a3;
 802             EndMessage();
 803         }
 804         catch (...)
 805         {
 806             AbortMessage();
 807             throw;
 808         }
 809     }
 810
 811     template<typename T1, typename T2, typename T3, typename T4>
 812     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4)
 813     {
 814         try
 815         {
 816             BeginMessage(pszCommand);
 817             vSend << a1 << a2 << a3 << a4;
 818             EndMessage();
 819         }
 820         catch (...)
 821         {
 822             AbortMessage();
 823             throw;
 824         }
 825     }
 826
 827     template<typename T1, typename T2, typename T3, typename T4, typename T5>
 828     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4, const T5& a5)
 829     {
 830         try
 831         {
 832             BeginMessage(pszCommand);
 833             vSend << a1 << a2 << a3 << a4 << a5;
 834             EndMessage();
 835         }
 836         catch (...)
 837         {
 838             AbortMessage();
 839             throw;
 840         }
 841     }
 842
 843     template<typename T1, typename T2, typename T3, typename T4, typename T5, typename T6>
 844     void PushMessage(const char* pszCommand, const T1& a1, const T2& a2, const T3& a3, const T4& a4, const T5& a5, const T6& a6)
 845     {
 846         try
 847         {
 848             BeginMessage(pszCommand);
 849             vSend << a1 << a2 << a3 << a4 << a5 << a6;
 850             EndMessage();
 851         }
 852         catch (...)
 853         {
 854             AbortMessage();
 855             throw;
 856         }
 857     }
 858
 859     template<typename T1, typename T2, typename T3, typename T4, typename T5, typename T6, typename T7>
 860     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)
 861     {
 862         try
 863         {
 864             BeginMessage(pszCommand);
 865             vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7;
 866             EndMessage();
 867         }
 868         catch (...)
 869         {
 870             AbortMessage();
 871             throw;
 872         }
 873     }
 874
 875     template<typename T1, typename T2, typename T3, typename T4, typename T5, typename T6, typename T7, typename T8>
 876     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)
 877     {
 878         try
 879         {
 880             BeginMessage(pszCommand);
 881             vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7 << a8;
 882             EndMessage();
 883         }
 884         catch (...)
 885         {
 886             AbortMessage();
 887             throw;
 888         }
 889     }
 890
 891     template<typename T1, typename T2, typename T3, typename T4, typename T5, typename T6, typename T7, typename T8, typename T9>
 892     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)
 893     {
 894         try
 895         {
 896             BeginMessage(pszCommand);
 897             vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7 << a8 << a9;
 898             EndMessage();
 899         }
 900         catch (...)
 901         {
 902             AbortMessage();
 903             throw;
 904         }
 905     }
 906
 907
 908     void PushRequest(const char* pszCommand,
 909                      void (*fn)(void*, CDataStream&), void* param1)
 910     {
 911         uint256 hashReply;
 912         RAND_bytes((unsigned char*)&hashReply, sizeof(hashReply));
 913
 914         CRITICAL_BLOCK(cs_mapRequests)
 915             mapRequests[hashReply] = CRequestTracker(fn, param1);
 916
 917         PushMessage(pszCommand, hashReply);
 918     }
 919
 920     template<typename T1>
 921     void PushRequest(const char* pszCommand, const T1& a1,
 922                      void (*fn)(void*, CDataStream&), void* param1)
 923     {
 924         uint256 hashReply;
 925         RAND_bytes((unsigned char*)&hashReply, sizeof(hashReply));
 926
 927         CRITICAL_BLOCK(cs_mapRequests)
 928             mapRequests[hashReply] = CRequestTracker(fn, param1);
 929
 930         PushMessage(pszCommand, hashReply, a1);
 931     }
 932
 933     template<typename T1, typename T2>
 934     void PushRequest(const char* pszCommand, const T1& a1, const T2& a2,
 935                      void (*fn)(void*, CDataStream&), void* param1)
 936     {
 937         uint256 hashReply;
 938         RAND_bytes((unsigned char*)&hashReply, sizeof(hashReply));
 939
 940         CRITICAL_BLOCK(cs_mapRequests)
 941             mapRequests[hashReply] = CRequestTracker(fn, param1);
 942
 943         PushMessage(pszCommand, hashReply, a1, a2);
 944     }
 945
 946
 947
 948     void PushGetBlocks(CBlockIndex* pindexBegin, uint256 hashEnd);
 949     bool IsSubscribed(unsigned int nChannel);
 950     void Subscribe(unsigned int nChannel, unsigned int nHops=0);
 951     void CancelSubscribe(unsigned int nChannel);
 952     void CloseSocketDisconnect();
 953     void Cleanup();
 954 };
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965 inline void RelayInventory(const CInv& inv)
 966 {
 967     // Put on lists to offer to the other nodes
 968     CRITICAL_BLOCK(cs_vNodes)
 969         foreach(CNode* pnode, vNodes)
 970             pnode->PushInventory(inv);
 971 }
 972
 973 template<typename T>
 974 void RelayMessage(const CInv& inv, const T& a)
 975 {
 976     CDataStream ss(SER_NETWORK);
 977     ss.reserve(10000);
 978     ss << a;
 979     RelayMessage(inv, ss);
 980 }
 981
 982 template<>
 983 inline void RelayMessage<>(const CInv& inv, const CDataStream& ss)
 984 {
 985     CRITICAL_BLOCK(cs_mapRelay)
 986     {
 987         // Expire old relay messages
 988         while (!vRelayExpiration.empty() && vRelayExpiration.front().first < GetTime())
 989         {
 990             mapRelay.erase(vRelayExpiration.front().second);
 991             vRelayExpiration.pop_front();
 992         }
 993
 994         // Save original serialized message so newer versions are preserved
 995         mapRelay[inv] = ss;
 996         vRelayExpiration.push_back(make_pair(GetTime() + 15 * 60, inv));
 997     }
 998
 999     RelayInventory(inv);
1000 }
1001
1002
1003
1004
1005
1006
1007
1008
1009 //
1010 // Templates for the publish and subscription system.
1011 // The object being published as T& obj needs to have:
1012 //   a set<unsigned int> setSources member
1013 //   specializations of AdvertInsert and AdvertErase
1014 // Currently implemented for CTable and CProduct.
1015 //
1016
1017 template<typename T>
1018 void AdvertStartPublish(CNode* pfrom, unsigned int nChannel, unsigned int nHops, T& obj)
1019 {
1020     // Add to sources
1021     obj.setSources.insert(pfrom->addr.ip);
1022
1023     if (!AdvertInsert(obj))
1024         return;
1025
1026     // Relay
1027     CRITICAL_BLOCK(cs_vNodes)
1028         foreach(CNode* pnode, vNodes)
1029             if (pnode != pfrom && (nHops < PUBLISH_HOPS || pnode->IsSubscribed(nChannel)))
1030                 pnode->PushMessage("publish", nChannel, nHops, obj);
1031 }
1032
1033 template<typename T>
1034 void AdvertStopPublish(CNode* pfrom, unsigned int nChannel, unsigned int nHops, T& obj)
1035 {
1036     uint256 hash = obj.GetHash();
1037
1038     CRITICAL_BLOCK(cs_vNodes)
1039         foreach(CNode* pnode, vNodes)
1040             if (pnode != pfrom && (nHops < PUBLISH_HOPS || pnode->IsSubscribed(nChannel)))
1041                 pnode->PushMessage("pub-cancel", nChannel, nHops, hash);
1042
1043     AdvertErase(obj);
1044 }
1045
1046 template<typename T>
1047 void AdvertRemoveSource(CNode* pfrom, unsigned int nChannel, unsigned int nHops, T& obj)
1048 {
1049     // Remove a source
1050     obj.setSources.erase(pfrom->addr.ip);
1051
1052     // If no longer supported by any sources, cancel it
1053     if (obj.setSources.empty())
1054         AdvertStopPublish(pfrom, nChannel, nHops, obj);
1055 }