#include <stdlib.h>
#include <stdio.h>
#include <stdint.h>
#include <assert.h>
#include <unistd.h>
#include <pthread.h>

#include <sophia.h>

FILE *stat_file = NULL;
FILE *bench_file = NULL;

unsigned long long now(void)
{
    unsigned long long tm;
    struct timeval tv;
    gettimeofday(&tv, NULL);
    tm = ((long)tv.tv_sec) * 1000;
    tm += tv.tv_usec / 1000;
    return tm;
}

unsigned long long begin = 0;
uint32_t n = 0;
void qos(int limit)
{
    if (n == limit) {
        unsigned long long c = now();
        unsigned long long diff = c - begin;
        float rps = n / (diff / 1000.0);
        if (rps > limit) {
            double t = ((rps - limit) * ( (double)diff / (double)n ));
            usleep(t * 1000);
        }
        begin = now();
        n = 0;
    }
    n++;
}

unsigned long long bench_last = 0;
uint32_t bench_n = 0;

float rps[16];
int rps_current = 0;

static inline void
print_current(uint64_t i)
{
    if (i > 0 && (i % 100000) == 0) {
        unsigned long long c = now();
        unsigned long long diff = c - bench_last;
        float rps = bench_n / (diff / 1000.0);
        float latency = (diff / 1000.0) / bench_n;
        bench_last = now();
        bench_n = 0;
        fprintf(bench_file, "%.1fM   rps: %f   latency: %f\n", i / 1000000.0, rps, latency);
        fflush(bench_file);
        printf("%.1fM   rps: %f   latency: %f\n", i / 1000000.0, rps, latency);
        fflush(stdout);
    }
    bench_n++;
}

static inline void
sophia_error(void *env)
{
    void *c = sp_ctl(env);
    void *o = sp_get(c, "sophia.error");
    char *value = (char*)sp_get(o, "value", NULL);
    if (value) {
        printf("error: %s\n", value);
    }
    sp_destroy(o);
}

static inline int
generate(char *name, int size)
{
    int max_len = 20;
    int count = 1000;
    int i = 0;
    int len = rand() % max_len;
    if (len <= 4)
        len += 7;
    int j = 0;
    while (j < len) {
        uint32_t seed = rand();
        if (j % 2 == 0)
            name[j] = 97 + seed % 25;
        else
            name[j] = 65 + seed % 25;
        j++;
    }
    int a = snprintf(name + j, size - j, "@mail.ru", name);
    return j + a + 1;
}

static inline void
set(void *env, void *db)
{
    uint64_t current = 0;
    uint64_t count = 6200000000;
    char key[256];
    char value[100];
    while (current < count)
    {
        //qos(600);
        void *tx = sp_begin(env);
        if (tx == NULL) {
            sophia_error(env);
            return;
        }
        char name[127];
        generate(name, sizeof(name));
        int count = rand() % 50;
        int j = 0;
        while (j < 150 + count) {
            int size = snprintf(key, sizeof(key), "%s_%d", name, rand());
            void *o = sp_object(db);
            assert(o != NULL);
            sp_set(o, "key", key, size);
            sp_set(o, "value", value, (rand() % 40) + 1);
            int rc = sp_set(tx, o);
            if (rc == -1) {
                sophia_error(env);
                return;
            }
            j++;
            current++;
            print_current(current);
        }
        int rc = sp_commit(tx);
        if (rc == -1) {
            sophia_error(env);
        }
    }
}

static inline void
seq(void *env, void *db)
{
    uint64_t current = 0;
    void *o = sp_object(db);
    void *cursor = sp_cursor(db, o);
    while (sp_get(cursor)) {
        current++;
        print_current(current);
    }
    sp_destroy(cursor);
}

int run = 0;
static void *stats(void *env)
{
    while (run)
    {
        sleep(5);
        fprintf(stat_file, "\n\n");
        void *c = sp_ctl(env);
        void *o, *cur = sp_cursor(c);
        while ((o = sp_get(cur))) {
            char *key = (char*)sp_get(o, "key", NULL);
            char *value = (char*)sp_get(o, "value", NULL);
            if (value) {
                fprintf(stat_file, "%s = %s\n", key, value);
            } else {
                fprintf(stat_file, "%s =\n", key);
            }
        }
        sp_destroy(cur);

        fflush(stat_file);
        sleep(5);
    }
    return NULL;
}

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

    void *env = sp_env();
    void *c = sp_ctl(env);
    sp_set(c, "sophia.path", "/mnt/1/benchmark_2");
    //sp_set(c, "memory.limit", "10737418240");
    sp_set(c, "compaction.0.compact_wm", "5");
    sp_set(c, "db", "test");
    void *db = sp_get(c, "db.test");
    int rc = sp_open(env);
    if (rc == -1) {
        sophia_error(env);
        sp_destroy(env);
        return 1;
    }

    bench_file = fopen("sophia_benchmark", "w");
    assert(bench_file != NULL);

    stat_file = fopen("sophia_stats", "w");
    assert(stat_file != NULL);

    run = 1;
    pthread_t id;
    pthread_create(&id, NULL, stats, env);

    set(env, db);
    //seq(env, db);
    run = 0;

    pthread_join(id, NULL);

    fclose(bench_file);
    fclose(stat_file);

    sp_destroy(env);
    return 0;
}