src/ioset.c

   1 /* ioset.h - srvx event loop
   2  * Copyright 2002-2004, 2006 srvx Development Team
   3  *
   4  * This file is part of srvx.
   5  *
   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.
  10  *
  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.
  15  *
  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.
  19  */
  20
  21 #include "ioset-impl.h"
  22 #include "log.h"
  23 #include "timeq.h"
  24 #include "saxdb.h"
  25 #include "conf.h"
  26
  27 #ifdef HAVE_FCNTL_H
  28 #include <fcntl.h>
  29 #endif
  30 #ifdef HAVE_SYS_SOCKET_H
  31 #include <sys/socket.h>
  32 #endif
  33
  34 #define IS_EOL(CH) ((CH) == '\n')
  35
  36 extern int uplink_connect(void);
  37 int clock_skew;
  38 int do_write_dbs;
  39 int do_reopen;
  40 static struct io_engine *engine;
  41 static struct io_fd *active_fd;
  42
  43 static void
  44 ioq_init(struct ioq *ioq, int size) {
  45     ioq->buf = malloc(size);
  46     ioq->get = ioq->put = 0;
  47     ioq->size = size;
  48 }
  49
  50 static unsigned int
  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;
  58     else
  59         return ioq->size - ioq->put;
  60 }
  61
  62 static unsigned int
  63 ioq_get_avail(const struct ioq *ioq) {
  64     return ((ioq->put < ioq->get) ? ioq->size : ioq->put) - ioq->get;
  65 }
  66
  67 static unsigned int
  68 ioq_used(const struct ioq *ioq) {
  69     return ((ioq->put < ioq->get) ? ioq->size : 0) + ioq->put - ioq->get;
  70 }
  71
  72 static unsigned int
  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);
  80     free(ioq->buf);
  81     ioq->put = ioq_used(ioq);
  82     ioq->get = 0;
  83     ioq->buf = new_buf;
  84     ioq->size = new_size;
  85     return new_size - ioq->put;
  86 }
  87
  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;
  92
  93 void
  94 ioset_init(void)
  95 {
  96     assert(engine == NULL);
  97
  98 #if WITH_IOSET_KQUEUE
  99     if (!engine && io_engine_kqueue.init())
 100         engine = &io_engine_kqueue;
 101 #endif
 102
 103 #if WITH_IOSET_EPOLL
 104     if (!engine && io_engine_epoll.init())
 105         engine = &io_engine_epoll;
 106 #endif
 107
 108 #if WITH_IOSET_WIN32
 109     if (!engine && io_engine_win32.init())
 110         engine = &io_engine_win32;
 111 #endif
 112
 113     if (engine) {
 114         /* we found one that works */
 115     } else if (io_engine_select.init())
 116         engine = &io_engine_select;
 117     else
 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);
 120 }
 121
 122 void
 123 ioset_cleanup(void) {
 124     engine->cleanup();
 125 }
 126
 127 struct io_fd *
 128 ioset_add(int fd) {
 129     struct io_fd *res;
 130     int flags;
 131
 132     if (fd < 0) {
 133         log_module(MAIN_LOG, LOG_ERROR, "Somebody called ioset_add(%d) on a negative fd!", fd);
 134         return 0;
 135     }
 136     res = calloc(1, sizeof(*res));
 137     if (!res)
 138         return 0;
 139     res->fd = fd;
 140     ioq_init(&res->send, 1024);
 141     ioq_init(&res->recv, 1024);
 142     engine->add(res);
 143     flags = fcntl(fd, F_GETFL);
 144     fcntl(fd, F_SETFL, flags|O_NONBLOCK);
 145     return res;
 146 }
 147
 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))
 149 {
 150     struct io_fd *io_fd;
 151     unsigned int opt;
 152     int res;
 153     int fd;
 154
 155     fd = socket(local ? local->sa_family : PF_INET, SOCK_STREAM, 0);
 156     if (fd < 0) {
 157         log_module(MAIN_LOG, LOG_ERROR, "Unable to create listening socket: %s", strerror(errno));
 158         return NULL;
 159     }
 160
 161     if (local && sa_size) {
 162         res = bind(fd, local, sa_size);
 163         if (res < 0) {
 164             log_module(MAIN_LOG, LOG_ERROR, "Unable to bind listening socket %d: %s", fd, strerror(errno));
 165             close(fd);
 166             return NULL;
 167         }
 168
 169         opt = 1;
 170         res = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
 171         if (res < 0) {
 172             log_module(MAIN_LOG, LOG_WARNING, "Unable to mark listener address as re-usable: %s", strerror(errno));
 173         }
 174     }
 175
 176     res = listen(fd, 1);
 177     if (res < 0) {
 178         log_module(MAIN_LOG, LOG_ERROR, "Unable to listen on socket %d: %s", fd, strerror(errno));
 179         close(fd);
 180         return NULL;
 181     }
 182
 183     io_fd = ioset_add(fd);
 184     if (!io_fd) {
 185         close(fd);
 186         return NULL;
 187     }
 188     io_fd->state = IO_LISTENING;
 189     io_fd->data = data;
 190     io_fd->accept_cb = accept_cb;
 191     engine->update(io_fd);
 192     return io_fd;
 193 }
 194
 195 struct 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     int fd, res;
 198     struct io_fd *io_fd;
 199     struct addrinfo hints, *ai;
 200     char portnum[10];
 201
 202     memset(&hints, 0, sizeof(hints));
 203     hints.ai_family = local ? local->sa_family : 0;
 204     hints.ai_socktype = SOCK_STREAM;
 205     snprintf(portnum, sizeof(portnum), "%u", port);
 206     if (getaddrinfo(peer, portnum, &hints, &ai)) {
 207         log_module(MAIN_LOG, LOG_ERROR, "getaddrinfo(%s, %s) failed.", peer, portnum);
 208         return NULL;
 209     }
 210
 211     if (local) {
 212         if ((fd = socket(local->sa_family, SOCK_STREAM, 0)) < 0) {
 213             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s)", peer, errno, strerror(errno));
 214             freeaddrinfo(ai);
 215             return NULL;
 216         }
 217         if (bind(fd, local, sa_size) < 0) {
 218             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));
 219         }
 220     } else {
 221         if ((fd = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
 222             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s).", peer, errno, strerror(errno));
 223             freeaddrinfo(ai);
 224             return NULL;
 225         }
 226     }
 227
 228     if (blocking) {
 229         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
 230         io_fd = ioset_add(fd);
 231     } else {
 232         io_fd = ioset_add(fd);
 233         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
 234     }
 235     freeaddrinfo(ai);
 236     if (!io_fd) {
 237         close(fd);
 238         return NULL;
 239     }
 240     io_fd->state = IO_CONNECTING;
 241     io_fd->data = data;
 242     io_fd->connect_cb = connect_cb;
 243     if (res < 0) {
 244         switch (errno) {
 245         case EINPROGRESS: /* only if !blocking */
 246             engine->update(io_fd);
 247             return io_fd;
 248         default:
 249             log_module(MAIN_LOG, LOG_ERROR, "connect(%s:%d) (fd %d) returned errno %d (%s).", peer, port, io_fd->fd, errno, strerror(errno));
 250             /* then fall through */
 251         case EHOSTUNREACH:
 252         case ECONNREFUSED:
 253             ioset_close(io_fd, 1);
 254             engine->update(io_fd);
 255             return NULL;
 256         }
 257     }
 258     io_fd->state = IO_CONNECTED;
 259     if (connect_cb)
 260         connect_cb(io_fd, ((res < 0) ? errno : 0));
 261     engine->update(io_fd);
 262     return io_fd;
 263 }
 264
 265 static void
 266 ioset_try_write(struct io_fd *fd) {
 267     int res;
 268     unsigned int req = ioq_get_avail(&fd->send);
 269     res = write(fd->fd, fd->send.buf+fd->send.get, req);
 270     if (res < 0) {
 271         switch (errno) {
 272         case EAGAIN:
 273             break;
 274         default:
 275             log_module(MAIN_LOG, LOG_ERROR, "write() on fd %d error %d: %s", fd->fd, errno, strerror(errno));
 276         }
 277     } else {
 278         fd->send.get += res;
 279         if (fd->send.get == fd->send.size)
 280             fd->send.get = 0;
 281         engine->update(fd);
 282     }
 283 }
 284
 285 void
 286 ioset_close(struct io_fd *fdp, int os_close) {
 287     if (!fdp)
 288         return;
 289     if (active_fd == fdp)
 290         active_fd = NULL;
 291     if (fdp->destroy_cb)
 292         fdp->destroy_cb(fdp);
 293     if (fdp->send.get != fdp->send.put) {
 294         int flags;
 295
 296         flags = fcntl(fdp->fd, F_GETFL);
 297         fcntl(fdp->fd, F_SETFL, flags&~O_NONBLOCK);
 298         ioset_try_write(fdp);
 299         /* it may need to send the beginning of the buffer now.. */
 300         if (fdp->send.get != fdp->send.put)
 301             ioset_try_write(fdp);
 302     }
 303     free(fdp->send.buf);
 304     free(fdp->recv.buf);
 305     if (os_close)
 306         close(fdp->fd);
 307     engine->remove(fdp);
 308     free(fdp);
 309 }
 310
 311 static void
 312 ioset_accept(struct io_fd *listener)
 313 {
 314     struct io_fd *old_active_fd;
 315     struct io_fd *new_fd;
 316     int fd;
 317
 318     fd = accept(listener->fd, NULL, 0);
 319     if (fd < 0) {
 320         log_module(MAIN_LOG, LOG_ERROR, "Unable to accept new connection on listener %d: %s", listener->fd, strerror(errno));
 321         return;
 322     }
 323
 324     new_fd = ioset_add(fd);
 325     new_fd->state = IO_CONNECTED;
 326     old_active_fd = active_fd;
 327     active_fd = new_fd;
 328     listener->accept_cb(listener, new_fd);
 329     assert(active_fd == NULL || active_fd == new_fd);
 330     if (active_fd == new_fd) {
 331         if (new_fd->send.get != new_fd->send.put)
 332             ioset_try_write(new_fd);
 333         else
 334             engine->update(new_fd);
 335     }
 336     active_fd = old_active_fd;
 337 }
 338
 339 static int
 340 ioset_find_line_length(struct io_fd *fd) {
 341     unsigned int pos, max, len;
 342     len = 0;
 343     max = (fd->recv.put < fd->recv.get) ? fd->recv.size : fd->recv.put;
 344     for (pos = fd->recv.get; pos < max; ++pos, ++len)
 345         if (IS_EOL(fd->recv.buf[pos]))
 346             return fd->line_len = len + 1;
 347     if (fd->recv.put < fd->recv.get)
 348         for (pos = 0; pos < fd->recv.put; ++pos, ++len)
 349             if (IS_EOL(fd->recv.buf[pos]))
 350                 return fd->line_len = len + 1;
 351     return fd->line_len = 0;
 352 }
 353
 354 static void
 355 ioset_buffered_read(struct io_fd *fd) {
 356     int put_avail, nbr, fdnum;
 357
 358     if (!(put_avail = ioq_put_avail(&fd->recv)))
 359         put_avail = ioq_grow(&fd->recv);
 360     nbr = read(fd->fd, fd->recv.buf + fd->recv.put, put_avail);
 361     if (nbr < 0) {
 362         switch (errno) {
 363         case EAGAIN:
 364             break;
 365         default:
 366             log_module(MAIN_LOG, LOG_ERROR, "Unexpected read() error %d on fd %d: %s", errno, fd->fd, strerror(errno));
 367             /* Just flag it as EOF and call readable_cb() to notify the fd's owner. */
 368             fd->state = IO_CLOSED;
 369             fd->readable_cb(fd);
 370             if (active_fd == fd)
 371                 engine->update(fd);
 372         }
 373     } else if (nbr == 0) {
 374         fd->state = IO_CLOSED;
 375         fd->readable_cb(fd);
 376         if (active_fd == fd)
 377             engine->update(fd);
 378     } else {
 379         if (fd->line_len == 0) {
 380             unsigned int pos;
 381             for (pos = fd->recv.put; pos < fd->recv.put + nbr; ++pos) {
 382                 if (IS_EOL(fd->recv.buf[pos])) {
 383                     if (fd->recv.put < fd->recv.get)
 384                         fd->line_len = fd->recv.size + pos + 1 - fd->recv.get;
 385                     else
 386                         fd->line_len = pos + 1 - fd->recv.get;
 387                     break;
 388                 }
 389             }
 390         }
 391         fd->recv.put += nbr;
 392         if (fd->recv.put == fd->recv.size)
 393             fd->recv.put = 0;
 394         fdnum = fd->fd;
 395         while (fd->wants_reads && (fd->line_len > 0)) {
 396             struct io_fd *old_active;
 397
 398             old_active = active_fd;
 399             active_fd = fd;
 400             fd->readable_cb(fd);
 401             if (active_fd)
 402                 ioset_find_line_length(fd);
 403             if (old_active != fd)
 404                 active_fd = old_active;
 405         }
 406     }
 407 }
 408
 409 int
 410 ioset_line_read(struct io_fd *fd, char *dest, int max) {
 411     int avail, done;
 412     if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) ||  (fd->line_len < 0)))
 413         return 0;
 414     if (fd->line_len < 0)
 415         return -1;
 416     if (fd->line_len < max)
 417         max = fd->line_len;
 418     avail = ioq_get_avail(&fd->recv);
 419     if (max > avail) {
 420         memcpy(dest, fd->recv.buf + fd->recv.get, avail);
 421         fd->recv.get += avail;
 422         assert(fd->recv.get == fd->recv.size);
 423         fd->recv.get = 0;
 424         done = avail;
 425     } else {
 426         done = 0;
 427     }
 428     memcpy(dest + done, fd->recv.buf + fd->recv.get, max - done);
 429     fd->recv.get += max - done;
 430     if (fd->recv.get == fd->recv.size)
 431         fd->recv.get = 0;
 432     dest[max] = 0;
 433     ioset_find_line_length(fd);
 434     return max;
 435 }
 436
 437 void
 438 ioset_events(struct io_fd *fd, int readable, int writable)
 439 {
 440     if (!fd || (!readable && !writable))
 441         return;
 442     active_fd = fd;
 443     switch (fd->state) {
 444     case IO_CLOSED:
 445         break;
 446     case IO_LISTENING:
 447         if (active_fd && readable)
 448             ioset_accept(fd);
 449         break;
 450     case IO_CONNECTING:
 451         assert(active_fd == NULL || active_fd == fd);
 452         if (active_fd && writable) {
 453             socklen_t arglen;
 454             int rc;
 455             arglen = sizeof(rc);
 456             if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
 457                 rc = errno;
 458             fd->state = IO_CONNECTED;
 459             if (fd->connect_cb)
 460                 fd->connect_cb(fd, rc);
 461             if (active_fd == fd)
 462                 engine->update(fd);
 463         }
 464         /* and fall through */
 465     case IO_CONNECTED:
 466         assert(active_fd == NULL || active_fd == fd);
 467         if (active_fd && readable) {
 468             if (fd->line_reads)
 469                 ioset_buffered_read(fd);
 470             else
 471                 fd->readable_cb(fd);
 472         }
 473
 474         assert(active_fd == NULL || active_fd == fd);
 475         if (active_fd && writable)
 476             ioset_try_write(fd);
 477         break;
 478     }
 479 }
 480
 481 void
 482 ioset_run(void) {
 483     extern struct io_fd *socket_io_fd;
 484     struct timeval timeout;
 485     time_t wakey;
 486
 487     while (!quit_services) {
 488         while (!socket_io_fd)
 489             uplink_connect();
 490
 491         /* How long to sleep? (fill in select_timeout) */
 492         wakey = timeq_next();
 493         if ((wakey - now) < 0)
 494             timeout.tv_sec = 0;
 495         else
 496             timeout.tv_sec = wakey - now;
 497         timeout.tv_usec = 0;
 498
 499         if (engine->loop(&timeout))
 500             continue;
 501
 502         /* Call any timeq events we need to call. */
 503         timeq_run();
 504         if (do_write_dbs) {
 505             saxdb_write_all();
 506             do_write_dbs = 0;
 507         }
 508         if (do_reopen) {
 509             extern char *services_config;
 510             conf_read(services_config);
 511             do_reopen = 0;
 512         }
 513     }
 514 }
 515
 516 void
 517 ioset_write(struct io_fd *fd, const char *buf, unsigned int nbw) {
 518     unsigned int avail;
 519     while (ioq_used(&fd->send) + nbw >= fd->send.size)
 520         ioq_grow(&fd->send);
 521     avail = ioq_put_avail(&fd->send);
 522     if (nbw > avail) {
 523         memcpy(fd->send.buf + fd->send.put, buf, avail);
 524         buf += avail;
 525         nbw -= avail;
 526         fd->send.put = 0;
 527     }
 528     memcpy(fd->send.buf + fd->send.put, buf, nbw);
 529     fd->send.put += nbw;
 530     if (fd->send.put == fd->send.size)
 531         fd->send.put = 0;
 532     engine->update(fd);
 533 }
 534
 535 int
 536 ioset_printf(struct io_fd *fd, const char *fmt, ...) {
 537     char tmpbuf[MAXLEN];
 538     va_list ap;
 539     int res;
 540
 541     va_start(ap, fmt);
 542     res = vsnprintf(tmpbuf, sizeof(tmpbuf), fmt, ap);
 543     va_end(ap);
 544     if (res > 0 && (size_t)res <= sizeof(tmpbuf))
 545         ioset_write(fd, tmpbuf, res);
 546     return res;
 547 }
 548
 549 void
 550 ioset_set_time(unsigned long new_now) {
 551     clock_skew = new_now - time(NULL);
 552     now = new_now;
 553 }