Added a sketch of a new array sorter implementation.

The interface is already hopefully fixed, I am going to add the missing bits
of implementation soon.

diff --git a/lib/sorter/array.h b/lib/sorter/array.h
new file mode 100644 (file)
index 0000000..73da690
--- /dev/null
+++ b/lib/sorter/array.h
@@ -0,0 +1,180 @@
+/*
+ *     UCW Library -- Optimized Array Sorter
+ *
+ *     (c) 2003--2007 Martin Mares <mj@ucw.cz>
+ *
+ *     This software may be freely distributed and used according to the terms
+ *     of the GNU Lesser General Public License.
+ */
+
+/*
+ *  This is a generator of routines for sorting arrays, very similar to the one
+ *  in lib/arraysort.h.
+ *
+ *  FIXME: <comments>
+ *  FIXME: Note on thread-safety
+ *
+ *  So much for advocacy, there are the parameters (those marked with [*]
+ *  are mandatory):
+ *
+ *  ASORT_PREFIX(x) [*]        add a name prefix (used on all global names
+ *                     defined by the sorter)
+ *  ASORT_KEY_TYPE  [*]        data type of a single array entry key
+ *  ASORT_LT(x,y)      x < y for ASORT_TYPE (default: "x<y")
+ *  ASORT_THRESHOLD    threshold for switching between quicksort and insertsort
+ *  ASORT_PAGE_ALIGNED the array is guaranteed to be aligned to a multiple of CPU_PAGE_SIZE
+ *  ASORT_HASH(x)      FIXME
+ *  ASORT_SWAP         FIXME: probably keep private
+ *
+ *  After including this file, a function
+ *     ASORT_KEY_TYPE *ASORT_PREFIX(sort)(ASORT_KEY_TYPE *array, uns num_elts, ASORT_KEY_TYPE *buf, uns hash_bits)
+ *  is declared and all parameter macros are automatically undef'd.
+ */
+
+#include "lib/sorter/common.h"
+
+#define Q(x) ASORT_PREFIX(x)
+
+typedef ASORT_KEY_TYPE Q(key);
+
+#ifndef ASORT_LT
+#define ASORT_LT(x,y) ((x) < (y))
+#endif
+
+#ifndef ASORT_SWAP
+#define ASORT_SWAP(i,j) do { Q(key) tmp = array[i]; array[i]=array[j]; array[j]=tmp; } while (0)
+#endif
+
+#ifndef ASORT_THRESHOLD
+#define ASORT_THRESHOLD 8              /* Guesswork and experimentation */
+#endif
+
+static void Q(raw_sort)(Q(key) *array, uns array_size)
+{
+  struct stk { int l, r; } stack[8*sizeof(uns)];
+  int l, r, left, right, m;
+  uns sp = 0;
+  Q(key) pivot;
+
+  if (array_size <= 1)
+    return;
+
+  /* QuickSort with optimizations a'la Sedgewick, but stop at ASORT_THRESHOLD */
+
+  left = 0;
+  right = array_size - 1;
+  for(;;)
+    {
+      l = left;
+      r = right;
+      m = (l+r)/2;
+      if (ASORT_LT(array[m], array[l]))
+       ASORT_SWAP(l,m);
+      if (ASORT_LT(array[r], array[m]))
+       {
+         ASORT_SWAP(m,r);
+         if (ASORT_LT(array[m], array[l]))
+           ASORT_SWAP(l,m);
+       }
+      pivot = array[m];
+      do
+       {
+         while (ASORT_LT(array[l], pivot))
+           l++;
+         while (ASORT_LT(pivot, array[r]))
+           r--;
+         if (l < r)
+           {
+             ASORT_SWAP(l,r);
+             l++;
+             r--;
+           }
+         else if (l == r)
+           {
+             l++;
+             r--;
+           }
+       }
+      while (l <= r);
+      if ((r - left) >= ASORT_THRESHOLD && (right - l) >= ASORT_THRESHOLD)
+       {
+         /* Both partitions ok => push the larger one */
+         if ((r - left) > (right - l))
+           {
+             stack[sp].l = left;
+             stack[sp].r = r;
+             left = l;
+           }
+         else
+           {
+             stack[sp].l = l;
+             stack[sp].r = right;
+             right = r;
+           }
+         sp++;
+       }
+      else if ((r - left) >= ASORT_THRESHOLD)
+       {
+         /* Left partition OK, right undersize */
+         right = r;
+       }
+      else if ((right - l) >= ASORT_THRESHOLD)
+       {
+         /* Right partition OK, left undersize */
+         left = l;
+       }
+      else
+       {
+         /* Both partitions undersize => pop */
+         if (!sp)
+           break;
+         sp--;
+         left = stack[sp].l;
+         right = stack[sp].r;
+       }
+    }
+
+  /*
+   * We have a partially sorted array, finish by insertsort. Inspired
+   * by qsort() in GNU libc.
+   */
+
+  /* Find minimal element which will serve as a barrier */
+  r = MIN(array_size, ASORT_THRESHOLD);
+  m = 0;
+  for (l=1; l<r; l++)
+    if (ASORT_LT(array[l], array[m]))
+      m = l;
+  ASORT_SWAP(0,m);
+
+  /* Insertion sort */
+  for (m=1; m<(int)array_size; m++)
+    {
+      l=m;
+      while (ASORT_LT(array[m], array[l-1]))
+       l--;
+      while (l < m)
+       {
+         ASORT_SWAP(l,m);
+         l++;
+       }
+    }
+}
+
+static Q(key) *Q(sort)(Q(key) *array, uns num_elts, Q(key) *buffer, uns hash_bits)
+{
+  (void) buffer;
+  (void) hash_bits;
+  Q(raw_sort)(array, num_elts);
+  return array;
+}
+
+/* FIXME */
+#undef ASORT_PREFIX
+#undef ASORT_KEY_TYPE
+#undef ASORT_LT
+#undef ASORT_SWAP
+#undef ASORT_THRESHOLD
+#undef ASORT_PAGE_ALIGNED
+#undef ASORT_HASH
+#undef Q

