省メモリ超高速マージソート


省メモリマージソートの私の最新のコード (2026年9月)

一般のマージソートでは,データのと同じ程度の追加メモリ領域が必要です。インプレースマージソート(in-place merge sort)とは,追加メモリ領域をほとんど使わずに行うマージソートです。
また,一般のマージソートでは,キーの値が同じ要素のソー ト前の順序が,ソート後も変わらずに維持されます。このようなソートを安定ソートと言います。
一般のマージソートは,最速と言われているクイックソートよりも遅いようです。ただし,クイックソートは,安定ソートではありません。(私が作ったマージソートはクイッ クソートよりも速いです。)
ここでは,作業用をデータのメモリ領域の 0.1% や 0.01% 程度に設定してもクイックソートより速くて,平均計算時間と最悪計算時間が共に O(n log n)である,安定な省メモリマージソートを示します。

基本的なインプレースマージソート
ウィキペディアの「ソート」のページでは,安定な In-place マージソートの平均計算時間と最悪計算時間は O(n(log n)2) となっています。
そのようなコードは,Google検索やyahoo検索で「インプレースマージソート  回転と再帰を用いた超高速なC言語のコード」などと上手く検索すれば,AIが提供してくれます。(AIによるコードには,バグがあり動作しない可能性があ ります。)
そのインプレースマージソートに上記の省メモリマージソートをうまく組み込めば,僅かなメモリでも実用に耐える速度の,安定マージソートが得られるので はないかと考えました。
まずは,検索でAIより得た動作不良のインプレースマージソートのコードを動作するように直し,少しだけ変更を加えた,計算時間が O(n(log n)2) の基本のインプレースマージソートのコードを示します。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>

// 挿入ソート
void insertion_sort(int arr[],int i,int n){
    for (;i<n;i++){
        int tmp=arr[i];
        int j=i-1;
        for (;j>=0 && arr[j]>tmp;j--) arr[j+1]=arr[j];
        arr[j+1]=tmp;
    }
}

// ヘルパー関数: メモリの特定領域を反転させる(rotateの実装に使用)
void reverse(int* first, int* last) {
    while (first < last) {
        int temp = *first;
        *first = *last;
        *last = temp;
        first++;
        last--;
    }
}

// [first, mid) と [mid, last) の要素を入れ替える
void rotate(int* first, int* mid, int* last) {
    if (first == mid || mid == last) return;
    reverse(first, mid-1);
    reverse(mid, last-1);
    reverse(first, last-1);
}

// キー以上の最初の位置
int lower_bound(int arr[], int low, int high, int key) {
    while (low < high) {
        int mid = low + (high - low) / 2;
        if (arr[mid] >= key)
            high = mid;
        else
            low = mid + 1;
    }
    return low;
}

// キーより大きい最初の位置
int upper_bound(int arr[], int low, int high, int key) {
    while (low < high) {
        int mid = low + (high - low) / 2;
        if (arr[mid] > key)
            high = mid;
        else
            low = mid + 1;
    }
    return low;
}

// 再帰的In-place マージ関数
void inPlaceMerge(int arr[], int l, int m, int r) {
    // マージ不要な場合
    if (m <= l || r < m) return;

    // 前半の最大値が後半の最小値以下の場合
    if (arr[m-1] <= arr[m]) return;

    // マージするデータ数が少なくなったとき
    if (r-l<=35){
        insertion_sort(arr+l, m-l, r-l+1);
        return;
    }

    // 分割点(回転させる境界)を見つける
    int i, j;
    if (m - l >= r - m + 1) {
        i = l + (m - l) / 2;
        // 範囲 [m, r+1) で arr[i] 以上に大きい最初の要素を探す
        j=lower_bound(arr, m, r+1, arr[i]);
    } else {
        j = m + (r - m + 1) / 2;
        // 範囲 [l, m) で arr[j] より大きい最初の要素を探す
        i=upper_bound(arr, l, m, arr[j]);
    }

    // 回転して領域を融合 (iからm-1とmからj-1を交換)
    rotate(arr+i, arr+m, arr+j);
    // 新しい分割点
    m = i + (j - m);

    // 再帰的にマージ
    inPlaceMerge(arr, l, i, m - 1);
    inPlaceMerge(arr, m , j, r);
}

// ソート関数
void inPlaceMergeSort(int arr[], int l, int r) {
    //if (r - l <= 0) return;
    int n = r - l + 1;
    if (n<50) { // 小さいところは挿入ソートで高速化
        insertion_sort(arr+l,1, n);
        return;
    }
       
    int m = l + n / 2;
    inPlaceMergeSort(arr, l, m - 1);
    inPlaceMergeSort(arr, m , r);
    inPlaceMerge(arr, l, m , r);
}

int main(){
    int n=100000000;
    int i;
    clock_t start,end;
    int *arr=malloc(n*sizeof(int));
    srand((unsigned) time(NULL));
    for (i=0;i<n;i++) arr[i]=rand()*(RAND_MAX+1)+rand();
    start=clock();
    inPlaceMergeSort(arr,0,n-1);
    end=clock();
    printf("%f秒  \n",(double)(end-start)/CLOCKS_PER_SEC);
    for (i=1;i<n;i++) if(arr[i-1]>arr[i]) printf("%d %d\n",i,arr[i]);
    free(arr);
    return 0;
}

