TimSort 排序算法

大家好，欢迎来到IT知识分享网。

之前给数组排序写比较方法时，出现了一个报错：

Comparison method violates its general contract!

然后报错提示到TimSort.mergeHi()方法抛出的异常，于是我就开始溯源问题，顺便研究一下世界上最快的归并排序——TimSort排序的实现。

首先在java的bug中第一次发现了TimSort的说明文档：
https://bugs.java.com/bugdatabase/view_bug.do?bug_id=6804124
http://svn.python.org/projects/python/trunk/Objects/listsort.txt

接下来我开始研读第二个链接指向的paper，它的大致内容如下

这是一个自适应的、稳定的自然归并算法。简单来说，主程序会自左向右遍历一遍数组，交替确认下一个run，并且“智能”的将它与之前的run进行合并。

关于和传统比较算法的性能对比，自行查阅文档。简而言之，只有在特定情况（有许多重复的值）下，TimSort耗时更长。

run可以是递增或者递减，且至少有两个值，除非它是数组的最后一个元素。递减的定义是严格的，因为主程序会将递减run转换为递增run，转换的方式是“交换头尾元素，直到中间位置”。
如果数组是随机的，几乎就不会存在长的run。如果一个初始run的数量小于minrun的数量（这是一个通过计算得到的值），主程序会通过标准的折半插入排序来将run扩充至正确的数量，因此在随机数组中，几乎所有的run都可能是minrun长度的。这会带来如下两点好处：
1.随机数据更倾向于完美的平衡合并，是最有效的方式。
2.因为run永远不会非常短（有扩充），其余代码不需要花费大力气去减少合并的开销。

关于如何计算minrun

如果N<64，则minrun为N，换句话说，二元插入排序用于整个数组，这很难避免其它优秀尝试所带来的开销。
当N是2的幂数，在随机数据测试的结果表明，minrun取16，32，64，128都表现不错，256对于二元插入排序开销太大，8则对于方法的调用次数开销过高。选择2的某次幂对于完美平衡合并很重要。我们选择32作为一个good value，选择一个值能够在最后更有机会利用更加短的natural runs。
通过sortperf.py花费大量时间的观察，发现32对于一般情况并不是一个好的选择，考虑N=2112：
divmod(2112, 32) = (66, 0) 取模操作
如果数据是随机有序的，我们可能最后得到66个长度为32的runs，最初的64个run顺序的进行完美平衡合并，最后剩下长度为2048和64的runs进行合并。自适应的创意可以执行少于2048+64次比较，但仍做了很多非必要的比较，同时，归并排序比普通排序要麻烦些，需要大量数据移动（O(n)次copy只能得到64个元素）。
如果这种情况下我们令minrun=33，我们将会的到长度为33的64个runs，这样所有合并都是完美平衡的！
我们想避免的是如下情况选择minrun：
q, r = divmod(N, minrun)
q是2的幂数同时r>0（最后的合并只能得到r个元素，同时r<minrun，minrun比N小），或者q比2的幂数大一点，不管r（我们得到一个类似“2112”的情况，最后的合并基本不需要做什么工作）。
因此选择minrun在(32, 65)范围内，这样N/minrun是一个确切的2的幂数，或者如果无法做到这种程度，则接近于2的幂数但严格小于它。实际上它的实现比听上去要容易的多：取N的前6个bit，如果有任何剩余位则+1。事实上，这个规则适用于任何场景，包括比较小的N以及2的幂数；merge_compute_minrun()是一个看似简单的函数。

合并模式

为了利用数据中的规律，我们自然的合并run长度，然后他们会变得非常不平衡，这对于这个排序是Good Thing！尽管这意味这我们不得不寻找一种方式来管理潜在的十分不同的run长度。
稳定性限制了可以使用的合并模式，比如，如果我们有如下3个连续的run长度：
A:10000 B:20000 C:10000
我们不敢先把A和C合并，因为如果A，B，C恰好包含同一元素，它在B中的顺序就乱了（破坏稳定性，如果A，B，C中的同一元素在排序后维持A，B，C的顺序，我们称之为排序具有稳定性）。因此合并必须是(A+B)+C或者A+(B+C)。
所以在同一时间同一地点永远是两个顺序run进行合并操作，尽管这需要一些临时内存。
当一个run被识别时，它的基础地址(base address)和长度都会被push到一个MergeState结构的堆栈中。merge_collapse()随后会调用来判断它是否需要和前一个run进行合并。我们希望尽量推迟merge以便能够利用后面出现的模式，但我们更希望尽快进行merge来利用刚发现的run处于内存层次结构的高位（缓存？）。我们也不能推迟合并太久，因为记住那些还未合并的runs会消耗内存，同时MergeState堆栈是有固定长度的。
一个好的折衷方案是在堆栈上维护两个不变量，其中A，B，C是最右边3个尚未合并的切片长度：

1. A > B+C
2. B > C

注意，通过归纳法，#2假设待定runs长度是递减序列的。#1假设，从右到左读取长度，待定run长度的增长速度至少和斐波那契数列一样快。因此堆栈永远不会大于log_base_phi(N)，因此对于非常大的数组，少量的堆栈槽位就足够了。
如果A <= B+C，则A和C中比较小的和B进行合并（推荐C，因为缓存机制），新的run将取代A，B或者B，C。
如最后3个元素是：
A:30 B:20 C:10
然后B和C合并，堆栈剩下
A:30 BC:30
即使它们是：
A:500 B:400 C: 1000
则A和B合并，堆栈剩余：
AB:900 C:1000
上述例子，合并后的堆栈不变量依然违反#2，则merge_collapse()继续合并runs直到两个不变量都得到满足。在极端例子中，假设我们不做minrun优化，原生runs的长度是128,64,32,16,8,4,2,和2。直到看到最后的2之前都不会进行任何合并，它会触发7次完美平衡合并。
触发合并的宗旨是尽可能平衡run的长度，使它们更加接近，同时尽可能将必须记录的run的数量维持在一个较低水平。对于随机数据的最有效的方法是，所有runs长度尽可能（人为控制）为minrun，之后我们会得到一系列完美平衡合并。
另一方面，这个排序方法对于部分排序数据如此有效与极度不平衡的run长度有关。

