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_SHIFT 2
  19
  20 #define ASORT_TRACE(x...) ASORT_XTRACE(1, x)
  21 #define ASORT_XTRACE(level, x...) do { if (sorter_trace_array >= level) msg(L_DEBUG, x); } while(0)
  22
  23 static void
  24 asort_radix(struct asort_context *ctx, void *array, void *buffer, uns num_elts, uns hash_bits, uns swapped_output)
  25 {
  26   // swap_output == 0 if result should be returned in `array', otherwise in `buffer'
  27   uns buckets = (1 << ctx->radix_bits);
  28   uns shift = (hash_bits > ctx->radix_bits) ? (hash_bits - ctx->radix_bits) : 0;
  29   uns cnt[buckets];
  30
  31 #if 0
  32   static int reported[64];
  33   if (!reported[hash_bits]++)
  34 #endif
  35   DBG(">>> n=%d h=%d s=%d sw=%d", num_elts, hash_bits, shift, swapped_output);
  36
  37   bzero(cnt, sizeof(cnt));
  38   ctx->radix_count(array, num_elts, cnt, shift);
  39
  40   uns pos = 0;
  41   for (uns i=0; i<buckets; i++)
  42     {
  43       uns j = cnt[i];
  44       cnt[i] = pos;
  45       pos += j;
  46     }
  47   ASSERT(pos == num_elts);
  48
  49   ctx->radix_split(array, buffer, num_elts, cnt, shift);
  50   pos = 0;
  51   for (uns i=0; i<buckets; i++)
  52     {
  53       uns n = cnt[i] - pos;
  54       if (n * ctx->elt_size < sorter_radix_threshold || shift < ASORT_MIN_SHIFT)
  55         {
  56           ctx->quicksort(buffer, n);
  57           if (!swapped_output)
  58             memcpy(array, buffer, n * ctx->elt_size);
  59         }
  60       else
  61         asort_radix(ctx, buffer, array, n, shift, !swapped_output);
  62       array += n * ctx->elt_size;
  63       buffer += n * ctx->elt_size;
  64       pos = cnt[i];
  65     }
  66 }
  67
  68 #ifdef CONFIG_UCW_THREADS
  69
  70 #include "lib/threads.h"
  71 #include "lib/workqueue.h"
  72 #include "lib/eltpool.h"
  73
  74 static uns asort_threads_use_count;
  75 static uns asort_threads_ready;
  76 static struct worker_pool asort_thread_pool;
  77
  78 static uns
  79 rs_estimate_stack(void)
  80 {
  81   // Stack space needed by the recursive radix-sorter
  82   uns ctrsize = sizeof(uns) * (1 << CONFIG_UCW_RADIX_SORTER_BITS);
  83   uns maxdepth = (64 / CONFIG_UCW_RADIX_SORTER_BITS) + 1;
  84   return ctrsize * maxdepth;
  85 }
  86
  87 void
  88 asort_start_threads(uns run)
  89 {
  90   ucwlib_lock();
  91   asort_threads_use_count++;
  92   if (run && !asort_threads_ready)
  93     {
  94       // XXX: If somebody overrides the radix-sorter parameters to insane values,
  95       // he also should override the stack size to insane values.
  96       asort_thread_pool.stack_size = default_thread_stack_size + rs_estimate_stack();
  97       asort_thread_pool.num_threads = sorter_threads;
  98       ASORT_TRACE("Initializing thread pool (%d threads, %dK stack)", sorter_threads, asort_thread_pool.stack_size >> 10);
  99       worker_pool_init(&asort_thread_pool);
 100       asort_threads_ready = 1;
 101     }
 102   ucwlib_unlock();
 103 }
 104
 105 void
 106 asort_stop_threads(void)
 107 {
 108   ucwlib_lock();
 109   if (!--asort_threads_use_count && asort_threads_ready)
 110     {
 111       ASORT_TRACE("Shutting down thread pool");
 112       worker_pool_cleanup(&asort_thread_pool);
 113       asort_threads_ready = 0;
 114     }
 115   ucwlib_unlock();
 116 }
 117
 118 struct qs_work {
 119   struct work w;
 120   struct asort_context *ctx;
 121   void *array;
 122   uns num_elts;
 123   int left, right;
 124 #define LR_UNDEF -100
 125 };
 126
 127 static void
 128 qs_handle_work(struct worker_thread *thr UNUSED, struct work *ww)
 129 {
 130   struct qs_work *w = (struct qs_work *) ww;
 131
 132   DBG("Thread %d: got %d elts", thr->id, w->num_elts);
 133   if (w->num_elts * w->ctx->elt_size < sorter_thread_threshold)
 134     {
 135       w->ctx->quicksort(w->array, w->num_elts);
 136       w->left = w->right = LR_UNDEF;
 137     }
 138   else
 139     w->ctx->quicksplit(w->array, w->num_elts, &w->left, &w->right);
 140   DBG("Thread %d: returning l=%d r=%d", thr->id, w->left, w->right);
 141 }
 142
 143 static struct qs_work *
 144 qs_alloc_work(struct asort_context *ctx)
 145 {
 146   struct qs_work *w = ep_alloc(ctx->eltpool);
 147   w->w.priority = 0;
 148   w->w.go = qs_handle_work;
 149   w->ctx = ctx;
 150   return w;
 151 }
 152
 153 static void
 154 threaded_quicksort(struct asort_context *ctx)
 155 {
 156   struct work_queue q;
 157   struct qs_work *v, *w;
 158
 159   asort_start_threads(1);
 160   work_queue_init(&asort_thread_pool, &q);
 161   ctx->eltpool = ep_new(sizeof(struct qs_work), 1000);
 162
 163   w = qs_alloc_work(ctx);
 164   w->array = ctx->array;
 165   w->num_elts = ctx->num_elts;
 166   work_submit(&q, &w->w);
 167
 168   while (v = (struct qs_work *) work_wait(&q))
 169     {
 170       if (v->left != LR_UNDEF)
 171         {
 172           if (v->right > 0)
 173             {
 174               w = qs_alloc_work(ctx);
 175               w->array = v->array;
 176               w->num_elts = v->right + 1;
 177               w->w.priority = v->w.priority + 1;
 178               work_submit(&q, &w->w);
 179             }
 180           if (v->left < (int)v->num_elts - 1)
 181             {
 182               w = qs_alloc_work(ctx);
 183               w->array = v->array + v->left * ctx->elt_size;
 184               w->num_elts = v->num_elts - v->left;
 185               w->w.priority = v->w.priority + 1;
 186               work_submit(&q, &w->w);
 187             }
 188         }
 189       ep_free(ctx->eltpool, v);
 190     }
 191
 192   ep_delete(ctx->eltpool);
 193   work_queue_cleanup(&q);
 194   asort_stop_threads();
 195 }
 196
 197 struct rs_work {
 198   struct work w;
 199   struct asort_context *ctx;
 200   void *array, *buffer;         // Like asort_radix().
 201   uns num_elts;
 202   uns shift;
 203   uns swap_output;
 204   uns cnt[0];
 205 };
 206
 207 static void
 208 rs_count(struct worker_thread *thr UNUSED, struct work *ww)
 209 {
 210   struct rs_work *w = (struct rs_work *) ww;
 211
 212   DBG("Thread %d: Counting %d items, shift=%d", thr->id, w->num_elts, w->shift);
 213   w->ctx->radix_count(w->array, w->num_elts, w->cnt, w->shift);
 214   DBG("Thread %d: Counting done", thr->id);
 215 }
 216
 217 static void
 218 rs_split(struct worker_thread *thr UNUSED, struct work *ww)
 219 {
 220   struct rs_work *w = (struct rs_work *) ww;
 221
 222   DBG("Thread %d: Splitting %d items, shift=%d", thr->id, w->num_elts, w->shift);
 223   w->ctx->radix_split(w->array, w->buffer, w->num_elts, w->cnt, w->shift);
 224   DBG("Thread %d: Splitting done", thr->id);
 225 }
 226
 227 static void
 228 rs_finish(struct worker_thread *thr UNUSED, struct work *ww)
 229 {
 230   struct rs_work *w = (struct rs_work *) ww;
 231
 232   if (thr)
 233     DBG("Thread %d: Finishing %d items, shift=%d", thr->id, w->num_elts, w->shift);
 234   if (w->shift < ASORT_MIN_SHIFT || w->num_elts * w->ctx->elt_size < sorter_radix_threshold)
 235     {
 236       w->ctx->quicksort(w->array, w->num_elts);
 237       if (w->swap_output)
 238         memcpy(w->buffer, w->array, w->num_elts * w->ctx->elt_size);
 239     }
 240   else
 241     asort_radix(w->ctx, w->array, w->buffer, w->num_elts, w->shift, w->swap_output);
 242   if (thr)
 243     DBG("Thread %d: Finishing done", thr->id);
 244 }
 245
 246 static void
 247 rs_wait_small(struct asort_context *ctx)
 248 {
 249   struct rs_work *w;
 250
 251   while (w = (struct rs_work *) work_wait(ctx->rs_work_queue))
 252     {
 253       DBG("Reaping small chunk of %d items", w->num_elts);
 254       ep_free(ctx->eltpool, w);
 255     }
 256 }
 257
 258 static void
 259 rs_radix(struct asort_context *ctx, void *array, void *buffer, uns num_elts, uns hash_bits, uns swapped_output)
 260 {
 261   uns buckets = (1 << ctx->radix_bits);
 262   uns shift = (hash_bits > ctx->radix_bits) ? (hash_bits - ctx->radix_bits) : 0;
 263   uns cnt[buckets];
 264   uns blksize = num_elts / sorter_threads;
 265   DBG(">>> n=%d h=%d s=%d blk=%d sw=%d", num_elts, hash_bits, shift, blksize, swapped_output);
 266
 267   // If there are any small chunks in progress, wait for them to finish
 268   rs_wait_small(ctx);
 269
 270   // Start parallel counting
 271   void *iptr = array;
 272   for (uns i=0; i<sorter_threads; i++)
 273     {
 274       struct rs_work *w = ctx->rs_works[i];
 275       w->w.priority = 0;
 276       w->w.go = rs_count;
 277       w->ctx = ctx;
 278       w->array = iptr;
 279       w->buffer = buffer;
 280       w->num_elts = blksize;
 281       if (i == sorter_threads-1)
 282         w->num_elts += num_elts % sorter_threads;
 283       w->shift = shift;
 284       iptr += w->num_elts * ctx->elt_size;
 285       bzero(w->cnt, sizeof(uns) * buckets);
 286       work_submit(ctx->rs_work_queue, &w->w);
 287     }
 288
 289   // Get bucket sizes from the counts
 290   bzero(cnt, sizeof(cnt));
 291   for (uns i=0; i<sorter_threads; i++)
 292     {
 293       struct rs_work *w = (struct rs_work *) work_wait(ctx->rs_work_queue);
 294       ASSERT(w);
 295       for (uns j=0; j<buckets; j++)
 296         cnt[j] += w->cnt[j];
 297     }
 298
 299   // Calculate bucket starts
 300   uns pos = 0;
 301   for (uns i=0; i<buckets; i++)
 302     {
 303       uns j = cnt[i];
 304       cnt[i] = pos;
 305       pos += j;
 306     }
 307   ASSERT(pos == num_elts);
 308
 309   // Start parallel splitting
 310   for (uns i=0; i<sorter_threads; i++)
 311     {
 312       struct rs_work *w = ctx->rs_works[i];
 313       w->w.go = rs_split;
 314       for (uns j=0; j<buckets; j++)
 315         {
 316           uns k = w->cnt[j];
 317           w->cnt[j] = cnt[j];
 318           cnt[j] += k;
 319         }
 320       work_submit(ctx->rs_work_queue, &w->w);
 321     }
 322   ASSERT(cnt[buckets-1] == num_elts);
 323
 324   // Wait for splits to finish
 325   while (work_wait(ctx->rs_work_queue))
 326     ;
 327
 328   // Recurse on buckets
 329   pos = 0;
 330   for (uns i=0; i<buckets; i++)
 331     {
 332       uns n = cnt[i] - pos;
 333       if (!n)
 334         continue;
 335       if (n * ctx->elt_size < sorter_thread_threshold || shift < ASORT_MIN_SHIFT)
 336         {
 337           struct rs_work *w = ep_alloc(ctx->eltpool);
 338           w->w.priority = 0;
 339           w->w.go = rs_finish;
 340           w->ctx = ctx;
 341           w->array = buffer;
 342           w->buffer = array;
 343           w->num_elts = n;
 344           w->shift = shift;
 345           w->swap_output = !swapped_output;
 346           if (n * ctx->elt_size < sorter_thread_chunk)
 347             {
 348               DBG("Sorting block %d+%d inline", pos, n);
 349               rs_finish(NULL, &w->w);
 350               ep_free(ctx->eltpool, w);
 351             }
 352           else
 353             {
 354               DBG("Scheduling block %d+%d", pos, n);
 355               work_submit(ctx->rs_work_queue, &w->w);
 356             }
 357         }
 358       else
 359         rs_radix(ctx, buffer, array, n, shift, !swapped_output);
 360       pos = cnt[i];
 361       array += n * ctx->elt_size;
 362       buffer += n * ctx->elt_size;
 363     }
 364 }
 365
 366 static void
 367 threaded_radixsort(struct asort_context *ctx, uns swap)
 368 {
 369   struct work_queue q;
 370
 371   asort_start_threads(1);
 372   work_queue_init(&asort_thread_pool, &q);
 373
 374   // Prepare work structures for counting and splitting.
 375   // We use big_alloc(), because we want to avoid cacheline aliasing between threads.
 376   ctx->rs_work_queue = &q;
 377   ctx->rs_works = alloca(sizeof(struct rs_work *) * sorter_threads);
 378   for (uns i=0; i<sorter_threads; i++)
 379     ctx->rs_works[i] = big_alloc(sizeof(struct rs_work) + sizeof(uns) * (1 << ctx->radix_bits));
 380
 381   // Prepare a pool for all remaining small bits which will be sorted on background.
 382   ctx->eltpool = ep_new(sizeof(struct rs_work), 1000);
 383
 384   // Do the big splitting
 385   rs_radix(ctx, ctx->array, ctx->buffer, ctx->num_elts, ctx->hash_bits, swap);
 386   for (uns i=0; i<sorter_threads; i++)
 387     big_free(ctx->rs_works[i], sizeof(struct rs_work) + sizeof(uns) * (1 << ctx->radix_bits));
 388
 389   // Finish the small blocks
 390   rs_wait_small(ctx);
 391
 392   ASSERT(!ctx->eltpool->num_allocated);
 393   ep_delete(ctx->eltpool);
 394   work_queue_cleanup(&q);
 395   asort_stop_threads();
 396 }
 397
 398 #else
 399
 400 void asort_start_threads(uns run UNUSED) { }
 401 void asort_stop_threads(void) { }
 402
 403 #endif
 404
 405 static uns
 406 predict_swap(struct asort_context *ctx)
 407 {
 408   uns bits = ctx->radix_bits;
 409   uns elts = ctx->num_elts;
 410   uns swap = 0;
 411
 412   while (elts * ctx->elt_size >= sorter_radix_threshold && bits >= ASORT_MIN_SHIFT)
 413     {
 414       DBG("Predicting pass: %d elts, %d bits", elts, bits);
 415       swap = !swap;
 416       elts >>= ctx->radix_bits;
 417       bits = MAX(bits, ctx->radix_bits) - ctx->radix_bits;
 418     }
 419   return swap;
 420 }
 421
 422 void
 423 asort_run(struct asort_context *ctx)
 424 {
 425   ASORT_TRACE("Array-sorting %d items per %d bytes, hash_bits=%d", ctx->num_elts, ctx->elt_size, ctx->hash_bits);
 426   uns allow_threads UNUSED = (sorter_threads > 1 &&
 427                               ctx->num_elts * ctx->elt_size >= sorter_thread_threshold &&
 428                               !(sorter_debug & SORT_DEBUG_ASORT_NO_THREADS));
 429
 430   if (ctx->num_elts * ctx->elt_size < sorter_radix_threshold ||
 431       ctx->hash_bits <= ASORT_MIN_SHIFT ||
 432       !ctx->radix_split ||
 433       (sorter_debug & SORT_DEBUG_ASORT_NO_RADIX))
 434     {
 435 #ifdef CONFIG_UCW_THREADS
 436       if (allow_threads)
 437         {
 438           ASORT_XTRACE(2, "Decided to use parallel quicksort");
 439           threaded_quicksort(ctx);
 440         }
 441       else
 442 #endif
 443         {
 444           ASORT_XTRACE(2, "Decided to use sequential quicksort");
 445           ctx->quicksort(ctx->array, ctx->num_elts);
 446         }
 447     }
 448   else
 449     {
 450       uns swap = predict_swap(ctx);
 451 #ifdef CONFIG_UCW_THREADS
 452       if (allow_threads)
 453         {
 454           ASORT_XTRACE(2, "Decided to use parallel radix-sort (swap=%d)", swap);
 455           threaded_radixsort(ctx, swap);
 456           return;
 457         }
 458       else
 459 #endif
 460         {
 461           ASORT_XTRACE(2, "Decided to use sequential radix-sort (swap=%d)", swap);
 462           asort_radix(ctx, ctx->array, ctx->buffer, ctx->num_elts, ctx->hash_bits, swap);
 463         }
 464       if (swap)
 465         ctx->array = ctx->buffer;
 466     }
 467
 468   ASORT_XTRACE(2, "Array-sort finished");
 469 }