基本的なインプレースマージソートを参考にして独自に作ったインプレースマージソー ト
基本的なインプレースマージソートは,平均計算時間と最悪計算時間は共に O(n (log n)2) で,追加のメモリ使用量が O(log n) であり,安定ソートです。
次のコードは,私が基本的なインプレースマージソートを参考にして独自に作ったインプレースマージソートで,基本的なコードより単純でデータ交換の負荷が 半分程度に小さくなっています。
データ数が1億個のときの平均計算時間は,クイックソートの2倍程度です。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>

typedef int d_type;// ソートするキーの型

// 挿入ソート
void insertion_sort(d_type arr[],int n){
    int i=1;
    for (;i<n;i++){
        d_type tmp=arr[i];
        int j=i-1;
        for (;j>=0 && arr[j]>tmp;j--) arr[j+1]=arr[j];
        arr[j+1]=tmp;
    }
}

// 2つのソート済み領域 [0, i) と [i , n) をインプレースでマージする関数
void inplace_merge(d_type arr[],int i,int n){
    if (arr[i]<arr[0]){
        d_type tmp=arr[i];
        int j=i-1;
        for (;j>=0;j--) arr[j+1]=arr[j];
        arr[0]=tmp;
        i++;
    }
    for (;i<n;i++){
        d_type tmp=arr[i];
        int j=i-1;
        for (;arr[j]>tmp;j--) arr[j+1]=arr[j];
        arr[j+1]=tmp;
    }
}

// 2つの変数の値を交換する
void swap(d_type *x, d_type *y){
    d_type tmp = *x;
    *x = *y;
    *y = tmp;
}

// i<mでarr[i]>arr[m+(m-1-i)]となる最小のi
int upper_bound(d_type arr[], int low, int m) {
    int high=m;
    while (low < high) {
        int mid = low + (high - low) / 2;
        if (arr[mid] > arr[m+(m-1-mid)])
            high = mid;
        else
            low = mid + 1;
    }
    return low;
}

// 再帰的In-place マージ関数
void inPlaceMerge(d_type arr[], int m, int n) {
    while (m>0 && n>0){
        // マージするデータ数が少なくなったとき
        if (m+n<50){
            inplace_merge(arr, m, m+n);
            return;
        }

        // 分割点を見つける
        int i;
        if (m<=n) i=0; else i=m-n;
        i=upper_bound(arr, i, m);

        int k=i,d=m-i;
        for (;k<m;k++) swap(arr+k,arr+k+d);

        // 小さい方の区間だけ再帰処理し、大きい方の区間は次のループで処理
        if (m  <= n) {
            inPlaceMerge(arr, i, m-i);
            // inPlaceMerge(arr+m, m-i, n-m+i) は次のループで処理
            arr=arr+m;
            n = n-m+i;
            m=m-i;
        } else {
            inPlaceMerge(arr+m, m-i, n-m+i);
            // inPlaceMerge(arr, i, m-i) は次のループで処理
            n=m-i;
            m=i;
        }
    }
}

// ソート関数
void inPlaceMergeSort(d_type arr[], int n) {
    if (n<50) { // 小さいところは挿入ソートで高速化
        insertion_sort(arr, n);
        return;
    }

    int m = n / 2;
    inPlaceMergeSort(arr, m);
    inPlaceMergeSort(arr+m, n-m);
    inPlaceMerge(arr, m , n-m);
}

int main(){
    int n=100000000;
    int i;
    clock_t start,end;
    d_type *array=malloc(n*sizeof(d_type));
    srand((unsigned) time(NULL));
    for (i=0;i<n;i++){
        array[i]=rand()*(RAND_MAX+1)+rand();
    }
    start=clock();
    inPlaceMergeSort(array,n);
    end=clock();
    printf("%f秒  \n",(double)(end-start)/CLOCKS_PER_SEC);
    for (i=1;i<n;i++) if(array[i-1]>array[i]) printf("%d %d\n",i,array[i]);
    free(array);
    return 0;
}

最初に考えた関数 inPlaceMerge() は,次のものです。

// 再帰的In-place マージ関数
void inPlaceMerge(d_type arr[], int m, int n) {
    // マージ不要な場合
    if (m<=0 || n<=0) return;

    // マージするデータ数が少なくなったとき
    if (m+n<50){
        inplace_merge(arr, m, m+n);
        return;
    }

    // 分割点を見つける
    int i;
    if (m<=n) i=0; else i=m-n;
    i=upper_bound(arr, i, m);

    int k=i,d=m-i;
    for (;k<m;k++) swap(arr+k,arr+k+d);

    // 再帰的にマージ
    inPlaceMerge(arr, i, m-i);
    inPlaceMerge(arr+m, m-i, n-m+i);
}

ところが,次のような1億個のデータに対してスタックオーバーフローを起こしてしまいました。そこで,クイックソートに対するスタックオーバーフロー対策法を参考にして改良を加えました。
    int n=100000000;
    for (i=0;i<n;i++){
        array[i]=i;
        if (rand()%100==0)
            array[i]=(int)((double)rand()/(RAND_MAX+1)*n);
    }


