アルゴリズム2：FizzBuzz問題
FizzBuzz問題はご存知ですね。1から任意の数までについて出力します。しかし、3の倍数の時は「Fizz」、5の倍数の時は「Buzz」と出力します。これを単純にコード化すると、次のようになります。
void FizzBuzz(const int この数まで)
{
for (int i = 1; i <= この数まで; ++i) {
if (i % 15 == 0) {
std::cout << "FizzBuzz ";
} else if (i % 3 == 0) {
std::cout << "Fizz ";
} else if (i % 5 == 0) {
std::cout << "Buzz ";
} else {
std::cout << i << " ";
}
}
}
しかし、こうすると、出力結果でコンソールが大変なことになります。また、このように出力すると並列化したときに「実行順」が問題になります。そのため、ここでは文字列領域を確保し、そこに入れていくことにします。これも動作の検証をして、並列化したコードも作ります。ここでは動作検証は省きますが、実際にやってみる場合は、しっかり検証してください。
#define FIZZBUZZMAX 5000000
char FB結果[FIZZBUZZMAX][12];
void FizzBuzz(const int この数まで)
{
for (int i = 1; i <= この数まで; ++i) {
if (i % 15 == 0) {
strcpy_s(FB結果[i], sizeof(FB結果[i]), "FizzBuzz");
} else if (i % 3 == 0) {
strcpy_s(FB結果[i], sizeof(FB結果[i]), "Fizz");
} else if (i % 5 == 0) {
strcpy_s(FB結果[i], sizeof(FB結果[i]), "Buzz");
} else {
sprintf_s(FB結果[i], sizeof(FB結果[i]), "%d", i);
}
}
}
void FizzBuzzParallel(const int この数まで)
{
#pragma omp parallel for
for (int i = 1; i <= この数まで; ++i) {
if (i % 15 == 0) {
strcpy_s(FB結果[i], sizeof(FB結果[i]), "FizzBuzz");
} else if (i % 3 == 0) {
strcpy_s(FB結果[i], sizeof(FB結果[i]), "Fizz");
} else if (i % 5 == 0) {
strcpy_s(FB結果[i], sizeof(FB結果[i]), "Buzz");
} else {
sprintf_s(FB結果[i], sizeof(FB結果[i]), "%d", i);
}
}
}
アルゴリズム3：バブルソート
3つ目のアルゴリズムは、バブルソートです。
バブルソートのアルゴリズムをおさらいしましょう。配列に数値が並んでいます。この数値を、小さい順に並べるとします。まず、配列の1番後ろと、その1つ手前を比べます。後ろにある方が大きければ、入れ替えます。次に、後ろから2番目と、その1つ手前を比べます。そして、1つ手前の方が大きければ入れ替えます。これを、最初の要素まで繰り返します。最初までくると、「1番小さい数値」が決定します。そこで今度は、「2番目に小さい数値」を決定すべく、また1番後ろから比較を繰り返します。
#define BSORTNUM 100000
int BS配列[BSORTNUM];
void BubbleSort(int* 対象配列, const int 配列数)
{
for (int 決定済み数 = 0; 決定済み数 < 配列数; ++決定済み数) {
for (int y = 配列数 - 1; y > 決定済み数; --y) {
if (対象配列[y] < 対象配列[y - 1]) {
int tmp = 対象配列[y];
対象配列[y] = 対象配列[y - 1];
対象配列[y - 1] = tmp;
}
}
}
}
void BubbleSortParallel(int* 対象配列, const int 配列数)
{
#pragma omp parallel for
for (int 決定済み数 = 0; 決定済み数 < 配列数; ++決定済み数) {
for (int y = 配列数 - 1; y > 決定済み数; --y) {
if (対象配列[y] < 対象配列[y - 1]) {
int tmp = 対象配列[y];
対象配列[y] = 対象配列[y - 1];
対象配列[y - 1] = tmp;
}
}
}
}
さて、これも検証しておきます。すると、並列化した方で、結果がおかしいことが分かります。以下の結果は、BSORTNUMを100として検証したときの結果です。「BS配列」は、1～BSORTNUMで埋め、任意の2つを入れ替えてシャッフルした後、実行しました。
void Shuffle(int *対象配列, const int 配列数)
{
for (int i = 0; i < 配列数; ++i) {
対象配列[i] = i + 1; }
int half = 配列数 / 2; int r = rand() % half;
// 配列数が奇数だと最後の数はシャッフルされない
for (int i = 0; i < (half + r); ++i) {
int b = rand() % half + half;
int s = rand() % half;
int tmp = 対象配列[b];
対象配列[b] = 対象配列[s];
対象配列[s] = tmp;
}
}
int _tmain(int argc, _TCHAR* argv[])
{
srand((unsigned int) time(0));
for (int i = 0; i < 1; ++i) {
Shuffle(BS配列, BSORTNUM);
std::cout << "ソート前\n";
for (int i = 0; i < BSORTNUM; ++i) {
std::cout << BS配列[i] << "\t";
}
std::cout << "\n----------\nソート後\n";
BubbleSortParallel(BS配列, BSORTNUM);
}
for (int i = 0; i < BSORTNUM; ++i) {
std::cout << BS配列[i] << "\t";
}
return 0;
}
ソート前 88 25 18 83 54 90 79 2 49 73 94 92 13 14 97 52 96 75 27 38 100 39 61 40 59 24 7 26 29 30 42 31 23 55 53 36 37 91 11 98 62 74 43 66 71 46 47 95 58 16 51 45 76 6 99 17 19 12 82 60 33 41 35 64 84 5 67 22 68 70 50 72 10 65 80 86 77 44 9 78 81 85 93 32 87 69 8 1 89 3 20 15 4 63 48 56 28 21 34 57 ---------- ソート後 1 2 3 4 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 21 22 23 24 25 26 27 28 29 30 31 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 88 50 51 83 54 90 79 52 53 73 94 92 55 56 97 57 96 75 58 59 100 60 61 62 63 64 65 66 67 68 69 70 71 72 74 76 77 91 78 98 80 81 82 84 86 87 89 95 93 99
検証するクセを付けておいてよかったでしょ？
実は、バブルソートのアルゴリズムは、並列化に向いていません。複数の並列に行われる処理が、同じ領域を書き換えようとするためです。上の実行結果では、「4」が2回現れています。代わりに、32がなくなっています。「4」のことを除くと50までは正しいように思われますが、51以上は並び方がおかしいです。これは「対象配列[y]」と「対象配列[y-1]」の比較入れ替えで、複数のスレッドが、yの値が同じところを操作しようとしたためです。
これを防ぐには、この範囲をクリティカルエリアとしてマークします。クリティカルとしてマークされた範囲は、必ず1つのスレッドでしか実行されません。バブルソートの場合、アルゴリズムの大部分が、クリティカルとしてマークされてしまうことになります。
void BubbleSortParallel(int* 対象配列, const int 配列数)
{
#pragma omp parallel for ordered
for (int 決定済み数 = 0; 決定済み数 < 配列数; ++決定済み数) {
for (int y = 配列数 - 1; y > 決定済み数; --y) {
#pragma omp critical
if (対象配列[y] < 対象配列[y - 1]) {
int tmp = 対象配列[y];
対象配列[y] = 対象配列[y - 1];
対象配列[y - 1] = tmp;
}
}
}
}
ソート前 63 2 3 100 5 81 16 82 86 53 11 1 13 14 34 35 58 74 19 40 8 22 23 24 73 18 27 28 9 30 55 56 33 76 99 12 98 4 26 90 68 37 57 44 67 95 47 48 32 69 51 52 78 54 80 38 94 17 59 60 72 49 43 64 65 66 45 41 50 70 88 7 25 84 77 15 62 10 36 92 6 21 83 61 85 29 87 42 89 96 91 20 93 79 46 31 97 71 39 75 ---------- ソート後 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100
