[adns] / src / event.c

/**/

static void autosys(adns_state ads, struct timeval now) {
  if (ads->iflags & adns_if_noautosys) return;
  adns_callback(ads,-1,0,0,0);
}

static int callb_checkfd(int maxfd, const fd_set *fds, int fd) {
  return maxfd<0 || !fds ? 1 :
         fd<maxfd && FD_ISSET(fd,fds);
}

int adns_callback(adns_state ads, int maxfd,
		  const fd_set *readfds, const fd_set *writefds,
		  const fd_set *exceptfds) {
  int skip, dgramlen, count;
  enum adns__tcpstate oldtcpstate;

  count= 0;
  oldtcpstate= ads->tcpstate;
  
  if (ads->tcpstate == server_connecting) {
    if (callb_checkfd(maxfd,writefds,ads->tcpsocket)) {
      count++;
      assert(ads->tcprecv.used==0);
      vbuf_ensure(&ads->tcprecv,1);
      if (ads->tcprecv.buf) {
	r= read(ads->tcpsocket,&ads->tcprecv.buf,1);
	if (r==0 || (r<0 && (errno==EAGAIN || errno==EWOULDBLOCK))) {
	  diag("nameserver %s TCP connection made",
	       inet_ntoa(ads->servers[ads->tcpserver].addr));
	  ads->tcpstate= server_connected;
	} else if (r>0) {
	  tcpserver_broken(ads,"connect/read","sent data before first request");
	} else if (errno!=EINTR) {
	  tcpserver_broken(ads,"connect",strerror(errno));
	}
      }
    }
  }
  if (ads->tcpstate == server_connected) {
    if (oldtcpstate == server_connected)
      count+= callb_checkfd(maxfd,readfds,ads->tcpsocket) +
	      callb_checkfd(maxfd,exceptfds,ads->tcpsocket) +
	(ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket));
    if (oldtcpstate != server_connected || callb_checkfd(maxfd,readfds,ads->tcpsocket)) {
      skip= 0;
      for (;;) {
	if (ads->tcprecv.used<skip+2) {
	  want= 2;
	} else {
	  dgramlen= (ads->tcprecv.buf[skip]<<8) | ads->tcprecv.buf[skip+1];
	  if (ads->tcprecv.used<skip+2+dgramlen) {
	    want= 2+dgramlen;
	  } else {
	    procdgram(ads,ads->tcprecv.buf+skip+2,dgramlen,-1);
	    skip+= 2+dgramlen; continue;
	  }
	}
	Ads->tcprecv.used -= skip;
	memmove(ads->tcprecv.buf,ads->tcprecv.buf+skip,ads->tcprecv.used);
	vbuf_ensure(&ads->tcprecv,want);
	if (ads->tcprecv.used >= ads->tcprecv.avail) break;
	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<0) {
	    if (errno==EAGAIN || errno==EWOULDBLOCK || errno==ENOMEM) break;
	    if (errno==EINTR) continue;
	  }
	  tcpserver_broken(ads->tcpserver,"read",r?strerror(errno):"closed");
	  break;
	}
      }
    } else if (callb_checkfd(maxfd,exceptfds,ads->tcpsocket)) {
      tcpserver_broken(ads->tcpserver,"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) {
	  tcpserver_broken(ads->tcpserver,"write",strerror(errno));
	}
      } else if (r>0) {
	ads->tcpsend.used -= r;
	memmove(ads->tcpsend.buf,ads->tcpsend.buf+r,ads->tcpsend.used);
      }
    }
  }

  if (
    break;
	
      
	}
	  
  tcpserver_broken(
		
	    if (ads-
	  used= 0;
	  for (;;) {
	  vbuf_ensure(&ads->tcprecv,2);
	  vbuf_ensure(&ads->tcprecv,
	  if (ads->tcprecv.avail<2) break;
      if (ads->tcprecv.used
      
      if (ads->tcprecv.used<2 && ads->tcprecv.avail
      if (ads->tcprecv.used<2 && ads->tcprecv.avail
      r= read(ads->tcpsocket,
      if (adns->tcprecv.used<2) {
	if (
	  
  if (ads->tcpstate != server_disc) {
    
      
    }
  if (maxfd<0 || !readfds || (FD_ISSET
      ads->
      
  abort(); /* FIXME */
}
	  diag("nameserver #%d (%s) TCP connection died: %s",
	       inet_ntoa(ads->servers[tcpserver].addr),

static void inter_maxto(struct timeval **tv_io, struct timeval *tvbuf,
			struct timeval maxto) {
  struct timeval rbuf;

  rbuf= *tv_io;
  if (!rbuf) { *tvbuf= maxto; *tv_io= tvbuf; return; }
  if (timercmp(rbuf,&maxto,>)) *rbuf= maxto;
}

static void inter_maxtoabs(struct timeval **tv_io, struct timeval *tvbuf,
			   struct timeval now, struct timeval maxtime) {
  ldiv_t dr;
  
  maxtime.tv_sec -= (now.tv_sec-1);
  maxtime.tv_usec += (1000-now.tv_usec);
  dr= ldiv(maxtime.tv_usec,1000);
  maxtime.tv_sec += dr.quot;
  maxtime.tv_usec -= dr.rem;
  inter_maxto(tv_io,tvbuf,maxtime);
}

static void localresourcerr(struct timeval **tv_io, struct timeval *tvbuf,
			    const char *syscall) {
  struct timeval tvto_lr;
  
  diag(ads,"local system resources scarce (during %s): %s",syscall,strerror(errno));
  timerclear(&tvto_lr); timevaladd(&tvto_lr,LOCALRESOURCEMS);
  inter_maxto(tv_io, tvbuf, tvto_lr);
  return;
}

static void inter_addfd(int *maxfd, fd_set *fds, int fd) {
  if (fd>=*maxfd) *maxfd= fd+1;
  FD_SET(fd,fds);
}

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;
  adns_query qu;
  int r;
  
  r= gettimeofday(&now,0);
  if (r) { localresourcerr(tv_io,tvbuf,"gettimeofday"); return; }

  for (qu= ads->timew; qu; qu= nqu) {
    nqu= qu->next;
    if (timercmp(&now,qu->timeout,>)) {
      DLIST_UNLINK(ads->timew,qu);
      if (qu->nextudpserver == -1) {
	query_fail(ads,qu,adns_s_notresponding);
      } else {
	DLIST_LINKTAIL(ads->tosend,qu);
      }
    } else {
      inter_maxtoabs(tv_io,tvbuf,now,qu->timeout);
    }
  }
  
  for (qu= ads->tosend; qu; qu= nqu) {
    nqu= qu->next;
    quproc_tosend(ads,qu,now);
  }

  inter_addfd(maxfd,readfds,ads->udpsocket);
  switch (ads->tcpstate) {
  case server_disc:
    break;
  case server_connecting:
    inter_addfd(maxfd,writefds,ads->tcpsocket);
    break;
  case server_connected:
    inter_addfd(maxfd,readfds,ads->tcpsocket);
    inter_addfd(maxfd,exceptfds,ads->tcpsocket);
    if (ads->opbufused) inter_addfd(maxfd,writefds,ads->tcpsocket);
  default:
    abort();
  }
  
}

static int internal_check(adns_state ads,
			  adns_query *query_io,
			  adns_answer **answer,
			  void **context_r) {
  adns_query qu;

  qu= *query_io;
  if (!qu) {
    if (!ads->output.head) return EWOULDBLOCK;
    qu= ads->output.head;
  } else {
    if (qu->id>=0) return EWOULDBLOCK;
  }
  LIST_UNLINK(ads->output,qu);
  *answer= qu->answer;
  if (context_r) *context_r= qu->context;
  free(qu);
  return 0;
}

int adns_wait(adns_state ads,
	      adns_query *query_io,
	      adns_answer **answer_r,
	      void **context_r) {
  int r, maxfd, rsel, rcb;
  fd_set readfds, writefds, exceptfds;
  struct timeval tvbuf, *tvp;
  
  for (;;) {
    r= internal_check(ads,query_io,answer_r,context_r);
    if (r && r != EWOULDBLOCK) return r;
    FD_ZERO(&readfds); FD_ZERO(&writefds); FD_ZERO(&exceptfds);
    maxfd= 0; tvp= 0;
    adns_interest(ads,&maxfd,&readfds,&writefds,&exceptfds,&tvp,&tvbuf);
    rsel= select(maxfd,&readfds,&writefds,&exceptfds,tvp);
    if (rsel==-1) return r;
    rcb= adns_callback(ads,maxfd,&readfds,&writefds,&exceptfds);
    assert(rcb==rsel);
  }
}

int adns_check(adns_state ads,
	       adns_query *query_io,
	       adns_answer **answer_r,
	       void **context_r) {
  autosys(ads);
  return internal_check(ads,query_io,answer_r,context_r);
}
Commit	Line	Data
656b2da9	1	/**/
	2
	3	static void autosys(adns_state ads, struct timeval now) {
	4	if (ads->iflags & adns_if_noautosys) return;
	5	adns_callback(ads,-1,0,0,0);
	6	}
	7
	8	static int callb_checkfd(int maxfd, const fd_set *fds, int fd) {
	9	return maxfd<0 \|\| !fds ? 1 :
	10	fd<maxfd && FD_ISSET(fd,fds);
	11	}
	12
	13	int adns_callback(adns_state ads, int maxfd,
	14	const fd_set readfds, const fd_set writefds,
	15	const fd_set *exceptfds) {
	16	int skip, dgramlen, count;
	17	enum adns__tcpstate oldtcpstate;
	18
	19	count= 0;
	20	oldtcpstate= ads->tcpstate;
	21
	22	if (ads->tcpstate == server_connecting) {
	23	if (callb_checkfd(maxfd,writefds,ads->tcpsocket)) {
	24	count++;
	25	assert(ads->tcprecv.used==0);
	26	vbuf_ensure(&ads->tcprecv,1);
	27	if (ads->tcprecv.buf) {
	28	r= read(ads->tcpsocket,&ads->tcprecv.buf,1);
	29	if (r==0 \|\| (r<0 && (errno==EAGAIN \|\| errno==EWOULDBLOCK))) {
	30	diag("nameserver %s TCP connection made",
	31	inet_ntoa(ads->servers[ads->tcpserver].addr));
	32	ads->tcpstate= server_connected;
	33	} else if (r>0) {
	34	tcpserver_broken(ads,"connect/read","sent data before first request");
	35	} else if (errno!=EINTR) {
	36	tcpserver_broken(ads,"connect",strerror(errno));
	37	}
	38	}
	39	}
	40	}
	41	if (ads->tcpstate == server_connected) {
	42	if (oldtcpstate == server_connected)
	43	count+= callb_checkfd(maxfd,readfds,ads->tcpsocket) +
	44	callb_checkfd(maxfd,exceptfds,ads->tcpsocket) +
	45	(ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket));
	46	if (oldtcpstate != server_connected \|\| callb_checkfd(maxfd,readfds,ads->tcpsocket)) {
	47	skip= 0;
	48	for (;;) {
	49	if (ads->tcprecv.used<skip+2) {
	50	want= 2;
	51	} else {
	52	dgramlen= (ads->tcprecv.buf[skip]<<8) \| ads->tcprecv.buf[skip+1];
	53	if (ads->tcprecv.used<skip+2+dgramlen) {
	54	want= 2+dgramlen;
	55	} else {
	56	procdgram(ads,ads->tcprecv.buf+skip+2,dgramlen,-1);
	57	skip+= 2+dgramlen; continue;
	58	}
	59	}
	60	Ads->tcprecv.used -= skip;
	61	memmove(ads->tcprecv.buf,ads->tcprecv.buf+skip,ads->tcprecv.used);
	62	vbuf_ensure(&ads->tcprecv,want);
	63	if (ads->tcprecv.used >= ads->tcprecv.avail) break;
	64	r= read(ads->tcpsocket,
65	ads->tcprecv.buf+ads->tcprecv.used,
66	ads->tcprecv.avail-ads->tcprecv.used);
67	if (r>0) {
68	ads->tcprecv.used+= r;
69	} else {
70	if (r<0) {
71	if (errno==EAGAIN \|\| errno==EWOULDBLOCK \|\| errno==ENOMEM) break;
72	if (errno==EINTR) continue;
73	}
74	tcpserver_broken(ads->tcpserver,"read",r?strerror(errno):"closed");
75	break;
76	}
77	}
78	} else if (callb_checkfd(maxfd,exceptfds,ads->tcpsocket)) {
79	tcpserver_broken(ads->tcpserver,"select","exceptional condition detected");
80	} else if (ads->tcpsend.used && callb_checkfd(maxfd,writefds,ads->tcpsocket)) {
81	r= write(ads->tcpsocket,ads->tcpsend.buf,ads->tcpsend.used);
82	if (r<0) {
83	if (errno!=EAGAIN && errno!=EWOULDBLOCK && errno!=ENOMEM && errno!=EINTR) {
84	tcpserver_broken(ads->tcpserver,"write",strerror(errno));
85	}
86	} else if (r>0) {
87	ads->tcpsend.used -= r;
88	memmove(ads->tcpsend.buf,ads->tcpsend.buf+r,ads->tcpsend.used);
89	}
90	}
91	}
92
93	if (
94	break;
95
96
97	}
98
99	tcpserver_broken(
100
101	if (ads-
102	used= 0;
103	for (;;) {
104	vbuf_ensure(&ads->tcprecv,2);
105	vbuf_ensure(&ads->tcprecv,
106	if (ads->tcprecv.avail<2) break;
107	if (ads->tcprecv.used
108
109	if (ads->tcprecv.used<2 && ads->tcprecv.avail
110	if (ads->tcprecv.used<2 && ads->tcprecv.avail
111	r= read(ads->tcpsocket,
112	if (adns->tcprecv.used<2) {
113	if (
114
115	if (ads->tcpstate != server_disc) {
116
117
118	}
119	if (maxfd<0 \|\| !readfds \|\| (FD_ISSET
120	ads->
121
122	abort(); /* FIXME */
123	}
124	diag("nameserver #%d (%s) TCP connection died: %s",
125	inet_ntoa(ads->servers[tcpserver].addr),
126
127	static void inter_maxto(struct timeval *tv_io, struct timeval tvbuf,
128	struct timeval maxto) {
129	struct timeval rbuf;
130
131	rbuf= *tv_io;
132	if (!rbuf) { tvbuf= maxto; tv_io= tvbuf; return; }
133	if (timercmp(rbuf,&maxto,>)) *rbuf= maxto;
134	}
135
136	static void inter_maxtoabs(struct timeval *tv_io, struct timeval tvbuf,
137	struct timeval now, struct timeval maxtime) {
138	ldiv_t dr;
139
140	maxtime.tv_sec -= (now.tv_sec-1);
141	maxtime.tv_usec += (1000-now.tv_usec);
142	dr= ldiv(maxtime.tv_usec,1000);
143	maxtime.tv_sec += dr.quot;
144	maxtime.tv_usec -= dr.rem;
145	inter_maxto(tv_io,tvbuf,maxtime);
146	}
147
148	static void localresourcerr(struct timeval *tv_io, struct timeval tvbuf,
149	const char *syscall) {
150	struct timeval tvto_lr;
151
152	diag(ads,"local system resources scarce (during %s): %s",syscall,strerror(errno));
153	timerclear(&tvto_lr); timevaladd(&tvto_lr,LOCALRESOURCEMS);
154	inter_maxto(tv_io, tvbuf, tvto_lr);
155	return;
156	}
157
158	static void inter_addfd(int maxfd, fd_set fds, int fd) {
159	if (fd>=maxfd) maxfd= fd+1;
160	FD_SET(fd,fds);
161	}
162
163	void adns_interest(adns_state ads, int *maxfd,
164	fd_set readfds, fd_set writefds, fd_set *exceptfds,
165	struct timeval *tv_io, struct timeval tvbuf) {
166	struct timeval now;
167	adns_query qu;
168	int r;
169
170	r= gettimeofday(&now,0);
171	if (r) { localresourcerr(tv_io,tvbuf,"gettimeofday"); return; }
172
173	for (qu= ads->timew; qu; qu= nqu) {
174	nqu= qu->next;
175	if (timercmp(&now,qu->timeout,>)) {
176	DLIST_UNLINK(ads->timew,qu);
177	if (qu->nextudpserver == -1) {
178	query_fail(ads,qu,adns_s_notresponding);
179	} else {
180	DLIST_LINKTAIL(ads->tosend,qu);
181	}
182	} else {
183	inter_maxtoabs(tv_io,tvbuf,now,qu->timeout);
184	}
185	}
186
187	for (qu= ads->tosend; qu; qu= nqu) {
188	nqu= qu->next;
189	quproc_tosend(ads,qu,now);
190	}
191
192	inter_addfd(maxfd,readfds,ads->udpsocket);
193	switch (ads->tcpstate) {
194	case server_disc:
195	break;
196	case server_connecting:
197	inter_addfd(maxfd,writefds,ads->tcpsocket);
198	break;
199	case server_connected:
200	inter_addfd(maxfd,readfds,ads->tcpsocket);
201	inter_addfd(maxfd,exceptfds,ads->tcpsocket);
202	if (ads->opbufused) inter_addfd(maxfd,writefds,ads->tcpsocket);
203	default:
204	abort();
205	}
206
207	}
208
209	static int internal_check(adns_state ads,
210	adns_query *query_io,
211	adns_answer **answer,
212	void **context_r) {
213	adns_query qu;
214
215	qu= *query_io;
216	if (!qu) {
217	if (!ads->output.head) return EWOULDBLOCK;
218	qu= ads->output.head;
219	} else {
220	if (qu->id>=0) return EWOULDBLOCK;
221	}
222	LIST_UNLINK(ads->output,qu);
223	*answer= qu->answer;
224	if (context_r) *context_r= qu->context;
225	free(qu);
226	return 0;
227	}
228
229	int adns_wait(adns_state ads,
230	adns_query *query_io,
231	adns_answer **answer_r,
232	void **context_r) {
233	int r, maxfd, rsel, rcb;
234	fd_set readfds, writefds, exceptfds;
235	struct timeval tvbuf, *tvp;
236
237	for (;;) {
238	r= internal_check(ads,query_io,answer_r,context_r);
239	if (r && r != EWOULDBLOCK) return r;
240	FD_ZERO(&readfds); FD_ZERO(&writefds); FD_ZERO(&exceptfds);
241	maxfd= 0; tvp= 0;
242	adns_interest(ads,&maxfd,&readfds,&writefds,&exceptfds,&tvp,&tvbuf);
243	rsel= select(maxfd,&readfds,&writefds,&exceptfds,tvp);
244	if (rsel==-1) return r;
245	rcb= adns_callback(ads,maxfd,&readfds,&writefds,&exceptfds);
246	assert(rcb==rsel);
247	}
248	}
249
250	int adns_check(adns_state ads,
251	adns_query *query_io,
252	adns_answer **answer_r,
253	void **context_r) {
254	autosys(ads);
255	return internal_check(ads,query_io,answer_r,context_r);
256	}