Implement epoll main loop in Server, and parse header.

[cubemap] / server.cpp

#include <stdio.h>
#include <string.h>
#include <stdint.h>
#include <assert.h>
#include <arpa/inet.h>
#include <curl/curl.h>
#include <sys/socket.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <sys/epoll.h>
#include <errno.h>
#include <vector>
#include <string>
#include <map>

#include "metacube.h"
#include "server.h"
#include "mutexlock.h"

using namespace std;

Server::Server()
{
        pthread_mutex_init(&mutex, NULL);

        epoll_fd = epoll_create(1024);  // Size argument is ignored.
        if (epoll_fd == -1) {
                perror("epoll_fd");
                exit(1);
        }
}

void Server::run()
{
        pthread_t thread;
        pthread_create(&thread, NULL, Server::do_work_thunk, this);
}

void *Server::do_work_thunk(void *arg)
{
        Server *server = static_cast<Server *>(arg);
        server->do_work();
        return NULL;
}

void Server::do_work()
{
        for ( ;; ) {
                int nfds = epoll_wait(epoll_fd, events, EPOLL_MAX_EVENTS, EPOLL_TIMEOUT_MS);

                MutexLock lock(&mutex);  // We release the mutex between iterations.
                if (nfds == -1) {
                        perror("epoll_wait");
                        exit(1);
                }
                
                for (int i = 0; i < nfds; ++i) {
                        int fd = events[i].data.fd;
                        assert(clients.count(fd) != 0);
                        Client *client = &clients[fd];

                        if (events[i].events & (EPOLLERR | EPOLLRDHUP | EPOLLHUP)) {
                                close_client(client);
                                continue;
                        }

                        process_client(client);
                }
        }
}
        
void Server::add_client(int sock)
{
        MutexLock lock(&mutex);
        Client new_client;
        new_client.sock = sock;
        new_client.client_request.reserve(1024);
        new_client.state = Client::READING_REQUEST;
        new_client.header_bytes_sent = 0;
        new_client.bytes_sent = 0;

        clients.insert(make_pair(sock, new_client));

        // Start listening on data from this socket.
        epoll_event ev;
        ev.events = EPOLLIN | EPOLLRDHUP;
        ev.data.fd = sock;
        if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev) == -1) {
                perror("epoll_ctl(EPOLL_CTL_ADD)");
                exit(1);
        }
}
        
void Server::add_stream(const string &stream_id)
{
        // TODO
}
        
void Server::set_header(const string &stream_id, const string &header)
{
        // TODO
        printf("got header! %lu bytes\n", header.size());
}
        
void Server::add_data(const string &stream_id, const char *data, size_t bytes)
{
        // TODO
}
        
void Server::process_client(Client *client)
{
        switch (client->state) {
        case Client::READING_REQUEST: {
                // Try to read more of the request.
                char buf[1024];
                int ret = read(client->sock, buf, sizeof(buf));
                if (ret == -1) {
                        perror("read");
                        close_client(client);
                        return;
                }
                if (ret == 0) {
                        // No data? This really means that we were triggered for something else than
                        // POLLIN (which suggests a logic error in epoll).
                        fprintf(stderr, "WARNING: fd %d returned unexpectedly 0 bytes!\n", client->sock);
                        close_client(client);
                        return;
                }

                // Guard against overlong requests gobbling up all of our space.
                if (client->client_request.size() + ret > MAX_CLIENT_REQUEST) {
                        fprintf(stderr, "WARNING: fd %d sent overlong request!\n", client->sock);
                        close_client(client);
                        return;
                }       

                // See if we have \r\n\r\n anywhere in the request. We start three bytes
                // before what we just appended, in case we just got the final character.
                size_t existing_req_bytes = client->client_request.size();
                client->client_request.append(string(buf, buf + ret));
        
                size_t start_at = (existing_req_bytes >= 3 ? existing_req_bytes - 3 : 0);
                const char *ptr = reinterpret_cast<char *>(
                        memmem(client->client_request.data() + start_at, client->client_request.size() - start_at,
                               "\r\n\r\n", 4));
                if (ptr == NULL) {
                        // OK, we don't have the entire header yet. Fine; we'll get it later.
                        return;
                }

                if (ptr != client->client_request.data() + client->client_request.size() - 4) {
                        fprintf(stderr, "WARNING: fd %d had junk data after request!\n", client->sock);
                        close_client(client);
                        return;
                }

                client->state = Client::SENDING_HEADER;

                epoll_event ev;
                ev.events = EPOLLOUT | EPOLLRDHUP;
                ev.data.fd = client->sock;

                if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) {
                        perror("epoll_ctl(EPOLL_CTL_MOD)");
                        exit(1);
                }
                break;
        }
        default:
                // TODO
                assert(false);
        }
}
        
void Server::close_client(Client *client)
{
        if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client->sock, NULL) == -1) {
                perror("epoll_ctl(EPOLL_CTL_DEL)");
                exit(1);
        }
        
        // Bye-bye!
        close(client->sock);
        clients.erase(client->sock);
}