Changeset 2287 for branches/parallel

Timestamp:

09/05/11 18:30:08 (13 years ago)

Author:

KohjiNakamura

Message:

modernQSort bug fix

File:

• branches/parallel/src/TableTraverse.cpp (modified) (5 diffs)

branches/parallel/src/TableTraverse.cpp

@@ -73 +73 @@
 #endif
 
+#ifdef HAVE_QSORT_R
+
+static int compWithComparator(void *arg, const void *a, const void *b)
+{
+        return ((asap::Comparator *)arg)->compare(a, b);
+}
+
+#else
+
 static inline void swap_(char *a, char *b, size_t size)
 {
@@ -91 +100 @@
 }
 
-static void sort_(char *left, char *right, const size_t size,
+// returning NULL means, already sorted.
+static inline char *quickFindPivot(char *const left, char *const right,
+                                   const size_t size, asap::Comparator *comp)
+{
+  char *const m = &left[(((right - left) / size) / 2 ) * size];
+  char *result = NULL;
+  if (left == m) { // This means there are at most 2 elements.
+    int diff = (left == right) ? 0 : comp->compare(left, right);
+    if (diff > 0) {
+      swap_(left, right, size);
+    }
+    if (diff != 0) {
+      return right;
+    }
+    return result;
+  }
+
+  int diff = comp->compare(left, m);
+  if (diff > 0) {
+    swap_(left, m, size);
+  }
+  int diff2 = comp->compare(m, right);
+  if (diff == 0 && diff2 == 0) {
+    return result;
+  }
+  result = m;
+  if (diff2 > 0) {
+    swap_(right, m, size);
+    int diff3 = comp->compare(left, m);
+    if (diff3 > 0) {
+      swap_(left, m, size);
+    } else if (diff3 == 0) {
+      result = right;
+    }
+  } else if (diff == 0) {
+    result = right;
+  }
+  return result;
+}
+
+static inline char *findPivot(char *const start, char *const end,
+                              size_t size, asap::Comparator *comp)
+{
+  char *result = quickFindPivot(start, end, size, comp);
+  if (result || &start[2*size] >= end) {
+    return result;
+  }
+
+  for (char *pivot = start + size; pivot <= end; pivot += size) {
+    int diff = comp->compare(start, pivot);
+    if (diff < 0) {
+      return pivot;
+    } else if (diff > 0) {
+      swap_(start, pivot, size);
+      return pivot;
+    }
+  }
+  return NULL; // all values are same.
+}
+
+static void sort_(char *const left, char *const right, const size_t size,
                   asap::Comparator *comp, int level)
 {
-  const size_t len = (right - left) / size;
-  if (len < 2) {
+  char *const m = findPivot(left, right, size, comp);
+  if (m == NULL) {
     return;
   }
-  size_t mid = len >> 1;
   char *l = left;
-  char *r = right - size;
+  char *r = right;
   char pivot[size]  __attribute__ ((aligned (16)));
-  cpy(pivot, &left[mid*size], size);
+  cpy(pivot, m, size);
   while (true) {
     while (l < r && comp->compare(l, pivot) < 0) {
       l += size;
     }
-    while (l < r && comp->compare(r, pivot) > 0) {
+    while (l < r && comp->compare(pivot, r) <= 0) {
       r -= size;
-        }
+    }
     if (l >= r) {
       break;
@@ -117 +185 @@
     r -= size;
   }
-  sort_(left, l, size, comp, level + 1);
+  if (l == r && comp->compare(l, pivot) < 0) {
+    l += size;
+  }
+  sort_(left, l - size, size, comp, level + 1);
   sort_(l, right, size, comp, level + 1);
 }
+
+#endif
 
 static inline void modernQSort(void *base, size_t elements, size_t size,
                                asap::Comparator *comp)
 {
-  sort_((char *)base, &((char *)base)[elements * size], size, comp, 0);
+  if (elements > 1) {
+#ifdef HAVE_QSORT_R
+    qsort_r(base, elements, size, comp, compWithComparator);
+#else
+    sort_((char *)base, &((char *)base)[(elements - 1) * size], size, comp, 0);
+#endif
+  }
 }
 
@@ -224 +303 @@
 }
 
-#if 0
+#if 0 && TIMING
+#include <assert.h>
+
+#define elementsof(x) (sizeof(x)/sizeof(*x))
+
 using namespace asap ;
 
-class MyComp: public Comparator {
+class IntComp: public Comparator {
 public:
   virtual int compare(const void *a, const void *b) {
@@ -234 +317 @@
 };
 
+static int compare(const void *a, const void *b) {
+  return (*(int*)a) - (*(int*)b);
+}
+
 int main()
 {
+  IntComp myComp;
   srand((int)currenttime());
-  int x[100];
-  printf("init\n");
-  for (int i = 0; i < 100; i++) {
-    x[i] = rand() % 100;
-  }
-  MyComp myComp;
-  printf("sorting\n");
+  const size_t N = 1024*1024*100;
+  int *x = new int[N];
+  int xx[] = {
+    5, 3,0,  1, 2,4, 4, 4
+  };
+#if 0
+  {
+    int x[] = {1};
+    char *p = findPivot((char *)x, (char *)&x[0], sizeof(x[0]), &myComp);
+    assert(p == NULL);
+  }
+  {
+    int x[] = {1, 1};
+    char *p = findPivot((char *)x, (char *)&x[1], sizeof(x[0]), &myComp);
+    assert(p == NULL);
+  }
+  {
+    int x[] = {2, 1};
+    char *p = findPivot((char *)x, (char *)&x[1], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1] && x[0] == 1 && x[1] == 2);
+  }
+  {
+    int x[] = {1, 2};
+    char *p = findPivot((char *)x, (char *)&x[1], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1] && x[0] == 1 && x[1] == 2);
+  }
+  {
+    int x[] = {1, 1, 1};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert(p == NULL);
+  }
+  {
+    int x[] = {2, 1, 1};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[2]);
+    assert(x[0] == 1 && x[1] == 1 && x[2] == 2);
+  }
+  {
+    int x[] = {1, 2, 1};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[2]);
+    assert(x[0] == 1 && x[1] == 1 && x[2] == 2);
+  }
+  {
+    int x[] = {1, 1, 2};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[2]);
+    assert(x[0] == 1 && x[1] == 1 && x[2] == 2);
+  }
+  {
+    int x[] = {2, 2, 1};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 2);
+  }
+  {
+    int x[] = {2, 1, 2};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 2);
+  }
+  {
+    int x[] = {1, 2, 2};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 2);
+  }
+  {
+    int x[] = {1, 2, 3};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 3);
+  }
+  {
+    int x[] = {2, 1, 3};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 3);
+  }
+  {
+    int x[] = {1, 3, 2};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 3);
+  }
+  {
+    int x[] = {3, 1, 2};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 3);
+  }
+  {
+    int x[] = {3, 2, 1};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 3);
+  }
+  {
+    int x[] = {2, 3, 1};
+    char *p = findPivot((char *)x, (char *)&x[2], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[1]);
+    assert(x[0] == 1 && x[1] == 2 && x[2] == 3);
+  }
+  {
+    int x[] = {36, 39, 42, 41, 36};
+    char *p = findPivot((char *)x, (char *)&x[4], sizeof(x[0]), &myComp);
+    assert((int*)p == &x[4]);
+    assert(x[4] == 42);
+  }
+
+  //assert(("assert enabled: OK", false));
+  modernQSort(x, 8, sizeof(x[0]), &myComp);
+  for (int i = 0; i < 8; i++) {
+    fprintf(stderr, "%d\n", x[i]);
+  }
+#endif
+  fprintf(stderr, "init\n");
+  for (int i = 0; i < N; i++) {
+    x[i] = rand();
+  }
+  fprintf(stderr, "sorting\n");
   modernQSort(x, 1, sizeof(x[0]), &myComp);
   modernQSort(x, 2, sizeof(x[0]), &myComp);
   modernQSort(x, 3, sizeof(x[0]), &myComp);
-  modernQSort(x, 100, sizeof(x[0]), &myComp);
-  for (int i = 0; i < 100; i++) {
+  if (! getenv("QSORT")) {
+    double _s = currenttime();
+    modernQSort(x, N, sizeof(x[0]), &myComp);
+    double _e = currenttime();
+    fprintf(stderr, "modernSorted %.3f\n", _e - _s);
+  } else {
+    double _s = currenttime();
+    qsort(x, N, sizeof(x[0]), compare);
+    double _e = currenttime();
+    fprintf(stderr, "qsorted %.3f\n", _e - _s);
+  }
+  fprintf(stderr, "outputting\n");
+  for (int i = 0; i < N; i++) {
     printf("%d\n", x[i]);
   }
-}
-#endif
+  delete[] x;
+}
+#endif