/alps/fwbench

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <errno.h>

#include <fcntl.h>

#include <glib.h>

#include <math.h>

#include <unistd.h>

#include <sys/stat.h>

#include <sys/time.h>

#include <fastwriter.h>

#include <ufo/ufo-graph.h>

#include "config.h"

#define FW_BUFFER 4096l

#define WRITE_INTERVAL 1

#define WRITE_SUMMARY 5

#define NEWFILE 10

#define run_time 3600

const char *fifo_name = ".fifo";

struct setup_s {

    size_t width;

    size_t height;

    size_t bpp;

    size_t fps;

    size_t iters;

    volatile int run_started;

};

typedef struct setup_s setup_t;

void handle_error(GError *error) {

    if (error != NULL) {

        g_print("%s\n", error->message); 

        g_error_free(error);

        exit(EXIT_FAILURE);

    }

}

void set_speed(setup_t *setup, size_t speed) {

    setup->bpp = sizeof(float);

    setup->width = 1024;

    setup->height = 768;

    setup->fps = 1 + speed / setup->width / setup->height / setup->bpp;

}

void set_size(setup_t *setup, size_t size) {

    setup->iters = size / setup->width / setup->height / setup->bpp;

}

void set_time(setup_t *setup, size_t time) {

    setup->iters = time * setup->fps;

}

static void *run(setup_t *setup) {

    GError *error = NULL;

    UfoGraph *graph = NULL;

    /* If you want to use system-wide installed filters: 

     * graph = ufo_graph_new(); */

    graph = g_object_new(UFO_TYPE_GRAPH,

#ifdef METABALLS_PATH

            "paths", METABALLS_PATH,

#endif /* METABALLS_PATH */

            NULL);

//    printf("%lu %lu %lu %lu\n", setup->width, setup->height, setup->iters, setup->width * setup->height * setup->iters * sizeof(float));

    UfoFilter *metaballs = ufo_graph_get_filter(graph, "metaballs", &error);

    handle_error(error);

    g_object_set(G_OBJECT(metaballs),

            "width", setup->width,

            "height", setup->height,

            "num-balls", 1,

            "num-iterations", setup->iters,

            "frames-per-second", setup->fps,

            NULL);

    UfoFilter *writer = ufo_graph_get_filter(graph, "pipeoutput", &error);

    handle_error(error);

    g_object_set(G_OBJECT(writer),

            "pipe-name", fifo_name,

            NULL);

    ufo_filter_connect_to(metaballs, writer, &error);

    handle_error(error);

    setup->run_started = 1;

    ufo_graph_run(graph, &error);

    handle_error(error);

//    g_print("Wrote %lu bytes\n", width * height * num_iterations * sizeof(float));

    g_thread_exit(NULL);

    return NULL;

}

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

