src/event.c

   1 /**/
   2
   3 #include "adns-internal.h"
   4
   5 static void autosys(adns_state ads, struct timeval now) {
   6   if (ads->iflags & adns_if_noautosys) return;
   7   adns_callback(ads,-1,0,0,0);
   8 }
   9
  10 static int callb_checkfd(int maxfd, const fd_set *fds, int fd) {
  11   return maxfd<0 || !fds ? 1 :
  12          fd<maxfd && FD_ISSET(fd,fds);
  13 }
  14
  15 void adns__tcp_broken(adns_state ads, const char *what, const char *why) {
  16   int serv;
  17
  18   assert(ads->tcpstate == server_connecting || ads->tcpstate == server_ok);
  19   warn("nameserver %s TCP connection lost: %s: %s",
  20        inet_ntoa(ads->servers[tcpserver].addr,what,why));
  21   close(ads->tcpsocket);
  22   ads->tcpstate= server_disconnected;
  23   serv= ads->tcpserver;
  24
  25   for (qu= ads->timew; qu; qu= nqu) {
  26     nqu= qu->next;
  27     if (qu->senttcpserver == -1) continue;
  28     assert(qu->senttcpserver == serv);
  29     DLIST_UNLINK(ads->timew,qu);
  30     adns__query_fail(ads,qu,adns_s_connlost); /* wishlist: send to other servers ? */
  31   }
  32
  33   ads->tcpbuf.used= 0;
  34   ads->tcpserver= (serv+1)%ads->nservers;
  35 }
  36
  37 void adns__tcp_tryconnect(adns_state ads) {
  38   int r, fd, tries;
  39   sockaddr_in addr;
  40
  41   for (tries=0; tries<ads->nservers; tries++) {
  42     if (ads->tcpstate == server_connecting || ads->tcpstate == server_ok) return;
  43     assert(ads->tcpstate == server_disconnected);
  44     assert(!ads->tcpbuf.used);
  45
  46     proto= getprotobyname("tcp");
  47     if (!proto) { diag(ads,"unable to find protocol number for TCP !",-1); return; }
  48     fd= socket(AF_INET,SOCK_STREAM,proto->p_proto);
  49     if (fd<0) { diag(ads,"cannot create TCP socket: %s",-1,strerror(errno)); return; }
  50     if (!adns__setnonblock(fd)) return;
  51     memset(&addr,0,sizeof(addr));
  52     addr.sin_family= AF_INET;
  53     addr.sin_port= htons(NSPORT);
  54     addr.sin_addr= ads->servers[ads->tcpserver].addr;
  55     r= connect(fd,&addr,sizeof(addr));
  56     ads->tcpsocket= fd;
  57     ads->tcpstate= server_connecting;
  58     if (r==0) { ads->tcpstate= server_ok; return; }
  59     if (errno == EWOULDBLOCK || errno == EINPROGRESS) return;
  60     tcpserver_broken(ads,"connect",strerror(errno));
  61   }
  62 }
  63
  64 int adns_callback(adns_state ads, int maxfd,
  65                   const fd_set *readfds, const fd_set *writefds,
  66                   const fd_set *exceptfds) {
  67   int skip, dgramlen, count, udpaddrlen;
  68   enum adns__tcpstate oldtcpstate;
  69   unsigned char udpbuf[UDPMAXDGRAM];
  70   struct sockaddr_in udpaddr;
  71
  72   count= 0;
  73   oldtcpstate= ads->tcpstate;
  74
  75   if (ads->tcpstate == server_connecting) {
  76     if (callb_checkfd(maxfd,writefds,ads->tcpsocket)) {
  77       count++;
  78       assert(ads->tcprecv.used==0);
  79       vbuf_ensure(&ads->tcprecv,1);
  80       if (ads->tcprecv.buf) {
  81         r= read(ads->tcpsocket,&ads->tcprecv.buf,1);
  82         if (r==0 || (r<0 && (errno==EAGAIN || errno==EWOULDBLOCK))) {
  83           debug("nameserver %s TCP connected",
  84                 inet_ntoa(ads->servers[ads->tcpserver].addr));
  85           ads->tcpstate= server_connected;
  86         } else if (r>0) {
  87           tcpserver_broken(ads,"connect/read","sent data before first request");
  88         } else if (errno!=EINTR) {
  89           tcpserver_broken(ads,"connect/read",strerror(errno));
  90         }
  91       }
  92     }
  93   }
  94   if (ads->tcpstate == server_connected) {
  95     if (oldtcpstate == server_connected)
  96       count+= callb_checkfd(maxfd,readfds,ads->tcpsocket) +
  97               callb_checkfd(maxfd,exceptfds,ads->tcpsocket) +
  98         (ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket));
  99     if (oldtcpstate != server_connected || callb_checkfd(maxfd,readfds,ads->tcpsocket)) {
 100       skip= 0;
 101       for (;;) {
 102         if (ads->tcprecv.used<skip+2) {
 103           want= 2;
 104         } else {
 105           dgramlen= (ads->tcprecv.buf[skip]<<8) | ads->tcprecv.buf[skip+1];
 106           if (ads->tcprecv.used<skip+2+dgramlen) {
 107             want= 2+dgramlen;
 108           } else {
 109             procdgram(ads,ads->tcprecv.buf+skip+2,dgramlen,ads->tcpserver);
 110             skip+= 2+dgramlen; continue;
 111           }
 112         }
 113         ads->tcprecv.used -= skip;
 114         memmove(ads->tcprecv.buf,ads->tcprecv.buf+skip,ads->tcprecv.used);
 115         vbuf_ensure(&ads->tcprecv,want);
 116         if (ads->tcprecv.used >= ads->tcprecv.avail) break;
 117         r= read(ads->tcpsocket,
 118                 ads->tcprecv.buf+ads->tcprecv.used,
 119                 ads->tcprecv.avail-ads->tcprecv.used);
 120         if (r>0) {
 121           ads->tcprecv.used+= r;
 122         } else {
 123           if (r<0) {
 124             if (errno==EAGAIN || errno==EWOULDBLOCK || errno==ENOMEM) break;
 125             if (errno==EINTR) continue;
 126           }
 127           tcpserver_broken(ads->tcpserver,"read",r?strerror(errno):"closed");
 128           break;
 129         }
 130       }
 131     } else if (callb_checkfd(maxfd,exceptfds,ads->tcpsocket)) {
 132       tcpserver_broken(ads->tcpserver,"select","exceptional condition detected");
 133     } else if (ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket)) {
 134       r= write(ads->tcpsocket,ads->tcpsend.buf,ads->tcpsend.used);
 135       if (r<0) {
 136         if (errno!=EAGAIN && errno!=EWOULDBLOCK && errno!=ENOMEM && errno!=EINTR) {
 137           tcpserver_broken(ads->tcpserver,"write",strerror(errno));
 138         }
 139       } else if (r>0) {
 140         ads->tcpsend.used -= r;
 141         memmove(ads->tcpsend.buf,ads->tcpsend.buf+r,ads->tcpsend.used);
 142       }
 143     }
 144   }
 145
 146   if (callb_checkfd(maxfd,readfds,ads->udpsocket)) {
 147     count++;
 148     for (;;) {
 149       udpaddrlen= sizeof(udpaddr);
 150       r= recvfrom(ads->udpsocket,udpbuf,sizeof(udpbuf),0,&udpaddr,&udpaddrlen);
 151       if (r<0) {
 152         if (!(errno == EAGAIN || errno == EWOULDBLOCK ||
 153               errno == EINTR || errno == ENOMEM || errno == ENOBUFS))
 154           warn("datagram receive error: %s",strerror(errno));
 155         break;
 156       }
 157       if (udpaddrlen != sizeof(udpaddr)) {
 158         diag("datagram received with wrong address length %d (expected %d)",
 159              udpaddrlen,sizeof(udpaddr));
 160         continue;
 161       }
 162       if (udpaddr.sin_family != AF_INET) {
 163         diag("datagram received with wrong protocol family %u (expected %u)",
 164              udpaddr.sin_family,AF_INET);
 165         continue;
 166       }
 167       if (ntohs(udpaddr.sin_port) != NSPORT) {
 168         diag("datagram received from wrong port %u (expected %u)",
 169              ntohs(udpaddr.sin_port),NSPORT);
 170         continue;
 171       }
 172       for (serv= 0;
 173            serv < ads->nservers &&
 174              ads->servers[serv].addr.s_addr != udpaddr.sin_addr.s_addr;
 175            serv++);
 176       if (serv >= ads->nservers) {
 177         warn("datagram received from unknown nameserver %s",inet_ntoa(udpaddr.sin_addr));
 178         continue;
 179       }
 180       procdgram(ads,udpbuf,r,serv);
 181     }
 182   }
 183 }
 184
 185 static void inter_maxto(struct timeval **tv_io, struct timeval *tvbuf,
 186                         struct timeval maxto) {
 187   struct timeval rbuf;
 188
 189   rbuf= *tv_io;
 190   if (!rbuf) { *tvbuf= maxto; *tv_io= tvbuf; return; }
 191   if (timercmp(rbuf,&maxto,>)) *rbuf= maxto;
 192 }
 193
 194 static void inter_maxtoabs(struct timeval **tv_io, struct timeval *tvbuf,
 195                            struct timeval now, struct timeval maxtime) {
 196   ldiv_t dr;
 197
 198   maxtime.tv_sec -= (now.tv_sec-1);
 199   maxtime.tv_usec += (1000-now.tv_usec);
 200   dr= ldiv(maxtime.tv_usec,1000);
 201   maxtime.tv_sec += dr.quot;
 202   maxtime.tv_usec -= dr.rem;
 203   inter_maxto(tv_io,tvbuf,maxtime);
 204 }
 205
 206 static void localresourcerr(struct timeval **tv_io, struct timeval *tvbuf,
 207                             const char *syscall) {
 208   struct timeval tvto_lr;
 209
 210   warn(ads,"local system resources scarce (during %s): %s",syscall,strerror(errno));
 211   timerclear(&tvto_lr); timevaladd(&tvto_lr,LOCALRESOURCEMS);
 212   inter_maxto(tv_io, tvbuf, tvto_lr);
 213   return;
 214 }
 215
 216 static void inter_addfd(int *maxfd, fd_set *fds, int fd) {
 217   if (fd>=*maxfd) *maxfd= fd+1;
 218   FD_SET(fd,fds);
 219 }
 220
 221 void adns_interest(adns_state ads, int *maxfd,
 222                    fd_set *readfds, fd_set *writefds, fd_set *exceptfds,
 223                    struct timeval **tv_io, struct timeval *tvbuf) {
 224   struct timeval now;
 225   adns_query qu;
 226   int r;
 227
 228   r= gettimeofday(&now,0);
 229   if (r) { localresourcerr(tv_io,tvbuf,"gettimeofday"); return; }
 230
 231   for (qu= ads->timew; qu; qu= nqu) {
 232     nqu= qu->next;
 233     if (timercmp(&now,qu->timeout,>)) {
 234       DLIST_UNLINK(ads->timew,qu);
 235       if (qu->nextudpserver == -1) {
 236         query_fail(ads,qu,adns_s_notresponding);
 237       } else {
 238         DLIST_LINKTAIL(ads->tosend,qu);
 239       }
 240     } else {
 241       inter_maxtoabs(tv_io,tvbuf,now,qu->timeout);
 242     }
 243   }
 244
 245   for (qu= ads->tosend; qu; qu= nqu) {
 246     nqu= qu->next;
 247     quproc_tosend(ads,qu,now);
 248   }
 249
 250   inter_addfd(maxfd,readfds,ads->udpsocket);
 251
 252   switch (ads->tcpstate) {
 253   case server_disc:
 254     break;
 255   case server_connecting:
 256     inter_addfd(maxfd,writefds,ads->tcpsocket);
 257     break;
 258   case server_connected:
 259     inter_addfd(maxfd,readfds,ads->tcpsocket);
 260     inter_addfd(maxfd,exceptfds,ads->tcpsocket);
 261     if (ads->opbufused) inter_addfd(maxfd,writefds,ads->tcpsocket);
 262   default:
 263     abort();
 264   }
 265 }
 266
 267 static int internal_check(adns_state ads,
 268                           adns_query *query_io,
 269                           adns_answer **answer,
 270                           void **context_r) {
 271   adns_query qu;
 272
 273   qu= *query_io;
 274   if (!qu) {
 275     if (!ads->output.head) return EWOULDBLOCK;
 276     qu= ads->output.head;
 277   } else {
 278     if (qu->id>=0) return EWOULDBLOCK;
 279   }
 280   LIST_UNLINK(ads->output,qu);
 281   *answer= qu->answer;
 282   if (context_r) *context_r= qu->context;
 283   free(qu);
 284   return 0;
 285 }
 286
 287 int adns_wait(adns_state ads,
 288               adns_query *query_io,
 289               adns_answer **answer_r,
 290               void **context_r) {
 291   int r, maxfd, rsel, rcb;
 292   fd_set readfds, writefds, exceptfds;
 293   struct timeval tvbuf, *tvp;
 294
 295   for (;;) {
 296     r= internal_check(ads,query_io,answer_r,context_r);
 297     if (r && r != EWOULDBLOCK) return r;
 298     maxfd= 0; tvp= 0;
 299     FD_ZERO(&readfds); FD_ZERO(&writefds); FD_ZERO(&exceptfds);
 300     adns_interest(ads,&maxfd,&readfds,&writefds,&exceptfds,&tvp,&tvbuf);
 301     rsel= select(maxfd,&readfds,&writefds,&exceptfds,tvp);
 302     if (rsel==-1) return r;
 303     rcb= adns_callback(ads,maxfd,&readfds,&writefds,&exceptfds);
 304     assert(rcb==rsel);
 305   }
 306 }
 307
 308 int adns_check(adns_state ads,
 309                adns_query *query_io,
 310                adns_answer **answer_r,
 311                void **context_r) {
 312   autosys(ads);
 313   return internal_check(ads,query_io,answer_r,context_r);
 314 }