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     if (engine) /* someone beat us to it */
  97         return;
  98
  99 #if WITH_IOSET_KQUEUE
 100     if (!engine && io_engine_kqueue.init())
 101         engine = &io_engine_kqueue;
 102 #endif
 103
 104 #if WITH_IOSET_EPOLL
 105     if (!engine && io_engine_epoll.init())
 106         engine = &io_engine_epoll;
 107 #endif
 108
 109 #if WITH_IOSET_WIN32
 110     if (!engine && io_engine_win32.init())
 111         engine = &io_engine_win32;
 112 #endif
 113
 114     if (engine) {
 115         /* we found one that works */
 116     } else if (io_engine_select.init())
 117         engine = &io_engine_select;
 118     else
 119         log_module(MAIN_LOG, LOG_FATAL, "No usable I/O engine found.");
 120     log_module(MAIN_LOG, LOG_DEBUG, "Using %s I/O engine.", engine->name);
 121 }
 122
 123 void
 124 ioset_cleanup(void) {
 125     engine->cleanup();
 126 }
 127
 128 struct io_fd *
 129 ioset_add(int fd) {
 130     struct io_fd *res;
 131     int flags;
 132
 133     if (fd < 0) {
 134         log_module(MAIN_LOG, LOG_ERROR, "Somebody called ioset_add(%d) on a negative fd!", fd);
 135         return 0;
 136     }
 137     if (!engine)
 138         ioset_init();
 139     res = calloc(1, sizeof(*res));
 140     if (!res)
 141         return 0;
 142     res->fd = fd;
 143     ioq_init(&res->send, 1024);
 144     ioq_init(&res->recv, 1024);
 145 #if defined(F_GETFL)
 146     flags = fcntl(fd, F_GETFL);
 147     fcntl(fd, F_SETFL, flags|O_NONBLOCK);
 148     flags = fcntl(fd, F_GETFD);
 149     fcntl(fd, F_SETFD, flags|FD_CLOEXEC);
 150 #else
 151     /* I hope you're using the Win32 backend or something else that
 152      * automatically marks the file descriptor non-blocking...
 153      */
 154     (void)flags;
 155 #endif
 156     engine->add(res);
 157     return res;
 158 }
 159
 160 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))
 161 {
 162     struct io_fd *io_fd;
 163     unsigned int opt;
 164     int res;
 165     int fd;
 166
 167     fd = socket(local ? local->sa_family : PF_INET, SOCK_STREAM, 0);
 168     if (fd < 0) {
 169         log_module(MAIN_LOG, LOG_ERROR, "Unable to create listening socket: %s", strerror(errno));
 170         return NULL;
 171     }
 172
 173     if (local && sa_size) {
 174         res = bind(fd, local, sa_size);
 175         if (res < 0) {
 176             log_module(MAIN_LOG, LOG_ERROR, "Unable to bind listening socket %d: %s", fd, strerror(errno));
 177             close(fd);
 178             return NULL;
 179         }
 180
 181         opt = 1;
 182         res = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
 183         if (res < 0) {
 184             log_module(MAIN_LOG, LOG_WARNING, "Unable to mark listener address as re-usable: %s", strerror(errno));
 185         }
 186     }
 187
 188     res = listen(fd, 1);
 189     if (res < 0) {
 190         log_module(MAIN_LOG, LOG_ERROR, "Unable to listen on socket %d: %s", fd, strerror(errno));
 191         close(fd);
 192         return NULL;
 193     }
 194
 195     io_fd = ioset_add(fd);
 196     if (!io_fd) {
 197         close(fd);
 198         return NULL;
 199     }
 200     io_fd->state = IO_LISTENING;
 201     io_fd->data = data;
 202     io_fd->accept_cb = accept_cb;
 203     engine->update(io_fd);
 204     return io_fd;
 205 }
 206
 207 struct io_fd *
 208 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)) {
 209     struct addrinfo hints;
 210     struct addrinfo *ai;
 211     struct io_fd *io_fd;
 212     struct io_fd *old_active;
 213     int res;
 214     int fd;
 215     char portnum[10];
 216
 217     memset(&hints, 0, sizeof(hints));
 218     hints.ai_family = local ? local->sa_family : 0;
 219     hints.ai_socktype = SOCK_STREAM;
 220     snprintf(portnum, sizeof(portnum), "%u", port);
 221     if (getaddrinfo(peer, portnum, &hints, &ai)) {
 222         log_module(MAIN_LOG, LOG_ERROR, "getaddrinfo(%s, %s) failed.", peer, portnum);
 223         return NULL;
 224     }
 225
 226     if (local) {
 227         if ((fd = socket(local->sa_family, SOCK_STREAM, 0)) < 0) {
 228             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s)", peer, errno, strerror(errno));
 229             freeaddrinfo(ai);
 230             return NULL;
 231         }
 232         if (bind(fd, local, sa_size) < 0) {
 233             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));
 234         }
 235     } else {
 236         if ((fd = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
 237             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s).", peer, errno, strerror(errno));
 238             freeaddrinfo(ai);
 239             return NULL;
 240         }
 241     }
 242
 243     if (blocking) {
 244         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
 245         io_fd = ioset_add(fd);
 246     } else {
 247         io_fd = ioset_add(fd);
 248         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
 249     }
 250     freeaddrinfo(ai);
 251     if (!io_fd) {
 252         close(fd);
 253         return NULL;
 254     }
 255     io_fd->state = IO_CONNECTING;
 256     io_fd->data = data;
 257     io_fd->connect_cb = connect_cb;
 258     if (res < 0) {
 259         switch (errno) {
 260         case EINPROGRESS: /* only if !blocking */
 261             engine->update(io_fd);
 262             return io_fd;
 263         default:
 264             log_module(MAIN_LOG, LOG_ERROR, "connect(%s:%d) (fd %d) returned errno %d (%s).", peer, port, io_fd->fd, errno, strerror(errno));
 265             /* then fall through */
 266         case EHOSTUNREACH:
 267         case ECONNREFUSED:
 268             ioset_close(io_fd, 1);
 269             return NULL;
 270         }
 271     }
 272     io_fd->state = IO_CONNECTED;
 273     old_active = active_fd;
 274     if (connect_cb)
 275         connect_cb(io_fd, ((res < 0) ? errno : 0));
 276     if (active_fd)
 277         engine->update(io_fd);
 278     if (old_active != io_fd)
 279         active_fd = old_active;
 280     return io_fd;
 281 }
 282
 283 void ioset_update(struct io_fd *fd) {
 284     engine->update(fd);
 285 }
 286
 287 static void
 288 ioset_try_write(struct io_fd *fd) {
 289     int res;
 290     unsigned int req;
 291
 292     req = ioq_get_avail(&fd->send);
 293     res = write(fd->fd, fd->send.buf+fd->send.get, req);
 294     if (res < 0) {
 295         switch (errno) {
 296         case EAGAIN:
 297             break;
 298         default:
 299             log_module(MAIN_LOG, LOG_ERROR, "write() on fd %d error %d: %s", fd->fd, errno, strerror(errno));
 300         }
 301     } else {
 302         fd->send.get += res;
 303         if (fd->send.get == fd->send.size)
 304             fd->send.get = 0;
 305         engine->update(fd);
 306     }
 307 }
 308
 309 void
 310 ioset_close(struct io_fd *fdp, int os_close) {
 311     if (!fdp)
 312         return;
 313     if (active_fd == fdp)
 314         active_fd = NULL;
 315     if (fdp->destroy_cb)
 316         fdp->destroy_cb(fdp);
 317 #if defined(HAVE_WSAEVENTSELECT)
 318     /* This is one huge kludge.  Sorry! */
 319     if (fdp->send.get != fdp->send.put && (os_close & 2)) {
 320         engine->remove(fdp, 0);
 321         ioset_try_write(fdp);
 322         /* it may need to send the beginning of the buffer now.. */
 323         if (fdp->send.get != fdp->send.put)
 324             ioset_try_write(fdp);
 325     }
 326     free(fdp->send.buf);
 327     free(fdp->recv.buf);
 328     if (os_close & 1)
 329         closesocket(fdp->fd);
 330 #else
 331     if (fdp->send.get != fdp->send.put && (os_close & 2)) {
 332         int flags;
 333
 334         flags = fcntl(fdp->fd, F_GETFL);
 335         fcntl(fdp->fd, F_SETFL, flags&~O_NONBLOCK);
 336         ioset_try_write(fdp);
 337         /* it may need to send the beginning of the buffer now.. */
 338         if (fdp->send.get != fdp->send.put)
 339             ioset_try_write(fdp);
 340     }
 341     free(fdp->send.buf);
 342     free(fdp->recv.buf);
 343     if (os_close & 1)
 344         close(fdp->fd);
 345     engine->remove(fdp, os_close & 1);
 346 #endif
 347     free(fdp);
 348 }
 349
 350 static void
 351 ioset_accept(struct io_fd *listener)
 352 {
 353     struct io_fd *old_active;
 354     struct io_fd *new_fd;
 355     int fd;
 356
 357     fd = accept(listener->fd, NULL, 0);
 358     if (fd < 0) {
 359         log_module(MAIN_LOG, LOG_ERROR, "Unable to accept new connection on listener %d: %s", listener->fd, strerror(errno));
 360         return;
 361     }
 362
 363     new_fd = ioset_add(fd);
 364     new_fd->state = IO_CONNECTED;
 365     old_active = active_fd;
 366     active_fd = new_fd;
 367     listener->accept_cb(listener, new_fd);
 368     assert(active_fd == NULL || active_fd == new_fd);
 369     if (active_fd == new_fd) {
 370         if (new_fd->send.get != new_fd->send.put)
 371             ioset_try_write(new_fd);
 372         else
 373             engine->update(new_fd);
 374     }
 375     active_fd = old_active;
 376 }
 377
 378 static int
 379 ioset_find_line_length(struct io_fd *fd) {
 380     unsigned int pos, max, len;
 381     len = 0;
 382     max = (fd->recv.put < fd->recv.get) ? fd->recv.size : fd->recv.put;
 383     for (pos = fd->recv.get; pos < max; ++pos, ++len)
 384         if (IS_EOL(fd->recv.buf[pos]))
 385             return fd->line_len = len + 1;
 386     if (fd->recv.put < fd->recv.get)
 387         for (pos = 0; pos < fd->recv.put; ++pos, ++len)
 388             if (IS_EOL(fd->recv.buf[pos]))
 389                 return fd->line_len = len + 1;
 390     return fd->line_len = 0;
 391 }
 392
 393 static void
 394 ioset_buffered_read(struct io_fd *fd) {
 395     int put_avail, nbr, fdnum;
 396
 397     if (!(put_avail = ioq_put_avail(&fd->recv)))
 398         put_avail = ioq_grow(&fd->recv);
 399     nbr = read(fd->fd, fd->recv.buf + fd->recv.put, put_avail);
 400     if (nbr < 0) {
 401         switch (errno) {
 402         case EAGAIN:
 403             break;
 404         default:
 405             log_module(MAIN_LOG, LOG_ERROR, "Unexpected read() error %d on fd %d: %s", errno, fd->fd, strerror(errno));
 406             /* Just flag it as EOF and call readable_cb() to notify the fd's owner. */
 407             fd->state = IO_CLOSED;
 408             fd->readable_cb(fd);
 409             if (active_fd == fd)
 410                 engine->update(fd);
 411         }
 412     } else if (nbr == 0) {
 413         fd->state = IO_CLOSED;
 414         fd->readable_cb(fd);
 415         if (active_fd == fd)
 416             engine->update(fd);
 417     } else {
 418         if (fd->line_len == 0) {
 419             unsigned int pos;
 420             for (pos = fd->recv.put; pos < fd->recv.put + nbr; ++pos) {
 421                 if (IS_EOL(fd->recv.buf[pos])) {
 422                     if (fd->recv.put < fd->recv.get)
 423                         fd->line_len = fd->recv.size + pos + 1 - fd->recv.get;
 424                     else
 425                         fd->line_len = pos + 1 - fd->recv.get;
 426                     break;
 427                 }
 428             }
 429         }
 430         fd->recv.put += nbr;
 431         if (fd->recv.put == fd->recv.size)
 432             fd->recv.put = 0;
 433         fdnum = fd->fd;
 434         while (fd->line_len > 0) {
 435             struct io_fd *old_active;
 436             int died = 0;
 437
 438             old_active = active_fd;
 439             active_fd = fd;
 440             fd->readable_cb(fd);
 441             if (active_fd)
 442                 ioset_find_line_length(fd);
 443             else
 444                 died = 1;
 445             if (old_active != fd)
 446                 active_fd = old_active;
 447             if (died)
 448                 break;
 449         }
 450     }
 451 }
 452
 453 int
 454 ioset_line_read(struct io_fd *fd, char *dest, int max) {
 455     int line_len;
 456     int avail;
 457     int done;
 458
 459     line_len = fd->line_len;
 460     if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) ||  (line_len < 0)))
 461         return 0;
 462     if (line_len < 0)
 463         return -1;
 464     if (line_len < max)
 465         max = line_len;
 466     avail = ioq_get_avail(&fd->recv);
 467     if (max > avail) {
 468         memcpy(dest, fd->recv.buf + fd->recv.get, avail);
 469         assert(fd->recv.get + avail == fd->recv.size);
 470         fd->recv.get = 0;
 471         done = avail;
 472     } else {
 473         done = 0;
 474     }
 475     memcpy(dest + done, fd->recv.buf + fd->recv.get, max - done);
 476     fd->recv.get += max - done;
 477     if (fd->recv.get == fd->recv.size)
 478         fd->recv.get = 0;
 479     dest[max - 1] = 0;
 480     ioset_find_line_length(fd);
 481     return line_len;
 482 }
 483
 484 void
 485 ioset_events(struct io_fd *fd, int readable, int writable)
 486 {
 487     if (!fd || (!readable && !writable))
 488         return;
 489     active_fd = fd;
 490     switch (fd->state) {
 491     case IO_CLOSED:
 492         break;
 493     case IO_LISTENING:
 494         if (active_fd && readable)
 495             ioset_accept(fd);
 496         break;
 497     case IO_CONNECTING:
 498         assert(active_fd == NULL || active_fd == fd);
 499         if (active_fd && readable) {
 500             socklen_t arglen;
 501             int rc;
 502             arglen = sizeof(rc);
 503             if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
 504                 rc = errno;
 505             fd->state = IO_CLOSED;
 506             if (fd->connect_cb)
 507                 fd->connect_cb(fd, rc);
 508         } else if (active_fd && writable) {
 509             fd->state = IO_CONNECTED;
 510             if (fd->connect_cb)
 511                 fd->connect_cb(fd, 0);
 512         }
 513         if (active_fd != fd)
 514             break;
 515         engine->update(fd);
 516         /* and fall through */
 517     case IO_CONNECTED:
 518         assert(active_fd == NULL || active_fd == fd);
 519         if (active_fd && readable) {
 520             if (fd->line_reads)
 521                 ioset_buffered_read(fd);
 522             else
 523                 fd->readable_cb(fd);
 524         }
 525
 526         assert(active_fd == NULL || active_fd == fd);
 527         if (active_fd && writable)
 528             ioset_try_write(fd);
 529         break;
 530     }
 531 }
 532
 533 void
 534 ioset_run(void) {
 535     extern struct io_fd *socket_io_fd;
 536     struct timeval timeout;
 537     time_t wakey;
 538
 539     while (!quit_services) {
 540         while (!socket_io_fd)
 541             uplink_connect();
 542
 543         /* How long to sleep? (fill in select_timeout) */
 544         wakey = timeq_next();
 545         if ((wakey - now) < 0)
 546             timeout.tv_sec = 0;
 547         else
 548             timeout.tv_sec = wakey - now;
 549         timeout.tv_usec = 0;
 550
 551         if (engine->loop(&timeout))
 552             continue;
 553
 554         /* Call any timeq events we need to call. */
 555         timeq_run();
 556         if (do_write_dbs) {
 557             saxdb_write_all();
 558             do_write_dbs = 0;
 559         }
 560         if (do_reopen) {
 561             extern char *services_config;
 562             conf_read(services_config);
 563             do_reopen = 0;
 564         }
 565     }
 566 }
 567
 568 void
 569 ioset_write(struct io_fd *fd, const char *buf, unsigned int nbw) {
 570     unsigned int avail;
 571     while (ioq_used(&fd->send) + nbw >= fd->send.size)
 572         ioq_grow(&fd->send);
 573     avail = ioq_put_avail(&fd->send);
 574     if (nbw > avail) {
 575         memcpy(fd->send.buf + fd->send.put, buf, avail);
 576         buf += avail;
 577         nbw -= avail;
 578         fd->send.put = 0;
 579     }
 580     memcpy(fd->send.buf + fd->send.put, buf, nbw);
 581     fd->send.put += nbw;
 582     if (fd->send.put == fd->send.size)
 583         fd->send.put = 0;
 584     engine->update(fd);
 585 }
 586
 587 int
 588 ioset_printf(struct io_fd *fd, const char *fmt, ...) {
 589     char tmpbuf[MAXLEN];
 590     va_list ap;
 591     int res;
 592
 593     va_start(ap, fmt);
 594     res = vsnprintf(tmpbuf, sizeof(tmpbuf), fmt, ap);
 595     va_end(ap);
 596     if (res > 0 && (size_t)res <= sizeof(tmpbuf))
 597         ioset_write(fd, tmpbuf, res);
 598     return res;
 599 }
 600
 601 void
 602 ioset_set_time(unsigned long new_now) {
 603     clock_skew = new_now - time(NULL);
 604     now = new_now;
 605 }