Merge pull request #123 from kevyang/121

Add data structures, interface, and basic unit tests for hotkey sampl…

src/CMakeLists.txt

@@ -1,4 +1,5 @@
 add_subdirectory(core ${PROJECT_BINARY_DIR}/core)
+add_subdirectory(hotkey ${PROJECT_BINARY_DIR}/hotkey)
 add_subdirectory(protocol ${PROJECT_BINARY_DIR}/protocol)
 add_subdirectory(storage ${PROJECT_BINARY_DIR}/storage)
 add_subdirectory(time ${PROJECT_BINARY_DIR}/time)

src/hotkey/CMakeLists.txt

@@ -0,0 +1,6 @@
+set(SOURCE
+    hotkey.c
+    kc_map.c
+    key_window.c)
+
+add_library(hotkey ${SOURCE})

src/hotkey/constant.h

@@ -0,0 +1,3 @@
+#pragma once
+
+#define MAX_KEY_LEN 250

src/hotkey/hotkey.c

@@ -0,0 +1,83 @@
+#include "hotkey.h"
+
+#include "constant.h"
+#include "kc_map.h"
+#include "key_window.h"
+
+#include <cc_bstring.h>
+#include <cc_debug.h>
+
+#define HOTKEY_MODULE_NAME "hotkey::hotkey"
+
+bool hotkey_enabled = false;
+
+static uint64_t hotkey_counter;
+
+static bool hotkey_init = false;
+static uint32_t hotkey_window_size = HOTKEY_WINDOW_SIZE;
+static uint32_t hotkey_rate = HOTKEY_RATE;
+static uint32_t hotkey_threshold = HOTKEY_THRESHOLD;
+static uint32_t hotkey_window_size_cur = 0;
+
+void
+hotkey_setup(hotkey_options_st *options)
+{
+    log_info("Set up the %s module", HOTKEY_MODULE_NAME);
+
+    if (options != NULL) {
+        hotkey_enabled = option_bool(&options->hotkey_enable);
+        hotkey_window_size = option_uint(&options->hotkey_sample_size);
+        hotkey_rate = option_uint(&options->hotkey_sample_rate);
+        hotkey_threshold = (uint32_t)(option_fpn(&options->hotkey_threshold_ratio) * hotkey_window_size);
+    }
+
+    hotkey_window_size_cur = 0;
+    hotkey_counter = 0;
+    key_window_setup(hotkey_window_size);
+    /* TODO: determine whether table size should be a tuneable parameter */
+    kc_map_setup(hotkey_window_size, hotkey_window_size);
+    hotkey_init = true;
+}
+
+void
+hotkey_teardown(void)
+{
+    log_info("Tear down the %s module", HOTKEY_MODULE_NAME);
+
+    if (!hotkey_init) {
+        log_warn("%s was not setup", HOTKEY_MODULE_NAME);
+        return;
+    }
+
+    hotkey_enabled = false;
+    key_window_teardown();
+    kc_map_teardown();
+    hotkey_init = false;
+}
+
+bool
+hotkey_sample(const struct bstring *key)
+{
+    if (++hotkey_counter % hotkey_rate == 0) {
+        /* sample this key */
+        uint32_t freq;
+
+        if (key_window_len() == hotkey_window_size) {
+            char buf[MAX_KEY_LEN];
+            struct bstring popped;
+
+            popped.data = buf;
+
+            /* pop from key_window, decrement in counter table */
+            popped.len = key_window_pop(popped.data);
+            kc_map_decr(&popped);
+        }
+
+        key_window_push(key);
+        freq = kc_map_incr(key);
+
+        return freq >= hotkey_threshold;
+    }
+
+    return false;
+}

src/hotkey/hotkey.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include <cc_option.h>
+
+#include <stdbool.h>
+#include <stdint.h>
+
+/* TODO(kevyang): add stats for hotkey module */
+
+#define HOTKEY_WINDOW_SIZE     10000 /* keep last 10000 keys sampled by default */
+#define HOTKEY_RATE            100   /* sample one in every 100 keys by default */
+#define HOTKEY_THRESHOLD_RATIO 0.01  /* signal hotkey if key takes up >= 0.01 of all keys by default */
+#define HOTKEY_THRESHOLD       (uint32_t)(HOTKEY_THRESHOLD_RATIO * HOTKEY_WINDOW_SIZE)
+
+/*          name                    type                default                 description */
+#define HOTKEY_OPTION(ACTION)                                                                                 \
+    ACTION( hotkey_enable,          OPTION_TYPE_BOOL,   false,                  "use hotkey detection?"      )\
+    ACTION( hotkey_sample_size,     OPTION_TYPE_UINT,   HOTKEY_WINDOW_SIZE,     "number of keys to maintain" )\
+    ACTION( hotkey_sample_rate,     OPTION_TYPE_UINT,   HOTKEY_RATE,            "hotkey sample ratio"        )\
+    ACTION( hotkey_threshold_ratio, OPTION_TYPE_UINT,   HOTKEY_THRESHOLD_RATIO, "threshold for hotkey signal")
+
+typedef struct {
+    HOTKEY_OPTION(OPTION_DECLARE)
+} hotkey_options_st;
+
+extern bool hotkey_enabled;
+
+void hotkey_setup(hotkey_options_st *options);
+void hotkey_teardown(void);
+bool hotkey_sample(const struct bstring *key);

