From b9ff2970b9fbb24e2fffc449b4ef478d019633d8 Mon Sep 17 00:00:00 2001
From: Jeff Garzik <jgarzik@exmulti.com>
Date: Thu, 15 Nov 2012 18:20:26 -0500
Subject: [PATCH] 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.
---
 src/net.cpp | 16 ++++++++++------
 src/net.h   |  5 +++++
 2 files changed, 15 insertions(+), 6 deletions(-)

diff --git a/src/net.cpp b/src/net.cpp
index 96719367c..eafb33564 100644
--- 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
index 78f8e72fb..03d32526b 100644
--- 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 @@ public:
             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);