合并内存

在原地址合并长度A和B的runs十分困难。理论结构能够做到，但对于实际使用太过复杂和缓慢。但如果我们使用一块长度为min(A, B)的临时内存，则容易多了。
如果A更加小(function_merge_lo)，将A复制到临时数组，留下B，然后我们开始从左到右合并，从临时区域和B中获取数据，开始放到A曾经保存的位置上。原有区域总是有一块由一些元素组成的空闲区域，元素数量和临时数组中还未进行合并的数量相同（通过归纳，在开始的时候是正确的）。唯一棘手的一点是，当比较过程中出现异常，我们必须记住将剩下的元素从临时区域复制回来，以免数组中出现来自B的重复项（也会丢失数据）。但如果我们先达到B的末尾，需要做的事一摸一样（将剩下的数据复制回去），因此在正常和异常的情况下都适用相同的退出代码。
如果B更小(function_merge_hi，是function_merge_lo的镜像)，基本相同，只不过是从右到左进行合并，将B复制到临时数组，同时从B原来位置的末尾开始设置值。
改进：当我们进行A和B的合并时，首先进行二分查找，来判断B[0]应该在A中的位置。在A中这个位置之前的元素已经在它们的最终位置上了，有效的缩小A的长度；同时我们找到A[-1](A的尾元素)在B中的位置，在B中这个位置之后的所有元素已经可以被忽略了，这同样减少了临时内存所需的大小。
这种预先查询在数据随机时可能完全无用。但它们总是能够获胜，尤其是在复制和内存削减上，即使真的没有起作用，也不过是付出了很短的时间而已。在这个算法中我们更希望通过下一点赌注来大胜（gamble a little to win a lot），即使随机数据的净期望是负值。

合并算法

大部分时间都花在merge_lo()和merge_hi()上了。merge_lo处理A <= B，merge_hi处理 A > B。它们不知道数据是聚集还是相同，但有趣的是，多数聚集通过连续多少次获胜的元素来自同一run中来“暴露自己”。我们只讨论merge_lo，merge_hi类似。
合并开始以通常，明显的方式，比较A的第一个元素和B的第一个元素，如果B[0]小于A[0]，则移动B[0]到合并区，否则移动A[0]到合并区。称为“一次一双(one pair at a time)”模式。这里唯一的转折就是追踪“胜者”来自同一run的次数。
如果数量达到MIN_GALLOP，我们切换到“飞驰模式(galloping mode)”。我们搜索B来查找A[0]的位置，然后移动在B中那个位置之前的所有元素到合并区，之后将A[0]移动到合并区。之后我们搜寻B[0]在A中的位置，同样移动A中的元素。之后再搜索A[0]在B中的位置…飞驰模式直到两个搜索发现的复制元素都少于MIN_GALLOP元素长度，就会切换回一次一双模式(one-pair-at-a-time mode)。
改进：MergeState结构包含min_gallop数量以控制我们何时进入飞驰模式，初始化为MIN_GALLOP。merge_lo()和merge_hi()调整这个值，当飞驰得不到回报时调高，当得到回报时调低。

飞驰模式-Galloping

一般情况，假设A是较短的run。在飞驰模式，我们首先看A[0]在B中的位置。我们通过“飞驰模式”比较A[0]和B[0],B[1],B[3],B[7],...,B[2**j - 1],...，直到找到k符合：
B[2**(k-1) - 1] < A[0] < B[2**k - 1]
这至少进行lg(B)次比较，相较于二分查找会更早的在B中找到正确的位置。
在找到这个k之后，不确定性区域减去连贯的2**(k-1) - 1个值（小于A[0]那些值），而二分查找需要k-1次额外比较。之后我们将B中前2**(k-1) - 1的值复制，然后复制A[0]。注意不论A[0]是否在B中，galloping+二分查找的组合最多不超过2*lg(B)次比较。
如果我们用二分查找，则比较次数不会超过lg(B+1)——但二分查找不论A[0]在哪都会进行那么多次操作；除非run很长，否则二分查找输于飞驰模式，而我们无法提前确定。
如果数据是随机的且run的长度相同，则有1/2的概率A[0]处于B[0]的位置，1/4的概率A[0]处于B[1]的位置，等等：B中长度为k的sub-run连续获胜的概率为1/2**(k+1)。因此在随机数据中不可能出现长的获胜的sub-runs，并且假设一个获胜的sub-run很长是十分危险的。
另一方面，如果数据是不平衡的，或者波浪起伏的，或者包含很多副本，则很可能出现长的获胜的sub-run，将比较次数从O(B)缩减为O(logB)是巨大的胜利。
当一个获胜的sub-run不长时，飞驰模式会迅速退出，而当处于飞驰模式时的搜索十分高效。

飞奔模式的优化

