Changeset 1316

Timestamp:

11/27/08 23:26:42 (21 months ago)

Author:

ahu

Message:

multithread the packet receiver - for great quad core justice

Files:

• trunk/pdns/pdns/common_startup.cc (modified) (5 diffs)

trunk/pdns/pdns/common_startup.cc

@@ -65 +65 @@
   ::arg().set("default-soa-name","name to insert in the SOA record if none set in the backend")="a.misconfigured.powerdns.server";
   ::arg().set("distributor-threads","Default number of Distributor (backend) threads to start")="3";
+  ::arg().set("receiver-threads","Default number of Distributor (backend) threads to start")="1";
   ::arg().set("queue-limit","Maximum number of milliseconds to queue a query")="1500"; 
   ::arg().set("recursor","If recursion is desired, IP address of a recursing nameserver")="no"; 
@@ -203 +204 @@
 }
 
+static DNSDistributor* g_distributor;
+static pthread_mutex_t d_distributorlock =PTHREAD_MUTEX_INITIALIZER;
+static bool g_mustlockdistributor;
 
 //! The qthread receives questions over the internet via the Nameserver class, and hands them to the Distributor for futher processing
-void *qthread(void *p)
-{
-  DNSDistributor *D=static_cast<DNSDistributor *>(p);
-
+void *qthread(void *number)
+{
   DNSPacket *P;
 
@@ -226 +228 @@
 
   for(;;) {
-    if(!((numreceived++)%50)) { // maintenance tasks
-      S.set("latency",(int)avg_latency);
-      int qcount, acount;
-      D->getQueueSizes(qcount, acount);
-      S.set("qsize-q",qcount);
+    if(number==0) {
+      if(!((numreceived++)%50)) { // maintenance tasks
+        S.set("latency",(int)avg_latency);
+        int qcount, acount;
+        g_distributor->getQueueSizes(qcount, acount);
+        S.set("qsize-q",qcount);
+      }
     }
-    
+
     if(!(P=N->receive(&question))) { // receive a packet         inline
       continue;                    // packet was broken, try again
@@ -265 +269 @@
       continue;
     }
-
-    D->question(P, &sendout); // otherwise, give to the distributor
+    if(g_mustlockdistributor) {
+      Lock l(&d_distributorlock);
+      g_distributor->question(P, &sendout); // otherwise, give to the distributor
+    }
+    else
+      g_distributor->question(P, &sendout); // otherwise, give to the distributor
   }
   return 0;
@@ -319 +327 @@
     
   //  fork(); (this worked :-))
-  DNSDistributor *D= new DNSDistributor(::arg().asNum("distributor-threads")); // the big dispatcher!
-  pthread_create(&qtid,0,qthread,static_cast<void *>(D)); // receives packets
+  g_distributor = new DNSDistributor(::arg().asNum("distributor-threads")); // the big dispatcher!
+  if(::arg().asNum("receiver-threads") > 1) {
+    g_mustlockdistributor=true;
+  }
+  for(int n=0; n < ::arg().asNum("receiver-threads"); ++n)
+    pthread_create(&qtid,0,qthread, reinterpret_cast<void *>(n)); // receives packets
 
   void *p;