diff --git a/lib/sorter/s-fixint.h b/lib/sorter/s-fixint.h
index 67fd17d06a7568dcbf85d5bcdffc8a81d861f114..6eac13bb49780fc5c78c5d8daa4796d94fcc7ee6 100644 (file)
--- a/lib/sorter/s-fixint.h
+++ b/lib/sorter/s-fixint.h
@@ -11,10 +11,9 @@
 
 #define ASORT_PREFIX(x) SORT_PREFIX(array_##x)
 #define ASORT_KEY_TYPE P(key)
-#define ASORT_ELT(i) ary[i]
 #define ASORT_LT(x,y) (P(compare)(&(x), &(y)) < 0)
-#define ASORT_EXTRA_ARGS , P(key) *ary
-#include "lib/arraysort.h"
+#define ASORT_PAGE_ALIGNED
+#include "lib/sorter/array.h"
 
 /*
  *  This is a more efficient implementation of the internal sorter,
@@ -31,7 +30,7 @@ static size_t P(internal_workspace)(void)
 #ifdef SORT_UNIFY
   workspace = sizeof(P(key) *);
 #endif
-#if 0          // FIXME: Workspace for radix-sort if needed
+#ifdef SORT_HASH_BITS  // FIXME: Another switch?
   workspace = MAX(workspace, sizeof(P(key)));
 #endif
   return workspace;
@@ -54,7 +53,7 @@ static int P(internal)(struct sort_context *ctx, struct sort_bucket *bin, struct
   P(key) *buf = ctx->big_buf;
   uns maxkeys = P(internal_num_keys)(ctx);
 
-  SORT_XTRACE(4, "s-fixint: Reading (maxkeys=%u)", maxkeys);
+  SORT_XTRACE(4, "s-fixint: Reading (maxkeys=%u, hash_bits=%d)", maxkeys, bin->hash_bits);
   uns n = 0;
   while (n < maxkeys && P(read_key)(in, &buf[n]))
     n++;
@@ -68,7 +67,13 @@ static int P(internal)(struct sort_context *ctx, struct sort_bucket *bin, struct
        stk_fsize(n * P(internal_workspace)()));
   timestamp_t timer;
   init_timer(&timer);
-  P(array_sort)(n, buf);
+  buf = P(array_sort)(buf, n,
+#ifdef SORT_HASH_BITS
+    workspace, bin->hash_bits
+#else
+    NULL, 0
+#endif
+    );
   ctx->total_int_time += get_timer(&timer);
 
   SORT_XTRACE(4, "s-fixint: Writing");

diff --git a/lib/sorter/s-internal.h b/lib/sorter/s-internal.h
index e921921e95241901544d48e88de817ae2befa55b..92a3ad64a8191409e5af1691f88b3a57b626b5d1 100644 (file)
--- a/lib/sorter/s-internal.h
+++ b/lib/sorter/s-internal.h
@@ -16,10 +16,9 @@ typedef struct {
 
 #define ASORT_PREFIX(x) SORT_PREFIX(array_##x)
 #define ASORT_KEY_TYPE P(internal_item_t)
-#define ASORT_ELT(i) ary[i]
 #define ASORT_LT(x,y) (P(compare)((x).key, (y).key) < 0)
-#define ASORT_EXTRA_ARGS , P(internal_item_t) *ary
-#include "lib/arraysort.h"
+#define ASORT_PAGE_ALIGNED
+#include "lib/sorter/array.h"
 
 /*
  *  The big_buf has the following layout:
@@ -66,8 +65,8 @@ static inline size_t P(internal_workspace)(P(key) *key UNUSED)
 #ifdef SORT_UNIFY_WORKSPACE
   ws += SORT_UNIFY_WORKSPACE(*key);
 #endif
-#if 0                                          /* FIXME: Shadow copy if radix-sorting */
-  ws = MAX(ws, sizeof(P(key) *));
+#ifdef SORT_HASH_BITS
+  ws = MAX(ws, sizeof(P(internal_item_t)));
 #endif
   return ws;
 }
@@ -146,7 +145,13 @@ static int P(internal)(struct sort_context *ctx, struct sort_bucket *bin, struct
        stk_fsize((byte*)ctx->big_buf + bufsize - end));
   timestamp_t timer;
   init_timer(&timer);
-  P(array_sort)(count, item_array);
+  item_array = P(array_sort)(item_array, count,
+#ifdef SORT_HASH_BITS
+    workspace, bin->hash_bits
+#else
+    NULL, 0
+#endif
+    );
   ctx->total_int_time += get_timer(&timer);
 
   SORT_XTRACE(4, "s-internal: Writing");