X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ianmdlvl/git?p=adns.git;a=blobdiff_plain;f=src%2Fevent.c;h=7702fe1d6d773c55818c16345fe8a527641e4555;hp=983e266949ef600677db2021f3a822571a206cfc;hb=bfc2c80e5bed3c17f3cf7451939458b0357bfd38;hpb=a8768e80b11709ee11b70a769f0ea86964988e3b diff --git a/src/event.c b/src/event.c index 983e266..7702fe1 100644 --- a/src/event.c +++ b/src/event.c @@ -22,16 +22,30 @@ * Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. */ -#include #include #include +#include +#include +#include #include +#include +#include #include #include "internal.h" -/* TCP connection management */ +/* TCP connection management. */ + +void adns__tcp_closenext(adns_state ads) { + int serv; + + serv= ads->tcpserver; + close(ads->tcpsocket); + ads->tcpstate= server_disconnected; + ads->tcprecv.used= ads->tcpsend.used= 0; + ads->tcpserver= (serv+1)%ads->nservers; +} void adns__tcp_broken(adns_state ads, const char *what, const char *why) { int serv; @@ -40,12 +54,11 @@ void adns__tcp_broken(adns_state ads, const char *what, const char *why) { assert(ads->tcpstate == server_connecting || ads->tcpstate == server_ok); serv= ads->tcpserver; adns__warn(ads,serv,0,"TCP connection lost: %s: %s",what,why); - close(ads->tcpsocket); - ads->tcpstate= server_disconnected; + adns__tcp_closenext(ads); for (qu= ads->timew.head; qu; qu= nqu) { nqu= qu->next; - if (qu->state == query_udp) continue; + if (qu->state == query_tosend) continue; assert(qu->state == query_tcpwait || qu->state == query_tcpsent); qu->state= query_tcpwait; qu->tcpfailed |= (1<tcprecv.used= ads->tcpsend.used= 0; - ads->tcpserver= (serv+1)%ads->nservers; } static void tcp_connected(adns_state ads, struct timeval now) { @@ -66,7 +76,7 @@ static void tcp_connected(adns_state ads, struct timeval now) { ads->tcpstate= server_ok; for (qu= ads->timew.head; qu; qu= nqu) { nqu= qu->next; - if (qu->state == query_udp) continue; + if (qu->state == query_tosend) continue; assert (qu->state == query_tcpwait); adns__query_tcp(qu,now); } @@ -100,7 +110,7 @@ void adns__tcp_tryconnect(adns_state ads, struct timeval now) { addr.sin_family= AF_INET; addr.sin_port= htons(DNS_PORT); addr.sin_addr= ads->servers[ads->tcpserver].addr; - r= connect(fd,&addr,sizeof(addr)); + r= connect(fd,(const struct sockaddr*)&addr,sizeof(addr)); ads->tcpsocket= fd; ads->tcpstate= server_connecting; if (r==0) { tcp_connected(ads,now); continue; } @@ -109,9 +119,20 @@ void adns__tcp_tryconnect(adns_state ads, struct timeval now) { } } -/* `Interest' functions - find out which fd's we might be interested in, - * and when we want to be called back for a timeout. - */ +/* Timeout handling functions. */ + +void adns__must_gettimeofday(adns_state ads, const struct timeval **now_io, + struct timeval *tv_buf) { + const struct timeval *now; + int r; + + now= *now_io; + if (now) return; + r= gettimeofday(tv_buf,0); if (!r) { *now_io= tv_buf; return; } + adns__diag(ads,-1,0,"gettimeofday failed: %s",strerror(errno)); + adns_globalsystemfailure(ads); + return; +} static void inter_maxto(struct timeval **tv_io, struct timeval *tvbuf, struct timeval maxto) { @@ -144,228 +165,136 @@ static void inter_maxtoabs(struct timeval **tv_io, struct timeval *tvbuf, inter_maxto(tv_io,tvbuf,maxtime); } -static void inter_addfd(int *maxfd, fd_set *fds, int fd) { - if (!maxfd || !fds) return; - if (fd>=*maxfd) *maxfd= fd+1; - FD_SET(fd,fds); -} - -static void checktimeouts(adns_state ads, struct timeval now, - struct timeval **tv_io, struct timeval *tvbuf) { +void adns__timeouts(adns_state ads, int act, + struct timeval **tv_io, struct timeval *tvbuf, + struct timeval now) { adns_query qu, nqu; - + for (qu= ads->timew.head; qu; qu= nqu) { nqu= qu->next; - if (timercmp(&now,&qu->timeout,>)) { + if (!timercmp(&now,&qu->timeout,>)) { + if (!tv_io) continue; + inter_maxtoabs(tv_io,tvbuf,now,qu->timeout); + } else { + if (!act) continue; LIST_UNLINK(ads->timew,qu); - if (qu->state != query_udp) { + if (qu->state != query_tosend) { adns__query_fail(qu,adns_s_timeout); } else { - adns__query_udp(qu,now); + adns__query_send(qu,now); } - } else { - inter_maxtoabs(tv_io,tvbuf,now,qu->timeout); + nqu= ads->timew.head; } } +} + +void adns_firsttimeout(adns_state ads, + struct timeval **tv_io, struct timeval *tvbuf, + struct timeval now) { + adns__consistency(ads,0,cc_entex); + adns__timeouts(ads, 0, tv_io,tvbuf, now); + adns__consistency(ads,0,cc_entex); } -static void checkfds(adns_state ads, int *maxfd, - fd_set *readfds, fd_set *writefds, fd_set *exceptfds, - int avail, struct pollfd *buf, int *count_r) { - int count; +void adns_processtimeouts(adns_state ads, const struct timeval *now) { + struct timeval tv_buf; + + adns__consistency(ads,0,cc_entex); + adns__must_gettimeofday(ads,&now,&tv_buf); + if (now) adns__timeouts(ads, 1, 0,0, *now); + adns__consistency(ads,0,cc_entex); +} + +/* fd handling functions. These are the top-level of the real work of + * reception and often transmission. + */ + +int adns__pollfds(adns_state ads, struct pollfd pollfds_buf[MAX_POLLFDS]) { + /* Returns the number of entries filled in. Always zeroes revents. */ - count= 0; - inter_addfd(ads->udpsocket,maxfd,readfds); - poll_addfd(ads->udpsocket,avail,buf,&count,POLLIN); + assert(MAX_POLLFDS==2); + + pollfds_buf[0].fd= ads->udpsocket; + pollfds_buf[0].events= POLLIN; + pollfds_buf[0].revents= 0; switch (ads->tcpstate) { case server_disconnected: - break; + return 1; case server_connecting: - inter_addfd(ads->tcpsocket,maxfd,writefds); - poll_addfd(ads->tcpsocket,avail,buf,&count,POLLOUT); + pollfds_buf[1].events= POLLOUT; break; case server_ok: - inter_addfd(ads->tcpsocket,maxfd,readfds); - inter_addfd(ads->tcpsocket,maxfd,exceptfds); - if (ads->tcpsend.used) inter_addfd(ads->tcpsocket,maxfd,writefds); - poll_addfd(ads->tcpsocket,avail,buf,&count, - ads->tcpsend.used ? POLLIN|POLLOUT : POLLIN); + pollfds_buf[1].events= ads->tcpsend.used ? POLLIN|POLLOUT|POLLPRI : POLLIN|POLLPRI; break; default: abort(); } + pollfds_buf[1].fd= ads->tcpsocket; + return 2; } -struct pollfd *adns_pollfds(adns_state ads, struct pollfd *buf, - int *len_io, int *timeout_io) { - struct timeval now, tvbuf, *tvp; - int timeout; - - r= gettimeofday(&now,0); - if (r) return 0; - - timeout= *timeout_io; - if (timeout < 0) { - tvtop= 0; - } else { - tvbuf.tv_sec= now.tv_sec + (timeout/1000); - tvbuf.tv_usec= now.tv_usec + (timeout%1000)*1000; - if (tvbuf.tv_sec >= 1000000) { - tvbuf.tv_sec += 1; - tvbuf.tv_usec -= 1000000; - } - tvp= &tvbuf; - } - checktimouts(ads,now,&tvp,&tvbuf); - - if (tvp) { - assert(tvbuf.tv_sectv_sec - now.tv_sec)*1000 + (tvp->tv_usec - now.tv_usec)/1000; - } else { - *timeout_io= -1; - } - - avail= *len_io; - if (avail == -1) { - buf= ads->pollfdsbuf; - avail= 2; - if (!buf) { - buf= ads->pollfdsbuf= malloc(sizeof(struct pollfd)*avail); - if (!buf) return 0; - } - } - checkfds(ads, 0,0,0,0, avail,buf,len_io); - if (*len_io > avail) { errno= ENOSPC; return 0; } - - return buf; -} - -void adns_interest(adns_state ads, int *maxfd, - fd_set *readfds, fd_set *writefds, fd_set *exceptfds, - struct timeval **tv_io, struct timeval *tvbuf) { - struct timeval now; - struct timeval tvto_lr; - int r; - -/*fprintf(stderr,"adns_interest\n");*/ - - r= gettimeofday(&now,0); - if (r) { - adns__warn(ads,-1,0,"gettimeofday failed - will sleep for a bit: %s", - strerror(errno)); - timerclear(&tvto_lr); timevaladd(&tvto_lr,LOCALRESOURCEMS); - inter_maxto(tv_io, tvbuf, tvto_lr); - } else { - checktimeouts(ads,now,tv_io,tvbuf); - } - - checkfds(ads, maxfd,readfds,writefds,exceptfds, 0,0,0); -} - -/* Callback procedures - these do the real work of reception and timeout, etc. */ - -static int callb_checkfd(int maxfd, const fd_set *fds, int fd) { - return maxfd<0 || !fds ? 1 : - fdtcpstate) { case server_disconnected: - break; case server_connecting: - if (callb_checkfd(maxfd,writefds,ads->tcpsocket)) { - count++; - assert(ads->tcprecv.used==0); - if (!adns__vbuf_ensure(&ads->tcprecv,1)) return -1; - if (ads->tcprecv.buf) { - r= read(ads->tcpsocket,&ads->tcprecv.buf,1); - if (r==0 || (r<0 && (errno==EAGAIN || errno==EWOULDBLOCK))) { - tcp_connected(ads,now); - } else if (r>0) { - adns__tcp_broken(ads,"connect/read","sent data before first request"); - } else if (errno!=EINTR) { - adns__tcp_broken(ads,"connect/read",strerror(errno)); - } - } - } break; case server_ok: - count+= callb_checkfd(maxfd,readfds,ads->tcpsocket) + - callb_checkfd(maxfd,exceptfds,ads->tcpsocket) + - (ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket)); - if (callb_checkfd(maxfd,readfds,ads->tcpsocket)) { - skip= 0; - for (;;) { - if (ads->tcprecv.usedtcprecv.buf[skip]<<8) | ads->tcprecv.buf[skip+1]; - if (ads->tcprecv.usedtcprecv.buf+skip+2,dgramlen,ads->tcpserver,now); - skip+= 2+dgramlen; continue; - } - } - ads->tcprecv.used -= skip; - memmove(ads->tcprecv.buf,ads->tcprecv.buf+skip,ads->tcprecv.used); - skip= 0; - if (!adns__vbuf_ensure(&ads->tcprecv,want)) return -1; - assert(ads->tcprecv.used <= ads->tcprecv.avail); - if (ads->tcprecv.used == ads->tcprecv.avail) continue; - r= read(ads->tcpsocket, - ads->tcprecv.buf+ads->tcprecv.used, - ads->tcprecv.avail-ads->tcprecv.used); - if (r>0) { - ads->tcprecv.used+= r; + if (fd != ads->tcpsocket) break; + skip= 0; + for (;;) { + if (ads->tcprecv.usedtcprecv.buf[skip]<<8) | ads->tcprecv.buf[skip+1]; + if (ads->tcprecv.usedtcprecv.buf+skip+2,dgramlen,ads->tcpserver,1,*now); + skip+= 2+dgramlen; continue; } } - } else if (callb_checkfd(maxfd,exceptfds,ads->tcpsocket)) { - adns__tcp_broken(ads,"select","exceptional condition detected"); - } else if (ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket)) { - r= write(ads->tcpsocket,ads->tcpsend.buf,ads->tcpsend.used); - if (r<0) { - if (errno!=EAGAIN && errno!=EWOULDBLOCK && errno!=ENOMEM && errno!=EINTR) { - adns__tcp_broken(ads,"write",strerror(errno)); + ads->tcprecv.used -= skip; + memmove(ads->tcprecv.buf,ads->tcprecv.buf+skip,ads->tcprecv.used); + skip= 0; + if (!adns__vbuf_ensure(&ads->tcprecv,want)) { r= ENOMEM; goto xit; } + assert(ads->tcprecv.used <= ads->tcprecv.avail); + if (ads->tcprecv.used == ads->tcprecv.avail) continue; + r= read(ads->tcpsocket, + ads->tcprecv.buf+ads->tcprecv.used, + ads->tcprecv.avail-ads->tcprecv.used); + if (r>0) { + ads->tcprecv.used+= r; + } else { + if (r) { + if (errno==EAGAIN || errno==EWOULDBLOCK) { r= 0; goto xit; } + if (errno==EINTR) continue; + if (errno_resources(errno)) { r= errno; goto xit; } } - } else if (r>0) { - ads->tcpsend.used -= r; - memmove(ads->tcpsend.buf,ads->tcpsend.buf+r,ads->tcpsend.used); + adns__tcp_broken(ads,"read",r?strerror(errno):"closed"); + r= 0; goto xit; } - } - break; + } /* never reached */ default: abort(); } - - if (callb_checkfd(maxfd,readfds,ads->udpsocket)) { - count++; + if (fd == ads->udpsocket) { for (;;) { udpaddrlen= sizeof(udpaddr); - r= recvfrom(ads->udpsocket,udpbuf,sizeof(udpbuf),0,&udpaddr,&udpaddrlen); + r= recvfrom(ads->udpsocket,udpbuf,sizeof(udpbuf),0, + (struct sockaddr*)&udpaddr,&udpaddrlen); if (r<0) { - if (!(errno == EAGAIN || errno == EWOULDBLOCK || - errno == EINTR || errno == ENOMEM || errno == ENOBUFS)) - adns__warn(ads,-1,0,"datagram receive error: %s",strerror(errno)); - break; + if (errno == EAGAIN || errno == EWOULDBLOCK) { r= 0; goto xit; } + if (errno == EINTR) continue; + if (errno_resources(errno)) { r= errno; goto xit; } + adns__warn(ads,-1,0,"datagram receive error: %s",strerror(errno)); + r= 0; goto xit; } if (udpaddrlen != sizeof(udpaddr)) { adns__diag(ads,-1,0,"datagram received with wrong address length %d" @@ -391,28 +320,224 @@ static int internal_callback(adns_state ads, int maxfd, inet_ntoa(udpaddr.sin_addr)); continue; } - adns__procdgram(ads,udpbuf,r,serv,now); + adns__procdgram(ads,udpbuf,r,serv,0,*now); } } - return count; + r= 0; +xit: + adns__consistency(ads,0,cc_entex); + return r; } -int adns_callback(adns_state ads, int maxfd, - const fd_set *readfds, const fd_set *writefds, - const fd_set *exceptfds) { - struct timeval now; +int adns_processwriteable(adns_state ads, int fd, const struct timeval *now) { int r; + + adns__consistency(ads,0,cc_entex); + + switch (ads->tcpstate) { + case server_disconnected: + break; + case server_connecting: + if (fd != ads->tcpsocket) break; + assert(ads->tcprecv.used==0); + for (;;) { + if (!adns__vbuf_ensure(&ads->tcprecv,1)) { r= ENOMEM; goto xit; } + r= read(ads->tcpsocket,&ads->tcprecv.buf,1); + if (r==0 || (r<0 && (errno==EAGAIN || errno==EWOULDBLOCK))) { + tcp_connected(ads,*now); + r= 0; goto xit; + } + if (r>0) { + adns__tcp_broken(ads,"connect/read","sent data before first request"); + r= 0; goto xit; + } + if (errno==EINTR) continue; + if (errno_resources(errno)) { r= errno; goto xit; } + adns__tcp_broken(ads,"connect/read",strerror(errno)); + r= 0; goto xit; + } /* not reached */ + case server_ok: + if (!(ads->tcpsend.used && fd == ads->tcpsocket)) break; + for (;;) { + adns__sigpipe_protect(ads); + r= write(ads->tcpsocket,ads->tcpsend.buf,ads->tcpsend.used); + adns__sigpipe_unprotect(ads); + if (r<0) { + if (errno==EINTR) continue; + if (errno==EAGAIN || errno==EWOULDBLOCK) { r= 0; goto xit; } + if (errno_resources(errno)) { r= errno; goto xit; } + adns__tcp_broken(ads,"write",strerror(errno)); + r= 0; goto xit; + } else if (r>0) { + ads->tcpsend.used -= r; + memmove(ads->tcpsend.buf,ads->tcpsend.buf+r,ads->tcpsend.used); + } + } /* not reached */ + default: + abort(); + } + r= 0; +xit: + adns__consistency(ads,0,cc_entex); + return r; +} + +int adns_processexceptional(adns_state ads, int fd, const struct timeval *now) { + adns__consistency(ads,0,cc_entex); + switch (ads->tcpstate) { + case server_disconnected: + break; + case server_connecting: + case server_ok: + if (fd != ads->tcpsocket) break; + adns__tcp_broken(ads,"poll/select","exceptional condition detected"); + break; + default: + abort(); + } + adns__consistency(ads,0,cc_entex); + return 0; +} + +static void fd_event(adns_state ads, int fd, + int revent, int pollflag, + int maxfd, const fd_set *fds, + int (*func)(adns_state, int fd, const struct timeval *now), + struct timeval now, int *r_r) { + int r; + + if (!(revent & pollflag)) return; + if (fds && !(fd= maxfd) maxfd= fd+1; + revents= pollfds[i].revents; + fd_event(ads,fd, revents,POLLIN, maxfd,readfds, adns_processreadable,now,r_r); + fd_event(ads,fd, revents,POLLOUT, maxfd,writefds, adns_processwriteable,now,r_r); + fd_event(ads,fd, revents,POLLPRI, maxfd,exceptfds, adns_processexceptional,now,r_r); + } } -/* User-visible functions and their implementation. */ +/* Wrappers for select(2). */ + +void adns_beforeselect(adns_state ads, int *maxfd_io, fd_set *readfds_io, + fd_set *writefds_io, fd_set *exceptfds_io, + struct timeval **tv_mod, struct timeval *tv_tobuf, + const struct timeval *now) { + struct timeval tv_nowbuf; + struct pollfd pollfds[MAX_POLLFDS]; + int i, fd, maxfd, npollfds; + + adns__consistency(ads,0,cc_entex); + + if (tv_mod && (!*tv_mod || (*tv_mod)->tv_sec || (*tv_mod)->tv_usec)) { + /* The caller is planning to sleep. */ + adns__must_gettimeofday(ads,&now,&tv_nowbuf); + if (!now) goto xit; + adns__timeouts(ads, 1, tv_mod,tv_tobuf, *now); + } + + npollfds= adns__pollfds(ads,pollfds); + maxfd= *maxfd_io; + for (i=0; i= maxfd) maxfd= fd+1; + if (pollfds[i].events & POLLIN) FD_SET(fd,readfds_io); + if (pollfds[i].events & POLLOUT) FD_SET(fd,writefds_io); + if (pollfds[i].events & POLLPRI) FD_SET(fd,exceptfds_io); + } + *maxfd_io= maxfd; + +xit: + adns__consistency(ads,0,cc_entex); +} + +void adns_afterselect(adns_state ads, int maxfd, const fd_set *readfds, + const fd_set *writefds, const fd_set *exceptfds, + const struct timeval *now) { + struct timeval tv_buf; + struct pollfd pollfds[MAX_POLLFDS]; + int npollfds, i; + + adns__consistency(ads,0,cc_entex); + adns__must_gettimeofday(ads,&now,&tv_buf); + if (!now) goto xit; + adns_processtimeouts(ads,now); + + npollfds= adns__pollfds(ads,pollfds); + for (i=0; itimew.head) { + adns__query_fail(ads->timew.head, adns_s_systemfail); + } + + switch (ads->tcpstate) { + case server_connecting: + case server_ok: + adns__tcp_closenext(ads); + break; + case server_disconnected: + break; + default: + abort(); + } + adns__consistency(ads,0,cc_entex); +} + +int adns_processany(adns_state ads) { + int r; + struct timeval now; + struct pollfd pollfds[MAX_POLLFDS]; + int npollfds; + + adns__consistency(ads,0,cc_entex); + + r= gettimeofday(&now,0); + if (!r) adns_processtimeouts(ads,&now); + + npollfds= adns__pollfds(ads,pollfds); + adns__fdevents(ads, + pollfds,npollfds, + 0,0,0,0, + now,&r); + + adns__consistency(ads,0,cc_entex); + return r; +} void adns__autosys(adns_state ads, struct timeval now) { if (ads->iflags & adns_if_noautosys) return; - adns_callback(ads,-1,0,0,0); + adns_processany(ads); } static int internal_check(adns_state ads, @@ -431,6 +556,7 @@ static int internal_check(adns_state ads, LIST_UNLINK(ads->output,qu); *answer= qu->answer; if (context_r) *context_r= qu->ctx.ext; + *query_io= qu; free(qu); return 0; } @@ -439,24 +565,32 @@ int adns_wait(adns_state ads, adns_query *query_io, adns_answer **answer_r, void **context_r) { - int r, maxfd, rsel, rcb; + int r, maxfd, rsel; fd_set readfds, writefds, exceptfds; struct timeval tvbuf, *tvp; + adns__consistency(ads,*query_io,cc_entex); for (;;) { r= internal_check(ads,query_io,answer_r,context_r); - if (r != EWOULDBLOCK) return r; + if (r != EWOULDBLOCK) break; maxfd= 0; tvp= 0; FD_ZERO(&readfds); FD_ZERO(&writefds); FD_ZERO(&exceptfds); - adns_interest(ads,&maxfd,&readfds,&writefds,&exceptfds,&tvp,&tvbuf); + adns_beforeselect(ads,&maxfd,&readfds,&writefds,&exceptfds,&tvp,&tvbuf,0); rsel= select(maxfd,&readfds,&writefds,&exceptfds,tvp); if (rsel==-1) { - if (errno == EINTR && !(ads->iflags & adns_if_eintr)) continue; - return errno; + if (errno == EINTR) { + if (ads->iflags & adns_if_eintr) { r= EINTR; break; } + } else { + adns__diag(ads,-1,0,"select failed in wait: %s",strerror(errno)); + adns_globalsystemfailure(ads); + } + } else { + assert(rsel >= 0); + adns_afterselect(ads,maxfd,&readfds,&writefds,&exceptfds,0); } - rcb= adns_callback(ads,maxfd,&readfds,&writefds,&exceptfds); - assert(rcb==rsel); } + adns__consistency(ads,0,cc_entex); + return r; } int adns_check(adns_state ads, @@ -466,7 +600,11 @@ int adns_check(adns_state ads, struct timeval now; int r; - r= gettimeofday(&now,0); if (r) return errno; - adns__autosys(ads,now); - return internal_check(ads,query_io,answer_r,context_r); + adns__consistency(ads,*query_io,cc_entex); + r= gettimeofday(&now,0); + if (!r) adns__autosys(ads,now); + + r= internal_check(ads,query_io,answer_r,context_r); + adns__consistency(ads,0,cc_entex); + return r; }