为什么我们不一直使用gallop呢？因为它有可能在如下两种情况表现不佳：
1.尽管我们愿意忍受每次合并的小开销，但每次比较的开销却是另一回事。每次比较调用另一个函数代价很高，并且gallop_left()和gallop_right()对于明智的内联来说过于冗长。（尽管内联可以节省调用函数带来的额外时间开支，但是需要相应的存储资源，标准的以空间换时间）
2.依托于数据，galloping可能比线性搜索需要更多的比较。
关于#2的更多细节。如果A[0]在B[0]之前，galloping和线性搜索一样需要一次比较去确认，只是gallop函数消耗更多。如果A[0]恰好在B[1]之前，galloping和线性搜索一样需要两次比较。在第三次比较，galloping确认A[0]对比B[3]，如果是<=，需要一次额外的比较确认A[0]属于B[2]还是B[3]。共需要4次比较，但如果A[0]在B[2]，线性搜索只需要3次比较就可以确定，这是极大的浪费！所需的比较次数提升了33%，而Python中比较代价高昂。

通常，如果A[0]属于B[i]，线性搜索需要i+1次比较确认，而galloping需要2*floor(lg(i))+2次比较。随着i的增加，galloping的优势是无限的，但它直到i=6才会赢。在那之前它输了两次（i=2和i=4），在其它值持平。只有当i=6之后，galloping才会一直获胜。
然而我们无法提前猜到它何时获胜，因此我们一次只做一对的比较，直到有强有力的证据表明进入galloping模式。MIN_GALLOP是7，同时这是一个强有力的证据。但，如果数据是随机的，那有可能偶尔幸运的触发galloping模式，然后下次又会退出。另一方面，像~sort案例，galloping模式总是触发，这时MIN_GALLOP比它应该的值要更大。因此MergeState结构维护由merge_lo和merge_hi调整的变量min_gallop：我们处于galloping模式越长，获得的min_gallop越小，能够让它更容易回到galloping模式（如果我们在当前合并曾经离开，并在下次合并开始进入）。但无论何时无法进入gallop循环，则min_gallop会加1，让它更难返回galloping模式（还是在合并内部和跨合并）。对于随机数据，这几乎消除了gallop的处罚：min_gallop增长到一个非常大的值以至于几乎不会进入galloping模式。对于像~sort例子，min_gallop可能会降到1。看起来这对于使用固定的MIN_GALLOP是一个微小的进步。

Galloping复杂化

上面的描述是针对merge_lo的，merge_hi则是从“末尾一端”进行合并，并且是从run的最后一个元素开始进行gallop，而不是第一个元素。Galloping开始仍然工作，但做了更多的比较（这一点很重要–两种方式都计时）。由于这个原因，gallop_left()和gallop_right()函数有一个“hint”提示galloping应该从哪个位置开始。因此galloping可能在任一位置开始，从开始执行的索引位置的偏移量1，3，7，15...（从头部开始）或者-1，-3，-7，-15...（从尾部开始）。
在我的代码中总是以0或者n-1调用它（n是run的元素数量）。做一些优化总是很诱人，融合galloping和某种插值搜索形式；比如，我们合并一个长度为1的run和一个长度为10000的run，索引5000可能比0或者9999是一个更好的猜测。但目前无法有效的进行猜测，同时合并极端不平衡的run已经获得了优秀的性能。
~sort是一个很好的例子，说明平衡的runs如何从更好的hint值中获益：在可能的情况下，它希望使用acount/bcount之前的一个值作为起始偏移量。这样做可能减少~sort的10%次对比。然而对于其它情况则有利有弊。

另外找到一篇博客画了很好的图例

以上是TimSort的paper中的主要内容，勉强翻译了一下，接下来我看看Java中是如何实现的，具体类为TimSort，可以通过Arrays类找到该私有类。

首先介绍就是

A stable, adaptive, iterative mergesort that requires far fewer than
n lg(n) comparisons when running on partially sorted arrays, while
offering performance comparable to a traditional mergesort when run
on random arrays. Like all proper mergesorts, this sort is stable and
runs O(n log n) time (worst case). In the worst case, this sort requires
temporary storage space for n/2 object references; in the best case,
it requires only a small constant amount of space.

---

一个稳定，自适应，迭代的归并排序，当运行在部分排序数组中需要远少于nlg(n)次比较，而运行在随机数组中和传统归并排序性能类似。像所有合适的归并算法一样，这个排序是稳定的，并且在最坏的情况下需要比较O(nlogn)次。在最坏的情况，需要的临时存储空间为n/2对象引用；最好的情况，只需要很小的常量空间。

变量

/** * 这是进行合并的序列的最小值。比这个值短的序列将会调用binarySort扩增。 * 如果所有数组都小于这个长度，则不会执行任何合并 */
private static final int MIN_MERGE = 32;
// 要被排序的数组
private final T[] a;
// 比较器
private final Comparator<? super T> c;
// 当我们进入galloping模式时，直到两次连续获胜的run都小于MIN_GALLOP就会退出
private static final int  MIN_GALLOP = 7;
/** * 这个值控制我们何时进入galloping模式，初始化为MIN_GALLOP。 * mergeLo和mergeHi方法会在随机数据时缓慢增加它，而在高度结构化数据是减少 */
private int minGallop = MIN_GALLOP;
// 临时数组的最大初始值，用于合并。这个数组可以根据需要增长。
private static final int INITIAL_TMP_STORAGE_LENGTH = 256;
// 用于合并的临时存储数组。一个工作区的数组由构造函数任意提供，只要足够大就行
private T[] tmp;
private int tmpBase; // 临时数组切片的基础索引
private int tmpLen;  // 临时数组切片的长度
/** * 用于待合并runs的栈。Run i 起始于base[i]地址，且拥有len[i]个元素。永远满足： * runBase[i] + runLen[i] == runBase[i+1] * 这样我们能够缩减存储空间，通过显示简化代码保存了所有信息 */
private int stackSize = 0;  // Number of pending runs on stack
private final int[] runBase;
private final int[] runLen;