省メモリ超高速マージソート
上記のインプレースマージソートに,データのメモリ領域の概ね半分のサイズの作業用メモリ領域を使用するマージソートを組み込み,僅かの追加メモリで高速に動 作するよう にしました。
追加のメモリは,マージするデータ数が少なくなったときの処理を高速化するための次のような変更に用いました。

       // マージするデータ数が少なくなったとき
        if (m+n<50){
            inplace_merge(arr, m, m+n);
            return;
        }

                ↓

        // マージするデータ数が少なくなったとき
        if (m<=wm){
            if (arr[m-1]>arr[m]) upper_merge(arr,m,n,work);
            return;
        }
        if (n<=wm){
            if (arr[m-1]>arr[m]) lower_merge(arr,m,n,work);
            return;
        }

作業用メモリ領域をデータのメモリ領域の 0.1% 程度に設定してもクイックソートより速くて,平均計算時間と最悪計算時間が共に O(n log n) である,安定な省メモリマージソートです。
また, 作業用メモリ領域を 0.01% 程度まで小さくするには,m=n*0.001+25 を m=n*0.0001+25 と変えるだけであり,このときの速度低下は極僅かです。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>

typedef int d_type;// ソートするキーの型

void insertion_sort(d_type arr[],int n){
    int i=1;
    for (;i<n;i++){
        d_type tmp=arr[i];
        int j=i-1;
        for (;j>=0 && arr[j]>tmp;j--) arr[j+1]=arr[j];
        arr[j+1]=tmp;
    }
}

// 2つの変数の値を交換する
void swap(d_type *x, d_type *y){
    d_type tmp = *x;
    *x = *y;
    *y = tmp;
}

//前側が作業用メモリに収まる場合の前方からのマージ
void upper_merge(d_type arr[],int m,int n,d_type work[]){
    d_type *a_last = arr+m+n-1;
    d_type *w_last = work+m-1;
    d_type *p1 = arr;
    d_type *p2 = work;
    while (p2<=w_last) *p2++ = *p1++;

    //p1=arr+m;
    p2=work;
    d_type *dest=arr;
    while (p1 <= a_last && p2 <= w_last) {
        if (*p1 < *p2)
            *dest++ = *p1++;
        else
            *dest++ = *p2++;
    }
    while (p2<=w_last) *dest++ = *p2++;
}

//後側が作業用メモリに収まる場合の後方からのマージ
void lower_merge(d_type arr[],int m,int n,d_type work[]){
    d_type *p1 = arr+m+n-1;
    d_type *p2 = work+n-1;
    while (p2>=work) *p2--=*p1--;

    //p1=arr+m-1;
    p2 = work+n-1;
    d_type *dest = arr+m+n-1;
    while (p1 >= arr && p2 >= work) {
        if (*p1 > *p2)
            *dest-- = *p1--;  
        else
            *dest-- = *p2--;
    }
    while (p2>=work) *dest--=*p2--;
}

// i<mでarr[i]>arr[m+(m-1-i)]となる最小のi
int upper_bound(d_type arr[], int low, int m) {
    int high=m;
    while (low < high) {
        int mid = low + (high - low) / 2;
        if (arr[mid] > arr[m+(m-1-mid)])
            high = mid;
        else
            low = mid + 1;
    }
    return low;
}

// 再帰的In-place マージ関数
void inPlaceMerge(d_type arr[], int m, int n, d_type work[], int wm) {
    while (m>0 && n>0){
        // マージするデータ数が少なくなったとき
        if (m<=wm){
            if (arr[m-1]>arr[m]) upper_merge(arr,m,n,work);
            return;
        }
        if (n<=wm){
            if (arr[m-1]>arr[m]) lower_merge(arr,m,n,work);
            return;
        }

        // 分割点を見つける
        int i;
        if (m<=n) i=0; else i=m-n;
        i=upper_bound(arr, i, m);

        int k=i,d=m-i;
        for (;k<m;k++) swap(arr+k,arr+k+d);

        // 小さい方の区間だけ再帰処理し、大きい方の区間は次のループで処理
        if (m <= n) {
            inPlaceMerge(arr, i, m-i, work, wm);
            // inPlaceMerge(arr+m, m-i,n-m+i, work, wm) は次のループで処理
            arr=arr+m;
            n = n-m+i;
            m=m-i;
        } else {
            inPlaceMerge(arr+m, m-i,n-m+i, work, wm);
            // inPlaceMerge(arr, i, m-i, work, wm) は次のループで処理
            n=m-i;
            m=i;
        }
    }
}

// ソート関数
void inPlaceMergeSort(d_type arr[], int n, d_type work[],int wm) {
    if (n<50) { // 小さいところは挿入ソートで高速化
        insertion_sort(arr, n);
        return;
    }

    int m = n/2;
    inPlaceMergeSort(arr, m,work,wm);
    inPlaceMergeSort(arr+m, n-m,work,wm);
    inPlaceMerge(arr, m , n-m,work,wm);
}

void MergeSort(d_type arr[], int n) {
    if (n<50) {
        insertion_sort(arr, n);
        return;
    }
    int wm=n*0.001+25;
    d_type *work=malloc(wm*sizeof(d_type));
    inPlaceMergeSort(arr,n,work,wm);
    free(work);
}

int main(){
    int n=100000000;
    int i;
    clock_t start,end;
    d_type *arr=malloc(n*sizeof(d_type));
    srand((unsigned) time(NULL));
    for (i=0;i<n;i++){
        arr[i]=rand()*(RAND_MAX+1)+rand();
    }
    start=clock();
    MergeSort(arr,n);
    end=clock();
    printf("%f秒  \n",(double)(end-start)/CLOCKS_PER_SEC);
    for (i=1;i<n;i++) if(arr[i-1]>arr[i]) printf("%d %d\n",i,arr[i]);
    free(arr);
    return 0;
}

