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 #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         res = bind(fd, local, sa_size);
 224         if (res < 0) {
 225             log_module(MAIN_LOG, LOG_ERROR, "Unable to bind listening socket %d: %s", fd, strerror(errno));
 226             close(fd);
 227             return NULL;
 228         }
 229
 230         opt = 1;
 231         res = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
 232         if (res < 0) {
 233             log_module(MAIN_LOG, LOG_WARNING, "Unable to mark listener address as re-usable: %s", strerror(errno));
 234         }
 235     }
 236
 237     res = listen(fd, 1);
 238     if (res < 0) {
 239         log_module(MAIN_LOG, LOG_ERROR, "Unable to listen on socket %d: %s", fd, strerror(errno));
 240         close(fd);
 241         return NULL;
 242     }
 243
 244     io_fd = ioset_add(fd);
 245     if (!io_fd) {
 246         close(fd);
 247         return NULL;
 248     }
 249     io_fd->state = IO_LISTENING;
 250     io_fd->data = data;
 251     io_fd->accept_cb = accept_cb;
 252     engine->update(io_fd);
 253     return io_fd;
 254 }
 255
 256 struct io_fd *
 257 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)) {
 258     struct addrinfo hints;
 259     struct addrinfo *ai;
 260     struct io_fd *io_fd;
 261     struct io_fd *old_active;
 262     int res;
 263     int fd;
 264     char portnum[10];
 265
 266     memset(&hints, 0, sizeof(hints));
 267     hints.ai_family = local ? local->sa_family : 0;
 268     hints.ai_socktype = SOCK_STREAM;
 269     snprintf(portnum, sizeof(portnum), "%u", port);
 270     if (getaddrinfo(peer, portnum, &hints, &ai)) {
 271         log_module(MAIN_LOG, LOG_ERROR, "getaddrinfo(%s, %s) failed.", peer, portnum);
 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, fdnum;
 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         fdnum = fd->fd;
 477         while (fd->line_len > 0) {
 478             struct io_fd *old_active;
 479             int died = 0;
 480
 481             old_active = active_fd;
 482             active_fd = fd;
 483             fd->readable_cb(fd);
 484             if (active_fd)
 485                 ioset_find_line_length(fd);
 486             else
 487                 died = 1;
 488             if (old_active != fd)
 489                 active_fd = old_active;
 490             if (died)
 491                 break;
 492         }
 493     }
 494 }
 495
 496 int
 497 ioset_line_read(struct io_fd *fd, char *dest, int max) {
 498     int line_len;
 499     int avail;
 500     int done;
 501
 502     line_len = fd->line_len;
 503     if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) ||  (line_len < 0)))
 504         return 0;
 505     if (line_len < 0)
 506         return -1;
 507     if (line_len < max)
 508         max = line_len;
 509     avail = ioq_get_avail(&fd->recv);
 510     if (max > avail) {
 511         memcpy(dest, fd->recv.buf + fd->recv.get, avail);
 512         assert(fd->recv.get + avail == fd->recv.size);
 513         fd->recv.get = 0;
 514         done = avail;
 515     } else {
 516         done = 0;
 517     }
 518     memcpy(dest + done, fd->recv.buf + fd->recv.get, max - done);
 519     fd->recv.get += max - done;
 520     if (fd->recv.get == fd->recv.size)
 521         fd->recv.get = 0;
 522     dest[max - 1] = 0;
 523     ioset_find_line_length(fd);
 524     return line_len;
 525 }
 526
 527 void
 528 ioset_events(struct io_fd *fd, int readable, int writable)
 529 {
 530     if (!fd || (!readable && !writable))
 531         return;
 532     active_fd = fd;
 533     switch (fd->state) {
 534     case IO_CLOSED:
 535         break;
 536     case IO_LISTENING:
 537         if (active_fd && readable)
 538             ioset_accept(fd);
 539         break;
 540     case IO_CONNECTING:
 541         assert(active_fd == NULL || active_fd == fd);
 542         if (active_fd && readable) {
 543             socklen_t arglen;
 544             int rc;
 545             arglen = sizeof(rc);
 546             if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
 547                 rc = errno;
 548             fd->state = IO_CLOSED;
 549             if (fd->connect_cb)
 550                 fd->connect_cb(fd, rc);
 551         } else if (active_fd && writable) {
 552             fd->state = IO_CONNECTED;
 553             if (fd->connect_cb)
 554                 fd->connect_cb(fd, 0);
 555         }
 556         if (active_fd != fd)
 557             break;
 558         engine->update(fd);
 559         /* and fall through */
 560     case IO_CONNECTED:
 561         assert(active_fd == NULL || active_fd == fd);
 562         if (active_fd && readable) {
 563             if (fd->line_reads)
 564                 ioset_buffered_read(fd);
 565             else
 566                 fd->readable_cb(fd);
 567         }
 568
 569         assert(active_fd == NULL || active_fd == fd);
 570         if (active_fd && writable)
 571             ioset_try_write(fd);
 572         break;
 573     }
 574 }
 575
 576 void
 577 ioset_run(void) {
 578     extern struct io_fd *socket_io_fd;
 579     struct timeval timeout;
 580     unsigned long wakey;
 581
 582     while (!quit_services) {
 583         while (!socket_io_fd)
 584             uplink_connect();
 585
 586         /* How long to sleep? (fill in select_timeout) */
 587         wakey = timeq_next();
 588         if (wakey < now)
 589             timeout.tv_sec = 0;
 590         else
 591             timeout.tv_sec = wakey - now;
 592         timeout.tv_usec = 0;
 593
 594         if (engine->loop(&timeout))
 595             continue;
 596
 597         /* Call any timeq events we need to call. */
 598         timeq_run();
 599         if (do_write_dbs) {
 600             saxdb_write_all();
 601             do_write_dbs = 0;
 602         }
 603         if (do_reopen) {
 604             extern char *services_config;
 605             conf_read(services_config);
 606             do_reopen = 0;
 607         }
 608     }
 609 }
 610
 611 void
 612 ioset_write(struct io_fd *fd, const char *buf, unsigned int nbw) {
 613     unsigned int avail;
 614     while (ioq_used(&fd->send) + nbw >= fd->send.size)
 615         ioq_grow(&fd->send);
 616     avail = ioq_put_avail(&fd->send);
 617     if (nbw > avail) {
 618         memcpy(fd->send.buf + fd->send.put, buf, avail);
 619         buf += avail;
 620         nbw -= avail;
 621         fd->send.put = 0;
 622     }
 623     memcpy(fd->send.buf + fd->send.put, buf, nbw);
 624     fd->send.put += nbw;
 625     if (fd->send.put == fd->send.size)
 626         fd->send.put = 0;
 627     engine->update(fd);
 628 }
 629
 630 int
 631 ioset_printf(struct io_fd *fd, const char *fmt, ...) {
 632     char tmpbuf[MAXLEN];
 633     va_list ap;
 634     int res;
 635
 636     va_start(ap, fmt);
 637     res = vsnprintf(tmpbuf, sizeof(tmpbuf), fmt, ap);
 638     va_end(ap);
 639     if (res > 0 && (size_t)res <= sizeof(tmpbuf))
 640         ioset_write(fd, tmpbuf, res);
 641     return res;
 642 }
 643
 644 void
 645 ioset_set_time(unsigned long new_now) {
 646     clock_skew = new_now - time(NULL);
 647     now = new_now;
 648 }