restartable cache

"-e /path/to/tmpfsmnt/file"
SIGUSR1 for graceful stop

restart requires the same memory limit, slab sizes, and some other
infrequently changed details. Most other options and features can
change between restarts. Binary can be upgraded between restarts.

Restart does some fixup work on start for every item in cache. Can take
over a minute with more than a few hundred million items in cache.

Keep in mind when a cache is down it may be missing invalidations,
updates, and so on.

Author: dormando

Makefile.am

@@ -24,7 +24,8 @@ memcached_SOURCES = memcached.c memcached.h \
                     crawler.c crawler.h \
                     itoa_ljust.c itoa_ljust.h \
                     slab_automove.c slab_automove.h \
-                    authfile.c authfile.h
+                    authfile.c authfile.h \
+                    restart.c restart.h
 
 if BUILD_CACHE
 memcached_SOURCES += cache.c

assoc.c

@@ -255,11 +255,14 @@ static void *assoc_maintenance_thread(void *arg) {
              * allow dynamic hash table expansion without causing significant
              * wait times.
              */
-            pause_threads(PAUSE_ALL_THREADS);
-            assoc_expand();
-            pause_threads(RESUME_ALL_THREADS);
+            if (do_run_maintenance_thread) {
+                pause_threads(PAUSE_ALL_THREADS);
+                assoc_expand();
+                pause_threads(RESUME_ALL_THREADS);
+            }
         }
     }
+    mutex_unlock(&maintenance_lock);
     return NULL;
 }
 

configure.ac

@@ -613,6 +613,7 @@ fi
 
 AC_CHECK_FUNCS(mlockall)
 AC_CHECK_FUNCS(getpagesizes)
+AC_CHECK_FUNCS(sysconf)
 AC_CHECK_FUNCS(memcntl)
 AC_CHECK_FUNCS(sigignore)
 AC_CHECK_FUNCS(clock_gettime)

crawler.c

@@ -460,23 +460,27 @@ static void *item_crawler_thread(void *arg) {
     pthread_mutex_unlock(&lru_crawler_lock);
     if (settings.verbose > 2)
         fprintf(stderr, "LRU crawler thread stopping\n");
+    settings.lru_crawler = false;
 
     return NULL;
 }
 
 static pthread_t item_crawler_tid;
 