マージソート,クイックソートとの比較
この省メモリマージソートで作業用メモリ領域を色々変えた場合と,ウキペディア等にあるクイックソートやマージソートに高速化の工夫を加えたソート,自作の安定インプレースマージソート,自作の不安定インプレースマージソートとの計算時間の比較をしてみました。

       実行例

準インプレースマージソート
実行結果は,この省メモリマージソートが,作業用メモリ領域を極端に減らしても速いことを示しています。1億個のデータに対して,作業用メモリ領域を僅か100個まで減らしても,クイックソートに匹敵する速さです。
10億個のデータに対してクイックソートに匹敵する速さを求めるなら,200個~300個分の作業用メモリ領域が必要です。
このソートは再帰を用いているため O(log n) の追加のメモリ領域が使われますが,作業用メモリ領域も同じ O(log n) となるようにしようと考えました。ただし,その場合,平均計算時間は O(n log n) とは成りません。
そこで,関数 MergeSort() を次のように変えると,追加のメモリ領域は O(log n) で,作業用メモリ領域はデータ数が百万個では wm=375 ,1千万個では wm=450 ,1億個では wm=525,10億個では wm=625 となるので,準インプレースマージソート呼んでも良いでしょう。
10億個のランダムなデータに対して,クイックソートと,作業用メモリが625個分の準インプレースマージソート,0.01%分(10万個分)のマージソート,0.1%分(100万個分)のマージソートで,実際に速度比較をしてみました。

void MergeSort(d_type arr[], int n) {
    if (n<50) {
        insertion_sort(arr, n);
        return;
    }
    int wm=0;
    int i=n;
    do{
        wm=wm+25;
        i=i/2;
    }while (i>=50);
    d_type *work=malloc(wm*sizeof(d_type));
    inPlaceMergeSort(arr,n,work,wm);
    free(work);
}

    実行例

10億個のランダムなデータに対して,作業用メモリ領域がデータ625個分のマージソートの方がクイックソートより速いという結果が得られました。
しかも,n≧100でnがある程度小さいときは,wmは n*0.001+25 より大きくなるので wm=n*0.001+25 の場合よりも高速です。
しかし,nが10億のときは,wm=n*0.0001 の場合よりも大分遅くなってしまいます。そこで,このコードの1行を次のように変えてみます。
    int wm=0;    →    int wm=n*0.0001;
この変更によりwmの値は,nが小さいときはほとんど変わらず,nが十分大きいときはほぼ wm=n*0.0001 となって,平均計算時間は O(n log n) になります。
コード全体は次のようです。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>

typedef int d_type;// ソートするキーの型

void insertion_sort(d_type arr[],int n){
    int i=1;
    for (;i<n;i++){
        d_type tmp=arr[i];
        int j=i-1;
        for (;j>=0 && arr[j]>tmp;j--) arr[j+1]=arr[j];
        arr[j+1]=tmp;
    }
}

// 2つの変数の値を交換する
void swap(d_type *x, d_type *y){
    d_type tmp = *x;
    *x = *y;
    *y = tmp;
}

//前側が作業用メモリに収まる場合の前方からのマージ
void upper_merge(d_type arr[],int m,int n,d_type work[]){
    d_type *a_last = arr+m+n-1;
    d_type *w_last = work+m-1;
    d_type *p1 = arr;
    d_type *p2 = work;
    while (p2<=w_last) *p2++ = *p1++;

    //p1=arr+m;
    p2=work;
    d_type *dest=arr;
    while (p1 <= a_last && p2 <= w_last) {
        if (*p1 < *p2)
            *dest++ = *p1++;
        else
            *dest++ = *p2++;
    }
    while (p2<=w_last) *dest++ = *p2++;
}

//後側が作業用メモリに収まる場合の後方からのマージ
void lower_merge(d_type arr[],int m,int n,d_type work[]){
    d_type *p1 = arr+m+n-1;
    d_type *p2 = work+n-1;
    while (p2>=work) *p2--=*p1--;

    //p1=arr+m-1;
    p2 = work+n-1;
    d_type *dest = arr+m+n-1;
    while (p1 >= arr && p2 >= work) {
        if (*p1 > *p2)
            *dest-- = *p1--;  
        else
            *dest-- = *p2--;
    }
    while (p2>=work) *dest--=*p2--;
}

// i<mでarr[i]>arr[m+(m-1-i)]となる最小のi
int upper_bound(d_type arr[], int low, int m) {
    int high=m;
    while (low < high) {
        int mid = low + (high - low) / 2;
        if (arr[mid] > arr[m+(m-1-mid)])
            high = mid;
        else
            low = mid + 1;
    }
    return low;
}

