ssize_t mk_http_send_file(struct client_session *cs, struct session_request *sr)
{
    ssize_t nbytes = 0;

    nbytes = mk_socket_send_file(cs->socket, sr->fd_file,
                                 &sr->bytes_offset, sr->bytes_to_send);

    if (nbytes > 0) {
        sr->bytes_to_send -= nbytes;
        if (sr->bytes_offset == nbytes) {
            mk_server_cork_flag(cs->socket, TCP_CORK_OFF);
        }
    }

    /*
     * In some circumstances when writing data, the connection can get broken.
     * So, we must be aware of that.
     *
     * Also, if for some reason the file that is being served changes it's size
     * we can get a zero bytes send as return value. We need to validate the
     * return values <= zero
     */
    if (mk_unlikely(nbytes <= 0)) {
        if (errno != EAGAIN) {
            MK_TRACE("sendfile() = %i", nbytes);
            return EXIT_ABORT;
        }
        MK_TRACE("sendfile() = EAGAIN");
    }

    return sr->bytes_to_send;
}

int mk_epoll_change_mode(int efd, int fd, int mode, int behavior)
{
    int ret;
    struct epoll_event event = {0, {0}};

    event.events = EPOLLERR | EPOLLHUP;
    event.data.fd = fd;

    if (behavior == MK_EPOLL_EDGE_TRIGGERED) {
        event.events |= EPOLLET;
    }

    switch (mode) {
    case MK_EPOLL_READ:
        MK_TRACE("[FD %i] EPoll changing mode to READ", fd);
        event.events |= EPOLLIN;
        break;
    case MK_EPOLL_WRITE:
        MK_TRACE("[FD %i] EPoll changing mode to WRITE", fd);
        event.events |= EPOLLOUT;
        break;
    case MK_EPOLL_RW:
        MK_TRACE("[FD %i] Epoll changing mode to READ/WRITE", fd);
        event.events |= EPOLLIN | EPOLLOUT;
        break;
    }

    ret = epoll_ctl(efd, EPOLL_CTL_MOD, fd, &event);
    if (ret < 0) {
        perror("epoll_ctl");
    }
    return ret;
}

int mk_conn_read(int socket)
{
    int ret;
    struct client_session *cs;
    struct sched_list_node *sched;

    MK_TRACE("[FD %i] Connection Handler / read", socket);

    /* Plugin hook */
    ret = mk_plugin_event_read(socket);

    switch(ret) {
    case MK_PLUGIN_RET_EVENT_OWNED:
        return MK_PLUGIN_RET_CONTINUE;
    case MK_PLUGIN_RET_EVENT_CLOSE:
        return -1;
    case MK_PLUGIN_RET_EVENT_CONTINUE:
        break; /* just return controller to invoker */
    }

    sched = mk_sched_get_thread_conf();
    cs = mk_session_get(socket);
    if (!cs) {
        /* Note: Linux don't set TCP_NODELAY socket flag by default */
        if (mk_socket_set_tcp_nodelay(socket) != 0) {
            mk_warn("TCP_NODELAY failed");
        }

        /* Create session for the client */
        MK_TRACE("[FD %i] Create session", socket);
        cs = mk_session_create(socket, sched);
        if (!cs) {
            return -1;
        }
    }

    /* Read incomming data */
    ret = mk_handler_read(socket, cs);
    if (ret > 0) {
        if (mk_http_pending_request(cs) == 0) {
            mk_epoll_change_mode(sched->epoll_fd,
                                 socket, MK_EPOLL_WRITE, MK_EPOLL_LEVEL_TRIGGERED);
        }
        else if (cs->body_length + 1 >= config->max_request_size) {
            /*
             * Request is incomplete and our buffer is full,
             * close connection
             */
            mk_session_remove(socket);
            return -1;
        }
        else {
            MK_TRACE("[FD %i] waiting for pending data", socket);
        }
    }

    return ret;
}

