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         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     res = getaddrinfo(peer, portnum, &hints, &ai);
 271     if (res != 0) {
 272         log_module(MAIN_LOG, LOG_ERROR, "getaddrinfo(%s, %s) failed: %s.", peer, portnum, gai_strerror(res));
 273         return NULL;
 274     }
 275
 276     if (local) {
 277         if ((fd = socket(local->sa_family, SOCK_STREAM, 0)) < 0) {
 278             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s)", peer, errno, strerror(errno));
 279             freeaddrinfo(ai);
 280             return NULL;
 281         }
 282         if (bind(fd, local, sa_size) < 0) {
 283             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));
 284         }
 285     } else {
 286         if ((fd = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
 287             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s).", peer, errno, strerror(errno));
 288             freeaddrinfo(ai);
 289             return NULL;
 290         }
 291     }
 292
 293     if (blocking) {
 294         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
 295         io_fd = ioset_add(fd);
 296     } else {
 297         io_fd = ioset_add(fd);
 298         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
 299     }
 300     freeaddrinfo(ai);
 301     if (!io_fd) {
 302         close(fd);
 303         return NULL;
 304     }
 305     io_fd->state = IO_CONNECTING;
 306     io_fd->data = data;
 307     io_fd->connect_cb = connect_cb;
 308     if (res < 0) {
 309         switch (errno) {
 310         case EINPROGRESS: /* only if !blocking */
 311             engine->update(io_fd);
 312             return io_fd;
 313         default:
 314             log_module(MAIN_LOG, LOG_ERROR, "connect(%s:%d) (fd %d) returned errno %d (%s).", peer, port, io_fd->fd, errno, strerror(errno));
 315             /* then fall through */
 316         case EHOSTUNREACH:
 317         case ECONNREFUSED:
 318             ioset_close(io_fd, 1);
 319             return NULL;
 320         }
 321     }
 322     io_fd->state = IO_CONNECTED;
 323     old_active = active_fd;
 324     if (connect_cb)
 325         connect_cb(io_fd, ((res < 0) ? errno : 0));
 326     if (active_fd)
 327         engine->update(io_fd);
 328     if (old_active != io_fd)
 329         active_fd = old_active;
 330     return io_fd;
 331 }
 332
 333 void ioset_update(struct io_fd *fd) {
 334     engine->update(fd);
 335 }
 336
 337 static void
 338 ioset_try_write(struct io_fd *fd) {
 339     int res;
 340     unsigned int req;
 341
 342     req = ioq_get_avail(&fd->send);
 343     res = send(fd->fd, fd->send.buf+fd->send.get, req, 0);
 344     if (res < 0) {
 345         if (errno != EAGAIN) {
 346             log_module(MAIN_LOG, LOG_ERROR, "send() on fd %d error %d: %s", fd->fd, errno, strerror(errno));
 347         }
 348     } else {
 349         fd->send.get += res;
 350         if (fd->send.get == fd->send.size)
 351             fd->send.get = 0;
 352         engine->update(fd);
 353     }
 354 }
 355
 356 void
 357 ioset_close(struct io_fd *fdp, int os_close) {
 358     if (!fdp)
 359         return;
 360     if (active_fd == fdp)
 361         active_fd = NULL;
 362     if (fdp->destroy_cb)
 363         fdp->destroy_cb(fdp);
 364 #if defined(HAVE_WSAEVENTSELECT)
 365     /* This is one huge kludge.  Sorry! */
 366     if (fdp->send.get != fdp->send.put && (os_close & 2)) {
 367         engine->remove(fdp, 0);
 368         ioset_try_write(fdp);
 369         /* it may need to send the beginning of the buffer now.. */
 370         if (fdp->send.get != fdp->send.put)
 371             ioset_try_write(fdp);
 372     }
 373     free(fdp->send.buf);
 374     free(fdp->recv.buf);
 375     if (os_close & 1)
 376         closesocket(fdp->fd);
 377 #else
 378     if (fdp->send.get != fdp->send.put && (os_close & 2)) {
 379         int flags;
 380
 381         flags = fcntl(fdp->fd, F_GETFL);
 382         fcntl(fdp->fd, F_SETFL, flags&~O_NONBLOCK);
 383         ioset_try_write(fdp);
 384         /* it may need to send the beginning of the buffer now.. */
 385         if (fdp->send.get != fdp->send.put)
 386             ioset_try_write(fdp);
 387     }
 388     free(fdp->send.buf);
 389     free(fdp->recv.buf);
 390     if (os_close & 1)
 391         close(fdp->fd);
 392     engine->remove(fdp, os_close & 1);
 393 #endif
 394     free(fdp);
 395 }
 396
 397 static void
 398 ioset_accept(struct io_fd *listener)
 399 {
 400     struct io_fd *old_active;
 401     struct io_fd *new_fd;
 402     int fd;
 403
 404     fd = accept(listener->fd, NULL, 0);
 405     if (fd < 0) {
 406         log_module(MAIN_LOG, LOG_ERROR, "Unable to accept new connection on listener %d: %s", listener->fd, strerror(errno));
 407         return;
 408     }
 409
 410     new_fd = ioset_add(fd);
 411     new_fd->state = IO_CONNECTED;
 412     old_active = active_fd;
 413     active_fd = new_fd;
 414     listener->accept_cb(listener, new_fd);
 415     assert(active_fd == NULL || active_fd == new_fd);
 416     if (active_fd == new_fd) {
 417         if (new_fd->send.get != new_fd->send.put)
 418             ioset_try_write(new_fd);
 419         else
 420             engine->update(new_fd);
 421     }
 422     active_fd = old_active;
 423 }
 424
 425 static int
 426 ioset_find_line_length(struct io_fd *fd) {
 427     unsigned int pos, max, len;
 428     len = 0;
 429     max = (fd->recv.put < fd->recv.get) ? fd->recv.size : fd->recv.put;
 430     for (pos = fd->recv.get; pos < max; ++pos, ++len)
 431         if (IS_EOL(fd->recv.buf[pos]))
 432             return fd->line_len = len + 1;
 433     if (fd->recv.put < fd->recv.get)
 434         for (pos = 0; pos < fd->recv.put; ++pos, ++len)
 435             if (IS_EOL(fd->recv.buf[pos]))
 436                 return fd->line_len = len + 1;
 437     return fd->line_len = 0;
 438 }
 439
 440 static void
 441 ioset_buffered_read(struct io_fd *fd) {
 442     int put_avail, nbr, fdnum;
 443
 444     if (!(put_avail = ioq_put_avail(&fd->recv)))
 445         put_avail = ioq_grow(&fd->recv);
 446     nbr = recv(fd->fd, fd->recv.buf + fd->recv.put, put_avail, 0);
 447     if (nbr < 0) {
 448         if (errno != EAGAIN) {
 449             log_module(MAIN_LOG, LOG_ERROR, "Unexpected recv() error %d on fd %d: %s", errno, fd->fd, strerror(errno));
 450             /* Just flag it as EOF and call readable_cb() to notify the fd's owner. */
 451             fd->state = IO_CLOSED;
 452             fd->readable_cb(fd);
 453             if (active_fd == fd)
 454                 engine->update(fd);
 455         }
 456     } else if (nbr == 0) {
 457         fd->state = IO_CLOSED;
 458         fd->readable_cb(fd);
 459         if (active_fd == fd)
 460             engine->update(fd);
 461     } else {
 462         if (fd->line_len == 0) {
 463             unsigned int pos;
 464             for (pos = fd->recv.put; pos < fd->recv.put + nbr; ++pos) {
 465                 if (IS_EOL(fd->recv.buf[pos])) {
 466                     if (fd->recv.put < fd->recv.get)
 467                         fd->line_len = fd->recv.size + pos + 1 - fd->recv.get;
 468                     else
 469                         fd->line_len = pos + 1 - fd->recv.get;
 470                     break;
 471                 }
 472             }
 473         }
 474         fd->recv.put += nbr;
 475         if (fd->recv.put == fd->recv.size)
 476             fd->recv.put = 0;
 477         fdnum = fd->fd;
 478         while (fd->line_len > 0) {
 479             struct io_fd *old_active;
 480             int died = 0;
 481
 482             old_active = active_fd;
 483             active_fd = fd;
 484             fd->readable_cb(fd);
 485             if (active_fd)
 486                 ioset_find_line_length(fd);
 487             else
 488                 died = 1;
 489             if (old_active != fd)
 490                 active_fd = old_active;
 491             if (died)
 492                 break;
 493         }
 494     }
 495 }
 496
 497 int
 498 ioset_line_read(struct io_fd *fd, char *dest, int max) {
 499     int line_len;
 500     int avail;
 501     int done;
 502
 503     line_len = fd->line_len;
 504     if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) ||  (line_len < 0)))
 505         return 0;
 506     if (line_len < 0)
 507         return -1;
 508     if (line_len < max)
 509         max = line_len;
 510     avail = ioq_get_avail(&fd->recv);
 511     if (max > avail) {
 512         memcpy(dest, fd->recv.buf + fd->recv.get, avail);
 513         assert(fd->recv.get + avail == fd->recv.size);
 514         fd->recv.get = 0;
 515         done = avail;
 516     } else {
 517         done = 0;
 518     }
 519     memcpy(dest + done, fd->recv.buf + fd->recv.get, max - done);
 520     fd->recv.get += max - done;
 521     if (fd->recv.get == fd->recv.size)
 522         fd->recv.get = 0;
 523     dest[max - 1] = 0;
 524     ioset_find_line_length(fd);
 525     return line_len;
 526 }
 527
 528 void
 529 ioset_events(struct io_fd *fd, int readable, int writable)
 530 {
 531     if (!fd || (!readable && !writable))
 532         return;
 533     active_fd = fd;
 534     switch (fd->state) {
 535     case IO_CLOSED:
 536         break;
 537     case IO_LISTENING:
 538         if (active_fd && readable)
 539             ioset_accept(fd);
 540         break;
 541     case IO_CONNECTING:
 542         assert(active_fd == NULL || active_fd == fd);
 543         if (active_fd && readable) {
 544             socklen_t arglen;
 545             int rc;
 546             arglen = sizeof(rc);
 547             if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
 548                 rc = errno;
 549             fd->state = IO_CLOSED;
 550             if (fd->connect_cb)
 551                 fd->connect_cb(fd, rc);
 552         } else if (active_fd && writable) {
 553             fd->state = IO_CONNECTED;
 554             if (fd->connect_cb)
 555                 fd->connect_cb(fd, 0);
 556         }
 557         if (active_fd != fd)
 558             break;
 559         engine->update(fd);
 560         /* and fall through */
 561     case IO_CONNECTED:
 562         assert(active_fd == NULL || active_fd == fd);
 563         if (active_fd && readable) {
 564             if (fd->line_reads)
 565                 ioset_buffered_read(fd);
 566             else
 567                 fd->readable_cb(fd);
 568         }
 569
 570         assert(active_fd == NULL || active_fd == fd);
 571         if (active_fd && writable)
 572             ioset_try_write(fd);
 573         break;
 574     }
 575 }
 576
 577 void
 578 ioset_run(void) {
 579     extern struct io_fd *socket_io_fd;
 580     struct timeval timeout;
 581     time_t wakey;
 582
 583     while (!quit_services) {
 584         while (!socket_io_fd)
 585             uplink_connect();
 586
 587         /* How long to sleep? (fill in select_timeout) */
 588         wakey = timeq_next();
 589         if (wakey < now)
 590             timeout.tv_sec = 0;
 591         else
 592             timeout.tv_sec = wakey - now;
 593         timeout.tv_usec = 0;
 594
 595         if (engine->loop(&timeout))
 596             continue;
 597
 598         /* Call any timeq events we need to call. */
 599         timeq_run();
 600         if (do_write_dbs) {
 601             saxdb_write_all();
 602             do_write_dbs = 0;
 603         }
 604         if (do_reopen) {
 605             extern char *services_config;
 606             conf_read(services_config);
 607             do_reopen = 0;
 608         }
 609     }
 610 }
 611
 612 void
 613 ioset_write(struct io_fd *fd, const char *buf, unsigned int nbw) {
 614     unsigned int avail;
 615     while (ioq_used(&fd->send) + nbw >= fd->send.size)
 616         ioq_grow(&fd->send);
 617     avail = ioq_put_avail(&fd->send);
 618     if (nbw > avail) {
 619         memcpy(fd->send.buf + fd->send.put, buf, avail);
 620         buf += avail;
 621         nbw -= avail;
 622         fd->send.put = 0;
 623     }
 624     memcpy(fd->send.buf + fd->send.put, buf, nbw);
 625     fd->send.put += nbw;
 626     if (fd->send.put == fd->send.size)
 627         fd->send.put = 0;
 628     engine->update(fd);
 629 }
 630
 631 int
 632 ioset_printf(struct io_fd *fd, const char *fmt, ...) {
 633     char tmpbuf[MAXLEN];
 634     va_list ap;
 635     int res;
 636
 637     va_start(ap, fmt);
 638     res = vsnprintf(tmpbuf, sizeof(tmpbuf), fmt, ap);
 639     va_end(ap);
 640     if (res > 0 && (size_t)res <= sizeof(tmpbuf))
 641         ioset_write(fd, tmpbuf, res);
 642     return res;
 643 }
 644
 645 void
 646 ioset_set_time(unsigned long new_now) {
 647     clock_skew = new_now - time(NULL);
 648     now = new_now;
 649 }