本文想要完成对twemproxy发送流程——msg_send的探索，对于twemproxy发送流程的数据结构已经在《twemproxy接收流程探索——剖析twemproxy代码正编》介绍过了，msg_send和msg_recv的流程大致类似。请在阅读代码时，查看注释，英文注释是作者对它的代码的注解，中文注释是我自己的感悟。
函数msg_send
rstatus_t
msg_send(struct context *ctx, struct conn *conn)
{
rstatus_t status;
struct msg *msg;
/*表示活跃的发送状态*/
ASSERT(conn->send_active);
/*表示准备发送*/
conn->send_ready = ;
do {
/*获取下一次发送的msg开头*/
msg = conn->send_next(ctx, conn);
if (msg == NULL) {
/* nothing to send */
return NC_OK;
}
/*发送框架，在此框架内conn->send_ready会改变*/
status = msg_send_chain(ctx, conn, msg);
if (status != NC_OK) {
return status;
} } while (conn->send_ready); return NC_OK;
}
发送框架msg_send_chain
由于在发送时，其底层采用writev的高效发送方式，难免出现数据发送到一边，系统的发送队列已满的情况，面对这种尴尬的情况，你应该如何处理？twemproxy的作者给出了自己的方式。
static rstatus_t
msg_send_chain(struct context *ctx, struct conn *conn, struct msg *msg)
{
struct msg_tqh send_msgq; /* send msg q */
struct msg *nmsg; /* next msg */
struct mbuf *mbuf, *nbuf; /* current and next mbuf */
size_t mlen; /* current mbuf data length */
struct iovec *ciov, iov[NC_IOV_MAX]; /* current iovec */
struct array sendv; /* send iovec */
size_t nsend, nsent; /* bytes to send; bytes sent */
size_t limit; /* bytes to send limit */
ssize_t n; /* bytes sent by sendv */ TAILQ_INIT(&send_msgq); array_set(&sendv, iov, sizeof(iov[]), NC_IOV_MAX); /* preprocess - build iovec */ nsend = ;
/*
* readv() and writev() returns EINVAL if the sum of the iov_len values
* overflows an ssize_t value Or, the vector count iovcnt is less than
* zero or greater than the permitted maximum.
*/
limit = SSIZE_MAX; /*
*send_msgq是一个临时的发送队列，将当前能进行发送的msg，即处理完的msg
*进行存储。发送队列仅仅自后面处理时能让调用者以msg的buf为单位处理。
*sendv是一个字符串数组，由于发送底层采用的函数是writev，为此sendv将发
*送的数据都存储在一起，sendv才是真正发送的数据内存。
*/
for (;;) {
ASSERT(conn->smsg == msg); TAILQ_INSERT_TAIL(&send_msgq, msg, m_tqe); for (mbuf = STAILQ_FIRST(&msg->mhdr);
mbuf != NULL && array_n(&sendv) < NC_IOV_MAX && nsend < limit;
mbuf = nbuf) {
nbuf = STAILQ_NEXT(mbuf, next);
/*
*发送的信息是否为空，即发送开始的字节位置是否和结束位置一致。
*在处理redis多key命令的mget，mdel，mset以及memcached多key命令
*get，gets时，由于分片的原因，分片后的msg也会在客户端发送队列
*中。在分片处理完要发送后，这些分片的msg应该不能被发送，为此，
*对于分片的msg的pos进行了将msg的发送量置为空，这边的sendv在添
*加发送内容时，忽视了这些分片。
*/
if (mbuf_empty(mbuf)) {
continue;
} mlen = mbuf_length(mbuf);
if ((nsend + mlen) > limit) {
mlen = limit - nsend;
} ciov = array_push(&sendv);
ciov->iov_base = mbuf->pos;
ciov->iov_len = mlen; nsend += mlen;
} /*超过发送限制*/
if (array_n(&sendv) >= NC_IOV_MAX || nsend >= limit) {
break;
} /*不存在发送内容*/
msg = conn->send_next(ctx, conn);
if (msg == NULL) {
break;
}
} /*
* (nsend == 0) is possible in redis multi-del
* see PR: https://github.com/twitter/twemproxy/pull/225
*/ /*发送函数conn_sendv*/
conn->smsg = NULL;
if (!TAILQ_EMPTY(&send_msgq) && nsend != ) {
n = conn_sendv(conn, &sendv, nsend);
} else {
n = ;
} nsent = n > ? (size_t)n : ; /* postprocess - process sent messages in send_msgq */
/*
*由于其发送函数底层采用writev，在发送过程中可能存在发送中断或者发送
*数据没有全部发出的情况，为此需要通过实际发送的字节数nsent来确认系统
*实际上发送到了哪一个msg的哪一个mbuf的哪一个字节pos，以便下一次从pos
*开始发送实际的内容，以免重复发送相同的内容，导致不可见的错误。
*/
for (msg = TAILQ_FIRST(&send_msgq); msg != NULL; msg = nmsg) {
nmsg = TAILQ_NEXT(msg, m_tqe); TAILQ_REMOVE(&send_msgq, msg, m_tqe); /*发送内容为空，进行发送完的处理*/
if (nsent == ) {
if (msg->mlen == ) {
conn->send_done(ctx, conn, msg);
}
continue;
} /* adjust mbufs of the sent message */
for (mbuf = STAILQ_FIRST(&msg->mhdr); mbuf != NULL; mbuf = nbuf) {
nbuf = STAILQ_NEXT(mbuf, next); if (mbuf_empty(mbuf)) {
continue;
} mlen = mbuf_length(mbuf);
if (nsent < mlen) {
/* mbuf was sent partially; process remaining bytes later */
/*此处确认了实际上发送到了哪一个msg的哪一个mbuf的哪一个字节pos*/
mbuf->pos += nsent;
ASSERT(mbuf->pos < mbuf->last);
nsent = ;
break;
} /* mbuf was sent completely; mark it empty */
mbuf->pos = mbuf->last;
nsent -= mlen;
} /* message has been sent completely, finalize it */
if (mbuf == NULL) {
conn->send_done(ctx, conn, msg);
}
} ASSERT(TAILQ_EMPTY(&send_msgq)); if (n >= ) {
return NC_OK;
} return (n == NC_EAGAIN) ? NC_OK : NC_ERROR;
}
发送函数conn_sendv
writev作为一个高效的网络io，它的正确用法一直是个问题，这里给出了twemproxy的作者给出了自己正确的注解。对于其的异常处理值得借鉴
ssize_t
conn_sendv(struct conn *conn, struct array *sendv, size_t nsend)
{
ssize_t n; ASSERT(array_n(sendv) > );
ASSERT(nsend != );
ASSERT(conn->send_ready); for (;;) {
/*这里的nc_writev就是writev*/
n = nc_writev(conn->sd, sendv->elem, sendv->nelem); log_debug(LOG_VERB, "sendv on sd %d %zd of %zu in %"PRIu32" buffers",
conn->sd, n, nsend, sendv->nelem); if (n > ) {
/*
*已发送数据长度比待发送数据长度小，说明系统发送队列已满或者不
*可写，此刻需要停止发送数据。
*/
if (n < (ssize_t) nsend) {
conn->send_ready = ;
}
conn->send_bytes += (size_t)n;
return n;
} if (n == ) {
log_warn("sendv on sd %d returned zero", conn->sd);
conn->send_ready = ;
return ;
}
/*
*EINTR表示由于信号中断，没发送成功任何数据，此刻需要停止发送数据。
*EAGAIN以及EWOULDBLOCK表示系统发送队列已满或者不可写，为此没发送
*成功任何数据，此刻需要停止发送数据，等待下次发送。
*除了上述两种错误，其他的错误为连接出现了问题需要停止发送数据并
*进行断链操作，conn->err非零时在程序流程中会触发断链。
*/
if (errno == EINTR) {
log_debug(LOG_VERB, "sendv on sd %d not ready - eintr", conn->sd);
continue;
} else if (errno == EAGAIN || errno == EWOULDBLOCK) {
conn->send_ready = ;
log_debug(LOG_VERB, "sendv on sd %d not ready - eagain", conn->sd);
return NC_EAGAIN;
} else {
conn->send_ready = ;
conn->err = errno;
log_error("sendv on sd %d failed: %s", conn->sd, strerror(errno));
return NC_ERROR;
}
} NOT_REACHED(); return NC_ERROR;
}
小结
在这短短的数百行代码中，我们获知了msg_send的简单过程，最最重要的是我们知道了writev函数的发送内容处理和异常处理，特别是它如教科书般的异常处理方式使我收益良多。