src/IOHandler.c

   1 /* IOHandler.c - IOMultiplexer
   2  * Copyright (C) 2012  Philipp Kreil (pk910)
   3  *
   4  * This program is free software: you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License as published by
   6  * the Free Software Foundation, either version 3 of the License, or
   7  * (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  * GNU General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License
  15  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  16  */
  17 #include "IOHandler.h"
  18 #include "IOEngine.h"
  19 #include "IOHandler_SSL.h"
  20 #include <errno.h>
  21 #include <stdio.h>
  22 #include <unistd.h>
  23 #ifdef WIN32
  24 #define _WIN32_WINNT 0x501
  25 #include <windows.h>
  26 #include <winsock2.h>
  27 #include <ws2tcpip.h>
  28 #else
  29 #include <signal.h>
  30 #include <stdarg.h>
  31 #include <stdlib.h>
  32 #include <string.h>
  33 #include <netdb.h>
  34 #include <sys/socket.h>
  35 #include <netinet/in.h>
  36 #include <netinet/tcp.h>
  37 #include <fcntl.h>
  38 #endif
  39
  40 #ifndef EWOULDBLOCK
  41 #define EWOULDBLOCK EAGAIN
  42 #endif
  43
  44 #define MAXLOG 1024
  45 iohandler_log_callback *iolog_backend = NULL;
  46
  47 struct IODescriptor *first_descriptor = NULL;
  48 struct IODescriptor *timer_priority = NULL;
  49
  50 #ifdef HAVE_PTHREAD_H
  51 static pthread_mutex_t io_thread_sync;
  52 static pthread_mutex_t io_poll_sync;
  53 #endif
  54
  55 void iohandler_log(enum IOLogType type, char *text, ...) {
  56     va_list arg_list;
  57     char logBuf[MAXLOG+1];
  58     int pos;
  59     logBuf[0] = '\0';
  60     va_start(arg_list, text);
  61     pos = vsnprintf(logBuf, MAXLOG - 1, text, arg_list);
  62     va_end(arg_list);
  63     if (pos < 0 || pos > (MAXLOG - 1)) pos = MAXLOG - 1;
  64     logBuf[pos] = '\n';
  65     logBuf[pos+1] = '\0';
  66
  67     if(iolog_backend)
  68         iolog_backend(type, logBuf);
  69 }
  70
  71 /* IO Engines */
  72 extern struct IOEngine engine_select; /* select system call (should always be useable) */
  73 extern struct IOEngine engine_kevent;
  74 extern struct IOEngine engine_epoll;
  75 extern struct IOEngine engine_win32;
  76
  77 int iohandler_settings = 0;
  78 struct IOEngine *engine = NULL;
  79
  80 void iohandler_set(int setting, int value) {
  81     if(value)
  82         iohandler_settings |= setting;
  83     else
  84         iohandler_settings &= ~setting;
  85 }
  86
  87 static void iohandler_init_engine() {
  88     if(engine) return;
  89     IOTHREAD_MUTEX_INIT(io_thread_sync);
  90     IOTHREAD_MUTEX_INIT(io_poll_sync);
  91
  92     //try other engines
  93     if(!(iohandler_settings & IOHANDLER_SETTING_HIGH_PRECISION_TIMER)) {
  94         if(!engine && engine_kevent.init && engine_kevent.init())
  95             engine = &engine_kevent;
  96         if(!engine && engine_epoll.init && engine_epoll.init())
  97             engine = &engine_epoll;
  98         if(!engine && engine_win32.init && engine_win32.init())
  99             engine = &engine_win32;
 100     }
 101
 102     if (!engine) {
 103         if(engine_select.init())
 104             engine = &engine_select;
 105         else {
 106             iohandler_log(IOLOG_FATAL, "found no useable IO engine");
 107             return;
 108         }
 109     }
 110     iohandler_log(IOLOG_DEBUG, "using %s IO engine", engine->name);
 111     iohandler_ssl_init();
 112 }
 113
 114 static void iohandler_append(struct IODescriptor *descriptor) {
 115     IOSYNCHRONIZE(io_thread_sync);
 116     struct timeval *timeout = ((descriptor->timeout.tv_sec || descriptor->timeout.tv_usec) ? &descriptor->timeout : NULL);
 117     if(timeout) {
 118         struct IODescriptor *iofd;
 119         int set_priority = 1;
 120         descriptor->timeout = *timeout;
 121         if(timer_priority)
 122             iofd = timer_priority;
 123         else
 124             iofd = first_descriptor;
 125         if(iofd) {
 126             for(;;iofd = iofd->next) {
 127                 if(timeval_is_smaler(timeout, (&iofd->timeout))) {
 128                     descriptor->prev = iofd->prev;
 129                     descriptor->next = iofd;
 130                     if(iofd->prev)
 131                         iofd->prev->next = descriptor;
 132                     iofd->prev = descriptor;
 133                     if(set_priority)
 134                         timer_priority = descriptor;
 135                     break;
 136                 }
 137                 if(iofd == timer_priority)
 138                     set_priority = 0;
 139                 if(iofd->next == NULL) {
 140                     descriptor->next = NULL;
 141                     descriptor->prev = iofd;
 142                     iofd->next = descriptor;
 143                     if(set_priority)
 144                         timer_priority = descriptor;
 145                     break;
 146                 }
 147             }
 148         } else {
 149             descriptor->prev = NULL;
 150             descriptor->next = NULL;
 151             first_descriptor = descriptor;
 152             timer_priority = descriptor;
 153         }
 154
 155     } else {
 156         descriptor->prev = NULL;
 157         descriptor->next = first_descriptor;
 158         if(first_descriptor)
 159             first_descriptor->prev = descriptor;
 160         first_descriptor = descriptor;
 161     }
 162     IODESYNCHRONIZE(io_thread_sync);
 163 }
 164
 165 static void iohandler_remove(struct IODescriptor *descriptor, int engine_remove) {
 166     //remove IODescriptor from the list
 167     IOSYNCHRONIZE(io_thread_sync);
 168     if(descriptor->prev)
 169         descriptor->prev->next = descriptor->next;
 170     else
 171         first_descriptor = descriptor->next;
 172     if(descriptor->next)
 173         descriptor->next->prev = descriptor->prev;
 174     if(descriptor == timer_priority)
 175         timer_priority = descriptor->next;
 176
 177     if(engine_remove)
 178         engine->remove(descriptor);
 179     if(descriptor->readbuf.buffer)
 180         free(descriptor->readbuf.buffer);
 181     if(descriptor->writebuf.buffer)
 182         free(descriptor->writebuf.buffer);
 183     iohandler_log(IOLOG_DEBUG, "removed IODescriptor (%d) of type `%s`", descriptor->fd, iohandler_iotype_name(descriptor->type));
 184     free(descriptor);
 185     IODESYNCHRONIZE(io_thread_sync);
 186 }
 187
 188 struct IODescriptor *iohandler_add(int sockfd, enum IOType type, struct timeval *timeout, iohandler_callback *callback) {
 189     //just add a new IODescriptor
 190     struct IODescriptor *descriptor = calloc(1, sizeof(*descriptor));
 191     if(!descriptor) {
 192         iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
 193         return NULL;
 194     }
 195     descriptor->fd = (type == IOTYPE_STDIN ? fileno(stdin) : sockfd);
 196     descriptor->type = type;
 197     descriptor->state = (type == IOTYPE_STDIN ? IO_CONNECTED : IO_CLOSED);
 198     descriptor->callback = callback;
 199     if(timeout) {
 200         descriptor->timeout = *timeout;
 201         if(descriptor->timeout.tv_usec > 1000000) {
 202             descriptor->timeout.tv_usec -= 1000000;
 203             descriptor->timeout.tv_sec++;
 204         }
 205     }
 206     if(type != IOTYPE_TIMER) {
 207         descriptor->readbuf.buffer = malloc(IO_READ_BUFLEN + 2);
 208         descriptor->readbuf.bufpos = 0;
 209         descriptor->readbuf.buflen = IO_READ_BUFLEN;
 210         descriptor->writebuf.buffer = malloc(IO_READ_BUFLEN + 2);
 211         descriptor->writebuf.bufpos = 0;
 212         descriptor->writebuf.buflen = IO_READ_BUFLEN;
 213     }
 214
 215     if(!engine) {
 216         iohandler_init_engine();
 217         if(!engine) {
 218             return NULL;
 219         }
 220     }
 221     engine->add(descriptor);
 222
 223     //add IODescriptor to the list
 224     iohandler_append(descriptor);
 225
 226     iohandler_log(IOLOG_DEBUG, "added custom socket descriptor (%d) as type `%s`", sockfd, iohandler_iotype_name(type));
 227     return descriptor;
 228 }
 229
 230 void iohandler_set_timeout(struct IODescriptor *descriptor, struct timeval *timeout) {
 231     if(descriptor->prev)
 232         descriptor->prev->next = descriptor->next;
 233     else
 234         first_descriptor = descriptor->next;
 235     if(descriptor->next)
 236         descriptor->next->prev = descriptor->prev;
 237     if(descriptor == timer_priority)
 238         timer_priority = descriptor->next;
 239     if(timeout && timeout->tv_sec == 0 && descriptor->constant_timeout) {
 240         descriptor->timeout.tv_usec += (descriptor->constant_timeout % 1000) * 1000;
 241         descriptor->timeout.tv_sec += (descriptor->constant_timeout / 1000);
 242         if(descriptor->timeout.tv_usec > 1000000) {
 243             descriptor->timeout.tv_sec += (descriptor->timeout.tv_usec / 1000000);
 244             descriptor->timeout.tv_usec %= 1000000;
 245         }
 246     } else if(timeout) {
 247         descriptor->timeout = *timeout;
 248         if(descriptor->timeout.tv_usec > 1000000) {
 249             descriptor->timeout.tv_usec -= 1000000;
 250             descriptor->timeout.tv_sec++;
 251         }
 252     } else {
 253         descriptor->timeout.tv_sec = 0;
 254         descriptor->timeout.tv_usec = 0;
 255     }
 256     iohandler_append(descriptor);
 257 }
 258
 259 static void iohandler_increase_iobuf(struct IOBuffer *iobuf, size_t required) {
 260     if(iobuf->buflen >= required) return;
 261     char *new_buf = realloc(iobuf->buffer, required + 2);
 262     if(new_buf) {
 263         iobuf->buffer = new_buf;
 264         iobuf->buflen = required;
 265     }
 266 }
 267
 268 struct IODescriptor *iohandler_timer(struct timeval timeout, iohandler_callback *callback) {
 269     struct IODescriptor *descriptor;
 270     descriptor = iohandler_add(-1, IOTYPE_TIMER, &timeout, callback);
 271     if(!descriptor) {
 272         iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
 273         return NULL;
 274     }
 275     iohandler_log(IOLOG_DEBUG, "added timer descriptor (sec: %d; usec: %d)", timeout.tv_sec, timeout.tv_usec);
 276     return descriptor;
 277 }
 278
 279 struct IODescriptor *iohandler_constant_timer(int msec, iohandler_callback *callback) {
 280     struct IODescriptor *descriptor;
 281     struct timeval timeout;
 282     gettimeofday(&timeout, NULL);
 283     timeout.tv_usec += (msec % 1000) * 1000;
 284     timeout.tv_sec += (msec / 1000);
 285     if(timeout.tv_usec > 1000000) {
 286         timeout.tv_sec += (timeout.tv_usec / 1000000);
 287         timeout.tv_usec %= 1000000;
 288     }
 289     descriptor = iohandler_add(-1, IOTYPE_TIMER, &timeout, callback);
 290     if(!descriptor) {
 291         iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
 292         return NULL;
 293     }
 294     descriptor->constant_timeout = msec;
 295     iohandler_log(IOLOG_DEBUG, "added timer descriptor (sec: %d; usec: %d)", timeout.tv_sec, timeout.tv_usec);
 296     return descriptor;
 297 }
 298
 299 struct IODescriptor *iohandler_connect(const char *hostname, unsigned int port, int ssl, const char *bindhost, iohandler_callback *callback) {
 300     return iohandler_connect_flags(hostname, port, ssl, bindhost, callback, IOHANDLER_CONNECT_IPV4 | IOHANDLER_CONNECT_IPV6);
 301 }
 302
 303 struct IODescriptor *iohandler_connect_flags(const char *hostname, unsigned int port, int ssl, const char *bindhost, iohandler_callback *callback, int flags) {
 304     //non-blocking connect
 305     int sockfd, result;
 306     struct addrinfo hints, *res;
 307     struct sockaddr_in *ip4 = NULL;
 308     struct sockaddr_in6 *ip6 = NULL;
 309     size_t dstaddrlen;
 310     struct sockaddr *dstaddr = NULL;
 311     struct IODescriptor *descriptor;
 312
 313     if(!engine) {
 314         iohandler_init_engine();
 315         if(!engine) return NULL;
 316     }
 317     memset (&hints, 0, sizeof (hints));
 318     hints.ai_family = PF_UNSPEC;
 319     hints.ai_socktype = SOCK_STREAM;
 320     hints.ai_flags |= AI_CANONNAME;
 321     if ((result = getaddrinfo (hostname, NULL, &hints, &res))) {
 322         iohandler_log(IOLOG_ERROR, "could not resolve %s to an IP address (%d)", hostname, result);
 323         return NULL;
 324     }
 325     while (res) {
 326         switch (res->ai_family) {
 327         case AF_INET:
 328             ip4 = (struct sockaddr_in *) res->ai_addr;
 329             break;
 330         case AF_INET6:
 331             ip6 = (struct sockaddr_in6 *) res->ai_addr;
 332             break;
 333         }
 334         res = res->ai_next;
 335         freeaddrinfo(res);
 336     }
 337
 338     if(ip6 && (flags & IOHANDLER_CONNECT_IPV6)) {
 339         sockfd = socket(AF_INET6, SOCK_STREAM, 0);
 340         if(sockfd == -1) {
 341             iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
 342             return NULL;
 343         }
 344
 345         ip6->sin6_family = AF_INET6;
 346         ip6->sin6_port = htons(port);
 347
 348         struct sockaddr_in6 *ip6vhost = NULL;
 349         if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
 350             while (res) {
 351                 switch (res->ai_family) {
 352                 case AF_INET6:
 353                     ip6vhost = (struct sockaddr_in6 *) res->ai_addr;
 354                     break;
 355                 }
 356                 res = res->ai_next;
 357                 freeaddrinfo(res);
 358             }
 359         }
 360         if(ip6vhost) {
 361             ip6vhost->sin6_family = AF_INET6;
 362             ip6vhost->sin6_port = htons(0);
 363             bind(sockfd, (struct sockaddr*)ip6vhost, sizeof(*ip6vhost));
 364         }
 365         dstaddr = (struct sockaddr*)ip6;
 366         dstaddrlen = sizeof(*ip6);
 367     } else if(ip4 && (flags & IOHANDLER_CONNECT_IPV4)) {
 368         sockfd = socket(AF_INET, SOCK_STREAM, 0);
 369         if(sockfd == -1) {
 370             iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
 371             return NULL;
 372         }
 373
 374         ip4->sin_family = AF_INET;
 375         ip4->sin_port = htons(port);
 376
 377         struct sockaddr_in *ip4vhost = NULL;
 378         if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
 379             while (res) {
 380                 switch (res->ai_family) {
 381                 case AF_INET:
 382                     ip4vhost = (struct sockaddr_in *) res->ai_addr;
 383                     break;
 384                 }
 385                 res = res->ai_next;
 386                 freeaddrinfo(res);
 387             }
 388         }
 389         if(ip4vhost) {
 390             ip4vhost->sin_family = AF_INET;
 391             ip4vhost->sin_port = htons(0);
 392             bind(sockfd, (struct sockaddr*)ip4vhost, sizeof(*ip4vhost));
 393         }
 394         dstaddr = (struct sockaddr*)ip4;
 395         dstaddrlen = sizeof(*ip4);
 396     } else
 397         return NULL;
 398     //prevent SIGPIPE
 399     #ifndef WIN32
 400     #if defined(SO_NOSIGPIPE)
 401     {
 402         int set = 1;
 403         setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
 404     }
 405     #else
 406     signal(SIGPIPE, SIG_IGN);
 407     #endif
 408     #endif
 409     //make sockfd unblocking
 410     #if defined(F_GETFL)
 411     {
 412         int fcntl_flags;
 413         fcntl_flags = fcntl(sockfd, F_GETFL);
 414         fcntl(sockfd, F_SETFL, fcntl_flags|O_NONBLOCK);
 415         fcntl_flags = fcntl(sockfd, F_GETFD);
 416         fcntl(sockfd, F_SETFD, fcntl_flags|FD_CLOEXEC);
 417     }
 418     #else
 419     /* I hope you're using the Win32 backend or something else that
 420      * automatically marks the file descriptor non-blocking...
 421      */
 422     #endif
 423     descriptor = iohandler_add(sockfd, IOTYPE_CLIENT, NULL, callback);
 424     if(!descriptor) {
 425         close(sockfd);
 426         return NULL;
 427     }
 428     connect(sockfd, dstaddr, dstaddrlen); //returns EINPROGRESS here (nonblocking)
 429     descriptor->state = IO_CONNECTING;
 430     descriptor->ssl = (ssl ? 1 : 0);
 431     descriptor->read_lines = 1;
 432     engine->update(descriptor);
 433     iohandler_log(IOLOG_DEBUG, "added client socket (%d) connecting to %s:%d", sockfd, hostname, port);
 434     return descriptor;
 435 }
 436
 437 struct IODescriptor *iohandler_listen(const char *hostname, unsigned int port, iohandler_callback *callback) {
 438     return iohandler_listen_flags(hostname, port, callback, IOHANDLER_LISTEN_IPV4 | IOHANDLER_LISTEN_IPV6);
 439 }
 440
 441 struct IODescriptor *iohandler_listen_flags(const char *hostname, unsigned int port, iohandler_callback *callback, int flags) {
 442     int sockfd;
 443     struct addrinfo hints, *res;
 444     struct sockaddr_in *ip4 = NULL;
 445     struct sockaddr_in6 *ip6 = NULL;
 446     struct IODescriptor *descriptor;
 447     unsigned int opt;
 448
 449     if(!engine) {
 450         iohandler_init_engine();
 451         if(!engine) return NULL;
 452     }
 453     memset (&hints, 0, sizeof (hints));
 454     hints.ai_family = PF_UNSPEC;
 455     hints.ai_socktype = SOCK_STREAM;
 456     hints.ai_flags |= AI_CANONNAME;
 457     if (getaddrinfo (hostname, NULL, &hints, &res)) {
 458         return NULL;
 459     }
 460     while (res) {
 461         switch (res->ai_family) {
 462         case AF_INET:
 463             ip4 = (struct sockaddr_in *) res->ai_addr;
 464             break;
 465         case AF_INET6:
 466             ip6 = (struct sockaddr_in6 *) res->ai_addr;
 467             break;
 468         }
 469         res = res->ai_next;
 470         freeaddrinfo(res);
 471     }
 472
 473     if(ip6 && (flags & IOHANDLER_LISTEN_IPV6)) {
 474         sockfd = socket(AF_INET6, SOCK_STREAM, 0);
 475         if(sockfd == -1) return NULL;
 476
 477         opt = 1;
 478         setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
 479
 480         ip6->sin6_family = AF_INET6;
 481         ip6->sin6_port = htons(port);
 482
 483         bind(sockfd, (struct sockaddr*)ip6, sizeof(*ip6));
 484     } else if(ip4 && (flags && IOHANDLER_LISTEN_IPV4)) {
 485         sockfd = socket(AF_INET, SOCK_STREAM, 0);
 486         if(sockfd == -1) return NULL;
 487
 488         opt = 1;
 489         setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
 490
 491         ip4->sin_family = AF_INET;
 492         ip4->sin_port = htons(port);
 493
 494         bind(sockfd, (struct sockaddr*)ip4, sizeof(*ip4));
 495     } else
 496         return NULL;
 497     //prevent SIGPIPE
 498     #ifndef WIN32
 499     #if defined(SO_NOSIGPIPE)
 500     {
 501         int set = 1;
 502         setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
 503     }
 504     #else
 505     signal(SIGPIPE, SIG_IGN);
 506     #endif
 507     #endif
 508     //make sockfd unblocking
 509     #if defined(F_GETFL)
 510     {
 511         int flag;
 512         flag = fcntl(sockfd, F_GETFL);
 513         fcntl(sockfd, F_SETFL, flag|O_NONBLOCK);
 514         flag = fcntl(sockfd, F_GETFD);
 515         fcntl(sockfd, F_SETFD, flag|FD_CLOEXEC);
 516     }
 517     #else
 518     /* I hope you're using the Win32 backend or something else that
 519      * automatically marks the file descriptor non-blocking...
 520      */
 521     #endif
 522     descriptor = iohandler_add(sockfd, IOTYPE_SERVER, NULL, callback);
 523     if(!descriptor) {
 524         close(sockfd);
 525         return NULL;
 526     }
 527     listen(sockfd, 1);
 528     descriptor->state = IO_LISTENING;
 529     engine->update(descriptor);
 530     iohandler_log(IOLOG_DEBUG, "added server socket (%d) listening on %s:%d", sockfd, hostname, port);
 531     return descriptor;
 532 }
 533
 534 struct IODescriptor *iohandler_listen_ssl(const char *hostname, unsigned int port, const char *certfile, const char *keyfile, iohandler_callback *callback) {
 535     return iohandler_listen_ssl_flags(hostname, port, certfile, keyfile, callback, IOHANDLER_LISTEN_IPV4 | IOHANDLER_LISTEN_IPV6);
 536 }
 537
 538 struct IODescriptor *iohandler_listen_ssl_flags(const char *hostname, unsigned int port, const char *certfile, const char *keyfile, iohandler_callback *callback, int flags) {
 539     struct IODescriptor *descriptor = iohandler_listen_flags(hostname, port, callback, flags);
 540     if(!descriptor)
 541         return NULL;
 542     //SSL Server Socket
 543     iohandler_ssl_listen(descriptor, certfile, keyfile);
 544     if(descriptor->sslnode)
 545         descriptor->ssl = 1;
 546     return descriptor;
 547 }
 548
 549 void iohandler_write(struct IODescriptor *iofd, const char *line) {
 550     size_t linelen = strlen(line);
 551     iohandler_send(iofd, line, linelen);
 552 }
 553
 554 void iohandler_send(struct IODescriptor *iofd, const char *data, size_t datalen) {
 555     if(iofd->type == IOTYPE_TIMER || iofd->state == IO_CLOSED) {
 556         iohandler_log(IOLOG_ERROR, "could not write to socket (%s)", (iofd->type == IOTYPE_TIMER ? "IOTYPE_TIMER" : "IO_CLOSED"));
 557         return;
 558     }
 559     iohandler_log(IOLOG_DEBUG, "add %d to writebuf (fd: %d): %s", datalen, iofd->fd, data);
 560     if(iofd->writebuf.buflen < iofd->writebuf.bufpos + datalen) {
 561         iohandler_log(IOLOG_DEBUG, "increase writebuf (curr: %d) to %d (+%d bytes)", iofd->writebuf.buflen, iofd->writebuf.bufpos + datalen, (iofd->writebuf.bufpos + datalen - iofd->writebuf.buflen));
 562         iohandler_increase_iobuf(&iofd->writebuf, iofd->writebuf.bufpos + datalen);
 563         if(iofd->writebuf.buflen < iofd->writebuf.bufpos + datalen) {
 564             iohandler_log(IOLOG_ERROR, "increase writebuf (curr: %d) to %d (+%d bytes) FAILED", iofd->writebuf.buflen, iofd->writebuf.bufpos + datalen, (iofd->writebuf.bufpos + datalen - iofd->writebuf.buflen));
 565             return;
 566         }
 567     }
 568     memcpy(iofd->writebuf.buffer + iofd->writebuf.bufpos, data, datalen);
 569     iofd->writebuf.bufpos += datalen;
 570     engine->update(iofd);
 571 }
 572
 573 void iohandler_printf(struct IODescriptor *iofd, const char *text, ...) {
 574     va_list arg_list;
 575     char sendBuf[IO_LINE_LEN];
 576     int pos;
 577     sendBuf[0] = '\0';
 578     va_start(arg_list, text);
 579     pos = vsnprintf(sendBuf, IO_LINE_LEN - 2, text, arg_list);
 580     va_end(arg_list);
 581     if (pos < 0 || pos > (IO_LINE_LEN - 2)) pos = IO_LINE_LEN - 2;
 582     sendBuf[pos] = '\n';
 583     sendBuf[pos+1] = '\0';
 584     iohandler_send(iofd, sendBuf, pos+1);
 585 }
 586
 587 static int iohandler_try_write(struct IODescriptor *iofd) {
 588     if(!iofd->writebuf.bufpos) return 0;
 589     iohandler_log(IOLOG_DEBUG, "write writebuf (%d bytes) to socket (fd: %d)", iofd->writebuf.bufpos, iofd->fd);
 590     int res;
 591     if(iofd->ssl_active)
 592         res = iohandler_ssl_write(iofd, iofd->writebuf.buffer, iofd->writebuf.bufpos);
 593     else
 594         res = send(iofd->fd, iofd->writebuf.buffer, iofd->writebuf.bufpos, 0);
 595     if(res < 0) {
 596         if (errno != EAGAIN && errno != EWOULDBLOCK)
 597             iohandler_log(IOLOG_ERROR, "could not write to socket (fd: %d): %d - %s", iofd->fd, errno, strerror(errno));
 598         else
 599             res = 0;
 600     } else {
 601         iofd->writebuf.bufpos -= res;
 602         if(iofd->state != IO_CLOSED)
 603             engine->update(iofd);
 604     }
 605     return res;
 606 }
 607
 608 void iohandler_close(struct IODescriptor *iofd) {
 609     int engine_remove = 1;
 610     iofd->state = IO_CLOSED;
 611     if(iofd->writebuf.bufpos) {
 612         //try to send everything before closing
 613 #if defined(F_GETFL)
 614         {
 615             int flags;
 616             flags = fcntl(iofd->fd, F_GETFL);
 617             fcntl(iofd->fd, F_SETFL, flags & ~O_NONBLOCK);
 618             flags = fcntl(iofd->fd, F_GETFD);
 619             fcntl(iofd->fd, F_SETFD, flags|FD_CLOEXEC);
 620         }
 621 #else
 622         engine_remove = 0;
 623         engine->remove(iofd);
 624 #endif
 625         iohandler_try_write(iofd);
 626     }
 627     //close IODescriptor
 628     if(iofd->ssl)
 629         iohandler_ssl_disconnect(iofd);
 630     if(iofd->type == IOTYPE_SERVER || iofd->type == IOTYPE_CLIENT || iofd->type == IOTYPE_STDIN)
 631         close(iofd->fd);
 632     iohandler_remove(iofd, engine_remove);
 633 }
 634
 635 void iohandler_update(struct IODescriptor *iofd) {
 636     iohandler_log(IOLOG_DEBUG, "external call to iohandler_update (fd: %d)", iofd->fd);
 637     engine->update(iofd);
 638 }
 639
 640 static void iohandler_trigger_event(struct IOEvent *event) {
 641     if(!event->iofd->callback) return;
 642     iohandler_log(IOLOG_DEBUG, "triggering event (%s) for %s (fd: %d)", iohandler_ioeventtype_name(event->type), iohandler_iotype_name(event->iofd->type), event->iofd->fd);
 643     event->iofd->callback(event);
 644 }
 645
 646 void iohandler_events(struct IODescriptor *iofd, int readable, int writeable) {
 647     struct IOEvent callback_event;
 648     callback_event.type = IOEVENT_IGNORE;
 649     callback_event.iofd = iofd;
 650     switch(iofd->state) {
 651         case IO_SSLWAIT:
 652             if(!readable && !writeable) {
 653                 if(!iofd->ssl_server_hs) {
 654                     callback_event.type = IOEVENT_SSLFAILED;
 655                     iofd->state = IO_CLOSED;
 656                     engine->update(iofd);
 657                 } else
 658                     iohandler_close(iofd);
 659             } else if(iofd->ssl_server_hs) {
 660                 iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_server_handshake for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
 661                 iohandler_ssl_server_handshake(iofd);
 662             } else {
 663                 iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_client_handshake for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
 664                 iohandler_ssl_client_handshake(iofd);
 665             }
 666             break;
 667         case IO_CLOSED:
 668             if(iofd->type == IOTYPE_TIMER)
 669                 callback_event.type = IOEVENT_TIMEOUT;
 670             break;
 671         case IO_LISTENING:
 672             if(readable) {
 673                 callback_event.data.accept_fd = accept(iofd->fd, NULL, 0);
 674                 if(callback_event.data.accept_fd < 0) {
 675                     iohandler_log(IOLOG_ERROR, "could not accept client (server fd: %d): %d - %s", iofd->fd, errno, strerror(errno));
 676                 } else if(iofd->ssl) {
 677                     struct IODescriptor *client_iofd = iohandler_add(callback_event.data.accept_fd, IOTYPE_CLIENT, NULL, NULL);
 678                     iohandler_ssl_client_accepted(iofd, client_iofd);
 679                 } else
 680                     callback_event.type = IOEVENT_ACCEPT;
 681             }
 682             break;
 683         case IO_CONNECTING:
 684             if(readable) { //could not connect
 685                 callback_event.type = IOEVENT_NOTCONNECTED;
 686                 //socklen_t arglen;
 687                 //arglen = sizeof(callback_event.data.errid);
 688                 //if (getsockopt(iofd->fd, SOL_SOCKET, SO_ERROR, &callback_event.data.errid, &arglen) < 0)
 689                 //    callback_event.data.errid = errno;
 690                 iofd->state = IO_CLOSED;
 691                                 engine->update(iofd);
 692             } else if(writeable) {
 693                 if(iofd->ssl && !iofd->ssl_active) {
 694                     iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_connect for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
 695                     iohandler_ssl_connect(iofd);
 696                     return;
 697                 }
 698                 if(iofd->ssl && iofd->ssl_server_hs) {
 699                     callback_event.type = IOEVENT_SSLACCEPT;
 700                     callback_event.iofd = iofd->data;
 701                     callback_event.data.accept_iofd = iofd;
 702                     iofd->data = NULL;
 703                 }
 704                 else
 705                     callback_event.type = IOEVENT_CONNECTED;
 706                 iofd->state = IO_CONNECTED;
 707                 engine->update(iofd);
 708             }
 709             break;
 710         case IO_CONNECTED:
 711             if(readable) {
 712                 if(iofd->read_lines) {
 713                     int bytes;
 714
 715                     if(iofd->ssl_active)
 716                         bytes = iohandler_ssl_read(iofd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos);
 717                     else {
 718                         if(iofd->type == IOTYPE_STDIN)
 719                             #ifdef WIN32
 720                             bytes = readable;
 721                             #else
 722                             bytes = read(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos);
 723                             #endif
 724                         else
 725                             bytes = recv(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos, 0);
 726                     }
 727                     if(bytes <= 0) {
 728                         if (errno != EAGAIN || errno != EWOULDBLOCK) {
 729                             iofd->state = IO_CLOSED;
 730                                                         engine->update(iofd);
 731                             callback_event.type = IOEVENT_CLOSED;
 732                             callback_event.data.errid = errno;
 733                         }
 734                     } else {
 735                         int i, used_bytes = 0;
 736                         iohandler_log(IOLOG_DEBUG, "received %d bytes (fd: %d). readbuf position: %d", bytes, iofd->fd, iofd->readbuf.bufpos);
 737                         iofd->readbuf.bufpos += bytes;
 738                         callback_event.type = IOEVENT_RECV;
 739                         for(i = 0; i < iofd->readbuf.bufpos; i++) {
 740                             if(iofd->readbuf.buffer[i] == '\r' && iofd->readbuf.buffer[i+1] == '\n')
 741                                 iofd->readbuf.buffer[i] = 0;
 742                             else if(iofd->readbuf.buffer[i] == '\n' || iofd->readbuf.buffer[i] == '\r') {
 743                                 iofd->readbuf.buffer[i] = 0;
 744                                 callback_event.data.recv_str = iofd->readbuf.buffer + used_bytes;
 745                                 iohandler_log(IOLOG_DEBUG, "parsed line (%d bytes): %s", i - used_bytes, iofd->readbuf.buffer + used_bytes);
 746                                 used_bytes = i+1;
 747                                 iohandler_trigger_event(&callback_event);
 748                             } else if(i + 1 - used_bytes >= IO_LINE_LEN) { //512 max
 749                                 iofd->readbuf.buffer[i] = 0;
 750                                 callback_event.data.recv_str = iofd->readbuf.buffer + used_bytes;
 751                                 iohandler_log(IOLOG_DEBUG, "parsed and stripped line (%d bytes): %s", i - used_bytes, iofd->readbuf.buffer + used_bytes);
 752                                 for(; i < iofd->readbuf.bufpos; i++) { //skip the rest of the line
 753                                     if(iofd->readbuf.buffer[i] == '\n' || (iofd->readbuf.buffer[i] == '\r' && iofd->readbuf.buffer[i+1] != '\n')) {
 754                                         break;
 755                                     }
 756                                 }
 757                                 used_bytes = i+1;
 758                                 iohandler_trigger_event(&callback_event);
 759                             }
 760                         }
 761                         if(used_bytes) {
 762                             if(used_bytes == iofd->readbuf.bufpos) {
 763                                 iofd->readbuf.bufpos = 0;
 764                                 iohandler_log(IOLOG_DEBUG, "readbuf fully processed (set buffer position to 0)");
 765                             } else {
 766                                 iohandler_log(IOLOG_DEBUG, "readbuf rest: %d bytes (used %d bytes)", iofd->readbuf.bufpos - used_bytes, used_bytes);
 767                                 memmove(iofd->readbuf.buffer, iofd->readbuf.buffer + used_bytes, iofd->readbuf.bufpos - used_bytes);
 768                                 iofd->readbuf.bufpos -= used_bytes;
 769                             }
 770                         }
 771                         callback_event.type = IOEVENT_IGNORE;
 772                     }
 773                 } else
 774                     callback_event.type = IOEVENT_READABLE;
 775             }
 776             if(writeable) {
 777                 int bytes;
 778                 bytes = iohandler_try_write(iofd);
 779                 if(bytes < 0) {
 780                     iofd->state = IO_CLOSED;
 781                     engine->update(iofd);
 782                     callback_event.type = IOEVENT_CLOSED;
 783                     callback_event.data.errid = errno;
 784                 }
 785             }
 786             break;
 787     }
 788     if(callback_event.type == IOEVENT_IGNORE && !readable && !writeable)
 789         callback_event.type = IOEVENT_TIMEOUT;
 790     if(callback_event.type != IOEVENT_IGNORE)
 791         iohandler_trigger_event(&callback_event);
 792 }
 793
 794 void iohandler_poll() {
 795     struct timeval timeout;
 796     timeout.tv_sec = IO_MAX_TIMEOUT;
 797     timeout.tv_usec = 0;
 798     iohandler_poll_timeout(timeout);
 799 }
 800
 801 void iohandler_poll_timeout(struct timeval timeout) {
 802     if(engine) {
 803         IOSYNCHRONIZE(io_poll_sync); //quite senceless multithread support... better support will follow
 804         engine->loop(&timeout);
 805         IODESYNCHRONIZE(io_poll_sync);
 806     }
 807 }
 808
 809 //debugging functions
 810 char *iohandler_iotype_name(enum IOType type) {
 811     switch(type) {
 812         case IOTYPE_UNKNOWN:
 813             return "IOTYPE_UNKNOWN";
 814         case IOTYPE_SERVER:
 815             return "IOTYPE_SERVER";
 816         case IOTYPE_CLIENT:
 817             return "IOTYPE_CLIENT";
 818         case IOTYPE_STDIN:
 819             return "IOTYPE_STDIN";
 820         case IOTYPE_TIMER:
 821             return "IOTYPE_TIMER";
 822         default:
 823             return "(UNDEFINED)";
 824     }
 825 }
 826
 827 char *iohandler_iostatus_name(enum IOStatus status) {
 828     switch(status) {
 829         case IO_CLOSED:
 830             return "IO_CLOSED";
 831         case IO_LISTENING:
 832             return "IO_LISTENING";
 833         case IO_CONNECTING:
 834             return "IO_CONNECTING";
 835         case IO_CONNECTED:
 836             return "IO_CONNECTED";
 837         case IO_SSLWAIT:
 838             return "IO_SSLWAIT";
 839         default:
 840             return "(UNDEFINED)";
 841     }
 842 }
 843
 844 char *iohandler_ioeventtype_name(enum IOEventType type) {
 845     switch(type) {
 846         case IOEVENT_IGNORE:
 847             return "IOEVENT_IGNORE";
 848         case IOEVENT_READABLE:
 849             return "IOEVENT_READABLE";
 850         case IOEVENT_RECV:
 851             return "IOEVENT_RECV";
 852         case IOEVENT_CONNECTED:
 853             return "IOEVENT_CONNECTED";
 854         case IOEVENT_NOTCONNECTED:
 855             return "IOEVENT_NOTCONNECTED";
 856         case IOEVENT_CLOSED:
 857             return "IOEVENT_CLOSED";
 858         case IOEVENT_ACCEPT:
 859             return "IOEVENT_ACCEPT";
 860         case IOEVENT_SSLACCEPT:
 861             return "IOEVENT_SSLACCEPT";
 862         case IOEVENT_TIMEOUT:
 863             return "IOEVENT_TIMEOUT";
 864         default:
 865             return "(UNDEFINED)";
 866     }
 867 }