构造函数

/** * Creates a TimSort instance to maintain the state of an ongoing sort. * * @param a the array to be sorted * @param c the comparator to determine the order of the sort * @param work a workspace array (slice) * @param workBase origin of usable space in work array * @param workLen usable size of work array */
private TimSort(T[] a, Comparator<? super T> c, T[] work, int workBase, int workLen) { 
   
    this.a = a;
    this.c = c;

    // 分配临时存储（必要情况会扩展）
    int len = a.length;
    int tlen = (len < 2 * INITIAL_TMP_STORAGE_LENGTH) ?
        len >>> 1 : INITIAL_TMP_STORAGE_LENGTH;
    if (work == null || workLen < tlen || workBase + tlen > work.length) { 
   
        @SuppressWarnings({ 
   "unchecked", "UnnecessaryLocalVariable"})
        T[] newArray = (T[])java.lang.reflect.Array.newInstance
            (a.getClass().getComponentType(), tlen);
        tmp = newArray;
        tmpBase = 0;
        tmpLen = tlen;
    }
    else { 
   
        tmp = work;
        tmpBase = workBase;
        tmpLen = workLen;
    }

    /* * 分配将要合并的runs的栈（无法扩展）。栈长度可以参考listsort.txt的描述。 * C版本总是使用相同的栈长（85），但在java中对于排序中等大小的数组（如100个元素）时过于浪费。 * 因此，我们对更小的数组使用更小的栈长（但足够大）。 * 如下计算的“魔法数”在MIN_MERGE减小时需要改变。可以在上述MIN_MERGE中查看更多信息。 * 最大值49允许数组长度直到Integer.MAX_VALUE-4，用于最坏情况下数组填充的栈长度。 * * More explanations are given in section 4 of: * http://envisage-project.eu/wp-content/uploads/2015/02/sorting.pdf */
    int stackLen = (len <    120  ?  5 :
                    len <   1542  ? 10 :
                    len < 119151  ? 24 : 49);
    runBase = new int[stackLen];
    runLen = new int[stackLen];
}

排序方法
sort()

/** * 排序给定范围，使用工作空间数组切片暂存数据。被Arrays的public方法调用。 * * @param a 待排序数组 * @param lo 待排序数组第一个元素的索引地址，包含 * @param hi 待排序数组最后一个元素的索引地址，不包含 * @param c 比较器 * @param work 工作空间数组（切片） * @param workBase 工作数组可用空间起始位置 * @param workLen 工作数组可用大小 * @since 1.8 */
static <T> void sort(T[] a, int lo, int hi, Comparator<? super T> c,
                     T[] work, int workBase, int workLen) { 
   
    // 参数校验 
    assert c != null && a != null && lo >= 0 && lo <= hi && hi <= a.length;
    // 数组大小
    int nRemaining  = hi - lo;
    if (nRemaining < 2)
        return;  // 0或1个元素的数组已经是排序的了

    // 如果数组过小，不做合并进行二分排序并直接返回
    if (nRemaining < MIN_MERGE) { 
   
        int initRunLen = countRunAndMakeAscending(a, lo, hi, c);
        binarySort(a, lo, hi, lo + initRunLen, c);
        return;
    }

    /** * 从左到右遍历数组一次，找到原始runs，扩展较短的原始runs到minRun个元素， * 合并runs维护栈不变量。 */
    TimSort<T> ts = new TimSort<>(a, c, work, workBase, workLen);
    // 计算run的最小长度
    int minRun = minRunLength(nRemaining);
    do { 
   
        // 确定下一个run
        int runLen = countRunAndMakeAscending(a, lo, hi, c);

        // 如果run太短, 扩展至 min(minRun, nRemaining)
        if (runLen < minRun) { 
   
            int force = nRemaining <= minRun ? nRemaining : minRun;
            // 通过二分排序排序从lo起，长度为force的run（因此称通过binarySort扩展run）
            binarySort(a, lo, lo + force, lo + runLen, c);
            runLen = force;
        }

        // 将run push到栈中，可能执行合并操作
        ts.pushRun(lo, runLen);
        ts.mergeCollapse();

        // 继续前进，确定下一个run
        lo += runLen;
        nRemaining -= runLen;
    } while (nRemaining != 0);

    // 合并所有剩下的run，以完成该排序方法
    assert lo == hi;
    ts.mergeForceCollapse();
    assert ts.stackSize == 1;
}

countRunAndMakeAscending()

/** * 返回指定数组指定位置开始的run长度，并且如果run是递减的则反转 * （保证方法返回时run永远是递增的） * 最长的递增run序列如下: * * a[lo] <= a[lo + 1] <= a[lo + 2] <= ... * * 最长的递减run序列如下: * * a[lo] > a[lo + 1] > a[lo + 2] > ... * * 因为该方法是在一个稳定的归并排序中使用，因此需要严格的“递减”定义， * 这样才能安全的反转递减序列而不违背稳定性。 * * @param a 对run计数并可能反转的数组 * @param lo run的第一个元素索引 * @param hi 最后一个元素的后一个索引，可能包含在run中。需要lo < hi。 * @param c 用于排序的比较器 * @return 指定数组在指定位置开始的run长度 */
private static <T> int countRunAndMakeAscending(T[] a, int lo, int hi,
                                                Comparator<? super T> c) { 
   
    assert lo < hi;
    // run的高位指针
    int runHi = lo + 1;
    if (runHi == hi)
        return 1;

    // 找到run的结尾，如果递减则反转（run就是一个递增或递减的序列）
    if (c.compare(a[runHi++], a[lo]) < 0) { 
    // 递减
        while (runHi < hi && c.compare(a[runHi], a[runHi - 1]) < 0)
            runHi++;
        reverseRange(a, lo, runHi);
    } else { 
                                 // 递增
        while (runHi < hi && c.compare(a[runHi], a[runHi - 1]) >= 0)
            runHi++;
    }

    return runHi - lo;
}