-int stop_item_crawler_thread(void) {
+int stop_item_crawler_thread(bool wait) {
     int ret;
     pthread_mutex_lock(&lru_crawler_lock);
+    if (do_run_lru_crawler_thread == 0) {
+        pthread_mutex_unlock(&lru_crawler_lock);
+        return 0;
+    }
     do_run_lru_crawler_thread = 0;
     pthread_cond_signal(&lru_crawler_cond);
     pthread_mutex_unlock(&lru_crawler_lock);
-    if ((ret = pthread_join(item_crawler_tid, NULL)) != 0) {
+    if (wait && (ret = pthread_join(item_crawler_tid, NULL)) != 0) {
         fprintf(stderr, "Failed to stop LRU crawler thread: %s\n", strerror(ret));
         return -1;
     }
-    settings.lru_crawler = false;
     return 0;
 }
 
@@ -584,6 +588,11 @@ int lru_crawler_start(uint8_t *ids, uint32_t remaining,
     STATS_LOCK();
     is_running = stats_state.lru_crawler_running;
     STATS_UNLOCK();
+    if (do_run_lru_crawler_thread == 0) {
+        pthread_mutex_unlock(&lru_crawler_lock);
+        return -2;
+    }
+
     if (is_running &&
             !(type == CRAWLER_AUTOEXPIRE && active_crawler_type == CRAWLER_AUTOEXPIRE)) {
         pthread_mutex_unlock(&lru_crawler_lock);

crawler.h

@@ -28,7 +28,9 @@ enum crawler_result_type {
     CRAWLER_OK=0, CRAWLER_RUNNING, CRAWLER_BADCLASS, CRAWLER_NOTSTARTED, CRAWLER_ERROR
 };
 int start_item_crawler_thread(void);
-int stop_item_crawler_thread(void);
+#define CRAWLER_WAIT true
+#define CRAWLER_NOWAIT false
+int stop_item_crawler_thread(bool wait);
 int init_lru_crawler(void *arg);
 enum crawler_result_type lru_crawler_crawl(char *slabs, enum crawler_run_type,
         void *c, const int sfd, unsigned int remaining);

items.c

@@ -60,6 +60,7 @@ static uint64_t sizes_bytes[LARGEST_ID];
 static unsigned int *stats_sizes_hist = NULL;
 static uint64_t stats_sizes_cas_min = 0;
 static int stats_sizes_buckets = 0;
+static uint64_t cas_id = 0;
 
 static volatile int do_run_lru_maintainer_thread = 0;
 static int lru_maintainer_initialized = 0;
@@ -109,13 +110,18 @@ static uint64_t lru_total_bumps_dropped(void);
 /* Get the next CAS id for a new item. */
 /* TODO: refactor some atomics for this. */
 uint64_t get_cas_id(void) {
-    static uint64_t cas_id = 0;
     pthread_mutex_lock(&cas_id_lock);
     uint64_t next_id = ++cas_id;
     pthread_mutex_unlock(&cas_id_lock);
     return next_id;
 }
 
+void set_cas_id(uint64_t new_cas) {
+    pthread_mutex_lock(&cas_id_lock);
+    cas_id = new_cas;
+    pthread_mutex_unlock(&cas_id_lock);
+}
+
 int item_is_flushed(item *it) {
     rel_time_t oldest_live = settings.oldest_live;
     uint64_t cas = ITEM_get_cas(it);
@@ -293,7 +299,7 @@ item *do_item_alloc(char *key, const size_t nkey, const unsigned int flags,
         return NULL;
     }
 
-    assert(it->slabs_clsid == 0);
+    assert(it->it_flags == 0 || it->it_flags == ITEM_CHUNKED);
     //assert(it != heads[id]);
 
     /* Refcount is seeded to 1 by slabs_alloc() */
@@ -373,6 +379,31 @@ bool item_size_ok(const size_t nkey, const int flags, const int nbytes) {
     return slabs_clsid(ntotal) != 0;
 }
 
+/* fixing stats/references during warm start */
+void do_item_link_fixup(item *it) {
+    item **head, **tail;
+    int ntotal = ITEM_ntotal(it);
+    uint32_t hv = hash(ITEM_key(it), it->nkey);
+    assoc_insert(it, hv);
+
+    head = &heads[it->slabs_clsid];
+    tail = &tails[it->slabs_clsid];
+    if (it->prev == 0 && *head == 0) *head = it;
+    if (it->next == 0 && *tail == 0) *tail = it;
+    sizes[it->slabs_clsid]++;
+    sizes_bytes[it->slabs_clsid] += ntotal;
+
+    STATS_LOCK();
+    stats_state.curr_bytes += ntotal;
+    stats_state.curr_items += 1;
+    stats.total_items += 1;
+    STATS_UNLOCK();
+
+    item_stats_sizes_add(it);
+
+    return;
+}
+
 static void do_item_link_q(item *it) { /* item is the new head */
     item **head, **tail;
     assert((it->it_flags & ITEM_SLABBED) == 0);

items.h

@@ -8,6 +8,7 @@
 
 /* See items.c */
 uint64_t get_cas_id(void);
+void set_cas_id(uint64_t new_cas);
 
 /*@null@*/
 item *do_item_alloc(char *key, const size_t nkey, const unsigned int flags, const rel_time_t exptime, const int nbytes);
@@ -23,6 +24,7 @@ void do_item_remove(item *it);
 void do_item_update(item *it);   /** update LRU time to current and reposition */
 void do_item_update_nolock(item *it);
 int  do_item_replace(item *it, item *new_it, const uint32_t hv);
+void do_item_link_fixup(item *it);
 
 int item_is_flushed(item *it);
 unsigned int do_get_lru_size(uint32_t id);

logger.c

@@ -88,8 +88,8 @@ static int logger_thread_poll_watchers(int force_poll, int watcher);
 
 /* Logger GID's can be used by watchers to put logs back into strict order
  */
-static uint64_t logger_get_gid(void) {
-    static uint64_t logger_gid = 0;
+static uint64_t logger_gid = 0;
+uint64_t logger_get_gid(void) {
 #ifdef HAVE_GCC_64ATOMICS
     return __sync_add_and_fetch(&logger_gid, 1);
 #elif defined(__sun)
@@ -102,6 +102,18 @@ static uint64_t logger_get_gid(void) {
 #endif
 }
 
+void logger_set_gid(uint64_t gid) {
+#ifdef HAVE_GCC_64ATOMICS
+    __sync_add_and_fetch(&logger_gid, gid);
+#elif defined(__sun)
+    atomic_add_64(&logger_gid);
+#else
+    mutex_lock(&logger_atomics_mutex);
+    logger_gid = gid;
+    mutex_unlock(&logger_atomics_mutex);
+#endif
+}
+
 /* TODO: genericize lists. would be nice to import queue.h if the impact is
  * studied... otherwise can just write a local one.
  */
@@ -505,6 +517,8 @@ static void logger_thread_sum_stats(struct logger_stats *ls) {
 static void *logger_thread(void *arg) {
     useconds_t to_sleep = MIN_LOGGER_SLEEP;
     L_DEBUG("LOGGER: Starting logger thread\n");
+    // TODO: If we ever have item references in the logger code, will need to
+    // ensure everything is dequeued before stopping the thread.
     while (do_run_logger_thread) {
         int found_logs = 0;
         logger *l;
@@ -553,12 +567,11 @@ static int start_logger_thread(void) {
     return 0;
 }
 
-// future.
-/*static int stop_logger_thread(void) {
+static int stop_logger_thread(void) {
     do_run_logger_thread = 0;
     pthread_join(logger_tid, NULL);
     return 0;
-}*/
+}
 
 /*************************
  * Public functions for submitting logs and starting loggers from workers.
@@ -578,19 +591,16 @@ void logger_init(void) {
         abort();
     }
 
-    /* This can be removed once the global stats initializer is improved */
-    STATS_LOCK();
-    stats.log_worker_dropped = 0;
-    stats.log_worker_written = 0;
-    stats.log_watcher_skipped = 0;
-    stats.log_watcher_sent = 0;
-    STATS_UNLOCK();
     /* This is what adding a STDERR watcher looks like. should replace old
      * "verbose" settings. */
     //logger_add_watcher(NULL, 0);
     return;
 }
 
+void logger_stop(void) {
+    stop_logger_thread();
+}
+
 /* called *from* the thread using a logger.
  * initializes the per-thread bipbuf, links it into the list of loggers
  */

logger.h

@@ -100,6 +100,9 @@ struct logentry_item_store {
 
 /* end intermediary structures */
 
+/* WARNING: cuddled items aren't compatible with warm restart. more code
+ * necessary to ensure log streams are all flushed/processed before stopping
+ */
 typedef struct _logentry {
     enum log_entry_subtype event;
     uint8_t pad;
@@ -108,7 +111,6 @@ typedef struct _logentry {
     struct timeval tv; /* not monotonic! */
     int size;
     union {
-        void *entry; /* probably an item */
         char end;
     } data[];
 } logentry;
@@ -165,6 +167,7 @@ extern pthread_key_t logger_key;
 /* public functions */
 
 void logger_init(void);
+void logger_stop(void);
 logger *logger_create(void);
 
 #define LOGGER_LOG(l, flag, type, ...) \
@@ -186,4 +189,8 @@ enum logger_add_watcher_ret {
 
 enum logger_add_watcher_ret logger_add_watcher(void *c, const int sfd, uint16_t f);
 
+/* functions used by restart system */
+uint64_t logger_get_gid(void);
+void logger_set_gid(uint64_t gid);
+
 #endif