int mk_utils_print_errno(int n)
{
        switch(n) {
        case EAGAIN:
            MK_TRACE("EAGAIN");
            return -1;
        case EBADF:
            MK_TRACE("EBADF");
            return -1;
        case EFAULT:
            MK_TRACE("EFAULT");
            return -1;
        case EFBIG:
            MK_TRACE("EFBIG");
            return -1;
        case EINTR:
            MK_TRACE("EINTR");
            return -1;
        case EINVAL:
            MK_TRACE("EINVAL");
            return -1;
        case EPIPE:
            MK_TRACE("EPIPE");
            return -1;
        default:
            MK_TRACE("DONT KNOW");
            return 0;
        }

        return 0;
}

void *mk_epoll_init(int efd, mk_epoll_handlers * handler, int max_events)
{
    int i, fd, ret = -1;
    int num_fds;
    int fds_timeout;

    struct epoll_event *events;
    struct sched_list_node *sched;

    /* Get thread conf */
    sched = mk_sched_get_thread_conf();

    fds_timeout = log_current_utime + config->timeout;
    events = mk_mem_malloc_z(max_events*sizeof(struct epoll_event));

    while (1) {
        ret = -1;
        num_fds = epoll_wait(efd, events, max_events, MK_EPOLL_WAIT_TIMEOUT);

        for (i = 0; i < num_fds; i++) {
            fd = events[i].data.fd;

            if (events[i].events & EPOLLIN) {
                MK_TRACE("[FD %i] EPoll Event READ", fd);
                ret = (*handler->read) (fd);
            }
            else if (events[i].events & EPOLLOUT) {
                MK_TRACE("[FD %i] EPoll Event WRITE", fd);
                ret = (*handler->write) (fd);
            }
            else if (events[i].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP)) {
                MK_TRACE("[FD %i] EPoll Event EPOLLHUP/EPOLLER", fd);
                ret = (*handler->error) (fd);
            }

            if (ret < 0) {
                MK_TRACE("[FD %i] Epoll Event FORCE CLOSE | ret = %i", fd, ret);
                (*handler->close) (fd);
            }
        }

        /* Check timeouts and update next one */
        if (log_current_utime >= fds_timeout) {
            mk_sched_check_timeouts(sched);
            fds_timeout = log_current_utime + config->timeout;
        }
    }
}

int mk_socket_ip_str(int socket_fd, char **buf, int size, unsigned long *len)
{
    int ret;
    struct sockaddr_storage addr;
    socklen_t s_len = sizeof(addr);

    ret = getpeername(socket_fd, (struct sockaddr *) &addr, &s_len);

    if (mk_unlikely(ret == -1)) {
        MK_TRACE("[FD %i] Can't get addr for this socket", socket_fd);
        return -1;
    }

    errno = 0;

    if(addr.ss_family == AF_INET) {
        if((inet_ntop(AF_INET, &((struct sockaddr_in *)&addr)->sin_addr,
                      *buf, size)) == NULL) {
            mk_warn("mk_socket_ip_str: Can't get the IP text form (%i)", errno);
            return -1;
        }
    }
    else if(addr.ss_family == AF_INET6) {
        if((inet_ntop(AF_INET6, &((struct sockaddr_in6 *)&addr)->sin6_addr,
                      *buf, size)) == NULL) {
            mk_warn("mk_socket_ip_str: Can't get the IP text form (%i)", errno);
            return -1;
        }
    }

    *len = strlen(*buf);
    return 0;
}

int mk_conn_close(int socket, int event)
{
    MK_TRACE("[FD %i] Connection Handler, closed", socket);

    /*
     * Remove the socket from the scheduler and make sure
     * to disable all notifications.
     */
    mk_sched_drop_connection(socket);

    /*
     * Plugin hook: this is a wrap-workaround to do not
     * break plugins until the whole interface events and
     * return values are re-worked.
     */
    if (event == MK_EP_SOCKET_CLOSED) {
        mk_plugin_event_close(socket);
    }
    else if (event == MK_EP_SOCKET_ERROR) {
        mk_plugin_event_error(socket);
    }
    else if (event == MK_EP_SOCKET_TIMEOUT) {
        mk_plugin_event_timeout(socket);
    }

    return 0;
}

