[#892] create pkt thread pool and handle processing using multiple threads

src/bin/dhcp4/dhcp4_srv.cc

@@ -39,7 +39,6 @@
 #include <dhcpsrv/subnet.h>
 #include <dhcpsrv/subnet_selector.h>
 #include <dhcpsrv/utils.h>
-#include <dhcpsrv/utils.h>
 #include <eval/evaluate.h>
 #include <eval/eval_messages.h>
 #include <hooks/callout_handle.h>
@@ -47,7 +46,6 @@
 #include <hooks/hooks_manager.h>
 #include <stats/stats_mgr.h>
 #include <util/strutil.h>
-#include <stats/stats_mgr.h>
 #include <log/logger.h>
 #include <cryptolink/cryptolink.h>
 #include <cfgrpt/config_report.h>
@@ -80,6 +78,7 @@ using namespace isc::dhcp_ddns;
 using namespace isc::hooks;
 using namespace isc::log;
 using namespace isc::stats;
+using namespace isc::util;
 using namespace std;
 
 namespace {
@@ -102,8 +101,8 @@ struct Dhcp4Hooks {
         hook_index_pkt4_receive_      = HooksManager::registerHook("pkt4_receive");
         hook_index_subnet4_select_    = HooksManager::registerHook("subnet4_select");
         hook_index_leases4_committed_ = HooksManager::registerHook("leases4_committed");
-        hook_index_pkt4_send_         = HooksManager::registerHook("pkt4_send");
         hook_index_lease4_release_    = HooksManager::registerHook("lease4_release");
+        hook_index_pkt4_send_         = HooksManager::registerHook("pkt4_send");
         hook_index_buffer4_send_      = HooksManager::registerHook("buffer4_send");
         hook_index_lease4_decline_    = HooksManager::registerHook("lease4_decline");
         hook_index_host4_identifier_  = HooksManager::registerHook("host4_identifier");
@@ -209,7 +208,7 @@ Dhcpv4Exchange::Dhcpv4Exchange(const AllocEnginePtr& alloc_engine,
             .arg(query_->getLabel())
             .arg(classes.toText());
     }
-};
+}
 
 void
 Dhcpv4Exchange::initResponse() {
@@ -470,14 +469,15 @@ const std::string Dhcpv4Srv::VENDOR_CLASS_PREFIX("VENDOR_CLASS_");
 
 Dhcpv4Srv::Dhcpv4Srv(uint16_t server_port, uint16_t client_port,
                      const bool use_bcast, const bool direct_response_desired)
-    : io_service_(new IOService()), shutdown_(true), alloc_engine_(),
-      use_bcast_(use_bcast), server_port_(server_port),
-      client_port_(client_port),
+    : io_service_(new IOService()), server_port_(server_port),
+      client_port_(client_port), shutdown_(true),
+      alloc_engine_(), use_bcast_(use_bcast),
       network_state_(new NetworkState(NetworkState::DHCPv4)),
       cb_control_(new CBControlDHCPv4()) {
 
     LOG_DEBUG(dhcp4_logger, DBG_DHCP4_START, DHCP4_OPEN_SOCKET)
         .arg(server_port);
+
     try {
         // Port 0 is used for testing purposes where we don't open broadcast
         // capable sockets. So, set the packet filter handling direct traffic
@@ -801,6 +801,11 @@ Dhcpv4Srv::run() {
         }
     }
 
+    // destroying the thread pool
+    if (Dhcpv4Srv::threadCount()) {
+        pkt_thread_pool_.reset();
+    }
+
     return (true);
 }
 
@@ -811,6 +816,17 @@ Dhcpv4Srv::run_one() {
     Pkt4Ptr rsp;
 
     try {
+
+        // Do not read more packets from socket if there are enough
+        // packets to be processed in the packet thread pool queue
+        const int max_queued_pkt_per_thread = Dhcpv4Srv::maxThreadQueueSize();
+        const auto queue_full_wait = std::chrono::milliseconds(1);
+        size_t pkt_queue_size = pkt_thread_pool_.count();
+        if (pkt_queue_size >= Dhcpv4Srv::threadCount() *
+            max_queued_pkt_per_thread) {
+            return;
+        }
+
         // Set select() timeout to 1s. This value should not be modified
         // because it is important that the select() returns control
         // frequently so as the IOService can be polled for ready handlers.
@@ -884,9 +900,33 @@ Dhcpv4Srv::run_one() {
             .arg(query->getLabel());
         return;
     } else {
-        processPacket(query, rsp);
+        if (Dhcpv4Srv::threadCount()) {
+            typedef function<void()> CallBack;
+            boost::shared_ptr<CallBack> call_back =
+                boost::make_shared<CallBack>(std::bind(&Dhcpv4Srv::processPacketAndSendResponseNoThrow,
+                                                       this, query, rsp));
+            pkt_thread_pool_.add(call_back);
+        } else {
+            processPacketAndSendResponse(query, rsp);
+        }
     }
+}
 
+void
+Dhcpv4Srv::processPacketAndSendResponseNoThrow(Pkt4Ptr& query, Pkt4Ptr& rsp) {
+    try {
+        processPacketAndSendResponse(query, rsp);
+    } catch (const std::exception& e) {
+        LOG_ERROR(packet4_logger, DHCP4_PACKET_PROCESS_STD_EXCEPTION)
+            .arg(e.what());
+    } catch (...) {
+        LOG_ERROR(packet4_logger, DHCP4_PACKET_PROCESS_EXCEPTION);
+    }
+}
+
+void
+Dhcpv4Srv::processPacketAndSendResponse(Pkt4Ptr& query, Pkt4Ptr& rsp) {
+    processPacket(query, rsp);
     if (!rsp) {
         return;
     }
@@ -3800,6 +3840,23 @@ int Dhcpv4Srv::getHookIndexLease4Decline() {
     return (Hooks.hook_index_lease4_decline_);
 }
 
+uint32_t Dhcpv4Srv::threadCount() {
+    uint32_t sys_threads = CfgMgr::instance().getCurrentCfg()->getServerThreadCount();
+    if (sys_threads) {
+        return sys_threads;
+    }
+    sys_threads = std::thread::hardware_concurrency();
+    return sys_threads * 0;
+}
+
+uint32_t Dhcpv4Srv::maxThreadQueueSize() {
+    uint32_t max_thread_queue_size = CfgMgr::instance().getCurrentCfg()->getServerMaxThreadQueueSize();
+    if (max_thread_queue_size) {
+        return max_thread_queue_size;
+    }
+    return 4;
+}
+
 void Dhcpv4Srv::discardPackets() {
     // Clear any packets held by the callhout handle store and
     // all parked packets