~drizzle-trunk/drizzle/development

/* Copyright (c) 2005 PrimeBase Technologies GmbH

 *

 * PrimeBase XT

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

 *

 * 2009-07-30	Paul McCullagh

 *

 * H&G2JCtL

 */

#include <pthread.h>

#include <stdlib.h>

#include <stddef.h>

#include <stdio.h>

#include <sys/time.h>

#include <unistd.h>

#include <dirent.h>

#include <stdio.h>

#include <sys/stat.h>

#include <fcntl.h>

#include <sys/types.h>

#include <ctype.h>

#include <string.h>

#include <errno.h>

#include <limits.h>

#include <assert.h>

#ifdef __APPLE__

#define MAC

#endif

/*

 * Define this if I/O should pause.

 */

//#define SHOULD_PAUSE

//#define PERIODIC_FLUSH

#define SIM_RECORD_SIZE			221

#ifdef MAC

#define SIM_FILE_SIZE			(256*1024*1024)

#define SIM_WRITE_AMOUNT		(1*1024*1024)

//#define SIM_WRITE_AMOUNT		(221 * 100)

#else

#define SIM_FILE_SIZE			((off_t) (2L*1024L*1024L*1024L))

#define SIM_WRITE_AMOUNT		(8*1024*1024)

#endif

#define SIM_FLUSH_THRESHOLD		(2*1024*1024)

#define SIM_PAUSE_THRESHOLD		(10*1024*1024)

#ifndef SHOULD_PAUSE

#undef  SIM_PAUSE_THRESHOLD

#define SIM_PAUSE_THRESHOLD		0

#endif

#ifndef PERIODIC_FLUSH

#undef  SIM_FLUSH_THRESHOLD

#define SIM_FLUSH_THRESHOLD		0

#endif

#define my_time					unsigned long long

#define u_long					unsigned long

#define TRUE					1

#define FALSE					0

typedef struct SortedRec {

	off_t		sr_offset;

	int			sr_order;

	char		*sr_data;

} SortedRec;

#define SORTED_MAX_RECORDS		10000

#define SORTED_DATA_SIZE		(SORTED_MAX_RECORDS * SIM_RECORD_SIZE)

#define SORTED_BUFFER_SIZE		(256*1024)

typedef struct RewriteRec {

	off_t		rr_offset;

	off_t		rr_size;

} RewriteRec;

#define REWRITE_MAX_RECORDS		1000

/* This is the maximum distance between to blocks that

 * will cause the blocks to be combined and written

 * as one block!

 */

#ifdef MAC

#define REWRITE_BLOCK_DISTANCE	(64*1024)

#else

#define REWRITE_BLOCK_DISTANCE	(1024*1024)

#endif

#define REWRITE_RECORD_LIMIT	256

typedef struct File {	

	int			file_fh;

	int			fi_monitor_index;

	char		fi_file_path[200];

	char		fi_test_name[200];

	char		fi_monitor_name[10];

	int			fi_monitor_active;

	my_time		total_time;

	my_time		flush_start;

	my_time		flush_time;

	u_long		flush_count;

	my_time		last_flush_time;

	u_long		last_flush_count;

	my_time		write_start;

	my_time		write_time;

	u_long		write_count;

	off_t		last_block_offset;

	size_t		last_block_size;

	u_long		block_write_count;

	my_time		last_write_time;

	u_long		last_write_count;

	my_time		read_start;

	my_time		read_time;

	u_long		read_count;

	my_time		last_read_time;

	u_long		last_read_count;

	/* Sorted file I/O */

	int			sf_rec_count;

	SortedRec	sf_records[SORTED_MAX_RECORDS];

	size_t		sf_alloc_pos;

	int			sf_order;

	char		sf_data[SORTED_DATA_SIZE];

	char		sf_buffer[SORTED_BUFFER_SIZE];

	/* Re-write sync: */

	off_t		rs_min_block_offset;

	off_t		rs_max_block_offset;

	size_t		rs_flush_block_total;

	size_t		rs_rec_count;

	RewriteRec	rs_records[REWRITE_MAX_RECORDS];

	void		(*fi_write)(struct File *f, void *block, size_t size, off_t start);

	void		(*fi_sync)(struct File *f);

	void		(*fi_write_all)(struct File *f);

} File;

/* -------------------  TIMING ------------------- */

static my_time my_clock(void)

