Merge pull request #717 from Netflix/dev

Merge dev into 0.6

Author: smukil

src/dyn_connection.c

@@ -234,6 +234,13 @@ rstatus_t conn_listen(struct context *ctx, struct conn *p) {
     return DN_ERROR;
   }
 
+  status = dn_set_keepalive(p->sd, 1);
+  if (status != DN_OK) {
+      log_error("set keepalive on p %d on addr '%.*s' failed: %s", p->sd,
+                p->pname.len, p->pname.data, strerror(errno));
+      // Continue since this is not catastrophic
+  }
+
   status = conn_event_add_conn(p);
   if (status < 0) {
     log_error("event add conn p %d on addr '%.*s' failed: %s", p->sd,

src/dyn_core.c

@@ -50,7 +50,8 @@ static void core_print_peer_status(void *arg1) {
       struct rack *rack = array_get(&dc->racks, rack_index);
       uint8_t i = 0;
       for (i = 0; i < rack->ncontinuum; i++) {
-        struct continuum *c = &rack->continuum[i];
+        struct continuum *c = (struct continuum*) array_get(&rack->continuums, i);
+        ASSERT(c != NULL);
         uint32_t peer_index = c->index;
         struct node *peer = *(struct node **)array_get(&sp->peers, peer_index);
         if (!peer) log_panic("peer is null. Topology not inited proerly");

src/dyn_core.h

@@ -126,7 +126,7 @@ struct instance {
 
 struct continuum {
   uint32_t index;          /* dyn_peer index */
-  uint32_t value;          /* hash value, used by ketama */
+  uint32_t value;          /* hash value, used ONLY by ketama */
   struct dyn_token *token; /* used in vnode/dyn_token situations */
 };
 
@@ -136,7 +136,7 @@ struct rack {
   uint32_t ncontinuum; /* # continuum points */
   uint32_t
       nserver_continuum; /* # servers - live and dead on continuum (const) */
-  struct continuum *continuum; /* continuum */
+  struct array continuums;
 };
 
 struct datacenter {

src/dyn_dnode_peer.c

@@ -785,7 +785,7 @@ uint32_t dnode_peer_idx_for_key_on_rack(struct server_pool *pool,
                                         uint32_t keylen) {
   struct dyn_token token;
   pool->key_hash(key, keylen, &token);
-  return vnode_dispatch(rack->continuum, rack->ncontinuum, &token);
+  return vnode_dispatch(&rack->continuums, rack->ncontinuum, &token);
 }
 
 static struct node *dnode_peer_for_key_on_rack(struct server_pool *pool,

src/dyn_message.c

@@ -632,6 +632,11 @@ void msg_put(struct msg *msg) {
     msg->keys = NULL;
   }
 
+  if (msg->args) {
+    array_destroy(msg->args);
+    msg->args = NULL;
+  }
+
   if (msg->orig_msg) {
     msg_put(msg->orig_msg);
     msg->orig_msg = NULL;
@@ -881,8 +886,10 @@ static rstatus_t msg_repair(struct context *ctx, struct conn *conn,
     return DN_ENOMEM;
   }
 
-  mbuf = STAILQ_LAST(&msg->mhdr, mbuf, next);
-  mbuf_remove(&msg->mhdr, mbuf);
+  // This was added to handle a specific case which doesn't seem reproducible
+  // now. Revisit if things seem off.
+  //mbuf = STAILQ_LAST(&msg->mhdr, mbuf, next);
+  //mbuf_remove(&msg->mhdr, mbuf);
   mbuf_insert(&msg->mhdr, nbuf);
   msg->pos = nbuf->pos;
 

src/dyn_server.c

@@ -508,7 +508,9 @@ dictType dc_string_dict_type = {
 };
 
 static rstatus_t rack_init(struct rack *rack) {
-  rack->continuum = dn_alloc(sizeof(struct continuum));
+
+  // TODO: Initialize the array to the size of the ring instead of to 1.
+  THROW_STATUS(array_init(&rack->continuums, 1, sizeof(struct continuum)));
   rack->ncontinuum = 0;
   rack->nserver_continuum = 0;
   rack->name = dn_alloc(sizeof(struct string));
@@ -521,9 +523,7 @@ static rstatus_t rack_init(struct rack *rack) {
 }
 
 static rstatus_t rack_deinit(struct rack *rack) {
-  if (rack->continuum != NULL) {
-    dn_free(rack->continuum);
-  }
+  array_deinit(&rack->continuums);
 
   return DN_OK;
 }

src/dyn_vnode.c

@@ -29,22 +29,26 @@
 #include <dyn_server.h>
 #include <dyn_vnode.h>
 
+
+// Similar to strcmp() but compares 2 'dyn_token' structs instead.
 static int vnode_item_cmp(const void *t1, const void *t2) {
   const struct continuum *ct1 = t1, *ct2 = t2;
 
   return cmp_dyn_token(ct1->token, ct2->token);
 }
 
+// Sorts the continuum for a rack based on their tokens.
 static rstatus_t vnode_rack_verify_continuum(void *elem) {
   struct rack *rack = elem;
-  qsort(rack->continuum, rack->ncontinuum, sizeof(*rack->continuum),
+  qsort(rack->continuums.elem, rack->ncontinuum, sizeof(struct continuum),
         vnode_item_cmp);
 
   log_debug(LOG_VERB, "**** printing continuums for rack '%.*s'",
             rack->name->len, rack->name->data);
   uint32_t i;
   for (i = 0; i < rack->ncontinuum; i++) {
-    struct continuum *c = &rack->continuum[i];
+    struct continuum *c = (struct continuum*) array_get(&rack->continuums, i);
+    ASSERT(c != NULL);
     log_debug(LOG_VERB, "next c[%d]: idx = %u, token->mag = %u", i, c->index,
               c->token->mag[0]);
   }
@@ -88,21 +92,18 @@ rstatus_t vnode_update(struct server_pool *sp) {
     uint32_t orig_cnt = rack->nserver_continuum;
     uint32_t new_cnt = orig_cnt + token_cnt;
 
-    if (new_cnt > 1) {
-      struct continuum *continuum =
-          dn_realloc(rack->continuum, sizeof(struct continuum) * new_cnt);
-      if (continuum == NULL) {
-        log_debug(LOG_ERR, "Are we failing? Why???? This is a serious issue");
-        return DN_ENOMEM;
-      }
-
-      rack->continuum = continuum;
+    struct continuum *continuum = array_push(&rack->continuums);
+    if (continuum == NULL) {
+      log_error("Could not allocate memory to expand the continuum.");
+      return DN_ENOMEM;
     }
     rack->nserver_continuum = new_cnt;
 
     uint32_t j;
     for (j = 0; j < token_cnt; j++) {
-      struct continuum *c = &rack->continuum[orig_cnt + j];
+      struct continuum *c = (struct continuum*) array_get(
+          &rack->continuums, orig_cnt + j);
+      ASSERT(c != NULL);
       c->index = i;
       c->value = 0; /* set this to an empty value, only used by ketama */
       c->token = array_get(&peer->tokens, j);
@@ -120,16 +121,15 @@ rstatus_t vnode_update(struct server_pool *sp) {
   return DN_OK;
 }
 
-// if token falls into interval (a,b], we return b.
-uint32_t vnode_dispatch(struct continuum *continuum, uint32_t ncontinuum,
+uint32_t vnode_dispatch(struct array *continuums, uint32_t ncontinuum,
                         struct dyn_token *token) {
   struct continuum *left, *right, *middle;
 
-  ASSERT(continuum != NULL);
+  ASSERT(continuums != NULL);
   ASSERT(ncontinuum != 0);
 
-  left = continuum;
-  right = continuum + ncontinuum - 1;
+  left = (struct continuum*) array_get(continuums, 0);
+  right = (struct continuum*) array_get(continuums, ncontinuum - 1);
 
   if (cmp_dyn_token(right->token, token) < 0 ||
       cmp_dyn_token(left->token, token) >= 0)

src/dyn_vnode.h

@@ -1,6 +1,16 @@
+/*
+ * Dynomite - A thin, distributed replication layer for multi non-distributed
+ * storages. Copyright (C) 2019 Netflix, Inc.
+ */
+
 #pragma once
 #include <dyn_types.h>
 
+// Initializes (on first call) and updates (on subsequent calls) the per rack continuums
+// and makes sure the tokens managed by the continuums are ascending.
 rstatus_t vnode_update(struct server_pool *pool);
-uint32_t vnode_dispatch(struct continuum *continuum, uint32_t ncontinuum,
+
+// Returns the index of the continuum from 'continuums' where 'token' falls.
+// If 'token' falls into interval (a,b], we return b.
+uint32_t vnode_dispatch(struct array *continuums, uint32_t ncontinuum,
                         struct dyn_token *token);

src/proto/dyn_proto_repair.h

@@ -77,7 +77,6 @@
  * compile time constants.
  *
  * This is still in the beta stage and has some limitations:
- * - Large values may cause overflows ( > 512 bytes)
  * - Limited command support due to parser limitations:
  *     - Sorted sets and lists need optional command parsing support.
  * - Repairing on the read path vs. the background has perf implications. In the future,
@@ -290,6 +289,71 @@
   "local value = redis.call(orig_cmd, key, field)\n"\
   "return {status_field, ts, value}\n\r\n"
 
+#define ZADD_SCRIPT "$4\r\nEVAL\r\n$2022\r\n"\
+  "local key = KEYS[1]\n"\
+  "local top_level_add_set = KEYS[2]\n"\
+  "local top_level_rem_set = KEYS[3]\n"\
+  "local add_set = top_level_add_set .. '_' .. key\n"\
+  "local rem_set = top_level_rem_set .. '_' .. key\n"\
+  "local orig_cmd = ARGV[1]\n"\
+  "local num_opts = ARGV[2]\n"\
+  "local num_fields = ARGV[3]\n"\
+  "local cur_ts = ARGV[4]\n"\
+  "local start_loop = 5 + num_opts\n"\
+  "local end_loop = (num_fields * 2) + 4 + num_opts\n"\
+  "local top_level_rem_set_ts = redis.call('ZSCORE', top_level_rem_set, key)\n"\
+  "if (top_level_rem_set_ts) then\n"\
+  "  if (tonumber(cur_ts) < tonumber(top_level_rem_set_ts)) then\n"\
+  "    return 0\n"\
+  "  end\n"\
+  "  redis.call('ZREM', top_level_rem_set, key)\n"\
+  "end\n"\
+  "local top_level_add_set_ts = redis.call('ZSCORE', top_level_add_set, key)\n"\
+  "if (top_level_add_set_ts) then\n"\
+  "  if (tonumber(cur_ts) > tonumber(top_level_add_set_ts)) then\n"\
+  "    redis.call('ZADD', top_level_add_set, cur_ts, key)\n"\
+  "  end\n"\
+  "else\n"\
+  "  redis.call('ZADD', top_level_add_set, cur_ts, key)\n"\
+  "end\n"\
+  "local skiploop\n"\
+  "local ret\n"\
+  "for i=start_loop,end_loop,2\n"\
+  "do\n"\
+  "  skiploop = false\n"\
+  "  local field = ARGV[i]\n"\
+  "  local value = ARGV[i+1]\n"\
+  "  local last_seen_ts_in_add = redis.call('ZSCORE', add_set, field)\n"\
+  "  local last_seen_ts_in_rem = redis.call('ZSCORE', rem_set, field)\n"\
+  "  if (last_seen_ts_in_rem) then\n"\
+  "    if (tonumber(cur_ts) < tonumber(last_seen_ts_in_rem)) then\n"\
+  "      skiploop = true\n"\
+  "    end\n"\
+  "    redis.call('ZREM', rem_set, field)\n"\
+  "  elseif (last_seen_ts_in_add) then\n"\
+  "    if (tonumber(cur_ts) < tonumber(last_seen_ts_in_add)) then\n"\
+  "      skiploop = true\n"\
+  "    end\n"\
+  "  end\n"\
+  "  if (skiploop == false) then\n"\
+  "    if (num_opts == '0') then\n"\
+  "      ret = redis.call(orig_cmd, key, value, field)\n"\
+  "    elseif (num_opts == '1') then\n"\
+  "      ret = redis.call(orig_cmd, key, ARGV[5], value, field)\n"\
+  "    elseif (num_opts == '2') then\n"\
+  "      ret = redis.call(orig_cmd, key, ARGV[5], ARGV[6], value, field)\n"\
+  "    elseif (num_opts == '3') then\n"\
+  "      ret = redis.call(orig_cmd, key, ARGV[5], ARGV[6], ARGV[7], value, field)\n"\
+  "    else\n"\
+  "      ret = false\n"\
+  "    end\n"\
+  "    if (type(ret) ~= 'boolean') then\n"\
+  "      redis.call('ZADD', add_set, cur_ts, field)\n"\
+  "    end\n"\
+  "  end\n"\
+  "end\n"\
+  "return ret\n\r\n"
+
 #define SADD_SCRIPT "$4\r\nEVAL\r\n$1526\r\n"\
   "local key = KEYS[1]\n"\
   "local top_level_add_set = KEYS[2]\n"\

src/proto/dyn_redis.c

@@ -1065,6 +1065,7 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
               // This is not to be confused with 'EXISTS'. This is the second half of the
               // command 'SCRIPT EXISTS'.
               r->type = MSG_REQ_REDIS_SCRIPT_EXISTS;
+              r->msg_routing = ROUTING_ALL_NODES_ALL_RACKS_ALL_DCS;
               r->is_read = 1;
               break;
             }
@@ -1829,7 +1830,8 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
 
             m = next_mbuf->pos + new_mbuf_offset;
             b = next_mbuf;
-            ++r->latest_parsed_mbuf_idx;
+
+            if (mbuf_full(b)) ++r->latest_parsed_mbuf_idx;
           }
           if (arg1_across_mbufs == false) {
             rstatus_t argstatus = record_arg(p , m , r->args);
@@ -1846,6 +1848,8 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
           r->rlen -= (uint32_t)(b->last - p);
           m = b->last - 1;
           p = m;
+
+          if (mbuf_full(b)) ++r->latest_parsed_mbuf_idx;
           break;
         }
 
@@ -1967,7 +1971,7 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
 
             m = next_mbuf->pos + new_mbuf_offset;
             b = next_mbuf;
-            ++r->latest_parsed_mbuf_idx;
+            if (mbuf_full(b)) ++r->latest_parsed_mbuf_idx;
           }
           if (arg2_across_mbufs == false) {
             // TODO: Verify if this is the correct behavior for EVAL/EVALSHA
@@ -1985,6 +1989,8 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
           r->rlen -= (uint32_t)(b->last - p);
           m = b->last - 1;
           p = m;
+
+          if (mbuf_full(b)) ++r->latest_parsed_mbuf_idx;
           break;
         }
 
@@ -2120,7 +2126,7 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
 
             m = next_mbuf->pos + new_mbuf_offset;
             b = next_mbuf;
-            ++r->latest_parsed_mbuf_idx;
+            if (mbuf_full(b)) ++r->latest_parsed_mbuf_idx;
           }
           if (arg3_across_mbufs == false) {
             rstatus_t argstatus = record_arg(p , m , r->args);
@@ -2137,6 +2143,8 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
           r->rlen -= (uint32_t)(b->last - p);
           m = b->last - 1;
           p = m;
+
+          if (mbuf_full(b)) ++r->latest_parsed_mbuf_idx;
           break;
         }
 
@@ -2227,7 +2235,7 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
 
             m = next_mbuf->pos + new_mbuf_offset;
             b = next_mbuf;
-            ++r->latest_parsed_mbuf_idx;
+            if (mbuf_full(b)) ++r->latest_parsed_mbuf_idx;
           }
           if (argn_across_mbufs == false) {
             rstatus_t argstatus = record_arg(p , m , r->args);
@@ -2244,6 +2252,8 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
           r->rlen -= (uint32_t)(b->last - p);
           m = b->last - 1;
           p = m;
+
+          if (mbuf_full(b)) ++r->latest_parsed_mbuf_idx;
           break;
         }
 
@@ -2288,15 +2298,13 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
   r->pos = p;
   r->state = state;
 
-  // We reached here since we finished parsing the current 'mbuf' in 'r->mhdr'.
-  ++r->latest_parsed_mbuf_idx;
-
   // If we have to parse again, we won't be able to write with the timestamp.
   r->rewrite_with_ts_possible = false;
   if (b->last == b->end && r->token != NULL) {
     r->pos = r->token;
     r->token = NULL;
     r->result = MSG_PARSE_REPAIR;
+
   } else {
     r->result = MSG_PARSE_AGAIN;
   }