src/client/client.c

/*
 * client.c
 *
 *  Created on: 2009-09-08
 *      Author: dziki_rysio_997
 */

#include <stdio.h>
#include <stdlib.h>
#include <syslog.h>
#include <errno.h>
#include <unistd.h>
#include <string.h>

/* sockets */
#include <sys/types.h>
#include <sys/socket.h>

/* for non blocking */
#include <fcntl.h>

#include <netinet/in.h>
#include <arpa/inet.h>

/* for struct timeval */
#include <time.h>

#include <pthread.h>

/* for libevent */
#include <event.h>

#include <assert.h>

#include <signal.h>

#include <stdlib.h>

int attack;

int admin_port;

char** keys;
char** values;

int* keyLength;
int* valueLength;

int count=0;
int threadCount;
int getsPerSet;

int reqSent;
int valReqSent;

int timeout;

struct timeval tv;

pthread_t * thread;

struct statistics {
        unsigned long long wrongAnswerCount;
        unsigned long long correctAnswerCount;
        unsigned long long hitCount;
        unsigned long long missCount;
        unsigned long long sentSetsCount;
        unsigned long long sentGetsCount;
        unsigned long long storedCount;
        unsigned long long notStoredCount;
        double totalTime;
        double totalSetResponseTime;
        double totalGetResponseTime;
};

struct timespec start, stop, getSent, getResp, setSent, setResp;

#define TOTAL_STATS_SIZE sizeof(struct statistics)

struct statistics stats;

#define OP_SET 1
#define OP_GET 2
#define OP_GETS 3

#define TMPL_SET "set %s %d %d %d\r\n%s\r\n"
#define TMPL_GET "get %s\r\n"
#define TMPL_GETS "gets %s\r\n"

struct request {
        struct event *ev;
        struct event *respev;
        unsigned int op :2;
        const char *req;
        int reqsize;
        struct event_base *base;
        char *response;
        off_t offset;
        off_t respoffset;
        off_t resppos;
        off_t tokcount;
        off_t flagpos;
        off_t objsizepos;
        off_t objpos;
        off_t mark;
};

/*
struct validation_request {
        struct event *ev;
        int pos;
        char reqBuf[256];
        off_t offset;
        int reqLen;
        struct event_base *base;
};

struct validation_response {
        struct event *ev;
        //      off_t offset;
        char reqRcvd[1024];
        char object[1048576];
        int objPos;

        off_t flagPosBeg;
        off_t flagPosEnd;

        char buf[1024];

        int bufLen;

        off_t bufPos;
        off_t reqPos;

        off_t lineEnd;
        int spaceCount;
        int completed;
        struct event_base *base;
};
*/

/*
struct response {
        struct event *ev;
        off_t offset;
        struct event_base *base;
};*/

void init_objects(const char * file_name) {
        FILE *fp = NULL;
        if ((fp = fopen(file_name, "rb")) == NULL) {
                fprintf(stderr, "Unable to open file.\n");
                return;
        }

        fread(&count, sizeof(int), 1, fp);

        printf("Read %d input values.\n", count);

        keys = (char**) malloc(count * sizeof(char*));
        values = (char**) malloc(count * sizeof(char*));

        keyLength = (int*) malloc(count * sizeof(int));
        valueLength = (int*) malloc(count * sizeof(int));

        int i;
        for (i = 0; i < count; i++) {
                fread(&keyLength[i], sizeof(int), 1, fp);
                fread(&valueLength[i], sizeof(int), 1, fp);
                keys[i] = (char*) malloc((keyLength[i]+1) * sizeof(char));
                fread(keys[i], keyLength[i], 1, fp);
                values[i] = (char*) malloc((valueLength[i]+1) * sizeof(char));
                fread(values[i], valueLength[i], 1, fp);
                keys[i][keyLength[i]]='\0';
                values[i][valueLength[i]]='\0';
//              printf("%d|keylen:%d|vallen:%d|%s|%s|\n", i, keyLength[i], valueLength[i], keys[i], values[i]);
        }

        fclose(fp);
}

/*struct arg {
        struct event *ev;
        int fd;
};*/

void nothing(int n) {
}

