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 static void
  21 asort_radix(struct asort_context *ctx, void *array, void *buffer, uns num_elts, uns hash_bits, uns swapped_output)
  22 {
  23   // swap_output == 0 if result should be returned in `array', otherwise in `buffer'
  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 * cts->elt_size < sorter_radix_threshold || 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   if (thr)
 218     DBG("Thread %d: Finishing %d items, shift=%d", thr->id, w->num_elts, w->shift);
 219   if (w->shift < ASORT_MIN_SHIFT || w->num_elts * ctx->elt_size < sorter_radix_threshold)
 220     {
 221       w->ctx->quicksort(w->in, w->num_elts);
 222       if (w->swap_output)
 223         memcpy(w->out, w->in, w->num_elts * w->ctx->elt_size);
 224     }
 225   else
 226     asort_radix(w->ctx, w->in, w->out, w->num_elts, w->shift, w->swap_output);
 227   if (thr)
 228     DBG("Thread %d: Finishing done", thr->id);
 229 }
 230
 231 static void
 232 rs_wait_small(struct asort_context *ctx)
 233 {
 234   struct rs_work *w;
 235
 236   while (w = (struct rs_work *) work_wait(ctx->rs_work_queue))
 237     {
 238       DBG("Reaping small chunk of %d items", w->num_elts);
 239       ep_free(ctx->eltpool, w);
 240     }
 241 }
 242
 243 static void
 244 rs_radix(struct asort_context *ctx, void *array, void *buffer, uns num_elts, uns hash_bits, uns swapped_output)
 245 {
 246   uns buckets = (1 << ctx->radix_bits);
 247   uns shift = (hash_bits > ctx->radix_bits) ? (hash_bits - ctx->radix_bits) : 0;
 248   uns cnt[buckets];
 249   uns blksize = num_elts / sorter_threads;
 250   DBG(">>> n=%d h=%d s=%d blk=%d sw=%d", num_elts, hash_bits, shift, blksize, swapped_output);
 251
 252   // If there are any small chunks in progress, wait for them to finish
 253   rs_wait_small(ctx);
 254
 255   // Start parallel counting
 256   void *iptr = array;
 257   for (uns i=0; i<sorter_threads; i++)
 258     {
 259       struct rs_work *w = ctx->rs_works[i];
 260       w->w.priority = 0;
 261       w->w.go = rs_count;
 262       w->ctx = ctx;
 263       w->in = iptr;
 264       w->out = buffer;
 265       w->num_elts = blksize;
 266       if (i == sorter_threads-1)
 267         w->num_elts += num_elts % sorter_threads;
 268       w->shift = shift;
 269       iptr += w->num_elts * ctx->elt_size;
 270       bzero(w->cnt, sizeof(uns) * buckets);
 271       work_submit(ctx->rs_work_queue, &w->w);
 272     }
 273
 274   // Get bucket sizes from the counts
 275   bzero(cnt, sizeof(cnt));
 276   for (uns i=0; i<sorter_threads; i++)
 277     {
 278       struct rs_work *w = (struct rs_work *) work_wait(ctx->rs_work_queue);
 279       ASSERT(w);
 280       for (uns j=0; j<buckets; j++)
 281         cnt[j] += w->cnt[j];
 282     }
 283
 284   // Calculate bucket starts
 285   uns pos = 0;
 286   for (uns i=0; i<buckets; i++)
 287     {
 288       uns j = cnt[i];
 289       cnt[i] = pos;
 290       pos += j;
 291     }
 292   ASSERT(pos == num_elts);
 293
 294   // Start parallel splitting
 295   for (uns i=0; i<sorter_threads; i++)
 296     {
 297       struct rs_work *w = ctx->rs_works[i];
 298       w->w.go = rs_split;
 299       for (uns j=0; j<buckets; j++)
 300         {
 301           uns k = w->cnt[j];
 302           w->cnt[j] = cnt[j];
 303           cnt[j] += k;
 304         }
 305       work_submit(ctx->rs_work_queue, &w->w);
 306     }
 307   ASSERT(cnt[buckets-1] == num_elts);
 308
 309   // Wait for splits to finish
 310   while (work_wait(ctx->rs_work_queue))
 311     ;
 312
 313   // Recurse on buckets
 314   pos = 0;
 315   for (uns i=0; i<buckets; i++)
 316     {
 317       uns n = cnt[i] - pos;
 318       if (!n)
 319         continue;
 320       if (n * ctx->elt_size < sorter_thread_threshold)
 321         {
 322           struct rs_work *w = ep_alloc(ctx->eltpool);
 323           w->w.priority = 0;
 324           w->w.go = rs_finish;
 325           w->ctx = ctx;
 326           w->in = buffer;
 327           w->out = array;
 328           w->num_elts = n;
 329           w->shift = shift;
 330           w->swap_output = !swapped_output;
 331           if (n * ctx->elt_size < sorter_thread_chunk)
 332             {
 333               DBG("Sorting block %d+%d inline", pos, n);
 334               rs_finish(NULL, &w->w);
 335               ep_free(ctx->eltpool, w);
 336             }
 337           else
 338             {
 339               DBG("Scheduling block %d+%d", pos, n);
 340               work_submit(ctx->rs_work_queue, &w->w);
 341             }
 342         }
 343       else
 344         rs_radix(ctx, buffer, array, n, shift, !swapped_output);
 345       pos = cnt[i];
 346       array += n * ctx->elt_size;
 347       buffer += n * ctx->elt_size;
 348     }
 349 }
 350
 351 static void
 352 threaded_radixsort(struct asort_context *ctx)
 353 {
 354   struct work_queue q;
 355
 356   asort_start_threads(1);
 357   work_queue_init(&asort_thread_pool, &q);
 358
 359   // Prepare work structures for counting and splitting.
 360   // We use big_alloc(), because we want to avoid cacheline aliasing between threads.
 361   ctx->rs_work_queue = &q;
 362   ctx->rs_works = alloca(sizeof(struct rs_work *) * sorter_threads);
 363   for (uns i=0; i<sorter_threads; i++)
 364     ctx->rs_works[i] = big_alloc(sizeof(struct rs_work) + sizeof(uns) * (1 << ctx->radix_bits));
 365
 366   // Prepare a pool for all remaining small bits which will be sorted on background.
 367   ctx->eltpool = ep_new(sizeof(struct rs_work), 1000);
 368
 369   // Do the big splitting
 370   // FIXME: Set the swap bit carefully.
 371   rs_radix(ctx, ctx->array, ctx->buffer, ctx->num_elts, ctx->hash_bits, 0);
 372   for (uns i=0; i<sorter_threads; i++)
 373     big_free(ctx->rs_works[i], sizeof(struct rs_work) + sizeof(uns) * (1 << ctx->radix_bits));
 374
 375   // Finish the small blocks
 376   rs_wait_small(ctx);
 377
 378   ASSERT(!ctx->eltpool->num_allocated);
 379   ep_delete(ctx->eltpool);
 380   work_queue_cleanup(&q);
 381   asort_stop_threads();
 382 }
 383
 384 #else
 385
 386 void asort_start_threads(uns run UNUSED) { }
 387 void asort_stop_threads(void) { }
 388
 389 #endif
 390
 391 void
 392 asort_run(struct asort_context *ctx)
 393 {
 394   SORT_XTRACE(10, "Array-sorting %d items per %d bytes, hash_bits=%d", ctx->num_elts, ctx->elt_size, ctx->hash_bits);
 395   uns allow_threads UNUSED = (sorter_threads > 1 &&
 396                               ctx->num_elts * ctx->elt_size >= sorter_thread_threshold &&
 397                               !(sorter_debug & SORT_DEBUG_ASORT_NO_THREADS));
 398
 399   if (ctx->num_elts * ctx->elt_size < sorter_radix_threshold ||
 400       ctx->hash_bits <= ASORT_MIN_SHIFT ||
 401       !ctx->radix_split ||
 402       (sorter_debug & SORT_DEBUG_ASORT_NO_RADIX))
 403     {
 404 #ifdef CONFIG_UCW_THREADS
 405       if (allow_threads)
 406         {
 407           SORT_XTRACE(12, "Decided to use parallel quicksort");
 408           threaded_quicksort(ctx);
 409           return;
 410         }
 411 #endif
 412       SORT_XTRACE(12, "Decided to use sequential quicksort");
 413       ctx->quicksort(ctx->array, ctx->num_elts);
 414     }
 415   else
 416     {
 417 #ifdef CONFIG_UCW_THREADS
 418       if (allow_threads)
 419         {
 420           SORT_XTRACE(12, "Decided to use parallel radix-sort");
 421           threaded_radixsort(ctx);
 422           return;
 423         }
 424 #endif
 425       SORT_XTRACE(12, "Decided to use sequential radix-sort");
 426       // FIXME: select dest buffer
 427       asort_radix(ctx, ctx->array, ctx->buffer, ctx->num_elts, ctx->hash_bits, 0);
 428     }
 429
 430   SORT_XTRACE(11, "Array-sort finished");
 431 }