// 再帰的In-place マージ関数
void inPlaceMerge(d_type arr[], int m, int n, d_type work[], int wm) {
    while (m>0 && n>0){
        // マージするデータ数が少なくなったとき
        if (m<=wm){
            if (arr[m-1]>arr[m]) upper_merge(arr,m,n,work);
            return;
        }
        if (n<=wm){
            if (arr[m-1]>arr[m]) lower_merge(arr,m,n,work);
            return;
        }

        // 分割点を見つける
        int i;
        if (m<=n) i=0; else i=m-n;
        i=upper_bound(arr, i, m);

        int k=i,d=m-i;
        for (;k<m;k++) swap(arr+k,arr+k+d);

        // 小さい方の区間だけ再帰処理し、大きい方の区間は次のループで処理
        if (m <= n) {
            inPlaceMerge(arr, i, m-i, work, wm);
            // inPlaceMerge(arr+m, m-i,n-m+i, work, wm) は次のループで処理
            arr=arr+m;
            n = n-m+i;
            m=m-i;
        } else {
            inPlaceMerge(arr+m, m-i,n-m+i, work, wm);
            // inPlaceMerge(arr, i, m-i, work, wm) は次のループで処理
            n=m-i;
            m=i;
        }
    }
}

// ソート関数
void inPlaceMergeSort(d_type arr[], int n, d_type work[],int wm) {
    if (n<50) { // 小さいところは挿入ソートで高速化
        insertion_sort(arr, n);
        return;
    }

    int m = n/2;
    inPlaceMergeSort(arr, m,work,wm);
    inPlaceMergeSort(arr+m, n-m,work,wm);
    inPlaceMerge(arr, m , n-m,work,wm);
}

void MergeSort(d_type arr[], int n) {
    if (n<50) {
        insertion_sort(arr, n);
        return;
    }
    int wm=n*0.0001;
    int i=n;
    do {
        wm=wm+25;
        i=i/2;
    }while (i>=50);
    d_type *work=malloc(wm*sizeof(d_type));
    inPlaceMergeSort(arr,n,work,wm);
    free(work);
}

int main(){
    int n=100000000;
    int i;
    clock_t start,end;
    d_type *arr=malloc(n*sizeof(d_type));
    srand((unsigned) time(NULL));
    for (i=0;i<n;i++){
        arr[i]=rand()*(RAND_MAX+1)+rand();
    }
    start=clock();
    MergeSort(arr,n);
    end=clock();
    printf("%f秒  \n",(double)(end-start)/CLOCKS_PER_SEC);
    for (i=1;i<n;i++) if(arr[i-1]>arr[i]) printf("%d %d\n",i,arr[i]);
    free(arr);
    return 0;
}

安定ソートであることを構造体を用いて確認するコード
#include <stdio.h>
#include <stdlib.h>
#include <time.h>

typedef struct{
    int index;    // 元の順序(インデックス)
    int value;    // ソートする値
} d_type;

int compare(const void *a, const void *b) {
    int x = ((d_type *)a)->value;
    int y = ((d_type *)b)->value;
    if(x<y) return -1;
    if(x>y) return 1;
    return 0;
}

void insertion_sort(d_type arr[],int n){
    int i=1;
    for (;i<n;i++){
        d_type tmp=arr[i];
        int j=i-1;
        for (;j>=0 && compare(arr+j,&tmp)>0;j--) arr[j+1]=arr[j];
        arr[j+1]=tmp;
    }
}

// 2つの変数の値を交換する
void swap(d_type *x, d_type *y){
    d_type tmp = *x;
    *x = *y;
    *y = tmp;
}

//前側が作業用メモリに収まる場合の前方からのマージ
void upper_merge(d_type arr[],int m,int n,d_type work[]){
    d_type *a_last = arr+m+n-1;
    d_type *w_last = work+m-1;
    d_type *p1 = arr;
    d_type *p2 = work;
    while (p2<=w_last) *p2++ = *p1++;

    p2=work;
    d_type *dest=arr;
    while (p1 <= a_last && p2 <= w_last) {
        if (compare(p1,p2)<0)
            *dest++ = *p1++;
        else
            *dest++ = *p2++;
    }
    while (p2<=w_last) *dest++ = *p2++;
}

//後側が作業用メモリに収まる場合の後方からのマージ
void lower_merge(d_type arr[],int m,int n,d_type work[]){
    d_type *p1 = arr+m+n-1;
    d_type *p2 = work+n-1;
    while (p2>=work) *p2--=*p1--;

    p2 = work+n-1;
    d_type *dest = arr+m+n-1;
    while (p1 >= arr && p2 >= work) {
        if (compare(p1,p2)>0)
            *dest-- = *p1--;   
        else
            *dest-- = *p2--;
    }
    while (p2>=work) *dest--=*p2--;
}

// i<mでarr[i]>arr[m+(m-1-i)]となる最小のi
int upper_bound(d_type arr[], int low, int m) {
    int high=m;
    while (low < high) {
        int mid = low + (high - low) / 2;
        if (compare(arr+mid,arr+m+(m-1-mid))>0)
            high = mid;
        else
            low = mid + 1;
    }
    return low;
}

// 再帰的In-place マージ関数
void inPlaceMerge(d_type arr[], int m, int n, d_type work[], int wm) {
    while (m>0 && n>0){
        // マージするデータ数が少なくなったとき
        if (m<=wm){
            if (compare(arr+m-1,arr+m)>0) upper_merge(arr,m,n,work);
            return;
        }
        if (n<=wm){
            if (compare(arr+m-1,arr+m)>0) lower_merge(arr,m,n,work);
            return;
        }

        // 分割点を見つける
        int i;
        if (m<=n) i=0; else i=m-n;
        i=upper_bound(arr, i, m);

        int k=i,d=m-i;
        for (;k<m;k++) swap(arr+k,arr+k+d);

        // 小さい方の区間だけ再帰処理し、大きい方の区間は次のループで処理
        if (m <= n) {
            inPlaceMerge(arr, i, m-i, work, wm);
            arr=arr+m; // 大きい方の区間を次のループへ
            n = n-m+i;
            m=m-i;
        } else {
            inPlaceMerge(arr+m, m-i,n-m+i, work, wm);
            n=m-i; // 大きい方の区間を次のループへ
            m=i;
        }
    }
}

