fwbench/fwbench.c

#define _BSD_SOURCE
#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>

#ifdef USE_UFO_GENERATOR
# include <ufo/ufo-graph.h>
#endif /* USE_UFO_GENERATOR */

#include "config.h"

//#define USE_FIFO
#define FW_BUFFER 4096l
#define WRITE_INTERVAL 1
#define WRITE_SUMMARY 5
#define DELTA_TOLERANCE 5
#define NEWFILE 10

#ifdef USE_UFO_GENERATOR
# ifndef USE_FIFO
#  define USE_FIFO
# endif /* !USE_FIFO */
#endif /* USE_UFO_GENERATOR */

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

    struct timeval tv_started;
    struct timeval tv_last_written;

    int broken_frame;
    unsigned long frames, lost;
    unsigned long last_frames, last_lost;

    size_t frame_size;
    size_t num_read;
    size_t buf_max;
    
    size_t writeouts;
    
    fastwriter_t *fw;
};

typedef struct setup_s setup_t;


#ifdef USE_UFO_GENERATOR
static void handle_error(GError *error) {
    if (error != NULL) {
        g_print("%s\n", error->message); 
        g_error_free(error);
        exit(EXIT_FAILURE);
    }
}
#endif /* USE_UFO_GENERATOR */


static void set_dim(setup_t *setup, size_t width, size_t height) {
    setup->bpp = sizeof(float);
    setup->width = width;
    setup->height = height;

    setup->frame_size = setup->width * setup->height * setup->bpp;
}

static void set_speed(setup_t *setup, size_t speed) {
    setup->fps = 1 + speed / setup->width / setup->height / setup->bpp;
}

static void set_size(setup_t *setup, size_t size) {
    setup->iters = size / setup->width / setup->height / setup->bpp;
}

static void set_time(setup_t *setup, size_t time) {
    setup->iters = time * setup->fps;
}

static int callback(setup_t *s, size_t result, void *buffer);

static void *run(setup_t *setup) {
#ifdef USE_UFO_GENERATOR
    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);
#else /* USE_UFO_GENERATOR */
    size_t i;
    struct timeval tv;
    size_t size = setup->width * setup->height * setup->bpp;
    char buffer[size];
    double interval = 1000000. / setup->fps, nextus;
    ssize_t tmp;
    size_t nexts;

    setup->run_started = 1;

#ifdef USE_FIFO
    int fd = open(fifo_name, O_WRONLY);
    g_assert(fd >= 0);
#endif /* USE_FIFO */

    gettimeofday(&tv, NULL); 
    nexts = tv.tv_sec;
    nextus = tv.tv_usec + interval;
    for (i = 0; i < setup->iters; i++) {
#ifdef USE_FIFO
	ssize_t res = write(fd, buffer, size);
	g_assert(res == size);
#else /* USE_FIFO */
	callback(setup, size, buffer);
#endif /* USE_FIFO */

	tmp = ((size_t)round(nextus)) / 1000000;
	nexts += tmp;
	nextus -= tmp * 1000000;

	gettimeofday(&tv, NULL);
	tmp = (nexts - tv.tv_sec)*1000000 + (nextus - tv.tv_usec);
	if (tmp > 10) usleep(tmp);
	
	nextus += interval;
    }

#ifdef USE_FIFO
    close(fd);
#endif /* USE_FIFO */

#endif /* USE_UFO_GENERATOR */

    g_thread_exit(NULL);
    return NULL;
}

