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 3 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 #ifdef WITH_IOSET_WIN32
  35
  36 # undef errno
  37 # define errno WSAGetLastError()
  38 # undef EINPROGRESS
  39 # define EINPROGRESS WSAEINPROGRESS
  40 # undef EHOSTUNREACH
  41 # define EHOSTUNREACH WSAEHOSTUNREACH
  42 # undef ECONNREFUSED
  43 # define ECONNREFUSED WSAECONNREFUSED
  44 # undef EAGAIN
  45 # define EAGAIN WSAEWOULDBLOCK
  46 # define strerror wsa_strerror
  47
  48 static const char *
  49 wsa_strerror(int wsa_err)
  50 {
  51     switch (wsa_err)
  52     {
  53     case WSAEINTR: return "Operation interrupted";
  54     case WSAEBADF: return "Bad file descriptor";
  55     case WSAEACCES: return "Permission denied";
  56     case WSAEFAULT: return "Invalid address";
  57     case WSAEINVAL: return "Invalid parameter";
  58     case WSAEMFILE: return "Too many open files";
  59     case WSAEWOULDBLOCK: return "Try again later";
  60     case WSAEINPROGRESS: return "Operation in progress";
  61     case WSAEALREADY: return "Operation already in progress";
  62     case WSAENOTSOCK: return "Not a socket";
  63     case WSAEDESTADDRREQ: return "Destination address required";
  64     case WSAEMSGSIZE: return "Invalid message size";
  65     case WSAEPROTOTYPE: return "Invalid protocol type for socket";
  66     case WSAENOPROTOOPT: return "Invalid protocol option";
  67     case WSAEPROTONOSUPPORT: return "Protocol not supported";
  68     case WSAEOPNOTSUPP: return "Operation not supported";
  69     case WSAEADDRINUSE: return "Address already in use";
  70     case WSAEADDRNOTAVAIL: return "Address not available";
  71     case WSAENETDOWN: return "Network down";
  72     case WSAENETUNREACH: return "Network unreachable";
  73     case WSAENETRESET: return "Network reset";
  74     case WSAECONNABORTED: return "Connection aborted";
  75     case WSAECONNRESET: return "Connection reset by peer";
  76     case WSAECONNREFUSED: return "Connection refused";
  77     }
  78     return "unknown error";
  79 }
  80
  81 #endif /* WITH_IOSET_WIN32 */
  82
  83 #define IS_EOL(CH) ((CH) == '\n')
  84
  85 extern int uplink_connect(void);
  86 int clock_skew;
  87 int do_write_dbs;
  88 int do_reopen;
  89 static struct io_engine *engine;
  90 static struct io_fd *active_fd;
  91
  92 static void
  93 ioq_init(struct ioq *ioq, int size) {
  94     ioq->buf = malloc(size);
  95     ioq->get = ioq->put = 0;
  96     ioq->size = size;
  97 }
  98
  99 static unsigned int
 100 ioq_put_avail(const struct ioq *ioq) {
 101     /* Subtract 1 from ioq->get to be sure we don't fill the buffer
 102      * and make it look empty even when there's data in it. */
 103     if (ioq->put < ioq->get)
 104         return ioq->get - ioq->put - 1;
 105     else if (ioq->get == 0)
 106         return ioq->size - ioq->put - 1;
 107     else
 108         return ioq->size - ioq->put;
 109 }
 110
 111 static unsigned int
 112 ioq_get_avail(const struct ioq *ioq) {
 113     return ((ioq->put < ioq->get) ? ioq->size : ioq->put) - ioq->get;
 114 }
 115
 116 static unsigned int
 117 ioq_used(const struct ioq *ioq) {
 118     return ((ioq->put < ioq->get) ? ioq->size : 0) + ioq->put - ioq->get;
 119 }
 120
 121 static unsigned int
 122 ioq_grow(struct ioq *ioq) {
 123     int new_size = ioq->size << 1;
 124     char *new_buf = malloc(new_size);
 125     int get_avail = ioq_get_avail(ioq);
 126     memcpy(new_buf, ioq->buf + ioq->get, get_avail);
 127     if (ioq->put < ioq->get)
 128         memcpy(new_buf + get_avail, ioq->buf, ioq->put);
 129     free(ioq->buf);
 130     ioq->put = ioq_used(ioq);
 131     ioq->get = 0;
 132     ioq->buf = new_buf;
 133     ioq->size = new_size;
 134     return new_size - ioq->put;
 135 }
 136
 137 extern struct io_engine io_engine_kevent;
 138 extern struct io_engine io_engine_epoll;
 139 extern struct io_engine io_engine_win32;
 140 extern struct io_engine io_engine_select;
 141
 142 void
 143 ioset_init(void)
 144 {
 145     if (engine) /* someone beat us to it */
 146         return;
 147
 148 #if WITH_IOSET_KEVENT
 149     if (!engine && io_engine_kevent.init())
 150         engine = &io_engine_kevent;
 151 #endif
 152
 153 #if WITH_IOSET_EPOLL
 154     if (!engine && io_engine_epoll.init())
 155         engine = &io_engine_epoll;
 156 #endif
 157
 158 #if WITH_IOSET_WIN32
 159     if (!engine && io_engine_win32.init())
 160         engine = &io_engine_win32;
 161 #endif
 162
 163     if (engine) {
 164         /* we found one that works */
 165     } else if (io_engine_select.init())
 166         engine = &io_engine_select;
 167     else
 168         log_module(MAIN_LOG, LOG_FATAL, "No usable I/O engine found.");
 169     log_module(MAIN_LOG, LOG_DEBUG, "Using %s I/O engine.", engine->name);
 170 }
 171
 172 void
 173 ioset_cleanup(void) {
 174     engine->cleanup();
 175 }
 176
 177 struct io_fd *
 178 ioset_add(int fd) {
 179     struct io_fd *res;
 180     int flags;
 181
 182     if (fd < 0) {
 183         log_module(MAIN_LOG, LOG_ERROR, "Somebody called ioset_add(%d) on a negative fd!", fd);
 184         return 0;
 185     }
 186     if (!engine)
 187         ioset_init();
 188     res = calloc(1, sizeof(*res));
 189     if (!res)
 190         return 0;
 191     res->fd = fd;
 192     ioq_init(&res->send, 1024);
 193     ioq_init(&res->recv, 1024);
 194 #if defined(F_GETFL)
 195     flags = fcntl(fd, F_GETFL);
 196     fcntl(fd, F_SETFL, flags|O_NONBLOCK);
 197     flags = fcntl(fd, F_GETFD);
 198     fcntl(fd, F_SETFD, flags|FD_CLOEXEC);
 199 #else
 200     /* I hope you're using the Win32 backend or something else that
 201      * automatically marks the file descriptor non-blocking...
 202      */
 203     (void)flags;
 204 #endif
 205     engine->add(res);
 206     return res;
 207 }
 208
 209 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))
 210 {
 211     struct io_fd *io_fd;
 212     unsigned int opt;
 213     int res;
 214     int fd;
 215
 216     fd = socket(local ? local->sa_family : PF_INET, SOCK_STREAM, 0);
 217     if (fd < 0) {
 218         log_module(MAIN_LOG, LOG_ERROR, "Unable to create listening socket: %s", strerror(errno));
 219         return NULL;
 220     }
 221
 222     if (local && sa_size) {
 223         opt = 1;
 224         res = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
 225         if (res < 0) {
 226             log_module(MAIN_LOG, LOG_WARNING, "Unable to mark listener address as re-usable: %s", strerror(errno));
 227         }
 228         res = bind(fd, local, sa_size);
 229         if (res < 0) {
 230             log_module(MAIN_LOG, LOG_ERROR, "Unable to bind listening socket %d: %s", fd, strerror(errno));
 231             close(fd);
 232             return NULL;
 233         }
 234     }
 235
 236     res = listen(fd, 1);
 237     if (res < 0) {
 238         log_module(MAIN_LOG, LOG_ERROR, "Unable to listen on socket %d: %s", fd, strerror(errno));
 239         close(fd);
 240         return NULL;
 241     }
 242
 243     io_fd = ioset_add(fd);
 244     if (!io_fd) {
 245         close(fd);
 246         return NULL;
 247     }
 248     io_fd->state = IO_LISTENING;
 249     io_fd->data = data;
 250     io_fd->accept_cb = accept_cb;
 251     engine->update(io_fd);
 252     return io_fd;
 253 }
 254
 255 struct io_fd *
 256 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)) {
 257     struct addrinfo hints;
 258     struct addrinfo *ai;
 259     struct io_fd *io_fd;
 260     struct io_fd *old_active;
 261     int res;
 262     int fd;
 263     char portnum[10];
 264
 265     memset(&hints, 0, sizeof(hints));
 266     hints.ai_family = local ? local->sa_family : 0;
 267     hints.ai_socktype = SOCK_STREAM;
 268     snprintf(portnum, sizeof(portnum), "%u", port);
 269     res = getaddrinfo(peer, portnum, &hints, &ai);
 270     if (res != 0) {
 271         log_module(MAIN_LOG, LOG_ERROR, "getaddrinfo(%s, %s) failed: %s.", peer, portnum, gai_strerror(res));
 272         return NULL;
 273     }
 274
 275     if (local) {
 276         if ((fd = socket(local->sa_family, SOCK_STREAM, 0)) < 0) {
 277             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s)", peer, errno, strerror(errno));
 278             freeaddrinfo(ai);
 279             return NULL;
 280         }
 281         if (bind(fd, local, sa_size) < 0) {
 282             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));
 283         }
 284     } else {
 285         if ((fd = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
 286             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s).", peer, errno, strerror(errno));
 287             freeaddrinfo(ai);
 288             return NULL;
 289         }
 290     }
 291
 292     if (blocking) {
 293         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
 294         io_fd = ioset_add(fd);
 295     } else {
 296         io_fd = ioset_add(fd);
 297         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
 298     }
 299     freeaddrinfo(ai);
 300     if (!io_fd) {
 301         close(fd);
 302         return NULL;
 303     }
 304     io_fd->state = IO_CONNECTING;
 305     io_fd->data = data;
 306     io_fd->connect_cb = connect_cb;
 307     if (res < 0) {
 308         switch (errno) {
 309         case EINPROGRESS: /* only if !blocking */
 310             engine->update(io_fd);
 311             return io_fd;
 312         default:
 313             log_module(MAIN_LOG, LOG_ERROR, "connect(%s:%d) (fd %d) returned errno %d (%s).", peer, port, io_fd->fd, errno, strerror(errno));
 314             /* then fall through */
 315         case EHOSTUNREACH:
 316         case ECONNREFUSED:
 317             ioset_close(io_fd, 1);
 318             return NULL;
 319         }
 320     }
 321     io_fd->state = IO_CONNECTED;
 322     old_active = active_fd;
 323     if (connect_cb)
 324         connect_cb(io_fd, ((res < 0) ? errno : 0));
 325     if (active_fd)
 326         engine->update(io_fd);
 327     if (old_active != io_fd)
 328         active_fd = old_active;
 329     return io_fd;
 330 }
 331
 332 void ioset_update(struct io_fd *fd) {
 333     engine->update(fd);
 334 }
 335
 336 static void
 337 ioset_try_write(struct io_fd *fd) {
 338     int res;
 339     unsigned int req;
 340
 341     req = ioq_get_avail(&fd->send);
 342     res = send(fd->fd, fd->send.buf+fd->send.get, req, 0);
 343     if (res < 0) {
 344         if (errno != EAGAIN) {
 345             log_module(MAIN_LOG, LOG_ERROR, "send() on fd %d error %d: %s", fd->fd, errno, strerror(errno));
 346         }
 347     } else {
 348         fd->send.get += res;
 349         if (fd->send.get == fd->send.size)
 350             fd->send.get = 0;
 351         engine->update(fd);
 352     }
 353 }
 354
 355 void
 356 ioset_close(struct io_fd *fdp, int os_close) {
 357     if (!fdp)
 358         return;
 359     if (active_fd == fdp)
 360         active_fd = NULL;
 361     if (fdp->destroy_cb)
 362         fdp->destroy_cb(fdp);
 363 #if defined(HAVE_WSAEVENTSELECT)
 364     /* This is one huge kludge.  Sorry! */
 365     if (fdp->send.get != fdp->send.put && (os_close & 2)) {
 366         engine->remove(fdp, 0);
 367         ioset_try_write(fdp);
 368         /* it may need to send the beginning of the buffer now.. */
 369         if (fdp->send.get != fdp->send.put)
 370             ioset_try_write(fdp);
 371     }
 372     free(fdp->send.buf);
 373     free(fdp->recv.buf);
 374     if (os_close & 1)
 375         closesocket(fdp->fd);
 376 #else
 377     if (fdp->send.get != fdp->send.put && (os_close & 2)) {
 378         int flags;
 379
 380         flags = fcntl(fdp->fd, F_GETFL);
 381         fcntl(fdp->fd, F_SETFL, flags&~O_NONBLOCK);
 382         ioset_try_write(fdp);
 383         /* it may need to send the beginning of the buffer now.. */
 384         if (fdp->send.get != fdp->send.put)
 385             ioset_try_write(fdp);
 386     }
 387     free(fdp->send.buf);
 388     free(fdp->recv.buf);
 389     if (os_close & 1)
 390         close(fdp->fd);
 391     engine->remove(fdp, os_close & 1);
 392 #endif
 393     free(fdp);
 394 }
 395
 396 static void
 397 ioset_accept(struct io_fd *listener)
 398 {
 399     struct io_fd *old_active;
 400     struct io_fd *new_fd;
 401     int fd;
 402
 403     fd = accept(listener->fd, NULL, 0);
 404     if (fd < 0) {
 405         log_module(MAIN_LOG, LOG_ERROR, "Unable to accept new connection on listener %d: %s", listener->fd, strerror(errno));
 406         return;
 407     }
 408
 409     new_fd = ioset_add(fd);
 410     new_fd->state = IO_CONNECTED;
 411     old_active = active_fd;
 412     active_fd = new_fd;
 413     listener->accept_cb(listener, new_fd);
 414     assert(active_fd == NULL || active_fd == new_fd);
 415     if (active_fd == new_fd) {
 416         if (new_fd->send.get != new_fd->send.put)
 417             ioset_try_write(new_fd);
 418         else
 419             engine->update(new_fd);
 420     }
 421     active_fd = old_active;
 422 }
 423
 424 static int
 425 ioset_find_line_length(struct io_fd *fd) {
 426     unsigned int pos, max, len;
 427     len = 0;
 428     max = (fd->recv.put < fd->recv.get) ? fd->recv.size : fd->recv.put;
 429     for (pos = fd->recv.get; pos < max; ++pos, ++len)
 430         if (IS_EOL(fd->recv.buf[pos]))
 431             return fd->line_len = len + 1;
 432     if (fd->recv.put < fd->recv.get)
 433         for (pos = 0; pos < fd->recv.put; ++pos, ++len)
 434             if (IS_EOL(fd->recv.buf[pos]))
 435                 return fd->line_len = len + 1;
 436     return fd->line_len = 0;
 437 }
 438
 439 static void
 440 ioset_buffered_read(struct io_fd *fd) {
 441     int put_avail, nbr;
 442
 443     if (!(put_avail = ioq_put_avail(&fd->recv)))
 444         put_avail = ioq_grow(&fd->recv);
 445     nbr = recv(fd->fd, fd->recv.buf + fd->recv.put, put_avail, 0);
 446     if (nbr < 0) {
 447         if (errno != EAGAIN) {
 448             log_module(MAIN_LOG, LOG_ERROR, "Unexpected recv() error %d on fd %d: %s", errno, fd->fd, strerror(errno));
 449             /* Just flag it as EOF and call readable_cb() to notify the fd's owner. */
 450             fd->state = IO_CLOSED;
 451             fd->readable_cb(fd);
 452             if (active_fd == fd)
 453                 engine->update(fd);
 454         }
 455     } else if (nbr == 0) {
 456         fd->state = IO_CLOSED;
 457         fd->readable_cb(fd);
 458         if (active_fd == fd)
 459             engine->update(fd);
 460     } else {
 461         if (fd->line_len == 0) {
 462             unsigned int pos;
 463             for (pos = fd->recv.put; pos < fd->recv.put + nbr; ++pos) {
 464                 if (IS_EOL(fd->recv.buf[pos])) {
 465                     if (fd->recv.put < fd->recv.get)
 466                         fd->line_len = fd->recv.size + pos + 1 - fd->recv.get;
 467                     else
 468                         fd->line_len = pos + 1 - fd->recv.get;
 469                     break;
 470                 }
 471             }
 472         }
 473         fd->recv.put += nbr;
 474         if (fd->recv.put == fd->recv.size)
 475             fd->recv.put = 0;
 476         while (fd->line_len > 0) {
 477             struct io_fd *old_active;
 478             int died = 0;
 479
 480             old_active = active_fd;
 481             active_fd = fd;
 482             fd->readable_cb(fd);
 483             if (active_fd)
 484                 ioset_find_line_length(fd);
 485             else
 486                 died = 1;
 487             if (old_active != fd)
 488                 active_fd = old_active;
 489             if (died)
 490                 break;
 491         }
 492     }
 493 }
 494
 495 int
 496 ioset_line_read(struct io_fd *fd, char *dest, int max) {
 497     int line_len;
 498     int avail;
 499     int done;
 500
 501     line_len = fd->line_len;
 502     if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) ||  (line_len < 0)))
 503         return 0;
 504     if (line_len < 0)
 505         return -1;
 506     if (line_len < max)
 507         max = line_len;
 508     avail = ioq_get_avail(&fd->recv);
 509     if (max > avail) {
 510         memcpy(dest, fd->recv.buf + fd->recv.get, avail);
 511         assert(fd->recv.get + avail == fd->recv.size);
 512         fd->recv.get = 0;
 513         done = avail;
 514     } else {
 515         done = 0;
 516     }
 517     memcpy(dest + done, fd->recv.buf + fd->recv.get, max - done);
 518     fd->recv.get += max - done;
 519     if (fd->recv.get == fd->recv.size)
 520         fd->recv.get = 0;
 521     dest[max - 1] = 0;
 522     ioset_find_line_length(fd);
 523     return line_len;
 524 }
 525
 526 void
 527 ioset_events(struct io_fd *fd, int readable, int writable)
 528 {
 529     if (!fd || (!readable && !writable))
 530         return;
 531     active_fd = fd;
 532     switch (fd->state) {
 533     case IO_CLOSED:
 534         break;
 535     case IO_LISTENING:
 536         if (active_fd && readable)
 537             ioset_accept(fd);
 538         break;
 539     case IO_CONNECTING:
 540         assert(active_fd == NULL || active_fd == fd);
 541         if (active_fd && readable) {
 542             socklen_t arglen;
 543             int rc;
 544             arglen = sizeof(rc);
 545             if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
 546                 rc = errno;
 547             fd->state = IO_CLOSED;
 548             if (fd->connect_cb)
 549                 fd->connect_cb(fd, rc);
 550         } else if (active_fd && writable) {
 551             fd->state = IO_CONNECTED;
 552             if (fd->connect_cb)
 553                 fd->connect_cb(fd, 0);
 554         }
 555         if (active_fd != fd)
 556             break;
 557         engine->update(fd);
 558         /* and fall through */
 559     case IO_CONNECTED:
 560         assert(active_fd == NULL || active_fd == fd);
 561         if (active_fd && readable) {
 562             if (fd->line_reads)
 563                 ioset_buffered_read(fd);
 564             else
 565                 fd->readable_cb(fd);
 566         }
 567
 568         assert(active_fd == NULL || active_fd == fd);
 569         if (active_fd && writable)
 570             ioset_try_write(fd);
 571         break;
 572     }
 573 }
 574
 575 void
 576 ioset_run(void) {
 577     extern struct io_fd *socket_io_fd;
 578     struct timeval timeout;
 579     time_t wakey;
 580
 581     while (!quit_services) {
 582         while (!socket_io_fd)
 583             uplink_connect();
 584
 585         /* How long to sleep? (fill in select_timeout) */
 586         wakey = timeq_next();
 587         if (wakey < now)
 588             timeout.tv_sec = 0;
 589         else
 590             timeout.tv_sec = wakey - now;
 591         timeout.tv_usec = 0;
 592
 593         if (engine->loop(&timeout))
 594             continue;
 595
 596         /* Call any timeq events we need to call. */
 597         timeq_run();
 598         if (do_write_dbs) {
 599             saxdb_write_all(NULL);
 600             do_write_dbs = 0;
 601         }
 602         if (do_reopen) {
 603             extern char *services_config;
 604             conf_read(services_config);
 605             do_reopen = 0;
 606         }
 607     }
 608 }
 609
 610 void
 611 ioset_write(struct io_fd *fd, const char *buf, unsigned int nbw) {
 612     unsigned int avail;
 613     while (ioq_used(&fd->send) + nbw >= fd->send.size)
 614         ioq_grow(&fd->send);
 615     avail = ioq_put_avail(&fd->send);
 616     if (nbw > avail) {
 617         memcpy(fd->send.buf + fd->send.put, buf, avail);
 618         buf += avail;
 619         nbw -= avail;
 620         fd->send.put = 0;
 621     }
 622     memcpy(fd->send.buf + fd->send.put, buf, nbw);
 623     fd->send.put += nbw;
 624     if (fd->send.put == fd->send.size)
 625         fd->send.put = 0;
 626     engine->update(fd);
 627 }
 628
 629 int
 630 ioset_printf(struct io_fd *fd, const char *fmt, ...) {
 631     char tmpbuf[MAXLEN];
 632     va_list ap;
 633     int res;
 634
 635     va_start(ap, fmt);
 636     res = vsnprintf(tmpbuf, sizeof(tmpbuf), fmt, ap);
 637     va_end(ap);
 638     if (res > 0 && (size_t)res <= sizeof(tmpbuf))
 639         ioset_write(fd, tmpbuf, res);
 640     return res;
 641 }
 642
 643 void
 644 ioset_set_time(unsigned long new_now) {
 645     clock_skew = new_now - time(NULL);
 646     now = new_now;
 647 }