dynomite简单分析

uint32_t dnode_peer_idx_for_key_on_rack(struct server_pool *pool,
                                        struct rack *rack, uint8_t *key,
                                        uint32_t keylen) {
  struct dyn_token token;
  pool->key_hash(key, keylen, &token);
  return vnode_dispatch(&rack->continuums, rack->ncontinuum, &token);
}

struct dyn_token {
  uint32_t signum;
  uint32_t mag[4];
  uint32_t len;
};

/**
 * Set unlimited core dump resource limits.
 */
static void dn_coredump_init(void) {
  struct rlimit core_limits;
  core_limits.rlim_cur = core_limits.rlim_max = RLIM_INFINITY;
  setrlimit(RLIMIT_CORE, &core_limits);
}

static rstatus_t dn_run(struct instance *nci) {
  rstatus_t status;
  THROW_STATUS(core_start(nci));
  struct context *ctx = nci->ctx;
  struct server_pool *sp = &ctx->pool;
  if (!sp->enable_gossip) core_set_local_state(ctx, NORMAL);
  /* run rabbit run */
  for (;;) {
    status = core_loop(ctx);
    if (status != DN_OK) {
      break;
    }
  }
  core_stop(ctx);
  return DN_OK;
}

typedef enum consistency {
  DC_ONE = 0,
  DC_QUORUM,
  DC_SAFE_QUORUM,
  DC_EACH_SAFE_QUORUM,
} consistency_t;

rstatus_t conn_listen(struct context *ctx, struct conn *p) {
  rstatus_t status;
  struct server_pool *pool = &ctx->pool;
  ASSERT((p->type == CONN_PROXY) || (p->type == CONN_DNODE_PEER_PROXY));
  p->sd = socket(p->family, SOCK_STREAM, 0);
  status = _conn_reuse(p);
  status = bind(p->sd, p->addr, p->addrlen);
  status = listen(p->sd, pool->backlog);
  status = dn_set_nonblocking(p->sd);
  status = dn_set_keepalive(p->sd, 1);
  status = conn_event_add_conn(p);
  status = conn_event_del_out(p);
}

typedef enum connection_type {
  CONN_UNSPECIFIED,
  CONN_PROXY,   // a dynomite proxy (listening) connection
  CONN_CLIENT,  // this is connected to a client connection
  CONN_SERVER,  // this is connected to underlying datastore ...redis/memcache
  CONN_DNODE_PEER_PROXY,   // this is a dnode (listening) connection...default
                           // 8101
  CONN_DNODE_PEER_CLIENT,  // this is connected to a dnode peer client
  CONN_DNODE_PEER_SERVER,  // this is connected to a dnode peer server
} connection_type_t;

struct conn_ops {
  func_recv_t recv;           /* recv (read) handler */
  func_recv_next_t recv_next; /* recv next message handler */
  func_recv_done_t recv_done; /* read done handler */
  func_send_t send;           /* send (write) handler */
  func_send_next_t send_next; /* write next message handler */
  func_send_done_t send_done; /* write done handler */
  func_close_t close;         /* close handler */
  func_active_t active;       /* active? handler */

  func_ref_t ref;     /* connection reference handler */
  func_unref_t unref; /* connection unreference handler */

  func_msgq_t enqueue_inq;  /* connection inq msg enqueue handler */
  func_msgq_t dequeue_inq;  /* connection inq msg dequeue handler */
  func_msgq_t enqueue_outq; /* connection outq msg enqueue handler */
  func_msgq_t dequeue_outq; /* connection outq msg dequeue handler */
  func_response_handler rsp_handler;
};

//这些指针封装了一系列的宏
#define conn_recv(ctx, conn) (conn)->ops->recv(ctx, conn)
#define conn_recv_next(ctx, conn, alloc) \
  (conn)->ops->recv_next(ctx, conn, alloc)
#define conn_recv_done(ctx, conn, msg, nmsg) \
  (conn)->ops->recv_done(ctx, conn, msg, nmsg)

#define conn_send(ctx, conn) (conn)->ops->send(ctx, conn)
#define conn_send_next(ctx, conn) (conn)->ops->send_next(ctx, conn)
#define conn_send_done(ctx, conn, msg) (conn)->ops->send_done(ctx, conn, msg)

#define conn_close(ctx, conn) (conn)->ops->close(ctx, conn)
#define conn_active(conn) (conn)->ops->active(conn)
#define conn_ref(conn, owner) (conn)->ops->ref(conn, owner)
#define conn_unref(conn) (conn)->ops->unref(conn)

#define conn_enqueue_inq(ctx, conn, msg) \
  (conn)->ops->enqueue_inq(ctx, conn, msg)
#define conn_dequeue_inq(ctx, conn, msg) \
  (conn)->ops->dequeue_inq(ctx, conn, msg)
#define conn_enqueue_outq(ctx, conn, msg) \
  (conn)->ops->enqueue_outq(ctx, conn, msg)
#define conn_dequeue_outq(ctx, conn, msg) \
  (conn)->ops->dequeue_outq(ctx, conn, msg)
TAILQ_HEAD(conn_tqh, conn);

struct conn_ops dnode_server_ops = {
    dnode_recv, NULL,        NULL, NULL,      NULL,
    NULL,       dnode_close, NULL, dnode_ref, dnode_unref,
    NULL,       NULL,        NULL, NULL,      conn_cant_handle_response};
void init_dnode_proxy_conn(struct conn *conn) {
  conn->dyn_mode = 1;
  conn->type = CONN_DNODE_PEER_PROXY;
  conn->ops = &dnode_server_ops;
}

