1 /* ioset.h - srvx event loop
2 * Copyright 2002-2004 srvx Development Team
4 * This file is part of srvx.
6 * srvx is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License as published by
8 * the Free Software Foundation; either version 2 of the License, or
9 * (at your option) any later version.
11 * This program is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 * GNU General Public License for more details.
16 * You should have received a copy of the GNU General Public License
17 * along with srvx; if not, write to the Free Software Foundation,
18 * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
30 #ifdef HAVE_SYS_SELECT_H
31 #include <sys/select.h>
33 #ifdef HAVE_SYS_SOCKET_H
34 #include <sys/socket.h>
41 #define IS_EOL(CH) ((CH) == '\n')
43 extern int uplink_connect(void);
44 static int clock_skew;
48 static struct io_fd **fds;
49 static unsigned int fds_size;
50 static fd_set read_fds, write_fds;
53 ioq_init(struct ioq *ioq, int size) {
54 ioq->buf = malloc(size);
55 ioq->get = ioq->put = 0;
60 ioq_put_avail(const struct ioq *ioq) {
61 /* Subtract 1 from ioq->get to be sure we don't fill the buffer
62 * and make it look empty even when there's data in it. */
63 if (ioq->put < ioq->get)
64 return ioq->get - ioq->put - 1;
65 else if (ioq->get == 0)
66 return ioq->size - ioq->put - 1;
68 return ioq->size - ioq->put;
72 ioq_get_avail(const struct ioq *ioq) {
73 return ((ioq->put < ioq->get) ? ioq->size : ioq->put) - ioq->get;
77 ioq_used(const struct ioq *ioq) {
78 return ((ioq->put < ioq->get) ? ioq->size : 0) + ioq->put - ioq->get;
82 ioq_grow(struct ioq *ioq) {
83 int new_size = ioq->size << 1;
84 char *new_buf = malloc(new_size);
85 int get_avail = ioq_get_avail(ioq);
86 memcpy(new_buf, ioq->buf + ioq->get, get_avail);
87 if (ioq->put < ioq->get)
88 memcpy(new_buf + get_avail, ioq->buf, ioq->put);
90 ioq->put = ioq_used(ioq);
94 return new_size - ioq->put;
108 log_module(MAIN_LOG, LOG_ERROR, "Somebody called ioset_add(%d) on a negative fd!", fd);
111 res = calloc(1, sizeof(*res));
115 ioq_init(&res->send, 1024);
116 ioq_init(&res->recv, 1024);
117 if ((unsigned)fd >= fds_size) {
118 unsigned int old_size = fds_size;
120 fds = realloc(fds, fds_size*sizeof(*fds));
121 memset(fds+old_size, 0, (fds_size-old_size)*sizeof(*fds));
124 flags = fcntl(fd, F_GETFL);
125 fcntl(fd, F_SETFL, flags|O_NONBLOCK);
130 ioset_connect(struct sockaddr *local, unsigned int sa_size, const char *peer, unsigned int port, int blocking, void *data, void (*connect_cb)(struct io_fd *fd, int error)) {
133 struct sockaddr_in sin;
136 if (!getipbyname(peer, &ip)) {
137 log_module(MAIN_LOG, LOG_ERROR, "getipbyname(%s) failed.", peer);
140 sin.sin_addr.s_addr = ip;
142 if ((fd = socket(local->sa_family, SOCK_STREAM, 0)) < 0) {
143 log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s)", peer, errno, strerror(errno));
146 if (bind(fd, local, sa_size) < 0) {
147 log_module(MAIN_LOG, LOG_ERROR, "bind() of socket for %s (fd %d) returned errno %d (%s). Will let operating system choose.", peer, fd, errno, strerror(errno));
150 if ((fd = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
151 log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s).", peer, errno, strerror(errno));
155 sin.sin_family = AF_INET;
156 sin.sin_port = htons(port);
158 res = connect(fd, (struct sockaddr*)&sin, sizeof(sin));
159 io_fd = ioset_add(fd);
161 io_fd = ioset_add(fd);
162 res = connect(fd, (struct sockaddr*)&sin, sizeof(sin));
169 io_fd->connect_cb = connect_cb;
172 case EINPROGRESS: /* only if !blocking */
175 log_module(MAIN_LOG, LOG_ERROR, "connect(%s:%d) (fd %d) returned errno %d (%s).", peer, port, io_fd->fd, errno, strerror(errno));
176 /* then fall through */
179 ioset_close(io_fd->fd, 1);
184 connect_cb(io_fd, ((res < 0) ? errno : 0));
189 ioset_try_write(struct io_fd *fd) {
191 unsigned int req = ioq_get_avail(&fd->send);
192 res = write(fd->fd, fd->send.buf+fd->send.get, req);
198 log_module(MAIN_LOG, LOG_ERROR, "write() on fd %d error %d: %s", fd->fd, errno, strerror(errno));
202 if (fd->send.get == fd->send.size)
208 ioset_close(int fd, int os_close) {
210 if (!(fdp = fds[fd]))
214 fdp->destroy_cb(fdp);
215 if (fdp->send.get != fdp->send.put) {
216 int flags = fcntl(fd, F_GETFL);
217 fcntl(fd, F_SETFL, flags&~O_NONBLOCK);
218 ioset_try_write(fdp);
219 /* it may need to send the beginning of the buffer now.. */
220 if (fdp->send.get != fdp->send.put)
221 ioset_try_write(fdp);
228 FD_CLR(fd, &read_fds);
229 FD_CLR(fd, &write_fds);
233 ioset_find_line_length(struct io_fd *fd) {
234 unsigned int pos, max, len;
236 max = (fd->recv.put < fd->recv.get) ? fd->recv.size : fd->recv.put;
237 for (pos = fd->recv.get; pos < max; ++pos, ++len)
238 if (IS_EOL(fd->recv.buf[pos]))
239 return fd->line_len = len + 1;
240 if (fd->recv.put < fd->recv.get)
241 for (pos = 0; pos < fd->recv.put; ++pos, ++len)
242 if (IS_EOL(fd->recv.buf[pos]))
243 return fd->line_len = len + 1;
244 return fd->line_len = 0;
248 ioset_buffered_read(struct io_fd *fd) {
249 int put_avail, nbr, fdnum;
251 if (!(put_avail = ioq_put_avail(&fd->recv)))
252 put_avail = ioq_grow(&fd->recv);
253 nbr = read(fd->fd, fd->recv.buf + fd->recv.put, put_avail);
259 log_module(MAIN_LOG, LOG_ERROR, "Unexpected read() error %d on fd %d: %s", errno, fd->fd, strerror(errno));
260 /* Just flag it as EOF and call readable_cb() to notify the fd's owner. */
265 } else if (nbr == 0) {
270 if (fd->line_len == 0) {
272 for (pos = fd->recv.put; pos < fd->recv.put + nbr; ++pos) {
273 if (IS_EOL(fd->recv.buf[pos])) {
274 if (fd->recv.put < fd->recv.get)
275 fd->line_len = fd->recv.size + pos + 1 - fd->recv.get;
277 fd->line_len = pos + 1 - fd->recv.get;
283 if (fd->recv.put == fd->recv.size)
286 while (fd->wants_reads && (fd->line_len > 0)) {
289 break; /* make sure they didn't close on us */
290 ioset_find_line_length(fd);
296 ioset_line_read(struct io_fd *fd, char *dest, int max) {
298 if (fd->eof && (!ioq_get_avail(&fd->recv) || (fd->line_len < 0)))
300 if (fd->line_len < 0)
302 if (fd->line_len < max)
304 avail = ioq_get_avail(&fd->recv);
306 memcpy(dest, fd->recv.buf + fd->recv.get, avail);
307 fd->recv.get += avail;
308 assert(fd->recv.get == fd->recv.size);
314 memcpy(dest + done, fd->recv.buf + fd->recv.get, max - done);
315 fd->recv.get += max - done;
316 if (fd->recv.get == fd->recv.size)
319 ioset_find_line_length(fd);
324 #define debug_fdsets(MSG, NFDS, READ_FDS, WRITE_FDS, EXCEPT_FDS, SELECT_TIMEOUT) (void)0
327 debug_fdsets(const char *msg, int nfds, fd_set *read_fds, fd_set *write_fds, fd_set *except_fds, struct timeval *select_timeout) {
328 static const char *flag_text[8] = { "---", "r", "w", "rw", "e", "er", "ew", "erw" };
333 for (pos=ii=0; ii<nfds; ++ii) {
334 flags = (read_fds && FD_ISSET(ii, read_fds)) ? 1 : 0;
335 flags |= (write_fds && FD_ISSET(ii, write_fds)) ? 2 : 0;
336 flags |= (except_fds && FD_ISSET(ii, except_fds)) ? 4 : 0;
339 pos += sprintf(buf+pos, " %d%s", ii, flag_text[flags]);
341 gettimeofday(&now, NULL);
342 if (select_timeout) {
343 log_module(MAIN_LOG, LOG_DEBUG, "%s, at "FMT_TIME_T".%06ld:%s (timeout "FMT_TIME_T".%06ld)", msg, now.tv_sec, now.tv_usec, buf, select_timeout->tv_sec, select_timeout->tv_usec);
345 log_module(MAIN_LOG, LOG_DEBUG, "%s, at "FMT_TIME_T".%06ld:%s (no timeout)", msg, now.tv_sec, now.tv_usec, buf);
352 extern struct io_fd *socket_io_fd;
353 struct timeval select_timeout;
355 int select_result, max_fd;
359 while (!quit_services) {
360 while (!socket_io_fd)
363 /* How long to sleep? (fill in select_timeout) */
364 wakey = timeq_next();
365 if ((wakey - now) < 0)
366 select_timeout.tv_sec = 0;
368 select_timeout.tv_sec = wakey - now;
369 select_timeout.tv_usec = 0;
371 /* Set up read_fds and write_fds fdsets. */
375 for (nn=0; nn<fds_size; nn++) {
380 FD_SET(nn, &read_fds);
381 if ((fd->send.get != fd->send.put) || !fd->connected)
382 FD_SET(nn, &write_fds);
385 /* Check for activity, update time. */
386 debug_fdsets("Entering select", max_fd+1, &read_fds, &write_fds, NULL, &select_timeout);
387 select_result = select(max_fd + 1, &read_fds, &write_fds, NULL, &select_timeout);
388 debug_fdsets("After select", max_fd+1, &read_fds, &write_fds, NULL, &select_timeout);
389 now = time(NULL) + clock_skew;
390 if (select_result < 0) {
391 if (errno != EINTR) {
392 log_module(MAIN_LOG, LOG_ERROR, "select() error %d: %s", errno, strerror(errno));
398 /* Call back anybody that has connect or read activity and wants to know. */
399 for (nn=0; nn<fds_size; nn++) {
402 if (FD_ISSET(nn, &read_fds)) {
404 ioset_buffered_read(fd);
408 if (FD_ISSET(nn, &write_fds) && !fd->connected) {
409 int rc, arglen = sizeof(rc);
410 if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
414 fd->connect_cb(fd, rc);
416 /* Note: check whether write FD is still set, since the
417 * connect_cb() might close the FD, making us dereference
418 * a free()'d pointer for the fd.
420 if (FD_ISSET(nn, &write_fds) && (fd->send.get != fd->send.put))
424 /* Call any timeq events we need to call. */
431 extern char *services_config;
432 conf_read(services_config);
439 ioset_write(struct io_fd *fd, const char *buf, unsigned int nbw) {
441 while (ioq_used(&fd->send) + nbw >= fd->send.size)
443 avail = ioq_put_avail(&fd->send);
445 memcpy(fd->send.buf + fd->send.put, buf, avail);
450 memcpy(fd->send.buf + fd->send.put, buf, nbw);
452 if (fd->send.put == fd->send.size)
457 ioset_set_time(unsigned long new_now) {
458 clock_skew = new_now - time(NULL);