redis protocol: tokenizer, hash and zset

.travis.yml

@@ -72,7 +72,12 @@ matrix:
 
     # clang 3.7 on osx
     - os: osx
-      osx_image: xcode7.1
+      osx_image: xcode7.3
+      env: C_COMPILER=clang
+
+    # clang 4.2 on osx
+    - os: osx
+      osx_image: xcode8.2
       env: C_COMPILER=clang
 
 
@@ -93,4 +98,5 @@ script:
   - cmake ..
   - make -j
   - make check
+  - egrep -r ":F:|:E:" . || true
   - cd ../test/integration && python test_twemcache.py

deps/ccommon/.travis.yml

@@ -75,6 +75,11 @@ matrix:
       osx_image: xcode7.1
       env: C_COMPILER=clang
 
+    # clang 4.2 on osx
+    - os: osx
+      osx_image: xcode8.2
+      env: C_COMPILER=clang
+
 
 before_install:
   # for osx: 0. update brew; 1. install cmake if missing; 2. (gcc) unlink pre-installed gcc; 3. (gcc) install desired version of gcc
@@ -93,3 +98,4 @@ script:
   - cmake ..
   - make -j
   - make check
+  - egrep -r ":F:|:E:" . || true

deps/ccommon/include/cc_print.h

@@ -44,7 +44,10 @@ extern "C" {
 
 /* behavior undefined if there isn't enough space in buf */
 size_t cc_print_uint64_unsafe(char *buf, uint64_t n);
+size_t cc_print_int64_unsafe(char *buf, int64_t n);
+
 size_t cc_print_uint64(char *buf, size_t size, uint64_t n);
+size_t cc_print_int64(char *buf, size_t size, int64_t n);
 
 size_t _scnprintf(char *buf, size_t size, const char *fmt, ...);
 size_t _vscnprintf(char *buf, size_t size, const char *fmt, va_list args);

deps/ccommon/include/cc_util.h

@@ -57,13 +57,17 @@ extern "C" {
  * # define UINT16_MAX  (65535)
  * # define UINT32_MAX  (4294967295U)
  * # define UINT64_MAX  (__UINT64_C(18446744073709551615))
+ *
+ * # define INT64_MIN   -9223372036854775808LL
  */
 #define CC_UINT8_MAXLEN     (3 + 1)
 #define CC_UINT16_MAXLEN    (5 + 1)
 #define CC_UINT32_MAXLEN    (10 + 1)
 #define CC_UINT64_MAXLEN    (20 + 1)
 #define CC_UINTMAX_MAXLEN   CC_UINT64_MAXLEN
 
+#define CC_INT64_MAXLEN    (1 + 19 + 1)
+
 /* alignment */
 /* Make data 'd' or pointer 'p', n-byte aligned, where n is a power of 2 */
 #define CC_ALIGNMENT        sizeof(unsigned long) /* platform word */

deps/ccommon/src/cc_print.c

@@ -22,6 +22,9 @@
  * implementation as a reference (folly/Conv.h)
  */
 
+/* use our own macro instead of llabs() to make sure it works with INT64_MIN */
+#define abs_int64(_x) ((_x) >= 0 ? (_x) : -(_x))
+
 static inline void
 _print_uint64(char *buf, size_t d, uint64_t n)
 {
@@ -46,6 +49,22 @@ cc_print_uint64_unsafe(char *buf, uint64_t n)
     return d;
 }
 
+size_t
+cc_print_int64_unsafe(char *buf, int64_t n)
+{
+    size_t d;
+    uint64_t ab = abs_int64(n);
+
+    if (n < 0) {
+        *buf++ = '-';
+    }
+
+    d = digits(ab);
+    _print_uint64(buf, d, n);
+
+    return (n < 0) ? d + 1 : d;
+}
+
 size_t
 cc_print_uint64(char *buf, size_t size, uint64_t n)
 {
@@ -61,6 +80,27 @@ cc_print_uint64(char *buf, size_t size, uint64_t n)
     return d;
 }
 
+size_t
+cc_print_int64(char *buf, size_t size, int64_t n)
+{
+    size_t d;
+    uint64_t ab = abs_int64(n);
+
+    d =  digits(ab) + (n < 0);
+    if (size < d) {
+        return 0;
+    }
+
+    if (n < 0) {
+        *buf++ = '-';
+        d--;
+    }
+
+    _print_uint64(buf, d, n);
+
+    return (n < 0) ? d + 1 : d;
+}
+
 size_t
 _vscnprintf(char *buf, size_t size, const char *fmt, va_list args)
 {

deps/ccommon/test/time/wheel/check_wheel.c

@@ -116,7 +116,7 @@ END_TEST
 
 START_TEST(test_timing_wheel_recur)
 {
-#define TICK_NS 10000000
+#define TICK_NS 50000000
 #define NSLOT 3
 #define NTICK 2
 
@@ -136,11 +136,13 @@ START_TEST(test_timing_wheel_recur)
 
     timing_wheel_insert(tw, &delay, true, _incr_cb, &i);
 
-    nanosleep(&ts, NULL);
+    /* tick unchanged */
     timing_wheel_execute(tw);
     ck_assert_int_eq(tw->nprocess, 0);
     ck_assert_int_eq(tw->nevent, 1);
 
+    /* next 2 tick */
+    nanosleep(&ts, NULL);
     nanosleep(&ts, NULL);
     timing_wheel_execute(tw);
     ck_assert_int_eq(tw->nevent, 1);
@@ -152,6 +154,7 @@ START_TEST(test_timing_wheel_recur)
     ck_assert_int_eq(tw->nprocess, 2);
     ck_assert_int_eq(i, 2);
 
+    /* flush events */
     timing_wheel_stop(tw);
     timing_wheel_flush(tw);
     ck_assert_int_eq(tw->nevent, 0);

src/protocol/data/CMakeLists.txt

@@ -1,2 +1,3 @@
 add_subdirectory(memcache)
 add_subdirectory(ping)
+add_subdirectory(redis)

src/protocol/data/memcache/compose.c

@@ -70,7 +70,7 @@ _write_uint64(struct buf **buf, uint64_t val)
     struct buf *b;
 
 
-    /* NOTE(yao): here we are being conservative on how many bytes wee need
+    /* NOTE(yao): here we are being conservative on how many bytes we need
      * to print a (64-bit) integer. The actual number might be smaller.
      * But since it is 21 bytes at most (including \0' while buffers usually
      * are KBs in size, it is unlikely to cause many extra expansions.

src/protocol/data/redis/CMakeLists.txt

@@ -0,0 +1,8 @@
+set(SOURCE
+    compose.c
+    parse.c
+    request.c
+    response.c
+    token.c)
+
+add_library(protocol_redis ${SOURCE})

src/protocol/data/redis/cmd_hash.h

@@ -0,0 +1,28 @@
+#pragma once
+
+/*
+ * Note: negative # of arguments means variable number of arguments:
+ *   e.g. `-2' means at least two arguments. This notation is inherited from
+ *   the original Redis server implementation.
+ */
+
+/*          type                string          # of args */
+#define REQ_HASH(ACTION)                            \
+    ACTION( REQ_HDEL,           "hdel",         -3 )\
+    ACTION( REQ_HDELALL,        "hdelall",      2  )\
+    ACTION( REQ_HEXISTS,        "hexists",      3  )\
+    ACTION( REQ_HGET,           "hget",         3  )\
+    ACTION( REQ_HGETALL,        "hgetall",      2  )\
+    ACTION( REQ_HINCRBY,        "hincrby",      4  )\
+    ACTION( REQ_HINCRBYFLOAT,   "hincrbyfloat", 4  )\
+    ACTION( REQ_HKEYS,          "hkeys",        2  )\
+    ACTION( REQ_HLEN,           "hlen",         2  )\
+    ACTION( REQ_HMGET,          "hmget",        -3 )\
+    ACTION( REQ_HMSET,          "hmset",        -4 )\
+    ACTION( REQ_HSET,           "hset",         4  )\
+    ACTION( REQ_HSETNX,         "hsetnx",       4  )\
+    ACTION( REQ_HSTRLEN,        "hstrlen",      3  )\
+    ACTION( REQ_HVALS,          "hvals",        2  )\
+    ACTION( REQ_HSCAN,          "hscan",        -3 )
+
+/* "hlen KEY" == "*2\r\n$4\r\nhlen\r\n$3\r\nKEY\r\n" */

src/protocol/data/redis/cmd_misc.h

@@ -0,0 +1,6 @@
+#pragma once
+
+/*          type            string      # of args */
+#define REQ_MISC(ACTION)                    \
+    ACTION( REQ_PING,       "ping",     -1 )\
+    ACTION( REQ_QUIT,       "quit",     1  )

src/protocol/data/redis/cmd_zset.h

@@ -0,0 +1,25 @@
+#pragma once
+
+/*          type                    string              # of args */
+#define REQ_ZSET(ACTION)                                    \
+    ACTION( REQ_ZADD,               "zadd",             -4 )\
+    ACTION( REQ_ZINCRBY,            "zincrby",          4  )\
+    ACTION( REQ_ZREM,               "zrem",             -3 )\
+    ACTION( REQ_ZREMRANGEBYSCORE,   "zremrangebyscore", 4  )\
+    ACTION( REQ_ZREMRANGEBYRANK,    "zremrangebyrank",  4  )\
+    ACTION( REQ_ZREMRANGEBYLEX,     "zremrangebylex",   4  )\
+    ACTION( REQ_ZUNIONSTORE,        "zunionstore",      -4 )\
+    ACTION( REQ_ZINTERSTORE,        "zinterstore",      -4 )\
+    ACTION( REQ_ZRANGE,             "zrange",           -4 )\
+    ACTION( REQ_ZRANGEBYSCORE,      "zrangebyscore",    -4 )\
+    ACTION( REQ_ZREVRANGEBYSCORE,   "zrevrangebyscore", -4 )\
+    ACTION( REQ_ZRANGEBYLEX,        "zrangebylex",      -4 )\
+    ACTION( REQ_ZREVRANGEBYLEX,     "zrevrangebylex",   -4 )\
+    ACTION( REQ_ZCOUNT,             "zcount",           4  )\
+    ACTION( REQ_ZLEXCOUNT,          "zlexcount",        4  )\
+    ACTION( REQ_ZREVRANGE,          "zrevrange",        -4 )\
+    ACTION( REQ_ZCARD,              "zcard",            2  )\
+    ACTION( REQ_ZSCORE,             "zscore",           3  )\
+    ACTION( REQ_ZRANK,              "zrank",            3  )\
+    ACTION( REQ_ZREVRANK,           "zrevrank",         3  )\
+    ACTION( REQ_ZSCAN,              "zscan",            -3 )

src/protocol/data/redis/compose.c

@@ -0,0 +1,92 @@
+#include "compose.h"
+
+#include "request.h"
+#include "response.h"
+#include "token.h"
+
+#include <cc_debug.h>
+#include <cc_print.h>
+
+#define COMPOSE_MODULE_NAME "protocol::redis::compose"
+
+static bool compose_init = false;
+static compose_req_metrics_st *compose_req_metrics = NULL;
+static compose_rsp_metrics_st *compose_rsp_metrics = NULL;
+
+void
+compose_setup(compose_req_metrics_st *req, compose_rsp_metrics_st *rsp)
+{
+    log_info("set up the %s module", COMPOSE_MODULE_NAME);
+
+    if (compose_init) {
+        log_warn("%s has already been setup, overwrite", COMPOSE_MODULE_NAME);
+    }
+
+    compose_req_metrics = req;
+    compose_rsp_metrics = rsp;
+
+    compose_init = true;
+}
+
+void
+compose_teardown(void)
+{
+    log_info("tear down the %s module", COMPOSE_MODULE_NAME);
+
+    if (!compose_init) {
+        log_warn("%s has never been setup", COMPOSE_MODULE_NAME);
+    }
+    compose_req_metrics = NULL;
+    compose_rsp_metrics = NULL;
+    compose_init = false;
+}
+
+int
+compose_req(struct buf **buf, struct request *req)
+{
+    int n;
+
+    n = compose_array_header(buf, req->token->nelem);
+    if (n < 0) {
+        return n;
+    }
+
+    for (int i = 0; i < req->token->nelem; i++) {
+        int ret;
+
+        ret = compose_element(buf, array_get(req->token, i));
+        if (ret < 0) {
+            return ret;
+        } else {
+            n += ret;
+        }
+    }
+
+    return n;
+}
+
+int
+compose_rsp(struct buf **buf, struct response *rsp)
+{
+    int n = 0;
+
+    if (rsp->type == ELEM_ARRAY) {
+        n = compose_array_header(buf, rsp->token->nelem);
+        if (n < 0) {
+            return n;
+        }
+    }
+
+    for (int i = 0; i < rsp->token->nelem; i++) {
+        int ret;
+
+        ret = compose_element(buf, array_get(rsp->token, i));
+        if (ret < 0) {
+            return ret;
+        } else {
+            n += ret;
+        }
+    }
+
+    return n;
+}

src/protocol/data/redis/compose.h

@@ -0,0 +1,44 @@
+#pragma once
+
+#include <buffer/cc_dbuf.h>
+#include <cc_define.h>
+#include <cc_metric.h>
+
+#include <stdint.h>
+
+/*          name                    Type            description */
+#define COMPOSE_REQ_METRIC(ACTION)                                          \
+    ACTION( request_compose,        METRIC_COUNTER, "# requests composed"  )\
+    ACTION( request_compose_ex,     METRIC_COUNTER, "# composing error"    )
+
+/*          name                    Type            description */
+#define COMPOSE_RSP_METRIC(ACTION)                                          \
+    ACTION( response_compose,       METRIC_COUNTER, "# responses composed" )\
+    ACTION( response_compose_ex,    METRIC_COUNTER, "# rsp composing error")
+
+typedef struct {
+    COMPOSE_REQ_METRIC(METRIC_DECLARE)
+} compose_req_metrics_st;
+
+typedef struct {
+    COMPOSE_RSP_METRIC(METRIC_DECLARE)
+} compose_rsp_metrics_st;
+
+typedef enum compose_rstatus {
+    COMPOSE_OK          = 0,
+    COMPOSE_EUNFIN      = -1,
+    COMPOSE_ENOMEM      = -2,
+    COMPOSE_EINVALID    = -3,
+    COMPOSE_EOTHER      = -4,
+} compose_rstatus_t;
+
+struct request;
+struct response;
+
+void compose_setup(compose_req_metrics_st *req, compose_rsp_metrics_st *rsp);
+void compose_teardown(void);
+
+/* if the return value is negative, it can be interpreted as compose_rstatus */
+int compose_req(struct buf **buf, struct request *req);
+
+int compose_rsp(struct buf **buf, struct response *rsp);