// ソート関数
void inPlaceMergeSort(d_type arr[], int n, d_type work[],int wm) {
    if (n<50) { // 小さいところは挿入ソートで高速化
        insertion_sort(arr, n);
        return;
    }

    int m = n/2;
    inPlaceMergeSort(arr, m,work,wm);
    inPlaceMergeSort(arr+m, n-m,work,wm);
    inPlaceMerge(arr, m , n-m,work,wm);
}

void MergeSort(d_type arr[], int n) {
    int wm=n*0.001+25;
    d_type *work=malloc(wm*sizeof(d_type));
    inPlaceMergeSort(arr,n,work,wm);
    free(work);
}

// 安定性をチェックする関数
void check_stability(d_type arr[], int n) {
    int i;
    for (i = 0; i < n - 1; i++) {
        if (arr[i].value==arr[i+1].value) {
            if (arr[i].index > arr[i + 1].index){
                printf("不安定です \n");
                return;
            }
        }
    }
    printf("安定の様です \n");
}

int main() {
    int n=100000000;
    int i;
    clock_t start,end;
    d_type *arr=(d_type *)malloc(n*sizeof(d_type));
    srand((unsigned) time(NULL));
    for (i=0;i<n;i++){
        arr[i].index=i;
        arr[i].value=rand();
    }
    start=clock();
    MergeSort(arr, n);
    //qsort(arr, n, sizeof(d_type), compare);
    end=clock();
    printf("%f秒  \n",(double)(end-start)/CLOCKS_PER_SEC);
    for (i=1;i<n;i++) if(arr[i-1].value > arr[i ].value) printf("%d %d\n",i,arr[i].value);

    // 結果判定
    check_stability(arr, n);
    return 0;
}

メモリ不足の場合,使用するメモリを自動的に削減する工夫
関数 MergeSort() を次の様に変えると,メモリ不足の場合には,work[] に使用するメモリを,使用できるメモリに応じてデータ1個分まで自動的に削減します。
ただし,wm=1 ではインプレースマージソートよりも大分遅くなってしまいますので,そのような場合はインプレースマージソートを用いて下さい。

void MergeSort(d_type arr[], int n) {
    if (n<50) {
        insertion_sort(arr, n);
        return;
    }
    int wm=n*0.001+25;
    do {
        d_type *work=malloc(wm*sizeof(d_type));
        if (work!=NULL){
            inPlaceMergeSort(arr,n,work,wm);
            free(work);
            return;
        }
        wm=wm/2;
    } while (wm>=1);
    //作業用メモリがデータ1個分も確保できないとき
}

汎用不安定ソートである qsort の代わりとなる汎用安定ソート msort
C言語の標準ライブラリ関数 qsort は, stdlib.h の中で宣言されている汎用ソートで,使いやすく実用には十分な速度で,欠点は不安定ソートであることです。
上記のマージソートを手直しすれば,qsort と同じインターフェース(比較関数の引数など)の汎用省メモリ安定ソートが作れるのではと考え,実際にそれを作成して msort と名付けました。
速度は,残念ながら極端に遅くなり,qsort よりもだいぶ遅くなってしまいました。qsortには,高速化のための高度な工夫がされている様です。
作業用メモリ量は、初期設定では元のデータのメモリ量の5%程度で,メモリ不足の場合は作業用メモリを自動的にデータ1個分まで削減します。
コードのコメントになっている qsort(arr, n, sizeof(d_type), compare) を実行して,msort と qsort を比較してみて下さい。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>

// qsort() と同じ使い方にするには、次の様にします
// 以下のコードの「msort.h の終わり」までを msort.h として保存し、その部分を削除する。
// #include "msort.h" をこのコードに付け加える。

//msort.h の始まり
#include <string.h>

#define int_CMP  int (*CMP)(const void *, const void *)

void insertion_sort(char *arr, size_t n, size_t size, int_CMP,char *work) {
    char *i=arr+size,*last=arr+n*size;
    for (; i<last; i+=size) {
        char *j=i;
        memcpy(work,j,size);
        for (;j>arr && CMP(j-size,work)>0;j-=size) memcpy(j,j-size,size);
        memcpy(j,work,size);
    }
}

//前側が作業用メモリに収まる場合の前方からのマージ
void upper_merge(char *arr,size_t m,size_t n, size_t size, int_CMP,char *work){
    memcpy(work,arr,m*size);
    char *a_last=arr+(m+n-1)*size;
    char *w_last=work+(m-1)*size;
    char *p1=arr+m*size;
    char *p2=work;

    char *dest=arr;
    while (p1<=a_last && p2<=w_last) {
        if (CMP(p1,p2)<0) {
            memcpy(dest,p1,size);
            dest+=size;p1+=size;
        } else {
            memcpy(dest,p2,size);
            dest+=size;p2+=size;
        }
    }
    while(p2<=w_last) {
        memcpy(dest,p2,size);
        dest+=size;p2+=size;
    }
}

