-void iohandler_log(enum IOLogType type, char *text, ...) {
- va_list arg_list;
- char logBuf[MAXLOG+1];
- int pos;
- logBuf[0] = '\0';
- va_start(arg_list, text);
- pos = vsnprintf(logBuf, MAXLOG - 1, text, arg_list);
- va_end(arg_list);
- if (pos < 0 || pos > (MAXLOG - 1)) pos = MAXLOG - 1;
- logBuf[pos] = '\n';
- logBuf[pos+1] = '\0';
-
- if(iolog_backend)
- iolog_backend(type, logBuf);
-}
-
-/* IO Engines */
-extern struct IOEngine engine_select; /* select system call (should always be useable) */
-extern struct IOEngine engine_kevent;
-extern struct IOEngine engine_epoll;
-extern struct IOEngine engine_win32;
-
-struct IOEngine *engine = NULL;
-
-static void iohandler_init_engine() {
- if(engine) return;
- IOTHREAD_MUTEX_INIT(io_thread_sync);
- IOTHREAD_MUTEX_INIT(io_poll_sync);
-
- //try other engines
- if(!engine && engine_kevent.init && engine_kevent.init())
- engine = &engine_kevent;
- if(!engine && engine_epoll.init && engine_epoll.init())
- engine = &engine_epoll;
- if(!engine && engine_win32.init && engine_win32.init())
- engine = &engine_win32;
-
- if (!engine) {
- if(engine_select.init())
- engine = &engine_select;
- else {
- iohandler_log(IOLOG_FATAL, "found no useable IO engine");
- return;
- }
- }
- iohandler_log(IOLOG_DEBUG, "using %s IO engine", engine->name);
- iohandler_ssl_init();
-}
-
-static void iohandler_append(struct IODescriptor *descriptor) {
- IOSYNCHRONIZE(io_thread_sync);
- struct timeval *timeout = ((descriptor->timeout.tv_sec || descriptor->timeout.tv_usec) ? &descriptor->timeout : NULL);
- if(timeout) {
- struct IODescriptor *iofd;
- int set_priority = 1;
- descriptor->timeout = *timeout;
- if(timer_priority)
- iofd = timer_priority;
- else
- iofd = first_descriptor;
- if(iofd) {
- for(;;iofd = iofd->next) {
- if(timeval_is_smaler(timeout, (&iofd->timeout))) {
- descriptor->prev = iofd->prev;
- descriptor->next = iofd;
- if(iofd->prev)
- iofd->prev->next = descriptor;
- iofd->prev = descriptor;
- if(set_priority)
- timer_priority = descriptor;
- break;
- }
- if(iofd == timer_priority)
- set_priority = 0;
- if(iofd->next == NULL) {
- descriptor->next = NULL;
- descriptor->prev = iofd;
- iofd->next = descriptor;
- if(set_priority)
- timer_priority = descriptor;
- break;
- }
- }
- } else {
- descriptor->prev = NULL;
- descriptor->next = NULL;
- first_descriptor = descriptor;
- timer_priority = descriptor;
- }
-
- } else {
- descriptor->prev = NULL;
- descriptor->next = first_descriptor;
- if(first_descriptor)
- first_descriptor->prev = descriptor;
- first_descriptor = descriptor;
- }
- IODESYNCHRONIZE(io_thread_sync);
-}
-
-static void iohandler_remove(struct IODescriptor *descriptor, int engine_remove) {
- //remove IODescriptor from the list
- IOSYNCHRONIZE(io_thread_sync);
- if(descriptor->prev)
- descriptor->prev->next = descriptor->next;
- else
- first_descriptor = descriptor->next;
- if(descriptor->next)
- descriptor->next->prev = descriptor->prev;
- if(descriptor == timer_priority)
- timer_priority = descriptor->next;
-
- if(engine_remove)
- engine->remove(descriptor);
- if(descriptor->readbuf.buffer)
- free(descriptor->readbuf.buffer);
- if(descriptor->writebuf.buffer)
- free(descriptor->writebuf.buffer);
- iohandler_log(IOLOG_DEBUG, "removed IODescriptor (%d) of type `%s`", descriptor->fd, iohandler_iotype_name(descriptor->type));
- free(descriptor);
- IODESYNCHRONIZE(io_thread_sync);
-}
-
-struct IODescriptor *iohandler_add(int sockfd, enum IOType type, struct timeval *timeout, iohandler_callback *callback) {
- //just add a new IODescriptor
- struct IODescriptor *descriptor = calloc(1, sizeof(*descriptor));
- if(!descriptor) {
- iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
- return NULL;
- }
- descriptor->fd = (type == IOTYPE_STDIN ? fileno(stdin) : sockfd);
- descriptor->type = type;
- descriptor->state = (type == IOTYPE_STDIN ? IO_CONNECTED : IO_CLOSED);
- descriptor->callback = callback;
- if(timeout)
- descriptor->timeout = *timeout;
- if(type != IOTYPE_TIMER) {
- descriptor->readbuf.buffer = malloc(IO_READ_BUFLEN + 2);
- descriptor->readbuf.bufpos = 0;
- descriptor->readbuf.buflen = IO_READ_BUFLEN;
- descriptor->writebuf.buffer = malloc(IO_READ_BUFLEN + 2);
- descriptor->writebuf.bufpos = 0;
- descriptor->writebuf.buflen = IO_READ_BUFLEN;
- }
-
- if(!engine) {
- iohandler_init_engine();
- if(!engine) {
- return NULL;
- }
- }
- engine->add(descriptor);
-
- //add IODescriptor to the list
- iohandler_append(descriptor);
-
- iohandler_log(IOLOG_DEBUG, "added custom socket descriptor (%d) as type `%s`", sockfd, iohandler_iotype_name(type));
- return descriptor;
-}
-
-void iohandler_set_timeout(struct IODescriptor *descriptor, struct timeval *timeout) {
- if(descriptor->prev)
- descriptor->prev->next = descriptor->next;
- else
- first_descriptor = descriptor->next;
- if(descriptor->next)
- descriptor->next->prev = descriptor->prev;
- if(descriptor == timer_priority)
- timer_priority = descriptor->next;
- if(timeout)
- descriptor->timeout = *timeout;
- else {
- descriptor->timeout.tv_sec = 0;
- descriptor->timeout.tv_usec = 0;
- }
- iohandler_append(descriptor);
-}
-
-static void iohandler_increase_iobuf(struct IOBuffer *iobuf, size_t required) {
- if(iobuf->buflen >= required) return;
- char *new_buf = realloc(iobuf->buffer, required + 2);
- if(new_buf) {
- iobuf->buffer = new_buf;
- iobuf->buflen = required;
- }
-}
-
-struct IODescriptor *iohandler_timer(struct timeval timeout, iohandler_callback *callback) {
- struct IODescriptor *descriptor;
- descriptor = iohandler_add(-1, IOTYPE_TIMER, &timeout, callback);
- if(!descriptor) {
- iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
- return NULL;
- }
- iohandler_log(IOLOG_DEBUG, "added timer descriptor (sec: %d; usec: %d)", timeout.tv_sec, timeout.tv_usec);
- return descriptor;
-}
-
-struct IODescriptor *iohandler_connect(const char *hostname, unsigned int port, int ssl, const char *bindhost, iohandler_callback *callback) {
- return iohandler_connect_flags(hostname, port, ssl, bindhost, callback, IOHANDLER_CONNECT_IPV4 | IOHANDLER_CONNECT_IPV6);
-}
-
-struct IODescriptor *iohandler_connect_flags(const char *hostname, unsigned int port, int ssl, const char *bindhost, iohandler_callback *callback, int flags) {
- //non-blocking connect
- int sockfd, result;
- struct addrinfo hints, *res;
- struct sockaddr_in *ip4 = NULL;
- struct sockaddr_in6 *ip6 = NULL;
- size_t dstaddrlen;
- struct sockaddr *dstaddr = NULL;
- struct IODescriptor *descriptor;
-
- if(!engine) {
- iohandler_init_engine();
- if(!engine) return NULL;
- }
- memset (&hints, 0, sizeof (hints));
- hints.ai_family = PF_UNSPEC;
- hints.ai_socktype = SOCK_STREAM;
- hints.ai_flags |= AI_CANONNAME;
- if ((result = getaddrinfo (hostname, NULL, &hints, &res))) {
- iohandler_log(IOLOG_ERROR, "could not resolve %s to an IP address (%d)", hostname, result);
- return NULL;
- }
- while (res) {
- switch (res->ai_family) {
- case AF_INET:
- ip4 = (struct sockaddr_in *) res->ai_addr;
- break;
- case AF_INET6:
- ip6 = (struct sockaddr_in6 *) res->ai_addr;
- break;
- }
- res = res->ai_next;
- freeaddrinfo(res);
- }
-
- if(ip6 && (flags & IOHANDLER_CONNECT_IPV6)) {
- sockfd = socket(AF_INET6, SOCK_STREAM, 0);
- if(sockfd == -1) {
- iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
- return NULL;
- }
-
- ip6->sin6_family = AF_INET6;
- ip6->sin6_port = htons(port);
-
- struct sockaddr_in6 *ip6vhost = NULL;
- if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
- while (res) {
- switch (res->ai_family) {
- case AF_INET6:
- ip6vhost = (struct sockaddr_in6 *) res->ai_addr;
- break;
- }
- res = res->ai_next;
- freeaddrinfo(res);
- }
- }
- if(ip6vhost) {
- ip6vhost->sin6_family = AF_INET6;
- ip6vhost->sin6_port = htons(0);
- bind(sockfd, (struct sockaddr*)ip6vhost, sizeof(*ip6vhost));
- }
- dstaddr = (struct sockaddr*)ip6;
- dstaddrlen = sizeof(*ip6);
- } else if(ip4 && (flags & IOHANDLER_CONNECT_IPV4)) {
- sockfd = socket(AF_INET, SOCK_STREAM, 0);
- if(sockfd == -1) {
- iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
- return NULL;
- }
-
- ip4->sin_family = AF_INET;
- ip4->sin_port = htons(port);
-
- struct sockaddr_in *ip4vhost = NULL;
- if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
- while (res) {
- switch (res->ai_family) {
- case AF_INET:
- ip4vhost = (struct sockaddr_in *) res->ai_addr;
- break;
- }
- res = res->ai_next;
- freeaddrinfo(res);
- }
- }
- if(ip4vhost) {
- ip4vhost->sin_family = AF_INET;
- ip4vhost->sin_port = htons(0);
- bind(sockfd, (struct sockaddr*)ip4vhost, sizeof(*ip4vhost));
- }
- dstaddr = (struct sockaddr*)ip4;
- dstaddrlen = sizeof(*ip4);
- } else
- return NULL;
- //prevent SIGPIPE
- #ifndef WIN32
- #if defined(SO_NOSIGPIPE)
- {
- int set = 1;
- setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
- }
- #else
- signal(SIGPIPE, SIG_IGN);
- #endif
- #endif
- //make sockfd unblocking
- #if defined(F_GETFL)
- {
- int flags;
- flags = fcntl(sockfd, F_GETFL);
- fcntl(sockfd, F_SETFL, flags|O_NONBLOCK);
- flags = fcntl(sockfd, F_GETFD);
- fcntl(sockfd, F_SETFD, flags|FD_CLOEXEC);
- }
- #else
- /* I hope you're using the Win32 backend or something else that
- * automatically marks the file descriptor non-blocking...
- */
- #endif
- descriptor = iohandler_add(sockfd, IOTYPE_CLIENT, NULL, callback);
- if(!descriptor) {
- close(sockfd);
- return NULL;
- }
- connect(sockfd, dstaddr, dstaddrlen); //returns EINPROGRESS here (nonblocking)
- descriptor->state = IO_CONNECTING;
- descriptor->ssl = (ssl ? 1 : 0);
- descriptor->read_lines = 1;
- engine->update(descriptor);
- iohandler_log(IOLOG_DEBUG, "added client socket (%d) connecting to %s:%d", sockfd, hostname, port);
- return descriptor;
-}