Merge branch 'sharding' of https://github.com/triAGENS/ArangoDB into sharding

lib/SimpleHttpClient/ConnectionManager.cpp

@@ -0,0 +1,290 @@
+////////////////////////////////////////////////////////////////////////////////
+/// @brief manages open HTTP connections on the client side
+///
+/// @file ConnectionManager.cpp
+///
+/// DISCLAIMER
+///
+/// Copyright 2010-2014 triagens GmbH, Cologne, Germany
+///
+/// Licensed under the Apache License, Version 2.0 (the "License");
+/// you may not use this file except in compliance with the License.
+/// You may obtain a copy of the License at
+///
+///     http://www.apache.org/licenses/LICENSE-2.0
+///
+/// Unless required by applicable law or agreed to in writing, software
+/// distributed under the License is distributed on an "AS IS" BASIS,
+/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/// See the License for the specific language governing permissions and
+/// limitations under the License.
+///
+/// Copyright holder is triAGENS GmbH, Cologne, Germany
+///
+/// @author Max Neunhoeffer
+/// @author Copyright 2014, triagens GmbH, Cologne, Germany
+////////////////////////////////////////////////////////////////////////////////
+
+#include "Basics/WriteLocker.h"
+
+#include "ConnectionManager.h"
+
+using namespace triagens::httpclient;
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief global options for connections
+////////////////////////////////////////////////////////////////////////////////
+
+ConnectionOptions ConnectionManager::_globalConnectionOptions = {
+  15.0,  // connectTimeout 
+  3.0,   // requestTimeout
+  3,     // numRetries
+  5.0,   // singleRequestTimeout
+  0      // sslProtocol
+};
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief the actual singleton instance
+////////////////////////////////////////////////////////////////////////////////
+
+ConnectionManager* ConnectionManager::_theinstance = 0;
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief destructor
+////////////////////////////////////////////////////////////////////////////////
+
+ConnectionManager::~ConnectionManager ( ) {
+  WRITE_LOCKER(allLock);
+  map<string,ServerConnections*>::iterator i;
+  for (i = allConnections.begin(); i != allConnections.end(); ++i) {
+    delete i->second;
+  }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief destructor for SingleServerConnection class
+////////////////////////////////////////////////////////////////////////////////
+
+ConnectionManager::SingleServerConnection::~SingleServerConnection () {
+  delete connection;
+  delete endpoint;
+  lastUsed = 0;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief destructor of ServerConnections class
+////////////////////////////////////////////////////////////////////////////////
+
+ConnectionManager::ServerConnections::~ServerConnections () {
+  vector<SingleServerConnection*>::iterator i;
+  WRITE_LOCKER(lock);
+
+  unused.clear();
+  for (i = connections.begin();i != connections.end();++i) {
+    delete *i;
+  }
+  connections.clear();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief open or get a previously cached connection to a server
+////////////////////////////////////////////////////////////////////////////////
+
+ConnectionManager::SingleServerConnection* 
+ConnectionManager::leaseConnection(string& endpoint) {
+  map<string,ServerConnections*>::iterator i;
+  ServerConnections* s;
+  SingleServerConnection* c;
+
+  // First find a connections list:
+  {
+    WRITE_LOCKER(allLock);
+
+    i = allConnections.find(endpoint);
+    if (i != allConnections.end()) {
+      s = i->second;
+    }
+    else {
+      s = new ServerConnections();
+      allConnections[endpoint] = s;
+    }
+  }
+
+  assert(s != 0);
+  
+  // Now get an unused one:
+  {
+    WRITE_LOCKER(s->lock);
+    if (!s->unused.empty()) {
+      c = s->unused.back();
+      s->unused.pop_back();
+      return c;
+    }
+  }
+  
+  triagens::rest::Endpoint* e 
+      = triagens::rest::Endpoint::clientFactory(endpoint);
+  if (0 == e) {
+    return 0;
+  }
+  triagens::httpclient::GeneralClientConnection*
+         g = triagens::httpclient::GeneralClientConnection::factory(
+                   e,
+                   _globalConnectionOptions._requestTimeout,
+                   _globalConnectionOptions._connectTimeout,
+                   _globalConnectionOptions._connectRetries,
+                   _globalConnectionOptions._sslProtocol);
+  if (0 == g) {
+    delete e;
+    return 0;
+  }
+  c = new SingleServerConnection(g,e,endpoint);
+  if (0 == c) {
+    delete g;
+    delete e;
+    return 0;
+  }
+
+  // Now put it into our administration:
+  {
+    WRITE_LOCKER(s->lock);
+    s->connections.push_back(c);
+  }
+  c->lastUsed = time(0);
+  return c;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief return leased connection to a server
+////////////////////////////////////////////////////////////////////////////////
+
+void ConnectionManager::returnConnection(SingleServerConnection* c) {
+  map<string,ServerConnections*>::iterator i;
+  ServerConnections* s;
+
+  // First find the collections list:
+  {
+    WRITE_LOCKER(allLock);
+
+    i = allConnections.find(c->ep_spec);
+    if (i != allConnections.end()) {
+      s = i->second;
+    }
+    else {
+      // How strange! We just destroy the connection in despair!
+      delete c;
+      return;
+    }
+  }
+  
+  c->lastUsed = time(0);
+
+  // Now mark it as unused:
+  {
+    WRITE_LOCKER(s->lock);
+    s->unused.push_back(c);
+  }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief report a leased connection as being broken
+////////////////////////////////////////////////////////////////////////////////
+
+void ConnectionManager::brokenConnection(SingleServerConnection* c) {
+  map<string,ServerConnections*>::iterator i;
+  ServerConnections* s;
+
+  // First find the collections list:
+  {
+    WRITE_LOCKER(allLock);
+
+    i = allConnections.find(c->ep_spec);
+    if (i != allConnections.end()) {
+      s = i->second;
+    }
+    else {
+      // How strange! We just destroy the connection in despair!
+      delete c;
+      return;
+    }
+  }
+  
+  // Now find it to get rid of it:
+  {
+    WRITE_LOCKER(s->lock);
+    vector<SingleServerConnection*>::iterator i;
+    for (i = s->connections.begin(); i != s->connections.end(); ++i) {
+      if (*i == c) {
+        // Got it, now remove it:
+        s->connections.erase(i);
+        delete c;
+        return;
+      }
+    }
+  }
+
+  // How strange! We should have known this one!
+  delete c;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief closes all connections that have been unused for more than
+/// limit seconds
+////////////////////////////////////////////////////////////////////////////////
+
+void ConnectionManager::closeUnusedConnections (double limit) {
+  WRITE_LOCKER(allLock);
+  map<string,ServerConnections*>::iterator s;
+  list<SingleServerConnection*>::iterator i;
+  list<SingleServerConnection*>::iterator prev;
+  ServerConnections* sc;
+  time_t t;
+  bool haveprev;
+
+  t = time(0);
+  for (s = allConnections.begin(); s != allConnections.end(); ++s) {
+    sc = s->second;
+    {
+      WRITE_LOCKER(sc->lock);
+      haveprev = false;
+      for (i = sc->unused.begin(); i != sc->unused.end(); ) {
+        if (t - (*i)->lastUsed > limit) {
+          vector<SingleServerConnection*>::iterator j;
+          for (j = sc->connections.begin(); j != sc->connections.end(); ++j) {
+            if (*j == *i) {
+              sc->connections.erase(j);
+              break;
+            }
+          }
+          delete (*i);
+          sc->unused.erase(i);
+          if (haveprev) {
+            i = prev;   // will be incremented in next iteration
+            i++;
+            haveprev = false;
+          }
+          else {
+            i = sc->unused.begin();
+          }
+        }
+        else {
+          prev = i;
+          ++i;
+          haveprev = true;
+        }
+      }
+    }
+  }
+}
+
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief
+////////////////////////////////////////////////////////////////////////////////
+
+// Local Variables:
+// mode: outline-minor
+// outline-regexp: "^\\(/// @brief\\|/// {@inheritDoc}\\|/// @addtogroup\\|// --SECTION--\\|/// @\\}\\)"
+// End:
+
+

lib/SimpleHttpClient/ConnectionManager.h

@@ -0,0 +1,214 @@
+////////////////////////////////////////////////////////////////////////////////
+/// @brief manages open HTTP connections on the client side
+///
+/// @file ConnectionManager.h
+///
+/// DISCLAIMER
+///
+/// Copyright 2010-2014 triagens GmbH, Cologne, Germany
+///
+/// Licensed under the Apache License, Version 2.0 (the "License");
+/// you may not use this file except in compliance with the License.
+/// You may obtain a copy of the License at
+///
+///     http://www.apache.org/licenses/LICENSE-2.0
+///
+/// Unless required by applicable law or agreed to in writing, software
+/// distributed under the License is distributed on an "AS IS" BASIS,
+/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/// See the License for the specific language governing permissions and
+/// limitations under the License.
+///
+/// Copyright holder is triAGENS GmbH, Cologne, Germany
+///
+/// @author Max Neunhoeffer
+/// @author Copyright 2014, triagens GmbH, Cologne, Germany
+////////////////////////////////////////////////////////////////////////////////
+
+#ifndef CONNECTION_MANAGER_H
+#define CONNECTION_MANAGER_H 1
+
+#include "BasicsC/common.h"
+#include "Basics/Common.h"
+#include "Basics/ReadWriteLock.h"
+#include "SimpleHttpClient/GeneralClientConnection.h"
+
+namespace triagens {
+  namespace httpclient {
+
+// -----------------------------------------------------------------------------
+// --SECTION--                                                ConnectionManager
+// -----------------------------------------------------------------------------
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief options for connections
+////////////////////////////////////////////////////////////////////////////////
+
+    struct ConnectionOptions {
+      double _connectTimeout;
+      double _requestTimeout;
+      size_t _connectRetries;
+      double _singleRequestTimeout;
+      uint32_t _sslProtocol;
+    };
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief the class to manage open client connections
+////////////////////////////////////////////////////////////////////////////////
+
+    class ConnectionManager {
+
+// -----------------------------------------------------------------------------
+// --SECTION--                                     constructors and destructors
+// -----------------------------------------------------------------------------
+      
+////////////////////////////////////////////////////////////////////////////////
+/// @brief initialises library
+///
+/// We are a singleton class, therefore nobody is allowed to create
+/// new instances or copy them, except we ourselves.
+////////////////////////////////////////////////////////////////////////////////
+      
+        ConnectionManager ( ) { }
+        ConnectionManager (ConnectionManager const&);    // not implemented
+        void operator= (ConnectionManager const&);       // not implemented
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief shuts down library
+////////////////////////////////////////////////////////////////////////////////
+      
+      public:
+
+        ~ConnectionManager ( );
+
+// -----------------------------------------------------------------------------
+// --SECTION--                                                public subclasses
+// -----------------------------------------------------------------------------
+      
+////////////////////////////////////////////////////////////////////////////////
+/// @brief class to administrate one connection to a server
+////////////////////////////////////////////////////////////////////////////////
+
+        struct SingleServerConnection {
+          GeneralClientConnection* connection;
+          rest::Endpoint* endpoint;
+          time_t lastUsed;
+          string ep_spec;
+
+          SingleServerConnection (GeneralClientConnection* c,
+                                  rest::Endpoint* e,
+                                  string& ep_spec)
+              : connection(c), endpoint(e), lastUsed(0), ep_spec(ep_spec) {}
+          ~SingleServerConnection ();
+        };
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief class to administrate all connections to a server
+////////////////////////////////////////////////////////////////////////////////
+
+        struct ServerConnections {
+          vector<SingleServerConnection*> connections;
+          list<SingleServerConnection*> unused;
+          triagens::basics::ReadWriteLock lock;
+
+          ServerConnections () {}
+          ~ServerConnections ();   // closes all connections
+        };
+
+// -----------------------------------------------------------------------------
+// --SECTION--                                                   public methods
+// -----------------------------------------------------------------------------
+      
+////////////////////////////////////////////////////////////////////////////////
+/// @brief get the unique instance
+////////////////////////////////////////////////////////////////////////////////
+      
+        static ConnectionManager* instance ( ) {
+          // This does not have to be thread-safe, because we guarantee that
+          // this is called very early in the startup phase when there is still
+          // a single thread.
+          if (0 == _theinstance) {
+            _theinstance = new ConnectionManager( );  
+            // this now happens exactly once
+          }
+          return _theinstance;
+        }
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief initialise function to call once when still single-threaded
+////////////////////////////////////////////////////////////////////////////////
+        
+        static void initialise () {
+        }
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief cleanup function to call once when shutting down
+////////////////////////////////////////////////////////////////////////////////
+        
+        static void cleanup () {
+          delete _theinstance;
+          _theinstance = 0;
+        }
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief open or get a previously cached connection to a server
+////////////////////////////////////////////////////////////////////////////////
+
+        SingleServerConnection* leaseConnection(std::string& endpoint);
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief return leased connection to a server
+////////////////////////////////////////////////////////////////////////////////
+
+        void returnConnection(SingleServerConnection* singleConnection);
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief report a leased connection as being broken
+////////////////////////////////////////////////////////////////////////////////
+
+        void brokenConnection(SingleServerConnection* singleConnection);
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief closes all connections that have been unused for more than
+/// limit seconds
+////////////////////////////////////////////////////////////////////////////////
+
+        void closeUnusedConnections(double limit);
+
+// -----------------------------------------------------------------------------
+// --SECTION--                                         private methods and data
+// -----------------------------------------------------------------------------
+
+      private: 
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief the pointer to the singleton instance
+////////////////////////////////////////////////////////////////////////////////
+
+        static ConnectionManager* _theinstance;
+         
+////////////////////////////////////////////////////////////////////////////////
+/// @brief global options for connections
+////////////////////////////////////////////////////////////////////////////////
+
+        static ConnectionOptions _globalConnectionOptions;
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief map to store all connections to all servers with corresponding lock
+////////////////////////////////////////////////////////////////////////////////
+
+        // We keep connections to servers open:
+        map<string, ServerConnections*> allConnections;
+        triagens::basics::ReadWriteLock allLock;
+
+    };
+  }
+}
+#endif
+
+// Local Variables:
+// mode: outline-minor
+// outline-regexp: "^\\(/// @brief\\|/// {@inheritDoc}\\|/// @addtogroup\\|// --SECTION--\\|/// @\\}\\)"
+// End:
+
+