SQLite3源码学习之链表的归并排序

一般数组排序采用快速排序最佳，但是在链表中2个元素的交换不是很方便，所以采用归并排序比较好，先把链表分割成一些已经排好序的子链表，最后再串起来就好了。

在SQLite中对于脏页链表按照页号重新排序采用的就是归并排序。

1.基本思路

子链表存放在一个临时数组里，这个数组定义为

PgHdr *a[N_SORT_BUCKET]

子链表是按照2的幂次方的大小依次存放，也就是说a[0]存放1个元素，a[1]存放2个元素，a[2]存放3个元素，依次类推。

下面举个最简单的例子：

假设初始链表的顺序为23 10 22 21 20,排序步骤如下：

23先存放在a[0]里:23

10和23归并存放到a[1]:10 23

22存放到a[0]里：22

21和22归并后本想存放到a[1],但是a[1]已经有值了，继续归并后存放到a[2]里:10 21 22 23

20存放到a[0]里:20

现在所有子链表已经划分完毕，将其全部归并就得到最后的结果：

10 20 21 22 23

2.代码实现

首先实现2个子链表的合并，result是合并后的链表，pTail 指向result的最后一个结点， pDirty可以理解成pNext(即指向下一个结点)，pA和pB是2个已经排序好的链表的头结点，先比较pA和pB哪个页序比较小，把较小的节点从链表中摘除，添加到pTail->pDirty，再更新pTail即可。

/*
** Merge two lists of pages connected by pDirty and in pgno order.
** Do not bother fixing the pDirtyPrev pointers.
*/
static PgHdr *pcacheMergeDirtyList(PgHdr *pA, PgHdr *pB){
  PgHdr result, *pTail;
  pTail = &result;
  assert( pA!=0 && pB!=0 );
  //遍历2个输入子链表，然后按顺序归并
  for(;;){
    //当pA的页序比较小时
if( pA->pgnopgno ){
  //把pA添加到result的末尾
      pTail->pDirty = pA;
      pTail = pA;//更新result的尾结点
      pA = pA->pDirty;//把pA从输入链表中摘除
      if( pA==0 ){
        //因为pA和pB事先已经排好序了
        //遍历到pA的末尾之后，直接把pB插入的结果中即可
        pTail->pDirty = pB;
        break;
      }
}else{
  //当pB的页序比较小时，逻辑和上面相同
      pTail->pDirty = pB;
      pTail = pB;
      pB = pB->pDirty;
      if( pB==0 ){
        pTail->pDirty = pA;
        break;
      }
    }
  }
  return result.pDirty;
}

下面的代码就是把要排序的链表拆分成2次幂长度的子链表，最后归并。这里假定要排序的元素个数不超过2^31

#define N_SORT_BUCKET  32  
static PgHdr *pcacheSortDirtyList(PgHdr *pIn){  
  PgHdr *a[N_SORT_BUCKET], *p;  
  int i;  
  memset(a, 0, sizeof(a));  
  //遍历输入链表的元素  
  while( pIn ){  
    //从pIn里摘下头结点  
    p = pIn;  
    pIn = p->pDirty;  
    p->pDirty = 0;  
    //遍历a[i]，直到找到空的地址存放子链表  
    for(i=0; ALWAYS(i<N_SORT_BUCKET-1); i++){  
      //如果a[i]为空，那么把子链表保存到a[i]  
      if( a[i]==0 ){  
        a[i] = p;  
        break;  
      }else{  
        //如果a[i]不为空，那么p与a[i]归并，再继续下一次循环  
        p = pcacheMergeDirtyList(a[i], p);  
        a[i] = 0;  
      }  
    }  
    //下面这个条件是永远进不去的，这里假定元素最多2^31个  
    if( NEVER(i==N_SORT_BUCKET-1) ){  
      /* To get here, there need to be 2^(N_SORT_BUCKET) elements in 
      ** the input list.  But that is impossible. 
      */  
      a[i] = pcacheMergeDirtyList(a[i], p);  
    }  
  }  
  p = a[0];  
  //子链表拆分完毕后，把所有的子链表再归并成一个链表  
  for(i=1; i<N_SORT_BUCKET; i++){  
    if( a[i]==0 ) continue;  
    p = p ? pcacheMergeDirtyList(p, a[i]) : a[i];  
  }  
  return p;  
}  

脏页的原始链表是按照添加到链表的先后顺序排列的，用pDirtyNext连接，下面代码先拷贝一份原始链表，然后返回按照页序排列的链表，通过pDirty来连接。

/* 
** Return a list of all dirty pages in the cache, sorted by page number. 
*/  
PgHdr *sqlite3PcacheDirtyList(PCache *pCache){  
  PgHdr *p;  
  for(p=pCache->pDirty; p; p=p->pDirtyNext){  
    p->pDirty = p->pDirtyNext;  
  }  
  return pcacheSortDirtyList(pCache->pDirty);  
}