binarySort()

/** * 对指定数组的指定位置使用二分插入排序。对于排序数量较少的数组是最好的算法。 * 它需要O(nlogn)次比较，但需要O(n^2)次数据移动（最坏情况）。 * * 如果指定的起始范围已经有序，该方法可以利用这一点： * 方法假设从lo（包含）开始到start（不包含）结束的数据已经有序。 * @param a 需要排序的数组 * @param lo 要排序范围的首个元素索引 * @param hi 要排序范围最后一个元素的后一位索引 * @param start 不确定是否已排序的范围的第一个元素的索引( lo <= start <= hi) * @param 用于排序的比较器 */
@SuppressWarnings("fallthrough")
private static <T> void binarySort(T[] a, int lo, int hi, int start,
                                   Comparator<? super T> c) { 
   
    assert lo <= start && start <= hi;
    if (start == lo)
        start++;
    for ( ; start < hi; start++) { 
   
        T pivot = a[start];

        // 取从lo到start的元素
        int left = lo;
        int right = start;
        assert left <= right;
        /* * 不变量: * pivot >= all in [lo, left). * pivot < all in [right, start). */
        while (left < right) { 
    // 二分查找pivot
            int mid = (left + right) >>> 1;
            if (c.compare(pivot, a[mid]) < 0)
                right = mid;
            else
                left = mid + 1;
        }
        assert left == right;

        /* * 不变量始终维持: pivot >= all in [lo, left) 以及 * pivot < all in [left, start), 因此pivot属于左边. * 如果有元素等于pivot, left 指向它们之后的第一个位置 * -- 这也是该排序是稳定的原因. * 滑动元素为pivot腾出位置. */
        int n = start - left;  // 需要移动的元素数量
        // 默认情况arraycopy是最优解（相当于把left的位置腾出来放pivot，其它n个元素右移1位）
        switch (n) { 
   
            case 2:  a[left + 2] = a[left + 1];
            case 1:  a[left + 1] = a[left];
                     break;
            default: System.arraycopy(a, left, a, left + 1, n);
        }
        a[left] = pivot;
    }
}

minRunLength()

/** * 返回指定长度数组的最小可接受run长度。原生runs元素数量小于该值时通过binarySort进行扩展 * 计算方式大概如下: * 当n < MIN_MERGE时，返回n（太小以至于不需要优化）。 * 当n时2的指数时，返回MIN_MERGE/2。 * 否则返回int常量k，满足MIN_MERGE/2 <= k <= MIN_MERGE，这样n/k接近但严格小于2的指数。 * * 基本原理可以查看 listsort.txt. * * @param n 需要排序的数组长度 * @return 要执行合并的run的最小长度 */
private static int minRunLength(int n) { 
   
    assert n >= 0;
    int r = 0;      // 如果有任何一位移动则变成1
    while (n >= MIN_MERGE) { 
   
        r |= (n & 1); // r为0或1
        n >>= 1; // n/2
    }
    return n + r; // 如果n为2的指数，保证最后返回的是MIN_MERGE/2；否则返回MIN_MERGE/2 + 1
}

pushRun()

/** * 将指定run push到存放run信息的栈中。runBase和runLen，同时栈长加1 * * @param runBase run中初始元素索引 * @param runLen run的元素数量 */
private void pushRun(int runBase, int runLen) { 
   
    this.runBase[stackSize] = runBase;
    this.runLen[stackSize] = runLen;
    stackSize++;
}

mergeCollapse()

/** * 检查等待合并runs的栈，并且合并相邻的runs直到栈不变量被更新为 * * 1. runLen[i - 3] > runLen[i - 2] + runLen[i - 1] * 2. runLen[i - 2] > runLen[i - 1] * * 每当一个新的run被push到栈中都会调用该方法，因此i<stackSize是保证不变量的条件 * */
private void mergeCollapse() { 
   
    while (stackSize > 1) { 
    // 只有栈长大于1才会执行合并操作
        int n = stackSize - 2; // 取栈顶的第二个run
        if (n > 0 && runLen[n-1] <= runLen[n] + runLen[n+1]) { 
   
            if (runLen[n - 1] < runLen[n + 1]) // 比较第三个run和第一个run的长度
                n--; // 当第三个run的长度小于第一个run，则n为第三个run
            mergeAt(n);
        } else if (runLen[n] <= runLen[n + 1]) { 
    // 比较第二个run和第一个run长度
            mergeAt(n); // 当第二个run的长度小于第一个run长度，n为第二个run
        } else { 
   
            break; // 不变量建立（符合上述不变量条件）
        }
    }
}

mergeAt()