static int callback(setup_t *s, size_t result, void *buffer) {
    int err;
    struct timeval tv;

    fastwriter_t *fw = s->fw;
    fastwriter_stats_t stats;

    size_t duration, last_duration, expected;
    long delta;


    if (!s->broken_frame) {
	err = fastwriter_push(fw, result, buffer);
    	if (err) {
    	    if (err == EWOULDBLOCK) {
		if (s->num_read) fastwriter_cancel(fw);
		s->broken_frame = 1;
    	    } else {
	    	if (err) printf("FastWriter returned error %i\n", err);
		g_assert(!err);
	    }
	}
    }

    s->num_read += result; 

    if (s->num_read < s->frame_size) return 0;

    s->num_read = 0;

    if (s->broken_frame) {
	s->lost++;
	s->broken_frame = 0;
    } else {
	err = fastwriter_commit(fw);
	s->frames++;
    }

    gettimeofday(&tv, NULL);
    if (!s->tv_started.tv_sec) {
    	memcpy(&s->tv_started, &tv, sizeof(struct timeval));
    	if (s->tv_started.tv_usec >= (1000000 / s->fps)) 
    	    s->tv_started.tv_usec -= (1000000 / s->fps);
    	else {
    	    s->tv_started.tv_sec--;
    	    s->tv_started.tv_usec += 1000000 - (1000000 / s->fps);
    	}
    	memcpy(&s->tv_last_written, &s->tv_started, sizeof(struct timeval));
    }

    fastwriter_get_stats(fw, &stats);
    if (stats.buffer_used > s->buf_max) s->buf_max = stats.buffer_used;

    if ((tv.tv_sec - s->tv_last_written.tv_sec) >= WRITE_INTERVAL) {
	last_duration = (tv.tv_sec - s->tv_last_written.tv_sec) * 1000000 + (tv.tv_usec - s->tv_last_written.tv_usec);

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

	if (((++s->writeouts)%WRITE_SUMMARY)==0) {
	    duration = (tv.tv_sec - s->tv_started.tv_sec) * 1000000 + (tv.tv_usec - s->tv_started.tv_usec);
	    expected = (tv.tv_sec - s->tv_started.tv_sec) * s->fps + round(1.*(tv.tv_usec - s->tv_started.tv_usec)*s->fps/1000000);
	    delta = expected - s->lost - s->frames;
    	    if ((delta > DELTA_TOLERANCE)||(delta < -DELTA_TOLERANCE))
    		printf(" *** Unexpected frame rate: %.2lf (%lu), delta: %li (%lu, %lu)\n", 1000000. * (s->frames + s->lost) / duration, s->fps, delta, s->lost + s->frames, expected); 
    		
	    printf("Total %6.2lf%% (% 8lu of % 8lu), %9.3lf GB at %8.3lf MB/s, buf:%6.2lf%%\n", 100. * s->lost / (s->lost + s->frames), s->lost, s->lost + s->frames, 1. * s->frames * s->frame_size / 1024 / 1024 / 1024, 1000000. * s->frames * s->frame_size / duration / 1024 / 1024,  100.*stats.buffer_max / stats.buffer_size);
	}
	
	s->buf_max = 0;
	memcpy(&s->tv_last_written, &tv, sizeof(struct timeval));
	s->last_frames = s->frames;
	s->last_lost = s->lost;
    }
    
    return 0;
}


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

#ifdef USE_FIFO
    GError *gerr;
    GThread *thr;
#endif /* USE_FIFO */

    setup_t setup;
    const char *out = "/dev/null";
    size_t width = 1024;
    size_t height = 768;
    size_t speed = 850;

    fastwriter_t *fw;
    fastwriter_stats_t stats;

    g_thread_init(NULL);

#ifdef USE_UFO_GENERATOR
    g_type_init();
#endif /* USE_UFO_GENERATOR */

    if (argc > 1) {
	out = argv[1];
    }
    
    if (argc > 2) {
	speed = atoi(argv[2]);
    }

#ifdef USE_FIFO
    unlink(fifo_name);
    g_assert(!mkfifo(fifo_name, S_IWUSR | S_IRUSR));
#endif /* USE_FIFO */

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

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

    setup.fw = fw;
    set_dim(&setup, width, height);
    set_speed(&setup, speed * 1024 * 1024);
    set_size(&setup, run_time * speed * 1024 * 1024);

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

    void *buffer = malloc(setup.frame_size);
    g_assert(buffer);

#ifdef USE_FIFO
    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, setup.frame_size);

    while (result > 0) {
	callback(&setup, result, buffer);
        result = read(fd, buffer, setup.frame_size - setup.num_read);
    }

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

    g_thread_join(thr);

    close(fd);
#else /* USE_FIFO */
    run(&setup);
#endif /* USE_FIFO */

    free(buffer);
    fastwriter_close(fw);
    fastwriter_destroy(fw);

#ifdef USE_FIFO
    unlink(fifo_name);
#endif /* USE_FIFO */

    return 0;
}