void mk_header_set_http_status(struct mk_http_request *sr, int status)
{
    mk_bug(!sr);
    sr->headers.status = status;

    MK_TRACE("Set HTTP status = %i", status);
}

int mk_conn_error(int socket)
{
    int ret = -1;
    struct client_session *cs;
    struct sched_list_node *sched;

    MK_TRACE("Connection Handler, error on FD %i", socket);

    /* Plugin hook */
    ret = mk_plugin_event_error(socket);
    switch(ret) {
    case MK_PLUGIN_RET_EVENT_OWNED:
        return MK_PLUGIN_RET_CONTINUE;
    case MK_PLUGIN_RET_EVENT_CLOSE:
        return -1;
    case MK_PLUGIN_RET_EVENT_CONTINUE:
        break; /* just return controller to invoker */
    }

    sched = mk_sched_get_thread_conf();
    mk_sched_remove_client(sched, socket);
    cs = mk_session_get(socket);
    if (cs) {
        mk_session_remove(socket);
    }

    return 0;
}

int mk_epoll_add(int efd, int fd, int init_mode, int behavior)
{
    int ret;
    struct epoll_event event = {0, {0}};

    event.data.fd = fd;
    event.events = EPOLLERR | EPOLLHUP | EPOLLRDHUP;

    if (behavior == MK_EPOLL_EDGE_TRIGGERED) {
        event.events |= EPOLLET;
    }

    switch (init_mode) {
    case MK_EPOLL_READ:
        event.events |= EPOLLIN;
        break;
    case MK_EPOLL_WRITE:
        event.events |= EPOLLOUT;
        break;
    case MK_EPOLL_RW:
        event.events |= EPOLLIN | EPOLLOUT;
        break;
    }

    ret = epoll_ctl(efd, EPOLL_CTL_ADD, fd, &event);
    if (ret < 0 && errno != EEXIST) {
        MK_TRACE("[FD %i] epoll_ctl()", fd, strerror(errno));
        return ret;
    }

    return ret;
}

static void mk_socket_safe_event_write(int socket)
{
    struct sched_list_node *sched;

    sched = mk_sched_get_thread_conf();
    MK_TRACE("[FD %i] Safe event write ON", socket);
    mk_event_add(sched->loop, socket, MK_EVENT_WRITE, NULL);
}

static void mk_socket_safe_event_write(int socket)
{
    struct sched_list_node *sched;

    sched = mk_sched_get_thread_conf();
    MK_TRACE("[FD %i] Safe event write ON", socket);
    mk_epoll_change_mode(sched->epoll_fd, socket,
                         MK_EPOLL_WRITE, MK_EPOLL_LEVEL_TRIGGERED);
}

int mk_user_init(struct client_session *cs, struct session_request *sr)
{
    int limit;
    const int offset = 2; /* The user is defined after the '/~' string, so offset = 2 */
    const int user_len = 255;
    char user[user_len], *user_uri;
    struct passwd *s_user;

    if (sr->uri_processed.len <= 2) {
        return -1;
    }

    limit = mk_string_char_search(sr->uri_processed.data + offset, '/',
                                  sr->uri_processed.len);

    if (limit == -1) {
        limit = (sr->uri_processed.len) - offset;
    }

    if (limit + offset >= (user_len)) {
        return -1;
    }

    memcpy(user, sr->uri_processed.data + offset, limit);
    user[limit] = '\0';

    MK_TRACE("user: '%s'", user);

    /* Check system user */
    if ((s_user = getpwnam(user)) == NULL) {
        mk_request_error(MK_CLIENT_NOT_FOUND, cs, sr);
        return -1;
    }

    if (sr->uri_processed.len > (unsigned int) (offset+limit)) {
        user_uri = mk_mem_malloc(sr->uri_processed.len);
        if (!user_uri) {
            return -1;
        }

        memcpy(user_uri,
               sr->uri_processed.data + (offset + limit),
               sr->uri_processed.len - offset - limit);
        user_uri[sr->uri_processed.len - offset - limit] = '\0';

        mk_string_build(&sr->real_path.data, &sr->real_path.len,
                        "%s/%s%s", s_user->pw_dir, config->user_dir, user_uri);
        mk_mem_free(user_uri);
    }
    else {
        mk_string_build(&sr->real_path.data, &sr->real_path.len,
                        "%s/%s", s_user->pw_dir, config->user_dir);
    }

    sr->user_home = MK_TRUE;
    return 0;
}