/** * 合并栈索引为i和i+1的两个run。run i必须为栈的倒数第二个或者倒数第三个元素。 * 换句话说，i必须等于stackSize-2 或 stackSize-3 * * @param i 两个合并run中第一个run的栈索引 */
private void mergeAt(int i) { 
   
    assert stackSize >= 2;
    assert i >= 0;
    assert i == stackSize - 2 || i == stackSize - 3;

    int base1 = runBase[i];
    int len1 = runLen[i];
    int base2 = runBase[i + 1];
    int len2 = runLen[i + 1];
    assert len1 > 0 && len2 > 0;
    assert base1 + len1 == base2;

    /* * 记录合并run的长度；如果i是倒数第三个run，同时滑向最后一个run（并不包含在这次merge中）。 * 含义为：记录run的栈用记录i+1 run（倒数第二个）的位置来记录i+2 run（栈顶run）。 * 当前的第 i+1 个run会消失。栈大小减1。 */
    runLen[i] = len1 + len2;
    if (i == stackSize - 3) { 
   
        runBase[i + 1] = runBase[i + 2];
        runLen[i + 1] = runLen[i + 2];
    }
    stackSize--;

    /* * 找到run2的首元素在run1中的位置。 * 在run1中这个位置之前的元素可以被忽略（因为它们已经在应该处于的位置上）。 */
    int k = gallopRight(a[base2], a, base1, len1, 0, c);
    assert k >= 0;
    base1 += k; // 忽略前k个元素
    len1 -= k; // run1长度减k
    if (len1 == 0)
        return;

    /* * 找到run1尾元素在run2中的位置。 * run2该位置之后的元素忽略。 */
    len2 = gallopLeft(a[base1 + len1 - 1], a, base2, len2, len2 - 1, c);
    assert len2 >= 0;
    if (len2 == 0)
        return;

    // 合并剩下的runs，使用临时数组，大小为min(len1, len2)
    if (len1 <= len2)
        mergeLo(base1, len1, base2, len2);
    else
        mergeHi(base1, len1, base2, len2);
}

gallopRight()

/** * 类似gallopLeft, 当范围内包含等于key的元素时，返回最右边相等值的后一位索引 * * @param key 需要插入位置的键 * @param a 执行搜索的数组 * @param base 范围内第一个元素的索引 * @param len 范围长度，必须>0 * @param hint 开始执行搜索的位置，0 <= hint < n。hint越接近结果，该方法返回的越快。 * @param c 执行排序和搜索的比较器 * @return the int k, 0 <= k <= n such that a[b + k - 1] <= key < a[b + k] */
private static <T> int gallopRight(T key, T[] a, int base, int len,
                                   int hint, Comparator<? super T> c) { 
   
    assert len > 0 && hint >= 0 && hint < len;

    int ofs = 1; // 右指针
    int lastOfs = 0; // 左指针
    if (c.compare(key, a[base + hint]) < 0) { 
    // key小于hint位置对应的值
        // 向左飞驰直到 a[b+hint - ofs] <= key < a[b+hint - lastOfs]
        int maxOfs = hint + 1;
        // 当右指针小于最大偏移，且key小于基于hint偏移ofs的值时
        while (ofs < maxOfs && c.compare(key, a[base + hint - ofs]) < 0) { 
   
            lastOfs = ofs; // 右指针赋值给左指针
            ofs = (ofs << 1) + 1; // 右指针*2 + 1
            if (ofs <= 0)   // 整形溢出
                ofs = maxOfs;
        }
        if (ofs > maxOfs)
            ofs = maxOfs;

        // 保证偏移基于base
        int tmp = lastOfs;
        lastOfs = hint - ofs; // 左指针是小的值（ofs大于lastOfs）
        ofs = hint - tmp;
    } else { 
    // a[b + hint] <= key
        // 向右飞驰直到 a[b+hint + lastOfs] <= key < a[b+hint + ofs]
        int maxOfs = len - hint;
        while (ofs < maxOfs && c.compare(key, a[base + hint + ofs]) >= 0) { 
   
            lastOfs = ofs;
            ofs = (ofs << 1) + 1;
            if (ofs <= 0)   // 整形溢出
                ofs = maxOfs;
        }
        if (ofs > maxOfs)
            ofs = maxOfs;

        // 偏移量基于base
        lastOfs += hint;
        ofs += hint;
    }
    assert -1 <= lastOfs && lastOfs < ofs && ofs <= len;

    /* * 现在 a[b + lastOfs] <= key < a[b + ofs], 因此key在lastOfs的右侧，ofs左侧某个位置 * 做二分查找，不变量为 a[base + lastOfs - 1] < key <= a[base + ofs] 。 */
    lastOfs++;
    while (lastOfs < ofs) { 
   
        int m = lastOfs + ((ofs - lastOfs) >>> 1);

        if (c.compare(key, a[base + m]) < 0)
            ofs = m;          // key < a[b + m]
        else
            lastOfs = m + 1;  // a[b + m] <= key
    }
    assert lastOfs == ofs;    // so a[b + ofs - 1] <= key < a[b + ofs]
    return ofs; // 最后找到key值所在索引
}

gallopLeft()

