root/trunk/pdns/pdns/pdns_recursor.cc @ 609

Revision 609, 30.8 KB (checked in by ahu, 7 years ago)

add sizing of UDP receive buffers, plus fix (somewhat) the latency
measurements

  • Property svn:eol-style set to native
  • Property svn:keywords set to author date id revision
Line 
1/*
2    PowerDNS Versatile Database Driven Nameserver
3    Copyright (C) 2003 - 2006  PowerDNS.COM BV
4
5    This program is free software; you can redistribute it and/or modify
6    it under the terms of the GNU General Public License version 2
7    as published by the Free Software Foundation
8
9    This program is distributed in the hope that it will be useful,
10    but WITHOUT ANY WARRANTY; without even the implied warranty of
11    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12    GNU General Public License for more details.
13
14    You should have received a copy of the GNU General Public License
15    along with this program; if not, write to the Free Software
16    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
17*/
18
19#include "utility.hh"
20#include <iostream>
21#include <errno.h>
22#include <map>
23#include <set>
24#ifndef WIN32
25#include <netdb.h>
26#endif // WIN32
27#include "recursor_cache.hh"
28#include <stdio.h>
29#include <signal.h>
30#include <stdlib.h>
31#include <unistd.h>
32#include "mtasker.hh"
33#include <utility>
34#include "dnspacket.hh"
35#include "statbag.hh"
36#include "arguments.hh"
37#include "syncres.hh"
38#include <fcntl.h>
39#include <fstream>
40#include "sstuff.hh"
41#include <boost/tuple/tuple.hpp>
42#include <boost/tuple/tuple_comparison.hpp>
43#include <boost/shared_array.hpp>
44#include <boost/lexical_cast.hpp>
45#include "dnsparser.hh"
46#include "dnswriter.hh"
47#include "dnsrecords.hh"
48#include "zoneparser-tng.hh"
49#include "rec_channel.hh"
50
51// using namespace boost;
52
53#ifdef __FreeBSD__           // see cvstrac ticket #26
54#include <pthread.h>
55#include <semaphore.h>
56#endif
57
58MemRecursorCache RC;
59RecursorStats g_stats;
60bool g_quiet;
61string s_programname="pdns_recursor";
62
63struct DNSComboWriter {
64  DNSComboWriter(const char* data, uint16_t len, const struct timeval& now) : d_mdp(data, len), d_now(now), d_tcp(false), d_socket(-1)
65  {}
66  MOADNSParser d_mdp;
67  void setRemote(struct sockaddr* sa, socklen_t len)
68  {
69    memcpy((void *)d_remote, (void *)sa, len);
70    d_socklen=len;
71  }
72
73  void setSocket(int sock)
74  {
75    d_socket=sock;
76  }
77
78  string getRemote() const
79  {
80    return sockAddrToString((struct sockaddr_in *)d_remote, d_socklen);
81  }
82
83  struct timeval d_now;
84  char d_remote[sizeof(sockaddr_in6)];
85  socklen_t d_socklen;
86  bool d_tcp;
87  int d_socket;
88};
89
90
91#ifndef WIN32
92#ifndef __FreeBSD__
93extern "C" {
94  int sem_init(sem_t*, int, unsigned int){return 0;}
95  int sem_wait(sem_t*){return 0;}
96  int sem_trywait(sem_t*){return 0;}
97  int sem_post(sem_t*){return 0;}
98  int sem_getvalue(sem_t*, int*){return 0;}
99  pthread_t pthread_self(void){return (pthread_t) 0;}
100  int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *mutexattr){ return 0; }
101  int pthread_mutex_lock(pthread_mutex_t *mutex){ return 0; }
102  int pthread_mutex_unlock(pthread_mutex_t *mutex) { return 0; }
103  int pthread_mutex_destroy(pthread_mutex_t *mutex) { return 0; }
104}
105#endif // __FreeBSD__
106#endif // WIN32
107
108StatBag S;
109ArgvMap &arg()
110{
111  static ArgvMap theArg;
112  return theArg;
113}
114static int d_clientsock;
115static vector<int> d_udpserversocks;
116
117typedef vector<int> tcpserversocks_t;
118static tcpserversocks_t s_tcpserversocks;
119
120static map<int,PacketID> d_tcpclientreadsocks, d_tcpclientwritesocks;
121
122MTasker<PacketID,string>* MT;
123
124int asendtcp(const string& data, Socket* sock) 
125{
126  PacketID pident;
127  pident.sock=sock;
128  pident.outMSG=data;
129  string packet;
130
131  d_tcpclientwritesocks[sock->getHandle()]=pident;
132
133  int ret=MT->waitEvent(pident,&packet,1);
134  if(!ret || ret==-1) { // timeout
135    d_tcpclientwritesocks.erase(sock->getHandle());
136  }
137  return ret;
138}
139
140// -1 is error, 0 is timeout, 1 is success
141int arecvtcp(string& data, int len, Socket* sock) 
142{
143  data="";
144  PacketID pident;
145  pident.sock=sock;
146  pident.inNeeded=len;
147
148  d_tcpclientreadsocks[sock->getHandle()]=pident;
149
150  int ret=MT->waitEvent(pident,&data,1);
151  if(!ret || ret==-1) { // timeout
152    d_tcpclientreadsocks.erase(sock->getHandle());
153  }
154  return ret;
155}
156
157
158/* these two functions are used by LWRes */
159// -1 is error, > 1 is success
160int asendto(const char *data, int len, int flags, struct sockaddr *toaddr, int addrlen, int id) 
161{
162  return sendto(d_clientsock, data, len, flags, toaddr, addrlen);
163}
164
165// -1 is error, 0 is timeout, 1 is success
166int arecvfrom(char *data, int len, int flags, struct sockaddr *toaddr, Utility::socklen_t *addrlen, int *d_len, int id)
167{
168  PacketID pident;
169  pident.id=id;
170  memcpy(&pident.remote, toaddr, sizeof(pident.remote));
171
172  string packet;
173  int ret=MT->waitEvent(pident, &packet, 1);
174  if(ret > 0) {
175    *d_len=packet.size();
176    memcpy(data,packet.c_str(),min(len,*d_len));
177  }
178  return ret;
179}
180
181void setReceiveBuffer(int fd, uint32_t size)
182{
183  uint32_t psize;
184  socklen_t len;
185  getsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char*)&psize, &len);
186  if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char*)&size, sizeof(size)) < 0 )
187    L<<Logger::Error<<"Warning: unable to raise socket buffer size to "<<size<<": "<<strerror(errno)<<"\n";
188}
189
190
191static void writePid(void)
192{
193  string fname=::arg()["socket-dir"]+"/"+s_programname+".pid";
194  ofstream of(fname.c_str());
195  if(of)
196    of<< getpid() <<endl;
197  else
198    L<<Logger::Error<<"Requested to write pid for "<<getpid()<<" to "<<fname<<" failed: "<<strerror(errno)<<endl;
199}
200
201void primeHints(void)
202{
203  // prime root cache
204  set<DNSResourceRecord>nsset;
205
206  if(::arg()["hint-file"].empty()) {
207    static char*ips[]={"198.41.0.4", "192.228.79.201", "192.33.4.12", "128.8.10.90", "192.203.230.10", "192.5.5.241", "192.112.36.4", "128.63.2.53", 
208                       "192.36.148.17","192.58.128.30", "193.0.14.129", "198.32.64.12", "202.12.27.33"};
209    DNSResourceRecord arr, nsrr;
210    arr.qtype=QType::A;
211    arr.ttl=time(0)+3600000;
212    nsrr.qtype=QType::NS;
213    nsrr.ttl=time(0)+3600000;
214   
215    for(char c='a';c<='m';++c) {
216      static char templ[40];
217      strncpy(templ,"a.root-servers.net", sizeof(templ) - 1);
218      *templ=c;
219      arr.qname=nsrr.content=templ;
220      arr.content=ips[c-'a'];
221      set<DNSResourceRecord> aset;
222      aset.insert(arr);
223      RC.replace(string(templ), QType(QType::A), aset);
224     
225      nsset.insert(nsrr);
226    }
227  }
228  else {
229    ZoneParserTNG zpt(::arg()["hint-file"]);
230    DNSResourceRecord rr;
231    set<DNSResourceRecord> aset;
232
233    while(zpt.get(rr)) {
234      rr.ttl+=time(0);
235      if(rr.qtype.getCode()==QType::A) {
236        set<DNSResourceRecord> aset;
237        aset.insert(rr);
238        RC.replace(rr.qname, QType(QType::A), aset);
239      }
240      if(rr.qtype.getCode()==QType::NS) {
241        nsset.insert(rr);
242      }
243    }
244  }
245  RC.replace("", QType(QType::NS), nsset); // and stuff in the cache
246}
247
248void startDoResolve(void *p)
249{
250  try {
251    DNSComboWriter* dc=(DNSComboWriter *)p;
252
253    uint16_t maxudpsize=512;
254    MOADNSParser::EDNSOpts edo;
255    if(dc->d_mdp.getEDNSOpts(&edo)) {
256      maxudpsize=edo.d_packetsize;
257    }
258
259    vector<DNSResourceRecord> ret;
260   
261    vector<uint8_t> packet;
262    DNSPacketWriter pw(packet, dc->d_mdp.d_qname, dc->d_mdp.d_qtype, dc->d_mdp.d_qclass);
263
264    pw.getHeader()->aa=0;
265    pw.getHeader()->ra=1;
266    pw.getHeader()->qr=1;
267    pw.getHeader()->id=dc->d_mdp.d_header.id;
268    pw.getHeader()->rd=dc->d_mdp.d_header.rd;
269
270    //    MT->setTitle("udp question for "+P.qdomain+"|"+P.qtype.getName());
271    SyncRes sr(dc->d_now);
272    if(!g_quiet)
273      L<<Logger::Error<<"["<<MT->getTid()<<"] " << (dc->d_tcp ? "TCP " : "") << "question for '"<<dc->d_mdp.d_qname<<"|"
274       <<DNSRecordContent::NumberToType(dc->d_mdp.d_qtype)<<"' from "<<dc->getRemote()<<endl;
275
276    sr.setId(MT->getTid());
277    if(!dc->d_mdp.d_header.rd)
278      sr.setCacheOnly();
279
280    int res=sr.beginResolve(dc->d_mdp.d_qname, QType(dc->d_mdp.d_qtype), ret);
281    if(res<0) {
282      pw.getHeader()->rcode=RCode::ServFail;
283      g_stats.servFails++;
284    }
285    else {
286      pw.getHeader()->rcode=res;
287      switch(res) {
288      case RCode::ServFail:
289        g_stats.servFails++;
290        break;
291      case RCode::NXDomain:
292        g_stats.nxDomains++;
293        break;
294      case RCode::NoError:
295        g_stats.noErrors++;
296        break;
297      }
298     
299      if(ret.size()) {
300        shuffle(ret);
301        for(vector<DNSResourceRecord>::const_iterator i=ret.begin();i!=ret.end();++i) {
302          pw.startRecord(i->qname, i->qtype.getCode(), i->ttl, 1, (DNSPacketWriter::Place)i->d_place);
303          shared_ptr<DNSRecordContent> drc(DNSRecordContent::mastermake(i->qtype.getCode(), 1, i->content)); 
304          drc->toPacket(pw);
305          if(!dc->d_tcp && pw.size() > maxudpsize) {
306            pw.rollback();
307            pw.getHeader()->tc=1;
308            goto sendit; // need to jump over pw.commit
309          }
310        }
311        pw.commit();
312      }
313    }
314  sendit:;
315    if(!dc->d_tcp) {
316      sendto(dc->d_socket, &*packet.begin(), packet.size(), 0, (struct sockaddr *)(dc->d_remote), dc->d_socklen);
317    }
318    else {
319      char buf[2];
320      buf[0]=packet.size()/256;
321      buf[1]=packet.size()%256;
322
323      struct iovec iov[2];
324
325      iov[0].iov_base=(void*)buf;              iov[0].iov_len=2;
326      iov[1].iov_base=(void*)&*packet.begin(); iov[1].iov_len = packet.size();
327
328      int ret=writev(dc->d_socket, iov, 2);
329
330      if(ret <= 0 ) 
331        L<<Logger::Error<<"Error writing TCP answer to "<<dc->getRemote()<<": "<< (ret ? strerror(errno) : "EOF") <<endl;
332      else if((unsigned int)ret != 2 + packet.size())
333        L<<Logger::Error<<"Oops, partial answer sent to "<<dc->getRemote()<<" - probably would have trouble receiving our answer anyhow (size="<<packet.size()<<")"<<endl;
334    }
335
336    //    MT->setTitle("DONE! udp question for "+P.qdomain+"|"+P.qtype.getName());
337    if(!g_quiet) {
338      L<<Logger::Error<<"["<<MT->getTid()<<"] answer to "<<(dc->d_mdp.d_header.rd?"":"non-rd ")<<"question '"<<dc->d_mdp.d_qname<<"|"<<DNSRecordContent::NumberToType(dc->d_mdp.d_qtype);
339      L<<"': "<<ntohs(pw.getHeader()->ancount)<<" answers, "<<ntohs(pw.getHeader()->arcount)<<" additional, took "<<sr.d_outqueries<<" packets, "<<
340        sr.d_throttledqueries<<" throttled, "<<sr.d_timeouts<<" timeouts, "<<sr.d_tcpoutqueries<<" tcp connections, rcode="<<res<<endl;
341    }
342   
343    sr.d_outqueries ? RC.cacheMisses++ : RC.cacheHits++; 
344    float spent=makeFloat(sr.d_now-dc->d_now);
345    if(spent < 0.001)
346      g_stats.answers0_1++;
347    else if(spent < 0.010)
348      g_stats.answers1_10++;
349    else if(spent < 0.1)
350      g_stats.answers10_100++;
351    else if(spent < 1.0)
352      g_stats.answers100_1000++;
353    else
354      g_stats.answersSlow++;
355
356    uint64_t newLat=(uint64_t)(spent*1000000);
357    g_stats.avgLatencyUsec=(1-0.0001)*g_stats.avgLatencyUsec + 0.0001*newLat;
358    delete dc;
359  }
360  catch(AhuException &ae) {
361    L<<Logger::Error<<"startDoResolve problem: "<<ae.reason<<endl;
362  }
363  catch(exception& e) {
364    L<<Logger::Error<<"STL error: "<<e.what()<<endl;
365  }
366  catch(...) {
367    L<<Logger::Error<<"Any other exception in a resolver context"<<endl;
368  }
369}
370
371RecursorControlChannel s_rcc;
372
373void makeControlChannelSocket()
374{
375  s_rcc.listen("pdns_recursor.controlsocket");
376}
377
378void makeClientSocket()
379{
380  d_clientsock=socket(AF_INET, SOCK_DGRAM,0);
381  if(d_clientsock<0) 
382    throw AhuException("Making a socket for resolver: "+stringerror());
383  setReceiveBuffer(d_clientsock, 250000); 
384  struct sockaddr_in sin;
385  memset((char *)&sin,0, sizeof(sin));
386 
387  sin.sin_family = AF_INET;
388
389  if(!IpToU32(::arg()["query-local-address"], &sin.sin_addr.s_addr))
390    throw AhuException("Unable to resolve local address '"+ ::arg()["query-local-address"] +"'"); 
391
392  int tries=10;
393  while(--tries) {
394    uint16_t port=10000+Utility::random()%10000;
395    sin.sin_port = htons(port); 
396   
397    if (::bind(d_clientsock, (struct sockaddr *)&sin, sizeof(sin)) >= 0) 
398      break;
399   
400  }
401  if(!tries)
402    throw AhuException("Resolver binding to local socket: "+stringerror());
403
404  Utility::setNonBlocking(d_clientsock);
405
406  L<<Logger::Error<<"Sending UDP queries from "<<inet_ntoa(sin.sin_addr)<<":"<< ntohs(sin.sin_port)  <<endl;
407}
408
409void makeTCPServerSockets()
410{
411  vector<string>locals;
412  stringtok(locals,::arg()["local-address"]," ,");
413
414  if(locals.empty())
415    throw AhuException("No local address specified");
416 
417  for(vector<string>::const_iterator i=locals.begin();i!=locals.end();++i) {
418    int fd=socket(AF_INET, SOCK_STREAM,0);
419    if(fd<0) 
420      throw AhuException("Making a server socket for resolver: "+stringerror());
421 
422    struct sockaddr_in sin;
423    memset((char *)&sin,0, sizeof(sin));
424   
425    sin.sin_family = AF_INET;
426    if(!IpToU32(*i, &sin.sin_addr.s_addr))
427      throw AhuException("Unable to resolve local address '"+ *i +"'"); 
428
429    int tmp=1;
430    if(setsockopt(fd,SOL_SOCKET,SO_REUSEADDR,(char*)&tmp,sizeof tmp)<0) {
431      L<<Logger::Error<<"Setsockopt failed for TCP listening socket"<<endl;
432      exit(1); 
433    }
434   
435    sin.sin_port = htons(::arg().asNum("local-port")); 
436   
437    if (::bind(fd, (struct sockaddr *)&sin, sizeof(sin))<0) 
438      throw AhuException("Binding TCP server socket for "+*i+": "+stringerror());
439   
440    Utility::setNonBlocking(fd);
441    listen(fd, 128);
442    s_tcpserversocks.push_back(fd);
443    L<<Logger::Error<<"Listening for TCP queries on "<<inet_ntoa(sin.sin_addr)<<":"<<::arg().asNum("local-port")<<endl;
444  }
445}
446
447void makeUDPServerSockets()
448{
449  vector<string>locals;
450  stringtok(locals,::arg()["local-address"]," ,");
451
452  if(locals.empty())
453    throw AhuException("No local address specified");
454 
455  if(::arg()["local-address"]=="0.0.0.0") {
456    L<<Logger::Warning<<"It is advised to bind to explicit addresses with the --local-address option"<<endl;
457  }
458
459  for(vector<string>::const_iterator i=locals.begin();i!=locals.end();++i) {
460    int fd=socket(AF_INET, SOCK_DGRAM,0);
461    if(fd<0) 
462      throw AhuException("Making a server socket for resolver: "+stringerror());
463    setReceiveBuffer(fd, 250000);
464    struct sockaddr_in sin;
465    memset((char *)&sin,0, sizeof(sin));
466   
467    sin.sin_family = AF_INET;
468    if(!IpToU32(*i, &sin.sin_addr.s_addr))
469      throw AhuException("Unable to resolve local address '"+ *i +"'"); 
470   
471    sin.sin_port = htons(::arg().asNum("local-port")); 
472   
473    if (::bind(fd, (struct sockaddr *)&sin, sizeof(sin))<0) 
474      throw AhuException("Resolver binding to server socket for "+*i+": "+stringerror());
475   
476    Utility::setNonBlocking(fd);
477    d_udpserversocks.push_back(fd);
478    L<<Logger::Error<<"Listening for UDP queries on "<<inet_ntoa(sin.sin_addr)<<":"<<::arg().asNum("local-port")<<endl;
479  }
480}
481
482
483#ifndef WIN32
484void daemonize(void)
485{
486  if(fork())
487    exit(0); // bye bye
488 
489  setsid(); 
490
491  // cleanup open fds, but skip sockets
492  close(0);
493  close(1);
494  close(2);
495}
496#endif
497
498uint64_t counter, qcounter;
499bool statsWanted;
500
501
502void usr1Handler(int)
503{
504  statsWanted=true;
505}
506
507
508
509void usr2Handler(int)
510{
511  SyncRes::setLog(true);
512  g_quiet=false;
513  ::arg().set("quiet")="no";
514
515}
516
517void doStats(void)
518{
519  if(qcounter) {
520    L<<Logger::Error<<"stats: "<<qcounter<<" questions, "<<RC.size()<<" cache entries, "<<SyncRes::s_negcache.size()<<" negative entries, "
521     <<(int)((RC.cacheHits*100.0)/(RC.cacheHits+RC.cacheMisses))<<"% cache hits"<<endl;
522    L<<Logger::Error<<"stats: throttle map: "<<SyncRes::s_throttle.size()<<", ns speeds: "
523     <<SyncRes::s_nsSpeeds.size()<<", bytes: "<<RC.bytes()<<endl;
524    L<<Logger::Error<<"stats: outpacket/query ratio "<<(int)(SyncRes::s_outqueries*100.0/SyncRes::s_queries)<<"%";
525    L<<Logger::Error<<", "<<(int)(SyncRes::s_throttledqueries*100.0/(SyncRes::s_outqueries+SyncRes::s_throttledqueries))<<"% throttled, "
526     <<SyncRes::s_nodelegated<<" no-delegation drops"<<endl;
527    L<<Logger::Error<<"stats: "<<SyncRes::s_tcpoutqueries<<" outgoing tcp connections, "<<MT->numProcesses()<<" queries running, "<<SyncRes::s_outgoingtimeouts<<" outgoing timeouts"<<endl;
528  }
529  else if(statsWanted) 
530    L<<Logger::Error<<"stats: no stats yet!"<<endl;
531
532  statsWanted=false;
533}
534
535static void houseKeeping(void *)
536{
537  static time_t last_stat, last_rootupdate, last_prune;
538  struct timeval now;
539  gettimeofday(&now, 0);
540
541  if(now.tv_sec - last_prune > 60) { 
542    DTime dt;
543    dt.setTimeval(now);
544    RC.doPrune();
545    int pruned=0;
546    for(SyncRes::negcache_t::iterator i = SyncRes::s_negcache.begin(); i != SyncRes::s_negcache.end();) 
547      if(i->second.ttd > now.tv_sec) {
548        SyncRes::s_negcache.erase(i++);
549        pruned++;
550      }
551      else
552        ++i;
553
554    time_t limit=now.tv_sec-300;
555    for(SyncRes::nsspeeds_t::iterator i = SyncRes::s_nsSpeeds.begin() ; i!= SyncRes::s_nsSpeeds.end(); )
556      if(i->second.stale(limit))
557        SyncRes::s_nsSpeeds.erase(i++);
558      else
559        ++i;
560
561    //    cerr<<"Pruned "<<pruned<<" records, left "<<SyncRes::s_negcache.size()<<"\n";
562//    cout<<"Prune took "<<dt.udiff()<<"usec\n";
563    last_prune=time(0);
564  }
565  if(now.tv_sec - last_stat>1800) { 
566    doStats();
567    last_stat=time(0);
568  }
569  if(now.tv_sec -last_rootupdate>7200) {
570    SyncRes sr(now);
571    vector<DNSResourceRecord> ret;
572
573    sr.setNoCache();
574    int res=sr.beginResolve("", QType(QType::NS), ret);
575    if(!res) {
576      L<<Logger::Error<<"Refreshed . records"<<endl;
577      last_rootupdate=now.tv_sec;
578    }
579    else
580      L<<Logger::Error<<"Failed to update . records, RCODE="<<res<<endl;
581  }
582}
583
584struct TCPConnection
585{
586  int fd;
587  enum {BYTE0, BYTE1, GETQUESTION} state;
588  int qlen;
589  int bytesread;
590  struct sockaddr_in remote;
591  char data[65535];
592  time_t startTime;
593};
594
595#if 0
596#include <execinfo.h>
597
598  multimap<uint32_t,string> rev;
599  for(map<string,uint32_t>::const_iterator i=casesptr->begin(); i!=casesptr->end(); ++i) {
600    rev.insert(make_pair(i->second,i->first));
601  }
602  for(multimap<uint32_t,string>::const_iterator i=rev.begin(); i!= rev.end(); ++i)
603    cout<<i->first<<" times: \n"<<i->second<<"\n";
604
605  cout.flush();
606
607map<string,uint32_t>* casesptr;
608static string maketrace()
609{
610  void *array[20]; //only care about last 17 functions (3 taken with tracing support)
611  size_t size;
612  char **strings;
613  size_t i;
614
615  size = backtrace (array, 5);
616  strings = backtrace_symbols (array, size); //Need -rdynamic gcc (linker) flag for this to work
617
618  string ret;
619
620  for (i = 0; i < size; i++) //skip useless functions
621    ret+=string(strings[i])+"\n";
622  return ret;
623}
624
625extern "C" {
626
627int gettimeofday (struct timeval *__restrict __tv,
628                  __timezone_ptr_t __tz)
629{
630  static map<string, uint32_t> s_cases;
631  casesptr=&s_cases;
632  s_cases[maketrace()]++;
633  __tv->tv_sec=time(0);
634  return 0;
635}
636
637}
638#endif
639
640int main(int argc, char **argv) 
641{
642  reportBasicTypes();
643
644  int ret = EXIT_SUCCESS;
645#ifdef WIN32
646    WSADATA wsaData;
647    WSAStartup( MAKEWORD( 2, 0 ), &wsaData );
648#endif // WIN32
649
650  try {
651    Utility::srandom(time(0));
652    ::arg().set("soa-minimum-ttl","Don't change")="0";
653    ::arg().set("soa-serial-offset","Don't change")="0";
654    ::arg().set("no-shuffle","Don't change")="off";
655    ::arg().set("aaaa-additional-processing","turn on to do AAAA additional processing (slow)")="off";
656    ::arg().set("local-port","port to listen on")="53";
657    ::arg().set("local-address","IP addresses to listen on, separated by spaces or commas")="0.0.0.0";
658    ::arg().set("trace","if we should output heaps of logging")="off";
659    ::arg().set("daemon","Operate as a daemon")="yes";
660    ::arg().set("chroot","switch to chroot jail")="";
661    ::arg().set("setgid","If set, change group id to this gid for more security")="";
662    ::arg().set("setuid","If set, change user id to this uid for more security")="";
663    ::arg().set("quiet","Suppress logging of questions and answers")="true";
664    ::arg().set("config-dir","Location of configuration directory (recursor.conf)")=SYSCONFDIR;
665    ::arg().set("socket-dir","Where the controlsocket will live")=LOCALSTATEDIR;
666    ::arg().set("delegation-only","Which domains we only accept delegations from")="";
667    ::arg().set("query-local-address","Source IP address for sending queries")="0.0.0.0";
668    ::arg().set("client-tcp-timeout","Timeout in seconds when talking to TCP clients")="2";
669    ::arg().set("max-tcp-clients","Maximum number of simultaneous TCP clients")="128";
670    ::arg().set("hint-file", "If set, load root hints from this file")="";
671
672    ::arg().setCmd("help","Provide a helpful message");
673    L.toConsole(Logger::Warning);
674    ::arg().laxParse(argc,argv); // do a lax parse
675
676    string configname=::arg()["config-dir"]+"/recursor.conf";
677    cleanSlashes(configname);
678
679    if(!::arg().file(configname.c_str())) 
680      L<<Logger::Warning<<"Unable to parse configuration file '"<<configname<<"'"<<endl;
681
682    ::arg().parse(argc,argv);
683
684    ::arg().set("delegation-only")=toLower(::arg()["delegation-only"]);
685
686    if(::arg().mustDo("help")) {
687      cerr<<"syntax:"<<endl<<endl;
688      cerr<<::arg().helpstring(::arg()["help"])<<endl;
689      exit(99);
690    }
691
692    L.setName("pdns_recursor");
693
694    L<<Logger::Warning<<"PowerDNS recursor "<<VERSION<<" (C) 2001-2006 PowerDNS.COM BV ("<<__DATE__", "__TIME__;
695#ifdef __GNUC__
696    L<<", gcc "__VERSION__;
697#endif // add other compilers here
698    L<<") starting up"<<endl;
699
700    L<<Logger::Warning<<"Operating in "<<(sizeof(unsigned long)*8) <<" bits mode"<<endl;
701  L<<Logger::Warning<<"PowerDNS comes with ABSOLUTELY NO WARRANTY. "
702    "This is free software, and you are welcome to redistribute it "
703    "according to the terms of the GPL version 2."<<endl;
704
705
706  g_quiet=::arg().mustDo("quiet");
707  if(::arg().mustDo("trace")) {
708      SyncRes::setLog(true);
709      ::arg().set("quiet")="no";
710      g_quiet=false;
711  }
712
713    makeClientSocket();
714    makeUDPServerSockets();
715    makeTCPServerSockets();
716    makeControlChannelSocket();
717   
718    MT=new MTasker<PacketID,string>(100000);
719
720    char data[1500];
721    struct sockaddr_in fromaddr;
722   
723    PacketID pident;
724    primeHints();   
725    L<<Logger::Warning<<"Done priming cache with root hints"<<endl;
726#ifndef WIN32
727    if(::arg().mustDo("daemon")) {
728      L.toConsole(Logger::Critical);
729      daemonize();
730    }
731    signal(SIGUSR1,usr1Handler);
732    signal(SIGUSR2,usr2Handler);
733    signal(SIGPIPE,SIG_IGN);
734
735    writePid();
736#endif
737
738    int newgid=0;
739    if(!::arg()["setgid"].empty())
740      newgid=Utility::makeGidNumeric(::arg()["setgid"]);
741    int newuid=0;
742    if(!::arg()["setuid"].empty())
743      newuid=Utility::makeUidNumeric(::arg()["setuid"]);
744
745
746    if (!::arg()["chroot"].empty()) {
747        if (chroot(::arg()["chroot"].c_str())<0) {
748            L<<Logger::Error<<"Unable to chroot to '"+::arg()["chroot"]+"': "<<strerror (errno)<<", exiting"<<endl;
749            exit(1);
750        }
751    }
752
753    Utility::dropPrivs(newuid, newgid);
754
755    vector<TCPConnection> tcpconnections;
756    counter=0;
757    struct timeval now;
758    unsigned int maxTcpClients=::arg().asNum("max-tcp-clients");
759    int tcpLimit=::arg().asNum("client-tcp-timeout");
760    for(;;) {
761      while(MT->schedule()); // housekeeping, let threads do their thing
762     
763      if(!((counter++)%500)) 
764        MT->makeThread(houseKeeping,0,"housekeeping");
765      if(statsWanted) {
766        doStats();
767      }
768
769      Utility::socklen_t addrlen=sizeof(fromaddr);
770      int d_len;
771     
772      struct timeval tv;
773      tv.tv_sec=0;
774      tv.tv_usec=500000;
775     
776      fd_set readfds, writefds;
777      FD_ZERO( &readfds );
778      FD_ZERO( &writefds );
779      FD_SET( d_clientsock, &readfds );
780      FD_SET( s_rcc.d_fd, &readfds);
781      int fdmax=max(d_clientsock, s_rcc.d_fd);
782
783      if(!tcpconnections.empty())
784        gettimeofday(&now, 0);
785
786      vector<TCPConnection> sweeped;
787
788      for(vector<TCPConnection>::iterator i=tcpconnections.begin();i!=tcpconnections.end();++i) {
789        if(now.tv_sec < i->startTime + tcpLimit) {
790          FD_SET(i->fd, &readfds);
791          fdmax=max(fdmax,i->fd);
792          sweeped.push_back(*i);
793        }
794        else {
795          L<<Logger::Error<<"TCP timeout from client "<<inet_ntoa(i->remote.sin_addr)<<endl;
796          close(i->fd);
797        }
798      }
799      sweeped.swap(tcpconnections);
800
801      for(vector<int>::const_iterator i=d_udpserversocks.begin(); i!=d_udpserversocks.end(); ++i) {
802        FD_SET( *i, &readfds );
803        fdmax=max(fdmax,*i);
804      }
805      if(tcpconnections.size() < maxTcpClients) 
806        for(tcpserversocks_t::const_iterator i=s_tcpserversocks.begin(); i!=s_tcpserversocks.end(); ++i) {
807          FD_SET(*i, &readfds );
808          fdmax=max(fdmax,*i);
809        }
810
811      for(map<int,PacketID>::const_iterator i=d_tcpclientreadsocks.begin(); i!=d_tcpclientreadsocks.end(); ++i) {
812        // cerr<<"Adding TCP socket "<<i->first<<" to read select set"<<endl;
813        FD_SET( i->first, &readfds );
814        fdmax=max(fdmax,i->first);
815      }
816
817      for(map<int,PacketID>::const_iterator i=d_tcpclientwritesocks.begin(); i!=d_tcpclientwritesocks.end(); ++i) {
818        // cerr<<"Adding TCP socket "<<i->first<<" to write select set"<<endl;
819        FD_SET( i->first, &writefds );
820        fdmax=max(fdmax,i->first);
821      }
822
823      int selret = select(  fdmax + 1, &readfds, &writefds, NULL, &tv );
824      gettimeofday(&now, 0);
825      if(selret<=0) 
826        if (selret == -1 && errno!=EINTR) 
827          throw AhuException("Select returned: "+stringerror());
828        else
829          continue;
830
831      if(FD_ISSET(s_rcc.d_fd, &readfds)) {
832        string remote;
833        string msg=s_rcc.recv(&remote);
834        RecursorControlParser rcp;
835        s_rcc.send(rcp.getAnswer(msg), &remote);
836      }
837
838      if(FD_ISSET(d_clientsock,&readfds)) { // do we have a UDP question response?
839        d_len=recvfrom(d_clientsock, data, sizeof(data), 0, (sockaddr *)&fromaddr, &addrlen);   
840        if(d_len<0) 
841          continue;
842
843        try {
844          DNSComboWriter dc(data, d_len, now);
845          dc.setRemote((struct sockaddr *)&fromaddr, addrlen);
846
847          if(dc.d_mdp.d_header.qr) {
848            pident.remote=fromaddr;
849            pident.id=dc.d_mdp.d_header.id;
850            string packet;
851            packet.assign(data, d_len);
852            MT->sendEvent(pident, &packet);
853          }
854          else 
855            L<<Logger::Warning<<"Ignoring question on outgoing socket from "<<dc.getRemote()<<endl;
856        }
857        catch(MOADNSException& mde) {
858          L<<Logger::Error<<"Unparseable packet from remote server "<< sockAddrToString((struct sockaddr_in*) &fromaddr, addrlen) <<": "<<mde.what()<<endl;
859        }
860      }
861     
862      for(vector<int>::const_iterator i=d_udpserversocks.begin(); i!=d_udpserversocks.end(); ++i) {
863        if(FD_ISSET(*i,&readfds)) { // do we have a new question on udp?
864          d_len=recvfrom(*i, data, sizeof(data), 0, (sockaddr *)&fromaddr, &addrlen);   
865          if(d_len<0) 
866            continue;
867
868          g_stats.queryrate.pulse(now);
869
870          try {
871            DNSComboWriter* dc = new DNSComboWriter(data, d_len, now);
872
873            dc->setRemote((struct sockaddr *)&fromaddr, addrlen);
874
875            if(dc->d_mdp.d_header.qr)
876              L<<Logger::Error<<"Ignoring answer on server socket!"<<endl;
877            else {
878              ++qcounter;
879              dc->setSocket(*i);
880              dc->d_tcp=false;
881              MT->makeThread(startDoResolve, (void*) dc, "udp");
882            }
883          }
884          catch(MOADNSException& mde) {
885            L<<Logger::Error<<"Unparseable packet from remote server "<< sockAddrToString((struct sockaddr_in*) &fromaddr, addrlen) <<": "<<mde.what()<<endl;
886          }
887        }
888      }
889
890      for(tcpserversocks_t::const_iterator i=s_tcpserversocks.begin(); i!=s_tcpserversocks.end(); ++i) { 
891        if(FD_ISSET(*i ,&readfds)) { // do we have a new TCP connection?
892          struct sockaddr_in addr;
893          socklen_t addrlen=sizeof(addr);
894          int newsock=accept(*i, (struct sockaddr*)&addr, &addrlen);
895         
896          if(newsock>0) {
897            Utility::setNonBlocking(newsock);
898            TCPConnection tc;
899            tc.fd=newsock;
900            tc.state=TCPConnection::BYTE0;
901            tc.remote=addr;
902            tc.startTime=now.tv_sec;
903            tcpconnections.push_back(tc);
904          }
905        }
906      }
907
908      // have any question answers come in over TCP?
909      for(map<int,PacketID>::iterator i=d_tcpclientreadsocks.begin(); i!=d_tcpclientreadsocks.end();) { 
910        bool haveErased=false;
911        if(FD_ISSET(i->first, &readfds)) { // can we receive
912          shared_array<char> buffer(new char[i->second.inNeeded]);
913
914          int ret=read(i->first, buffer.get(), min(i->second.inNeeded,200));
915          // cerr<<"Read returned "<<ret<<endl;
916          if(ret > 0) {
917            i->second.inMSG.append(&buffer[0], &buffer[ret]);
918            i->second.inNeeded-=ret;
919            if(!i->second.inNeeded) {
920              // cerr<<"Got entire load of "<<i->second.inMSG.size()<<" bytes"<<endl;
921              PacketID pid=i->second;
922              string msg=i->second.inMSG;
923             
924              d_tcpclientreadsocks.erase((i++));
925              haveErased=true;
926              MT->sendEvent(pid, &msg);   // XXX DODGY
927            }
928            else {
929              // cerr<<"Still have "<<i->second.inNeeded<<" left to go"<<endl;
930            }
931          }
932          else {
933            //      cerr<<"when reading ret="<<ret<<endl;
934            // XXX FIXME I think some stuff needs to happen here - like send an EOF event
935          }
936        }
937        if(!haveErased)
938          ++i;
939      }
940     
941      // is there data we can send to remote nameservers over TCP?
942      for(map<int,PacketID>::iterator i=d_tcpclientwritesocks.begin(); i!=d_tcpclientwritesocks.end(); ) { 
943        bool haveErased=false;
944        if(FD_ISSET(i->first, &writefds)) { // can we send over TCP
945          // cerr<<"Socket "<<i->first<<" available for writing"<<endl;
946          int ret=write(i->first, i->second.outMSG.c_str(), i->second.outMSG.size() - i->second.outPos);
947          if(ret > 0) {
948            i->second.outPos+=ret;
949            if(i->second.outPos==i->second.outMSG.size()) {
950              // cerr<<"Sent out entire load of "<<i->second.outMSG.size()<<" bytes"<<endl;
951              PacketID pid=i->second;
952              d_tcpclientwritesocks.erase(i++);   // erase!
953              haveErased=true;
954              MT->sendEvent(pid, 0);
955            }
956
957          }
958          else { 
959            //      cerr<<"ret="<<ret<<" when writing"<<endl;
960            // XXX FIXME I think some stuff needs to happen here - like send an EOF event
961          }
962        }
963        if(!haveErased)
964          ++i;
965      }
966     
967      // very braindead TCP incoming question parser
968      for(vector<TCPConnection>::iterator i=tcpconnections.begin();i!=tcpconnections.end();++i) {
969        if(FD_ISSET(i->fd, &readfds)) {
970          if(i->state==TCPConnection::BYTE0) {
971            int bytes=read(i->fd,i->data,2);
972            if(bytes==1)
973              i->state=TCPConnection::BYTE1;
974            if(bytes==2) { 
975              i->qlen=(i->data[0]<<8)+i->data[1];
976              i->bytesread=0;
977              i->state=TCPConnection::GETQUESTION;
978            }
979            if(!bytes || bytes < 0) {
980              close(i->fd);
981              tcpconnections.erase(i);
982              break;
983            }
984          }
985          else if(i->state==TCPConnection::BYTE1) {
986            int bytes=read(i->fd,i->data+1,1);
987            if(bytes==1) {
988              i->state=TCPConnection::GETQUESTION;
989              i->qlen=(i->data[0]<<8)+i->data[1];
990              i->bytesread=0;
991            }
992            if(!bytes || bytes < 0) {
993              L<<Logger::Error<<"TCP Remote "<<sockAddrToString(&i->remote,sizeof(i->remote))<<" disconnected after first byte"<<endl;
994              close(i->fd);
995              tcpconnections.erase(i);
996              break;
997            }
998           
999          }
1000          else if(i->state==TCPConnection::GETQUESTION) {
1001            int bytes=read(i->fd,i->data + i->bytesread,i->qlen - i->bytesread);
1002            if(!bytes || bytes < 0) {
1003              L<<Logger::Error<<"TCP Remote "<<sockAddrToString(&i->remote,sizeof(i->remote))<<" disconnected while reading question body"<<endl;
1004              close(i->fd);
1005              tcpconnections.erase(i);
1006              break;
1007            }
1008            i->bytesread+=bytes;
1009            if(i->bytesread==i->qlen) {
1010              i->state=TCPConnection::BYTE0;
1011              DNSComboWriter* dc=0;
1012              try {
1013                dc=new DNSComboWriter(i->data, i->qlen, now);
1014              }
1015              catch(MOADNSException &mde) {
1016                L<<Logger::Error<<"Unparseable packet from remote client "<<sockAddrToString(&i->remote,sizeof(i->remote))<<endl;
1017                close(i->fd);
1018                tcpconnections.erase(i);
1019                break;
1020              }
1021
1022              dc->setSocket(i->fd);
1023              dc->d_tcp=true;
1024              dc->setRemote((struct sockaddr *)&i->remote,sizeof(i->remote));
1025              if(dc->d_mdp.d_header.qr)
1026                L<<Logger::Error<<"Ignoring answer on server socket!"<<endl;
1027              else {
1028                ++qcounter;
1029                MT->makeThread(startDoResolve, dc, "tcp");
1030              }
1031            }
1032          }
1033        }
1034      }
1035    }
1036  }
1037  catch(AhuException &ae) {
1038    L<<Logger::Error<<"Exception: "<<ae.reason<<endl;
1039    ret=EXIT_FAILURE;
1040  }
1041  catch(exception &e) {
1042    L<<Logger::Error<<"STL Exception: "<<e.what()<<endl;
1043    ret=EXIT_FAILURE;
1044  }
1045  catch(...) {
1046    L<<Logger::Error<<"any other exception in main: "<<endl;
1047    ret=EXIT_FAILURE;
1048  }
1049 
1050#ifdef WIN32
1051  WSACleanup();
1052#endif // WIN32
1053
1054  return ret;
1055}
Note: See TracBrowser for help on using the browser.