pdns/distributor.hh

   1 /*
   2  * This file is part of PowerDNS or dnsdist.
   3  * Copyright -- PowerDNS.COM B.V. and its contributors
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of version 2 of the GNU General Public License as
   7  * published by the Free Software Foundation.
   8  *
   9  * In addition, for the avoidance of any doubt, permission is granted to
  10  * link this program with OpenSSL and to (re)distribute the binaries
  11  * produced as the result of such linking.
  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 Street, Fifth Floor, Boston, MA 02110-1301 USA.
  21  */
  22 #pragma once
  23 #include <string>
  24 #include <deque>
  25 #include <queue>
  26 #include <vector>
  27 #include <pthread.h>
  28 #include "threadname.hh"
  29 #include <unistd.h>
  30 #include "logger.hh"
  31 #include "dns.hh"
  32 #include "dnsbackend.hh"
  33 #include "pdnsexception.hh"
  34 #include "arguments.hh"
  35 #include <atomic>
  36 #include "statbag.hh"
  37
  38 extern StatBag S;
  39
  40 /** the Distributor template class enables you to multithread slow question/answer
  41     processes.
  42
  43     Questions are posed to the Distributor, which returns the answer via a callback.
  44
  45     The Distributor spawns sufficient backends, and if they thrown an exception,
  46     it will cycle the backend but drop the query that was active during the exception.
  47 */
  48
  49 template<class Answer, class Question, class Backend> class Distributor
  50 {
  51 public:
  52   static Distributor* Create(int n=1); //!< Create a new Distributor with \param n threads
  53   typedef std::function<void(std::unique_ptr<Answer>&)> callback_t;
  54   virtual int question(Question&, callback_t callback) =0; //!< Submit a question to the Distributor
  55   virtual int getQueueSize() =0; //!< Returns length of question queue
  56   virtual bool isOverloaded() =0;
  57   virtual ~Distributor() { cerr<<__func__<<endl;}
  58 };
  59
  60 template<class Answer, class Question, class Backend> class SingleThreadDistributor
  61     : public Distributor<Answer, Question, Backend>
  62 {
  63 public:
  64   SingleThreadDistributor(const SingleThreadDistributor&) = delete;
  65   void operator=(const SingleThreadDistributor&) = delete;
  66   SingleThreadDistributor();
  67   typedef std::function<void(std::unique_ptr<Answer>&)> callback_t;
  68   int question(Question&, callback_t callback) override; //!< Submit a question to the Distributor
  69   int getQueueSize() override {
  70     return 0;
  71   }
  72
  73   bool isOverloaded() override
  74   {
  75     return false;
  76   }
  77
  78 private:
  79   std::unique_ptr<Backend> b{nullptr};
  80 };
  81
  82 template<class Answer, class Question, class Backend> class MultiThreadDistributor
  83     : public Distributor<Answer, Question, Backend>
  84 {
  85 public:
  86   MultiThreadDistributor(const MultiThreadDistributor&) = delete;
  87   void operator=(const MultiThreadDistributor&) = delete;
  88   MultiThreadDistributor(int n);
  89   typedef std::function<void(std::unique_ptr<Answer>&)> callback_t;
  90   int question(Question&, callback_t callback) override; //!< Submit a question to the Distributor
  91   static void* makeThread(void *); //!< helper function to create our n threads
  92   int getQueueSize() override {
  93     return d_queued;
  94   }
  95
  96   struct QuestionData
  97   {
  98     QuestionData(const Question& query): Q(query)
  99     {
 100     }
 101
 102     Question Q;
 103     callback_t callback;
 104     int id;
 105   };
 106
 107   bool isOverloaded() override
 108   {
 109     return d_overloadQueueLength && (d_queued > d_overloadQueueLength);
 110   }
 111
 112 private:
 113   int nextid;
 114   time_t d_last_started;
 115   unsigned int d_overloadQueueLength, d_maxQueueLength;
 116   int d_num_threads;
 117   std::atomic<unsigned int> d_queued{0}, d_running{0};
 118   std::vector<std::pair<int,int>> d_pipes;
 119 };
 120
 121 //template<class Answer, class Question, class Backend>::nextid;
 122 template<class Answer, class Question, class Backend> Distributor<Answer,Question,Backend>* Distributor<Answer,Question,Backend>::Create(int n)
 123 {
 124     if( n == 1 )
 125       return new SingleThreadDistributor<Answer,Question,Backend>();
 126     else
 127       return new MultiThreadDistributor<Answer,Question,Backend>( n );
 128 }
 129
 130 template<class Answer, class Question, class Backend>SingleThreadDistributor<Answer,Question,Backend>::SingleThreadDistributor()
 131 {
 132   g_log<<Logger::Error<<"Only asked for 1 backend thread - operating unthreaded"<<endl;
 133   try {
 134     b=make_unique<Backend>();
 135   }
 136   catch(const PDNSException &AE) {
 137     g_log<<Logger::Error<<"Distributor caught fatal exception: "<<AE.reason<<endl;
 138     _exit(1);
 139   }
 140   catch(...) {
 141     g_log<<Logger::Error<<"Caught an unknown exception when creating backend, probably"<<endl;
 142     _exit(1);
 143   }
 144 }
 145
 146 template<class Answer, class Question, class Backend>MultiThreadDistributor<Answer,Question,Backend>::MultiThreadDistributor(int n)
 147 {
 148   d_num_threads=n;
 149   d_overloadQueueLength=::arg().asNum("overload-queue-length");
 150   d_maxQueueLength=::arg().asNum("max-queue-length");
 151   nextid=0;
 152   d_last_started=time(0);
 153
 154   pthread_t tid;
 155
 156
 157   for(int i=0; i < n; ++i) {
 158     int fds[2];
 159     if(pipe(fds) < 0)
 160       unixDie("Creating pipe");
 161     d_pipes.push_back({fds[0],fds[1]});
 162   }
 163
 164   if (n<1) {
 165     g_log<<Logger::Error<<"Asked for fewer than 1 threads, nothing to do"<<endl;
 166     _exit(1);
 167   }
 168
 169   g_log<<Logger::Warning<<"About to create "<<n<<" backend threads for UDP"<<endl;
 170   for(int i=0;i<n;i++) {
 171     pthread_create(&tid,0,&makeThread,static_cast<void *>(this));
 172     Utility::usleep(50000); // we've overloaded mysql in the past :-)
 173   }
 174   g_log<<Logger::Warning<<"Done launching threads, ready to distribute questions"<<endl;
 175 }
 176
 177
 178 // start of a new thread
 179 template<class Answer, class Question, class Backend>void *MultiThreadDistributor<Answer,Question,Backend>::makeThread(void *p)
 180 {
 181   setThreadName("pdns/distributo");
 182   pthread_detach(pthread_self());
 183   MultiThreadDistributor *us=static_cast<MultiThreadDistributor *>(p);
 184   int ournum=us->d_running++;
 185
 186   try {
 187     std::unique_ptr<Backend> b= make_unique<Backend>(); // this will answer our questions
 188     int queuetimeout=::arg().asNum("queue-limit");
 189
 190     for(;;) {
 191
 192       QuestionData* tempQD = nullptr;
 193       if(read(us->d_pipes[ournum].first, &tempQD, sizeof(tempQD)) != sizeof(tempQD))
 194         unixDie("read");
 195       --us->d_queued;
 196       std::unique_ptr<QuestionData> QD = std::unique_ptr<QuestionData>(tempQD);
 197       tempQD = nullptr;
 198       std::unique_ptr<Answer> a = nullptr;
 199
 200       if(queuetimeout && QD->Q.d_dt.udiff()>queuetimeout*1000) {
 201         S.inc("timedout-packets");
 202         continue;
 203       }
 204
 205       bool allowRetry=true;
 206 retry:
 207       // this is the only point where we interact with the backend (synchronous)
 208       try {
 209         if (!b) {
 210           allowRetry=false;
 211           b=make_unique<Backend>();
 212         }
 213         a=b->question(QD->Q);
 214       }
 215       catch(const PDNSException &e) {
 216         b.reset();
 217         if (!allowRetry) {
 218           g_log<<Logger::Error<<"Backend error: "<<e.reason<<endl;
 219           a=QD->Q.replyPacket();
 220
 221           a->setRcode(RCode::ServFail);
 222           S.inc("servfail-packets");
 223           S.ringAccount("servfail-queries", QD->Q.qdomain, QD->Q.qtype);
 224         } else {
 225           g_log<<Logger::Notice<<"Backend error (retry once): "<<e.reason<<endl;
 226           goto retry;
 227         }
 228       }
 229       catch(...) {
 230         b.reset();
 231         if (!allowRetry) {
 232           g_log<<Logger::Error<<"Caught unknown exception in Distributor thread "<<(long)pthread_self()<<endl;
 233           a=QD->Q.replyPacket();
 234
 235           a->setRcode(RCode::ServFail);
 236           S.inc("servfail-packets");
 237           S.ringAccount("servfail-queries", QD->Q.qdomain, QD->Q.qtype);
 238         } else {
 239           g_log<<Logger::Warning<<"Caught unknown exception in Distributor thread "<<(long)pthread_self()<<" (retry once)"<<endl;
 240           goto retry;
 241         }
 242       }
 243
 244       QD->callback(a);
 245       QD.reset();
 246     }
 247
 248     b.reset();
 249   }
 250   catch(const PDNSException &AE) {
 251     g_log<<Logger::Error<<"Distributor caught fatal exception: "<<AE.reason<<endl;
 252     _exit(1);
 253   }
 254   catch(...) {
 255     g_log<<Logger::Error<<"Caught an unknown exception when creating backend, probably"<<endl;
 256     _exit(1);
 257   }
 258   return 0;
 259 }
 260
 261 template<class Answer, class Question, class Backend>int SingleThreadDistributor<Answer,Question,Backend>::question(Question& q, callback_t callback)
 262 {
 263   std::unique_ptr<Answer> a = nullptr;
 264   bool allowRetry=true;
 265 retry:
 266   try {
 267     if (!b) {
 268       allowRetry=false;
 269       b=make_unique<Backend>();
 270     }
 271     a=b->question(q); // a can be NULL!
 272   }
 273   catch(const PDNSException &e) {
 274     b.reset();
 275     if (!allowRetry) {
 276       g_log<<Logger::Error<<"Backend error: "<<e.reason<<endl;
 277       a=q.replyPacket();
 278
 279       a->setRcode(RCode::ServFail);
 280       S.inc("servfail-packets");
 281       S.ringAccount("servfail-queries", q.qdomain, q.qtype);
 282     } else {
 283       g_log<<Logger::Notice<<"Backend error (retry once): "<<e.reason<<endl;
 284       goto retry;
 285     }
 286   }
 287   catch(...) {
 288     b.reset();
 289     if (!allowRetry) {
 290       g_log<<Logger::Error<<"Caught unknown exception in Distributor thread "<<(unsigned long)pthread_self()<<endl;
 291       a=q.replyPacket();
 292
 293       a->setRcode(RCode::ServFail);
 294       S.inc("servfail-packets");
 295       S.ringAccount("servfail-queries", q.qdomain, q.qtype);
 296     } else {
 297       g_log<<Logger::Warning<<"Caught unknown exception in Distributor thread "<<(unsigned long)pthread_self()<<" (retry once)"<<endl;
 298       goto retry;
 299     }
 300   }
 301   callback(a);
 302   return 0;
 303 }
 304
 305 struct DistributorFatal{};
 306
 307 template<class Answer, class Question, class Backend>int MultiThreadDistributor<Answer,Question,Backend>::question(Question& q, callback_t callback)
 308 {
 309   // this is passed to other process over pipe and released there
 310   auto QD=new QuestionData(q);
 311   auto ret = QD->id = nextid++; // might be deleted after write!
 312   QD->callback=callback;
 313
 314   ++d_queued;
 315   if(write(d_pipes[QD->id % d_pipes.size()].second, &QD, sizeof(QD)) != sizeof(QD)) {
 316     --d_queued;
 317     delete QD;
 318     unixDie("write");
 319   }
 320
 321   if(d_queued > d_maxQueueLength) {
 322     g_log<<Logger::Error<< d_queued <<" questions waiting for database/backend attention. Limit is "<<::arg().asNum("max-queue-length")<<", respawning"<<endl;
 323     // this will leak the entire contents of all pipes, nothing will be freed. Respawn when this happens!
 324     throw DistributorFatal();
 325   }
 326
 327   return ret;
 328 }