mdw@git.distorted.org.uk Git - adns/blob - src/event.c

   1 /**/
   2
   3 #include <string.h>
   4 #include <errno.h>
   5 #include <stdlib.h>
   6
   7 #include <netdb.h>
   8 #include <arpa/inet.h>
   9
  10 #include "internal.h"
  11
  12 /* TCP connection management */
  13
  14 void adns__tcp_broken(adns_state ads, const char *what, const char *why) {
  15   int serv;
  16   adns_query qu, nqu;
  17
  18   assert(ads->tcpstate == server_connecting || ads->tcpstate == server_ok);
  19   serv= ads->tcpserver;
  20   adns__warn(ads,serv,"TCP connection lost: %s: %s",what,why);
  21   close(ads->tcpsocket);
  22   ads->tcpstate= server_disconnected;
  23
  24   for (qu= ads->timew.head; qu; qu= nqu) {
  25     nqu= qu->next;
  26     if (qu->state == query_udp) continue;
  27     assert(qu->state == query_tcpwait || qu->state == query_tcpsent);
  28     qu->state= query_tcpwait;
  29     qu->tcpfailed |= (1<<serv);
  30     if (qu->tcpfailed == (1<<ads->nservers)-1) {
  31       LIST_UNLINK(ads->timew,qu);
  32       adns__query_fail(ads,qu,adns_s_allservfail);
  33     }
  34   }
  35
  36   ads->tcprecv.used= ads->tcpsend.used= 0;
  37   ads->tcpserver= (serv+1)%ads->nservers;
  38 }
  39
  40 static void tcp_connected(adns_state ads, struct timeval now) {
  41   adns_query qu, nqu;
  42
  43   adns__debug(ads,ads->tcpserver,"TCP connected");
  44   ads->tcpstate= server_ok;
  45   for (qu= ads->timew.head; qu; qu= nqu) {
  46     nqu= qu->next;
  47     if (qu->state == query_udp) continue;
  48     assert (qu->state == query_tcpwait);
  49     adns__query_tcp(ads,qu,now);
  50   }
  51 }
  52
  53 void adns__tcp_tryconnect(adns_state ads, struct timeval now) {
  54   int r, fd, tries;
  55   struct sockaddr_in addr;
  56   struct protoent *proto;
  57   /* fixme: single TCP timeout, not once per server */
  58
  59   for (tries=0; tries<ads->nservers; tries++) {
  60     if (ads->tcpstate == server_connecting || ads->tcpstate == server_ok) return;
  61     assert(ads->tcpstate == server_disconnected);
  62     assert(!ads->tcpsend.used);
  63     assert(!ads->tcprecv.used);
  64
  65     proto= getprotobyname("tcp");
  66     if (!proto) { adns__diag(ads,-1,"unable to find protocol no. for TCP !"); return; }
  67     fd= socket(AF_INET,SOCK_STREAM,proto->p_proto);
  68     if (fd<0) {
  69       adns__diag(ads,-1,"cannot create TCP socket: %s",strerror(errno));
  70       return;
  71     }
  72     r= adns__setnonblock(ads,fd);
  73     if (r) {
  74       adns__diag(ads,-1,"cannot make TCP socket nonblocking: %s",strerror(r));
  75       close(fd);
  76       return;
  77     }
  78     memset(&addr,0,sizeof(addr));
  79     addr.sin_family= AF_INET;
  80     addr.sin_port= htons(NSPORT);
  81     addr.sin_addr= ads->servers[ads->tcpserver].addr;
  82     r= connect(fd,&addr,sizeof(addr));
  83     ads->tcpsocket= fd;
  84     ads->tcpstate= server_connecting;
  85     if (r==0) { tcp_connected(ads,now); continue; }
  86     if (errno == EWOULDBLOCK || errno == EINPROGRESS) return;
  87     adns__tcp_broken(ads,"connect",strerror(errno));
  88   }
  89 }
  90
  91 /* `Interest' functions - find out which fd's we might be interested in,
  92  * and when we want to be called back for a timeout.
  93  */
  94
  95 static void inter_maxto(struct timeval **tv_io, struct timeval *tvbuf,
  96                         struct timeval maxto) {
  97   struct timeval *rbuf;
  98
  99   if (!tv_io) return;
 100   rbuf= *tv_io;
 101   if (!rbuf) {
 102     *tvbuf= maxto; *tv_io= tvbuf;
 103   } else {
 104     if (timercmp(rbuf,&maxto,>)) *rbuf= maxto;
 105   }
 106 fprintf(stderr,"inter_maxto maxto=%ld.%06ld result=%ld.%06ld\n",
 107         maxto.tv_sec,maxto.tv_usec,(**tv_io).tv_sec,(**tv_io).tv_usec);
 108 }
 109
 110 static void inter_maxtoabs(struct timeval **tv_io, struct timeval *tvbuf,
 111                            struct timeval now, struct timeval maxtime) {
 112   ldiv_t dr;
 113
 114 fprintf(stderr,"inter_maxtoabs now=%ld.%06ld maxtime=%ld.%06ld\n",
 115         now.tv_sec,now.tv_usec,maxtime.tv_sec,maxtime.tv_usec);
 116   if (!tv_io) return;
 117   maxtime.tv_sec -= (now.tv_sec+2);
 118   maxtime.tv_usec -= (now.tv_usec-2000000);
 119   dr= ldiv(maxtime.tv_usec,1000000);
 120   maxtime.tv_sec += dr.quot;
 121   maxtime.tv_usec -= dr.quot*1000000;
 122   if (maxtime.tv_sec<0) timerclear(&maxtime);
 123   inter_maxto(tv_io,tvbuf,maxtime);
 124 }
 125
 126 static void inter_addfd(int *maxfd, fd_set *fds, int fd) {
 127   if (!maxfd || !fds) return;
 128   if (fd>=*maxfd) *maxfd= fd+1;
 129   FD_SET(fd,fds);
 130 }
 131
 132 static void checktimeouts(adns_state ads, struct timeval now,
 133                           struct timeval **tv_io, struct timeval *tvbuf) {
 134   adns_query qu, nqu;
 135
 136   for (qu= ads->timew.head; qu; qu= nqu) {
 137     nqu= qu->next;
 138     if (timercmp(&now,&qu->timeout,>)) {
 139       LIST_UNLINK(ads->timew,qu);
 140       if (qu->state != query_udp) {
 141         adns__query_fail(ads,qu,adns_s_timeout);
 142       } else {
 143         adns__query_udp(ads,qu,now);
 144       }
 145     } else {
 146       inter_maxtoabs(tv_io,tvbuf,now,qu->timeout);
 147     }
 148   }
 149 }
 150
 151 void adns_interest(adns_state ads, int *maxfd,
 152                    fd_set *readfds, fd_set *writefds, fd_set *exceptfds,
 153                    struct timeval **tv_io, struct timeval *tvbuf) {
 154   struct timeval now;
 155   struct timeval tvto_lr;
 156   int r;
 157
 158 fprintf(stderr,"adns_interest\n");
 159
 160 r= gettimeofday(&now,0);
 161   if (r) {
 162     adns__warn(ads,-1,"gettimeofday failed - will sleep for a bit: %s",strerror(errno));
 163     timerclear(&tvto_lr); timevaladd(&tvto_lr,LOCALRESOURCEMS);
 164     inter_maxto(tv_io, tvbuf, tvto_lr);
 165   } else {
 166     checktimeouts(ads,now,tv_io,tvbuf);
 167   }
 168
 169   inter_addfd(maxfd,readfds,ads->udpsocket);
 170
 171   switch (ads->tcpstate) {
 172   case server_disconnected:
 173     break;
 174   case server_connecting:
 175     inter_addfd(maxfd,writefds,ads->tcpsocket);
 176     break;
 177   case server_ok:
 178     inter_addfd(maxfd,readfds,ads->tcpsocket);
 179     inter_addfd(maxfd,exceptfds,ads->tcpsocket);
 180     if (ads->tcpsend.used) inter_addfd(maxfd,writefds,ads->tcpsocket);
 181   default:
 182     abort();
 183   }
 184 }
 185
 186 /* Callback procedures - these do the real work of reception and timeout, etc. */
 187
 188 static int callb_checkfd(int maxfd, const fd_set *fds, int fd) {
 189   return maxfd<0 || !fds ? 1 :
 190          fd<maxfd && FD_ISSET(fd,fds);
 191 }
 192
 193 static int internal_callback(adns_state ads, int maxfd,
 194                              const fd_set *readfds, const fd_set *writefds,
 195                              const fd_set *exceptfds,
 196                              struct timeval now) {
 197   int skip, want, dgramlen, count, udpaddrlen, r, serv;
 198   byte udpbuf[MAXUDPDGRAM];
 199   struct sockaddr_in udpaddr;
 200
 201   count= 0;
 202
 203   switch (ads->tcpstate) {
 204   case server_disconnected:
 205     break;
 206   case server_connecting:
 207     if (callb_checkfd(maxfd,writefds,ads->tcpsocket)) {
 208       count++;
 209       assert(ads->tcprecv.used==0);
 210       adns__vbuf_ensure(&ads->tcprecv,1);
 211       if (ads->tcprecv.buf) {
 212         r= read(ads->tcpsocket,&ads->tcprecv.buf,1);
 213         if (r==0 || (r<0 && (errno==EAGAIN || errno==EWOULDBLOCK))) {
 214           tcp_connected(ads,now);
 215         } else if (r>0) {
 216           adns__tcp_broken(ads,"connect/read","sent data before first request");
 217         } else if (errno!=EINTR) {
 218           adns__tcp_broken(ads,"connect/read",strerror(errno));
 219         }
 220       }
 221     }
 222     break;
 223   case server_ok:
 224     count+= callb_checkfd(maxfd,readfds,ads->tcpsocket) +
 225             callb_checkfd(maxfd,exceptfds,ads->tcpsocket) +
 226       (ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket));
 227     if (callb_checkfd(maxfd,readfds,ads->tcpsocket)) {
 228       skip= 0;
 229       for (;;) {
 230         if (ads->tcprecv.used<skip+2) {
 231           want= 2;
 232         } else {
 233           dgramlen= (ads->tcprecv.buf[skip]<<8) | ads->tcprecv.buf[skip+1];
 234           if (ads->tcprecv.used<skip+2+dgramlen) {
 235             want= 2+dgramlen;
 236           } else {
 237             adns__procdgram(ads,ads->tcprecv.buf+skip+2,dgramlen,ads->tcpserver);
 238             skip+= 2+dgramlen; continue;
 239           }
 240         }
 241         ads->tcprecv.used -= skip;
 242         memmove(ads->tcprecv.buf,ads->tcprecv.buf+skip,ads->tcprecv.used);
 243         adns__vbuf_ensure(&ads->tcprecv,want);
 244         if (ads->tcprecv.used >= ads->tcprecv.avail) break;
 245         r= read(ads->tcpsocket,
 246                 ads->tcprecv.buf+ads->tcprecv.used,
 247                 ads->tcprecv.avail-ads->tcprecv.used);
 248         if (r>0) {
 249           ads->tcprecv.used+= r;
 250         } else {
 251           if (r<0) {
 252             if (errno==EAGAIN || errno==EWOULDBLOCK || errno==ENOMEM) break;
 253             if (errno==EINTR) continue;
 254           }
 255           adns__tcp_broken(ads,"read",r?strerror(errno):"closed");
 256           break;
 257         }
 258       }
 259     } else if (callb_checkfd(maxfd,exceptfds,ads->tcpsocket)) {
 260       adns__tcp_broken(ads,"select","exceptional condition detected");
 261     } else if (ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket)) {
 262       r= write(ads->tcpsocket,ads->tcpsend.buf,ads->tcpsend.used);
 263       if (r<0) {
 264         if (errno!=EAGAIN && errno!=EWOULDBLOCK && errno!=ENOMEM && errno!=EINTR) {
 265           adns__tcp_broken(ads,"write",strerror(errno));
 266         }
 267       } else if (r>0) {
 268         ads->tcpsend.used -= r;
 269         memmove(ads->tcpsend.buf,ads->tcpsend.buf+r,ads->tcpsend.used);
 270       }
 271     }
 272   default:
 273     abort();
 274   }
 275
 276   if (callb_checkfd(maxfd,readfds,ads->udpsocket)) {
 277     count++;
 278     for (;;) {
 279       udpaddrlen= sizeof(udpaddr);
 280       r= recvfrom(ads->udpsocket,udpbuf,sizeof(udpbuf),0,&udpaddr,&udpaddrlen);
 281       if (r<0) {
 282         if (!(errno == EAGAIN || errno == EWOULDBLOCK ||
 283               errno == EINTR || errno == ENOMEM || errno == ENOBUFS))
 284           adns__warn(ads,-1,"datagram receive error: %s",strerror(errno));
 285         break;
 286       }
 287       if (udpaddrlen != sizeof(udpaddr)) {
 288         adns__diag(ads,-1,"datagram received with wrong address length %d (expected %d)",
 289                    udpaddrlen,sizeof(udpaddr));
 290         continue;
 291       }
 292       if (udpaddr.sin_family != AF_INET) {
 293         adns__diag(ads,-1,"datagram received with wrong protocol family"
 294                    " %u (expected %u)",udpaddr.sin_family,AF_INET);
 295         continue;
 296       }
 297       if (ntohs(udpaddr.sin_port) != NSPORT) {
 298         adns__diag(ads,-1,"datagram received from wrong port %u (expected %u)",
 299                    ntohs(udpaddr.sin_port),NSPORT);
 300         continue;
 301       }
 302       for (serv= 0;
 303            serv < ads->nservers &&
 304              ads->servers[serv].addr.s_addr != udpaddr.sin_addr.s_addr;
 305            serv++);
 306       if (serv >= ads->nservers) {
 307         adns__warn(ads,-1,"datagram received from unknown nameserver %s",
 308                    inet_ntoa(udpaddr.sin_addr));
 309         continue;
 310       }
 311       adns__procdgram(ads,udpbuf,r,serv);
 312     }
 313   }
 314   return count;
 315 }
 316
 317 int adns_callback(adns_state ads, int maxfd,
 318                   const fd_set *readfds, const fd_set *writefds,
 319                   const fd_set *exceptfds) {
 320   struct timeval now;
 321   int r;
 322
 323   r= gettimeofday(&now,0); if (r) return -1;
 324   checktimeouts(ads,now,0,0);
 325   return internal_callback(ads,maxfd,readfds,writefds,exceptfds,now);
 326 }
 327
 328 /* User-visible functions and their implementation. */
 329
 330 void adns__autosys(adns_state ads, struct timeval now) {
 331   if (ads->iflags & adns_if_noautosys) return;
 332   adns_callback(ads,-1,0,0,0);
 333 }
 334
 335 static int internal_check(adns_state ads,
 336                           adns_query *query_io,
 337                           adns_answer **answer,
 338                           void **context_r) {
 339   adns_query qu;
 340
 341   qu= *query_io;
 342   if (!qu) {
 343     if (!ads->output.head) return EWOULDBLOCK;
 344     qu= ads->output.head;
 345   } else {
 346     if (qu->id>=0) return EWOULDBLOCK;
 347   }
 348   LIST_UNLINK(ads->output,qu);
 349   *answer= (adns_answer*)qu->answer.buf;
 350   if (context_r) *context_r= qu->context.ext;
 351   free(qu);
 352   return 0;
 353 }
 354
 355 int adns_wait(adns_state ads,
 356               adns_query *query_io,
 357               adns_answer **answer_r,
 358               void **context_r) {
 359   int r, maxfd, rsel, rcb;
 360   fd_set readfds, writefds, exceptfds;
 361   struct timeval tvbuf, *tvp;
 362
 363   for (;;) {
 364     r= internal_check(ads,query_io,answer_r,context_r);
 365     if (r != EWOULDBLOCK) return r;
 366     maxfd= 0; tvp= 0;
 367     FD_ZERO(&readfds); FD_ZERO(&writefds); FD_ZERO(&exceptfds);
 368     adns_interest(ads,&maxfd,&readfds,&writefds,&exceptfds,&tvp,&tvbuf);
 369     rsel= select(maxfd,&readfds,&writefds,&exceptfds,tvp);
 370     if (rsel==-1) return r;
 371     rcb= adns_callback(ads,maxfd,&readfds,&writefds,&exceptfds);
 372     assert(rcb==rsel);
 373   }
 374 }
 375
 376 int adns_check(adns_state ads,
 377                adns_query *query_io,
 378                adns_answer **answer_r,
 379                void **context_r) {
 380   struct timeval now;
 381   int r;
 382
 383   r= gettimeofday(&now,0); if (r) return errno;
 384   adns__autosys(ads,now);
 385   return internal_check(ads,query_io,answer_r,context_r);
 386 }