それでもおかしい並列qsort

segvで落ちる並列qsortについて、どこからともなく神の声が聞こえた。

main()のpはstackにおかないでmallocする、ret1とret2を0以外で初期化する、if (...||...) free(pnext); の行を削除、myqsort()の最後でちゃんとreturn NULL;する、とかすれば多分動きそうです。

なるほど。特にret1、ret2を0以外で初期化していなかったのがまずそう。スレッド生成が発生しないケースで、たまたまretの値が0となってjoinしようとすれば落ちておかしくない。

あと、mainのスタックにデータを置いていたのは、何となくmain関数は特別なグローバルな存在という意識があったことからくる間違いだ。mainだろうが、なんだろうが、スレッドはすべて対等。ということは、すべてのスレッドからアクセスするデータはヒープにおかないとダメだ。

以下のように書き換えた。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <pthread.h>
#include "psort.h"

#define SWAP(type, x, y) do {type tmp = x; x = y; y = tmp;}while(0)

void *myqsort(void *p) {

  pthread_t thread1, thread2; 
  int ret1, ret2;
  ret1 = ret2 = 1;

  partition *part, *pnext1, *pnext2;

  part = (partition *)p;

  int left = part->begin;
  int right = part->end;

  if (left >= right) {
    pthread_exit(NULL);
  }

  int pivot = part->ary[(left + right) / 2];
  
  while (1) {
    while(part->ary[left] < pivot) left++;
    while(part->ary[right] > pivot) right--;
    if (left >= right) break;

    SWAP(int, part->ary[left], part->ary[right]);
    left++; right--;
  }

  if (left - part->begin >= 2) {
    pnext1 = (partition *)malloc(sizeof(partition));
    pnext1->ary = part->ary;
    pnext1->begin = part->begin;
    pnext1->end = left - 1;
    ret1 = pthread_create(&thread1, NULL, myqsort, (void *)pnext1);
  }
  if (part->end - right >= 2) {
    pnext2 = (partition *)malloc(sizeof(partition));
    pnext2->ary = part->ary;
    pnext2->begin = right + 1;
    pnext2->end = part->end;
    ret2 = pthread_create(&thread2, NULL, myqsort, (void *)pnext2);
  }
  if (ret1 == 0) {
    pthread_join(thread1, NULL);
  }
  if (ret2 == 0) {
    pthread_join(thread2, NULL);
  }
  return NULL;
}

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <string.h>
#include <assert.h>
#include <stdbool.h>
#include "psort.h"

int main (int argc, char *argv[]) {
  int size, i;
  int *x;
  partition *p;

  if (argc < 1) exit(1);
  size = atoi(argv[1]);

  p = (partition *)malloc(sizeof(partition));
  x = (int *)malloc(sizeof(int)*size);

  srand(time(NULL));
  for (i = 0; i < size; i++) {
    x[i] = rand() % size;
  }

  show_ary(x, size);
  p->ary = x;
  p->begin = 0;
  p->end = size - 1;
  myqsort(p);
  //  sleep(1);
  printf("--sort--\n");
  show_ary(x, size);
}

これでsegvで落ちなくなった。いい感じ、いい感じと思ったのもつかの間、少し大きな並びを突っ込むとどうも並びきってない箇所が現れる。要素数300だとキレイに並ぶけど、要素数3万だと、どうもおかしなところがあっちこっちにある。

グラフにしてみた。もとのデータは乱数なので、ややでこぼこしながらも直線的に並ぶのが正解。3万個のときには明らかに一部のデータがおかしい。

しかし、グラフがちょっとおもしろい。これはどういう意味だろうか。sleep(1)で寝かせてみても、ちゃんとソートされない領域が残る。