{

	static my_time	my_start_clock = 0;

	struct timeval	tv;

	my_time			now;

	gettimeofday(&tv, NULL);

	now = (my_time) tv.tv_sec * (my_time) 1000000 + tv.tv_usec;

	if (my_start_clock)

		return now - my_start_clock;

	my_start_clock = now;

	return 0;

}

/* -------------------  ERRORS ------------------- */

static void print_error(char *file, int err)

{

	printf("ERROR %s: %s\n", file, strerror(err));

}

static void error_exit(char *file, int err)

{

	print_error(file, err);

	exit(1);

}

static void fatal_error(char *message)

{

	printf("%s", message);

	exit(1);

}

/* -------------------  ERRORS ------------------- */

static void *my_bsearch(void *key, register const void *base, size_t count, size_t size, size_t *idx, int (*compare)(void *key, void *rec))

{

	register size_t		i;

	register size_t		guess;

	register int		r;

	i = 0;

	while (i < count) {

		guess = (i + count - 1) >> 1;

		r = compare(key, ((char *) base) + guess * size);

		if (r == 0) {

			*idx = guess;

			return ((char *) base) + guess * size;

		}

		if (r < 0)

			count = guess;

		else

			i = guess + 1;

	}

	*idx = i;

	return NULL;

}

/* -------------------  MONITORING THREAD ------------------- */

/* Monitor file types: */

#define MAX_FILES			10

static int		monitor_file_count;

static int		monitor_files_in_use;

static int		monitor_running;

static	File	*monitor_files[MAX_FILES];

static void monitor_file(File *f, const char *test_name, const char *monitor_name)

{

	if (monitor_file_count == MAX_FILES)

		fatal_error("Too many files to monitor");

	monitor_files[monitor_file_count] = f;

	f->fi_monitor_index = monitor_file_count;

	sprintf(f->fi_file_path, "test-data-%d", (int) monitor_file_count);

	strcpy(f->fi_test_name, test_name);

	strcpy(f->fi_monitor_name, monitor_name);

	monitor_file_count++;

	f->fi_monitor_active = TRUE;

	monitor_files_in_use++;

}

static void unmonitor_file(File *f)

{

	/* Wait for the last activity to be reported! */

	while (monitor_running && f->fi_monitor_active) {

		usleep(100);

	}

	monitor_files_in_use--;

	monitor_files[f->fi_monitor_index] = NULL;

}

static void print_header()

{

	File	*f;

	int		i;

	printf("time ");

	for (i=0; i<MAX_FILES; i++) {

		if ((f = monitor_files[i])) {

			printf("%7s %5s %9s %5s %9s %5s ", f->fi_monitor_name, f->fi_monitor_name, f->fi_monitor_name, f->fi_monitor_name, f->fi_monitor_name, f->fi_monitor_name);

		}

	}

	printf("\n");

	printf("     ");

	for (i=0; i<MAX_FILES; i++) {

		if ((f = monitor_files[i])) {

			printf("%7s %5s %9s %5s %9s %5s ", "flush", "ftime", "write", "wtime", "read", "rtime");

		}

	}

	printf("\n");

}

static void *iotest_monitor(void *data)