void * writeEv(void* arg) {

        static int num = 0;
        int ret;

        struct sigaction action;

        action.sa_handler = nothing;
        sigemptyset(&action.sa_mask);
        action.sa_flags = SA_NOCLDSTOP;

        if (sigaction(SIGUSR1, &action, NULL)) {
                perror("sigaction");
                exit(1);
        }

        usleep(rand() % 100000);
        fprintf(stderr, "start thread %d\n", ++num);

        do {
                ret = event_base_loop((struct event_base *) arg, 0);
                printf("X");
                fflush(stdout);
        } while (ret > -1);
        fprintf(stderr, "Error occurred\n");
        return NULL;

}

void * admin(void* arg) {
        struct sockaddr_in serv_addr, cli_addr;
        int fd, cli_fd;

        fd = socket(AF_INET, SOCK_STREAM, 0);
        memset(&serv_addr, 0, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;
        serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
        serv_addr.sin_port = htons(admin_port);

        int flag = 1;
        if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag)) < 0) {
                perror("setsockopt(SO_REUSEADDR)");
        }

        if (bind(fd, (struct sockaddr*) &serv_addr, sizeof(serv_addr))) {
                perror("bind");
                exit(1);
        }

        if (listen(fd, 1)) {
                perror("listen");
                exit(1);
        }

        socklen_t cli_len;
        cli_len = sizeof(struct sockaddr_in);

        printf("Started admin channel.\n");

        while (1) {
                cli_fd = accept(fd, (struct sockaddr*) &cli_addr, &cli_len);

#define START '1'
#define STOP '9'
#define STATS '5'

                char type;
                if (recv(cli_fd, (void*) &type, 1, 0) == 1) {

                        switch (type) {
                        case START:
                                printf("waking threads...\n");
                                attack = 1;
                                int i;
                                for (i = 0; i < threadCount; i++) {
                                        pthread_kill(thread[i], SIGUSR1);
                                }
                                break;
                        case STOP:
                                printf("stoping...\n");
                                attack = 0;
                                clock_gettime(CLOCK_REALTIME, &stop);
                                stats.totalTime += (double) (stop.tv_sec - start.tv_sec)
                                                + (double) (stop.tv_nsec - start.tv_nsec)
                                                                / 1000000000.0;
                                break;
                        case STATS:
                                printf("Stats requested.\n");

                                char sendBuf[TOTAL_STATS_SIZE];

                                memcpy((void*) (sendBuf), (void*) &stats, TOTAL_STATS_SIZE);

                                if (send(cli_fd, (void*) sendBuf, TOTAL_STATS_SIZE, 0)
                                                == TOTAL_STATS_SIZE) {
                                        printf("Stats sent.\n");
                                }

                                break;
                        }
                }

                close(cli_fd);
        }

        close(fd);
}

const char * build_req(short type, int key_index, int *req_len) {
        int len;
        char *req;

        if (type == OP_SET) {
                len = sizeof(TMPL_SET) / sizeof(TMPL_SET[0]) + keyLength[key_index]
                                + valueLength[key_index] + 10 + 1; //+50 + 1;
                req = malloc(len);
                len = sprintf(req, TMPL_SET, keys[key_index], key_index, rand()
                                % timeout + 1, valueLength[key_index], values[key_index]);
                //snprintf(req, len, TMPL_SET, keys[key_index], keyLength[key_index]);
        } else if (type == OP_GET) { //get ala\r\n
                len = sizeof(TMPL_GET) / sizeof(TMPL_GET[0]) + keyLength[key_index] + 1;
                req = malloc(len);
                //len = snprintf(req, len, TMPL_GET, keys[key_index]);
                len = sprintf(req, TMPL_GET, keys[key_index]);
        } else {
                assert(0);
        }
        *req_len = len;
        return req;
}

#define RET_NEED_MORE 1
#define RET_OK 2
#define RET_ERROR -1

#define SHORTEST_SET_RESPONSE "STORED\r\n"
#define LONGEST_SET_RESPONSE "NOT_STORED\r\n"
#define SHORTEST_SET_RESPONSE_LENGTH sizeof(SHORTEST_SET_RESPONSE)-1
#define LONGEST_SET_RESPONSE_LENGTH sizeof(LONGEST_SET_RESPONSE)-1

#define SHORTEST_GET_RESPONSE "END\r\n"

#define LONGEST_GET_RESPONSE_LENGTH 2*1024*1024
#define SHORTEST_GET_RESPONSE_LENGTH 5

