src/main/EventServer.h

Go to the documentation of this file.

#ifndef EVENTSERVER_H_
#define EVENTSERVER_H_

#include <ctime>
#include <iostream>
#include <fstream>
#include <string>
#include <cstring>
#include <vector>
#include <cstdlib>
#include <unistd.h>
#include <ctime>
#include <csignal>
#include <pthread.h>
#include <event.h>
#include <fcntl.h>
#include <cerrno>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <cmath>
#include <cstdlib>
#include <cstdio>
#include <cassert>
#include <byteswap.h>

//definitions of macros used in EventServer.cpp

#define HAVE_CONFIG_H 1

#ifdef HAVE_CONFIG_H
#include "config.h"
#else
#error "No config.h provided!"
#endif

#include "Cache.h"
#include "ServerLogger.h"
#include "functions.h"

#define DEBUG_BINARY 0

#define DEBUG_ASCII 0

#if DEBUG_BINARY >= 20

FILE * f;
#endif

enum State {
        SET_HEADER, SET_VALUE, CAS_HEADER, CAS_VALUE, ADD_HEADER, ADD_VALUE, // those six states have to be first!
        REPLACE_HEADER,
        REPLACE_VALUE,
        APPEND_HEADER,
        APPEND_VALUE,
        PREPEND_HEADER,
        PREPEND_VALUE, // sequence matters up there! *_HEADER -> *_VALUE
        UNDEFINED,
        QUIT,
        VERSION_CMD,
        GET_HEADER,
        GETS_HEADER,
        ERROR_STATE,
        GET_RESPONSE_END,
        STATS_HEADER,
        STATS_RESPONSE,
        INCR_HEADER,
        DECR_HEADER,
        INCR_DECR_RESPONSE
};

#define DEBUG_EV_SERVER 1

const char STORED_MSG[] = "STORED\r\n";
const char NOT_STORED_MSG[] = "NOT_STORED\r\n";
const char EXISTS_MSG[] = "EXISTS\r\n";
const char NOT_FOUND_MSG[] = "NOT_FOUND\r\n";
const char ERROR_MSG[] = "ERROR\r\n";
const char LINE_TERMINATOR[] = "\r\n";
const char REQUEST_TERMINATOR[] = "END\r\n";

#define MAX_HEADER_LENGTH 300

#define REPORT_ERROR { \
        i->cursor = i->offset = i->tokenCount = i->writeOffset = 0; \
        i->sendIt = ERROR_MSG; \
        i->length = sizeof(ERROR_MSG) - 1; \
\
        i->state = ERROR_STATE; \
\
        event_del(&i->readEvent); \
        event_add(&i->writeEvent, &connectionTimeout); \
        /*cout << "error in line " << __LINE__ << endl;*/ \
}

struct Thread {

        int index;
        pthread_t threadID;
        struct event_base * eventBase;
        struct event pipeEvent;
        int in;
        int out;

        uint64_t connections;

        uint64_t totalConnections;

};

struct Info {

        Thread * t;

        struct event readEvent;
        struct event writeEvent;

        int clientSocket;
        char buffer[2 * Cache::MAX_OBJECT_SIZE];

        char sendBuffer[Cache::MAX_OBJECT_SIZE + MAX_HEADER_LENGTH + sizeof(LINE_TERMINATOR) - 1
                        + sizeof(REQUEST_TERMINATOR) - 1];
        const char * sendIt;

        Value value;
        unsigned int offset;
        unsigned int writeOffset;
        unsigned int cursor;
        int tokens[6];
        int tokGet;
        int tokenCount;
        bool tokenFlag;
        bool endFlag;
        bool continueFlag;
        unsigned int length;
        bool active;

        State state;

        char protocol;

        char quit;

        char * queueBuffer;

        unsigned int queueLength;

        unsigned int queueCapacity;

};

struct workerData {
        int clientSocket;

        int threadNo;

}__attribute__((__packed__));

const int QUEUE_CHUNK_LENGTH = 1024 * 1024;

struct raw_request_header {
        uint8_t magic;
        uint8_t opcode;
        uint16_t keylen;
        uint8_t extlen;
        uint8_t datatype;
        uint16_t reserved;
        uint32_t bodylen;
        uint32_t opaque;
        uint64_t cas;
}__attribute__((__packed__));

struct set_request_header {
        raw_request_header header;
        uint32_t flags;
        uint32_t expiration;
}__attribute__((__packed__));

struct increment_request_header {
        raw_request_header header;
        uint64_t delta;
        uint64_t initial_value;
        uint32_t expiration;
}__attribute__((__packed__));

union request_header {
        raw_request_header header;
        set_request_header set_header;
        increment_request_header increment_header;
};

struct raw_response_header {
        uint8_t magic;
        uint8_t opcode;
        uint16_t keylen;
        uint8_t extlen;
        uint8_t datatype;
        uint16_t status;
        uint32_t bodylen;
        uint32_t opaque;
        uint64_t cas;
}__attribute__((__packed__));

struct hit_response_header {
        raw_response_header header;
        uint32_t flags;
}__attribute__((__packed__));

struct increment_response_header {
        raw_response_header header;
        uint64_t value;
}__attribute__((__packed__));

union response_header {
        raw_response_header header;
        hit_response_header hit_header;
        increment_response_header increment_header;
};

inline void loadConfFile(const string& serverConfFilename);
inline void * threadWorker(void * arg);
inline void acceptCallback(int serverSocket, short event, void * arg);
inline void pipeCallback(int fd, short event, void * arg);
inline void readCallback(int socket, short event, void * arg);
inline void writeCallback(int socket, short event, void * arg);
inline void binaryHandler(Info * i);
inline void asciiHandler(Info * i);
inline void parseState(Info * i);
inline void resetBuffer(Info * i);

// * closes client socket and frees memory related to that connection
// */
//inline void sweepConnection(Info * i);

#define sweepConnectionTemplate(i, event) { \
        i->t->connections--; \
        event_del(&event); \
        close(i->clientSocket); \
        if (i->queueBuffer) { \
                free(i->queueBuffer); \
        } \
        free(i); \
}

#define sweepConnection(i) sweepConnectionTemplate(i, i->readEvent)

#define sweepConnectionWrite(i) sweepConnectionTemplate(i, i->writeEvent)

void printBinaryBuffer(void * buffer, int n, const char * label = NULL);

void printBinaryCommand(uint8_t cmd);

void signalHandler(int n);

void prepareStatBinaryResponse(request_header * header, Info * i);

int getStat(Stats * stats, const char * arg, char * buffer, int maxLength);

void freeQueue(Info * i);

void daemonize();

/****************************
 * configuration
 ****************************/

unsigned port;

std::string userName = SINGLE_USER;

std::string spaceName = SINGLE_SPACE;

unsigned softQuota;

unsigned hardQuota;

unsigned serverLoggerInterval;

bool logSwitch;

std::string serverLogsDirectory;

int threadCount;

timeval connectionTimeout;

Thread * threads;

time_t startTime;

Cache * cache;

Space * space;

ServerLogger * serverLogger;

#endif /* EVENTSERVER_H_ */

---