epoll_writev_4threads

#define _GNU_SOURCE

#include <errno.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <sys/uio.h>
#include <arpa/inet.h>
#include <sys/mman.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>

#define THREADS 4
#define PORT 12345

#define NUM_EVENTS 128
#define NUM_RESPONSES 128
#define READ_BUFFER_SIZE (32 * 1024)
#define HTTP_RESPONSE "HTTP/1.1 200 Ok\r\nContent-Length: 11\r\n\r\nHello World"

static struct iovec responses[NUM_RESPONSES];

int find_clrf(const char* buffer, int len) {
    const unsigned int CLRF = '\r' | ('\n' << 8) | ('\r' << 16) | ('\n' << 24);
    int offset = 0;
    while (len >= sizeof(unsigned int)) {
        if (*((unsigned int*) &buffer[offset]) == CLRF)
            return offset + 4;
        offset++;
        len--;
    }
    return -1;
}

void* worker(void* arg) {
    int client;
    struct epoll_event event;
    char buffer[READ_BUFFER_SIZE];

    int epoll = ((int*) arg)[0];
    int server = ((int*) arg)[1];
    struct epoll_event events[NUM_EVENTS];

    while (1) {
        int n = epoll_wait(epoll, events, NUM_EVENTS, -1);
        while (n--) {
            struct epoll_event* ev = &events[n];
            if (ev->events & (EPOLLERR | EPOLLHUP)) {
                close(ev->data.fd);

            } else if (ev->data.fd == server) {
                while ((client = accept4(server, NULL, NULL, SOCK_NONBLOCK)) != -1) {
                    int enable = 1;
                    setsockopt(client, IPPROTO_TCP, TCP_NODELAY, &enable, sizeof(enable));
                    event.data.fd = client;
                    event.events = EPOLLIN | EPOLLONESHOT;
                    epoll_ctl(epoll, EPOLL_CTL_ADD, client, &event);
                }
                event.data.fd = server;
                event.events = EPOLLIN | EPOLLONESHOT;
                epoll_ctl(epoll, EPOLL_CTL_MOD, server, &event);

            } else {
                int buf_len, offset = 0, num_requests = 0;
                while ((buf_len = read(ev->data.fd, buffer + offset, READ_BUFFER_SIZE)) > 0) {
                    int clrf_offset = 0;
                    while ((clrf_offset = find_clrf(buffer + clrf_offset + offset, buf_len)) != -1)
                        num_requests++;
                    offset += buf_len;
                }

                writev(ev->data.fd, responses, num_requests);
                if (offset == 0 || (buf_len <= 0 && errno != EAGAIN)) {
                    close(ev->data.fd);
                } else {
                    event.data.fd = ev->data.fd;
                    event.events = EPOLLIN | EPOLLONESHOT;
                    epoll_ctl(epoll, EPOLL_CTL_MOD, event.data.fd, &event);
                }
            }
        }
    }

    return NULL;
}

int main() {
    struct sockaddr_in server_addr;
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(PORT);
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);

    int server = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
    bind(server, (struct sockaddr*)&server_addr, sizeof(server_addr));
    listen(server, SOMAXCONN);

    struct epoll_event event;
    event.data.fd = server;
    event.events = EPOLLIN | EPOLLONESHOT;
    int epoll = epoll_create1(0);
    epoll_ctl(epoll, EPOLL_CTL_ADD, server, &event);

    for (int i = 0; i < NUM_RESPONSES; i++) {
        responses[i].iov_base = HTTP_RESPONSE;
        responses[i].iov_len = sizeof(HTTP_RESPONSE) - 1;
    }

    pthread_t thread;
    cpu_set_t cpuset;
    int fds[2] = { epoll, server };
    for (int i = 0; i < THREADS - 1; i++) {
        CPU_ZERO(&cpuset);
        CPU_SET(i, &cpuset);
        pthread_create(&thread, NULL, worker, &fds);
        //pthread_setaffinity_np(thread, sizeof(cpu_set_t), &cpuset);
    }

    printf("Listener staretd on :%d\n", PORT);
    worker(&fds);
}