From 660dd268f667ea6ef7452a33fd0631cc687255a6 Mon Sep 17 00:00:00 2001
From: Aki Tuomi <cmouse@desteem.org>
Date: Wed, 5 Nov 2014 12:05:45 +0200
Subject: [PATCH] Reuse existing connection when possible

---
 modules/remotebackend/httpconnector.cc | 92 +++++++++++++++++++-------
 modules/remotebackend/remotebackend.hh |  1 +
 2 files changed, 70 insertions(+), 23 deletions(-)

diff --git a/modules/remotebackend/httpconnector.cc b/modules/remotebackend/httpconnector.cc
index f407edc5e..21e5e3741 100644
--- a/modules/remotebackend/httpconnector.cc
+++ b/modules/remotebackend/httpconnector.cc
@@ -6,7 +6,7 @@
 #include <sstream>
 #include "rapidjson/stringbuffer.h"
 #include "rapidjson/writer.h"
-#include "polarssl/ssl.h"
+#include "pdns/lock.hh"
 
 #ifndef UNIX_PATH_MAX
 #define UNIX_PATH_MAX 108
@@ -22,6 +22,7 @@ HTTPConnector::HTTPConnector(std::map<std::string,std::string> options) {
     this->timeout = 2;
     this->d_post = false;
     this->d_post_json = false;
+    this->d_socket = NULL;
 
     if (options.find("timeout") != options.end()) {
       this->timeout = boost::lexical_cast<int>(options.find("timeout")->second)/1000;
@@ -38,11 +39,11 @@ HTTPConnector::HTTPConnector(std::map<std::string,std::string> options) {
         this->d_post_json = true;
       }
     }
-    if (options.find("capath") != options.end()) this->d_capath = options.find("capath")->second;
-    if (options.find("cafile") != options.end()) this->d_cafile = options.find("cafile")->second;
 }
 
 HTTPConnector::~HTTPConnector() {
+    if (d_socket != NULL)
+      delete d_socket;
 }
 
 // converts json value into string