src/hotkey/kc_map.c

@@ -0,0 +1,244 @@
+#include "kc_map.h"
+
+#include "constant.h"
+
+#include <cc_bstring.h>
+#include <cc_debug.h>
+#include <cc_hash.h>
+#include <cc_mm.h>
+#include <cc_pool.h>
+
+#define KC_MAP_MODULE_NAME "hotkey::kc_map"
+
+struct kc_map_entry {
+    STAILQ_ENTRY(kc_map_entry) next; /* entry in hash table or pool */
+
+    char                       key[MAX_KEY_LEN];
+    uint32_t                   klen;
+    uint32_t                   count;
+};
+
+STAILQ_HEAD(kcme_slh, kc_map_entry);
+
+static struct kcme_slh *table = NULL;
+static uint32_t table_size = 0; /* number of buckets in table */
+static bool kc_map_init = false;
+
+FREEPOOL(kcme_pool, kcmeq, kc_map_entry);
+static struct kcme_pool kcmep;
+static bool kcmep_init = false;
+
+static void
+kc_map_entry_reset(struct kc_map_entry *kcme)
+{
+    kcme->klen = 0;
+    kcme->count = 0;
+}
+
+static struct kc_map_entry *
+kc_map_entry_create(void)
+{
+    struct kc_map_entry *kcme = cc_alloc(sizeof(*kcme));
+
+    if (kcme == NULL) {
+        return NULL;
+    }
+
+    kc_map_entry_reset(kcme);
+
+    return kcme;
+}
+
+static void
+kc_map_entry_destroy(struct kc_map_entry **kc_map_entry)
+{
+    struct kc_map_entry *kcme = *kc_map_entry;
+    ASSERT(kcme != NULL);
+
+    cc_free(kcme);
+    *kc_map_entry = NULL;
+}
+
+static void
+kc_map_entry_pool_destroy(void)
+{
+    struct kc_map_entry *kcme, *tkcme;
+
+    if (!kcmep_init) {
+        log_warn("kc_map_entry pool was not created, ignore");
+        return;
+    }
+
+    log_info("destroying kc_map_entry pool: free %"PRIu32, kcmep.nfree);
+
+    FREEPOOL_DESTROY(kcme, tkcme, &kcmep, next, kc_map_entry_destroy);
+    kcmep_init = false;
+}
+
+static struct kc_map_entry *
+kc_map_entry_borrow(void)
+{
+    struct kc_map_entry *kcme;
+
+    FREEPOOL_BORROW(kcme, &kcmep, next, kc_map_entry_create);
+    if (kcme == NULL) {
+        log_debug("borrow kc_map_entry failed: OOM");
+        return NULL;
+    }
+    kc_map_entry_reset(kcme);
+
+    return kcme;
+}
+
+static void
+kc_map_entry_return(struct kc_map_entry **kc_map_entry)
+{
+    struct kc_map_entry *kcme = *kc_map_entry;
+
+    if (kcme == NULL) {
+        return;
+    }
+
+    FREEPOOL_RETURN(kcme, &kcmep, next);
+
+    *kc_map_entry = NULL;
+}
+
+static void
+kc_map_entry_pool_create(uint32_t max)
+{
+    struct kc_map_entry *kcme;
+
+    if (kcmep_init) {
+        log_warn("kc_map_entry pool has already been created, re-creating");
+        kc_map_entry_pool_destroy();
+    }
+
+    log_info("creating kc_map_entry pool: max %"PRIu32, max);
+
+    FREEPOOL_CREATE(&kcmep, max);
+    kcmep_init = true;
+
+    FREEPOOL_PREALLOC(kcme, &kcmep, max, next, kc_map_entry_create);
+    if (kcmep.nfree < max) {
+        log_crit("cannot preallocate kc_map_entry pool, OOM. abort");
+        exit(EXIT_FAILURE);
+    }
+}
+
+void
+kc_map_setup(uint32_t size, uint32_t poolsize)
+{
+    uint32_t i;
+
+    log_info("Set up the %s module", KC_MAP_MODULE_NAME);
+
+    if (kc_map_init) {
+        log_warn("%s has already been setup, ignore", KC_MAP_MODULE_NAME);
+        return;
+    }
+
+    table = cc_alloc(sizeof(*table) * size);
+    table_size = size;
+
+    if (table == NULL) {
+        log_crit("Could not allocate counter table for hotkey - OOM");
+        exit(EXIT_FAILURE);
+    }
+
+    for (i = 0; i < size; ++i) {
+        STAILQ_INIT(&table[i]);
+    }
+
+    kc_map_entry_pool_create(poolsize);
+}
+
+void
+kc_map_teardown(void)
+{
+    log_info("Tear down the %s module", KC_MAP_MODULE_NAME);
+
+    if (!kc_map_init) {
+        log_warn("%s was not setup", KC_MAP_MODULE_NAME);
+    }
+
+    if (table != NULL) {
+        uint32_t i;
+        /* free all entries in table */
+        for (i = 0; i < table_size; ++i) {
+            struct kc_map_entry *kcme, *tkcme;
+            STAILQ_FOREACH_SAFE(kcme, &(table[i]), next, tkcme) {
+                kc_map_entry_return(&kcme);
+            }
+        }
+    }
+
+    kc_map_entry_pool_destroy();
+    kc_map_init = false;
+}
+
+static inline struct kcme_slh *
+_get_bucket(const struct bstring *key)
+{
+    return &(table[hash(key->data, key->len, 0) % table_size]);
+}
+
+uint32_t
+kc_map_incr(const struct bstring *key)
+{
+    struct kc_map_entry *kcme;
+    struct kcme_slh *bucket;
+
+    ASSERT(key->len <= MAX_KEY_LEN);
+
+    bucket = _get_bucket(key);
+
+    /* iterate through bucket looking for item */
+    for (kcme = STAILQ_FIRST(bucket); kcme != NULL; kcme = STAILQ_NEXT(kcme, next)) {
+        if ((key->len == kcme->klen) && cc_memcmp(key->data, kcme->key, key->len) == 0) {
+            /* found item */
+            return ++kcme->count;
+        }
+    }
+
+    /* not found, insert entry */
+    kcme = kc_map_entry_borrow();
+    cc_memcpy(kcme->key, key->data, key->len);
+    kcme->klen = key->len;
+    kcme->count = 1;
+    STAILQ_INSERT_HEAD(bucket, kcme, next);
+
+    return 1;
+}
+
+void
+kc_map_decr(const struct bstring *key)
+{
+    struct kc_map_entry *kcme, *prev;
+    struct kcme_slh *bucket;
+
+    ASSERT(key->len <= MAX_KEY_LEN);
+
+    bucket = _get_bucket(key);
+
+    /* iterate through bucket looking for item */
+    for (prev = NULL, kcme = STAILQ_FIRST(bucket); kcme != NULL;
+         prev = kcme, kcme = STAILQ_NEXT(kcme, next)) {
+        if ((key->len == kcme->klen) && cc_memcmp(key->data, kcme->key, key->len) == 0) {
+            /* found item */
+            if (--(kcme->count) == 0) {
+                /* remove entry */
+                if (prev == NULL) {
+                    STAILQ_REMOVE_HEAD(bucket, next);
+                } else {
+                    STAILQ_REMOVE_AFTER(bucket, prev, next);
+                }
+            }
+
+            return;
+        }
+    }
+
+    /* item not found, should never happen for hotkey detection scheme */
+    ASSERT(false);
+}

src/hotkey/kc_map.h

@@ -0,0 +1,22 @@
+#pragma once
+
+#include <stdint.h>
+
+/*
+ * The kc_map module provides a utility for counting the frequency at which keys
+ * appear. When a key is sampled, it should be incremented in the table after
+ * enqueueing it. When a key is dequeued, it should be decremented from the
+ * table.
+ */
+
+struct bstring;
+
+/* setup/teardown module */
+void kc_map_setup(uint32_t size, uint32_t poolsize);
+void kc_map_teardown(void);
+
+/* increment count and return count, insert into table if count == 0 */
+uint32_t kc_map_incr(const struct bstring *key);
+
+/* decrement count, remove from table if count == 0 after decr */
+void kc_map_decr(const struct bstring *key);