pdns/delaypipe.cc

   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 #include "delaypipe.hh"
  23 #include "misc.hh"
  24 #include "gettime.hh"
  25 #include <thread>
  26 #include "threadname.hh"
  27
  28 template<class T>
  29 ObjectPipe<T>::ObjectPipe()
  30 {
  31   if(pipe(d_fds))
  32     unixDie("pipe");
  33 }
  34
  35 template<class T>
  36 ObjectPipe<T>::~ObjectPipe()
  37 {
  38   ::close(d_fds[0]);
  39   if(d_fds[1] >= 0)
  40     ::close(d_fds[1]);
  41 }
  42
  43 template<class T>
  44 void ObjectPipe<T>::close()
  45 {
  46   if(d_fds[1] < 0)
  47     return;
  48   ::close(d_fds[1]); // the writing side
  49   d_fds[1]=-1;
  50 }
  51
  52 template<class T>
  53 void ObjectPipe<T>::write(T& t)
  54 {
  55   auto ptr = new T(t);
  56   if(::write(d_fds[1], &ptr, sizeof(ptr)) != sizeof(ptr)) {
  57     delete ptr;
  58     unixDie("write");
  59   }
  60 }
  61
  62 template<class T>
  63 bool ObjectPipe<T>::read(T* t)
  64 {
  65   T* ptr;
  66   int ret = ::read(d_fds[0], &ptr, sizeof(ptr));
  67
  68   if(ret < 0)
  69     unixDie("read");
  70   if(ret==0)
  71     return false;
  72   if(ret != sizeof(ptr))
  73     throw std::runtime_error("Partial read, should not happen");
  74   *t=*ptr;
  75   delete ptr;
  76   return true;
  77 }
  78
  79 template<class T>
  80 int ObjectPipe<T>::readTimeout(T* t, double msec)
  81 {
  82   T* ptr;
  83
  84   int ret = waitForData(d_fds[0], 0, 1000*msec);
  85   if(ret < 0)
  86     unixDie("waiting for data in object pipe");
  87   if(ret == 0)
  88     return -1;
  89
  90   ret = ::read(d_fds[0], &ptr, sizeof(ptr)); // this is BLOCKING!
  91
  92   if(ret < 0)
  93     unixDie("read");
  94   if(ret==0)
  95     return false;
  96   if(ret != sizeof(ptr))
  97     throw std::runtime_error("Partial read, should not happen 2");
  98   *t=*ptr;
  99   delete ptr;
 100   return 1;
 101 }
 102
 103
 104 template<class T>
 105 DelayPipe<T>::DelayPipe() : d_thread(&DelayPipe<T>::worker, this)
 106 {
 107 }
 108
 109 template<class T>
 110 void DelayPipe<T>::gettime(struct timespec* ts)
 111 {
 112   ::gettime(ts);
 113 }
 114
 115
 116 template<class T>
 117 void DelayPipe<T>::submit(T& t, int msec)
 118 {
 119   struct timespec now;
 120   gettime(&now);
 121   now.tv_nsec += msec*1e6;
 122   while(now.tv_nsec > 1e9) {
 123     now.tv_sec++;
 124     now.tv_nsec-=1e9;
 125   }
 126   Combo c{t, now};
 127   d_pipe.write(c);
 128 }
 129
 130 template<class T>
 131 DelayPipe<T>::~DelayPipe()
 132 {
 133   d_pipe.close();
 134   d_thread.join();
 135 }
 136
 137
 138
 139 template<class T>
 140 void DelayPipe<T>::worker()
 141 {
 142   setThreadName("dnsdist/delayPi");
 143   Combo c;
 144   for(;;) {
 145     /* this code is slightly too subtle, but I don't see how it could be any simpler.
 146        So we have a set of work to do, and we need to wait until the time arrives to do it.
 147        Simultaneously new work might come in. So we try to combine both of these things by
 148        setting a timeout on listening to the pipe over which new work comes in. This timeout
 149        is equal to the wait until the first thing that needs to be done.
 150
 151        Two additional cases exist: we have no work to wait for, so we can wait infinitely long.
 152        The other special case is that the first we have to do.. is in the past, so we need to do it
 153        immediately. */
 154
 155
 156     double delay=-1;  // infinite
 157     struct timespec now;
 158     if(!d_work.empty()) {
 159       gettime(&now);
 160       delay=1000*tsdelta(d_work.begin()->first, now);
 161       if(delay < 0) {
 162         delay=0;   // don't wait - we have work that is late already!
 163       }
 164     }
 165     if(delay != 0 ) {
 166       int ret = d_pipe.readTimeout(&c, delay);
 167       if(ret > 0) {  // we got an object
 168         d_work.insert(make_pair(c.when, c.what));
 169       }
 170       else if(ret==0) { // EOF
 171         break;
 172       }
 173       else {
 174         ;
 175       }
 176       gettime(&now);
 177     }
 178
 179     tscomp cmp;
 180
 181     for(auto iter = d_work.begin() ; iter != d_work.end(); ) { // do the needful
 182       if(cmp(iter->first, now)) {
 183         iter->second();
 184         d_work.erase(iter++);
 185       }
 186       else {
 187         break;
 188       }
 189     }
 190   }
 191 }