敗者ツリーの多重バランス集計外部ソート

外部ソートの基本構想
72 KBのファイルがあると仮定し、18 K個の整数が格納され、ディスク内の物理ブロックのサイズは4 KBであり、ファイルを18グループに分け、各グループはちょうど4 KBである.
まず18回の内部ソートにより、18組のデータを順番に並べ、初期の18個の集計セグメントR 1~R 18が得られ、集計セグメントごとに1024個の整数が得られる.
次に、この18個の集計セグメントについて、4パスバランス集計ソートを使用します.
1回目の集計:集計セグメントを5つ生成
R11   R12    R13    R14    R15
内
R 11は、{R 1,R 2,R 3,R 4}のデータを合成したものである
R 12は、{R 5,R 6,R 7,R 8}のデータを合成したものである
R 13は、{R 9,R 10,R 11,R 12}のデータを統合したものである
R 14は、{R 13,R 14,R 15,R 16}のデータを統合したものである
R 15は、{R 17,R 18}のデータを統合したものである
この5つの集計セグメントのデータを5つのファイルに書き込みます.
foo_1.dat    foo_2.dat    foo_3.dat     foo_4.dat     foo_5.dat
 
2回目の集計:1回目の集計で生成された5つのファイルからデータを読み出し、集計し、2つの集計セグメントを生成する
R21  R22
ここで、R 21は{R 11,R 12,R 13,R 14}のデータを合成したものである
ここでR 22は{R 15}におけるデータの合成によるものである
この2つの集計セグメントを2つのファイルに書き込む
bar_1.dat   bar_2.dat
 
3回目の集計:2回目の集計で生成された2つのファイルからデータを読み出し、集計し、1つの集計セグメントを生成する
R31
R 31は、{R 21,R 22}のデータを合成したものである
このファイルを1つのファイルに書き込む
foo_1.dat
これが最終的にソートされたファイルです.
 
二敗者ツリーを使用して集計ソートを高速化する
外部ソートに最も時間がかかる操作時のディスク読み書き、m個の初期集計セグメントがある場合、k路バランスの集計ソート、ディスク読み書き回数は
|logkm|,kの値を大きくするとディスクの読み書き回数が減少することがわかるが,kの値を大きくするとkパスマージという負の効果をもたらす.
の場合はアルゴリズムの複雑さが増すので、一例を見てみましょう.
n個の整数をkグループに分けて、各グループの整数はすべてすでに並べ替えて、今kグループのデータを1グループの順序の整数に合併して、アルゴリズムの複雑さを求めます
u1: xxxxxxxx
u2: xxxxxxxx
u3: xxxxxxxx
.......
uk: xxxxxxxx
アルゴリズムのステップは、k個のグループの最初の要素から最小の数を選択するたびに、新しいグループに追加し、毎回k-1回比較するので、
アルゴリズム複雑度はO((n−1)*(k−1))であるが,敗者ツリーを用いるとO(logk)の複雑度で最小の数を得ることができ,アルゴリズムが複雑である.
度はO((n−1)*logk)となり,外部ソートというデータ量の大きいソートにとって,これは小さくない向上である.
 
敗者ツリーの作成と調整については、清華大学「データ構造-C言語版」を参照してください.
 
三バイナリテストデータの生成
Linux端末を開き、コマンドを入力
dd if=/dev/urandom of=random.dat bs=1M count=512
これにより、現在のディレクトリの下に512 Mの大きなバイナリファイルが生成され、ファイル内のデータはランダムであり、ファイルは4バイトごとに読み出される.
1個の整数として見ると、128 M個のランダム整数を得ることに相当する.
 
四プログラム実装
 

#include <assert.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>

#define MAX_INT ~(1<<31)
#define MIN_INT 1<<31

//#define DEBUG

#ifdef DEBUG
#define debug(...) debug( __VA_ARGS__) 
#else
#define debug(...)
#endif

#define MAX_WAYS 100

typedef struct run_t {
	int *buf;		/*       */
	int length;		/*            */
	int offset;		/*               */
	int idx;		/*        */
} run_t;

static unsigned int K;				/* K    */
static unsigned int BUF_PAGES;		/*        page */
static unsigned int PAGE_SIZE;		/* page    */
static unsigned int BUF_SIZE;		/*       , BUF_SIZE = BUF_PAGES*PAGE_SIZE */

static int *buffer;					/*       */

