Implemented swapping in/out buckets.

In some sorting methods, we want to use lots of buckets, but avoid
keeping all the files open and their buffers allocated. The bucket
machinery is now able to swap out a bucket (close its fastbuf) and
then transparently swap it in as needed.

diff --git a/lib/sorter/common.h b/lib/sorter/common.h
index 6372b5f450f9f02114b9367b7f474f47ba7f1d1c..a17a809a75b19b479814aa2dfdfc3e1f90e7dc80 100644 (file)
--- a/lib/sorter/common.h
+++ b/lib/sorter/common.h
@@ -27,9 +27,10 @@ enum sort_debug {
 
 struct sort_bucket {
   cnode n;
+  struct sort_context *ctx;
   uns flags;
   struct fastbuf *fb;
-  byte *name;
+  byte *filename;
   u64 size;                            // Size in bytes (not valid when writing)
   uns runs;                            // Number of runs, 0 if not sorted
   uns hash_bits;                       // Remaining bits of the hash function
@@ -43,6 +44,7 @@ enum sort_bucket_flags {
   SBF_OPEN_WRITE = 256,                        // We are currently writing to the fastbuf
   SBF_OPEN_READ = 512,                 // We are reading from the fastbuf
   SBF_DESTROYED = 1024,                        // Already done with, no further references allowed
+  SBF_SWAPPED_OUT = 2048,              // Swapped out to a named file
 };
 
 struct sort_context {
@@ -81,5 +83,6 @@ int sbuck_have(struct sort_bucket *b);
 sh_off_t sbuck_size(struct sort_bucket *b);
 struct fastbuf *sbuck_read(struct sort_bucket *b);
 struct fastbuf *sbuck_write(struct sort_bucket *b);
+void sbuck_swap_out(struct sort_bucket *b);
 
 #endif

diff --git a/lib/sorter/govern.c b/lib/sorter/govern.c
index f962810b3ba0ce72a47d8d63c4f3f8b4b54329eb..8ab251b9fca7deab45ffce24d56603dcd629c426 100644 (file)
--- a/lib/sorter/govern.c
+++ b/lib/sorter/govern.c
@@ -12,6 +12,8 @@
 #include "lib/mempool.h"
 #include "lib/sorter/common.h"
 
+#include <fcntl.h>
+
 void *
 sorter_alloc(struct sort_context *ctx, uns size)
 {
@@ -21,7 +23,9 @@ sorter_alloc(struct sort_context *ctx, uns size)
 struct sort_bucket *
 sbuck_new(struct sort_context *ctx)
 {
-  return sorter_alloc(ctx, sizeof(struct sort_bucket));
+  struct sort_bucket *b = sorter_alloc(ctx, sizeof(struct sort_bucket));
+  b->ctx = ctx;
+  return b;
 }
 
 void
@@ -41,7 +45,7 @@ sbuck_drop(struct sort_bucket *b)
 sh_off_t
 sbuck_size(struct sort_bucket *b)
 {
-  if (b->flags & SBF_OPEN_WRITE)
+  if ((b->flags & SBF_OPEN_WRITE) && !(b->flags & SBF_SWAPPED_OUT))
     return btell(b->fb);
   else
     return b->size;
@@ -53,11 +57,24 @@ sbuck_have(struct sort_bucket *b)
   return b && sbuck_size(b);
 }
 
+static void
+sbuck_swap_in(struct sort_bucket *b)
+{
+  if (b->flags & SBF_SWAPPED_OUT)
+    {
+      b->fb = bopen(b->filename, O_RDWR, sorter_stream_bufsize);
+      if (b->flags & SBF_OPEN_WRITE)
+       bseek(b->fb, 0, SEEK_END);
+      bconfig(b->fb, BCONFIG_IS_TEMP_FILE, 1);
+      b->flags &= ~SBF_SWAPPED_OUT;
+      SORT_XTRACE("Swapped in %s", b->filename);
+    }
+}
+
 struct fastbuf *
 sbuck_read(struct sort_bucket *b)
 {
-  /* FIXME: These functions should handle buckets with no fb and only name. */
-  ASSERT(b->fb);
+  sbuck_swap_in(b);
   if (b->flags & SBF_OPEN_READ)
     return b->fb;
   else if (b->flags & SBF_OPEN_WRITE)
@@ -74,6 +91,7 @@ sbuck_read(struct sort_bucket *b)
 struct fastbuf *
 sbuck_write(struct sort_bucket *b)
 {
+  sbuck_swap_in(b);
   if (b->flags & SBF_OPEN_WRITE)
     ASSERT(b->fb);
   else
@@ -81,10 +99,26 @@ sbuck_write(struct sort_bucket *b)
       ASSERT(!(b->flags & (SBF_OPEN_READ | SBF_DESTROYED)));
       b->fb = bopen_tmp(sorter_stream_bufsize);
       b->flags |= SBF_OPEN_WRITE;
+      b->filename = mp_strdup(b->ctx->pool, b->fb->name);
     }
   return b->fb;
 }
 
+void
+sbuck_swap_out(struct sort_bucket *b)
+{
+  if ((b->flags & (SBF_OPEN_READ | SBF_OPEN_WRITE)) && b->fb)
+    {
+      if (b->flags & SBF_OPEN_WRITE)
+       b->size = btell(b->fb);
+      bconfig(b->fb, BCONFIG_IS_TEMP_FILE, 0);
+      bclose(b->fb);
+      b->fb = NULL;
+      b->flags |= SBF_SWAPPED_OUT;
+      SORT_XTRACE("Swapped out %s", b->filename);
+    }
+}
+
 void
 sorter_alloc_buf(struct sort_context *ctx)
 {
@@ -156,7 +190,6 @@ sorter_twoway(struct sort_context *ctx, struct sort_bucket *b)
     {
       SORT_TRACE("Presorting");
       ins[0] = sbuck_new(ctx);
-      sbuck_read(b);
       if (!sorter_presort(ctx, b, ins[0], join ? : ins[0]))
        {
          if (join)

diff --git a/lib/sorter/s-internal.h b/lib/sorter/s-internal.h
index c05ca077e574868bb7e27cd94c85ddc6cf24ee10..329c3b4a46a1703e9c18bd2e84f5338dc5bf8063 100644 (file)
--- a/lib/sorter/s-internal.h
+++ b/lib/sorter/s-internal.h
@@ -22,8 +22,7 @@ typedef struct {
 static int P(internal)(struct sort_context *ctx, struct sort_bucket *bin, struct sort_bucket *bout, struct sort_bucket *bout_only)
 {
   sorter_alloc_buf(ctx);
-  ASSERT(bin->fb);                     // Expects the input bucket to be already open for reading
-  struct fastbuf *in = bin->fb;
+  struct fastbuf *in = sbuck_read(bin);
 
   P(key) key, *keybuf = ctx->key_buf;
   if (!keybuf)