//後側が作業用メモリに収まる場合の後方からのマージ
void lower_merge(char *arr,size_t m,size_t n, size_t size, int_CMP,char *work){
    memcpy(work,arr+m*size,n*size);
    char *p1=arr+(m-1)*size;
    char *p2=work+(n-1)*size;

    char *dest=arr+(m+n-1)*size;
    while (p1>=arr && p2>=work) {
        if (CMP(p1,p2)>0) {
            memcpy(dest,p1,size);
            dest-=size;p1-=size;
        } else {
            memcpy(dest,p2,size);
            dest-=size;p2-=size;
        }
    }
    while(p2>=work) {
        memcpy(dest,p2,size);
        dest-=size;p2-=size;
    }
}

// i<mでarr[i]>arr[m+(m-1-i)]となる最小のi
size_t upper_bound(char *arr, size_t low, size_t m, size_t size, int_CMP) {
    size_t high=m;
    while (low < high) {
        size_t mid = low + (high - low) / 2;
        if (CMP(arr+mid*size,arr+(m+(m-1-mid))*size)>0)
            high = mid;
        else
            low = mid + 1;
    }
    return low;
}

// 再帰的In-place マージ関数
void inPlaceMerge(char *arr, size_t m, size_t n, size_t size, int_CMP, char *work, size_t wm) {
    while (m>0 && n>0){
        // マージするデータ数が少なくなったとき
        if (m<=wm){
            if (CMP(arr+(m-1)*size,arr+m*size)>0)
                upper_merge(arr, m, n, size, CMP, work);
            return;
        }
        if (n<=wm){
            if (CMP(arr+(m-1)*size,arr+m*size)>0)
                lower_merge(arr, m, n, size, CMP, work);
            return;
        }

        // 分割点を見つける
        size_t i;
        if (m<=n) i=0; else i=m-n;
        i=upper_bound(arr, i, m, size, CMP);

        char *k=arr+i*size,*right=arr+m*size;
        size_t dsize=right-k;
        for (; k < right; k+=size) { //変数の値を交換する
            char *j = k+dsize;
            memcpy(work, k, size);
            memcpy(k, j, size);
            memcpy(j, work, size);
        }

        // 小さい方の区間だけ再帰処理し、大きい方の区間は次のループで処理
        if (m <= n) {
            inPlaceMerge(arr, i, m-i, size, CMP, work, wm);
            arr=arr+m*size; // 大きい方の区間を次のループへ
            n = n-m+i;
            m=m-i;
        } else {
            inPlaceMerge(arr+m*size, m-i,n-m+i, size, CMP, work, wm);
            n=m-i; // 大きい方の区間を次のループへ
            m=i;
        }
    }
}

// ソート関数
void inPlaceMergeSort(char *arr, size_t n, size_t size, int_CMP, char *work, size_t wm) {
    if (n<=16) { // 小さいところは挿入ソートで高速化
        insertion_sort(arr, n, size, CMP, work);
        return;
    }

    size_t m = n/2;
    inPlaceMergeSort(arr, m, size, CMP, work, wm);
    inPlaceMergeSort(arr+m*size, n-m, size, CMP, work, wm);
    inPlaceMerge(arr, m , n-m, size, CMP, work, wm);
}

void msort(void *base, size_t n, size_t size, int_CMP) {
    char *arr = (char *)base;
    size_t wm=n*0.05+6;
    do {
        char *work=(char *)malloc(wm*size);
        if (work!=NULL){
            inPlaceMergeSort(arr,n, size, CMP,work,wm);
            free(work);
            return;
        }
        wm=wm/2;
    } while (wm>=1);
    //作業用メモリがデータ1個分も確保できないとき
}

#undef int_CMP
// msort.h の終わり

typedef struct{
    int index;    // 元の順序(インデックス)
    int value;    // ソートする値
} d_type;

int compare(const void *a, const void *b) {
    int x = ((d_type *)a)->value;
    int y = ((d_type *)b)->value;
    if(x<y) return -1;
    if(x>y) return 1;
    return 0;
}

// 安定性をチェックする関数
void check_stability(d_type arr[], int n) {
    int i;
    for (i = 0; i < n - 1; i++) {
        if (arr[i].value==arr[i+1].value) {
            if (arr[i].index > arr[i + 1].index){
                printf("不安定です \n");
                return;
            }
        }
    }
    printf("安定の様です \n");
}

int main() {
    size_t n=100000000;
    int i;
    clock_t start,end;
    d_type *arr=(d_type *)malloc(n*sizeof(d_type));
    srand((unsigned) time(NULL));
    for (i=0;i<n;i++){
        arr[i].index=i;
        arr[i].value=rand()*(RAND_MAX+1)+rand();
    }
    start=clock();
    msort(arr, n, sizeof(d_type), compare);
    //qsort(arr, n, sizeof(d_type), compare);
    end=clock();
    printf("%f秒  \n",(double)(end-start)/CLOCKS_PER_SEC);
    for (i=1;i<n;i++) if(arr[i-1].value > arr[i ].value) printf("%d %d\n",i,arr[i].value);

    check_stability(arr, n);// 安定性の判定
    free(arr);
    return 0;
}



ランダムなデータに対してクイックソートに匹敵する速さの不安定インプレースマージソート

