add support for hash tags

conf/nutcracker.yml

@@ -12,6 +12,7 @@ alpha:
 beta:
   listen: 127.0.0.1:22122
   hash: fnv1a_64
+  hash_tag: "{}"
   distribution: ketama
   auto_eject_hosts: false
   timeout: 400

notes/recommendation.md

@@ -76,7 +76,7 @@ The memcache ascii protocol [specification](https://github.com/twitter/twemproxy
 
 ## Node Names for Consistent Hashing
 
-The server cluster in twemproxy can either be specified as list strings in format 'host:port:weight' or 'host:port:weight name'. 
+The server cluster in twemproxy can either be specified as list strings in format 'host:port:weight' or 'host:port:weight name'.
 
     servers:
      - 127.0.0.1:6379:1
@@ -96,3 +96,23 @@ Or,
 In the former configuration, keys are mapped **directly** to **'host:port:weight'** triplet and in the latter they are mapped to **node names** which are then mapped to nodes i.e. host:port pair. The latter configuration gives us the freedom to relocate nodes to a different server without disturbing the hash ring and hence makes this configuration ideal when auto_eject_hosts is set to false. See [issue 25](https://github.com/twitter/twemproxy/issues/25) for details.
 
 Note that when using node names for consistent hashing, twemproxy ignores the weight value in the 'host:port:weight name' format string.
+
+## Hash Tags
+
+[Hash Tags](http://antirez.com/post/redis-presharding.html) enables you to use part of the key for calculating the hash. When the hash tag is present, we use part of the key within the tag as the key to be used for consistent hashing. Otherwise, we use the full key as is. Hash tags enable you to map different keys to the same server as long as the part of the key within the tag is the same.
+
+For example, the configuration of server pool _beta_, aslo shown below, specifies a two character hash_tag string - "{}". This means that keys "user:{user1}:ids" and "user:{user1}:tweets" map to the same server because we compute the hash on "user1". For a key like "user:user1:ids", we use the entire string "user:user1:ids" to compute the hash and it may map to a different server.
+
+    beta:
+      listen: 127.0.0.1:22122
+      hash: fnv1a_64
+      hash_tag: "{}"
+      distribution: ketama
+      auto_eject_hosts: false
+      timeout: 400
+      redis: true
+      servers:
+       - 127.0.0.1:6380:1 server1
+       - 127.0.0.1:6381:1 server2
+       - 127.0.0.1:6382:1 server3
+       - 127.0.0.1:6383:1 server4

src/nc_conf.c

@@ -50,6 +50,10 @@ static struct command conf_commands[] = {
       conf_set_hash,
       offsetof(struct conf_pool, hash) },
 
+    { string("hash_tag"),
+      conf_set_hashtag,
+      offsetof(struct conf_pool, hash_tag) },
+
     { string("distribution"),
       conf_set_distribution,
       offsetof(struct conf_pool, distribution) },
@@ -171,7 +175,9 @@ conf_pool_init(struct conf_pool *cp, struct string *name)
     cp->listen.valid = 0;
 
     cp->hash = CONF_UNSET_HASH;
+    string_init(&cp->hash_tag);
     cp->distribution = CONF_UNSET_DIST;
+
     cp->timeout = CONF_UNSET_NUM;
     cp->backlog = CONF_UNSET_NUM;
 
@@ -256,10 +262,11 @@ conf_pool_each_transform(void *elem, void *data)
     sp->addrlen = cp->listen.info.addrlen;
     sp->addr = (struct sockaddr *)&cp->listen.info.addr;
 
-    sp->dist_type = cp->distribution;
-
     sp->key_hash_type = cp->hash;
     sp->key_hash = hash_algos[cp->hash];
+    sp->dist_type = cp->distribution;
+    sp->hash_tag = cp->hash_tag;
+
     sp->redis = cp->redis ? 1 : 0;
     sp->timeout = cp->timeout;
     sp->backlog = cp->backlog;
@@ -304,9 +311,11 @@ conf_dump(struct conf *cf)
         log_debug(LOG_VVERB, "%.*s", cp->name.len, cp->name.data);
         log_debug(LOG_VVERB, "  listen: %.*s",
                   cp->listen.pname.len, cp->listen.pname.data);
-        log_debug(LOG_VVERB, "  hash: %d", cp->hash);
         log_debug(LOG_VVERB, "  timeout: %d", cp->timeout);
         log_debug(LOG_VVERB, "  backlog: %d", cp->backlog);
+        log_debug(LOG_VVERB, "  hash: %d", cp->hash);
+        log_debug(LOG_VVERB, "  hash_tag: \"%.*s\"", cp->hash_tag.len,
+                  cp->hash_tag.data);
         log_debug(LOG_VVERB, "  distribution: %d", cp->distribution);
         log_debug(LOG_VVERB, "  client_connections: %d",
                   cp->client_connections);
@@ -1727,3 +1736,31 @@ conf_set_distribution(struct conf *cf, struct command *cmd, void *conf)
 
     return "is not a valid distribution";
 }
+
+char *
+conf_set_hashtag(struct conf *cf, struct command *cmd, void *conf)
+{
+    rstatus_t status;
+    uint8_t *p;
+    struct string *field, *value;
+
+    p = conf;
+    field = (struct string *)(p + cmd->offset);
+
+    if (field->data != CONF_UNSET_PTR) {
+        return "is a duplicate";
+    }
+
+    value = array_top(&cf->arg);
+
+    if (value->len != 2) {
+        return "is not a valid hash tag string with two characters";
+    }
+
+    status = string_duplicate(field, value);
+    if (status != NC_OK) {
+        return CONF_ERROR;
+    }
+
+    return CONF_OK;
+}

src/nc_conf.h

@@ -75,6 +75,7 @@ struct conf_pool {
     struct string      name;                  /* pool name (root node) */
     struct conf_listen listen;                /* listen: */
     hash_type_t        hash;                  /* hash: */
+    struct string      hash_tag;              /* hash_tag: */
     dist_type_t        distribution;          /* distribution: */
     int                timeout;               /* timeout: */
     int                backlog;               /* backlog: */
@@ -122,6 +123,7 @@ char *conf_set_num(struct conf *cf, struct command *cmd, void *conf);
 char * conf_set_bool(struct conf *cf, struct command *cmd, void *conf);
 char *conf_set_hash(struct conf *cf, struct command *cmd, void *conf);
 char *conf_set_distribution(struct conf *cf, struct command *cmd, void *conf);
+char *conf_set_hashtag(struct conf *cf, struct command *cmd, void *conf);
 
 rstatus_t conf_server_each_transform(void *elem, void *data);
 rstatus_t conf_pool_each_transform(void *elem, void *data);

src/nc_request.c

@@ -434,6 +434,7 @@ req_forward(struct context *ctx, struct conn *c_conn, struct msg *msg)
 {
     rstatus_t status;
     struct conn *s_conn;
+    struct server_pool *pool;
     uint8_t *key;
     uint32_t keylen;
 
@@ -444,8 +445,33 @@ req_forward(struct context *ctx, struct conn *c_conn, struct msg *msg)
         c_conn->enqueue_outq(ctx, c_conn, msg);
     }
 
-    key = msg->key_start;
-    keylen = (uint32_t)(msg->key_end - msg->key_start);
+    pool = c_conn->owner;
+    key = NULL;
+    keylen = 0;
+
+    /*
+     * If hash_tag: is configured for this server pool, we use the part of
+     * the key within the hash tag as an input to the distributor. Otherwise
+     * we use the full key
+     */
+    if (!string_empty(&pool->hash_tag)) {
+        struct string *tag = &pool->hash_tag;
+        uint8_t *tag_start, *tag_end;
+
+        tag_start = nc_strchr(msg->key_start, msg->key_end, tag->data[0]);
+        if (tag_start != NULL) {
+            tag_end = nc_strchr(tag_start + 1, msg->key_end, tag->data[1]);
+            if (tag_end != NULL) {
+                key = tag_start + 1;
+                keylen = (uint32_t)(tag_end - key);
+            }
+        }
+    }
+
+    if (keylen == 0) {
+        key = msg->key_start;
+        keylen = (uint32_t)(msg->key_end - msg->key_start);
+    }
 
     s_conn = server_pool_conn(ctx, c_conn->owner, key, keylen);
     if (s_conn == NULL) {

src/nc_server.h

@@ -109,6 +109,7 @@ struct server_pool {
     int                dist_type;            /* distribution type (dist_type_t) */
     int                key_hash_type;        /* key hash type (hash_type_t) */
     hash_t             key_hash;             /* key hasher */
+    struct string      hash_tag;             /* key hash tag (ref in conf_pool) */
     int                timeout;              /* timeout in msec */
     int                backlog;              /* listen backlog */
     uint32_t           client_connections;   /* maximum # client connection */