lib/sorter/govern.c

   1 /*
   2  *      UCW Library -- Universal Sorter: Governing Routines
   3  *
   4  *      (c) 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 #include "lib/lib.h"
  11 #include "lib/fastbuf.h"
  12 #include "lib/mempool.h"
  13 #include "lib/sorter/common.h"
  14
  15 #include <string.h>
  16
  17 static int
  18 sorter_presort(struct sort_context *ctx, struct sort_bucket *in, struct sort_bucket *out, struct sort_bucket *out_only)
  19 {
  20   sorter_alloc_buf(ctx);
  21   if (in->flags & SBF_CUSTOM_PRESORT)
  22     {
  23       struct fastbuf *f = sbuck_write(out);
  24       return ctx->custom_presort(f, ctx->big_buf, ctx->big_buf_size);   // FIXME: out_only optimization?
  25     }
  26   return ctx->internal_sort(ctx, in, out, out_only);
  27 }
  28
  29 static inline struct sort_bucket *
  30 sbuck_join_to(struct sort_bucket *b)
  31 {
  32   if (sorter_debug & SORT_DEBUG_NO_JOIN)
  33     return NULL;
  34
  35   struct sort_bucket *out = (struct sort_bucket *) b->n.prev;   // Such bucket is guaranteed to exist
  36   return (out->flags & SBF_FINAL) ? out : NULL;
  37 }
  38
  39 static void
  40 sorter_join(struct sort_bucket *b)
  41 {
  42   struct sort_bucket *join = (struct sort_bucket *) b->n.prev;
  43   ASSERT(join->flags & SBF_FINAL);
  44   ASSERT(b->runs == 1);
  45
  46   if (!sbuck_has_file(join))
  47     {
  48       // The final bucket doesn't have any file associated yet, so replace
  49       // it with the new bucket.
  50       SORT_XTRACE(2, "Replaced final bucket");
  51       b->flags |= SBF_FINAL;
  52       sbuck_drop(join);
  53     }
  54   else
  55     {
  56       SORT_TRACE("Copying to output file: %s", F_BSIZE(b));
  57       struct fastbuf *src = sbuck_read(b);
  58       struct fastbuf *dest = sbuck_write(join);
  59       bbcopy(src, dest, ~0U);
  60       sbuck_drop(b);
  61     }
  62 }
  63
  64 static void
  65 sorter_twoway(struct sort_context *ctx, struct sort_bucket *b)
  66 {
  67   struct sort_bucket *ins[3] = { NULL }, *outs[3] = { NULL };
  68   cnode *list_pos = b->n.prev;
  69   struct sort_bucket *join = sbuck_join_to(b);
  70
  71   if (!(sorter_debug & SORT_DEBUG_NO_PRESORT) || (b->flags & SBF_CUSTOM_PRESORT))
  72     {
  73       SORT_XTRACE(2, "Presorting");
  74       ins[0] = sbuck_new(ctx);
  75       if (!sorter_presort(ctx, b, ins[0], join ? : ins[0]))
  76         {
  77           SORT_TRACE("Sorted in memory");
  78           if (join)
  79             sbuck_drop(ins[0]);
  80           else
  81             clist_insert_after(&ins[0]->n, list_pos);
  82           sbuck_drop(b);
  83           return;
  84         }
  85
  86       ins[1] = sbuck_new(ctx);
  87       int i = 1;
  88       while (sorter_presort(ctx, b, ins[i], ins[i]))
  89         i = 1-i;
  90       sbuck_drop(b);
  91     }
  92   else
  93     {
  94       SORT_XTRACE(2, "Presorting disabled");
  95       ins[0] = b;
  96     }
  97
  98   SORT_XTRACE(2, "Main sorting");
  99   uns pass = 0;
 100   do {
 101     ++pass;
 102     if (ins[0]->runs == 1 && ins[1]->runs == 1 && join)
 103       {
 104         // This is guaranteed to produce a single run, so join if possible
 105         outs[0] = join;
 106         outs[1] = NULL;
 107         ctx->twoway_merge(ctx, ins, outs);
 108         ASSERT(outs[0]->runs == 2);
 109         outs[0]->runs--;
 110         SORT_TRACE("Mergesort pass %d (final run, %s)", pass, F_BSIZE(outs[0]));
 111         sbuck_drop(ins[0]);
 112         sbuck_drop(ins[1]);
 113         return;
 114       }
 115     outs[0] = sbuck_new(ctx);
 116     outs[1] = sbuck_new(ctx);
 117     outs[2] = NULL;
 118     ctx->twoway_merge(ctx, ins, outs);
 119     SORT_TRACE("Mergesort pass %d (%d+%d runs, %s+%s)", pass, outs[0]->runs, outs[1]->runs, F_BSIZE(outs[0]), F_BSIZE(outs[1]));
 120     sbuck_drop(ins[0]);
 121     sbuck_drop(ins[1]);
 122     memcpy(ins, outs, 3*sizeof(struct sort_bucket *));
 123   } while (sbuck_have(ins[1]));
 124
 125   sbuck_drop(ins[1]);
 126   clist_insert_after(&ins[0]->n, list_pos);
 127 }
 128
 129 void
 130 sorter_run(struct sort_context *ctx)
 131 {
 132   ctx->pool = mp_new(4096);
 133   clist_init(&ctx->bucket_list);
 134
 135   /* FIXME: There should be a way how to detect size of the input file */
 136   /* FIXME: Remember to test sorting of empty files */
 137
 138   // Create bucket containing the source
 139   struct sort_bucket *bin = sbuck_new(ctx);
 140   bin->flags = SBF_SOURCE | SBF_OPEN_READ;
 141   if (ctx->custom_presort)
 142     bin->flags |= SBF_CUSTOM_PRESORT;
 143   else
 144     bin->fb = ctx->in_fb;
 145   bin->ident = "in";
 146   bin->size = ~(u64)0;
 147   bin->hash_bits = ctx->hash_bits;
 148   clist_add_tail(&ctx->bucket_list, &bin->n);
 149
 150   // Create bucket for the output
 151   struct sort_bucket *bout = sbuck_new(ctx);
 152   bout->flags = SBF_FINAL;
 153   bout->fb = ctx->out_fb;
 154   bout->ident = "out";
 155   bout->runs = 1;
 156   clist_add_head(&ctx->bucket_list, &bout->n);
 157
 158   struct sort_bucket *b;
 159   while (bout = clist_head(&ctx->bucket_list), b = clist_next(&ctx->bucket_list, &bout->n))
 160     {
 161       if (!sbuck_have(b))
 162         sbuck_drop(b);
 163       else if (b->runs == 1)
 164         sorter_join(b);
 165       else
 166         sorter_twoway(ctx, b);
 167     }
 168
 169   sorter_free_buf(ctx);
 170   sbuck_write(bout);            // Force empty bucket to a file
 171   SORT_XTRACE(2, "Final size: %s", F_BSIZE(bout));
 172   ctx->out_fb = sbuck_read(bout);
 173 }