improved & fixed ThriftBroker

 * renamed swc.ThriftBroker.handlers to *.clients.handlers
 * fixed process_results CompactResults&
 * added cfg swc.ThriftBroker.connections.max
 * added default swc.ThriftBroker.cfg.dyn=swc_thriftbroker.dyn.cfg
 * added thrift::transport::TSocket to AppHandler ctor & storage
 * added LOG_INFO Open & Close Connection + total m_connections
 * added SIGINT app-context at  server stopped-serving
 * fixed ThreadManager stop , 1st  stop TThreadPoolServer

Author: kashirin-alex

docs/configure/properties/thriftbroker.md

@@ -23,22 +23,27 @@ _default_ **```swc.ThriftBroker.port=18000```**
 The number of workers a comm-reactor initalizes. \
 _default_ **```swc.ThriftBroker.workers=32```**
 
-* ### swc.ThriftBroker.handlers
+* ### swc.ThriftBroker.connections.max
 ```INT32```
-The number of Application handlers. \
-_default_ **```swc.ThriftBroker.handlers=8```**
+The Max client Connections allowed, any new connections above the Open-Connections will be dropped and the Max-Total is the number of Endpoints(Thrift-Broker is listening-on) by `swc.ThriftBroker.connections.max`. If open-file-descriptors is above allowed-limit Thrift-Broker will shutdown, unplanned shutdown can be avoided by the max limit. \
+_default_ **```swc.ThriftBroker.workers=INT64_MAX```**
 
 * ### swc.ThriftBroker.transport
 ```STRING```
 The thrift transport that should be used (framed/). \
 _default_ **```swc.ThriftBroker.transport=framed```**
 
-
 * ### swc.ThriftBroker.timeout
 ```INT32```
 The ThriftBroker timeout in milliseconds. \
 _default_ **```swc.ThriftBroker.timeout=900000```**
 
+* ### swc.ThriftBroker.clients.handlers
+```INT32```
+The number of SWC-DB clients handlers. \
+_default_ **```swc.ThriftBroker.clients.handlers=8```**
+
+
 
 ***
 

src/cc/bin/swcdb/thriftbroker/main.cc

