From 0ce1209ffd78ede3a5d353393e281cc6a7995cd3 Mon Sep 17 00:00:00 2001
From: Martin Mares <mj@ucw.cz>
Date: Thu, 6 Sep 2007 17:16:10 +0200
Subject: [PATCH] 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.
---
 lib/sorter/array.h      | 180 ++++++++++++++++++++++++++++++++++++++++
 lib/sorter/s-fixint.h   |  17 ++--
 lib/sorter/s-internal.h |  17 ++--
 3 files changed, 202 insertions(+), 12 deletions(-)
 create mode 100644 lib/sorter/array.h

diff --git a/lib/sorter/array.h b/lib/sorter/array.h
new file mode 100644
index 00000000..73da6908
--- /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 67fd17d0..6eac13bb 100644
--- 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 e921921e..92a3ad64 100644
--- 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");
-- 
2.39.2