lib/sorter/array.c

   1 /*
   2  *      UCW Library -- Optimized Array Sorter
   3  *
   4  *      (c) 2003--2007 Martin Mares <mj@ucw.cz>
   5  *
   6  *      This software may be freely distributed and used according to the terms
   7  *      of the GNU Lesser General Public License.
   8  */
   9
  10 #undef LOCAL_DEBUG
  11
  12 #include "lib/lib.h"
  13 #include "lib/sorter/common.h"
  14
  15 #include <string.h>
  16 #include <alloca.h>
  17
  18 #define ASORT_MIN_RADIX 5000            // FIXME: var?
  19 #define ASORT_MIN_SHIFT 2
  20
  21 static void
  22 asort_radix(struct asort_context *ctx, void *array, void *buffer, uns num_elts, uns hash_bits, uns swapped_output)
  23 {
  24   uns buckets = (1 << ctx->radix_bits);
  25   uns shift = (hash_bits > ctx->radix_bits) ? (hash_bits - ctx->radix_bits) : 0;
  26   uns cnt[buckets];
  27
  28 #if 0
  29   static int reported[64];
  30   if (!reported[hash_bits]++)
  31 #endif
  32   DBG(">>> n=%d h=%d s=%d sw=%d", num_elts, hash_bits, shift, swapped_output);
  33
  34   bzero(cnt, sizeof(cnt));
  35   ctx->radix_count(array, num_elts, cnt, shift);
  36
  37   uns pos = 0;
  38   for (uns i=0; i<buckets; i++)
  39     {
  40       uns j = cnt[i];
  41       cnt[i] = pos;
  42       pos += j;
  43     }
  44   ASSERT(pos == num_elts);
  45
  46   ctx->radix_split(array, buffer, num_elts, cnt, shift);
  47   pos = 0;
  48   for (uns i=0; i<buckets; i++)
  49     {
  50       uns n = cnt[i] - pos;
  51       if (n < ASORT_MIN_RADIX || shift < ASORT_MIN_SHIFT)
  52         {
  53           ctx->quicksort(buffer, n);
  54           if (!swapped_output)
  55             memcpy(array, buffer, n * ctx->elt_size);
  56         }
  57       else
  58         asort_radix(ctx, buffer, array, n, shift, !swapped_output);
  59       array += n * ctx->elt_size;
  60       buffer += n * ctx->elt_size;
  61       pos = cnt[i];
  62     }
  63 }
  64
  65 #ifdef CONFIG_UCW_THREADS
  66
  67 #include "lib/threads.h"
  68 #include "lib/workqueue.h"
  69 #include "lib/eltpool.h"
  70
  71 static uns asort_threads_use_count;
  72 static uns asort_threads_ready;
  73 static struct worker_pool asort_thread_pool;
  74
  75 void
  76 asort_start_threads(uns run)
  77 {
  78   ucwlib_lock();
  79   asort_threads_use_count++;
  80   if (run && !asort_threads_ready)
  81     {
  82       SORT_XTRACE(2, "Initializing thread pool (%d threads)", sorter_threads);
  83       asort_thread_pool.num_threads = sorter_threads;
  84       worker_pool_init(&asort_thread_pool);
  85       asort_threads_ready = 1;
  86     }
  87   ucwlib_unlock();
  88 }
  89
  90 void
  91 asort_stop_threads(void)
  92 {
  93   ucwlib_lock();
  94   if (!--asort_threads_use_count && asort_threads_ready)
  95     {
  96       SORT_XTRACE(2, "Shutting down thread pool");
  97       worker_pool_cleanup(&asort_thread_pool);
  98       asort_threads_ready = 0;
  99     }
 100   ucwlib_unlock();
 101 }
 102
 103 struct qs_work {
 104   struct work w;
 105   struct asort_context *ctx;
 106   void *array;
 107   uns num_elts;
 108   int left, right;
 109 #define LR_UNDEF -100
 110 };
 111
 112 static void
 113 qs_handle_work(struct worker_thread *thr UNUSED, struct work *ww)
 114 {
 115   struct qs_work *w = (struct qs_work *) ww;
 116
 117   DBG("Thread %d: got %d elts", thr->id, w->num_elts);
 118   if (w->num_elts * w->ctx->elt_size < sorter_thread_threshold)
 119     {
 120       w->ctx->quicksort(w->array, w->num_elts);
 121       w->left = w->right = LR_UNDEF;
 122     }
 123   else
 124     w->ctx->quicksplit(w->array, w->num_elts, &w->left, &w->right);
 125   DBG("Thread %d: returning l=%d r=%d", thr->id, w->left, w->right);
 126 }
 127
 128 static struct qs_work *
 129 qs_alloc_work(struct asort_context *ctx)
 130 {
 131   struct qs_work *w = ep_alloc(ctx->eltpool);
 132   w->w.priority = 0;
 133   w->w.go = qs_handle_work;
 134   w->ctx = ctx;
 135   return w;
 136 }
 137
 138 static void
 139 threaded_quicksort(struct asort_context *ctx)
 140 {
 141   struct work_queue q;
 142   struct qs_work *v, *w;
 143
 144   asort_start_threads(1);
 145   work_queue_init(&asort_thread_pool, &q);
 146   ctx->eltpool = ep_new(sizeof(struct qs_work), 1000);
 147
 148   w = qs_alloc_work(ctx);
 149   w->array = ctx->array;
 150   w->num_elts = ctx->num_elts;
 151   work_submit(&q, &w->w);
 152
 153   while (v = (struct qs_work *) work_wait(&q))
 154     {
 155       if (v->left != LR_UNDEF)
 156         {
 157           if (v->right > 0)
 158             {
 159               w = qs_alloc_work(ctx);
 160               w->array = v->array;
 161               w->num_elts = v->right + 1;
 162               w->w.priority = v->w.priority + 1;
 163               work_submit(&q, &w->w);
 164             }
 165           if (v->left < (int)v->num_elts - 1)
 166             {
 167               w = qs_alloc_work(ctx);
 168               w->array = v->array + v->left * ctx->elt_size;
 169               w->num_elts = v->num_elts - v->left;
 170               w->w.priority = v->w.priority + 1;
 171               work_submit(&q, &w->w);
 172             }
 173         }
 174       ep_free(ctx->eltpool, v);
 175     }
 176
 177   ep_delete(ctx->eltpool);
 178   work_queue_cleanup(&q);
 179   asort_stop_threads();
 180 }
 181
 182 struct rs_work {
 183   struct work w;
 184   struct asort_context *ctx;
 185   void *in, *out;
 186   uns num_elts;
 187   uns shift;
 188   uns swap_output;
 189   uns cnt[0];
 190 };
 191
 192 static void
 193 rs_count(struct worker_thread *thr UNUSED, struct work *ww)
 194 {
 195   struct rs_work *w = (struct rs_work *) ww;
 196
 197   DBG("Thread %d: Counting %d items, shift=%d", thr->id, w->num_elts, w->shift);
 198   w->ctx->radix_count(w->in, w->num_elts, w->cnt, w->shift);
 199   DBG("Thread %d: Counting done", thr->id);
 200 }
 201
 202 static void
 203 rs_split(struct worker_thread *thr UNUSED, struct work *ww)
 204 {
 205   struct rs_work *w = (struct rs_work *) ww;
 206
 207   DBG("Thread %d: Splitting %d items, shift=%d", thr->id, w->num_elts, w->shift);
 208   w->ctx->radix_split(w->in, w->out, w->num_elts, w->cnt, w->shift);
 209   DBG("Thread %d: Splitting done", thr->id);
 210 }
 211
 212 static void
 213 rs_finish(struct worker_thread *thr UNUSED, struct work *ww)
 214 {
 215   struct rs_work *w = (struct rs_work *) ww;
 216
 217   DBG("Thread %d: Finishing %d items, shift=%d", thr->id, w->num_elts, w->shift);
 218   if (w->shift < ASORT_MIN_SHIFT || w->num_elts < ASORT_MIN_RADIX)
 219     {
 220       w->ctx->quicksort(w->out, w->num_elts);
 221       if (!w->swap_output)
 222         memcpy(w->in, w->out, w->num_elts * w->ctx->elt_size);
 223     }
 224   else
 225     asort_radix(w->ctx, w->out, w->in, w->num_elts, w->shift, !w->swap_output);
 226   DBG("Thread %d: Finishing done", thr->id);
 227 }
 228
 229 static void
 230 rs_radix(struct asort_context *ctx, void *array, void *buffer, uns num_elts, uns hash_bits, uns swapped_output)
 231 {
 232   uns buckets = (1 << ctx->radix_bits);
 233   uns shift = (hash_bits > ctx->radix_bits) ? (hash_bits - ctx->radix_bits) : 0;
 234   uns cnt[buckets];
 235   uns blksize = num_elts / sorter_threads;
 236   DBG(">>> n=%d h=%d s=%d blk=%d sw=%d", num_elts, hash_bits, shift, blksize, swapped_output);
 237
 238   // Start parallel counting
 239   void *iptr = array;
 240   for (uns i=0; i<sorter_threads; i++)
 241     {
 242       struct rs_work *w = ctx->rs_works[i];
 243       w->w.priority = 0;
 244       w->w.go = rs_count;
 245       w->ctx = ctx;
 246       w->in = iptr;
 247       w->out = ctx->buffer;
 248       w->num_elts = blksize;
 249       if (i == sorter_threads-1)
 250         w->num_elts += num_elts % sorter_threads;
 251       w->shift = shift;
 252       iptr += w->num_elts * ctx->elt_size;
 253       work_submit(ctx->rs_work_queue, &w->w);
 254     }
 255
 256   // Get bucket sizes from the counts
 257   bzero(cnt, sizeof(cnt));
 258   for (uns i=0; i<sorter_threads; i++)
 259     {
 260       struct rs_work *w = (struct rs_work *) work_wait(ctx->rs_work_queue);
 261       ASSERT(w);
 262       for (uns j=0; j<buckets; j++)
 263         cnt[j] += w->cnt[j];
 264     }
 265
 266   // Calculate bucket starts
 267   uns pos = 0;
 268   for (uns i=0; i<buckets; i++)
 269     {
 270       uns j = cnt[i];
 271       cnt[i] = pos;
 272       pos += j;
 273     }
 274   ASSERT(pos == num_elts);
 275
 276   // Start parallel splitting
 277   for (uns i=0; i<sorter_threads; i++)
 278     {
 279       struct rs_work *w = ctx->rs_works[i];
 280       w->w.go = rs_split;
 281       for (uns j=0; j<buckets; j++)
 282         {
 283           uns k = w->cnt[j];
 284           w->cnt[j] = cnt[j];
 285           cnt[j] += k;
 286         }
 287       work_submit(ctx->rs_work_queue, &w->w);
 288     }
 289   ASSERT(cnt[buckets-1] == num_elts);
 290
 291   // Wait for splits to finish
 292   while (work_wait(ctx->rs_work_queue))
 293     ;
 294
 295   // Recurse on buckets
 296   pos = 0;
 297   for (uns i=0; i<buckets; i++)
 298     {
 299       uns n = cnt[i] - pos;
 300       if (n < sorter_thread_threshold)
 301         {
 302           struct rs_work *w = ep_alloc(ctx->eltpool);
 303           w->w.priority = 0;
 304           w->w.go = rs_finish;
 305           w->ctx = ctx;
 306           w->in = array;
 307           w->out = buffer;
 308           w->num_elts = n;
 309           w->shift = shift;
 310           w->swap_output = swapped_output;
 311           clist_add_tail(&ctx->rs_bits, &w->w.n);
 312           DBG("Scheduling block %d+%d", pos, n);
 313         }
 314       else
 315         rs_radix(ctx, buffer, array, n, shift, !swapped_output);
 316       pos = cnt[i];
 317       array += n * ctx->elt_size;
 318       buffer += n * ctx->elt_size;
 319     }
 320 }
 321
 322 static void
 323 threaded_radixsort(struct asort_context *ctx)
 324 {
 325   struct work_queue q;
 326
 327   asort_start_threads(1);
 328   work_queue_init(&asort_thread_pool, &q);
 329
 330   // Prepare work structures for counting and splitting.
 331   // We use big_alloc(), because we want to avoid cacheline aliasing between threads.
 332   ctx->rs_work_queue = &q;
 333   ctx->rs_works = alloca(sizeof(struct rs_work *) * sorter_threads);
 334   for (uns i=0; i<sorter_threads; i++)
 335     ctx->rs_works[i] = big_alloc(sizeof(struct rs_work) + sizeof(uns) * (1 << ctx->radix_bits));
 336
 337   // Prepare work structures for all remaining small bits which will be sorted later.
 338   clist_init(&ctx->rs_bits);
 339   ctx->eltpool = ep_new(sizeof(struct rs_work), 1000);
 340
 341   // Do the big splitting
 342   // FIXME: Set the swap bit carefully.
 343   rs_radix(ctx, ctx->array, ctx->buffer, ctx->num_elts, ctx->hash_bits, 0);
 344   for (uns i=0; i<sorter_threads; i++)
 345     big_free(ctx->rs_works[i], sizeof(struct rs_work) + sizeof(uns) * (1 << ctx->radix_bits));
 346
 347   // Finish the small blocks
 348   struct rs_work *w, *tmp;
 349   CLIST_WALK_DELSAFE(w, ctx->rs_bits, tmp)
 350     work_submit(&q, &w->w);
 351   while (work_wait(&q))
 352     ;
 353
 354   ep_delete(ctx->eltpool);
 355   work_queue_cleanup(&q);
 356   asort_stop_threads();
 357 }
 358
 359 #else
 360
 361 void asort_start_threads(uns run UNUSED) { }
 362 void asort_stop_threads(void) { }
 363
 364 #endif
 365
 366 void
 367 asort_run(struct asort_context *ctx)
 368 {
 369   SORT_XTRACE(10, "Array-sorting %d items per %d bytes, hash_bits=%d", ctx->num_elts, ctx->elt_size, ctx->hash_bits);
 370   uns allow_threads UNUSED = (sorter_threads > 1 &&
 371                               ctx->num_elts * ctx->elt_size >= sorter_thread_threshold &&
 372                               !(sorter_debug & SORT_DEBUG_ASORT_NO_THREADS));
 373
 374   if (ctx->num_elts < ASORT_MIN_RADIX ||
 375       ctx->hash_bits <= ASORT_MIN_SHIFT ||
 376       !ctx->radix_split ||
 377       (sorter_debug & SORT_DEBUG_ASORT_NO_RADIX))
 378     {
 379 #ifdef CONFIG_UCW_THREADS
 380       if (allow_threads)
 381         {
 382           SORT_XTRACE(12, "Decided to use parallel quicksort");
 383           threaded_quicksort(ctx);
 384           return;
 385         }
 386 #endif
 387       SORT_XTRACE(12, "Decided to use sequential quicksort");
 388       ctx->quicksort(ctx->array, ctx->num_elts);
 389     }
 390   else
 391     {
 392 #ifdef CONFIG_UCW_THREADS
 393       if (allow_threads)
 394         {
 395           SORT_XTRACE(12, "Decided to use parallel radix-sort");
 396           threaded_radixsort(ctx);
 397           return;
 398         }
 399 #endif
 400       SORT_XTRACE(12, "Decided to use sequential radix-sort");
 401       // FIXME: select dest buffer
 402       asort_radix(ctx, ctx->array, ctx->buffer, ctx->num_elts, ctx->hash_bits, 0);
 403     }
 404
 405   SORT_XTRACE(11, "Array-sort finished");
 406 }