X-Git-Url: http://mj.ucw.cz/gitweb/?a=blobdiff_plain;f=lib%2Fsorter%2Fgovern.c;h=738186e34da8a6eebddb0384f6aa3781b4387fcb;hb=396f0bc99fd49bfdac3bcc6681342fe76d4f79df;hp=aece061c528b9093eadeb8dd057663195ffed07e;hpb=5ff0a5b047624c8f215da9cd6c7182d609fcbd01;p=libucw.git diff --git a/lib/sorter/govern.c b/lib/sorter/govern.c index aece061c..738186e3 100644 --- a/lib/sorter/govern.c +++ b/lib/sorter/govern.c @@ -12,79 +12,159 @@ #include "lib/mempool.h" #include "lib/sorter/common.h" -struct sort_bucket * -sorter_new_bucket(struct sort_context *ctx) -{ - return mp_alloc_zero(ctx->pool, sizeof(struct sort_bucket)); -} +#include -struct fastbuf * -sorter_open_read(struct sort_bucket *b) +static int sorter_presort(struct sort_context *ctx, struct sort_bucket *in, struct sort_bucket *out, struct sort_bucket *out_only) { - /* FIXME: These functions should handle buckets with no fb and only name. */ - ASSERT(b->fb); - return b->fb; + sorter_alloc_buf(ctx); + if (in->flags & SBF_CUSTOM_PRESORT) + { + struct fastbuf *f = sbuck_write(out); + return ctx->custom_presort(f, ctx->big_buf, ctx->big_buf_size); // FIXME: out_only optimization? + } + return ctx->internal_sort(ctx, in, out, out_only); } -struct fastbuf * -sorter_open_write(struct sort_bucket *b) +static inline struct sort_bucket * +sbuck_join_to(struct sort_bucket *b) { - if (!b->fb) - b->fb = bopen_tmp(sorter_stream_bufsize); - return b->fb; + if (sorter_debug & SORT_DEBUG_NO_JOIN) + return NULL; + + struct sort_bucket *out = (struct sort_bucket *) b->n.prev; // Such bucket is guaranteed to exist + return (out->flags & SBF_FINAL) ? out : NULL; } -void -sorter_close_read(struct sort_bucket *b) +static void +sorter_join(struct sort_bucket *b) { - if (!b) - return; - ASSERT(b->fb); - bclose(b->fb); - b->fb = NULL; + struct sort_bucket *join = (struct sort_bucket *) b->n.prev; + ASSERT(join->flags & SBF_FINAL); + ASSERT(b->runs == 1); + + if (!sbuck_has_file(join)) + { + // The final bucket doesn't have any file associated yet, so replace + // it with the new bucket. + SORT_XTRACE("Replaced final bucket"); + b->flags |= SBF_FINAL; + sbuck_drop(join); + } + else + { + SORT_TRACE("Copying %jd bytes to output file", (uintmax_t) sbuck_size(b)); + struct fastbuf *src = sbuck_read(b); + struct fastbuf *dest = sbuck_write(join); + bbcopy(src, dest, ~0U); + sbuck_drop(b); + } } -void -sorter_close_write(struct sort_bucket *b) +static void +sorter_twoway(struct sort_context *ctx, struct sort_bucket *b) { - if (b->fb) + struct sort_bucket *ins[3] = { NULL }, *outs[3] = { NULL }; + cnode *list_pos = b->n.prev; + struct sort_bucket *join = sbuck_join_to(b); + + if (!(sorter_debug & SORT_DEBUG_NO_PRESORT) || (b->flags & SBF_CUSTOM_PRESORT)) { - b->size = btell(b->fb); - brewind(b->fb); + SORT_TRACE("Presorting"); + ins[0] = sbuck_new(ctx); + if (!sorter_presort(ctx, b, ins[0], join ? : ins[0])) + { + SORT_XTRACE("Sorted in memory"); + if (join) + sbuck_drop(ins[0]); + else + clist_insert_after(&ins[0]->n, list_pos); + sbuck_drop(b); + return; + } + + ins[1] = sbuck_new(ctx); + int i = 1; + while (sorter_presort(ctx, b, ins[i], ins[i])) + i = 1-i; + sbuck_drop(b); } - /* FIXME: Remove empty buckets from the list automatically? */ + else + { + SORT_TRACE("Skipped presorting"); + ins[0] = b; + } + + SORT_TRACE("Main sorting"); + do { + if (ins[0]->runs == 1 && ins[1]->runs == 1 && join) + { + // This is guaranteed to produce a single run, so join if possible + outs[0] = join; + outs[1] = NULL; + ctx->twoway_merge(ctx, ins, outs); + ASSERT(outs[0]->runs == 2); + outs[0]->runs--; + SORT_TRACE("Pass done (joined final run)"); + sbuck_drop(ins[0]); + sbuck_drop(ins[1]); + return; + } + outs[0] = sbuck_new(ctx); + outs[1] = sbuck_new(ctx); + outs[2] = NULL; + ctx->twoway_merge(ctx, ins, outs); + 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])); + sbuck_drop(ins[0]); + sbuck_drop(ins[1]); + memcpy(ins, outs, 3*sizeof(struct sort_bucket *)); + } while (sbuck_have(ins[1])); + + sbuck_drop(ins[1]); + clist_insert_after(&ins[0]->n, list_pos); } void sorter_run(struct sort_context *ctx) { ctx->pool = mp_new(4096); - ASSERT(!ctx->custom_presort); - ASSERT(!ctx->out_fb); clist_init(&ctx->bucket_list); /* FIXME: There should be a way how to detect size of the input file */ + /* FIXME: Remember to test sorting of empty files */ - /* Trivial 2-way merge with no presorting (just a testing hack) */ - struct sort_bucket *bin = sorter_new_bucket(ctx); - bin->flags = SBF_SOURCE; - bin->fb = ctx->in_fb; - bin->ident = "src"; - struct sort_bucket *ins[3], *outs[3]; - ins[0] = bin; - ins[1] = NULL; + // Create bucket containing the source + struct sort_bucket *bin = sbuck_new(ctx); + bin->flags = SBF_SOURCE | SBF_OPEN_READ; + if (ctx->custom_presort) + bin->flags |= SBF_CUSTOM_PRESORT; + else + bin->fb = ctx->in_fb; + bin->ident = "in"; + bin->size = ~(u64)0; + bin->hash_bits = ctx->hash_bits; + clist_add_tail(&ctx->bucket_list, &bin->n); - do { - outs[0] = sorter_new_bucket(ctx); - outs[1] = sorter_new_bucket(ctx); - outs[2] = NULL; - log(L_DEBUG, "Pass..."); - ctx->twoway_merge(ctx, ins, outs); - log(L_DEBUG, "Done (%d+%d runs)", outs[0]->runs, outs[1]->runs); - sorter_close_write(outs[0]); - sorter_close_write(outs[1]); - memcpy(ins, outs, 3*sizeof(struct sort_bucket *)); - } while (ins[1]->fb); + // Create bucket for the output + struct sort_bucket *bout = sbuck_new(ctx); + bout->flags = SBF_FINAL; + bout->fb = ctx->out_fb; + bout->ident = "out"; + bout->runs = 1; + clist_add_head(&ctx->bucket_list, &bout->n); + + struct sort_bucket *b; + while (bout = clist_head(&ctx->bucket_list), b = clist_next(&ctx->bucket_list, &bout->n)) + { + if (!sbuck_have(b)) + sbuck_drop(b); + else if (b->runs == 1) + sorter_join(b); + else + sorter_twoway(ctx, b); + } - ctx->out_fb = sorter_open_read(ins[0]); + sorter_free_buf(ctx); + sbuck_write(bout); // Force empty bucket to a file + SORT_XTRACE("Final size: %jd", (uintmax_t) sbuck_size(bout)); + ctx->out_fb = sbuck_read(bout); }