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 <fcntl.h>
  16
  17 void *
  18 sorter_alloc(struct sort_context *ctx, uns size)
  19 {
  20   return mp_alloc_zero(ctx->pool, size);
  21 }
  22
  23 struct sort_bucket *
  24 sbuck_new(struct sort_context *ctx)
  25 {
  26   struct sort_bucket *b = sorter_alloc(ctx, sizeof(struct sort_bucket));
  27   b->ctx = ctx;
  28   return b;
  29 }
  30
  31 void
  32 sbuck_drop(struct sort_bucket *b)
  33 {
  34   if (b)
  35     {
  36       ASSERT(!(b->flags & SBF_DESTROYED));
  37       if (b->n.prev)
  38         clist_remove(&b->n);
  39       bclose(b->fb);
  40       bzero(b, sizeof(*b));
  41       b->flags = SBF_DESTROYED;
  42     }
  43 }
  44
  45 sh_off_t
  46 sbuck_size(struct sort_bucket *b)
  47 {
  48   if ((b->flags & SBF_OPEN_WRITE) && !(b->flags & SBF_SWAPPED_OUT))
  49     return btell(b->fb);
  50   else
  51     return b->size;
  52 }
  53
  54 int
  55 sbuck_have(struct sort_bucket *b)
  56 {
  57   return b && sbuck_size(b);
  58 }
  59
  60 static void
  61 sbuck_swap_in(struct sort_bucket *b)
  62 {
  63   if (b->flags & SBF_SWAPPED_OUT)
  64     {
  65       b->fb = bopen(b->filename, O_RDWR, sorter_stream_bufsize);
  66       if (b->flags & SBF_OPEN_WRITE)
  67         bseek(b->fb, 0, SEEK_END);
  68       bconfig(b->fb, BCONFIG_IS_TEMP_FILE, 1);
  69       b->flags &= ~SBF_SWAPPED_OUT;
  70       SORT_XTRACE("Swapped in %s", b->filename);
  71     }
  72 }
  73
  74 struct fastbuf *
  75 sbuck_read(struct sort_bucket *b)
  76 {
  77   sbuck_swap_in(b);
  78   if (b->flags & SBF_OPEN_READ)
  79     return b->fb;
  80   else if (b->flags & SBF_OPEN_WRITE)
  81     {
  82       b->size = btell(b->fb);
  83       b->flags = (b->flags & ~SBF_OPEN_WRITE) | SBF_OPEN_READ;
  84       brewind(b->fb);
  85       return b->fb;
  86     }
  87   else
  88     ASSERT(0);
  89 }
  90
  91 struct fastbuf *
  92 sbuck_write(struct sort_bucket *b)
  93 {
  94   sbuck_swap_in(b);
  95   if (b->flags & SBF_OPEN_WRITE)
  96     ASSERT(b->fb);
  97   else
  98     {
  99       ASSERT(!(b->flags & (SBF_OPEN_READ | SBF_DESTROYED)));
 100       b->fb = bopen_tmp(sorter_stream_bufsize);
 101       b->flags |= SBF_OPEN_WRITE;
 102       b->filename = mp_strdup(b->ctx->pool, b->fb->name);
 103     }
 104   return b->fb;
 105 }
 106
 107 void
 108 sbuck_swap_out(struct sort_bucket *b)
 109 {
 110   if ((b->flags & (SBF_OPEN_READ | SBF_OPEN_WRITE)) && b->fb)
 111     {
 112       if (b->flags & SBF_OPEN_WRITE)
 113         b->size = btell(b->fb);
 114       bconfig(b->fb, BCONFIG_IS_TEMP_FILE, 0);
 115       bclose(b->fb);
 116       b->fb = NULL;
 117       b->flags |= SBF_SWAPPED_OUT;
 118       SORT_XTRACE("Swapped out %s", b->filename);
 119     }
 120 }
 121
 122 void
 123 sorter_alloc_buf(struct sort_context *ctx)
 124 {
 125   if (ctx->big_buf)
 126     return;
 127   u64 bs = MAX(sorter_bufsize/2, 1);
 128   bs = ALIGN_TO(bs, (u64)CPU_PAGE_SIZE);
 129   ctx->big_buf = big_alloc(2*bs);
 130   ctx->big_buf_size = 2*bs;
 131   ctx->big_buf_half = ((byte*) ctx->big_buf) + bs;
 132   ctx->big_buf_half_size = bs;
 133   SORT_XTRACE("Allocated sorting buffer (%jd bytes)", (uintmax_t) bs);
 134 }
 135
 136 void
 137 sorter_free_buf(struct sort_context *ctx)
 138 {
 139   if (!ctx->big_buf)
 140     return;
 141   big_free(ctx->big_buf, ctx->big_buf_size);
 142   ctx->big_buf = NULL;
 143   SORT_XTRACE("Freed sorting buffer");
 144 }
 145
 146 static int sorter_presort(struct sort_context *ctx, struct sort_bucket *in, struct sort_bucket *out, struct sort_bucket *out_only)
 147 {
 148   /* FIXME: Mode with no presorting (mostly for debugging) */
 149   sorter_alloc_buf(ctx);
 150   if (in->flags & SBF_CUSTOM_PRESORT)
 151     {
 152       struct fastbuf *f = sbuck_write(out);
 153       return ctx->custom_presort(f, ctx->big_buf, ctx->big_buf_size);   // FIXME: out_only optimization?
 154     }
 155   return ctx->internal_sort(ctx, in, out, out_only);
 156 }
 157
 158 static inline struct sort_bucket *
 159 sbuck_join_to(struct sort_bucket *b)
 160 {
 161   struct sort_bucket *out = (struct sort_bucket *) b->n.prev;   // Such bucket is guaranteed to exist
 162   return (out->flags & SBF_FINAL) ? out : NULL;
 163 }
 164
 165 static void
 166 sorter_join(struct sort_bucket *b)
 167 {
 168   struct sort_bucket *join = sbuck_join_to(b);
 169   ASSERT(join);
 170
 171   // FIXME: What if the final bucket doesn't contain any file yet?
 172
 173   SORT_TRACE("Copying %jd bytes to output file", (uintmax_t) sbuck_size(b));
 174   struct fastbuf *src = sbuck_read(b);
 175   struct fastbuf *dest = sbuck_write(join);
 176   bbcopy(src, dest, ~0U);
 177   sbuck_drop(b);
 178 }
 179
 180 static void
 181 sorter_twoway(struct sort_context *ctx, struct sort_bucket *b)
 182 {
 183   struct sort_bucket *ins[3] = { NULL }, *outs[3] = { NULL };
 184   cnode *list_pos = b->n.prev;
 185   struct sort_bucket *join = sbuck_join_to(b);
 186   if (sorter_debug & SORT_DEBUG_NO_JOIN)
 187     join = NULL;
 188
 189   if (!(sorter_debug & SORT_DEBUG_NO_PRESORT) || (b->flags & SBF_CUSTOM_PRESORT))
 190     {
 191       SORT_TRACE("Presorting");
 192       ins[0] = sbuck_new(ctx);
 193       if (!sorter_presort(ctx, b, ins[0], join ? : ins[0]))
 194         {
 195           if (join)
 196             sbuck_drop(ins[0]);
 197           else
 198             clist_insert_after(&ins[0]->n, list_pos);
 199           sbuck_drop(b);
 200           return;
 201         }
 202
 203       ins[1] = sbuck_new(ctx);
 204       int i = 1;
 205       while (sorter_presort(ctx, b, ins[i], ins[i]))
 206         i = 1-i;
 207       sbuck_drop(b);
 208     }
 209   else
 210     {
 211       SORT_TRACE("Skipped presorting");
 212       ins[0] = b;
 213     }
 214
 215   SORT_TRACE("Main sorting");
 216   do {
 217     if (ins[0]->runs == 1 && ins[1]->runs == 1 && join)
 218       {
 219         // This is guaranteed to produce a single run, so join if possible
 220         outs[0] = join;
 221         outs[1] = NULL;
 222         ctx->twoway_merge(ctx, ins, outs);
 223         ASSERT(outs[0]->runs == 2);
 224         outs[0]->runs--;
 225         SORT_TRACE("Pass done (joined final run)");
 226         sbuck_drop(ins[0]);
 227         sbuck_drop(ins[1]);
 228         return;
 229       }
 230     outs[0] = sbuck_new(ctx);
 231     outs[1] = sbuck_new(ctx);
 232     outs[2] = NULL;
 233     ctx->twoway_merge(ctx, ins, outs);
 234     SORT_TRACE("Pass done (%d+%d runs, %jd+%jd bytes)", outs[0]->runs, outs[1]->runs, (uintmax_t) sbuck_size(outs[0]), (uintmax_t) sbuck_size(outs[1]));
 235     sbuck_drop(ins[0]);
 236     sbuck_drop(ins[1]);
 237     memcpy(ins, outs, 3*sizeof(struct sort_bucket *));
 238   } while (sbuck_have(ins[1]));
 239
 240   sbuck_drop(ins[1]);
 241   clist_insert_after(&ins[0]->n, list_pos);
 242 }
 243
 244 void
 245 sorter_run(struct sort_context *ctx)
 246 {
 247   ctx->pool = mp_new(4096);
 248   clist_init(&ctx->bucket_list);
 249
 250   /* FIXME: There should be a way how to detect size of the input file */
 251   /* FIXME: Remember to test sorting of empty files */
 252
 253   // Create bucket containing the source
 254   struct sort_bucket *bin = sbuck_new(ctx);
 255   bin->flags = SBF_SOURCE | SBF_OPEN_READ;
 256   if (ctx->custom_presort)
 257     bin->flags |= SBF_CUSTOM_PRESORT;
 258   else
 259     bin->fb = ctx->in_fb;
 260   bin->ident = "in";
 261   bin->size = ~(u64)0;
 262   bin->hash_bits = ctx->hash_bits;
 263   clist_add_tail(&ctx->bucket_list, &bin->n);
 264
 265   // Create bucket for the output
 266   struct sort_bucket *bout = sbuck_new(ctx);
 267   bout->flags = SBF_FINAL;
 268   bout->fb = ctx->out_fb;
 269   bout->ident = "out";
 270   bout->runs = 1;
 271   clist_add_head(&ctx->bucket_list, &bout->n);
 272
 273   struct sort_bucket *b;
 274   while (b = clist_next(&ctx->bucket_list, &bout->n))
 275     {
 276       if (!sbuck_have(b))
 277         sbuck_drop(b);
 278       else if (b->runs == 1)
 279         sorter_join(b);
 280       else
 281         sorter_twoway(ctx, b);
 282     }
 283
 284   sorter_free_buf(ctx);
 285   SORT_XTRACE("Final size: %jd", (uintmax_t) sbuck_size(bout));
 286   ctx->out_fb = sbuck_read(bout);
 287 }