aboutsummaryrefslogtreecommitdiff
path: root/src/net.h
diff options
context:
space:
mode:
Diffstat (limited to 'src/net.h')
-rw-r--r--src/net.h137
1 files changed, 101 insertions, 36 deletions
diff --git a/src/net.h b/src/net.h
index 3b46523cd9..368e4cd4bb 100644
--- a/src/net.h
+++ b/src/net.h
@@ -27,7 +27,7 @@ extern int nBestHeight;
-inline unsigned int ReceiveBufferSize() { return 1000*GetArg("-maxreceivebuffer", 5*1000); }
+inline unsigned int ReceiveFloodSize() { return 1000*GetArg("-maxreceivebuffer", 5*1000); }
inline unsigned int SendBufferSize() { return 1000*GetArg("-maxsendbuffer", 1*1000); }
void AddOneShot(std::string strDest);
@@ -42,6 +42,7 @@ unsigned short GetListenPort();
bool BindListenPort(const CService &bindAddr, std::string& strError=REF(std::string()));
void StartNode(void* parg);
bool StopNode();
+void SocketSendData(CNode *pnode);
enum
{
@@ -126,6 +127,44 @@ public:
+class CNetMessage {
+public:
+ bool in_data; // parsing header (false) or data (true)
+
+ CDataStream hdrbuf; // partially received header
+ CMessageHeader hdr; // complete header
+ unsigned int nHdrPos;
+
+ CDataStream vRecv; // received message data
+ unsigned int nDataPos;
+
+ CNetMessage(int nTypeIn, int nVersionIn) : hdrbuf(nTypeIn, nVersionIn), vRecv(nTypeIn, nVersionIn) {
+ hdrbuf.resize(24);
+ in_data = false;
+ nHdrPos = 0;
+ nDataPos = 0;
+ }
+
+ bool complete() const
+ {
+ if (!in_data)
+ return false;
+ return (hdr.nMessageSize == nDataPos);
+ }
+
+ void SetVersion(int nVersionIn)
+ {
+ hdrbuf.SetVersion(nVersionIn);
+ vRecv.SetVersion(nVersionIn);
+ }
+
+ int readHeader(const char *pch, unsigned int nBytes);
+ int readData(const char *pch, unsigned int nBytes);
+};
+
+
+
+
/** Information about a peer */
class CNode
@@ -134,16 +173,21 @@ public:
// socket
uint64 nServices;
SOCKET hSocket;
- CDataStream vSend;
- CDataStream vRecv;
+ CDataStream ssSend;
+ size_t nSendSize; // total size of all vSendMsg entries
+ size_t nSendOffset; // offset inside the first vSendMsg already sent
+ std::deque<CSerializeData> vSendMsg;
CCriticalSection cs_vSend;
- CCriticalSection cs_vRecv;
+
+ std::deque<CInv> vRecvGetData;
+ std::deque<CNetMessage> vRecvMsg;
+ CCriticalSection cs_vRecvMsg;
+ int nRecvVersion;
+
int64 nLastSend;
int64 nLastRecv;
int64 nLastSendEmpty;
int64 nTimeConnected;
- int nHeaderStart;
- unsigned int nMessageStart;
CAddress addr;
std::string addrName;
CService addrLocal;
@@ -191,16 +235,15 @@ public:
CCriticalSection cs_inventory;
std::multimap<int64, CInv> mapAskFor;
- CNode(SOCKET hSocketIn, CAddress addrIn, std::string addrNameIn = "", bool fInboundIn=false) : vSend(SER_NETWORK, MIN_PROTO_VERSION), vRecv(SER_NETWORK, MIN_PROTO_VERSION)
+ CNode(SOCKET hSocketIn, CAddress addrIn, std::string addrNameIn = "", bool fInboundIn=false) : ssSend(SER_NETWORK, MIN_PROTO_VERSION)
{
nServices = 0;
hSocket = hSocketIn;
+ nRecvVersion = MIN_PROTO_VERSION;
nLastSend = 0;
nLastRecv = 0;
nLastSendEmpty = GetTime();
nTimeConnected = GetTime();
- nHeaderStart = -1;
- nMessageStart = -1;
addr = addrIn;
addrName = addrNameIn == "" ? addr.ToStringIPPort() : addrNameIn;
nVersion = 0;
@@ -213,6 +256,8 @@ public:
fDisconnect = false;
nRefCount = 0;
nReleaseTime = 0;
+ nSendSize = 0;
+ nSendOffset = 0;
hashContinue = 0;
pindexLastGetBlocksBegin = 0;
hashLastGetBlocksEnd = 0;
@@ -250,6 +295,26 @@ public:
return std::max(nRefCount, 0) + (GetTime() < nReleaseTime ? 1 : 0);
}
+ // requires LOCK(cs_vRecvMsg)
+ unsigned int GetTotalRecvSize()
+ {
+ unsigned int total = 0;
+ BOOST_FOREACH(const CNetMessage &msg, vRecvMsg)
+ total += msg.vRecv.size() + 24;
+ return total;
+ }
+
+ // requires LOCK(cs_vRecvMsg)
+ bool ReceiveMsgBytes(const char *pch, unsigned int nBytes);
+
+ // requires LOCK(cs_vRecvMsg)
+ void SetRecvVersion(int nVersionIn)
+ {
+ nRecvVersion = nVersionIn;
+ BOOST_FOREACH(CNetMessage &msg, vRecvMsg)
+ msg.SetVersion(nVersionIn);
+ }
+
CNode* AddRef(int64 nTimeout=0)
{
if (nTimeout != 0)
@@ -324,11 +389,8 @@ public:
void BeginMessage(const char* pszCommand) EXCLUSIVE_LOCK_FUNCTION(cs_vSend)
{
ENTER_CRITICAL_SECTION(cs_vSend);
- if (nHeaderStart != -1)
- AbortMessage();
- nHeaderStart = vSend.size();
- vSend << CMessageHeader(pszCommand, 0);
- nMessageStart = vSend.size();
+ assert(ssSend.size() == 0);
+ ssSend << CMessageHeader(pszCommand, 0);
if (fDebug)
printf("sending: %s ", pszCommand);
}
@@ -336,11 +398,8 @@ public:
// TODO: Document the precondition of this function. Is cs_vSend locked?
void AbortMessage() UNLOCK_FUNCTION(cs_vSend)
{
- if (nHeaderStart < 0)
- return;
- vSend.resize(nHeaderStart);
- nHeaderStart = -1;
- nMessageStart = -1;
+ ssSend.clear();
+
LEAVE_CRITICAL_SECTION(cs_vSend);
if (fDebug)
@@ -357,26 +416,32 @@ public:
return;
}
- if (nHeaderStart < 0)
+ if (ssSend.size() == 0)
return;
// Set the size
- unsigned int nSize = vSend.size() - nMessageStart;
- memcpy((char*)&vSend[nHeaderStart] + CMessageHeader::MESSAGE_SIZE_OFFSET, &nSize, sizeof(nSize));
+ unsigned int nSize = ssSend.size() - CMessageHeader::HEADER_SIZE;
+ memcpy((char*)&ssSend[CMessageHeader::MESSAGE_SIZE_OFFSET], &nSize, sizeof(nSize));
// Set the checksum
- uint256 hash = Hash(vSend.begin() + nMessageStart, vSend.end());
+ uint256 hash = Hash(ssSend.begin() + CMessageHeader::HEADER_SIZE, ssSend.end());
unsigned int nChecksum = 0;
memcpy(&nChecksum, &hash, sizeof(nChecksum));
- assert(nMessageStart - nHeaderStart >= CMessageHeader::CHECKSUM_OFFSET + sizeof(nChecksum));
- memcpy((char*)&vSend[nHeaderStart] + CMessageHeader::CHECKSUM_OFFSET, &nChecksum, sizeof(nChecksum));
+ assert(ssSend.size () >= CMessageHeader::CHECKSUM_OFFSET + sizeof(nChecksum));
+ memcpy((char*)&ssSend[CMessageHeader::CHECKSUM_OFFSET], &nChecksum, sizeof(nChecksum));
if (fDebug) {
printf("(%d bytes)\n", nSize);
}
- nHeaderStart = -1;
- nMessageStart = -1;
+ std::deque<CSerializeData>::iterator it = vSendMsg.insert(vSendMsg.end(), CSerializeData());
+ ssSend.GetAndClear(*it);
+ nSendSize += (*it).size();
+
+ // If write queue empty, attempt "optimistic write"
+ if (it == vSendMsg.begin())
+ SocketSendData(this);
+
LEAVE_CRITICAL_SECTION(cs_vSend);
}
@@ -403,7 +468,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1;
+ ssSend << a1;
EndMessage();
}
catch (...)
@@ -419,7 +484,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1 << a2;
+ ssSend << a1 << a2;
EndMessage();
}
catch (...)
@@ -435,7 +500,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1 << a2 << a3;
+ ssSend << a1 << a2 << a3;
EndMessage();
}
catch (...)
@@ -451,7 +516,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1 << a2 << a3 << a4;
+ ssSend << a1 << a2 << a3 << a4;
EndMessage();
}
catch (...)
@@ -467,7 +532,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1 << a2 << a3 << a4 << a5;
+ ssSend << a1 << a2 << a3 << a4 << a5;
EndMessage();
}
catch (...)
@@ -483,7 +548,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1 << a2 << a3 << a4 << a5 << a6;
+ ssSend << a1 << a2 << a3 << a4 << a5 << a6;
EndMessage();
}
catch (...)
@@ -499,7 +564,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7;
+ ssSend << a1 << a2 << a3 << a4 << a5 << a6 << a7;
EndMessage();
}
catch (...)
@@ -515,7 +580,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7 << a8;
+ ssSend << a1 << a2 << a3 << a4 << a5 << a6 << a7 << a8;
EndMessage();
}
catch (...)
@@ -531,7 +596,7 @@ public:
try
{
BeginMessage(pszCommand);
- vSend << a1 << a2 << a3 << a4 << a5 << a6 << a7 << a8 << a9;
+ ssSend << a1 << a2 << a3 << a4 << a5 << a6 << a7 << a8 << a9;
EndMessage();
}
catch (...)
generated by cgit v1.2.3 (git 2.26.2) at 2024-09-13 09:09:26 +0000