From c5d6b9e6fb3920433914140bc94e3158ec5eecfc Mon Sep 17 00:00:00 2001
From: abadcafe <fwlei@live.com>
Date: Fri, 3 Apr 2015 22:23:04 +0800
Subject: [PATCH 1/2] THRIFT-3080: fix connection leak of C++ Nonblocking
 Server while huge number connections are accepted and unix socket stream fd
 is busy.

---
 .../src/thrift/server/TNonblockingServer.cpp  | 37 +++++++++++++++++--
 1 file changed, 34 insertions(+), 3 deletions(-)

diff --git a/lib/cpp/src/thrift/server/TNonblockingServer.cpp b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
index 587560c0931..31bc34b9caf 100644
--- a/lib/cpp/src/thrift/server/TNonblockingServer.cpp
+++ b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
@@ -28,6 +28,7 @@
 #include <thrift/transport/PlatformSocket.h>
 
 #include <iostream>
+#include <poll.h>
 
 #ifdef HAVE_SYS_SOCKET_H
 #include <sys/socket.h>
@@ -1393,9 +1394,39 @@ bool TNonblockingIOThread::notify(TNonblockingServer::TConnection* conn) {
     return false;
   }
 
-  const int kSize = sizeof(conn);
-  if (send(fd, const_cast_sockopt(&conn), kSize, 0) != kSize) {
-    return false;
+  int ret = -1;
+  struct pollfd pfd = {fd, POLLOUT, 0};
+  int kSize = sizeof(conn);
+  const char * pos = (const char *)const_cast_sockopt(&conn);
+
+  while (kSize > 0) {
+    pfd.revents = 0;
+    ret = poll(&pfd, 1, -1);
+    if (ret < 0) {
+      return false;
+    } else if (ret == 0) {
+      continue;
+    }
+
+    if (pfd.revents & POLLHUP || pfd.revents & POLLERR) {
+      ::close(fd);
+      return false;
+    }
+
+    if (pfd.revents & POLLOUT) {
+      ret = send(fd, pos, kSize, 0);
+      if (ret < 0) {
+        if (errno == EAGAIN) {
+          continue;
+        }
+
+        ::close(fd);
+        return false;
+      }
+
+      kSize -= ret;
+      pos += ret;
+    }
   }
 
   return true;

From ba9fc4a21e318257cea2de5638bd23cc2f9a75b4 Mon Sep 17 00:00:00 2001
From: Lei Feiwei <leifeiwei@baidu.com>
Date: Sat, 4 Apr 2015 22:12:07 +0800
Subject: [PATCH 2/2] THRIFT-3080: use select() instead poll() for early
 windows compatibility.

---
 .../src/thrift/server/TNonblockingServer.cpp  | 22 ++++++++++++-------
 1 file changed, 14 insertions(+), 8 deletions(-)

diff --git a/lib/cpp/src/thrift/server/TNonblockingServer.cpp b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
index 31bc34b9caf..8590bff72a4 100644
--- a/lib/cpp/src/thrift/server/TNonblockingServer.cpp
+++ b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
@@ -28,7 +28,10 @@
 #include <thrift/transport/PlatformSocket.h>
 
 #include <iostream>
-#include <poll.h>
+
+#ifdef HAVE_SYS_SELECT_H
+#include <sys/select.h>
+#endif
 
 #ifdef HAVE_SYS_SOCKET_H
 #include <sys/socket.h>
@@ -1394,33 +1397,36 @@ bool TNonblockingIOThread::notify(TNonblockingServer::TConnection* conn) {
     return false;
   }
 
+  fd_set wfds, efds;
   int ret = -1;
-  struct pollfd pfd = {fd, POLLOUT, 0};
   int kSize = sizeof(conn);
   const char * pos = (const char *)const_cast_sockopt(&conn);
 
   while (kSize > 0) {
-    pfd.revents = 0;
-    ret = poll(&pfd, 1, -1);
+    FD_ZERO(&wfds);
+    FD_ZERO(&efds);
+    FD_SET(fd, &wfds);
+    FD_SET(fd, &efds);
+    ret = select(fd + 1, NULL, &wfds, &efds, NULL);
     if (ret < 0) {
       return false;
     } else if (ret == 0) {
       continue;
     }
 
-    if (pfd.revents & POLLHUP || pfd.revents & POLLERR) {
-      ::close(fd);
+    if (FD_ISSET(fd, &efds)) {
+      ::THRIFT_CLOSESOCKET(fd);
       return false;
     }
 
-    if (pfd.revents & POLLOUT) {
+    if (FD_ISSET(fd, &wfds)) {
       ret = send(fd, pos, kSize, 0);
       if (ret < 0) {
         if (errno == EAGAIN) {
           continue;
         }
 
-        ::close(fd);
+        ::THRIFT_CLOSESOCKET(fd);
         return false;
       }