C++でマルチスレッドを使ってマージソートを実装する方法

本記事では、ソートされていない整数型配列が与えられたとき、マルチスレッドを活用したマージソート（Merge Sort）で配列を並べ替える方法を解説します。

マージソートとは

マージソートは「分割統治法（Divide and Conquer）」に基づくソートアルゴリズムです。配列を半分ずつに分割していき、最後にそれらを整列させながら統合（マージ）することで、全体を昇順に並べ替えます。

マージソートのアルゴリズム

• リスト内の要素が1つだけなら、その要素をそのまま返します。
• そうでなければ、データを再帰的に2つに分割し、それ以上分割できなくなるまで繰り返します。
• 最後に、小さなリスト同士を整列順に保ちながら新しいリストへマージしていきます。

マルチスレッドとは

オペレーティングシステムにおけるスレッド（Thread）は、タスクの一部を実行する軽量プロセスです。複数のスレッドは共通のリソースを共有しながら、処理を並行して実行できます。

マルチスレッドはマルチタスクの一種の実装方式で、1つのプロセッサ上で複数のスレッドを動かし、タスクを同時に実行することを可能にします。1つのアプリケーション内部の特定の処理を個別のスレッドに細分化し、それぞれのスレッドを並列に走らせることで、処理速度の向上が期待できます。

実行例

入力：int arr[] = {3, 2, 1, 10, 8, 5, 7, 9, 4}

出力：ソート済み配列： 1, 2, 3, 4, 5, 7, 8, 9, 10

説明：整数値を持つ未ソート配列が与えられます。これをマルチスレッドによるマージソートで並べ替えます。

入力：int arr[] = {5, 3, 1, 45, 32, 21, 50}

出力：ソート済み配列： 1, 3, 5, 21, 32, 45, 50

説明：こちらも同様に、未ソートの整数配列をマルチスレッド版マージソートで整列させます。

プログラムのアプローチ

• C++ STL の rand() メソッドを使って乱数を生成し、配列の初期値とします。
• pthread_t 型の配列 P_TH[thread_size] を作成します。
• i = 0 から i がスレッド数未満になるまでループを回し、pthread_create(&P_TH[i], NULL, Sorting_Threading, (void*)NULL) を呼び出してスレッドを生成します。
• pthread_join() ですべてのスレッドの完了を待ち合わせた後、combine_array(0, (size / 2 - 1) / 2, size / 2 - 1)、combine_array(size / 2, size/2 + (size-1-size/2)/2, size - 1)、combine_array(0, (size - 1)/2, size - 1) の順に呼び出して、各部分配列を統合します。
• 整数型配列 arr[] に格納されたソート結果を出力します。

関数 void* Sorting_Threading(void* arg) の内部

• 変数 set_val を temp_val++ の値として宣言し、first = set_val * (size / 4)、end = (set_val + 1) * (size / 4) - 1、mid_val = first + (end - first) / 2 を求めます。
• first < end である場合、Sorting_Threading(first, mid_val)、Sorting_Threading(mid_val + 1, end)、combine_array(first, mid_val, end) を呼び出します。

関数 void Sorting_Threading(int first, int end) の内部

• 変数 mid_val を first + (end - first) / 2 として宣言します。
• first < end である場合、再帰的に Sorting_Threading(first, mid_val)、Sorting_Threading(mid_val + 1, end)、combine_array(first, mid_val, end) を呼び出します。

関数 void combine_array(int first, int mid_val, int end) の内部

• int* start = new int[mid_val - first + 1]、int* last = new int[end - mid_val]、temp_1 = mid_val - first + 1、temp_2 = end - mid_val、i、j、k = first を宣言します。
• i = 0 から i < temp_1 までのループで start[i] = arr[i + first] を設定します。
• i = 0 から i < temp_2 までのループで last[i] = arr[i + mid_val + 1] を設定します。
• i と j を 0 に初期化し、i < temp_1 かつ j < temp_2 の間ループします。start[i] <= last[j] なら arr[k++] = start[i++]、そうでなければ arr[k++] = last[j++] とします。
• 残りの要素についても、i < temp_1 の間は arr[k++] = start[i++]、j < temp_2 の間は arr[k++] = last[j++] としてコピーします。

サンプルコード

#include <iostream>
#include <pthread.h>
#include <time.h>
#define size 20
#define thread_size 4
using namespace std;
int arr[size];
int temp_val = 0;
void combine_array(int first, int mid_val, int end){
    int* start = new int[mid_val - first + 1];
    int* last = new int[end - mid_val];
    int temp_1 = mid_val - first + 1;
    int temp_2 = end - mid_val;
    int i, j;
    int k = first;
    for(i = 0; i < temp_1; i++){
        start[i] = arr[i + first];
    }
    for (i = 0; i < temp_2; i++){
        last[i] = arr[i + mid_val + 1];
    }
    i = j = 0;
    while(i < temp_1 && j < temp_2){
        if(start[i] <= last[j]){
            arr[k++] = start[i++];
        }
        else{
            arr[k++] = last[j++];
        }
    }
    while (i < temp_1){
        arr[k++] = start[i++];
    }
    while (j < temp_2){
        arr[k++] = last[j++];
    }
}
void Sorting_Threading(int first, int end){
    int mid_val = first + (end - first) / 2;
    if(first < end){
        Sorting_Threading(first, mid_val);
        Sorting_Threading(mid_val + 1, end);
        combine_array(first, mid_val, end);
    }
}
void* Sorting_Threading(void* arg){
    int set_val = temp_val++;
    int first = set_val * (size / 4);
    int end = (set_val + 1) * (size / 4) - 1;
    int mid_val = first + (end - first) / 2;
    if (first < end){
        Sorting_Threading(first, mid_val);
        Sorting_Threading(mid_val + 1, end);
        combine_array(first, mid_val, end);
    }
}
int main(){
    for(int i = 0; i < size; i++){
        arr[i] = rand() % 100;
    }
    pthread_t P_TH[thread_size];
    for(int i = 0; i < thread_size; i++){
        pthread_create(&P_TH[i], NULL, Sorting_Threading, (void*)NULL);
    }
    for(int i = 0; i < 4; i++){
        pthread_join(P_TH[i], NULL);
    }
    combine_array(0, (size / 2 - 1) / 2, size / 2 - 1);
    combine_array(size / 2, size/2 + (size-1-size/2)/2, size - 1);
    combine_array(0, (size - 1)/2, size - 1);
    cout<<"Merge Sort using Multi-threading: ";
    for (int i = 0; i < size; i++){
        cout << arr[i] << " ";
    }
    return 0;
}

出力結果

上記のコードを実行すると、次のような出力が得られます。

Merge Sort using Multi-threading: 15 21 26 26 27 35 36 40 49 59 62 63 72 77 83 86 86 90 92 93

このように、POSIXスレッド（pthread）を利用して配列を4つのブロックに分割し、各スレッドが担当範囲を独立してマージソートすることで、大規模なデータの並べ替えを効率化できます。最後にメインスレッドがすべての部分配列を統合し、完全にソートされた配列が完成します。