{

    int err;

    GError *gerr;

    GThread *thr;

    setup_t setup;

    fastwriter_t *fw;

    fastwriter_stats_t stats;

    const char *out = "/dev/null";

    size_t speed = 850;

    size_t num_read = 0;

    size_t buf_max = 0;

    int broken_frame = 0;

    size_t writeouts = 0;

    size_t duration, last_duration;

    struct timeval tv;

    struct timeval tv_started = {0};

    struct timeval tv_last_written = {0};

    unsigned long frames = 0, lost = 0;

    unsigned long delta, expected;

    unsigned long last_frames = 0, last_lost = 0;

    g_thread_init(NULL);    

    g_type_init();

    if (argc > 1) {

	out = argv[1];

    }

    if (argc > 2) {

	speed = atoi(argv[2]);

    }

    unlink(fifo_name);

    g_assert(!mkfifo(fifo_name, S_IWUSR | S_IRUSR));

    fw = fastwriter_init(out, FASTWRITER_FLAGS_OVERWRITE);

    g_assert(fw);

    err = fastwriter_open(fw, out,  FASTWRITER_FLAGS_OVERWRITE);

    if (err) printf("FastWriter returned error %i\n", err);

    g_assert(!err);

    fastwriter_set_buffer_size(fw, FW_BUFFER * 1024 * 1024);

    fastwriter_get_stats(fw, &stats);

    printf("*** Writing to %s, speed: %lu MB/s, buffer: %lu MB\n", out, speed, stats.buffer_size);

    memset(&setup, 0, sizeof(setup_t));

    set_speed(&setup, speed * 1024 * 1024);

    set_size(&setup, run_time * speed * 1024 * 1024);

    size_t frame_size = setup.width * setup.height * setup.bpp;

    void *buffer = malloc(frame_size);

    g_assert(buffer);

    thr = g_thread_create((GThreadFunc)run, &setup, 1, &gerr);

    g_assert(thr);

    while (!setup.run_started);

    int fd = open(fifo_name, O_RDONLY);

    g_assert(fd);

    ssize_t result = read(fd, buffer, frame_size);

    while (result > 0) {

	if (!broken_frame) {

	    err = fastwriter_push(fw, result, buffer);

    	    if (err) {

    		if (err == EWOULDBLOCK) {

		    if (num_read) fastwriter_cancel(fw);

		    broken_frame = 1;

    		} else {

	    	    if (err) printf("FastWriter returned error %i\n", err);

		    g_assert(!err);

		}

	    }

    	}

        num_read += result; 

        if (num_read < frame_size) {

	    result = read(fd, buffer, frame_size - num_read);

	    continue;

	}

        gettimeofday(&tv, NULL);

        if (!tv_started.tv_sec) {

    	    memcpy(&tv_started, &tv, sizeof(struct timeval));

    	    if (tv_started.tv_usec >= (1000000 / setup.fps)) 

    		tv_started.tv_usec -= (1000000 / setup.fps);

    	    else {

    		tv_started.tv_sec--;

    		tv_started.tv_usec += 1000000 - (1000000 / setup.fps);

    	    }

    	    memcpy(&tv_last_written, &tv_started, sizeof(struct timeval));

    	}

        num_read = 0;

        if (broken_frame) {

    	    lost++;

        } else {

    	    err = fastwriter_commit(fw);

    	    frames++;

        }

	fastwriter_get_stats(fw, &stats);

	if (stats.buffer_used > buf_max) buf_max = stats.buffer_used;

	if ((tv.tv_sec - tv_last_written.tv_sec) >= WRITE_INTERVAL) {

	    last_duration = (tv.tv_sec - tv_last_written.tv_sec) * 1000000 + (tv.tv_usec - tv_last_written.tv_usec);

/*	    expected = (tv.tv_sec - tv_last_summary.tv_sec) * setup.fps + (tv.tv_usec - tv_last_summary.tv_usec) * setup.fps / 1000000;

	    delta = expected - lost - frames;

    	    if (delta > 0) delta--;*/

	    printf("Lost  %6.2lf%% (% 8lu of % 8lu), %9.3lf GB at %8.3lf MB/s, buf:%6.2lf%%\n", 100.*(lost - last_lost) / (lost + frames - last_lost - last_frames), lost - last_lost, lost + frames - (last_lost + last_frames), 1. * frame_size * (frames - last_frames) / 1024 / 1024 / 1024, 1000000. * frame_size * (frames - last_frames) / last_duration / 1024 / 1024, 100.*buf_max/stats.buffer_size);

	    if (((++writeouts)%WRITE_SUMMARY)==0) {

		duration = (tv.tv_sec - tv_started.tv_sec) * 1000000 + (tv.tv_usec - tv_started.tv_usec);

		expected = (tv.tv_sec - tv_started.tv_sec) * setup.fps + round(1.*(tv.tv_usec - tv_started.tv_usec)*setup.fps/1000000);

	        delta = expected - lost - frames;

    		if ((delta > 1)||(delta < 1))

    		    printf(" *** Unexpected frame rate: %.2lf (%lu), delta: %li (%lu, %lu)\n", 1000000. * (frames + lost) / duration, setup.fps, delta, lost + frames, expected); 

		printf("Total %6.2lf%% (% 8lu of % 8lu), %9.3lf GB at %8.3lf MB/s, buf:%6.2lf%%\n", 100.*lost / (lost + frames), lost, lost + frames, 1. * frames * frame_size / 1024 / 1024 / 1024, 1000000. * frames * frame_size / duration / 1024 / 1024,  100.*stats.buffer_max / stats.buffer_size);

	    }

	    buf_max = 0;

	    memcpy(&tv_last_written, &tv, sizeof(struct timeval));

	    last_frames = frames;

	    last_lost = lost;

	}

	result = read(fd, buffer, frame_size);

    }

    printf("Wrote %lu GB\n", frame_size * frames / 1024 / 1024 / 1024);

    g_thread_join(thr);

    close(fd);

    free(buffer);

    fastwriter_close(fw);

    fastwriter_destroy(fw);

    unlink(fifo_name);

    return 0;

}