一 外部排序的基本思路
假設有一個72KB的文件，其中存儲了18K個整數，磁盤中物理塊的大小為4KB，將文件分成18組，每組剛好4KB。
首先通過18次內部排序，把18組數據排好序，得到初始的18個歸並段R1~R18，每個歸並段有1024個整數。
然後對這18個歸並段使用4路平衡歸並排序：
第1次歸並：產生5個歸並段
R11 R12 R13 R14 R15
其中
R11是由{R1,R2,R3,R4}中的數據合並而來
R12是由{R5,R6,R7,R8}中的數據合並而來
R13是由{R9,R10,R11,R12}中的數據合並而來
R14是由{R13,R14,R15,R16}中的數據合並而來
R15是由{R17,R18}中的數據合並而來
把這5個歸並段的數據寫入5個文件：
foo_1.dat foo_2.dat foo_3.dat foo_4.dat foo_5.dat
第2次歸並：從第1次歸並產生的5個文件中讀取數據，合並，產生2個歸並段
R21 R22
其中R21是由{R11,R12,R13,R14}中的數據合並而來
其中R22是由{R15}中的數據合並而來
把這2個歸並段寫入2個文件
bar_1.dat bar_2.dat
第3次歸並：從第2次歸並產生的2個文件中讀取數據，合並，產生1個歸並段
R31
R31是由{R21,R22}中的數據合並而來
把這個文件寫入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
這樣在當前目錄下產生一個512M大的二進制文件，文件內的數據是隨機的，讀取文件，每4個字節
看成1個整數，相當於得到128M個隨機整數。
四 程序實現
[cpp]
#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\n", 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\n", 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.\n", secs);
printf("sorting result saved in %s%d.dat.\n", 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\n", 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 !!!!!!!!!\n");
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>\n");
printf("\tfilename: filename of file to be sorted\n");
printf("\tK-ways: how many ways to merge\n");
exit(1);
}
五 編譯運行
gcc sort.c -o sort -g
./sort random.dat 64
以64路平衡歸並對random.dat內的數據進行外部排序。在I5處理器，4G內存的硬件環境下，實驗結果如下
文件大小 耗時
128M 14.72 秒
256M 30.89 秒
512M 71.65 秒
1G 169.18秒
六 讀取二進制文件，查看排序結
[cpp]
#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!\n", 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("\n");
}
}
printf("\n");
close(fd);
free(buffer);
return 0;
}
