胜者树和败者树
轉自：http://blog.csdn.net/whz_zb/article/details/7425152
勝者樹與敗者樹??
?????? 勝者樹和敗者樹都是完全二叉樹，是樹形選擇排序的一種變型。每個葉子結點相當于一個選手，每個中間結點相當于一場比賽，每一層相當于一輪比賽。
?
????? 不同的是，勝者樹的中間結點記錄的是勝者的標號；而敗者樹的中間結點記錄的敗者的標號。
?
?????? 勝者樹與敗者樹可以在log(n)的時間內找到最值。任何一個葉子結點的值改變后，利用中間結點的信息，還是能夠快速地找到最值。在k路歸并排序中經常用到。
?
一、勝者樹
??????
?????? 勝者樹的一個優點是，如果一個選手的值改變了，可以很容易地修改這棵勝者樹。只需要沿著從該結點到根結點的路徑修改這棵二叉樹，而不必改變其他比賽的結果。
Fig. 1
Fig.1是一個勝者樹的示例。規定數值小者勝。
1.?????????b3 PK b4，b3勝b4負，內部結點ls[4]的值為3；
2.?????????b3 PK b0，b3勝b0負，內部結點ls[2]的值為3；
3.?????????b1 PK b2，b1勝b2負，內部結點ls[3]的值為1；
4.?????????b3 PK b1，b3勝b1負，內部結點ls[1]的值為3。.
當Fig. 1中葉子結點b3的值變為11時，重構的勝者樹如Fig. 2所示。
1.?????????b3 PK b4，b3勝b4負，內部結點ls[4]的值為3；
2.?????????b3 PK b0，b0勝b3負，內部結點ls[2]的值為0；
3.?????????b1 PK b2，b1勝b2負，內部結點ls[3]的值為1；
4.?????????b0 PK b1，b1勝b0負，內部結點ls[1]的值為1。.
Fig. 2
?
?
二、敗者樹
?
?????? 敗者樹是勝者樹的一種變體。在敗者樹中，用父結點記錄其左右子結點進行比賽的敗者，而讓勝者參加下一輪的比賽。敗者樹的根結點記錄的是敗者，需要加一個結點來記錄整個比賽的勝利者。采用敗者樹可以簡化重構的過程。
?
Fig. 3
Fig. 3是一棵敗者樹。規定數大者敗。
1.?????????b3 PK b4，b3勝b4負，內部結點ls[4]的值為4；
2.?????????b3 PK b0，b3勝b0負，內部結點ls[2]的值為0；
3.?????????b1 PK b2，b1勝b2負，內部結點ls[3]的值為2；
4.?????????b3 PK b1，b3勝b1負，內部結點ls[1]的值為1；
5.?????????在根結點ls[1]上又加了一個結點ls[0]=3，記錄的最后的勝者。
敗者樹重構過程如下：
·????????????將新進入選擇樹的結點與其父結點進行比賽：將敗者存放在父結點中；而勝者再與上一級的父結點比較。
·????????????比賽沿著到根結點的路徑不斷進行，直到ls[1]處。把敗者存放在結點ls[1]中，勝者存放在ls[0]中。
Fig. 4
?????? Fig. 4是當b3變為13時，敗者樹的重構圖。
?
?????? 注意，敗者樹的重構跟勝者樹是不一樣的，敗者樹的重構只需要與其父結點比較。對照Fig. 3來看，b3與結點ls[4]的原值比較，ls[4]中存放的原值是結點4，即b3與b4比較，b3負b4勝，則修改ls[4]的值為結點3。同理，以此類推，沿著根結點不斷比賽，直至結束。
?
??????? 由上可知，敗者樹簡化了重構。敗者樹的重構只是與該結點的父結點的記錄有關，而勝者樹的重構還與該結點的兄弟結點有關。
敗者樹 多路平衡歸并外部排序
一 外部排序的基本思路
假設有一個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個隨機整數。
程序：
#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 100typedef 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); }
上文的代碼過于復雜，因為考慮實際情況。下面我寫得程序，條理比較清晰。
#include <iostream> #include <limits> #include <algorithm>using namespace std;const int MaxInt = numeric_limits<int>::max(); const int MinInt = numeric_limits<int>::min(); const int K=4;int a[][8] = {{0,3,5,8,MaxInt},{2,6,9,MaxInt},{1,3,8,MaxInt},{5,9,12,17,MaxInt}};int loserTree[K];//存放內節點 int b[K+1];//存放葉子節點，在初始化的時候，要在最后一位放在最小標志位，這是為了在初始化敗者樹時，在每個節點均是葉子節點的值void swap(int &a,int &b) {int t = a;a = b;b = t; }void Adjust(int leaf) {int parent = (leaf +K)/2;//leaf+K是葉子節點在loserTree中的節點號，(leaf +K)/2是leaf的父節點int winner = leaf;//winner記錄在一路中勝者的下標while (parent >0){if (b[winner] > b[loserTree[parent]])//loserTree只記錄失敗一方{swap(winner,loserTree[parent]);}parent /=2;}loserTree[0] = winner;//0節點只記錄最后的勝者}void CreateLoserTree()//初始化之后，在loserTree[0]中就是最小值 {for (int i=0;i<K;++i){loserTree[i] = K;}for (int i=K-1;i>=0;--i){Adjust(i);}}void kMerge() {int pa[K];//是指向a的指針memset(pa,0,sizeof(pa));for(int i=0;i<K;++i){b[i] = a[i][pa[i]];++pa[i];}b[K] = MinInt;//CreateLoserTree();while (b[loserTree[0]]!=MaxInt){cout<<b[loserTree[0]]<<" ";int i = loserTree[0];b[i] = a[i][pa[i]++];Adjust(i);}cout<<endl; }int main() {kMerge(); }
維護敗者樹， 從更新的葉子節點網上，依次與內部節點比較，將“敗者”更新，勝者往上繼續比較。由于更新節點占用的是之前的最小值的葉子節點，它往上一直到根節點的路徑 與之前的最小值的路徑是完全相同的。內部節點記錄的“敗者”雖然稱為“敗者”，但卻是其左子樹或者右子樹中最小的數。也就是說，只要與“敗者”比較得到的勝者，就 是該子樹中最小的那個數。
總結
