transaction.c

// Run like:
// transaction <db_dir> <num threads> <num repeats>
// F.e:> transaction db 2 10000
// Make:
// gcc -Wall -O0 -g -I.. -o transaction transaction.c -pthread ../libsophia.a
//
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/time.h>
#include <unistd.h>

#include "sophia.h"

#define MAX_THREADS 1024
#define VALUE_LEN   4096

int  qn_repeat = 10;
int  no_write  = 0;
char fname[256];

void *db  = NULL;
void *env = NULL;
void *ctl = NULL;

char* itoa(int value, char* result, int base) {
    // check that the base if valid
    if (base < 2 || base > 36) { *result = '\0'; return result; }

    char* ptr = result, *ptr1 = result, tmp_char;
    int tmp_value;

    do {
        tmp_value = value;
        value /= base;
        *ptr++ = "zyxwvutsrqponmlkjihgfedcba9876543210123456789abcdefghijklmnopqrstuvwxyz" [35 + (tmp_value - value * base)];
    } while ( value );
    // Apply negative sign
    if (tmp_value < 0) *ptr++ = '-';
    *ptr-- = '\0';
    while(ptr1 < ptr) {
        tmp_char = *ptr;
        *ptr--= *ptr1;
        *ptr1++ = tmp_char;
    }

    return result;
}

void ptime(const char *message, int number) {
    int sec, msec;
#ifdef _WIN32
    DWORD t;
    t = timeGetTime();
    sec = t / 1000;
    msec = t % 1000;
#else
    struct timeval tv;
    struct timezone tz;
    gettimeofday(&tv, &tz);
    sec  = tv.tv_sec % 1000;
    msec = tv.tv_usec / 1000;
#endif
    printf("[%03d.%03d sec]: %s [%d]\n",sec, msec, message, number);
}

void run_test(void *arg) {

    int i, rc;
    int k = (int) arg;

    char  buf[16];
    char  key[16];
    char  res[16];

    char *data;

    pthread_t id=pthread_self();
    printf("Thread [%d] started, arg=[%d]!\n", (int) id, k);

    data=malloc(VALUE_LEN);
    memset(data, '*', VALUE_LEN);

    if (k==0) {
        if (no_write==0) {
            void *transaction;
            void *obj;

            for (i=0;i<qn_repeat;i++) {
                if (i%1000==0) {
                    printf("Write key [%d]...\n",i);
                }

                transaction = sp_begin(env);
                obj = sp_object(db);

                if (obj==NULL) {
                    printf("Object is empty in get thread!\n");
                    return;
                }

                itoa(i, buf, 10);

                sp_set(obj, "key", buf, strlen(buf));
                sp_set(obj, "value", data, VALUE_LEN);

                rc = sp_set(transaction, obj);
                sp_commit(transaction);

                if (rc == -1) {
                    printf("sp_set() error for key [%s]\n", buf);
                }
                else {
                    // printf("Key [%d] set to [%d]\n", i, i);
                }

                // Time to die?
                sp_destroy(obj);
                sp_destroy(transaction);
            }
            ptime("Data added!",0);
        }
        else {
            printf("Write distabled!");
        }
    }
    else {
        for (i=0;i<qn_repeat;i++) {
            if (i%1000==0) {
                printf("Read key [%d]...\n",i);
            }
            itoa(i, key, 10);

            void *transaction = sp_begin(env);
            void *obj = sp_object(db);

            if (obj==NULL) {
                printf("Object is empty in get thread!\n");
                return;
            }

            sp_set(obj, "key", key, strlen(key));
            void *result = sp_get(transaction, obj);

            if (result!=NULL) {
                int valuesize;
                char *value = sp_get(result, "value", &valuesize);
                if (memcmp(value, data, VALUE_LEN)!=0) {
                    printf("Error! Data corrupted!");
                }
                sp_destroy(result);
            }
            else {
                printf("[%s] key not found!\n", key);
            }

            // Ok, free
            sp_destroy(obj);
            sp_destroy(transaction);
        }
    }
}

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

    int i;
    int thread_count=1;
    pthread_t thread[MAX_THREADS];

    if (argc>1) {
        strcpy(fname, argv[1]);
    }
    else {
        strcpy(fname, "./db");
    }

    if (argc>2) {
        thread_count=atoi(argv[2]);
        if (thread_count<0) {
            no_write=1;
            thread_count=-1*thread_count;
        }
    }

    if (argc>3) {
        qn_repeat=atoi(argv[3]);
    }

    printf("Starting test with [%d] threads, [%d] repeats...\n", thread_count, qn_repeat);
    ptime("Go!", 1);

    env = sp_env();
    ctl = sp_ctl(env);

    sp_set(ctl, "sophia.path", fname);
    sp_set(ctl, "db", "test");

    int rc = sp_open(env);
    if (rc == -1) {
        printf("Cannot open env!\n");
        return rc;
    }

    db = sp_get(ctl, "db.test");

    for (i=0; i<thread_count; i++) {
        pthread_create(&(thread[i]), NULL, (void *) &run_test,(void *) i);
        if (i==0) {
            usleep(30*1000);
            ptime("Add delay finished!",0);
        }
    }

    /*
     * Wait for tests to finish
     */
    for (i=1; i<=thread_count; i++) {
        pthread_join(thread[i], NULL);
    }

    ptime("All thread finished!",0);
    sp_destroy(db);
    sp_destroy(env);
    printf("DB closed\n");
    sleep(10);

    return 0;
}