{

	File	*f;

	int		i;

	int		row = 0;

	my_time	curr, last;

	my_time	now, fstart, wstart, rstart;

	int		version = 0;

	int		activity;

	my_time	curr_flush_time;

	u_long	curr_flush_count;

	my_time	curr_write_time;

	u_long	curr_write_count;

	my_time	curr_read_time;

	u_long	curr_read_count;

	my_time	flush_time;

	u_long	flush_count;

	my_time	write_time;

	u_long	write_count;

	my_time	read_time;

	u_long	read_count;

	monitor_running = TRUE;

	last = my_clock();

	for (;;) {

		curr = my_clock();

		if (!monitor_files_in_use)

			goto wait_phase;

		if ((row % 20) == 0 || version != monitor_file_count) {

			version = monitor_file_count;

			print_header();

		}

		printf("%4.0f ", ((double) curr - (double) last) / (double) 1000);

		activity = FALSE;

		for (i=0; i<MAX_FILES; i++) {

			if ((f = monitor_files[i])) {

				curr_flush_time = f->flush_time;

				fstart = f->flush_start;

				curr_write_time = f->write_time;

				wstart = f->write_start;

				curr_read_time = f->read_time;

				rstart = f->read_start;

				now = my_clock();

				if (fstart)

					curr_flush_time += now - fstart;

				flush_time = curr_flush_time - f->last_flush_time;

				f->last_flush_time = curr_flush_time;

				curr_flush_count = f->flush_count;

				flush_count = curr_flush_count - f->last_flush_count;

				f->last_flush_count = curr_flush_count;

				if (wstart)

					curr_write_time += now - wstart;

				write_time = curr_write_time - f->last_write_time;

				f->last_write_time = curr_write_time;

				curr_write_count = f->write_count;

				write_count = curr_write_count - f->last_write_count;

				f->last_write_count = curr_write_count;

				if (rstart)

					curr_read_time += now - rstart;

				read_time = curr_read_time - f->last_read_time;

				f->last_read_time = curr_read_time;

				curr_read_count = f->read_count;

				read_count = curr_read_count - f->last_read_count;

				f->last_read_count = curr_read_count;

				printf("%7lu %5.0f %9.2f %5.0f %9.2f %5.0f ", flush_count, (double) flush_time / (double) 1000,

					(double) write_count / (double) 1024, (double) write_time / (double) 1000,

					(double) read_count / (double) 1024, (double) read_time / (double) 1000);

				if (flush_count || flush_time || write_count || write_time || read_count || read_time) {

					f->fi_monitor_active = TRUE;

					activity = TRUE;

				}

				else

					f->fi_monitor_active = FALSE;

			}

		}

		printf("\n");

		row++;

		wait_phase:

		/* Leave the loop, only when there is no more activity. */

		if (!monitor_running && !activity)

			break;

		do {

			usleep(1000);

		} while (my_clock() - curr < 1000000);

		last = curr;

	}

	return NULL;

}

/* -------------------  BASIC FILE I/O ------------------- */

#define PREFILL				0

#define SET_EOF				1

#define TRUNCATE			2

#define XT_MASK				((S_IRUSR | S_IWUSR) | (S_IRGRP | S_IWGRP) | (S_IROTH))

static void create_file(File *f, size_t size, int type)

{

	int		fd;

	size_t	tfer;

	char	*block;

	size_t	i;

	off_t	eof;

	off_t	offset;

	if (!(block = (char *) malloc(512)))

		error_exit(f->fi_file_path, errno);

	for (i=0; i<512; i++)

		block[i] = (char) i;

	fd = open(f->fi_file_path, O_CREAT | O_RDWR, XT_MASK);

	if (fd == -1)

		error_exit(f->fi_file_path, errno);

	eof = lseek(fd, 0, SEEK_END);

	if (type == PREFILL && size == eof)

		goto done;

	if (ftruncate(fd, 0) == -1)

		error_exit(f->fi_file_path, errno);

	if (type == SET_EOF) {

		if (size > 512)

			offset = size - 512;

		else

			offset = 0;

		if (pwrite(fd, block, 512, offset) != 512)

			error_exit(f->fi_file_path, errno);

	}

	else {

		offset = 0;

		while (size > 0) {

			tfer = size;

			if (tfer > 512)

				tfer = 512;

			if (pwrite(fd, block, tfer, offset) != tfer)

				error_exit(f->fi_file_path, errno);

			size -= tfer;

			offset += tfer;

		}

	}

	if (fsync(fd) == -1)

		error_exit(f->fi_file_path, errno);

	done:

	close(fd);

	free(block);

}

/*

static void delete_file(char *file)

{

	unlink(file);

}

*/

static void write_file(File *f, void *block, size_t size, off_t offset)

{

	my_time t, s;

	s = my_clock();

	f->write_start = s;

	if (pwrite(f->file_fh, block, size, offset) != size)

		error_exit(f->fi_file_path, errno);

	t = my_clock();

	f->write_start = 0;

	f->write_time += (t - s);

	f->write_count += size;

	/* Does this block touch the previous block? */

	if (f->last_block_offset == -1 ||

		offset < f->last_block_offset ||

		offset > f->last_block_offset + f->last_block_size)

		/* If not, it is a new block (with a gap): */

		f->block_write_count++;

	f->last_block_offset = offset;

	f->last_block_size = size;

}

static void read_file(File *f, void *block, size_t size, off_t start)

{

	my_time t, s;

	s = my_clock();

	f->read_start = s;

	if (pread(f->file_fh, block, size, start) != size)

		error_exit(f->fi_file_path, errno);

	t = my_clock();

	f->read_start = 0;

	f->read_time += (t - s);

	f->read_count += size;

}