/** * 定位要将指定的键插入到指定的排序范围的位置；如果该范围包含等于key的元素， * 返回等于该元素的最左边元素的索引值。 * * @param key 需要插入位置的键 * @param a 执行搜索的数组 * @param base 范围内第一个元素的索引 * @param len 范围长度，必须>0 * @param hint 开始执行搜索的位置，0 <= hint < n。hint越接近结果，该方法返回的越快。 * * @param c 执行排序和搜索的比较器 * @return the int k, 0 <= k <= n 并且 a[b + k - 1] < key <= a[b + k], * 假设 a[b - 1] 是负无穷并且 a[b + n] 是正无穷。 * 换句话说, key 位于索引 b + k; 或者, * 前k个元素小于 key, 后 n - k 个元素大于key。 */
private static <T> int gallopLeft(T key, T[] a, int base, int len, int hint,
                                  Comparator<? super T> c) { 
   
    assert len > 0 && hint >= 0 && hint < len;
    int lastOfs = 0; // 左指针，相对于hint
    int ofs = 1;  // 右指针，相对于hint
    if (c.compare(key, a[base + hint]) > 0) { 
    // 当要比较的第一个元素小于key
        // 向右飞驰直到 a[base+hint+lastOfs] < key <= a[base+hint+ofs]
        int maxOfs = len - hint; // 最大偏移
        // 当右指针小于最大偏移，且右指针对应的值小于key
        while (ofs < maxOfs && c.compare(key, a[base + hint + ofs]) > 0) { 
   
            lastOfs = ofs; // 将右指针赋给左指针
            ofs = (ofs << 1) + 1; // 右指针*2 + 1 （1，3，7，15...）
            if (ofs <= 0)   // 整形溢出
                ofs = maxOfs;
        }
        if (ofs > maxOfs) // 当右指针超过最大偏移，则赋值为最大偏移
            ofs = maxOfs;

        // 相对于base的偏移量
        lastOfs += hint;
        ofs += hint;
    } else { 
    // key <= a[base + hint]
        // 向左飞驰直到 a[base+hint-ofs] < key <= a[base+hint-lastOfs]
        final int maxOfs = hint + 1;
        // 当右指针小于最大偏移，且从hint向左偏移ofs的值大于等于key时
        while (ofs < maxOfs && c.compare(key, a[base + hint - ofs]) <= 0) { 
   
            lastOfs = ofs; // 左指针取右指针的值
            ofs = (ofs << 1) + 1; // 右指针*2 + 1
            if (ofs <= 0)   // 整形溢出
                ofs = maxOfs;
        }
        if (ofs > maxOfs)
            ofs = maxOfs;

        // 保证基于base的偏移量
        int tmp = lastOfs;
        lastOfs = hint - ofs; // 左指针依然是小的那个（ofs大于lastOfs）
        ofs = hint - tmp; // 右指针依然是大的那个
    }
    assert -1 <= lastOfs && lastOfs < ofs && ofs <= len;

    /* * 现在 a[base+lastOfs] < key <= a[base+ofs], 因此key在lastOfs的右侧，ofs左侧某个位置 * 做二分查找，不变量为 a[base + lastOfs - 1] < key <= a[base + ofs] 。 */
    lastOfs++;
    while (lastOfs < ofs) { 
    // 左指针到右指针
        int m = lastOfs + ((ofs - lastOfs) >>> 1); // 二分查找确定中间值

        if (c.compare(key, a[base + m]) > 0)
            lastOfs = m + 1;  // a[base + m] < key
        else
            ofs = m;          // key <= a[base + m]
    }
    assert lastOfs == ofs;    // so a[base + ofs - 1] < key <= a[base + ofs]
    return ofs; // 最后找到key值所在索引
}

mergeLo()

