twemproxy发送流程探索——剖析twemproxy代码正编

 1 rstatus_t
 2 msg_send(struct context *ctx, struct conn *conn)
 3 {
 4     rstatus_t status;
 5     struct msg *msg;
 6     /*表示活跃的发送状态*/
 7     ASSERT(conn->send_active);
 8     /*表示准备发送*/
 9     conn->send_ready = 1;
10     do {
11         /*获取下一次发送的msg开头*/
12         msg = conn->send_next(ctx, conn);
13         if (msg == NULL) {
14             /* nothing to send */
15             return NC_OK;
16         }
17         /*发送框架，在此框架内conn->send_ready会改变*/
18         status = msg_send_chain(ctx, conn, msg);
19         if (status != NC_OK) {
20             return status;
21         }
22 
23     } while (conn->send_ready);
24 
25     return NC_OK;
26 }

  1 static rstatus_t
  2 msg_send_chain(struct context *ctx, struct conn *conn, struct msg *msg)
  3 {
  4     struct msg_tqh send_msgq;            /* send msg q */
  5     struct msg *nmsg;                    /* next msg */
  6     struct mbuf *mbuf, *nbuf;            /* current and next mbuf */
  7     size_t mlen;                         /* current mbuf data length */
  8     struct iovec *ciov, iov[NC_IOV_MAX]; /* current iovec */
  9     struct array sendv;                  /* send iovec */
 10     size_t nsend, nsent;                 /* bytes to send; bytes sent */
 11     size_t limit;                        /* bytes to send limit */
 12     ssize_t n;                           /* bytes sent by sendv */
 13 
 14     TAILQ_INIT(&send_msgq);
 15 
 16     array_set(&sendv, iov, sizeof(iov[0]), NC_IOV_MAX);
 17 
 18     /* preprocess - build iovec */
 19 
 20     nsend = 0;
 21     /*
 22      * readv() and writev() returns EINVAL if the sum of the iov_len values
 23      * overflows an ssize_t value Or, the vector count iovcnt is less than
 24      * zero or greater than the permitted maximum.
 25      */
 26     limit = SSIZE_MAX;
 27 
 28     /*
 29      *send_msgq是一个临时的发送队列，将当前能进行发送的msg，即处理完的msg
 30      *进行存储。发送队列仅仅自后面处理时能让调用者以msg的buf为单位处理。
 31      *sendv是一个字符串数组，由于发送底层采用的函数是writev，为此sendv将发
 32      *送的数据都存储在一起，sendv才是真正发送的数据内存。
 33      */
 34     for (;;) {
 35         ASSERT(conn->smsg == msg);
 36 
 37         TAILQ_INSERT_TAIL(&send_msgq, msg, m_tqe);
 38 
 39         for (mbuf = STAILQ_FIRST(&msg->mhdr);
 40              mbuf != NULL && array_n(&sendv) < NC_IOV_MAX && nsend < limit;
 41              mbuf = nbuf) {
 42             nbuf = STAILQ_NEXT(mbuf, next);
 43             /*
 44              *发送的信息是否为空，即发送开始的字节位置是否和结束位置一致。
 45              *在处理redis多key命令的mget，mdel，mset以及memcached多key命令
 46              *get，gets时，由于分片的原因，分片后的msg也会在客户端发送队列
 47              *中。在分片处理完要发送后，这些分片的msg应该不能被发送，为此，
 48              *对于分片的msg的pos进行了将msg的发送量置为空，这边的sendv在添
 49              *加发送内容时，忽视了这些分片。
 50              */
 51             if (mbuf_empty(mbuf)) {
 52                 continue;
 53             }
 54 
 55             mlen = mbuf_length(mbuf);
 56             if ((nsend + mlen) > limit) {
 57                 mlen = limit - nsend;
 58             }
 59 
 60             ciov = array_push(&sendv);
 61             ciov->iov_base = mbuf->pos;
 62             ciov->iov_len = mlen;
 63 
 64             nsend += mlen;
 65         }
 66 
 67         /*超过发送限制*/
 68         if (array_n(&sendv) >= NC_IOV_MAX || nsend >= limit) {
 69             break;
 70         }
 71 
 72         /*不存在发送内容*/
 73         msg = conn->send_next(ctx, conn);
 74         if (msg == NULL) {
 75             break;
 76         }
 77     }
 78 
 79     /*
 80      * (nsend == 0) is possible in redis multi-del
 81      * see PR: https://github.com/twitter/twemproxy/pull/225
 82      */
 83 
 84     /*发送函数conn_sendv*/
 85     conn->smsg = NULL;
 86     if (!TAILQ_EMPTY(&send_msgq) && nsend != 0) {
 87         n = conn_sendv(conn, &sendv, nsend);
 88     } else {
 89         n = 0;
 90     }
 91 
 92     nsent = n > 0 ? (size_t)n : 0;
 93 
 94     /* postprocess - process sent messages in send_msgq */
 95     /*
 96      *由于其发送函数底层采用writev，在发送过程中可能存在发送中断或者发送
 97      *数据没有全部发出的情况，为此需要通过实际发送的字节数nsent来确认系统
 98      *实际上发送到了哪一个msg的哪一个mbuf的哪一个字节pos，以便下一次从pos
 99      *开始发送实际的内容，以免重复发送相同的内容，导致不可见的错误。
100      */
101     for (msg = TAILQ_FIRST(&send_msgq); msg != NULL; msg = nmsg) {
102         nmsg = TAILQ_NEXT(msg, m_tqe);
103 
104         TAILQ_REMOVE(&send_msgq, msg, m_tqe);
105 
106         /*发送内容为空，进行发送完的处理*/
107         if (nsent == 0) {
108             if (msg->mlen == 0) {
109                 conn->send_done(ctx, conn, msg);
110             }
111             continue;
112         }
113 
114         /* adjust mbufs of the sent message */
115         for (mbuf = STAILQ_FIRST(&msg->mhdr); mbuf != NULL; mbuf = nbuf) {
116             nbuf = STAILQ_NEXT(mbuf, next);
117 
118             if (mbuf_empty(mbuf)) {
119                 continue;
120             }
121 
122             mlen = mbuf_length(mbuf);
123             if (nsent < mlen) {
124                 /* mbuf was sent partially; process remaining bytes later */
125                 /*此处确认了实际上发送到了哪一个msg的哪一个mbuf的哪一个字节pos*/
126                 mbuf->pos += nsent;
127                 ASSERT(mbuf->pos < mbuf->last);
128                 nsent = 0;
129                 break;
130             }
131 
132             /* mbuf was sent completely; mark it empty */
133             mbuf->pos = mbuf->last;
134             nsent -= mlen;
135         }
136 
137         /* message has been sent completely, finalize it */
138         if (mbuf == NULL) {
139             conn->send_done(ctx, conn, msg);
140         }
141     }
142 
143     ASSERT(TAILQ_EMPTY(&send_msgq));
144 
145     if (n >= 0) {
146         return NC_OK;
147     }
148 
149     return (n == NC_EAGAIN) ? NC_OK : NC_ERROR;
150 }

 1 ssize_t
 2 conn_sendv(struct conn *conn, struct array *sendv, size_t nsend)
 3 {
 4     ssize_t n;
 5 
 6     ASSERT(array_n(sendv) > 0);
 7     ASSERT(nsend != 0);
 8     ASSERT(conn->send_ready);
 9 
10     for (;;) {
11         /*这里的nc_writev就是writev*/
12         n = nc_writev(conn->sd, sendv->elem, sendv->nelem);
13 
14         log_debug(LOG_VERB, "sendv on sd %d %zd of %zu in %"PRIu32" buffers",
15                   conn->sd, n, nsend, sendv->nelem);
16 
17         if (n > 0) {
18             /*
19              *已发送数据长度比待发送数据长度小，说明系统发送队列已满或者不
20              *可写，此刻需要停止发送数据。
21              */
22             if (n < (ssize_t) nsend) {
23                 conn->send_ready = 0;
24             }
25             conn->send_bytes += (size_t)n;
26             return n;
27         }
28 
29         if (n == 0) {
30             log_warn("sendv on sd %d returned zero", conn->sd);
31             conn->send_ready = 0;
32             return 0;
33         }
34         /*
35          *EINTR表示由于信号中断，没发送成功任何数据，此刻需要停止发送数据。
36          *EAGAIN以及EWOULDBLOCK表示系统发送队列已满或者不可写，为此没发送
37          *成功任何数据，此刻需要停止发送数据，等待下次发送。
38          *除了上述两种错误，其他的错误为连接出现了问题需要停止发送数据并
39          *进行断链操作，conn->err非零时在程序流程中会触发断链。
40          */
41         if (errno == EINTR) {
42             log_debug(LOG_VERB, "sendv on sd %d not ready - eintr", conn->sd);
43             continue;
44         } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
45             conn->send_ready = 0;
46             log_debug(LOG_VERB, "sendv on sd %d not ready - eagain", conn->sd);
47             return NC_EAGAIN;
48         } else {
49             conn->send_ready = 0;
50             conn->err = errno;
51             log_error("sendv on sd %d failed: %s", conn->sd, strerror(errno));
52             return NC_ERROR;
53         }
54     }
55 
56     NOT_REACHED();
57 
58     return NC_ERROR;
59 }