/alps/fwbench

#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 <getopt.h>

#include <stdarg.h>

#include <fastwriter.h>

#ifdef USE_UFO_GENERATOR

# include <ufo/ufo-graph.h>

#endif /* USE_UFO_GENERATOR */

#include "config.h"

//#define USE_FIFO

#define WRITE_RANDOM "/dev/frandom"

#define WRITE_INTERVAL 1

#define WRITE_SUMMARY 5

#define DELTA_TOLERANCE 5

#ifdef USE_UFO_GENERATOR

# ifndef USE_FIFO

#  define USE_FIFO

# endif /* !USE_FIFO */

#endif /* USE_UFO_GENERATOR */

const char *fifo_name = ".fifo";

struct setup_s {

    size_t uc;

    size_t width;

    size_t height;

    size_t bpp;

    size_t fps;

    size_t fpf;

    size_t iters;

    double tolerance;

    const char *output;

    const char *fs;

    int verbose;

    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;

    char *current_output;

};

typedef struct setup_s setup_t;

typedef enum {

    OPT_OUTPUT = 'o',

    OPT_BUFFER = 'b',

    OPT_SIZE = 's',

    OPT_TIME = 't',

    OPT_RATE = 'r',

    OPT_FRAMES = 'n',

    OPT_GEOMETRY = 'g',

    OPT_FPF = 'f',

    OPT_MISSING = 'm',

    OPT_QUIET = 'q',

    OPT_UC = 'u',

    OPT_HELP = 'h'

} options_t;

void Usage(int argc,  char* const argv[], const char *format, ...) {

    if (format) {

	va_list ap;

	va_start(ap, format);

	printf("Error %i: ", errno);

	vprintf(format, ap);

	printf("\n");

	va_end(ap);

        printf("\n");

    }

    printf(

"Usage:\n"

" %s [options]\n"

"   -o <file|device>	- Output to file/device [/dev/null]\n"

"			use %%zu to replace with frame number\n"

"   -f <number>		- Number of frames per file [all]\n"

"   -b size		- Buffer Size (MB)\n"

"   -r rate		- Write rate (MB/s)\n"

"   -s size		- Total size of data to write (GB)\n"

"   -t run_time		- Run time (s)\n"

"   -n frames		- Number of frames to write\n"

"   -g <width>x<height>	- Geometry [1024]\n"

"   -g <pixels>		- Number of megapixels [0.7]\n"

"   -m <percent>		- Tolerable missing frames  [100%%]\n"

"   -u <size>		- Use uncompressable data, repeating block (MB)\n"

"   -q			- Quiete\n"

"   -h			- Help\n"

"\n\n",

argv[0]);

    exit(0);

}

#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_verbosity(setup_t *setup, int level) {

    setup->verbose = level;

}

static void set_current_output(setup_t *setup, size_t frame) {

    sprintf(setup->current_output, setup->output, frame);

}

static void set_output(setup_t *setup, const char *name) {

    setup->output = name;

    setup->fs = name;

    setup->current_output = malloc(strlen(setup->output) + 32);

    g_assert(setup->current_output);

    set_current_output(setup, 0);

}

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_pixels(setup_t *setup, size_t pixels) {

    int width = sqrt(4 * pixels / 3);

    set_dim(setup, width, width * 3 / 4);

}

static void set_speed(setup_t *setup, size_t speed) {

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

}

static void set_iters(setup_t *setup, size_t iters) {

    setup->iters = iters;

}

static void set_time(setup_t *setup, size_t run_time) {

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

}

static void set_size(setup_t *setup, size_t size) {

    setup->iters = size / setup->frame_size + ((size % setup->frame_size)?1:0);

}

static void set_frames_per_file(setup_t *setup, size_t fpf) {

    setup->fpf = fpf;

}

static void set_fail_tolerance(setup_t *setup, double fail_rate) {

    setup->tolerance = fail_rate;

}

static void request_uncompressable_data(setup_t *setup, size_t uc) {

    setup->uc = uc;

}

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

    g_thread_exit(NULL);

#else /* USE_UFO_GENERATOR */

    size_t i;

    struct timeval tv;

    size_t size = setup->width * setup->height * setup->bpp;

    size_t pos = 0, mult = setup->uc / size;

    char *buffer = malloc((mult + 1) * size);

    double interval = 1000000. / setup->fps, nextus;

    ssize_t tmp;

    size_t nexts;

    g_assert(buffer);

    if (!mult) mult = 1;

# ifdef WRITE_RANDOM

    if (setup->uc) {

	printf("Generating %zu MB of randomness...\n", mult * size / 1024 / 1024);

	int rfd = open(WRITE_RANDOM, O_RDONLY);

	g_assert(rfd >= 0);

	read(rfd, buffer, mult * size);

	close(rfd);

    }

# endif /* WRITE_RANDOM */

    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 + pos * size, size);

	g_assert(res == size);

