Merge pull request #730 from Netflix/dev

Merge dev into v0.6

src/dyn_client.c

@@ -343,6 +343,7 @@ struct msg *req_recv_next(struct context *ctx, struct conn *conn, bool alloc) {
   if (is_read_repairs_enabled()) {
     req->timestamp = current_timestamp_in_millis();
   }
+
   return req;
 }
 
@@ -364,9 +365,18 @@ static bool req_filter(struct context *ctx, struct conn *conn,
   if (req->quit) {
     ASSERT(conn->rmsg == NULL);
     log_debug(LOG_VERB, "%s filter quit %s", print_obj(conn), print_obj(req));
+
+    // The client expects to receive an "+OK\r\n" response, so make sure
+    // to do that.
+    IGNORE_RET_VAL(simulate_ok_rsp(ctx, conn, req));
+
     conn->eof = 1;
     conn->recv_ready = 0;
-    req_put(req);
+    return true;
+  }
+
+  // If this is a Dynomite configuration message, don't forward it.
+  if (is_msg_type_dyno_config(req->type)) {
     return true;
   }
 

src/dyn_connection.c

@@ -127,17 +127,15 @@ inline void conn_set_read_consistency(struct conn *conn, consistency_t cons) {
 }
 
 inline consistency_t conn_get_read_consistency(struct conn *conn) {
-  // return conn->read_consistency;
-  return g_read_consistency;
+  return conn->read_consistency;
 }
 
 inline void conn_set_write_consistency(struct conn *conn, consistency_t cons) {
   conn->write_consistency = cons;
 }
 
 inline consistency_t conn_get_write_consistency(struct conn *conn) {
-  // return conn->write_consistency;
-  return g_write_consistency;
+  return conn->write_consistency;
 }
 
 rstatus_t conn_event_del_conn(struct conn *conn) {

src/dyn_core.c

@@ -132,6 +132,9 @@ static rstatus_t core_event_base_create(struct context *ctx) {
 }
 
 /**
+ *
+ * NOTE: DEPRECATED and not currently used in the codebase.
+ *
  * Initialize anti-entropy.
  * @param[in,out] ctx Context.
  * @return rstatus_t Return status code.
@@ -269,11 +272,6 @@ rstatus_t core_start(struct instance *nci) {
     goto error;
   }
 
-  status = core_entropy_init(ctx);
-  if (status != DN_OK) {
-    goto error;
-  }
-
   status = core_event_base_create(ctx);
   if (status != DN_OK) {
     goto error;

src/dyn_gossip.c

@@ -840,12 +840,12 @@ static void *gossip_loop(void *arg) {
 }
 
 rstatus_t gossip_start(struct server_pool *sp) {
-  rstatus_t status;
   pthread_t tid;
 
-  status = pthread_create(&tid, NULL, gossip_loop, sp);
-  if (status < 0) {
-    log_error("gossip service create failed: %s", strerror(status));
+  int pthread_status;
+  pthread_status = pthread_create(&tid, NULL, gossip_loop, sp);
+  if (pthread_status < 0) {
+    log_error("gossip service create failed: %s", strerror(pthread_status));
     return DN_ERROR;
   }
 

src/dyn_message.c

@@ -915,6 +915,93 @@ static rstatus_t msg_repair(struct context *ctx, struct conn *conn,
   return DN_OK;
 }
 
+/*
+ * Crafts a success response message for the respective datastore.
+ *
+ * TODO: This currently does only Redis. The Redis specific code should
+ *       be moved out of this file.
+ *
+ * Returns a 'msg' with the expected success response.
+ */
+static struct msg *craft_ok_rsp(struct context *ctx, struct conn *conn,
+    struct msg *req) {
+
+  ASSERT(req->is_request);
+
+  rstatus_t ret_status = DN_OK;
+  const char *QUIT_FMT_STRING = "+OK\r\n";
+
+  struct msg *rsp = msg_get(conn, false, __FUNCTION__);
+  if (rsp == NULL) {
+    conn->err = errno;
+    return NULL;
+  }
+
+  rstatus_t append_status = msg_append(rsp, QUIT_FMT_STRING, strlen(QUIT_FMT_STRING));
+  if (append_status != DN_OK) {
+    rsp_put(rsp);
+    return NULL;
+  }
+
+  rsp->peer = req;
+  rsp->is_request = 0;
+
+  req->done = 1;
+
+  return rsp;
+}
+
+rstatus_t simulate_ok_rsp(struct context *ctx, struct conn *conn,
+    struct msg *msg) {
+  // Create an OK response.
+  struct msg *ok_rsp = craft_ok_rsp(ctx, conn, msg);
+
+  // Add it to the outstanding messages dictionary, so that 'conn_handle_response'
+  // can process it appropriately.
+  dictAdd(conn->outstanding_msgs_dict, &msg->id, msg);
+
+  // Enqueue the message in the outbound queue so that the code on the response
+  // path can find it.
+  conn_enqueue_outq(ctx, conn, msg);
+
+  THROW_STATUS(conn_handle_response(ctx, conn,
+      msg->parent_id ? msg->parent_id : msg->id, ok_rsp));
+
+  return DN_OK;
+}
+
+/*
+ * If the command sent to Dynomite was a special Dynomite configuration
+ * command, we process and apply the configuration here.
+ *
+ * Returns: DN_OK on successful application, DN_ERROR otherwise.
+ */
+static rstatus_t msg_apply_config(struct context *ctx, struct conn *conn,
+    struct msg *msg) {
+
+  // We only support one type of configuration now.
+  // TODO: If we support more, convert this to a switch case.
+  ASSERT(msg->type == MSG_HACK_SETTING_CONN_CONSISTENCY);
+
+  struct argpos *consistency_string = (struct argpos*) array_get(msg->args, 0);
+
+  // We must have a consistency string, else we wouldn't have reached here.
+  ASSERT(consistency_string != NULL);
+
+  consistency_t cons = get_consistency_enum_from_string(consistency_string->start);
+  if (cons == -1) return DN_ERROR;
+
+  conn_set_read_consistency(conn, cons);
+  conn_set_write_consistency(conn, cons);
+
+  // Set the consistency to DC_ONE, since this is just a configuration setting.
+  msg->consistency = DC_ONE;
+
+  THROW_STATUS(simulate_ok_rsp(ctx, conn, msg));
+
+  return DN_OK;
+}
+
 static rstatus_t msg_parse(struct context *ctx, struct conn *conn,
                            struct msg *msg) {
   rstatus_t status;
@@ -929,19 +1016,24 @@ static rstatus_t msg_parse(struct context *ctx, struct conn *conn,
 
   switch (msg->result) {
     case MSG_PARSE_OK:
-      // log_debug(LOG_VVERB, "MSG_PARSE_OK");
       status = msg_parsed(ctx, conn, msg);
       break;
-
     case MSG_PARSE_REPAIR:
-      // log_debug(LOG_VVERB, "MSG_PARSE_REPAIR");
       status = msg_repair(ctx, conn, msg);
       break;
-
     case MSG_PARSE_AGAIN:
-      // log_debug(LOG_VVERB, "MSG_PARSE_AGAIN");
       status = DN_OK;
       break;
+    case MSG_PARSE_DYNO_CONFIG:
+      status = msg_apply_config(ctx, conn, msg);
+
+      // No more data to parse.
+      conn_recv_done(ctx, conn, msg, NULL);
+      break;
+
+    case MSG_PARSE_NOOP:
+      status = DN_NOOPS;
+      break;
 
     default:
       /*
@@ -1611,3 +1703,9 @@ rstatus_t msg_append_format(struct msg *msg, const char *fmt, int num_args, ...)
 
   return DN_OK;
 }
+
+bool is_msg_type_dyno_config(msg_type_t msg_type) {
+  // TODO: Convert to a switch case if we support more.
+  if (msg_type == MSG_HACK_SETTING_CONN_CONSISTENCY) return true;
+  return false;
+}

src/dyn_message.h

@@ -218,6 +218,7 @@
   ACTION(RSP_REDIS_ERROR_EXECABORT)                                            \
   ACTION(RSP_REDIS_ERROR_MASTERDOWN)                                           \
   ACTION(RSP_REDIS_ERROR_NOREPLICAS)                                           \
+  ACTION(HACK_SETTING_CONN_CONSISTENCY)                                        \
   ACTION(SENTINEL)                                                             \
   ACTION(END_IDX)                                                              \
   /* ACTION( REQ_REDIS_AUTH) */                                                \
@@ -260,11 +261,21 @@ extern func_msg_repair_t g_make_repair_query; /* Create a repair msg. */
 void set_datastore_ops(void);
 
 typedef enum msg_parse_result {
-  MSG_PARSE_OK,       /* parsing ok */
-  MSG_PARSE_ERROR,    /* parsing error */
-  MSG_PARSE_REPAIR,   /* more to parse -> repair parsed & unparsed data */
-  MSG_PARSE_FRAGMENT, /* multi-vector request -> fragment */
-  MSG_PARSE_AGAIN,    /* incomplete -> parse again */
+  // Parsing OK
+  MSG_PARSE_OK,
+  // Parsing error
+  MSG_PARSE_ERROR,
+  // More to parse -> Repair parsed & unparsed data
+  MSG_PARSE_REPAIR,
+  // Multi-vector request -> fragment
+  MSG_PARSE_FRAGMENT,
+  // Incomplete, parse again.
+  MSG_PARSE_AGAIN,
+  // Parsing done, but do nothing after
+  MSG_PARSE_NOOP,
+  // Parsing done, command was a dynomite configuration
+  MSG_PARSE_DYNO_CONFIG,
+  // OOM error during parsing (TODO: consider removing)
   MSG_OOM_ERROR
 } msg_parse_result_t;
 
@@ -350,6 +361,19 @@ static inline char *get_consistency_string(consistency_t cons) {
   return "INVALID CONSISTENCY";
 }
 
+static inline consistency_t get_consistency_enum_from_string(char *cons) {
+  if (dn_strcasecmp(cons, "DC_ONE") == 0) {
+    return DC_ONE;
+  } else if (dn_strcasecmp(cons, "DC_QUORUM") == 0) {
+    return DC_QUORUM;
+  } else if (dn_strcasecmp(cons, "DC_SAFE_QUORUM") == 0) {
+    return DC_SAFE_QUORUM;
+  } else if (dn_strcasecmp(cons, "DC_EACH_SAFE_QUORUM") == 0) {
+    return DC_EACH_SAFE_QUORUM;
+  }
+  return -1;
+}
+
 #define DEFAULT_READ_CONSISTENCY DC_ONE
 #define DEFAULT_WRITE_CONSISTENCY DC_ONE
 extern consistency_t g_write_consistency;
@@ -616,4 +640,18 @@ rstatus_t dnode_peer_req_forward(struct context *ctx, struct conn *c_conn,
 // string *data);
 void dnode_peer_gossip_forward(struct context *ctx, struct conn *conn,
                                struct mbuf *data);
+
+/*
+ * Simulates a successful response as though the datastore sent it.
+ * Also, does the necessary to make sure that the response path is
+ * able to send this response back to the client.
+ *
+ * Returns DN_OK on success and an appropriate error otherwise.
+ */
+rstatus_t simulate_ok_rsp(struct context *ctx, struct conn *conn,
+    struct msg *msg);
+
+// Returns 'true' if 'msg_type' is a Dynomite configuration command.
+bool is_msg_type_dyno_config(msg_type_t msg_type);
+
 #endif

src/dyn_response_mgr.c

@@ -182,7 +182,7 @@ static void rspmgr_incr_non_quorum_responses_stats(
  */
 bool perform_repairs_if_necessary(struct context *ctx, struct response_mgr *rspmgr) {
 
-  struct msg* repair_msg;
+  struct msg* repair_msg = NULL;
   rstatus_t repair_create_status = g_make_repair_query(ctx, rspmgr, &repair_msg);
 
   if (repair_create_status == DN_OK && repair_msg != NULL) {

src/dyn_server.c

@@ -886,10 +886,11 @@ void req_send_done(struct context *ctx, struct conn *conn, struct msg *req) {
    * enqueue message (request) in server outq, if response is expected.
    * Otherwise, free the request
    */
-  if (req->expect_datastore_reply || (conn->type == CONN_SERVER))
+  if (req->expect_datastore_reply || (conn->type == CONN_SERVER)) {
     conn_enqueue_outq(ctx, conn, req);
-  else
+  } else {
     req_put(req);
+  }
 }
 
 static void req_server_enqueue_imsgq(struct context *ctx, struct conn *conn,

src/dyn_stats.c

@@ -1346,17 +1346,16 @@ static rstatus_t stats_listen(struct stats *st) {
 }
 
 static rstatus_t stats_start_aggregator(struct stats *st) {
-  rstatus_t status;
-
   if (!stats_enabled) {
     return DN_OK;
   }
 
   THROW_STATUS(stats_listen(st));
 
-  status = pthread_create(&st->tid, NULL, stats_loop, st);
-  if (status < 0) {
-    log_error("stats aggregator create failed: %s", strerror(status));
+  int pthread_status;
+  pthread_status = pthread_create(&st->tid, NULL, stats_loop, st);
+  if (pthread_status < 0) {
+    log_error("stats aggregator create failed: %s", strerror(pthread_status));
     return DN_ERROR;
   }
 

src/entropy/dyn_entropy_util.c

@@ -405,6 +405,9 @@ rstatus_t entropy_key_iv_load(struct context *ctx) {
 }
 
 /*
+ *
+ * NOTE: DEPRECATED and not currently used in the codebase.
+ *
  * Function:  entropy_snd_init
  * --------------------
  * Initiates the data for the connection towards another cluster for
@@ -627,14 +630,13 @@ void *entropy_loop(void *arg) {
  */
 
 rstatus_t entropy_conn_start(struct entropy *cn) {
-  rstatus_t status;
-
   THROW_STATUS(entropy_listen(cn));
 
-  status = pthread_create(&cn->tid, NULL, entropy_loop, cn);
-  if (status < 0) {
+  int pthread_status;
+  pthread_status = pthread_create(&cn->tid, NULL, entropy_loop, cn);
+  if (pthread_status < 0) {
     log_error("reconciliation thread for socket create failed: %s",
-              strerror(status));
+              strerror(pthread_status));
     return DN_ERROR;
   }