Merge pull request #720 from Netflix/dev

[Beta] Add DC_EACH_SAFE_QUORUM consistency level

Author: smukil

src/dyn_client.c

@@ -50,6 +50,8 @@
 #include "dyn_util.h"
 
 static rstatus_t msg_quorum_rsp_handler(struct context *ctx, struct msg *req, struct msg *rsp);
+static rstatus_t msg_each_quorum_rsp_handler(struct context *ctx, struct msg *req,
+    struct msg *rsp);
 static msg_response_handler_t msg_get_rsp_handler(struct context *ctx, struct msg *req);
 
 static rstatus_t rewrite_query_if_necessary(struct msg **req,
@@ -653,8 +655,10 @@ rstatus_t req_forward_to_peer(struct context *ctx, struct conn *c_conn,
   }
 
   if (!(same_dc && same_rack) || force_swallow) {
-    // Swallow responses from remote racks or DCs.
-    rack_msg->swallow = true;
+    if (req->consistency != DC_EACH_SAFE_QUORUM || force_swallow) {
+      // Swallow responses from remote racks or DCs.
+      rack_msg->swallow = true;
+    }
   }
 
   // Get a connection to the node.
@@ -701,13 +705,20 @@ rstatus_t req_forward_to_peer(struct context *ctx, struct conn *c_conn,
   return status;
 }
 
-void req_forward_all_local_racks(struct context *ctx, struct conn *c_conn,
+void req_forward_all_racks_for_dc(struct context *ctx, struct conn *c_conn,
                                  struct msg *req, struct mbuf *orig_mbuf,
                                  uint8_t *key, uint32_t keylen,
                                  struct datacenter *dc) {
   uint8_t rack_cnt = (uint8_t)array_n(&dc->racks);
   uint8_t rack_index;
-  init_response_mgr(&req->rspmgr, req, req->is_read, rack_cnt, c_conn);
+
+  if (req->rspmgrs_inited == false) {
+    if (req->consistency == DC_EACH_SAFE_QUORUM) {
+      init_response_mgr_all_dcs(ctx, req, c_conn, dc);
+    } else {
+      init_response_mgr(req, &req->rspmgr, rack_cnt, c_conn);
+    }
+  }
   log_info("%s %s same DC racks:%d expect replies %d", print_obj(c_conn),
            print_obj(req), rack_cnt, req->rspmgr.max_responses);
 
@@ -732,6 +743,10 @@ static bool request_send_to_all_dcs(struct msg *req) {
   // There is a routing override
   if (req->msg_routing != ROUTING_NORMAL) return false;
 
+  // Under DC_EACH_SAFE_QUORUM, we need to send reads and writes to all
+  // DCs.
+  if (req->consistency == DC_EACH_SAFE_QUORUM) return true;
+
   // Reads are not propagated
   if (req->is_read) return false;
 
@@ -753,8 +768,11 @@ static bool request_send_to_all_local_racks(struct msg *req) {
   // A write should go to all racks
   if (!req->is_read) return true;
 
-  if ((req->consistency == DC_QUORUM) || (req->consistency == DC_SAFE_QUORUM))
+  if ((req->consistency == DC_QUORUM)
+    || (req->consistency == DC_SAFE_QUORUM)
+    || (req->consistency == DC_EACH_SAFE_QUORUM)) {
     return true;
+  }
   return false;
 }
 
@@ -794,6 +812,17 @@ static void req_forward_remote_dc(struct context *ctx, struct conn *c_conn,
   const uint32_t rack_cnt = array_n(&dc->racks);
   if (rack_cnt == 0) return;
 
+  if (req->consistency == DC_EACH_SAFE_QUORUM) {
+    // Under 'DC_EACH_SAFE_QUORUM', we want to hear back from at least
+    // quorum racks in each DC, so send it to all racks in remote DCs.
+    req_forward_all_racks_for_dc(ctx, c_conn, req, orig_mbuf, key, keylen, dc);
+    return;
+  }
+
+  // If we're not expecting a consistency level of 'DC_EACH_SAFE_QUORUM', then
+  // we send it to only to the preselected rack in the remote DCs. If that's not
+  // reachable, we failover to another in the remote DC.
+
   // Pick the preferred pre-selected rack for this DC.
   struct rack *rack = dc->preselected_rack_for_replication;
   if (rack == NULL) rack = array_get(&dc->racks, 0);
@@ -846,7 +875,7 @@ static void req_forward_local_dc(struct context *ctx, struct conn *c_conn,
   req->rsp_handler = msg_get_rsp_handler(ctx, req);
   if (request_send_to_all_local_racks(req)) {
     // send request to all local racks
-    req_forward_all_local_racks(ctx, c_conn, req, orig_mbuf, key, keylen, dc);
+    req_forward_all_racks_for_dc(ctx, c_conn, req, orig_mbuf, key, keylen, dc);
   } else {
     // send request to only local token owner
     struct rack *rack =
@@ -1002,7 +1031,7 @@ rstatus_t rewrite_query_if_necessary(struct msg **req, struct context *ctx) {
  * the new query and free up the original msg.
  *
  */
-rstatus_t rewrite_query_with_timestamp_md(struct msg **req, struct context *ctx) {
+static rstatus_t rewrite_query_with_timestamp_md(struct msg **req, struct context *ctx) {
 
   if (is_read_repairs_enabled() == false) return DN_OK;
 
@@ -1099,6 +1128,8 @@ static msg_response_handler_t msg_get_rsp_handler(struct context *ctx, struct ms
     // Check if its quorum
     if ((req->consistency == DC_QUORUM) || (req->consistency == DC_SAFE_QUORUM)) {
       return msg_quorum_rsp_handler;
+    } else if (req->consistency == DC_EACH_SAFE_QUORUM) {
+      return msg_each_quorum_rsp_handler;
     }
   }
 
@@ -1149,6 +1180,104 @@ static rstatus_t msg_quorum_rsp_handler(struct context *ctx, struct msg *req,
   return DN_OK;
 }
 
+static int find_rspmgr_idx(struct context *ctx, struct response_mgr **rspmgrs,
+    struct string *target_dc_name) {
+  int num_dcs = (int) array_n(&ctx->pool.datacenters);
+
+  int i = 0;
+  for (i = 0; i < num_dcs; ++i) {
+    struct response_mgr *rspmgr = rspmgrs[i];
+    if (string_compare(&rspmgr->dc_name, target_dc_name) == 0) {
+      return i;
+    }
+  }
+  return -1;
+}
+
+static bool all_rspmgrs_done(struct context *ctx, struct response_mgr **rspmgrs) {
+  int num_dcs = (int) array_n(&ctx->pool.datacenters);
+  int i = 0;
+  for (i = 0; i < num_dcs; ++i) {
+    struct response_mgr *rspmgr = rspmgrs[i];
+    if (!rspmgr->done) return false;
+  }
+
+  return true;
+}
+
+static struct msg *all_rspmgrs_get_response(struct context *ctx, struct msg *req) {
+  int num_dcs = (int) array_n(&ctx->pool.datacenters);
+  struct msg *rsp = NULL;
+  int i;
+  for (i = 0; i < num_dcs; ++i) {
+    struct response_mgr *rspmgr = req->additional_each_rspmgrs[i];
+    struct msg *dc_rsp = NULL;
+    if (!rsp) {
+      rsp = rspmgr_get_response(ctx, rspmgr);
+      ASSERT(rsp);
+    } else if (rsp->is_error) {
+      // If any of the DCs errored out, we just clean up responses from the
+      // remaining DCs.
+      rspmgr_free_other_responses(rspmgr, NULL);
+      continue;
+    } else {
+      ASSERT(rsp->is_error == false);
+      // If the DCs we've processed so far have not seen errors, we need to
+      // make sure that the remaining DCs don't have errors too.
+      dc_rsp = rspmgr_get_response(ctx, rspmgr);
+      ASSERT(dc_rsp);
+      if (dc_rsp->is_error) {
+        rsp_put(rsp);
+        rsp = dc_rsp;
+      } else {
+        // If it's not an error, clear all responses from this DC.
+        rspmgr_free_other_responses(rspmgr, NULL);
+        continue;
+      }
+    }
+
+    rspmgr_free_other_responses(rspmgr, rsp);
+    rsp->peer = req;
+    req->selected_rsp = rsp;
+    req->error_code = rsp->error_code;
+    req->is_error = rsp->is_error;
+    req->dyn_error_code = rsp->dyn_error_code;
+
+  }
+
+  return rsp;
+}
+
+static rstatus_t msg_each_quorum_rsp_handler(struct context *ctx, struct msg *req,
+    struct msg *rsp) {
+
+  if (all_rspmgrs_done(ctx, req->additional_each_rspmgrs)) return swallow_extra_rsp(req, rsp);
+
+  int rspmgr_idx = -1;
+  struct conn *rsp_conn = rsp->owner;
+  if (rsp_conn == NULL) {
+    rspmgr_idx = 0;
+  } else if (rsp_conn->type == CONN_DNODE_PEER_SERVER) {
+    struct node *peer_instance = (struct node*) rsp_conn->owner;
+    struct string *peer_dc_name = &peer_instance->dc;
+    rspmgr_idx = find_rspmgr_idx(ctx, req->additional_each_rspmgrs, peer_dc_name);
+    if (rspmgr_idx == -1) {
+      log_error("Could not find which DC response was from");
+    }
+  } else if (rsp_conn->type == CONN_SERVER) {
+    // If this is a 'CONN_SERVER' connection, then it is from the same DC.
+    rspmgr_idx = 0;
+  }
+
+  struct response_mgr *rspmgr = req->additional_each_rspmgrs[rspmgr_idx];
+  rspmgr_submit_response(rspmgr, rsp);
+  if (!rspmgr_check_is_done(rspmgr)) return DN_EAGAIN;
+  if (!all_rspmgrs_done(ctx, req->additional_each_rspmgrs)) return DN_EAGAIN;
+
+  rsp = all_rspmgrs_get_response(ctx, req);
+  return DN_OK;
+}
+
 static void req_client_enqueue_omsgq(struct context *ctx, struct conn *conn,
                                      struct msg *req) {
   ASSERT(req->is_request);

src/dyn_conf.c

@@ -2010,10 +2010,12 @@ static rstatus_t conf_validate_pool(struct conf *cf, struct conf_pool *cp) {
     g_read_consistency = DC_SAFE_QUORUM;
   else if (!dn_strcasecmp(cp->read_consistency.data, CONF_STR_DC_QUORUM))
     g_read_consistency = DC_QUORUM;
+  else if (!dn_strcasecmp(cp->read_consistency.data, CONF_STR_DC_EACH_SAFE_QUORUM))
+    g_read_consistency = DC_EACH_SAFE_QUORUM;
   else {
     log_error(
         "conf: directive \"read_consistency:\"must be one of 'DC_ONE' "
-        "'DC_QUORUM' 'DC_SAFE_QUORUM'");
+        "'DC_QUORUM' 'DC_SAFE_QUORUM' 'DC_EACH_SAFE_QUORUM'");
     return DN_ERROR;
   }
 
@@ -2023,10 +2025,12 @@ static rstatus_t conf_validate_pool(struct conf *cf, struct conf_pool *cp) {
     g_write_consistency = DC_SAFE_QUORUM;
   else if (!dn_strcasecmp(cp->write_consistency.data, CONF_STR_DC_QUORUM))
     g_write_consistency = DC_QUORUM;
+  else if (!dn_strcasecmp(cp->write_consistency.data, CONF_STR_DC_EACH_SAFE_QUORUM))
+    g_write_consistency = DC_EACH_SAFE_QUORUM;
   else {
     log_error(
         "conf: directive \"write_consistency:\"must be one of 'DC_ONE' "
-        "'DC_QUORUM' 'DC_SAFE_QUORUM'");
+        "'DC_QUORUM' 'DC_SAFE_QUORUM' 'DC_EACH_SAFE_QUORUM'");
     return DN_ERROR;
   }
 

src/dyn_conf.h

@@ -47,6 +47,7 @@
 #define CONF_STR_DC_ONE "dc_one"
 #define CONF_STR_DC_QUORUM "dc_quorum"
 #define CONF_STR_DC_SAFE_QUORUM "dc_safe_quorum"
+#define CONF_STR_DC_EACH_SAFE_QUORUM "dc_each_safe_quorum"
 
 #define UNSET_NUM 0
 

src/dyn_dnode_client.c

@@ -306,7 +306,8 @@ static void dnode_req_forward(struct context *ctx, struct conn *conn,
     // racks
     struct mbuf *orig_mbuf = STAILQ_FIRST(&req->mhdr);
     struct datacenter *dc = server_get_dc(pool, &pool->dc);
-    req_forward_all_local_racks(ctx, conn, req, orig_mbuf, key, keylen, dc);
+
+    req_forward_all_racks_for_dc(ctx, conn, req, orig_mbuf, key, keylen, dc);
   }
 }
 

src/dyn_message.c

@@ -415,6 +415,8 @@ static struct msg *_msg_get(struct conn *conn, bool request,
   msg->orig_msg = NULL;
   msg->needs_repair = false;
   msg->rewrite_with_ts_possible = true;
+  msg->additional_each_rspmgrs = NULL;
+  msg->rspmgrs_inited = false;
 
   return msg;
 }
@@ -641,6 +643,23 @@ void msg_put(struct msg *msg) {
     msg_put(msg->orig_msg);
     msg->orig_msg = NULL;
   }
+
+  if (msg->additional_each_rspmgrs) {
+    ASSERT(msg->consistency == DC_EACH_SAFE_QUORUM);
+    // Only requests have their connection's owner as the 'struct server_pool' object,
+    // and only requests would have 'additional_each_rspmgrs', so it's safe to cast to
+    // 'struct server_pool'.
+    struct server_pool *sp = msg->owner->owner;
+    uint8_t num_dcs = array_n(&sp->datacenters);
+
+    int i;
+    // Skip the 0th index as that points back to the statically allocated 'rspmgr' struct
+    // in 'msg'.
+    for (i = 1; i < num_dcs; ++i) {
+      dn_free(msg->additional_each_rspmgrs[i]);
+    }
+    dn_free(msg->additional_each_rspmgrs);
+  }
   TAILQ_INSERT_HEAD(&free_msgq, msg, m_tqe);
 }
 

src/dyn_message.h

@@ -333,6 +333,7 @@ typedef enum consistency {
   DC_ONE = 0,
   DC_QUORUM,
   DC_SAFE_QUORUM,
+  DC_EACH_SAFE_QUORUM,
 } consistency_t;
 
 static inline char *get_consistency_string(consistency_t cons) {
@@ -343,6 +344,8 @@ static inline char *get_consistency_string(consistency_t cons) {
       return "DC_QUORUM";
     case DC_SAFE_QUORUM:
       return "DC_SAFE_QUORUM";
+    case DC_EACH_SAFE_QUORUM:
+      return "DC_EACH_SAFE_QUORUM";
   }
   return "INVALID CONSISTENCY";
 }
@@ -430,7 +433,7 @@ struct msg {
                                              or remote region or cross rack */
   usec_t request_send_time; /* when message was sent: either to the data store
                                or remote region or cross rack */
-  uint8_t awaiting_rsps;
+  uint32_t awaiting_rsps;
   struct msg *selected_rsp;
 
   struct rbnode tmo_rbe; /* entry in rbtree */
@@ -506,7 +509,16 @@ struct msg {
   msg_response_handler_t rsp_handler;
   consistency_t consistency;
   msgid_t parent_id; /* parent message id */
+
+  // Primary response_mgr for this instance's DC.
   struct response_mgr rspmgr;
+
+  // Additional response_mgrs if we choose to use DC_EACH_SAFE_QUORUM
+  struct response_mgr **additional_each_rspmgrs;
+
+  // Indicates whether the rspmgr and additional_each_rspmgrs(if applicable)
+  // are init-ed.
+  bool rspmgrs_inited;
 };
 
 TAILQ_HEAD(msg_tqh, msg);
@@ -588,7 +600,7 @@ void dnode_rsp_gos_syn(struct context *ctx, struct conn *p_conn,
 
 void req_forward_error(struct context *ctx, struct conn *conn, struct msg *req,
                        err_t error_code, err_t dyn_error_code);
-void req_forward_all_local_racks(struct context *ctx, struct conn *c_conn,
+void req_forward_all_racks_for_dc(struct context *ctx, struct conn *c_conn,
                                  struct msg *req, struct mbuf *orig_mbuf,
                                  uint8_t *key, uint32_t keylen,
                                  struct datacenter *dc);