static char input_prefix[] = "foo_";
static char output_prefix[] = "bar_";

static int ls[MAX_WAYS];			/* loser tree */

void swap(int *p, int *q);
int partition(int *a, int s, int t);
void quick_sort(int *a, int s, int t);
void adjust(run_t ** runs, int n, int s);
void create_loser_tree(run_t **runs, int n);
long get_time_usecs();
void k_merge(run_t** runs, char* input_prefix, int num_runs, int base, int n_merge);
void usage();


int main(int argc, char **argv)
{
	char 				filename[100];
	unsigned int	data_size;
	unsigned int 	num_runs;				/*              */
	unsigned int	num_merges;				/*               num_merges = num_runs/K */
	unsigned int	run_length;				/*       ，      */
	unsigned int	num_runs_in_merge;		/*     merge K runs    ，     merge    K runs */
	int					fd, rv, i, j, bytes;
	struct stat 		sbuf;

	if (argc != 3) {
		usage();
		return 0;
	}
	long start_usecs = get_time_usecs();

	strcpy(filename, argv[1]);
	fd = open(filename, O_RDONLY);
	if (fd < 0) {
		printf("can't open file %s
", filename);
		exit(0);
	}
	rv = fstat(fd, &sbuf);
	data_size = sbuf.st_size;

	K = atoi(argv[2]);
	PAGE_SIZE = 4096;							/* page = 4KB */
	BUF_PAGES = 32;
	BUF_SIZE = PAGE_SIZE*BUF_PAGES;
	num_runs = data_size / PAGE_SIZE;			/*          ，      4096 byte,  1024    */
	buffer = (int *)malloc(BUF_SIZE);

	run_length = 1;
	run_t **runs = (run_t **)malloc(sizeof(run_t *)*(K+1));
	for (i = 0; i < K; i++) {
		runs[i] = (run_t *)malloc(sizeof(run_t));
		runs[i]->buf = (int *)calloc(1, BUF_SIZE+4);
	}
	while (num_runs > 1) {
		num_merges = num_runs / K;
		int left_runs = num_runs % K;
		if(left_runs > 0) num_merges++;
		for (i = 0; i < num_merges; i++) {
			num_runs_in_merge = K;
			if ((i+1) == num_merges && left_runs > 0) {
				num_runs_in_merge = left_runs;
			}
			int base = 0;
			printf("Merge %d of %d,%d ways
", i, num_merges, num_runs_in_merge);
			for (j = 0; j < num_runs_in_merge; j++) {
				if (run_length == 1) {
					base = 1;
					bytes = read(fd, runs[j]->buf, PAGE_SIZE);
					runs[j]->length = bytes/sizeof(int);
					quick_sort(runs[j]->buf, 0, runs[j]->length-1);
				} else {
					snprintf(filename, 20, "%s%d.dat", input_prefix, i*K+j);
					int infd = open(filename, O_RDONLY);
					bytes = read(infd, runs[j]->buf, BUF_SIZE);
					runs[j]->length = bytes/sizeof(int);
					close(infd);	
				}
				runs[j]->idx = 0;
				runs[j]->offset = bytes;
			}
			k_merge(runs, input_prefix, num_runs_in_merge, base, i);
		}

		strcpy(filename, output_prefix);
		strcpy(output_prefix, input_prefix);
		strcpy(input_prefix, filename);

		run_length *= K;
		num_runs = num_merges;
	}

	for (i = 0; i < K; i++) {
		free(runs[i]->buf);
		free(runs[i]);
	}
	free(runs);
	free(buffer);
	close(fd);

	long end_usecs = get_time_usecs();
	double secs = (double)(end_usecs - start_usecs) / (double)1000000;
	printf("Sorting took %.02f seconds.
", secs);
	printf("sorting result saved in %s%d.dat.
", input_prefix, 0);

	return 0;
}

void k_merge(run_t** runs, char* input_prefix, int num_runs, int base, int n_merge)
{
	int bp, bytes, output_fd;
	int live_runs = num_runs;
	run_t *mr;
	char filename[20];

	bp = 0;
	create_loser_tree(runs, num_runs);

	snprintf(filename, 100, "%s%d.dat", output_prefix, n_merge);
	output_fd = open(filename, O_CREAT|O_WRONLY|O_TRUNC, 
			S_IRWXU|S_IRWXG);
	if (output_fd < 0) {
		printf("create file %s fail
", filename);
		exit(0);
	}

	while (live_runs > 0) {
		mr = runs[ls[0]];
		buffer[bp++] = mr->buf[mr->idx++];
		//        
		if (bp*4 == BUF_SIZE) {
			bytes = write(output_fd, buffer, BUF_SIZE);
			bp = 0;
		}
		// mr        
		if (mr->idx == mr->length) {
			snprintf(filename, 20, "%s%d.dat", input_prefix, ls[0]+n_merge*K);
			if (base) {
				mr->buf[mr->idx] = MAX_INT;
				live_runs--;
			} else {
				int fd = open(filename, O_RDONLY);
				lseek(fd, mr->offset, SEEK_SET);
				bytes = read(fd, mr->buf, BUF_SIZE);
				close(fd);
				if (bytes == 0) {
					mr->buf[mr->idx] = MAX_INT;
					live_runs--;
				}
				else {
					mr->length = bytes/sizeof(int);
					mr->offset += bytes;
					mr->idx = 0;
				}
			}
		}
		adjust(runs, num_runs, ls[0]);
	}
	bytes = write(output_fd, buffer, bp*4);
	if (bytes != bp*4) {
		printf("!!!!!! Write Error !!!!!!!!!
");
		exit(0);
	}
	close(output_fd);
}

long get_time_usecs()
{
	struct timeval time;
	struct timezone tz;
	memset(&tz, '\0', sizeof(struct timezone));
	gettimeofday(&time, &tz);
	long usecs = time.tv_sec*1000000 + time.tv_usec;

	return usecs;
}

void swap(int *p, int *q)
{
	int		tmp;

	tmp = *p;
	*p = *q;
	*q = tmp;
}

int partition(int *a, int s, int t)
{
	int		i, j;	/* i    a[s]...a[t-1], j    x         */

	for (i = j = s; i < t; i++) {
		if (a[i] < a[t]) {
			swap(a+i, a+j);
			j++;
		}
	}
	swap(a+j, a+t);

	return j;
}

void quick_sort(int *a, int s, int t)
{
	int 	p;

	if (s < t) {
		p = partition(a, s, t);
		quick_sort(a, s, p-1);
		quick_sort(a, p+1, t);
	}
}

void adjust(run_t ** runs, int n, int s)
{
	int t, tmp;

	t = (s+n)/2;
	while (t > 0) {
		if (s == -1) {
			break;
		}
		if (ls[t] == -1 || runs[s]->buf[runs[s]->idx] > runs[ls[t]]->buf[runs[ls[t]]->idx]) {
			tmp = s;
			s = ls[t];
			ls[t] = tmp;
		}
		t >>= 1;
	}
	ls[0] = s;
}

void create_loser_tree(run_t **runs, int n)
{
	int		i;

	for (i = 0; i < n; i++) {
		ls[i] = -1;
	}
	for (i = n-1; i >= 0; i--) {
		adjust(runs, n, i);
	}
}

void usage()
{
	printf("sort <filename> <K-ways>
");
	printf("\tfilename: filename of file to be sorted
");
	printf("\tK-ways: how many ways to merge
");
	exit(1);
}

 
 
五コンパイル運転
gcc sort.c -o sort -g
./sort random.dat 64
64ウェイバランスでrandomを集計する.dat内のデータは外部ソートを行う.I 5プロセッサ,4 Gメモリのハードウェア環境において,実験結果は以下の通りである.
ファイルサイズに時間がかかる
128 M 14.72秒
256 M 30.89秒
512 M 71.65秒
1 G 169.18秒
 
六バイナリファイルを読み込み、ソート・ノットを表示する
 

#include <assert.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>

int main(int argc, char **argv)
{
	char *filename = argv[1];
	int *buffer = (int *)malloc(1<<20);
	struct stat 	sbuf;
	int rv, data_size, i, bytes, fd;

	fd = open(filename, O_RDONLY);
	if (fd < 0) {
		printf("%s not found!
", filename);
		exit(0);
	}
	rv = fstat(fd, &sbuf);
	data_size = sbuf.st_size;

	bytes = read(fd, buffer, data_size);
	for (i = 0; i < bytes/4; i++) {
		printf("%d ", buffer[i]);
		if ((i+1) % 10 == 0) {
			printf("
");
		}
	}
	printf("
");
	close(fd);
	free(buffer);
	return 0;
}