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