extstore: crawler fix and ext_low_ttl option

LRU crawler was not marking reclaimed expired items as removed from the
storage engine. This could cause fragmentation to persist much longer than it
should, but would not cause any problems once compaction started.

Adds "ext_low_ttl" option. Items with a remaining expiration age below this
value are grouped into special pages. If you have a mixed TTL workload this
would help prevent low TTL items from causing excess fragmentation/compaction.

Pages with low ttl items are excluded from compaction.

10 files changed, 117 insertions, 31 deletions

diff --git a/crawler.c b/crawler.c
index d43fa92..4fac58b 100644
--- a/crawler.c
+++ b/crawler.c
@@ -213,6 +213,9 @@ static void crawler_expired_eval(crawler_module_t *cm, item *search, uint32_t hv
         if ((search->it_flags & ITEM_FETCHED) == 0 && !is_flushed) {
             crawlers[i].unfetched++;
         }
+#ifdef EXTSTORE
+        STORAGE_delete(storage, search);
+#endif
         do_item_unlink_nolock(search, hv);
         do_item_remove(search);
         assert(search->slabs_clsid == 0);
diff --git a/extstore.c b/extstore.c
index 2e26462..e499178 100644
--- a/extstore.c
+++ b/extstore.c
@@ -785,6 +785,7 @@ static void *extstore_maint_thread(void *arg) {
             if (p->obj_count > 0 && !p->closed) {
                 pd[p->id].version = p->version;
                 pd[p->id].bytes_used = p->bytes_used;
+                pd[p->id].bucket = p->bucket;
                 if (p->version < low_version) {
                     low_version = p->version;
                     low_page = i;
diff --git a/extstore.h b/extstore.h
index 6a6b2bd..84157a0 100644
--- a/extstore.h
+++ b/extstore.h
@@ -7,6 +7,7 @@
 struct extstore_page_data {
     uint64_t version;
     uint64_t bytes_used;
+    unsigned int bucket;
 };
 
 /* Pages can have objects deleted from them at any time. This creates holes
diff --git a/memcached.c b/memcached.c
index 297e251..b436ad3 100644
--- a/memcached.c
+++ b/memcached.c
@@ -4469,32 +4469,31 @@ static void process_lru_command(conn *c, token_t *tokens, const size_t ntokens)
 #ifdef EXTSTORE
 static void process_extstore_command(conn *c, token_t *tokens, const size_t ntokens) {
     set_noreply_maybe(c, tokens, ntokens);
-    if (strcmp(tokens[1].value, "item_size") == 0 && ntokens >= 3) {
-        if (!safe_strtoul(tokens[2].value, &settings.ext_item_size)) {
-            out_string(c, "ERROR");
-        } else {
-            out_string(c, "OK");
-        }
-    } else if (strcmp(tokens[1].value, "item_age") == 0 && ntokens >= 3) {
-        if (!safe_strtoul(tokens[2].value, &settings.ext_item_age)) {
-            out_string(c, "ERROR");
-        } else {
-            out_string(c, "OK");
-        }
-    } else if (strcmp(tokens[1].value, "recache_rate") == 0 && ntokens >= 3) {
-        if (!safe_strtoul(tokens[2].value, &settings.ext_recache_rate)) {
-            out_string(c, "ERROR");
-        } else {
-            out_string(c, "OK");
-        }
-    } else if (strcmp(tokens[1].value, "max_frag") == 0 && ntokens >= 3) {
-        if (!safe_strtod(tokens[2].value, &settings.ext_max_frag)) {
-            out_string(c, "ERROR");
-        } else {
-            out_string(c, "OK");
-        }
+    bool ok = true;
+    if (ntokens < 3) {
+        ok = false;
+    } else if (strcmp(tokens[1].value, "item_size") == 0) {
+        if (!safe_strtoul(tokens[2].value, &settings.ext_item_size))
+            ok = false;
+    } else if (strcmp(tokens[1].value, "item_age") == 0) {
+        if (!safe_strtoul(tokens[2].value, &settings.ext_item_age))
+            ok = false;
+    } else if (strcmp(tokens[1].value, "low_ttl") == 0) {
+        if (!safe_strtoul(tokens[2].value, &settings.ext_low_ttl))
+            ok = false;
+    } else if (strcmp(tokens[1].value, "recache_rate") == 0) {
+        if (!safe_strtoul(tokens[2].value, &settings.ext_recache_rate))
+            ok = false;
+    } else if (strcmp(tokens[1].value, "max_frag") == 0) {
+        if (!safe_strtod(tokens[2].value, &settings.ext_max_frag))
+            ok = false;
     } else {
+        ok = false;
+    }
+    if (!ok) {
         out_string(c, "ERROR");
+    } else {
+        out_string(c, "OK");
     }
 }
 #endif
@@ -6161,6 +6160,7 @@ static void usage(void) {
            "   - ext_threads:         number of IO threads to run.\n"
            "   - ext_item_size:       store items larger than this (bytes)\n"
            "   - ext_item_age:        store items idle at least this long\n"
+           "   - ext_low_ttl:         consider TTLs lower than this specially\n"
            "   - ext_recache_rate:    recache an item every N accesses\n"
            "   - ext_max_frag:        max page fragmentation to tolerage\n"
            "                          (see doc/storage.txt for more info)"
@@ -6496,6 +6496,7 @@ int main (int argc, char **argv) {
         EXT_PATH,
         EXT_ITEM_SIZE,
         EXT_ITEM_AGE,
+        EXT_LOW_TTL,
         EXT_RECACHE_RATE,
         EXT_MAX_FRAG,
 #endif
@@ -6549,6 +6550,7 @@ int main (int argc, char **argv) {
         [EXT_PATH] = "ext_path",
         [EXT_ITEM_SIZE] = "ext_item_size",
         [EXT_ITEM_AGE] = "ext_item_age",
+        [EXT_LOW_TTL] = "ext_low_ttl",
         [EXT_RECACHE_RATE] = "ext_recache_rate",
         [EXT_MAX_FRAG] = "ext_max_frag",
 #endif
@@ -6568,16 +6570,17 @@ int main (int argc, char **argv) {
 #ifdef EXTSTORE
     settings.ext_item_size = 512;
     settings.ext_item_age = 0;
+    settings.ext_low_ttl = 0;
     settings.ext_recache_rate = 2000;
     settings.ext_max_frag = 0.8;
     settings.ext_wbuf_size = 1024 * 1024 * 4;
     ext_cf.page_size = 1024 * 1024 * 64;
     ext_cf.page_count = 64;
     ext_cf.wbuf_size = settings.ext_wbuf_size;
-    ext_cf.wbuf_count = 3;
+    ext_cf.wbuf_count = 4;
     ext_cf.io_threadcount = 1;
     ext_cf.io_depth = 1;
-    ext_cf.page_buckets = 3;
+    ext_cf.page_buckets = 4;
 #endif
 
     /* Run regardless of initializing it later */
@@ -7188,6 +7191,16 @@ int main (int argc, char **argv) {
                     return 1;
                 }
                 break;
+            case EXT_LOW_TTL:
+                if (subopts_value == NULL) {
+                    fprintf(stderr, "Missing ext_low_ttl argument\n");
+                    return 1;
+                }
+                if (!safe_strtoul(subopts_value, &settings.ext_low_ttl)) {
+                    fprintf(stderr, "could not parse argument to ext_low_ttl\n");
+                    return 1;
+                }
+                break;
             case EXT_RECACHE_RATE:
                 if (subopts_value == NULL) {
                     fprintf(stderr, "Missing ext_recache_rate argument\n");
diff --git a/memcached.h b/memcached.h
index 5e6f185..5a6cec8 100644
--- a/memcached.h
+++ b/memcached.h
@@ -408,6 +408,7 @@ struct settings {
 #ifdef EXTSTORE
     unsigned int ext_item_size; /* minimum size of items to store externally */
     unsigned int ext_item_age; /* max age of tail item before storing ext. */
+    unsigned int ext_low_ttl; /* remaining TTL below this uses own pages */
     unsigned int ext_recache_rate; /* counter++ % recache_rate == 0 > recache */
     unsigned int ext_wbuf_size; /* read only note for the engine */
     double ext_max_frag; /* ideal maximum page fragmentation */
diff --git a/storage.c b/storage.c
index cf2f7fb..aee0dbe 100644
--- a/storage.c
+++ b/storage.c
@@ -10,6 +10,7 @@
 #define PAGE_BUCKET_DEFAULT 0
 #define PAGE_BUCKET_COMPACT 1
 #define PAGE_BUCKET_CHUNKED 2
+#define PAGE_BUCKET_LOWTTL  3
 
 int lru_maintainer_store(void *storage, const int clsid) {
     //int i;
@@ -25,7 +26,7 @@ int lru_maintainer_store(void *storage, const int clsid) {
     chunks_free = slabs_available_chunks(clsid, &mem_limit_reached,
             NULL, &chunks_perslab);
     // if we are low on chunks and no spare, push out early.
-    if (chunks_free < (chunks_perslab / 2) && mem_limit_reached)
+    if (chunks_free < chunks_perslab && mem_limit_reached)
         item_age = 0;
 
     it_info.it = NULL;
@@ -57,6 +58,10 @@ int lru_maintainer_store(void *storage, const int clsid) {
         if (hdr_it != NULL) {
             int bucket = (it->it_flags & ITEM_CHUNKED) ?
                 PAGE_BUCKET_CHUNKED : PAGE_BUCKET_DEFAULT;
+            // Compres soon to expire items into similar pages.
+            if (it->exptime - current_time < settings.ext_low_ttl) {
+                bucket = PAGE_BUCKET_LOWTTL;
+            }
             hdr_it->it_flags |= ITEM_HDR;
             io.len = orig_ntotal;
             io.mode = OBJ_IO_WRITE;
@@ -156,7 +161,8 @@ static int storage_compact_check(void *storage, logger *l,
 
     // find oldest page by version that violates the constraint
     for (x = 0; x < st.page_count; x++) {
-        if (st.page_data[x].version == 0)
+        if (st.page_data[x].version == 0 ||
+            st.page_data[x].bucket == PAGE_BUCKET_LOWTTL)
             continue;
         if (st.page_data[x].bytes_used < frag_limit) {
             if (st.page_data[x].version < low_version) {
diff --git a/t/binary-extstore.t b/t/binary-extstore.t
index 5a52bf4..9028422 100755
--- a/t/binary-extstore.t
+++ b/t/binary-extstore.t
@@ -17,7 +17,7 @@ if (!supports_extstore()) {
 
 $ext_path = "/tmp/extstore.$$";
 
-my $server = new_memcached("-m 64 -U 0 -o ext_page_size=8,ext_page_count=8,ext_wbuf_size=2,ext_wbuf_count=3,ext_threads=1,ext_io_depth=2,ext_item_size=512,ext_item_age=2,ext_recache_rate=10000,ext_max_frag=0.9,ext_path=$ext_path,no_lru_crawler");
+my $server = new_memcached("-m 64 -U 0 -o ext_page_size=8,ext_page_count=8,ext_wbuf_size=2,ext_threads=1,ext_io_depth=2,ext_item_size=512,ext_item_age=2,ext_recache_rate=10000,ext_max_frag=0.9,ext_path=$ext_path,no_lru_crawler");
 ok($server, "started the server");
 
 # Based almost 100% off testClient.py which is:
diff --git a/t/chunked-extstore.t b/t/chunked-extstore.t
index e80c768..f3cf903 100644
--- a/t/chunked-extstore.t
+++ b/t/chunked-extstore.t
@@ -18,7 +18,7 @@ if (!supports_extstore()) {
 
 $ext_path = "/tmp/extstore.$$";
 
-my $server = new_memcached("-m 64 -U 0 -o ext_page_size=8,ext_page_count=8,ext_wbuf_size=2,ext_wbuf_count=3,ext_threads=1,ext_io_depth=2,ext_item_size=512,ext_item_age=2,ext_recache_rate=10000,ext_max_frag=0.9,ext_path=$ext_path,slab_chunk_max=16384");
+my $server = new_memcached("-m 64 -U 0 -o ext_page_size=8,ext_page_count=8,ext_wbuf_size=2,ext_threads=1,ext_io_depth=2,ext_item_size=512,ext_item_age=2,ext_recache_rate=10000,ext_max_frag=0.9,ext_path=$ext_path,slab_chunk_max=16384");
 my $sock = $server->sock;
 
 # We're testing to ensure item chaining doesn't corrupt or poorly overlap
diff --git a/t/extstore-buckets.t b/t/extstore-buckets.t
new file mode 100644
index 0000000..f6143af
--- /dev/null
+++ b/t/extstore-buckets.t
@@ -0,0 +1,61 @@
+#!/usr/bin/perl
+
+use strict;
+use warnings;
+use Test::More;
+use FindBin qw($Bin);
+use lib "$Bin/lib";
+use MemcachedTest;
+use Data::Dumper qw/Dumper/;
+
+my $ext_path;
+
+if (!supports_extstore()) {
+    plan skip_all => 'extstore not enabled';
+    exit 0;
+}
+
+$ext_path = "/tmp/extstore.$$";
+
+my $server = new_memcached("-m 64 -U 0 -o ext_page_size=8,ext_page_count=8,ext_wbuf_size=2,ext_threads=1,ext_io_depth=2,ext_item_size=512,ext_item_age=2,ext_recache_rate=10000,ext_max_frag=0,ext_path=$ext_path,ext_low_ttl=60");
+my $sock = $server->sock;
+
+my $value;
+{
+    my @chars = ("C".."Z");
+    for (1 .. 20000) {
+        $value .= $chars[rand @chars];
+    }
+}
+
+# fill some larger objects
+{
+    # interleave sets with 0 ttl vs long ttl's.
+    my $keycount = 1200;
+    for (1 .. $keycount) {
+        print $sock "set nfoo$_ 0 0 20000 noreply\r\n$value\r\n";
+        print $sock "set lfoo$_ 0 5 20000 noreply\r\n$value\r\n";
+    }
+    # wait for a flush
+    sleep 10;
+    print $sock "lru_crawler crawl all\r\n";
+    <$sock>;
+    sleep 2;
+    # fetch
+    mem_get_is($sock, "nfoo1", $value);
+    # check extstore counters
+    my $stats = mem_stats($sock);
+    cmp_ok($stats->{extstore_page_allocs}, '>', 0, 'at least one page allocated');
+    cmp_ok($stats->{extstore_objects_written}, '>', $keycount / 2, 'some objects written');
+    cmp_ok($stats->{extstore_bytes_written}, '>', length($value) * 2, 'some bytes written');
+    cmp_ok($stats->{get_extstore}, '>', 0, 'one object was fetched');
+    cmp_ok($stats->{extstore_objects_read}, '>', 0, 'one object read');
+    cmp_ok($stats->{extstore_bytes_read}, '>', length($value), 'some bytes read');
+    cmp_ok($stats->{extstore_page_reclaims}, '>', 1, 'at least two pages reclaimed');
+}
+
+done_testing();
+
+END {
+    unlink $ext_path if $ext_path;
+}
diff --git a/t/extstore.t b/t/extstore.t
index 74038af..8a8180e 100644
--- a/t/extstore.t
+++ b/t/extstore.t
@@ -17,7 +17,7 @@ if (!supports_extstore()) {
 
 $ext_path = "/tmp/extstore.$$";
 
-my $server = new_memcached("-m 64 -U 0 -o ext_page_size=8,ext_page_count=8,ext_wbuf_size=2,ext_wbuf_count=3,ext_threads=1,ext_io_depth=2,ext_item_size=512,ext_item_age=2,ext_recache_rate=10000,ext_max_frag=0.9,ext_path=$ext_path");
+my $server = new_memcached("-m 64 -U 0 -o ext_page_size=8,ext_page_count=8,ext_wbuf_size=2,ext_threads=1,ext_io_depth=2,ext_item_size=512,ext_item_age=2,ext_recache_rate=10000,ext_max_frag=0.9,ext_path=$ext_path");
 my $sock = $server->sock;
 
 my $value;