@@ -35,6 +35,7 @@ int run() {
 
   uint32_t reactors = 1; // settings->get_i32("swc.ThriftBroker.reactors");
   int workers = settings->get_i32("swc.ThriftBroker.workers");
+  uint64_t conns_max = settings->get_i64("swc.ThriftBroker.connections.max");
   uint32_t timeout_ms = settings->get_i16("swc.ThriftBroker.timeout");
   std::string transport = settings->get_str("swc.ThriftBroker.transport");
 
@@ -77,6 +78,7 @@ int run() {
 
 
   auto app_ctx = std::make_shared<AppContext>();
+  std::vector<std::unique_ptr<std::thread>> threads;
   std::vector<std::shared_ptr<thrift::server::TThreadPoolServer>> servers;
 
   for(uint32_t reactor=0; reactor < reactors; ++reactor) {
@@ -114,22 +116,45 @@ int run() {
         protocol,
         threadManager
       );
+      server->setConcurrentClientLimit(conns_max);
+
       servers.push_back(server);
-      std::thread([server]{ server->serve(); }).detach();
- 
-      SWC_LOG_OUT(LOG_INFO,
-        SWC_LOG_OSTREAM << "Listening On: " << endpoint
-          << " fd=" << (ssize_t)server->getServerTransport()->getSocketFD()
-          << ' ' << (is_plain ? "PLAIN" : "SECURE");
+      threads.emplace_back(
+        new std::thread([app_ctx, is_plain, endpoint, server] {
+          SWC_LOG_OUT(LOG_INFO, SWC_LOG_OSTREAM 
+            << "Listening On: " << endpoint
+            << " fd=" << (ssize_t)server->getServerTransport()->getSocketFD()
+            << ' ' << (is_plain ? "PLAIN" : "SECURE");
+          );
+
+          server->serve();
+
+          SWC_LOG_OUT(LOG_INFO, SWC_LOG_OSTREAM 
+            << "Stopping to Listen On: " << endpoint
+            << " fd=" << (ssize_t)server->getServerTransport()->getSocketFD()
+            << ' ' << (is_plain ? "PLAIN" : "SECURE");
+          );
+          app_ctx->shutting_down(std::error_code(), SIGINT);
+        })
       );
     }
   }
 
   app_ctx->wait_while_run();
 
-  quick_exit(0);
-  for(auto& server : servers)
+  for(auto& server : servers) {
+    server->stop();
     server->getThreadManager()->stop();
+  }
+  servers.clear();
+
+  for(auto& th : threads) 
+    th->join();
+  threads.clear();
+
+  SWC_LOG(LOG_INFO, "Exit");
+  //quick_exit(0);
+
 
   return 0);
 

src/cc/include/swcdb/thrift/broker/AppContext.h

@@ -28,7 +28,7 @@ class AppContext final : virtual public BrokerIfFactory {
 
   AppContext() : m_run(true) { 
     auto settings = Env::Config::settings();
-    Env::IoCtx::init(settings->get_i32("swc.ThriftBroker.handlers"));
+    Env::IoCtx::init(settings->get_i32("swc.ThriftBroker.clients.handlers"));
 
     int sig = 0;
     Env::IoCtx::io()->set_signals();
@@ -55,62 +55,90 @@ class AppContext final : virtual public BrokerIfFactory {
   }
 
   virtual ~AppContext() { }
-  
+
   void wait_while_run() {
     std::unique_lock lock_wait(m_mutex);
     m_cv.wait(lock_wait, [this]{return !m_run;});
   }
 
-  BrokerIf* getHandler(const thrift::TConnectionInfo&) override { //connInfo
-    return new AppHandler;
+  BrokerIf* getHandler(const thrift::TConnectionInfo& connInfo) override {
+    AppHandler* handler = new AppHandler(
+      std::dynamic_pointer_cast<thrift::transport::TSocket>(
+        connInfo.transport));
+    if(handler->socket) try {
+      SWC_LOG_OUT(LOG_INFO,
+        SWC_LOG_OSTREAM << "Connection Opened(hdlr=" << size_t(handler)
+                        << " [" << handler->socket->getPeerAddress() << "]:"
+                        << handler->socket->getPeerPort()
+                        << ") open=" << m_connections.increment_and_count();
+      );
+    } catch(...) {
+      SWC_LOG_CURRENT_EXCEPTION("");
+    }
+    return handler;
   }
 
   void releaseHandler(ServiceIf* hdlr) override {
     AppHandler* handler = dynamic_cast<AppHandler*>(hdlr);
     handler->disconnected();
-    
-    delete hdlr;
-  }
-
-  private:
 
-  
-  void shutting_down(const std::error_code &ec, const int &sig) {
+    if(handler->socket) try {
+      SWC_LOG_OUT(LOG_INFO,
+        SWC_LOG_OSTREAM << "Connection Closed(hdlr=" << size_t(handler)
+                        << " [" << handler->socket->getPeerAddress() << "]:"
+                        << handler->socket->getPeerPort()
+                        << ") open=" << m_connections.decrement_and_count();
+      );
+    } catch(...) {
+      SWC_LOG_CURRENT_EXCEPTION("");
+    }
+    delete handler;
+  }
 
-    if(sig==0){ // set signals listener
+  void shutting_down(const std::error_code& ec, const int& sig) {
+    if(sig == 0) { // set signals listener
       Env::IoCtx::io()->signals()->async_wait(
-        [ptr=this](const std::error_code &ec, const int &sig){
+        [this](const std::error_code& ec, const int &sig) {
+          if(ec == asio::error::operation_aborted)
+            return;
           SWC_LOGF(LOG_INFO, "Received signal, sig=%d ec=%s", sig, ec.message().c_str());
-          ptr->shutting_down(ec, sig); 
+          shutting_down(ec, sig); 
         }
       ); 
       SWC_LOGF(LOG_INFO, "Listening for Shutdown signal, set at sig=%d ec=%s", 
               sig, ec.message().c_str());
       return;
+    } else {
+
+      std::scoped_lock lock(m_mutex);
+      if(!m_run)
+        return;
+      m_run = false;
     }
+
     SWC_LOGF(LOG_INFO, "Shutdown signal, sig=%d ec=%s", sig, ec.message().c_str());
 
     stop();
   }
 
+  private:
+
   void stop() {
 
     Env::Clients::get()->rgr->stop();
     Env::Clients::get()->mngr->stop();
+    Env::IoCtx::io()->stop();
 
     //Env::FsInterface::interface()->stop();
-    Env::IoCtx::io()->stop();
 
-    SWC_LOG(LOG_INFO, "Exit");
-    
     std::scoped_lock lock(m_mutex);
-    m_run = false;
     m_cv.notify_all();
   }
 
   std::mutex                                    m_mutex;
   bool                                          m_run;
   std::condition_variable                       m_cv;
+  Common::Stats::CompletionCounter<size_t>      m_connections;
 };
 
 

src/cc/include/swcdb/thrift/broker/AppHandler.h

@@ -7,10 +7,13 @@
 #define swcdb_app_thriftbroker_AppHandler_h
 
 
-#include "swcdb/thrift/gen-cpp/Broker.h"
+#include "swcdb/core/Semaphore.h"
 #include "swcdb/db/client/sql/SQL.h"
+#include "swcdb/thrift/gen-cpp/Broker.h"
 #include "swcdb/thrift/utils/Converter.h"
 
+#include <thrift/transport/TSocket.h>
+
 
 namespace SWC {
 namespace thrift = apache::thrift;
@@ -25,6 +28,12 @@ using namespace Thrift;
 class AppHandler final : virtual public BrokerIf {
   public:
 
+  const std::shared_ptr<thrift::transport::TSocket> socket;
+
+  AppHandler(const std::shared_ptr<thrift::transport::TSocket>& socket)
+            : socket(socket) {
+  }
+
   virtual ~AppHandler() { }
 
   /* SQL any */
@@ -600,25 +609,23 @@ class AppHandler final : virtual public BrokerIf {
         Converter::exception(err);
     }
 
-    std::mutex mutex;
-    std::promise<void> res;
-    for(auto& schema : dbschemas) {
+    size_t sz = dbschemas.size();
+    Core::Semaphore sem(sz, sz);
+    _return.resize(sz);
+    for(size_t idx = 0; idx < sz; ++idx) {
+      auto& r = _return[idx];
       Comm::Protocol::Mngr::Req::ColumnCompact::request(
-        schema->cid,
-        [&mutex, &_return, await=&res, cid=schema->cid, sz=dbschemas.size()]
+        (r.cid = dbschemas[idx]->cid),
+        [&sem, err=&r.err]
         (const Comm::client::ConnQueue::ReqBase::Ptr&, 
          const Comm::Protocol::Mngr::Params::ColumnCompactRsp& rsp) {
-          std::scoped_lock lock(mutex);
-          auto& r = _return.emplace_back();
-          r.cid=cid;
-          r.err=rsp.err;
-          if(_return.size() == sz)
-            await->set_value();
+          *err = rsp.err;
+          sem.release();
         },
         300000
       );
     }
-    res.get_future().wait();
+    sem.wait_all();
   }
 
   static void process_results(

src/cc/include/swcdb/thrift/broker/Settings.h

@@ -19,16 +19,25 @@ void Settings::init_app_options() {
   init_client_options();
 
   file_desc.add_options()
-    ("swc.ThriftBroker.cfg", str(), "Specific cfg-file for ThriftBroker")
+    ("swc.ThriftBroker.cfg", str(), 
+      "Specific cfg-file for ThriftBroker")
     ("swc.ThriftBroker.cfg.dyn", strs(), 
       "Specific dyn. cfg-file for ThriftBroker")
 
-    ("swc.ThriftBroker.port", i16(18000), "ThriftBroker port")
-    ("swc.ThriftBroker.transport", str("framed"), "ThriftBroker timeout")
-
-    ("swc.ThriftBroker.workers", i32(32), "Number of Comm-Workers")
-    ("swc.ThriftBroker.timeout", i32(900000), "ThriftBroker timeout")
-    ("swc.ThriftBroker.handlers", i32(8), "Number of App Handlers")
+    ("swc.ThriftBroker.port", i16(18000), 
+      "ThriftBroker port")
+    ("swc.ThriftBroker.transport", str("framed"), 
+      "ThriftBroker timeout")
+
+    ("swc.ThriftBroker.workers", i32(32),
+      "Number of Comm-Workers")
+    ("swc.ThriftBroker.connections.max", i64(INT64_MAX),
+      "The Max client Connections allowed , total= addrs X this")
+      
+    ("swc.ThriftBroker.timeout", i32(900000), 
+      "ThriftBroker timeout")
+    ("swc.ThriftBroker.clients.handlers", i32(8), 
+      "The number of SWC-DB clients handlers")
   ;
 }
 

src/etc/swcdb/swc_thriftbroker.cfg

@@ -10,11 +10,12 @@
 
 
 
-# swc.ThriftBroker.cfg.dyn=swc_thriftbroker.dyn.cfg
+swc.ThriftBroker.cfg.dyn=swc_thriftbroker.dyn.cfg
 
 # swc.ThriftBroker.port=18000 # default ThriftBroker port
 swc.ThriftBroker.transport=framed # /zlib
 
 swc.ThriftBroker.workers=32
+swc.ThriftBroker.connections.max=1000
 swc.ThriftBroker.timeout=900000
-swc.ThriftBroker.handlers=8
+swc.ThriftBroker.clients.handlers=8

src/etc/swcdb/swc_thriftbroker.dyn.cfg

@@ -0,0 +1,14 @@
+# SWC-DB© Copyright since 2019 Alex Kashirin <[email protected]>
+# License details at <https://github.com/kashirin-alex/swc-db/#license>
+
+
+
+# Dynamic Configurations
+## swcdbThriftBroker read the file
+# documentations at 
+#  https://www.swcdb.org/configure/properties/thriftbroker.html
+##
+
+
+
+swc.logging.level=info

tests/integration/thrift/client.cc

@@ -204,17 +204,25 @@ void sql_list_columns_all(Client& client) {
 void sql_compact_columns(Client& client) {
   std::cout << std::endl << "test: sql_compact_columns all: " << std::endl;
   CompactResults res;
-  client.sql_compact_columns(
-    res, 
-    "compact columns "
-  );
-  assert(res.size() >= 2);
-  std::cout << std::endl << "CompactResults.size=" << res.size() << std::endl;
-  for(auto& r : res) {
-    assert(!r.err);
-    r.printTo(std::cout << " ");
-    std::cout << std::endl;
-  }
+
+  bool err = false;
+  size_t count = 0;
+  do {
+    res.clear();
+    client.sql_compact_columns(res,  "compact columns ");
+    assert(res.size() >= 2);
+    std::cout << std::endl 
+              << "CompactResults.size=" << res.size() 
+              << " waited=" << count << "ms" << std::endl;
+    for(auto& r : res) {
+      r.printTo(std::cout << " ");
+      std::cout << std::endl;
+      if((err = r.err))
+        break;
+    }
+    std::this_thread::sleep_for(std::chrono::microseconds(1000));
+    ++count;
+  } while(err);
 
   std::cout << std::endl << "test: sql_compact_columns test-columns: " << std::endl;
   res.clear();