@@ -289,7 +290,7 @@ void HTTPConnector::post_requestbuilder(const rapidjson::Document &input, YaHTTP
 }
 
 int HTTPConnector::send_message(const rapidjson::Document &input) {
-    int rv,ec;
+    int rv,ec,fd;
     
     std::vector<std::string> members;
     std::string method;
@@ -304,34 +305,65 @@ int HTTPConnector::send_message(const rapidjson::Document &input) {
       restful_requestbuilder(input["method"].GetString(), input["parameters"], req);
 
     rv = -1;
-    req.headers["connection"] = "close"; // make sure the other ends knows we are not going to hang around
+    req.headers["connection"] = "Keep-Alive"; // see if we can streamline requests (not needed, strictly speaking)
 
     out << req;
 
+    // try sending with current socket, if it fails retry with new socket
+    if (this->d_socket != NULL) {
+      fd = this->d_socket->getHandle();
+      // there should be no data waiting
+      if (waitForRWData(fd, true, 0, 1000) < 1) {
+        try {
+          d_socket->writenWithTimeout(out.str().c_str(), out.str().size(), timeout);
+          rv = 1;
+        } catch (NetworkError& ne) {
+          L<<Logger::Error<<"While writing to HTTP endpoint "<<d_addr.toStringWithPort()<<": "<<ne.what()<<std::endl;
+        } catch (...) {
+          L<<Logger::Error<<"While writing to HTTP endpoint "<<d_addr.toStringWithPort()<<": exception caught"<<std::endl;
+        }
+      }
+    }
+
+    if (rv == 1) return rv;
+
+    delete this->d_socket;
+    this->d_socket = NULL;
+
     if (req.url.protocol == "unix") {
       // connect using unix socket
     } else {
       // connect using tcp
-      struct addrinfo *gAddr, *gAddrPtr;
+      struct addrinfo *gAddr, *gAddrPtr, hints;
       std::string sPort = boost::lexical_cast<std::string>(req.url.port);
-      if ((ec = getaddrinfo(req.url.host.c_str(), sPort.c_str(), NULL, &gAddr)) == 0) {
+      memset(&hints,0,sizeof hints);
+      hints.ai_family = AF_UNSPEC;
+      hints.ai_flags = AI_ADDRCONFIG; 
+      hints.ai_socktype = SOCK_STREAM;
+      hints.ai_protocol = 6; // tcp
+      if ((ec = getaddrinfo(req.url.host.c_str(), sPort.c_str(), &hints, &gAddr)) == 0) {
         // try to connect to each address. 
         gAddrPtr = gAddr;
+  
         while(gAddrPtr) {
-          d_socket = new Socket(gAddrPtr->ai_family, gAddrPtr->ai_socktype, gAddrPtr->ai_protocol);
           try {
-            ComboAddress addr = *reinterpret_cast<ComboAddress*>(gAddrPtr->ai_addr);
-            d_socket->connect(addr);
+            d_socket = new Socket(gAddrPtr->ai_family, gAddrPtr->ai_socktype, gAddrPtr->ai_protocol);
+            d_addr.setSockaddr(gAddrPtr->ai_addr, gAddrPtr->ai_addrlen);
+            d_socket->connect(d_addr);
             d_socket->setNonBlocking();
             d_socket->writenWithTimeout(out.str().c_str(), out.str().size(), timeout);
             rv = 1;
           } catch (NetworkError& ne) {
-            L<<Logger::Error<<"While writing to HTTP endpoint: "<<ne.what()<<std::endl;
+            L<<Logger::Error<<"While writing to HTTP endpoint "<<d_addr.toStringWithPort()<<": "<<ne.what()<<std::endl;
+          } catch (...) {
+            L<<Logger::Error<<"While writing to HTTP endpoint "<<d_addr.toStringWithPort()<<": exception caught"<<std::endl;
           }
+
           if (rv > -1) break;
           delete d_socket;
           d_socket = NULL;
           gAddrPtr = gAddrPtr->ai_next;
+          
         }
         freeaddrinfo(gAddr);
       } else {
@@ -349,18 +381,35 @@ int HTTPConnector::recv_message(rapidjson::Document &output) {
     if (d_socket == NULL ) return -1; // cannot receive :(
     char buffer[4096];
     int rd = -1;
+    bool fail = false;
 
     arl.initialize(&resp);
 
-    while(arl.ready() == false) {
-       rd = d_socket->readWithTimeout(buffer, sizeof(buffer), timeout);
-       if (rd<0) {
-         delete d_socket;
-         d_socket = NULL;
-         return -1;
-       }
-       buffer[rd] = 0;
-       arl.feed(std::string(buffer, rd));
+    try {
+      while(arl.ready() == false) {
+        rd = d_socket->readWithTimeout(buffer, sizeof(buffer), timeout);
+        if (rd<0) {
+          delete d_socket;
+          d_socket = NULL;
+          fail = true;
+          break;
+        }
+        buffer[rd] = 0;
+        arl.feed(std::string(buffer, rd));
+      }
+    } catch (NetworkError &ne) {
+      L<<Logger::Error<<"While reading from HTTP endpoint "<<d_addr.toStringWithPort()<<": "<<ne.what()<<std::endl; 
+      delete d_socket;
+      d_socket = NULL;
+      fail = true;
+    } catch (...) {
+      L<<Logger::Error<<"While reading from HTTP endpoint "<<d_addr.toStringWithPort()<<": exception caught"<<std::endl;
+      delete d_socket;
+      fail = true;
+    }
+
+    if (fail) {
+      return -1;
     }
 
     arl.finalize();
@@ -380,8 +429,5 @@ int HTTPConnector::recv_message(rapidjson::Document &output) {
     else
        rv = -1;
 
-    delete d_socket;
-    d_socket = NULL;
- 
     return rv;
 }
diff --git a/modules/remotebackend/remotebackend.hh b/modules/remotebackend/remotebackend.hh
index 49770e0cb..76a1d0ec6 100644
--- a/modules/remotebackend/remotebackend.hh
+++ b/modules/remotebackend/remotebackend.hh
@@ -81,6 +81,7 @@ class HTTPConnector: public Connector {
     void post_requestbuilder(const rapidjson::Document &input, YaHTTP::Request& req);
     void addUrlComponent(const rapidjson::Value &parameters, const char *element, std::stringstream& ss);
     Socket* d_socket;
+    ComboAddress d_addr;
 };
 
 #ifdef REMOTEBACKEND_ZEROMQ
-- 
2.40.0