src/libmemcached-based-client/lm_client.c

#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include <libmemcached/memcached.h>

#include <syslog.h>
#include <errno.h>
#include <unistd.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>

#include <signal.h>

#include <libmemcached/memcached.h>

pthread_t * thread;
int attack;

int single_mode;
int binary_mode;

int admin_port;
char ** server_names;
int server_count;

char** keys;
char** values;

int* keyLength;
int* valueLength;

int count = 0;
int threadCount;
int getsPerSet;

int reqSent;
int valReqSent;

int timeout;

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;

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);
}

void nothing(int n) {
}

void * worker(void * arg) {

        static int num = 0;
        int i;

        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);

        //zaczyna sie zabawa z libmemcached:

        memcached_return rc;

        memcached_server_st **servers = malloc(server_count * sizeof(memcached_server_st *));
        memcached_st ** memc = malloc(server_count * sizeof(memcached_st *));

        for (i = 0; i < 2 * server_count; i += 2) {
                memc[i] = memcached_create(NULL);
                if (binary_mode){
                        memcached_behavior_set(memc[i], MEMCACHED_BEHAVIOR_BINARY_PROTOCOL, 1);
                }
                servers[i]
                                = memcached_server_list_append(NULL, server_names[0], atoi(server_names[1]), &rc);
                memcached_server_push(memc[i], servers[i]);
        }

        //the client sleeps when turned off by client manager
        printf("atacking thread pauses...\n");

//      int index = 20000;

        while (1) {
                if (!attack) {
                        pause();
                        printf("atacking thread continues to attack!\n");
                        clock_gettime(CLOCK_REALTIME, &start);
                }

                int index = rand() % count;

                if (single_mode) {
                        for (i = 0; i < 2 * server_count; i += 2) {
                                memcached_quit(memc[i]);
                        }
                }

                static struct timespec start, stop;
                if (getsPerSet == -1 || (rand() % (getsPerSet + 1))) { //get
                        uint32_t flags;
                        size_t objSize;
                        for (i = 0; i < server_count; i++) {
                                clock_gettime(CLOCK_REALTIME, &start);
                                char * ret = memcached_get(memc[i], keys[index], keyLength[index], &objSize,
                                                &flags, &rc);
                                clock_gettime(CLOCK_REALTIME, &stop);
                                stats.totalGetResponseTime += (double) (stop.tv_sec - start.tv_sec)
                                                + (double) (stop.tv_nsec - start.tv_nsec) / 1000000000.0;
                                if (ret) {
                                        if (!strncmp(values[index], ret, valueLength[index])) {
                                                stats.hitCount++;
                                                stats.correctAnswerCount++;
                                        } else {
                                                stats.wrongAnswerCount++;
                                        }
                                } else {
                                        stats.missCount++;
                                        stats.correctAnswerCount++;
                                }
                                stats.sentGetsCount++;
                                free((void*) ret);

                        }
                } else { //set

                        for (i = 0; i < server_count; i++) {
                                clock_gettime(CLOCK_REALTIME, &start);
                                rc = memcached_set(memc[i], keys[index], keyLength[index], values[index],
                                                valueLength[index], (time_t) (rand() % timeout + 1),
                                                (uint32_t) index);
                                clock_gettime(CLOCK_REALTIME, &stop);
                                stats.totalSetResponseTime += (double) (stop.tv_sec - start.tv_sec)
                                                + (double) (stop.tv_nsec - start.tv_nsec) / 1000000000.0;

                                stats.sentSetsCount++;
                                if (rc == MEMCACHED_SUCCESS) {
                                        stats.storedCount++;
                                        stats.correctAnswerCount++;
                                } else {
                                        stats.notStoredCount++;
                                        stats.correctAnswerCount++;
                                }
                        }

                }

//              index++;
//              index %= count;

        }

        // we never arrive here..
        //      memcached_server_free(servers);
        //      memcached_free(memc);

        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("stopping...\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);
}

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

        attack = 0;

        srand(time(NULL));

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

        single_mode = atoi(argv[6]);
        binary_mode = atoi(argv[7]);

        admin_port = atoi(argv[1]);

        server_names = argv + 8;
        server_count = (argc - 8) / 2;

        pthread_t adminThread;

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

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

        reqSent = 0;
        valReqSent = 0;

        threadCount = atoi(argv[3]);
        getsPerSet = atoi(argv[4]);
        timeout = atoi(argv[5]);

        init_objects(argv[2]);

        int i;

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

        for (i = 0; i < threadCount; i++)
                pthread_create(&thread[i], NULL, worker, NULL);

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

        return 0;
}

---