static rstatus_t dnode_accept(struct context *ctx, struct conn *p) {
  rstatus_t status;
  struct conn *c;
  struct sockaddr_in client_address;
  socklen_t client_len = sizeof(client_address);
  int sd = 0;
  ASSERT(p->type == CONN_DNODE_PEER_PROXY);
  ASSERT(p->sd > 0);
  ASSERT(p->recv_active && p->recv_ready);
  for (;;) {
    sd = accept(p->sd, (struct sockaddr *)&client_address, &client_len);
    if (sd < 0) {
      if (errno == EINTR) {
        log_warn("accept on %s not ready - eintr", print_obj(p));
        continue;
      }

      if (errno == EAGAIN || errno == EWOULDBLOCK) {
        p->recv_ready = 0;
        return DN_OK;
      }

      /*
       * FIXME: On EMFILE or ENFILE mask out IN event on the proxy; mask
       * it back in when some existing connection gets closed
       */

      log_error("accept on %s failed: %s", print_obj(p), strerror(errno));
      return DN_ERROR;
    }

    break;
  }
  char clntName[INET_ADDRSTRLEN];
  inet_ntop(AF_INET, &client_address.sin_addr.s_addr, clntName,
                sizeof(clntName)) != NULL);
  c = conn_get(p->owner, init_dnode_client_conn); //注意这行
  c->sd = sd;
  string_copy_c(&c->pname, (unsigned char *)dn_unresolve_peer_desc(c->sd));
  stats_pool_incr(ctx, dnode_client_connections);
  status = dn_set_nonblocking(c->sd);
  if (p->family == AF_INET || p->family == AF_INET6) {
    status = dn_set_tcpnodelay(c->sd);
  }
  status = conn_event_add_conn(c);
}

struct conn_ops dnode_client_ops = {msg_recv,
                                    dnode_req_recv_next,
                                    dnode_req_recv_done,
                                    msg_send,
                                    dnode_rsp_send_next,
                                    dnode_rsp_send_done,
                                    dnode_client_close,
                                    dnode_client_active,
                                    dnode_client_ref,
                                    dnode_client_unref,
                                    NULL,
                                    NULL,
                                    dnode_req_client_enqueue_omsgq,
                                    dnode_req_client_dequeue_omsgq,
                                    dnode_client_handle_response};

void init_dnode_client_conn(struct conn *conn) {
  conn->dyn_mode = 1;
  conn->type = CONN_DNODE_PEER_CLIENT;
  conn->ops = &dnode_client_ops;
}

/*

 *             Client+             Proxy           Server+
 *                              (dynomite)
 *                                   .
 *       msg_recv {read event}       .       msg_recv {read event}
 *         +                         .                         +
 *         |                         .                         |
 *         \                         .                         /
 *         req_recv_next             .             rsp_recv_next
 *           +                       .                       +
 *           |                       .                       |       Rsp
 *           req_recv_done           .           rsp_recv_done      <===
 *             +                     .                     +
 *             |                     .                     |
 *    Req      \                     .                     /
 *    ===>     req_filter*           .           *rsp_filter
 *               +                   .                   +
 *               |                   .                   |
 *               \                   .                   /
 *               req_forward-//  (a) . (c)  \\-rsp_forward
 *                                   .
 *                                   .
 *       msg_send {write event}      .      msg_send {write event}
 *         +                         .                         +
 *         |                         .                         |
 *    Rsp' \                         .                         /     Req'
 *   <===  rsp_send_next             .             req_send_next     ===>
 *           +                       .                       +
 *           |                       .                       |
 *           \                       .                       /
 *           rsp_send_done-//    (d) . (b)    //-req_send_done

*/

struct conn_ops proxy_ops = {proxy_recv, NULL, NULL, NULL, NULL, NULL,
                             proxy_close, NULL, proxy_ref, proxy_unref,
                             // enqueue, dequeues
                             NULL, NULL, NULL, NULL, conn_cant_handle_response};

void init_proxy_conn(struct conn *conn) {
  conn->dyn_mode = 0;
  conn->type = CONN_PROXY;
  conn->ops = &proxy_ops;
}

struct conn_ops client_ops = {msg_recv,
                              req_recv_next,
                              req_recv_done,
                              msg_send,
                              rsp_send_next,
                              rsp_send_done,
                              client_close,
                              client_active,
                              client_ref,
                              client_unref,
                              NULL,
                              NULL,
                              req_client_enqueue_omsgq,
                              req_client_dequeue_omsgq,
                              client_handle_response};

void init_client_conn(struct conn *conn) {
  conn->dyn_mode = 0;
  conn->type = CONN_CLIENT;
  conn->ops = &client_ops;
}

  if (request_send_to_all_local_racks(req)) {
    // send request to all local racks
    req_forward_all_racks_for_dc(ctx, c_conn, req, orig_mbuf, key, keylen, dc);
  } 

// Wait for only quorum number of responses before responding
bool rspmgr_check_is_done(struct response_mgr *rspmgr) {
  uint8_t pending_responses = (uint8_t)(
      rspmgr->max_responses - rspmgr->good_responses - rspmgr->error_responses);
  // do the required calculation and tell if we are done here
  if (rspmgr->good_responses >= rspmgr->quorum_responses) {
    // We received enough good responses but do their checksum match?
    if (rspmgr_is_quorum_achieved(rspmgr)) {
      log_info("req %lu quorum achieved", rspmgr->msg->id);
      rspmgr->done = true;
    } else if (pending_responses) {
      // There's a mismatch in checksum. Wait for any pending responses
      rspmgr->done = false;
    } else {
      // no pending responses, and the checksum do not match.
      rspmgr->done = true;
    }
  } else if ((pending_responses + rspmgr->good_responses) <
             rspmgr->quorum_responses) {
    // Even if we receive all the pending responses, still we do not form
    // a quorum, So decision is done, no quorum possible
    rspmgr->done = true;
  }
  return rspmgr->done;
}