1 /* ioset.h - srvx event loop
2 * Copyright 2002-2004, 2006 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.
21 #include "ioset-impl.h"
30 #ifdef HAVE_SYS_SOCKET_H
31 #include <sys/socket.h>
34 #define IS_EOL(CH) ((CH) == '\n')
36 extern int uplink_connect(void);
40 static struct io_engine *engine;
41 static struct io_fd *active_fd;
44 ioq_init(struct ioq *ioq, int size) {
45 ioq->buf = malloc(size);
46 ioq->get = ioq->put = 0;
51 ioq_put_avail(const struct ioq *ioq) {
52 /* Subtract 1 from ioq->get to be sure we don't fill the buffer
53 * and make it look empty even when there's data in it. */
54 if (ioq->put < ioq->get)
55 return ioq->get - ioq->put - 1;
56 else if (ioq->get == 0)
57 return ioq->size - ioq->put - 1;
59 return ioq->size - ioq->put;
63 ioq_get_avail(const struct ioq *ioq) {
64 return ((ioq->put < ioq->get) ? ioq->size : ioq->put) - ioq->get;
68 ioq_used(const struct ioq *ioq) {
69 return ((ioq->put < ioq->get) ? ioq->size : 0) + ioq->put - ioq->get;
73 ioq_grow(struct ioq *ioq) {
74 int new_size = ioq->size << 1;
75 char *new_buf = malloc(new_size);
76 int get_avail = ioq_get_avail(ioq);
77 memcpy(new_buf, ioq->buf + ioq->get, get_avail);
78 if (ioq->put < ioq->get)
79 memcpy(new_buf + get_avail, ioq->buf, ioq->put);
81 ioq->put = ioq_used(ioq);
85 return new_size - ioq->put;
88 extern struct io_engine io_engine_kqueue;
89 extern struct io_engine io_engine_epoll;
90 extern struct io_engine io_engine_win32;
91 extern struct io_engine io_engine_select;
96 assert(engine == NULL);
99 if (!engine && io_engine_kqueue.init())
100 engine = &io_engine_kqueue;
104 if (!engine && io_engine_epoll.init())
105 engine = &io_engine_epoll;
109 if (!engine && io_engine_win32.init())
110 engine = &io_engine_win32;
114 /* we found one that works */
115 } else if (io_engine_select.init())
116 engine = &io_engine_select;
118 log_module(MAIN_LOG, LOG_FATAL, "No usable I/O engine found.");
119 log_module(MAIN_LOG, LOG_DEBUG, "Using %s I/O engine.", engine->name);
123 ioset_cleanup(void) {
133 log_module(MAIN_LOG, LOG_ERROR, "Somebody called ioset_add(%d) on a negative fd!", fd);
136 res = calloc(1, sizeof(*res));
140 ioq_init(&res->send, 1024);
141 ioq_init(&res->recv, 1024);
143 flags = fcntl(fd, F_GETFL);
144 fcntl(fd, F_SETFL, flags|O_NONBLOCK);
148 struct io_fd *ioset_listen(struct sockaddr *local, unsigned int sa_size, void *data, void (*accept_cb)(struct io_fd *listener, struct io_fd *new_connect))
155 fd = socket(local ? local->sa_family : PF_INET, SOCK_STREAM, 0);
157 log_module(MAIN_LOG, LOG_ERROR, "Unable to create listening socket: %s", strerror(errno));
161 if (local && sa_size) {
162 res = bind(fd, local, sa_size);
164 log_module(MAIN_LOG, LOG_ERROR, "Unable to bind listening socket %d: %s", fd, strerror(errno));
170 res = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
172 log_module(MAIN_LOG, LOG_WARNING, "Unable to mark listener address as re-usable: %s", strerror(errno));
178 log_module(MAIN_LOG, LOG_ERROR, "Unable to listen on socket %d: %s", fd, strerror(errno));
183 io_fd = ioset_add(fd);
188 io_fd->state = IO_LISTENING;
190 io_fd->accept_cb = accept_cb;
191 engine->update(io_fd);
196 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)) {
197 struct addrinfo hints;
200 struct io_fd *old_active;
205 memset(&hints, 0, sizeof(hints));
206 hints.ai_family = local ? local->sa_family : 0;
207 hints.ai_socktype = SOCK_STREAM;
208 snprintf(portnum, sizeof(portnum), "%u", port);
209 if (getaddrinfo(peer, portnum, &hints, &ai)) {
210 log_module(MAIN_LOG, LOG_ERROR, "getaddrinfo(%s, %s) failed.", peer, portnum);
215 if ((fd = socket(local->sa_family, SOCK_STREAM, 0)) < 0) {
216 log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s)", peer, errno, strerror(errno));
220 if (bind(fd, local, sa_size) < 0) {
221 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));
224 if ((fd = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
225 log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s).", peer, errno, strerror(errno));
232 res = connect(fd, ai->ai_addr, ai->ai_addrlen);
233 io_fd = ioset_add(fd);
235 io_fd = ioset_add(fd);
236 res = connect(fd, ai->ai_addr, ai->ai_addrlen);
243 io_fd->state = IO_CONNECTING;
245 io_fd->connect_cb = connect_cb;
248 case EINPROGRESS: /* only if !blocking */
249 engine->update(io_fd);
252 log_module(MAIN_LOG, LOG_ERROR, "connect(%s:%d) (fd %d) returned errno %d (%s).", peer, port, io_fd->fd, errno, strerror(errno));
253 /* then fall through */
256 ioset_close(io_fd, 1);
257 engine->update(io_fd);
261 io_fd->state = IO_CONNECTED;
262 old_active = active_fd;
264 connect_cb(io_fd, ((res < 0) ? errno : 0));
266 engine->update(io_fd);
267 if (old_active != io_fd)
268 active_fd = old_active;
273 ioset_try_write(struct io_fd *fd) {
275 unsigned int req = ioq_get_avail(&fd->send);
276 res = write(fd->fd, fd->send.buf+fd->send.get, req);
282 log_module(MAIN_LOG, LOG_ERROR, "write() on fd %d error %d: %s", fd->fd, errno, strerror(errno));
286 if (fd->send.get == fd->send.size)
293 ioset_close(struct io_fd *fdp, int os_close) {
296 if (active_fd == fdp)
299 fdp->destroy_cb(fdp);
300 if (fdp->send.get != fdp->send.put) {
303 flags = fcntl(fdp->fd, F_GETFL);
304 fcntl(fdp->fd, F_SETFL, flags&~O_NONBLOCK);
305 ioset_try_write(fdp);
306 /* it may need to send the beginning of the buffer now.. */
307 if (fdp->send.get != fdp->send.put)
308 ioset_try_write(fdp);
319 ioset_accept(struct io_fd *listener)
321 struct io_fd *old_active;
322 struct io_fd *new_fd;
325 fd = accept(listener->fd, NULL, 0);
327 log_module(MAIN_LOG, LOG_ERROR, "Unable to accept new connection on listener %d: %s", listener->fd, strerror(errno));
331 new_fd = ioset_add(fd);
332 new_fd->state = IO_CONNECTED;
333 old_active = active_fd;
335 listener->accept_cb(listener, new_fd);
336 assert(active_fd == NULL || active_fd == new_fd);
337 if (active_fd == new_fd) {
338 if (new_fd->send.get != new_fd->send.put)
339 ioset_try_write(new_fd);
341 engine->update(new_fd);
343 active_fd = old_active;
347 ioset_find_line_length(struct io_fd *fd) {
348 unsigned int pos, max, len;
350 max = (fd->recv.put < fd->recv.get) ? fd->recv.size : fd->recv.put;
351 for (pos = fd->recv.get; pos < max; ++pos, ++len)
352 if (IS_EOL(fd->recv.buf[pos]))
353 return fd->line_len = len + 1;
354 if (fd->recv.put < fd->recv.get)
355 for (pos = 0; pos < fd->recv.put; ++pos, ++len)
356 if (IS_EOL(fd->recv.buf[pos]))
357 return fd->line_len = len + 1;
358 return fd->line_len = 0;
362 ioset_buffered_read(struct io_fd *fd) {
363 int put_avail, nbr, fdnum;
365 if (!(put_avail = ioq_put_avail(&fd->recv)))
366 put_avail = ioq_grow(&fd->recv);
367 nbr = read(fd->fd, fd->recv.buf + fd->recv.put, put_avail);
373 log_module(MAIN_LOG, LOG_ERROR, "Unexpected read() error %d on fd %d: %s", errno, fd->fd, strerror(errno));
374 /* Just flag it as EOF and call readable_cb() to notify the fd's owner. */
375 fd->state = IO_CLOSED;
380 } else if (nbr == 0) {
381 fd->state = IO_CLOSED;
386 if (fd->line_len == 0) {
388 for (pos = fd->recv.put; pos < fd->recv.put + nbr; ++pos) {
389 if (IS_EOL(fd->recv.buf[pos])) {
390 if (fd->recv.put < fd->recv.get)
391 fd->line_len = fd->recv.size + pos + 1 - fd->recv.get;
393 fd->line_len = pos + 1 - fd->recv.get;
399 if (fd->recv.put == fd->recv.size)
402 while (fd->wants_reads && (fd->line_len > 0)) {
403 struct io_fd *old_active;
406 old_active = active_fd;
410 ioset_find_line_length(fd);
413 if (old_active != fd)
414 active_fd = old_active;
422 ioset_line_read(struct io_fd *fd, char *dest, int max) {
424 if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) || (fd->line_len < 0)))
426 if (fd->line_len < 0)
428 if (fd->line_len < max)
430 avail = ioq_get_avail(&fd->recv);
432 memcpy(dest, fd->recv.buf + fd->recv.get, avail);
433 fd->recv.get += avail;
434 assert(fd->recv.get == fd->recv.size);
440 memcpy(dest + done, fd->recv.buf + fd->recv.get, max - done);
441 fd->recv.get += max - done;
442 if (fd->recv.get == fd->recv.size)
445 ioset_find_line_length(fd);
450 ioset_events(struct io_fd *fd, int readable, int writable)
452 if (!fd || (!readable && !writable))
459 if (active_fd && readable)
463 assert(active_fd == NULL || active_fd == fd);
464 if (active_fd && readable) {
468 if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
470 fd->state = IO_CLOSED;
472 fd->connect_cb(fd, rc);
473 } else if (active_fd && writable) {
474 fd->state = IO_CONNECTED;
476 fd->connect_cb(fd, 0);
481 /* and fall through */
483 assert(active_fd == NULL || active_fd == fd);
484 if (active_fd && readable) {
486 ioset_buffered_read(fd);
491 assert(active_fd == NULL || active_fd == fd);
492 if (active_fd && writable)
500 extern struct io_fd *socket_io_fd;
501 struct timeval timeout;
504 while (!quit_services) {
505 while (!socket_io_fd)
508 /* How long to sleep? (fill in select_timeout) */
509 wakey = timeq_next();
510 if ((wakey - now) < 0)
513 timeout.tv_sec = wakey - now;
516 if (engine->loop(&timeout))
519 /* Call any timeq events we need to call. */
526 extern char *services_config;
527 conf_read(services_config);
534 ioset_write(struct io_fd *fd, const char *buf, unsigned int nbw) {
536 while (ioq_used(&fd->send) + nbw >= fd->send.size)
538 avail = ioq_put_avail(&fd->send);
540 memcpy(fd->send.buf + fd->send.put, buf, avail);
545 memcpy(fd->send.buf + fd->send.put, buf, nbw);
547 if (fd->send.put == fd->send.size)
553 ioset_printf(struct io_fd *fd, const char *fmt, ...) {
559 res = vsnprintf(tmpbuf, sizeof(tmpbuf), fmt, ap);
561 if (res > 0 && (size_t)res <= sizeof(tmpbuf))
562 ioset_write(fd, tmpbuf, res);
567 ioset_set_time(unsigned long new_now) {
568 clock_skew = new_now - time(NULL);