In-Place Sorting With Merge Sort に,平均計算時間と最悪計算時間が共に O(n log n)で,安定ソートではないインプレースマージソートがありますが,それは一般のマージソートより遅いものです。
私も,平均計算時間と最悪計算時間が共に O(n log n)で,安定ソートではないインプレースマージソートを作りました。再帰呼 び出しを用いていて,追加メモリ領域は O(log n) です。(2025年1月)
ここで示すインプレースマージソートは,In-Place Sorting With Merge Sort にあるソートに幾つかの改良を施した形で,高速になっています。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>

typedef int d_type;// ソートするキーの型

//挿入ソート
#define INSERTION_SORT(arr,m,n){\
    int i=m;\
    for (;i<n;i++){\
        d_type tmp=arr[i];\
        int j=i-1;\
        for (;j>=0 && arr[j]>tmp;j--) arr[j+1]=arr[j];\
        arr[j+1]=tmp;\
    }\
}

#define SWAP(x,y){\
    d_type tmp=*x;\
    *x=*y;\
    *y=tmp;\
}

#define SWAP__(x,y){\
    SWAP(x,y)\
    x--;y--;\
}

// マージ (整列されたm個,整列されていないn 個,整列されたn個)
// → (整列されたm+n個,整列されていないn個) を行う関数
void mergeD(d_type arr[], int m, int n){
    d_type *p1=arr+m-1;
    d_type *p2=p1+2*n;
    d_type *dest=p1+n;

    while (1) {
        if (*p1>*p2) {
            SWAP__(dest,p1)
            if (p1<arr) {
                while(dest>=arr) SWAP__(dest,p2)
                return;
            }
        } else {
            SWAP__(dest,p2)
            if (dest<=p1) return;
        }
    }
}
#undef SWAP__

#define RATIO 0.18 //RATIO<=0.5
//整列されたn個の要素の配列を作る関数
void merge_sortD(d_type arr[], int n){
    if (n<50) {
        INSERTION_SORT(arr,1,n)
        return;
    }
    int k=n*RATIO,m=n-k;//k=n/2をk=n*0.18に変えたら,なぜか高速化
    merge_sortD(arr,m);
    merge_sortD(arr+n,k);
    mergeD(arr,m,k);
}

#define CompSWAP(a,b){\
    d_type *x=arr+a,*y=arr+b;\
    if (*x<*y) SWAP(x,y)\
}

void merge_sort(d_type arr[], int n){
    int m=n*(1.0-RATIO); //mをできるだけ大きくした
    merge_sortD(arr,m); //整列されたm個の要素の配列を作る
    while (1){ //m個の要素の配列にk個の要素をマージすることを繰り返す
        int k=(n-m)/2;
        if (k<2)break;
        int i=m+1;
        for (;i<m+k;i++) {
            CompSWAP(i-1,i)
            CompSWAP(i,i-1+k)
        }
        merge_sort(arr+i,n-i); //i=m+k
        mergeD(arr,m,k);
        m=i;
    }
    INSERTION_SORT(arr,m,n) //残った要素を挿入ソートする
}
#undef CompSWAP
#undef SWAP
#undef RATIO
#undef INSERTION_SORT

int main(){
    int n=100000000;
    int i;
    clock_t start,end;
    d_type *arr=malloc(n*sizeof(d_type));
    srand((unsigned) time(NULL));
    for (i=0;i<n;i++) arr[i]=rand()*(RAND_MAX+1)+rand();
    start=clock();
    merge_sort(arr,n);
    end=clock();
    printf("%f秒  \n",(double)(end-start)/CLOCKS_PER_SEC);
    for (i=1;i<n;i++) if(arr[i-1]>arr[i]) printf("%d %d\n",i,arr[i]);
    free(arr);
    return 0;
}

上記の merge_sort() は,基本的なコードに次のコードを付加することにより,大きく高速化されています。

        int i=m+1;
        for (;i<m+k;i++) {
            Exchange(i-1,i);
            Exchange(i,i-1+k);
        }

上記のプログラムがやっていることの概略は,次の様です。
1 全データの82%分のソートされた配列を作る。(この作業は,関数 merge_sortD() により高速に行われる)
2 残りの18%のうちの半分の9%をソートされた配列にし,それをソートされた82%の配列にマージして,ソートされた91%分の配列を作る。
3 残りの9%のうちの半分の4.5%をソートされた配列にし,それをソートされた91%の配列にマージして,ソートされた95.5%分の配列を作る。
4 残りの4.5%のうちの半分の2.25%をソートされた配列にし,それをソートされた95.5%の配列にマージして,ソートされた97.75%分の配列を作る。
          …………

最初のコードでは,できるだけデータ数がほぼ等しいデータ群をマージしようとしていました。すなわち,m≒k となる割合を多くすれば高速になると思い込んでいました。
ところが,コードの一部の「k=n/2」を「k=n*0.18」に変えたところ,不思議にも大きく速くなりました。

注 関数 merge_sort() にある「m=n*(1.0-RATIO);//mをできるだけ大きくした」は,関数 merge_sortD()を用いるに当たり,次の条件が満たされれば良いと考えたからです。
    m+m×RATIO+m×(RATIO)2+m×(RATIO)3+m×(RATIO)4+……≦n
ここで,無限等比級数の和の公式により
    左辺=m/(1-RATIO)
であるから
    m/(1-RATIO)≦n
よって
    m≦n×(1-RATIO)


理由が大して分からずマジックのように速くなったソートですが,マー ジソートの考案者のフォン・ノイマンもマジマジと見ているかもしれません。
最初のページ に戻る