static void sync_file(File *f)

{

	my_time t, s;

	s = my_clock();

	f->flush_start = s;

	if (fsync(f->file_fh) == -1)

		error_exit(f->fi_file_path, errno);

	t = my_clock();

	f->flush_start = 0;

	f->flush_time += t - s;

	f->flush_count++;

}

static void new_file(File **ret_f, const char *test_name, const char *monitor_name)

{

	File *f;

	f = malloc(sizeof(File));

	memset(f, 0, sizeof(File));

	f->last_block_offset = (off_t) -1;

	f->rs_min_block_offset = (off_t) -1;

	monitor_file(f, test_name, monitor_name);

	f->fi_write = write_file;

	f->fi_sync = sync_file;

	*ret_f = f;

}

static void open_file(File *f)

{

	f->file_fh = open(f->fi_file_path, O_RDWR, 0);

	if (f->file_fh == -1)

		error_exit(f->fi_file_path, errno);

	f->total_time = my_clock();

}

static void close_file(File *f)

{

	f->fi_sync(f);

	f->total_time = my_clock() - f->total_time;

	if (f->file_fh != -1)

		close(f->file_fh);

	unmonitor_file(f);

	printf("\n=* TEST: %s (%s) *=\n", f->fi_test_name, f->fi_monitor_name);

	printf("Written K:  %.2f\n", (double) f->write_count / (double) 1024);

	printf("Run time:   %.2f ms\n", (double) f->total_time / (double) 1000);

	if (f->rs_flush_block_total > 0)

		printf("Flush blks: %lu\n", f->rs_flush_block_total);

	if (f->write_count > 0) {

		printf("Tot blocks: %lu\n", f->block_write_count);

		printf("Seek time:  %.3f ms\n", (double) f->flush_time / (double) f->block_write_count / (double) 1000);

	}

	printf("\n");

	if (f->write_time)

		printf("Write K/s: %.2f\n", (double) f->write_count * (double) 1000000 / (double) 1024 / (double) f->write_time);

	if (f->read_time)

		printf("Read  K/s: %.2f\n", (double) f->read_count * (double) 1000000 / (double) 1024 / (double) f->read_time);

	if (f->flush_time)

		printf("Flush K/s: %.2f\n", (double) f->write_count * (double) 1000000 / (double) 1024 / (double) f->flush_time);

	if (f->write_time + f->read_time + f->flush_time)

		printf("Total K/s: %.2f\n", (double) f->write_count * (double) 1000000 / (double) 1024 / ((double) f->write_time + (double) f->read_time + (double) f->flush_time));

	printf("=*=\n");

}

/* -------------------  SORTED I/O ------------------- */

/* Sort records before they are writing.

 * Options are also added to write records,

 * in large blocks. This requires reading the block

 * first.

 */

static int compare_rec(const void *a, const void *b)

{

	SortedRec	*ra = (SortedRec *) a;

	SortedRec	*rb = (SortedRec *) b;

	if (ra->sr_offset == rb->sr_offset) {

		if (ra->sr_order == rb->sr_order)

			return 0;

		if (ra->sr_order < rb->sr_order)

			return -1;

		return 1;

	}

	if (ra->sr_offset < rb->sr_offset)

		return -1;

	return 1;

}

static void sorted_write_all(File *f)

{

	SortedRec	*rec;

	int			i;

	qsort(f->sf_records, f->sf_rec_count, sizeof(SortedRec), compare_rec);

	rec = f->sf_records;

	for (i=0; i<f->sf_rec_count; i++) {

		write_file(f, rec->sr_data, SIM_RECORD_SIZE, rec->sr_offset);

		rec++;

	}

	f->sf_rec_count = 0;

	f->sf_alloc_pos = 0;

	f->sf_order = 0;

}

static void sorted_write_rw_all(File *f)

{

	SortedRec	*rec;

	int			i;

	off_t		offset = 0;

	size_t		size = 0;

	qsort(f->sf_records, f->sf_rec_count, sizeof(SortedRec), compare_rec);

	rec = f->sf_records;

	for (i=0; i<f->sf_rec_count; i++) {

		reread:

		if (!size) {

			offset = (rec->sr_offset / 1024) * 1024;

			size = SORTED_BUFFER_SIZE;

			if (offset+(off_t)size > SIM_FILE_SIZE)

				size = (size_t) (SIM_FILE_SIZE - offset);

			read_file(f, f->sf_buffer, size, offset);

		}

		if (rec->sr_offset >= offset && rec->sr_offset+SIM_RECORD_SIZE <= offset+size)

			memcpy(&f->sf_buffer[rec->sr_offset - offset], rec->sr_data, SIM_RECORD_SIZE);

		else {

			write_file(f, f->sf_buffer, size, offset);

			size = 0;

			goto reread;

		}

		rec++;

	}

	f->sf_rec_count = 0;

	f->sf_alloc_pos = 0;

	f->sf_order = 0;

}