static int validate_set_response(const char *response, int len) /* expecting STORED or NOT_STORED */{
#define OK_CMD "STORED\r\n"
#define FAIL_CMD "NOT_STORED\r\n"
        assert(response != NULL);

        if (len < SHORTEST_SET_RESPONSE_LENGTH) { /* shortest set response has 8 bytes length */
                return RET_NEED_MORE;
        }

        if (response[0] == 'S') {
                if (strncmp(response, OK_CMD, len) == 0) {
                        return RET_OK;
                }
        }

        if (response[0] == 'N') {
                if (len < LONGEST_SET_RESPONSE_LENGTH) {
                        return RET_NEED_MORE;
                }
        }
        return RET_ERROR;
}

void write_callback(int fd, short event, void *arg);

void read_callback(int fd, short event, void *arg) {

        struct request *req = (struct request *) arg;
        int ret;

        if (event == EV_TIMEOUT) {
                /* XXX: timeout */
                return;
        }

        if (req->op == OP_SET) {
                if (req->response == NULL) {
                        req->response = malloc(LONGEST_SET_RESPONSE_LENGTH);
                }
                assert(req->response != NULL);
                int toRead = 0;
                if (req->respoffset < SHORTEST_SET_RESPONSE_LENGTH) {
                        toRead = SHORTEST_SET_RESPONSE_LENGTH - req->respoffset;
                } else if (req->respoffset > SHORTEST_SET_RESPONSE_LENGTH
                                && req->respoffset < LONGEST_SET_RESPONSE_LENGTH) {
                        toRead = LONGEST_SET_RESPONSE_LENGTH - req->respoffset;
                }
                //              printf("[");
                //              fflush(stdout);
                ret = read(fd, (req->response + req->respoffset), toRead);
                //              printf("]");
                //              fflush(stdout);
                if (ret < 0) {
                        if (errno == EAGAIN || errno == EINPROGRESS) {
                                event_add(req->respev, &tv);
                                return;
                        }
                }
                req->respoffset += ret;
                ret = validate_set_response(req->response, req->respoffset);
                if (ret == RET_NEED_MORE) {
                        event_add(req->respev, &tv);
                        return;
                } else if (ret == RET_OK) { /* request successful */
                        stats.storedCount++;
                        stats.correctAnswerCount++;
                        free((void *) req->req);
                        req->req = NULL;
                        req->offset = 0;
                        event_set(req->respev, fd, EV_WRITE, write_callback, req);
                        event_base_set(req->base, req->respev);
                        event_add(req->respev, &tv);
                        req->respoffset = 0;
                        clock_gettime(CLOCK_REALTIME, &setResp);
                        stats.totalSetResponseTime += (double) (setResp.tv_sec
                                        - setSent.tv_sec) + (double) (setResp.tv_nsec
                                        - setSent.tv_nsec) / 1000000000.0;
                } else {
                        stats.notStoredCount++;
                        stats.correctAnswerCount++;
                        free((void *) req->req);
                        req->req = NULL;
                        req->offset = 0;
                        event_set(req->respev, fd, EV_WRITE, write_callback, req);
                        event_base_set(req->base, req->respev);
                        event_add(req->respev, &tv);
                        req->respoffset = 0;
                        clock_gettime(CLOCK_REALTIME, &setResp);
                        stats.totalSetResponseTime += (double) (setResp.tv_sec
                                        - setSent.tv_sec) + (double) (setResp.tv_nsec
                                        - setSent.tv_nsec) / 1000000000.0;
                }
        }
}

