1 /* IOHandler.c - IOMultiplexer
2 * Copyright (C) 2012 Philipp Kreil (pk910)
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.
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.
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/>.
17 #include "IOHandler.h"
19 #include "IOHandler_SSL.h"
24 #define _WIN32_WINNT 0x501
34 #include <sys/socket.h>
35 #include <netinet/in.h>
36 #include <netinet/tcp.h>
41 #define EWOULDBLOCK EAGAIN
45 iohandler_log_callback *iolog_backend = NULL;
47 struct IODescriptor *first_descriptor = NULL;
48 struct IODescriptor *timer_priority = NULL;
51 static pthread_mutex_t io_thread_sync;
52 static pthread_mutex_t io_poll_sync;
55 void iohandler_log(enum IOLogType type, char *text, ...) {
57 char logBuf[MAXLOG+1];
60 va_start(arg_list, text);
61 pos = vsnprintf(logBuf, MAXLOG - 1, text, arg_list);
63 if (pos < 0 || pos > (MAXLOG - 1)) pos = MAXLOG - 1;
68 iolog_backend(type, logBuf);
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;
77 int iohandler_settings = 0;
78 struct IOEngine *engine = NULL;
80 void iohandler_set(int setting, int value) {
82 iohandler_settings |= setting;
84 iohandler_settings &= ~setting;
87 static void iohandler_init_engine() {
89 IOTHREAD_MUTEX_INIT(io_thread_sync);
90 IOTHREAD_MUTEX_INIT(io_poll_sync);
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;
103 if(engine_select.init())
104 engine = &engine_select;
106 iohandler_log(IOLOG_FATAL, "found no useable IO engine");
110 iohandler_log(IOLOG_DEBUG, "using %s IO engine", engine->name);
111 iohandler_ssl_init();
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);
118 struct IODescriptor *iofd;
119 int set_priority = 1;
120 descriptor->timeout = *timeout;
122 iofd = timer_priority;
124 iofd = first_descriptor;
126 for(;;iofd = iofd->next) {
127 if(timeval_is_smaler(timeout, (&iofd->timeout))) {
128 descriptor->prev = iofd->prev;
129 descriptor->next = iofd;
131 iofd->prev->next = descriptor;
132 iofd->prev = descriptor;
134 timer_priority = descriptor;
137 if(iofd == timer_priority)
139 if(iofd->next == NULL) {
140 descriptor->next = NULL;
141 descriptor->prev = iofd;
142 iofd->next = descriptor;
144 timer_priority = descriptor;
149 descriptor->prev = NULL;
150 descriptor->next = NULL;
151 first_descriptor = descriptor;
152 timer_priority = descriptor;
156 descriptor->prev = NULL;
157 descriptor->next = first_descriptor;
159 first_descriptor->prev = descriptor;
160 first_descriptor = descriptor;
162 IODESYNCHRONIZE(io_thread_sync);
165 static void iohandler_remove(struct IODescriptor *descriptor, int engine_remove) {
166 //remove IODescriptor from the list
167 IOSYNCHRONIZE(io_thread_sync);
169 descriptor->prev->next = descriptor->next;
171 first_descriptor = descriptor->next;
173 descriptor->next->prev = descriptor->prev;
174 if(descriptor == timer_priority)
175 timer_priority = descriptor->next;
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));
185 IODESYNCHRONIZE(io_thread_sync);
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));
192 iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
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;
200 descriptor->timeout = *timeout;
201 if(descriptor->timeout.tv_usec > 1000000) {
202 descriptor->timeout.tv_usec -= 1000000;
203 descriptor->timeout.tv_sec++;
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;
216 iohandler_init_engine();
221 engine->add(descriptor);
223 //add IODescriptor to the list
224 iohandler_append(descriptor);
226 iohandler_log(IOLOG_DEBUG, "added custom socket descriptor (%d) as type `%s`", sockfd, iohandler_iotype_name(type));
230 void iohandler_set_timeout(struct IODescriptor *descriptor, struct timeval *timeout) {
232 descriptor->prev->next = descriptor->next;
234 first_descriptor = 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;
247 descriptor->timeout = *timeout;
248 if(descriptor->timeout.tv_usec > 1000000) {
249 descriptor->timeout.tv_usec -= 1000000;
250 descriptor->timeout.tv_sec++;
253 descriptor->timeout.tv_sec = 0;
254 descriptor->timeout.tv_usec = 0;
256 iohandler_append(descriptor);
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);
263 iobuf->buffer = new_buf;
264 iobuf->buflen = required;
268 struct IODescriptor *iohandler_timer(struct timeval timeout, iohandler_callback *callback) {
269 struct IODescriptor *descriptor;
270 descriptor = iohandler_add(-1, IOTYPE_TIMER, &timeout, callback);
272 iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
275 iohandler_log(IOLOG_DEBUG, "added timer descriptor (sec: %d; usec: %d)", timeout.tv_sec, timeout.tv_usec);
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;
289 descriptor = iohandler_add(-1, IOTYPE_TIMER, &timeout, callback);
291 iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
294 descriptor->constant_timeout = msec;
295 iohandler_log(IOLOG_DEBUG, "added timer descriptor (sec: %d; usec: %d)", timeout.tv_sec, timeout.tv_usec);
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);
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
306 struct addrinfo hints, *res;
307 struct sockaddr_in *ip4 = NULL;
308 struct sockaddr_in6 *ip6 = NULL;
310 struct sockaddr *dstaddr = NULL;
311 struct IODescriptor *descriptor;
314 iohandler_init_engine();
315 if(!engine) return NULL;
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);
326 switch (res->ai_family) {
328 ip4 = (struct sockaddr_in *) res->ai_addr;
331 ip6 = (struct sockaddr_in6 *) res->ai_addr;
338 if(ip6 && (flags & IOHANDLER_CONNECT_IPV6)) {
339 sockfd = socket(AF_INET6, SOCK_STREAM, 0);
341 iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
345 ip6->sin6_family = AF_INET6;
346 ip6->sin6_port = htons(port);
348 struct sockaddr_in6 *ip6vhost = NULL;
349 if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
351 switch (res->ai_family) {
353 ip6vhost = (struct sockaddr_in6 *) res->ai_addr;
361 ip6vhost->sin6_family = AF_INET6;
362 ip6vhost->sin6_port = htons(0);
363 bind(sockfd, (struct sockaddr*)ip6vhost, sizeof(*ip6vhost));
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);
370 iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
374 ip4->sin_family = AF_INET;
375 ip4->sin_port = htons(port);
377 struct sockaddr_in *ip4vhost = NULL;
378 if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
380 switch (res->ai_family) {
382 ip4vhost = (struct sockaddr_in *) res->ai_addr;
390 ip4vhost->sin_family = AF_INET;
391 ip4vhost->sin_port = htons(0);
392 bind(sockfd, (struct sockaddr*)ip4vhost, sizeof(*ip4vhost));
394 dstaddr = (struct sockaddr*)ip4;
395 dstaddrlen = sizeof(*ip4);
400 #if defined(SO_NOSIGPIPE)
403 setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
406 signal(SIGPIPE, SIG_IGN);
409 //make sockfd unblocking
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);
419 /* I hope you're using the Win32 backend or something else that
420 * automatically marks the file descriptor non-blocking...
423 descriptor = iohandler_add(sockfd, IOTYPE_CLIENT, NULL, callback);
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);
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);
441 struct IODescriptor *iohandler_listen_flags(const char *hostname, unsigned int port, iohandler_callback *callback, int flags) {
443 struct addrinfo hints, *res;
444 struct sockaddr_in *ip4 = NULL;
445 struct sockaddr_in6 *ip6 = NULL;
446 struct IODescriptor *descriptor;
450 iohandler_init_engine();
451 if(!engine) return NULL;
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)) {
461 switch (res->ai_family) {
463 ip4 = (struct sockaddr_in *) res->ai_addr;
466 ip6 = (struct sockaddr_in6 *) res->ai_addr;
473 if(ip6 && (flags & IOHANDLER_LISTEN_IPV6)) {
474 sockfd = socket(AF_INET6, SOCK_STREAM, 0);
475 if(sockfd == -1) return NULL;
478 setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
480 ip6->sin6_family = AF_INET6;
481 ip6->sin6_port = htons(port);
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;
489 setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
491 ip4->sin_family = AF_INET;
492 ip4->sin_port = htons(port);
494 bind(sockfd, (struct sockaddr*)ip4, sizeof(*ip4));
499 #if defined(SO_NOSIGPIPE)
502 setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
505 signal(SIGPIPE, SIG_IGN);
508 //make sockfd unblocking
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);
518 /* I hope you're using the Win32 backend or something else that
519 * automatically marks the file descriptor non-blocking...
522 descriptor = iohandler_add(sockfd, IOTYPE_SERVER, NULL, callback);
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);
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);
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);
543 iohandler_ssl_listen(descriptor, certfile, keyfile);
544 if(descriptor->sslnode)
549 void iohandler_write(struct IODescriptor *iofd, const char *line) {
550 size_t linelen = strlen(line);
551 iohandler_send(iofd, line, linelen);
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"));
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));
568 memcpy(iofd->writebuf.buffer + iofd->writebuf.bufpos, data, datalen);
569 iofd->writebuf.bufpos += datalen;
570 engine->update(iofd);
573 void iohandler_printf(struct IODescriptor *iofd, const char *text, ...) {
575 char sendBuf[IO_LINE_LEN];
578 va_start(arg_list, text);
579 pos = vsnprintf(sendBuf, IO_LINE_LEN - 2, text, arg_list);
581 if (pos < 0 || pos > (IO_LINE_LEN - 2)) pos = IO_LINE_LEN - 2;
583 sendBuf[pos+1] = '\0';
584 iohandler_send(iofd, sendBuf, pos+1);
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);
592 res = iohandler_ssl_write(iofd, iofd->writebuf.buffer, iofd->writebuf.bufpos);
594 res = send(iofd->fd, iofd->writebuf.buffer, iofd->writebuf.bufpos, 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));
601 iofd->writebuf.bufpos -= res;
602 if(iofd->state != IO_CLOSED)
603 engine->update(iofd);
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
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);
623 engine->remove(iofd);
625 iohandler_try_write(iofd);
629 iohandler_ssl_disconnect(iofd);
630 if(iofd->type == IOTYPE_SERVER || iofd->type == IOTYPE_CLIENT || iofd->type == IOTYPE_STDIN)
632 iohandler_remove(iofd, engine_remove);
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);
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);
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) {
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);
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);
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);
668 if(iofd->type == IOTYPE_TIMER)
669 callback_event.type = IOEVENT_TIMEOUT;
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);
680 callback_event.type = IOEVENT_ACCEPT;
684 if(readable) { //could not connect
685 callback_event.type = IOEVENT_NOTCONNECTED;
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);
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;
705 callback_event.type = IOEVENT_CONNECTED;
706 iofd->state = IO_CONNECTED;
707 engine->update(iofd);
712 if(iofd->read_lines) {
716 bytes = iohandler_ssl_read(iofd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos);
718 if(iofd->type == IOTYPE_STDIN)
722 bytes = read(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos);
725 bytes = recv(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos, 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;
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);
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')) {
758 iohandler_trigger_event(&callback_event);
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)");
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;
771 callback_event.type = IOEVENT_IGNORE;
774 callback_event.type = IOEVENT_READABLE;
778 bytes = iohandler_try_write(iofd);
780 iofd->state = IO_CLOSED;
781 engine->update(iofd);
782 callback_event.type = IOEVENT_CLOSED;
783 callback_event.data.errid = errno;
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);
794 void iohandler_poll() {
795 struct timeval timeout;
796 timeout.tv_sec = IO_MAX_TIMEOUT;
798 iohandler_poll_timeout(timeout);
801 void iohandler_poll_timeout(struct timeval timeout) {
803 IOSYNCHRONIZE(io_poll_sync); //quite senceless multithread support... better support will follow
804 engine->loop(&timeout);
805 IODESYNCHRONIZE(io_poll_sync);
809 //debugging functions
810 char *iohandler_iotype_name(enum IOType type) {
813 return "IOTYPE_UNKNOWN";
815 return "IOTYPE_SERVER";
817 return "IOTYPE_CLIENT";
819 return "IOTYPE_STDIN";
821 return "IOTYPE_TIMER";
823 return "(UNDEFINED)";
827 char *iohandler_iostatus_name(enum IOStatus status) {
832 return "IO_LISTENING";
834 return "IO_CONNECTING";
836 return "IO_CONNECTED";
840 return "(UNDEFINED)";
844 char *iohandler_ioeventtype_name(enum IOEventType type) {
847 return "IOEVENT_IGNORE";
848 case IOEVENT_READABLE:
849 return "IOEVENT_READABLE";
851 return "IOEVENT_RECV";
852 case IOEVENT_CONNECTED:
853 return "IOEVENT_CONNECTED";
854 case IOEVENT_NOTCONNECTED:
855 return "IOEVENT_NOTCONNECTED";
857 return "IOEVENT_CLOSED";
859 return "IOEVENT_ACCEPT";
860 case IOEVENT_SSLACCEPT:
861 return "IOEVENT_SSLACCEPT";
862 case IOEVENT_TIMEOUT:
863 return "IOEVENT_TIMEOUT";
865 return "(UNDEFINED)";