static void sorted_write_rw_no_gaps_all(File *f)

{

	SortedRec	*rec;

	int			i;

	off_t		offset;

	size_t		size = 0;

	qsort(f->sf_records, f->sf_rec_count, sizeof(SortedRec), compare_rec);

	rec = f->sf_records;

	if (!f->sf_rec_count)

		goto done;

	offset = (rec->sr_offset / 1024) * 1024;

	for (i=0; i<f->sf_rec_count; i++) {

		reread:

		if (!size) {

			size = SORTED_BUFFER_SIZE;

			if (offset+size > SIM_FILE_SIZE)

				size = SIM_FILE_SIZE - offset;

			read_file(f, f->sf_buffer, size, offset);

		}

		if (rec->sr_offset >= offset && rec->sr_offset+SIM_RECORD_SIZE <= offset+size)

			memcpy(&f->sf_buffer[rec->sr_offset - offset], rec->sr_data, SIM_RECORD_SIZE);

		else {

			write_file(f, f->sf_buffer, size, offset);

			offset += size;

			if (rec->sr_offset < offset)

				offset = (rec->sr_offset / 1024) * 1024;

			size = 0;

			goto reread;

		}

		rec++;

	}

	done:

	f->sf_rec_count = 0;

	f->sf_alloc_pos = 0;

	f->sf_order = 0;

}

static void sorted_sync_file(File *f)

{

	f->fi_write_all(f);

	sync_file(f);

}

static void sorted_write_file(File *f, void *block, size_t size, off_t offset)

{

	SortedRec *rec;

	if (size != SIM_RECORD_SIZE)

		printf("ooops\n");

	if (f->sf_rec_count == SORTED_MAX_RECORDS ||

		f->sf_alloc_pos + size > SORTED_DATA_SIZE) {

		f->fi_write_all(f);

	}

	rec = &f->sf_records[f->sf_rec_count];

	rec->sr_offset = offset;

	rec->sr_order = f->sf_order;

	rec->sr_data = &f->sf_data[f->sf_alloc_pos];

	memcpy(rec->sr_data, block, size);

	f->sf_alloc_pos += size;

	f->sf_rec_count++;

	f->sf_order++;

}

/* -------------------  RE-WRITE FLUSH ------------------- */

/* The idea is that it is better to re-write the file

 * sequentially, then allow the FS to write scattered

 * dirty blocks!

 *

 * This comes from the fact that seeking is a lot more

 * expensive than sequential write.

 */

static void rewrite_all_sync_file(File *f)

{

	off_t	offset = 0;

	size_t	size = 0;

	while (offset < SIM_FILE_SIZE) {

		size = SORTED_BUFFER_SIZE;

		if (offset + size > SIM_FILE_SIZE)

			size = SIM_FILE_SIZE - offset;

		read_file(f, f->sf_buffer, size, offset);

		write_file(f, f->sf_buffer, size, offset);

		offset += size;

	}

	sync_file(f);

}

static void rewrite_min_max_sync_file(File *f)

{

	off_t	offset = 0;

	size_t	size = 0;

	off_t	eof = SIM_FILE_SIZE;

	if (f->rs_min_block_offset != (off_t) -1)

		offset = f->rs_min_block_offset / 1024 * 1024;

	eof = (f->rs_max_block_offset + 1023) / 1024 * 1024;

	if (eof > SIM_FILE_SIZE)

		eof = SIM_FILE_SIZE;

	while (offset < eof) {

		size = SORTED_BUFFER_SIZE;

		if (offset + size > eof)

			size = eof - offset;

		read_file(f, f->sf_buffer, size, offset);

		write_file(f, f->sf_buffer, size, offset);

		offset += size;

	}

	sync_file(f);

}

static void rewrite_min_max_write_file(File *f, void *block, size_t size, off_t offset)

{

	off_t		top_offset;

	write_file(f, block, size, offset);