void read_get_callback(int fd, short event, void *arg) {

        struct request *req = (struct request *) arg;

        if (event == EV_TIMEOUT) {
                /* XXX: timeout */
                return;
        }

        if (req->response == NULL) {
                req->resppos = 0;
                req->tokcount = 0;
                req->response = malloc(LONGEST_GET_RESPONSE_LENGTH);
                req->mark = 0;
        }
        assert(req->response != NULL);

        int len = recv(fd, req->response + req->respoffset,
                        LONGEST_GET_RESPONSE_LENGTH, MSG_DONTWAIT);

        if (len == 0) {
                event_set(req->respev, fd, EV_READ, read_get_callback, req);
                event_base_set(req->base, req->respev);
                event_add(req->respev, &tv);
                return;
        } else if (len > 0) {
                req->respoffset += len;
        }

        //get response looks like this: VALUE <key> <flag> <size>\r\n<object>\r\n
        while (req->resppos < req->respoffset) {

                if (req->tokcount < 3 || req->tokcount == 7) {
                        switch (req->response[req->resppos]) {
                        case '\r':
                                req->resppos++;
                                break;
                        case '\n':
                                if (req->resppos - req->mark == SHORTEST_GET_RESPONSE_LENGTH
                                                - 1) {
                                        if (!strncmp(req->response + req->mark,
                                                        SHORTEST_GET_RESPONSE, SHORTEST_GET_RESPONSE_LENGTH)) {
                                                if (req->tokcount == 7) {
                                                        int size = atoi(req->response + req->objsizepos);
                                                        int flag = atoi(req->response + req->flagpos);

                                                        if (size == valueLength[flag] && !strncmp(
                                                                        req->response + req->objpos, values[flag],
                                                                        size)) {
                                                                stats.hitCount++;
                                                                stats.correctAnswerCount++;
                                                        } else {
                                                                stats.hitCount++;
                                                                stats.wrongAnswerCount++;
                                                        }
                                                } else {
                                                        stats.missCount++;
                                                        stats.correctAnswerCount++;
                                                }
                                                clock_gettime(CLOCK_REALTIME, &getResp);
                                                stats.totalGetResponseTime += (double) (getResp.tv_sec
                                                                - getSent.tv_sec) + (double) (getResp.tv_nsec
                                                                - getSent.tv_nsec) / 1000000000.0;
                                        } else {
                                                assert(0);
                                        }
                                        free((void *) req->req);
                                        req->mark = 0;
                                        req->tokcount = 0;
                                        req->req = NULL;
                                        req->offset = 0;
                                        req->respoffset = 0;
                                        req->resppos = 0;

                                        event_set(req->respev, fd, EV_WRITE, write_callback, req);
                                        event_base_set(req->base, req->respev);
                                        event_add(req->respev, &tv);
                                        return;
                                }
                                break;
                        case ' ':
                                if (req->tokcount == 1) {
                                        req->flagpos = req->resppos + 1;
                                } else if (req->tokcount == 2) {
                                        req->objsizepos = req->resppos + 1;
                                }
                                req->response[req->resppos] = '\0';
                                req->tokcount++;
                                req->resppos++;
                                break;
                        default:
                                req->resppos++;
                                break;
                        }
                } else if (req->tokcount == 3) { //we finish parsing the first line
                        switch (req->response[req->resppos]) {
                        case '\r':
                                req->tokcount++;
                                req->response[req->resppos] = '\0';
                                req->resppos++;
                                break;
                        case '\n':
                                assert(0);
                                break;
                        default:
                                req->resppos++;
                                break;
                        }
                } else if (req->tokcount == 4) {
                        switch (req->response[req->resppos]) {
                        case '\r':
                                assert(0);
                                break;
                        case '\n': //end of first line
                                req->tokcount++;
                                req->objpos = ++req->resppos;
                                break;
                        default:
                                assert(0);
                                break;
                        }
                } else if (req->tokcount == 5) {
                        switch (req->response[req->resppos]) {
                        case '\r':
                                req->tokcount++;
                                req->resppos++;
                                break;
                        case '\n':
                                assert(0);
                                break;
                        default:
                                req->resppos++;
                                break;
                        }
                } else if (req->tokcount == 6) {
                        switch (req->response[req->resppos]) {
                        case '\r':
                                assert(0);
                                break;
                        case '\n':
                                req->tokcount++;
                                req->mark = ++req->resppos;
                                break;
                        default:
                                assert(0);
                                break;
                        }
                }
        }

        event_set(req->respev, fd, EV_READ, read_get_callback, req);
        event_base_set(req->base, req->respev);
        event_add(req->respev, &tv);
}

