Changeset 442

Timestamp:

07/14/05 19:40:18 (8 years ago)

Author:

ahu

Message:

add titles to mthreads
fix confusion in mtasker when waiting for duplicated keys
clarified error handling in waiting for events
improve error checking in mtasker
fix dnsreplay timing fix

Location:

trunk/pdns/pdns

Files:

• dnsreplay.cc (modified) (1 diff)
• mtasker.cc (modified) (8 diffs)
• mtasker.hh (modified) (2 diffs)
• pdns_recursor.cc (modified) (13 diffs)

trunk/pdns/pdns/dnsreplay.cc

@@ -303 +303 @@
           
           nanosleep(&tosleep, 0);
+          lastsent=pr.d_pheader.ts;
         }
-        lastsent=pr.d_pheader.ts;
+
         //      cout<<"sending!"<<endl;
         s_socket->sendTo(string(pr.d_payload, pr.d_payload + pr.d_len), remote);

trunk/pdns/pdns/mtasker.cc

@@ -167 +167 @@
   w.tid=d_tid;
   
+  if(d_waiters.count(key)) { // there was already an exact same waiter
+    return -1;
+  }
   d_waiters[key]=w;
   
-  swapcontext(d_waiters[key].context,&d_kernel); // 'A' will return here when 'key' has arrived, hands over control to kernel first
+  if(swapcontext(d_waiters[key].context,&d_kernel)) { // 'A' will return here when 'key' has arrived, hands over control to kernel first
+    perror("swapcontext");
+    exit(EXIT_FAILURE); // no way we can deal with this 
+  }
   if(val && d_waitstatus==Answer) 
     *val=d_waitval;
@@ -182 +188 @@
 {
   d_runQueue.push(d_tid);
-  swapcontext(d_threads[d_tid],&d_kernel); // give control to the kernel
+  if(swapcontext(d_threads[d_tid].first ,&d_kernel) < 0) { // give control to the kernel
+    perror("swapcontext in  yield");
+    exit(EXIT_FAILURE);
+  }
 }
 
