pdns/dnstcpbench.cc

   1 /*
   2     PowerDNS Versatile Database Driven Nameserver
   3     Copyright (C) 2002-2013  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     Additionally, the license of this program contains a special
  10     exception which allows to distribute the program in binary form when
  11     it is linked against OpenSSL.
  12
  13     This program is distributed in the hope that it will be useful,
  14     but WITHOUT ANY WARRANTY; without even the implied warranty of
  15     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16     GNU General Public License for more details.
  17
  18     You should have received a copy of the GNU General Public License
  19     along with this program; if not, write to the Free Software
  20     Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  21 */
  22 #ifdef HAVE_CONFIG_H
  23 #include "config.h"
  24 #endif
  25 #include <boost/accumulators/statistics/median.hpp>
  26 #include <boost/accumulators/statistics/mean.hpp>
  27 #include <boost/accumulators/accumulators.hpp>
  28
  29 #include <boost/accumulators/statistics.hpp>
  30
  31 #include "dnsparser.hh"
  32 #include "sstuff.hh"
  33 #include "misc.hh"
  34 #include "dnswriter.hh"
  35 #include "dnsrecords.hh"
  36 #include "statbag.hh"
  37 #include <netinet/tcp.h>
  38 #include <boost/array.hpp>
  39 #include <boost/program_options.hpp>
  40
  41
  42 StatBag S;
  43 namespace po = boost::program_options;
  44
  45 po::variables_map g_vm;
  46 bool g_verbose;
  47 bool g_onlyTCP;
  48 bool g_tcpNoDelay;
  49 unsigned int g_timeoutMsec;
  50 AtomicCounter g_networkErrors, g_otherErrors, g_OK, g_truncates, g_authAnswers, g_timeOuts;
  51 ComboAddress g_dest;
  52
  53 unsigned int makeUsec(const struct timeval& tv)
  54 {
  55   return 1000000*tv.tv_sec + tv.tv_usec;
  56 }
  57
  58 /* On Linux, run echo 1 > /proc/sys/net/ipv4/tcp_tw_recycle
  59    to prevent running out of free TCP ports */
  60
  61 struct BenchQuery
  62 {
  63   BenchQuery(const std::string& qname_, uint16_t qtype_) : qname(qname_), qtype(qtype_), udpUsec(0), tcpUsec(0), answerSecond(0) {}
  64   BenchQuery(){}
  65   DNSName qname;
  66   uint16_t qtype;
  67   uint32_t udpUsec, tcpUsec;
  68   time_t answerSecond;
  69 };
  70
  71 void doQuery(BenchQuery* q)
  72 try
  73 {
  74   vector<uint8_t> packet;
  75   DNSPacketWriter pw(packet, q->qname, q->qtype);
  76   int res;
  77   string reply;
  78
  79   struct timeval tv, now;
  80   gettimeofday(&tv, 0);
  81
  82   if(!g_onlyTCP) {
  83     Socket udpsock(g_dest.sin4.sin_family, SOCK_DGRAM);
  84
  85     udpsock.sendTo(string((char*)&*packet.begin(), (char*)&*packet.end()), g_dest);
  86     ComboAddress origin;
  87     res = waitForData(udpsock.getHandle(), 0, 1000 * g_timeoutMsec);
  88     if(res < 0)
  89       throw NetworkError("Error waiting for response");
  90     if(!res) {
  91       g_timeOuts++;
  92       return;
  93     }
  94
  95     udpsock.recvFrom(reply, origin);
  96
  97     gettimeofday(&now, 0);
  98     q->udpUsec = makeUsec(now - tv);
  99     tv=now;
 100
 101     MOADNSParser mdp(reply);
 102     if(!mdp.d_header.tc)
 103       return;
 104     g_truncates++;
 105   }
 106
 107   Socket sock(g_dest.sin4.sin_family, SOCK_STREAM);
 108   int tmp=1;
 109   if(setsockopt(sock.getHandle(),SOL_SOCKET,SO_REUSEADDR,(char*)&tmp,sizeof tmp)<0)
 110     throw runtime_error("Unable to set socket reuse: "+string(strerror(errno)));
 111
 112   if(g_tcpNoDelay && setsockopt(sock.getHandle(), IPPROTO_TCP, TCP_NODELAY,(char*)&tmp,sizeof tmp)<0)
 113     throw runtime_error("Unable to set socket no delay: "+string(strerror(errno)));
 114
 115   sock.connect(g_dest);
 116   uint16_t len = htons(packet.size());
 117   string tcppacket((char*)& len, 2);
 118   tcppacket.append((char*)&*packet.begin(), (char*)&*packet.end());
 119
 120   sock.writen(tcppacket);
 121
 122   res = waitForData(sock.getHandle(), 0, 1000 * g_timeoutMsec);
 123   if(res < 0)
 124     throw NetworkError("Error waiting for response");
 125   if(!res) {
 126     g_timeOuts++;
 127     return;
 128   }
 129
 130   if(sock.read((char *) &len, 2) != 2)
 131     throw PDNSException("tcp read failed");
 132
 133   len=ntohs(len);
 134   char *creply = new char[len];
 135   int n=0;
 136   int numread;
 137   while(n<len) {
 138     numread=sock.read(creply+n, len-n);
 139     if(numread<0)
 140       throw PDNSException("tcp read failed");
 141     n+=numread;
 142   }
 143
 144   reply=string(creply, len);
 145   delete[] creply;
 146
 147   gettimeofday(&now, 0);
 148   q->tcpUsec = makeUsec(now - tv);
 149   q->answerSecond = now.tv_sec;
 150
 151   MOADNSParser mdp(reply);
 152   //  cout<<"Had correct TCP/IP response, "<<mdp.d_answers.size()<<" answers, aabit="<<mdp.d_header.aa<<endl;
 153   if(mdp.d_header.aa)
 154     g_authAnswers++;
 155   g_OK++;
 156 }
 157 catch(NetworkError& ne)
 158 {
 159   cerr<<"Network error: "<<ne.what()<<endl;
 160   g_networkErrors++;
 161 }
 162 catch(...)
 163 {
 164   g_otherErrors++;
 165 }
 166
 167 /* read queries from stdin, put in vector
 168    launch n worker threads, each picks a query using AtomicCounter
 169    If a worker reaches the end of its queue, it stops */
 170
 171 AtomicCounter g_pos;
 172
 173 vector<BenchQuery> g_queries;
 174
 175 static void* worker(void*)
 176 {
 177   for(;;) {
 178     unsigned int pos = g_pos++;
 179     if(pos >= g_queries.size())
 180       break;
 181
 182     doQuery(&g_queries[pos]); // this is safe as long as nobody *inserts* to g_queries
 183   }
 184   return 0;
 185 }
 186
 187 int main(int argc, char** argv)
 188 try
 189 {
 190   po::options_description desc("Allowed options"), hidden, alloptions;
 191   desc.add_options()
 192     ("help,h", "produce help message")
 193     ("verbose,v", "be verbose")
 194     ("udp-first,u", "try UDP first")
 195     ("file,f", po::value<string>(), "source file - if not specified, defaults to stdin")
 196     ("tcp-no-delay", po::value<bool>()->default_value(true), "use TCP_NODELAY socket option")
 197     ("timeout-msec", po::value<int>()->default_value(10), "wait for this amount of milliseconds for an answer")
 198     ("workers", po::value<int>()->default_value(100), "number of parallel workers");
 199
 200   hidden.add_options()
 201     ("remote-host", po::value<string>(), "remote-host")
 202     ("remote-port", po::value<int>()->default_value(53), "remote-port");
 203   alloptions.add(desc).add(hidden);
 204
 205   po::positional_options_description p;
 206   p.add("remote-host", 1);
 207   p.add("remote-port", 1);
 208
 209   po::store(po::command_line_parser(argc, argv).options(alloptions).positional(p).run(), g_vm);
 210   po::notify(g_vm);
 211
 212   if(g_vm.count("help")) {
 213     cout << desc<<endl;
 214     exit(EXIT_SUCCESS);
 215   }
 216   g_tcpNoDelay = g_vm["tcp-no-delay"].as<bool>();
 217
 218   g_onlyTCP = !g_vm.count("udp-first");
 219   g_verbose = g_vm.count("verbose");
 220   g_timeoutMsec = g_vm["timeout-msec"].as<int>();
 221
 222   reportAllTypes();
 223
 224   if(g_vm["remote-host"].empty()) {
 225     cerr<<"Syntax: tcpbench remote [port] < queries"<<endl;
 226     cerr<<"Where queries is one query per line, format: qname qtype, just 1 space"<<endl;
 227     cerr<<desc<<endl;
 228     exit(EXIT_FAILURE);
 229   }
 230
 231   g_dest = ComboAddress(g_vm["remote-host"].as<string>().c_str(), g_vm["remote-port"].as<int>());
 232
 233   unsigned int numworkers=g_vm["workers"].as<int>();
 234
 235   if(g_verbose) {
 236     cout<<"Sending queries to: "<<g_dest.toStringWithPort()<<endl;
 237     cout<<"Attempting UDP first: " << (g_onlyTCP ? "no" : "yes") <<endl;
 238     cout<<"Timeout: "<< g_timeoutMsec<<"msec"<<endl;
 239     cout << "Using TCP_NODELAY: "<<g_tcpNoDelay<<endl;
 240   }
 241
 242
 243   pthread_t workers[numworkers];
 244
 245   FILE* fp;
 246   if(!g_vm.count("file"))
 247     fp=fdopen(0, "r");
 248   else {
 249     fp=fopen(g_vm["file"].as<string>().c_str(), "r");
 250     if(!fp)
 251       unixDie("Unable to open "+g_vm["file"].as<string>()+" for input");
 252   }
 253   pair<string, string> q;
 254   string line;
 255   while(stringfgets(fp, line)) {
 256     trim_right(line);
 257     q=splitField(line, ' ');
 258     g_queries.push_back(BenchQuery(q.first, DNSRecordContent::TypeToNumber(q.second)));
 259   }
 260   fclose(fp);
 261
 262   for(unsigned int n = 0; n < numworkers; ++n) {
 263     pthread_create(&workers[n], 0, worker, 0);
 264   }
 265   for(unsigned int n = 0; n < numworkers; ++n) {
 266     void* status;
 267     pthread_join(workers[n], &status);
 268   }
 269
 270   using namespace boost::accumulators;
 271   typedef accumulator_set<
 272     double
 273     , stats<boost::accumulators::tag::median(with_p_square_quantile),
 274       boost::accumulators::tag::mean(immediate)
 275     >
 276   > acc_t;
 277
 278   acc_t udpspeeds, tcpspeeds, qps;
 279
 280   typedef map<time_t, uint32_t> counts_t;
 281   counts_t counts;
 282
 283   for(const BenchQuery& bq :  g_queries) {
 284     counts[bq.answerSecond]++;
 285     udpspeeds(bq.udpUsec);
 286     tcpspeeds(bq.tcpUsec);
 287   }
 288
 289   for(const counts_t::value_type& val :  counts) {
 290     qps(val.second);
 291   }
 292
 293   cout<<"Average qps: "<<mean(qps)<<", median qps: "<<median(qps)<<endl;
 294   cout<<"Average UDP latency: "<<mean(udpspeeds)<<"usec, median: "<<median(udpspeeds)<<"usec"<<endl;
 295   cout<<"Average TCP latency: "<<mean(tcpspeeds)<<"usec, median: "<<median(tcpspeeds)<<"usec"<<endl;
 296
 297   cout<<"OK: "<<g_OK<<", network errors: "<<g_networkErrors<<", other errors: "<<g_otherErrors<<endl;
 298   cout<<"Timeouts: "<<g_timeOuts<<endl;
 299   cout<<"Truncateds: "<<g_truncates<<", auth answers: "<<g_authAnswers<<endl;
 300 }
 301 catch(std::exception &e)
 302 {
 303   cerr<<"Fatal: "<<e.what()<<endl;
 304 }