void write_callback(int fd, short event, void *arg) {

        //      printf("<");
        //      fflush(stdout);

        struct request *req = (struct request *) arg;
        int ret;

        if (event == EV_TIMEOUT) {
                /* XXX: timeout */
                printf("EV_TIMEOUT\n");
                return;
        }

        if (req->req == NULL) {

                //the client sleeps when turned off by client manager
                if (!attack) {
                        printf("atacking thread pauses...\n");
                        while (!attack) {
                                pause();
                        }
                        printf("atacking thread continues to attack!\n");
                        clock_gettime(CLOCK_REALTIME, &start);
                }
                assert(req->offset == 0);

                if (getsPerSet == -1) {
                        req->op = OP_GET;
                } else if (!(rand() % (getsPerSet + 1))) {
                        req->op = OP_SET;
                } else
                        req->op = OP_GET;

                req->req = build_req(req->op, rand() % count, &(req->reqsize));
        }
        //      printf(".");
        //      fflush(stdout);

        if ((ret = write(fd, req->req, req->reqsize - req->offset)) < 0) {
                if (errno == EAGAIN || errno == EINPROGRESS) {
                        fprintf(stderr, "EAGAIN\n");
                        event_add(req->ev, &tv);
                } else {
                        fprintf(stderr, "Write error: %s", strerror(errno));
                        /* XXX: new connection */
                }
        } else {
                req->offset += ret;
                if (req->offset == req->reqsize) { /* complete request sent */
                        if (req->op == OP_SET) {
                                clock_gettime(CLOCK_REALTIME, &setSent);
                                stats.sentSetsCount++;
                                event_set(req->respev, fd, EV_READ, read_callback, req);
                        } else if (req->op == OP_GET) {
                                clock_gettime(CLOCK_REALTIME, &getSent);
                                stats.sentGetsCount++;
                                event_set(req->respev, fd, EV_READ, read_get_callback, req);
                        }
                        free((void *) req->req);
                        req->req = NULL;
                        req->offset = 0;

                        event_base_set(req->base, req->respev);
                        event_add(req->respev, &tv);
                } else {
                        event_add(req->respev, &tv);
                }
        }
        //      printf(">");
        //      fflush(stdout);
}

int main(int argc, char** argv) {

        attack = 0;

        srand(time(NULL));

        if (argc < 8) {
                printf(
                                "Usage: ./client admin_port host port fileName threadCount getsPerSet timeout\n");
                return 1;
        }

        admin_port = atoi(argv[1]);

        pthread_t adminThread;

        pthread_create(&adminThread, NULL, admin, NULL);

        memset((void*) &stats, 0, TOTAL_STATS_SIZE);

        reqSent = 0;
        valReqSent = 0;

        threadCount = atoi(argv[5]);
        getsPerSet = atoi(argv[6]);
        timeout = atoi(argv[7]);

        init_objects(argv[4]);

//      char *req;
//
//      int no = 4;
//
//      int len1 = sizeof(TMPL_SET) / sizeof(TMPL_SET[0]) + keyLength[no]
//              + valueLength[no] + 10 + 1; //+50 + 1;
//      req = malloc(len1);
//      int len2 = sprintf(req, TMPL_SET, keys[no], 0, rand()
//                              % timeout + 1, valueLength[no], values[no]);
//
//      printf("len1:%d|len2:%d|%s|", len1, len2, req);

        tv.tv_sec = 1000;
        tv.tv_usec = 0;

        int fd[threadCount];

        struct event_base * eb[threadCount + 2];

        struct event ev[threadCount + 2];
        struct event respEv[threadCount];

        int i;
        for (i = 0; i < threadCount + 2; i++)
                eb[i] = event_init();

        struct sockaddr_in serv_addr;

        for (i = 0; i < threadCount; i++) {
                fd[i] = socket(AF_INET, SOCK_STREAM, 0);
                memset(&serv_addr, 0, sizeof(serv_addr));
                serv_addr.sin_family = AF_INET;
                serv_addr.sin_addr.s_addr = inet_addr(argv[2]);
                serv_addr.sin_port = htons(atoi(argv[3]));
                if (connect(fd[i], (struct sockaddr*) &serv_addr, sizeof(serv_addr))
                                == -1)
                        printf("Connection failed\n");
                else
                        printf("Connected\n");
        }

        struct request reqTab[threadCount];
        for (i = 0; i < threadCount; i++) {
                reqTab[i].ev = &ev[i];
                reqTab[i].respev = &respEv[i];
                reqTab[i].req = NULL;
                reqTab[i].offset = 0;
                reqTab[i].respoffset = 0;
                reqTab[i].reqsize = 0;
                reqTab[i].response = NULL;
                reqTab[i].op = OP_SET;
                reqTab[i].base = eb[i];
        }

        for (i = 0; i < threadCount; i++) {
                event_set(&ev[i], fd[i], EV_WRITE, write_callback, (void*) &reqTab[i]);
                event_base_set(eb[i], &ev[i]);
                event_add(&ev[i], &tv);
        }

        thread = malloc(threadCount * sizeof(pthread_t));

        for (i = 0; i < threadCount; i++)
                pthread_create(&thread[i], NULL, writeEv, (void *) eb[i]);

        char c;
        scanf("%c", &c);

        return 0;
}

---