@@ -194 +203 @@
 {
   if(!d_waiters.count(key)) {
+    //    cout<<"Event sent nobody was waiting for!"<<endl;
     return 0;
   }
@@ -205 +215 @@
   
   d_waiters.erase(key);             // removes the waitpoint 
-  swapcontext(&d_kernel,userspace); // swaps back to the above point 'A'
-  
+  if(swapcontext(&d_kernel,userspace)) { // swaps back to the above point 'A'
+    perror("swapcontext in sendEvent");
+    exit(EXIT_FAILURE);
+  }
   delete userspace;
   return 1;
@@ -216 +228 @@
     \param val A void pointer that can be used to pass data to the thread
 */
-template<class Key, class Val>void MTasker<Key,Val>::makeThread(tfunc_t *start, void* val)
+template<class Key, class Val>void MTasker<Key,Val>::makeThread(tfunc_t *start, void* val, const string& name)
 {
   ucontext_t *uc=new ucontext_t;
@@ -230 +242 @@
   makecontext (uc, (void (*)(void))threadWrapper, 4, this, start, d_maxtid, val);
 #endif
-  d_threads[d_maxtid]=uc;
+  d_threads[d_maxtid]=make_pair(uc, name);
   d_runQueue.push(d_maxtid++); // will run at next schedule invocation
 }
@@ -250 +262 @@
   if(!d_runQueue.empty()) {
     d_tid=d_runQueue.front();
-    swapcontext(&d_kernel, d_threads[d_tid]);
+    if(swapcontext(&d_kernel, d_threads[d_tid].first)) {
+      perror("swapcontext in schedule");
+      exit(EXIT_FAILURE);
+    }
+      
     d_runQueue.pop();
     return true;
   }
   if(!d_zombiesQueue.empty()) {
-    delete[] (char *)d_threads[d_zombiesQueue.front()]->uc_stack.ss_sp;
-    delete d_threads[d_zombiesQueue.front()];
+    delete[] (char *)d_threads[d_zombiesQueue.front()].first->uc_stack.ss_sp;
+    delete d_threads[d_zombiesQueue.front()].first;
     d_threads.erase(d_zombiesQueue.front());
     d_zombiesQueue.pop();
@@ -266 +282 @@
       if(i->second.ttd && i->second.ttd<now) {
         d_waitstatus=TimeOut;
-        swapcontext(&d_kernel,i->second.context); // swaps back to the above point 'A'
+        if(swapcontext(&d_kernel,i->second.context)) { // swaps back to the above point 'A'
+          perror("swapcontext in schedule2");
+          exit(EXIT_FAILURE);
+        }
         delete i->second.context;              
         d_waiters.erase(i->first);                  // removes the waitpoint 

trunk/pdns/pdns/mtasker.hh

@@ -53 +53 @@
   typedef std::map<EventKey,Waiter> waiters_t;
   waiters_t d_waiters;
-  std::map<int,ucontext_t*> d_threads;
+  typedef std::map<int,pair<ucontext_t*,string> > mthreads_t;
+  mthreads_t d_threads;
   int d_tid;
   int d_maxtid;
@@ -77 +78 @@
   int sendEvent(const EventKey& key, const EventVal* val=0);
   void getEvents(std::vector<EventKey>& events);
-  void makeThread(tfunc_t *start, void* val);
+  void makeThread(tfunc_t *start, void* val, const string& name="");
   bool schedule();
   bool noProcesses();
   unsigned int numProcesses();
   int getTid(); 
+  void setTitle(const string& name)
+  {
+    d_threads[d_tid].second=name;
+  }
+
 private:
   static void threadWrapper(MTasker *self, tfunc_t *tf, int tid, void* val);

trunk/pdns/pdns/pdns_recursor.cc

@@ -123 +123 @@
   d_tcpclientwritesocks[sock->getHandle()]=pident;
 
-  if(!MT->waitEvent(pident,&packet,1)) { // timeout
+  int ret=MT->waitEvent(pident,&packet,1);
+  if(!ret || ret==-1) { // timeout
     d_tcpclientwritesocks.erase(sock->getHandle());
-    return 0; 
-  }
-  //  cerr<<"asendtcp happy"<<endl;
-  return 1;
-}
-
+  }
+  return ret;
+}
+
+// -1 is error, 0 is timeout, 1 is success
 int arecvtcp(string& data, int len, Socket* sock) 
 {
@@ -142 +142 @@
   d_tcpclientreadsocks[sock->getHandle()]=pident;
 
-  if(!MT->waitEvent(pident,&data,1)) { // timeout
+  int ret=MT->waitEvent(pident,&data,1);
+  if(!ret || ret==-1) { // timeout
     d_tcpclientreadsocks.erase(sock->getHandle());
-    return 0; 
-  }
-
-  // cerr<<"arecvtcp happy, data.size(): "<<data.size()<<endl;
-  return 1;
+  }
+  return ret;
 }
 
 
 /* these two functions are used by LWRes */
+// -1 is error, > 1 is success
 int asendto(const char *data, int len, int flags, struct sockaddr *toaddr, int addrlen, int id) 
 {
@@ -158 +157 @@
 }
 
+// -1 is error, 0 is timeout, 1 is success
 int arecvfrom(char *data, int len, int flags, struct sockaddr *toaddr, Utility::socklen_t *addrlen, int *d_len, int id)
 {
@@ -165 +165 @@
 
   string packet;
-  if(!MT->waitEvent(pident,&packet,1)) { // timeout
-    return 0; 
-  }
-
-  *d_len=packet.size();
-  memcpy(data,packet.c_str(),min(len,*d_len));
-
-  return 1;
+  int ret=MT->waitEvent(pident,&packet,1);
+  if(ret > 0) {
+    *d_len=packet.size();
+    memcpy(data,packet.c_str(),min(len,*d_len));
+  }
+
+  return ret;
 }
 
@@ -221 +220 @@
 
     delete (DNSPacket *)p;
-
+    
     vector<DNSResourceRecord>ret;
     DNSPacket *R=P.replyPacket();
     R->setA(false);
     R->setRA(true);
-
+    //    MT->setTitle("udp question for "+P.qdomain+"|"+P.qtype.getName());
     SyncRes sr;
     if(!quiet)
@@ -274 +273 @@
     }
 
+    //    MT->setTitle("DONE! udp question for "+P.qdomain+"|"+P.qtype.getName());
     if(!quiet) {
       L<<Logger::Error<<"["<<MT->getTid()<<"] answer to "<<(P.d.rd?"":"non-rd ")<<"question '"<<P.qdomain<<"|"<<P.qtype.getName();
@@ -419 +419 @@
   statsWanted=true;
 }
+
+void usr2Handler(int)
+{
+  SyncRes::setLog(true);
+}
+
 
 
@@ -558 +564 @@
     }
     signal(SIGUSR1,usr1Handler);
+    signal(SIGUSR2,usr2Handler);
 
     writePid();
@@ -587 +594 @@
       
       if(!((counter++)%100)) 
-        MT->makeThread(houseKeeping,0);
+        MT->makeThread(houseKeeping,0,"housekeeping");
       if(statsWanted)
         doStats();
@@ -690 +697 @@
               P.setSocket(*i);
               P.d_tcp=false;
-              MT->makeThread(startDoResolve,(void*)new DNSPacket(P));
+              MT->makeThread(startDoResolve,(void*)new DNSPacket(P), "udp");
             }
           }
@@ -738 +745 @@
           }
           else {
-            cerr<<"when reading ret="<<ret<<endl;
+            //      cerr<<"when reading ret="<<ret<<endl;
+            // XXX FIXME I think some stuff needs to happen here - like send an EOF event
           }
         }
@@ -763 +771 @@
           }
           else { 
-            cerr<<"ret="<<ret<<" when writing"<<endl;
+            //      cerr<<"ret="<<ret<<" when writing"<<endl;
+            // XXX FIXME I think some stuff needs to happen here - like send an EOF event
           }
         }
@@ -829 +838 @@
                 else {
                   ++qcounter;
-                  MT->makeThread(startDoResolve,(void*)new DNSPacket(P));
+                  MT->makeThread(startDoResolve,(void*)new DNSPacket(P), "tcp");
                 }
               }