/** * 稳定合并两个相邻runs. 第一个run的首元素必须大于第二个run的第首元素(a[base1] > a[base2]), * 且第一个run的尾元素(a[base1 + len1-1]) 必须大于第二个run的所有元素。 * * 出于性能考虑, 该方法只有在 len1 <= len2 才会被调用; * 它的孪生方法, mergeHi 在 len1 >= len2 时调用. (当 len1 == len2 可调用任一方法.) * * @param base1 要被合并的第一个run的首元素的索引 * @param len1 要合并的第一个run的长度（必须>0） * @param base2 要合并的第二个run的首元素的索引（等于aBase + aLen） * @param len2 要合并的第二个run的长度 (必须 > 0) */
private void mergeLo(int base1, int len1, int base2, int len2) { 
   
    assert len1 > 0 && len2 > 0 && base1 + len1 == base2;

    T[] a = this.a; // 性能，a是整个待排序数组
    T[] tmp = ensureCapacity(len1);
    int cursor1 = tmpBase; // 临时数组的索引
    int cursor2 = base2;   // run2的索引
    int dest = base1;      // run1的索引
    // 将run1复制到临时数组
    System.arraycopy(a, base1, tmp, cursor1, len1);

    // 移动run2的首元素，并处理退化的情况
    a[dest++] = a[cursor2++]; // 因为run2的首元素小于run1的首元素，因此要将run2首元素放到第一个位置
    if (--len2 == 0) { 
    // 当run2只有一个值，则将剩下的run1元素放回所属位置
        System.arraycopy(tmp, cursor1, a, dest, len1);
        return;
    }
    if (len1 == 1) { 
    // 当run1只有一个值，将剩下run2的值放到所属位置
        System.arraycopy(a, cursor2, a, dest, len2);
        a[dest + len2] = tmp[cursor1]; // 并将run1的这个值放到末尾，因为run1尾元素最大
        return;
    }

    Comparator<? super T> c = this.c;  // 出于性能考虑，使用方法的局部变量，而不使用类的实例变量
    int minGallop = this.minGallop;    // 同上
outer:
    while (true) { 
   
        int count1 = 0; // run1连续获胜的次数
        int count2 = 0; // run2连续获胜的次数

        /* * 做简单的操作，直到（如果有）一个run开始持续获胜 */
        do { 
   
            assert len1 > 1 && len2 > 0;
            if (c.compare(a[cursor2], tmp[cursor1]) < 0) { 
    // 当run2的值比run1的值小
                a[dest++] = a[cursor2++]; // 将run2的值放到a的对应位置上
                count2++; // run2连续获胜次数增加
                count1 = 0; // run1连续获胜次数置零
                if (--len2 == 0) // 当run2为空，则跳出循环
                    break outer;
            } else { 
   
                a[dest++] = tmp[cursor1++]; // 将run1值放到a对应位置上
                count1++;
                count2 = 0;
                if (--len1 == 1) // 当run1只剩尾元素，跳出循环
                    break outer;
            }
        } while ((count1 | count2) < minGallop); // 当run1或run2连续获胜次数小于minGallop时

        /* * 当一个run持续获胜则galloping将会获得巨大胜利。 * 因此尝试galloping, 直到run不再出现连续获胜。 */
        do { 
   
            assert len1 > 1 && len2 > 0;
            // 通过飞驰模式找到run2的值再run1中的位置
            count1 = gallopRight(a[cursor2], tmp, cursor1, len1, 0, c);
            if (count1 != 0) { 
   
                System.arraycopy(tmp, cursor1, a, dest, count1); // 将该位置之前的所有run1中的值复制到a中
                // 更新指针
                dest += count1;
                cursor1 += count1;
                len1 -= count1;
                if (len1 <= 1) // len1 == 1 || len1 == 0
                    break outer;
            }
            a[dest++] = a[cursor2++]; // 将run2的值放到a中
            if (--len2 == 0) // 如果run2为空，则退出循环
                break outer;
            // 找到run1的值在run2中的位置
            count2 = gallopLeft(tmp[cursor1], a, cursor2, len2, 0, c);
            if (count2 != 0) { 
   
                // 复制run1对应位置前run2的所有元素到a中
                System.arraycopy(a, cursor2, a, dest, count2);
                // 更新指针
                dest += count2;
                cursor2 += count2;
                len2 -= count2;
                if (len2 == 0) // run2长度为空退出循环
                    break outer;
            }
            a[dest++] = tmp[cursor1++]; // 将run1的值放到a中
            if (--len1 == 1) // 当run1只剩尾元素，退出循环
                break outer;
            minGallop--; // 减少minGallop，使进入飞驰模式更容易
        } while (count1 >= MIN_GALLOP | count2 >= MIN_GALLOP); // 当run的连续获胜次数大于MIN_GALLOP时
        if (minGallop < 0)
            minGallop = 0;
        minGallop += 2;  // 退出飞驰模式的惩罚
    }  // End of "outer" loop
    this.minGallop = minGallop < 1 ? 1 : minGallop;  // 回写字段

    if (len1 == 1) { 
    // run1只剩尾元素
        assert len2 > 0;
        System.arraycopy(a, cursor2, a, dest, len2); // 复制剩下的run2元素
        a[dest + len2] = tmp[cursor1]; // run1尾元素
    } else if (len1 == 0) { 
    // 如果run1为空，抛出异常
        throw new IllegalArgumentException(
            "Comparison method violates its general contract!");
    } else { 
    // run2为空，则将run1的所有元素复制到a中
        assert len2 == 0;
        assert len1 > 1;
        System.arraycopy(tmp, cursor1, a, dest, len1);
    }
}

ensureCapacity()

/** * 确保外部临时数组至少能存放指定数量的元素, 必要时扩展。 * 指数扩展以抵消线性时间复杂度。 * * @param minCapacity 临时数组所需最小容量 * @return tmp, 不管扩不扩充 */
private T[] ensureCapacity(int minCapacity) { 
   
    if (tmpLen < minCapacity) { 
   
        // 计算最小的大于 minCapacity的2的指数
        int newSize = minCapacity;
        newSize |= newSize >> 1;
        newSize |= newSize >> 2;
        newSize |= newSize >> 4;
        newSize |= newSize >> 8;
        newSize |= newSize >> 16;
        newSize++;

        if (newSize < 0) // 基本不可能
            newSize = minCapacity;
        else
            newSize = Math.min(newSize, a.length >>> 1);

        @SuppressWarnings({ 
   "unchecked", "UnnecessaryLocalVariable"})
        T[] newArray = (T[])java.lang.reflect.Array.newInstance
            (a.getClass().getComponentType(), newSize);
        tmp = newArray; // 临时数组
        tmpLen = newSize; // 临时数组长度
        tmpBase = 0;  // 临时数组首元素索引
    }
    return tmp;
}

mergeHi()方法和mergeLo()方法基本思路相同，这里就不再赘述，只不过索引都变成了从右到左的，需要做差取正确的索引等。

至此，整个TimSort算法就全部看完了。
这其中也出现了关键报错信息：

“Comparison method violates its general contract!”

也就是整个问题的来源。
根源在于mergeLo()中这部分代码

count1 = gallopRight(a[cursor2], tmp, cursor1, len1, 0, c);
if (count1 != 0) { 
   
    System.arraycopy(tmp, cursor1, a, dest, count1);
    dest += count1;
    cursor1 += count1;
    len1 -= count1;
    if (len1 <= 1) // len1 == 1 || len1 == 0
        break outer;
}

也就是说，在元素相同的时候，没有正确规定comparator中的行为，导致飞驰模式取了run1的尾元素，进而导致len1为0.
即gallopRight中的这部分代码

int maxOfs = len - hint;
while (ofs < maxOfs && c.compare(key, a[base + hint + ofs]) >= 0) { 
   
    lastOfs = ofs;
    ofs = (ofs << 1) + 1;
    if (ofs <= 0)   // int overflow
        ofs = maxOfs;
}
if (ofs > maxOfs)
    ofs = maxOfs;

// Make offsets relative to b
lastOfs += hint;
ofs += hint;

ofs取了maxOfs，最后等于len。ofs在计算机术语中，是偏移量的含义。

最后关于那个报错，可以看下这个文章：貌似是违背了自反性?