# else /* USE_FIFO */

	callback(setup, size, buffer + pos * size);

# endif /* USE_FIFO */

	if ((++pos) == mult) pos = 0;

	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 */

    free(buffer);

#endif /* USE_UFO_GENERATOR */

    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;

    gettimeofday(&tv, NULL);

    if (!s->broken_frame) {

	err = fastwriter_push(fw, result, buffer);

    	if (err) {

    	    if (err == EWOULDBLOCK) {

    		if (!s->tolerance) {

    		    if (s->verbose >= 0) printf("Lost frame...\n");

    		    exit(1);

    		}

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

	if ((s->fpf)&&((s->frames%s->fpf) == 0)) {

	    fastwriter_close(s->fw);

	    set_current_output(s, s->frames);

	    err = fastwriter_open(s->fw, s->current_output,  FASTWRITER_FLAGS_OVERWRITE);

	    if (err) {

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

		g_assert(!err);

	    }

	// reopen

    }

    }

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

	if ((s->tolerance > 0)&&(s->tolerance < 100)) {

	    double lost = (100. * s->lost / (s->lost + s->frames));

	    double last_lost = (100. * (s->lost - s->last_lost) / (s->lost + s->frames - s->last_lost - s->last_frames));

	    if ((lost > s->tolerance)||(last_lost > s->tolerance)) {

		if (s->verbose >= 0)

		    printf("Lost %.2lf%% (%lu) frames, total: %.2lf%% (%lu)\n", last_lost, s->lost - s->last_lost, lost, s->lost);

		exit(1);

	    }

	}

	if (s->verbose >= 0) {

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

    	    if (s->verbose >= 0) {

		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;

    unsigned char c;

    const char *sptr;

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

    double mpix = 0;

    size_t width = 1024;

    size_t height = 768;

    size_t speed = 850;

    size_t size = 0;

    size_t uc = 0;

    size_t run_time = 0;

    size_t iters = 0x7FFFFFFF;

    size_t fw_buffer = 0;

    double fail_rate = 100.;

    size_t fpf = 0;

    int quiet = 0;

    fastwriter_stats_t stats;

    g_thread_init(NULL);

#ifdef USE_UFO_GENERATOR

    g_type_init();

#endif /* USE_UFO_GENERATOR */

    while ((c = getopt(argc, argv, "hqo:s:u:t:r:n:g:f:m:b:")) != (unsigned char)-1) {

	switch (c) {

	 case OPT_OUTPUT:

	    out = optarg;

	    break;

	 case OPT_BUFFER:

            fw_buffer = atol(optarg);

	    break;

         case OPT_SIZE:

            size = atol(optarg);

	    break;

	 case OPT_UC:

	    uc = atol(optarg);

	    break;

         case OPT_TIME:

            run_time = atol(optarg);

	    break;

         case OPT_FRAMES:

            iters = atol(optarg);

	    break;

         case OPT_RATE:

            speed = atoi(optarg);

            break;

         case OPT_GEOMETRY:

            sptr = strchr(optarg, 'x');

            if (sptr) {

        	width = atol(optarg);

        	height = atol(sptr + 1);

            } else {

        	mpix = atof(optarg);

            }

            break;

         case OPT_FPF:

            fpf = atol(optarg);

            break;

         case OPT_MISSING:

            fail_rate = atof(optarg);

            break;

        case OPT_QUIET:

    	    quiet = 1;

	    break;

         case OPT_HELP:

	    Usage(argc, argv, NULL);

     }

    }

#ifdef USE_FIFO

    unlink(fifo_name);

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

#endif /* USE_FIFO */

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

    if (quiet) set_verbosity(&setup, -1);

    set_output(&setup, out);

    if (mpix) set_pixels(&setup, mpix * 1000000);

    else set_dim(&setup, width, height);

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

    if (size) set_size(&setup, size * 1024 * 1024 * 1024);

    else if (run_time) set_time(&setup, run_time);

    else set_iters(&setup, iters);

    set_fail_tolerance(&setup, fail_rate);

    set_frames_per_file(&setup, fpf);

    request_uncompressable_data(&setup, uc * 1024 * 1024);

    setup.fw = fastwriter_init(setup.fs, FASTWRITER_FLAGS_OVERWRITE);

    g_assert(setup.fw);

    if (fw_buffer) 

	fastwriter_set_buffer_size(setup.fw, fw_buffer * 1024 * 1024);

    else

	fastwriter_set_buffer_size(setup.fw, FASTWRITER_BUFFER_MAX);

    err = fastwriter_open(setup.fw, setup.current_output,  FASTWRITER_FLAGS_OVERWRITE);

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

    g_assert(!err);

    fastwriter_get_stats(setup.fw, &stats);

    if (!quiet)

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

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

    }

    if (!quiet)

	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(setup.fw);

    fastwriter_destroy(setup.fw);

#ifdef USE_FIFO

    unlink(fifo_name);

#endif /* USE_FIFO */

    return 0;

}