/*
 * Example from:
 * http://www.baus.net/on-tcp_cork
 */
int mk_socket_set_cork_flag(int fd, int state)
{
    if (config->corking == MK_FALSE) {
        return 0;
    }

    MK_TRACE("Socket, set Cork Flag FD %i to %s", fd, (state ? "ON" : "OFF"));

    return setsockopt(fd, SOL_TCP, TCP_CORK, &state, sizeof(state));
}

static inline size_t channel_write_stream_file(struct mk_channel *channel,
                                               struct mk_stream *stream)
{
    ssize_t bytes = 0;

    MK_TRACE("[CH %i] STREAM_FILE %i, bytes=%lu",
             channel->fd, stream->fd, stream->bytes_total);

    /* Direct write */
    bytes = mk_sched_conn_sendfile(channel,
                                    stream->fd,
                                    &stream->bytes_offset,
                                    stream->bytes_total
                                    );
    MK_TRACE("[CH=%d] [FD=%i] WRITE STREAM FILE: %lu bytes",
             channel->fd, stream->fd, bytes);

    return bytes;
}

/*
 * Example from:
 * http://www.baus.net/on-tcp_cork
 */
int mk_socket_set_cork_flag(int fd, int state)
{
    MK_TRACE("Socket, set Cork Flag FD %i to %s", fd, (state ? "ON" : "OFF"));

#if defined (TCP_CORK)
    return setsockopt(fd, SOL_TCP, TCP_CORK, &state, sizeof(state));
#elif defined (TCP_NOPUSH)
    return setsockopt(fd, SOL_SOCKET, TCP_NOPUSH, &state, sizeof(state));
#endif
}

ssize_t mk_iov_send(int fd, struct mk_iov *mk_io)
{
    ssize_t n = writev(fd, mk_io->io, mk_io->iov_idx);
    if (mk_unlikely(n < 0)) {
        MK_TRACE( "[FD %i] writev() '%s'", fd, strerror(errno));
        return -1;
    }

    return n;
}

int mk_socket_set_nonblocking(int sockfd)
{

    MK_TRACE("Socket, set FD %i to non-blocking", sockfd);

    if (fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, 0) | O_NONBLOCK | O_CLOEXEC) == -1) {
        mk_err("Can't set to non-blocking mode socket %i", sockfd);
        return -1;
    }
    return 0;
}

void mk_server_loop(int server_fd)
{
    int ret;
    int remote_fd;

    /* Activate TCP_DEFER_ACCEPT */
    if (mk_socket_set_tcp_defer_accept(server_fd) != 0) {
            mk_warn("TCP_DEFER_ACCEPT failed");
    }

    /* Rename worker */
    mk_utils_worker_rename("monkey: server");
    mk_info("HTTP Server started");

    while (1) {
        remote_fd = mk_socket_accept(server_fd);

        if (mk_unlikely(remote_fd == -1)) {
            continue;
        }

#ifdef TRACE
        MK_TRACE("New connection arrived: FD %i", remote_fd);

        int i;
        struct sched_list_node *node;

        node = sched_list;
        for (i=0; i < config->workers; i++) {
            MK_TRACE("Worker Status");
            MK_TRACE(" WID %i / conx = %llu", node[i].idx, node[i].accepted_connections - node[i].closed_connections);
        }
#endif

        /* Assign socket to worker thread */
        ret = mk_sched_add_client(remote_fd);
        if (ret == -1) {
            close(remote_fd);
        }
    }
}

int mk_epoll_del(int efd, int fd)
{
    int ret;

    ret = epoll_ctl(efd, EPOLL_CTL_DEL, fd, NULL);
    MK_TRACE("Epoll, removing fd %i from efd %i", fd, efd);

    if (ret < 0) {
        perror("epoll_ctl");
    }
    return ret;
}