P2P: improve RX/TX flow control

1) "optimistic write": Push each message to kernel socket buffer immediately. 2) If there is write data at select time, that implies send() blocked during optimistic write. Drain write queue, before receiving any more messages. This avoids needlessly queueing received data, if the remote peer is not themselves receiving data. Result: write buffer (and thus memory usage) is kept small, DoS potential is slightly lower, and TCP flow control signalling is properly utilized. The kernel will queue data into the socket buffer, then signal the remote peer to stop sending data, until we resume reading again.
TrezarCoin · Mar 29, 2013 · b9ff297 · b9ff297
1 parent bc2f5aa
commit b9ff297
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 6 deletions.
diff --git a/src/net.cpp b/src/net.cpp
@@ -855,14 +855,18 @@ void ThreadSocketHandler2(void* parg)
             {
                 if (pnode->hSocket == INVALID_SOCKET)
                     continue;
-                FD_SET(pnode->hSocket, &fdsetRecv);
-                FD_SET(pnode->hSocket, &fdsetError);
-                hSocketMax = max(hSocketMax, pnode->hSocket);
-                have_fds = true;
                 {
                     TRY_LOCK(pnode->cs_vSend, lockSend);
-                    if (lockSend && !pnode->vSend.empty())
-                        FD_SET(pnode->hSocket, &fdsetSend);
+                    if (lockSend) {
+                        // do not read, if draining write queue
+                        if (!pnode->vSend.empty())
+                            FD_SET(pnode->hSocket, &fdsetSend);
+                        else
+                            FD_SET(pnode->hSocket, &fdsetRecv);
+                        FD_SET(pnode->hSocket, &fdsetError);
+                        hSocketMax = max(hSocketMax, pnode->hSocket);
+                        have_fds = true;
+                    }
                 }
             }
         }

diff --git a/src/net.h b/src/net.h
@@ -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
 {
@@ -437,6 +438,10 @@ class CNode
             printf("(%d bytes)\n", nSize);
         }
 
+        // If write queue empty, attempt "optimistic write"
+        if (nHeaderStart == 0)
+            SocketSendData(this);
+
         nHeaderStart = -1;
         nMessageStart = -1;
         LEAVE_CRITICAL_SECTION(cs_vSend);