Powered by WP Greet Box WordPress Plugin
原创文章，转载请注明： 转载自pagefault
本文链接地址: nginx中upstream的设计和实现(二)
这次主要来看upstream的几个相关的hook函数。
首先要知道，对于upstream，同时有两个连接，一个时client和nginx，一个是nginx和upstream，这个时候就会有两个回调，然后上篇blog中，我们能看到在upstream中，会改变read_event_handler和write_event_handler,不过这里有三个条件，分别是
1 没有使用cache，
2 不忽略client的提前终止
3 不是post_action
1 2 3 4 5 6 | //条件赋值 if (!u->store && !r->post_action && !u->conf->ignore_client_abort) {//然后给读写handler赋值 r->read_event_handler = ngx_http_upstream_rd_check_broken_connection; r->write_event_handler = ngx_http_upstream_wr_check_broken_connection; } |
然后我们来看这个两个函数，这两个都会调用ngx_http_upstream_check_broken_connection，因此我们就先来详细分析这个函数。
这个函数主要是用来检测client的连接是否完好。因此它使用了MSG_PEEK这个参数，也就是预读，然后通过recv的返回值来判断是否连接已经断开。
这里的代码分为两部分，第一部分是本身连接在进入这个回调函数之前连接都已经有错误了，这个时候如果是水平触发，则删除事件，然后finalize这个upstream(没有cache’的情况下），否则就直接finalize这个upstream。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | c = r->connection; u = r->upstream;//如果连接已经出现错误。 if (c->error) {//如果是水平触发 if ((ngx_event_flags & NGX_USE_LEVEL_EVENT) && ev->active) { event = ev->write ? NGX_WRITE_EVENT : NGX_READ_EVENT;//删除事件 if (ngx_del_event(ev, event, 0) != NGX_OK) { ngx_http_upstream_finalize_request(r, u, NGX_HTTP_INTERNAL_SERVER_ERROR); return; } } if (!u->cacheable) {//清理upstream request ngx_http_upstream_finalize_request(r, u, NGX_HTTP_CLIENT_CLOSED_REQUEST); } return; } |
紧接着就是第二部分，这部分的工作就是预读取１个字节，然后来判断是否连接已经被client断掉。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 | //读取１个字节 n = recv(c->fd, buf, 1, MSG_PEEK); err = ngx_socket_errno; ngx_log_debug1(NGX_LOG_DEBUG_HTTP, ev->log, err, "http upstream recv(): %d", n); if (ev->write && (n >= 0 || err == NGX_EAGAIN)) { return; }//如果水平触发则删除事件 if ((ngx_event_flags & NGX_USE_LEVEL_EVENT) && ev->active) { event = ev->write ? NGX_WRITE_EVENT : NGX_READ_EVENT; if (ngx_del_event(ev, event, 0) != NGX_OK) { ngx_http_upstream_finalize_request(r, u, NGX_HTTP_INTERNAL_SERVER_ERROR); return; } }//如果还有数据，则直接返回 if (n > 0) { return; } if (n == -1) { if (err == NGX_EAGAIN) { return; } ev->error = 1; } else { /* n == 0 */ err = 0; }//到达这里说明有错误产生了 ev->eof = 1;//设置错误，可以看到这个值在函数一开始有检测. c->error = 1;//如果没有cache，则finalize upstream request if (!u->cacheable && u->peer.connection) { ngx_log_error(NGX_LOG_INFO, ev->log, err, "client closed prematurely connection, " "so upstream connection is closed too"); ngx_http_upstream_finalize_request(r, u, NGX_HTTP_CLIENT_CLOSED_REQUEST); return; } ngx_log_error(NGX_LOG_INFO, ev->log, err, "client closed prematurely connection");//如果有cache，并且后端的upstream还在处理，则此时继续处理upstream，忽略对端的错误. if (u->peer.connection == NULL) { ngx_http_upstream_finalize_request(r, u, NGX_HTTP_CLIENT_CLOSED_REQUEST); } |
然后我们来看nginx如何连接后端的upstream，在上篇blog的结束的时候，我们看到最终会调用ngx_http_upstream_connect来进入连接upstream的处理，因此我们来详细分析这个函数以及相关的函数。
函数一开始是初始化请求开始事件一些参数
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | ..........................................................//取得upstream的状态 u->state = ngx_array_push(r->upstream_states); if (u->state == NULL) { ngx_http_upstream_finalize_request(r, u, NGX_HTTP_INTERNAL_SERVER_ERROR); return; } ngx_memzero(u->state, sizeof(ngx_http_upstream_state_t)); tp = ngx_timeofday();//初始化时间 u->state->response_sec = tp->sec; u->state->response_msec = tp->msec; |
然后是调用ngx_event_connect_peer开始连接后端upstream.并且对返回值进行处理，等下会详细分析ngx_event_connect_peer这个函数.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | //连接后端 rc = ngx_event_connect_peer(&u->peer); ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, "http upstream connect: %i", rc); if (rc == NGX_ERROR) { ngx_http_upstream_finalize_request(r, u, NGX_HTTP_INTERNAL_SERVER_ERROR); return; }//这个是很关键的一个结构peer，后面的blog会详细分析 u->state->peer = u->peer.name; if (rc == NGX_BUSY) { ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, "no live upstreams"); ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_NOLIVE); return; } if (rc == NGX_DECLINED) { ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR); return; } |
当返回值为NGX_OK或者NGX_AGAIN的话，就说明连接成功或者暂时异步的连接还没成功，所以需要挂载upstream端的回调函数.这里要注意就是NGX_AGAIN的情况，因为是异步的connect，所以可能会连接不成功。所以如果返回NGX_AGAIN的话，需要挂载写函数.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 | /* rc == NGX_OK || rc == NGX_AGAIN */ c = u->peer.connection; c->data = r; c->write->handler = ngx_http_upstream_handler; c->read->handler = ngx_http_upstream_handler;//开始挂载回调函数，一个是读，一个是写。 u->write_event_handler = ngx_http_upstream_send_request_handler; u->read_event_handler = ngx_http_upstream_process_header; c->sendfile &= r->connection->sendfile; u->output.sendfile = c->sendfile; c->pool = r->pool; c->log = r->connection->log; c->read->log = c->log; c->write->log = c->log; /* init or reinit the ngx_output_chain() and ngx_chain_writer() contexts */ u->writer.out = NULL; u->writer.last = &u->writer.out; u->writer.connection = c; u->writer.limit = 0; |
然后时对request_body的一些处理以及如果request_sent已经设置，也就是这个upstream已经发送过一部分数据了，此时需要重新初始化upstream.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 | if (u->request_sent) {//重新初始化upstream if (ngx_http_upstream_reinit(r, u) != NGX_OK) { ngx_http_upstream_finalize_request(r, u, NGX_HTTP_INTERNAL_SERVER_ERROR); return; } }//如果request_body存在的话，保存request_body if (r->request_body && r->request_body->buf && r->request_body->temp_file && r == r->main) { /* * the r->request_body->buf can be reused for one request only, * the subrequests should allocate their own temporay bufs */ u->output.free = ngx_alloc_chain_link(r->pool); if (u->output.free == NULL) { ngx_http_upstream_finalize_request(r, u, NGX_HTTP_INTERNAL_SERVER_ERROR); return; }//保存到output u->output.free->buf = r->request_body->buf; u->output.free->next = NULL; u->output.allocated = 1;//重置request_body r->request_body->buf->pos = r->request_body->buf->start; r->request_body->buf->last = r->request_body->buf->start; r->request_body->buf->tag = u->output.tag; } |
最后则是先判断rc的返回值，如果是NGX_AGAIN,则说明连接没有返回，则设置定时器，然后返回，否则说明连接成功，这时就需要发送请求到后端。
1 2 3 4 5 6 7 | if (rc == NGX_AGAIN) {//添加定时器 ngx_add_timer(c->write, u->conf->connect_timeout); return; } ngx_http_upstream_send_request(r, u); |
紧接着我们来看最后的两个函数，分别是上面的ngx_event_connect_peer和ngx_http_upstream_send_request，我们来一个个看。
先来看ngx_event_connect_peer。它主要是用来连接后端，函数比较长，一部分一部分来看。
下面这部分主要是建立socket，然后设置属性，从连接池取出来connection.这里后面的一部分和我们前面client请求上来之后，我们初始化connect类似.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | //取得我们将要发送的upstream对端 rc = pc->get(pc, pc->data); if (rc != NGX_OK) { return rc; }//新建socket s = ngx_socket(pc->sockaddr->sa_family, SOCK_STREAM, 0); ngx_log_debug1(NGX_LOG_DEBUG_EVENT, pc->log, 0, "socket %d", s); if (s == -1) { ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno, ngx_socket_n " failed"); return NGX_ERROR; }//取得连接 c = ngx_get_connection(s, pc->log); if (c == NULL) { if (ngx_close_socket(s) == -1) { ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno, ngx_close_socket_n "failed"); } return NGX_ERROR; }//设置rcvbuf的大小 if (pc->rcvbuf) { if (setsockopt(s, SOL_SOCKET, SO_RCVBUF, (const void *) &pc->rcvbuf, sizeof(int)) == -1) { ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno, "setsockopt(SO_RCVBUF) failed"); goto failed; } }//设置非阻塞 if (ngx_nonblocking(s) == -1) { ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno, ngx_nonblocking_n " failed"); goto failed; } if (pc->local) { if (bind(s, pc->local->sockaddr, pc->local->socklen) == -1) { ngx_log_error(NGX_LOG_CRIT, pc->log, ngx_socket_errno, "bind(%V) failed", &pc->local->name); goto failed; } }//开始挂载对应的读写函数. c->recv = ngx_recv; c->send = ngx_send; c->recv_chain = ngx_recv_chain; c->send_chain = ngx_send_chain; c->sendfile = 1; c->log_error = pc->log_error; if (pc->sockaddr->sa_family != AF_INET) { c->tcp_nopush = NGX_TCP_NOPUSH_DISABLED; c->tcp_nodelay = NGX_TCP_NODELAY_DISABLED;#if (NGX_SOLARIS) /* Solaris's sendfilev() supports AF_NCA, AF_INET, and AF_INET6 */ c->sendfile = 0;#endif } rev = c->read; wev = c->write; rev->log = pc->log; wev->log = pc->log; pc->connection = c; c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);#if (NGX_THREADS) /* TODO: lock event when call completion handler */ rev->lock = pc->lock; wev->lock = pc->lock; rev->own_lock = &c->lock; wev->own_lock = &c->lock;#endif if (ngx_add_conn) {//添加读写事件 if (ngx_add_conn(c) == NGX_ERROR) { goto failed; } } |
等socket设置完毕，nginx就开始连接后端的upstream，这段代码可以学习一个好的代码是如何处理错误的，
下面这段主要是处理当返回值为-１，并且err不等于NGX_EINPROGRESS的时候，而NGX_EINPROGRESS表示非阻塞的socket，然后connect，然后连接还没有完成，可是提前返回，就回设置这个errno.这个error不算出错，因此需要过滤掉.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | rc = connect(s, pc->sockaddr, pc->socklen); if (rc == -1) { err = ngx_socket_errno;//判断错误号 if (err != NGX_EINPROGRESS#if (NGX_WIN32) /* Winsock returns WSAEWOULDBLOCK (NGX_EAGAIN) */ && err != NGX_EAGAIN#endif ) { if (err == NGX_ECONNREFUSED#if (NGX_LINUX) /* * Linux returns EAGAIN instead of ECONNREFUSED * for unix sockets if listen queue is full */ || err == NGX_EAGAIN#endif || err == NGX_ECONNRESET || err == NGX_ENETDOWN || err == NGX_ENETUNREACH || err == NGX_EHOSTDOWN || err == NGX_EHOSTUNREACH) { level = NGX_LOG_ERR; } else { level = NGX_LOG_CRIT; } ngx_log_error(level, c->log, err, "connect() to %V failed", pc->name);//返回declined return NGX_DECLINED; } } |
然后就是下面的部门就是处理连接成功和错误号为NGX_EINPROGRESS的情况，
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | //如果当前的事件模型支持add_conn，则事件在开始已经加好了，因此如果rc==-1则直接返回 if (ngx_add_conn) { if (rc == -1) { /* NGX_EINPROGRESS */ return NGX_AGAIN; } ngx_log_debug0(NGX_LOG_DEBUG_EVENT, pc->log, 0, "connected"); wev->ready = 1; return NGX_OK; }..............................................//添加可读事件 if (ngx_add_event(rev, NGX_READ_EVENT, event) != NGX_OK) { goto failed; } if (rc == -1) { /* NGX_EINPROGRESS *///如果错误号是 EINPROGRES 添加可写事件 if (ngx_add_event(wev, NGX_WRITE_EVENT, event) != NGX_OK) { goto failed; } return NGX_AGAIN; } ngx_log_debug0(NGX_LOG_DEBUG_EVENT, pc->log, 0, "connected"); wev->ready = 1; return NGX_OK;.............................................. |
最后我们来看下ngx_http_upstream_send_request的实现，这个函数是用来发送数据到后端的upstream，然后这里有一个需要注意的地方，那就是在linux下当非阻塞的connect，然后没有连接完成，如果挂载写事件，此时如果写事件上报上来，并不代表连接成功，此时还需要调用getsockopt来判断SO_ERROR，如果没有错误才能保证连接成功。
SOL_SOCKET
to determine whether connect() completed successfully (SO_ERROR is zero) or unsuccessfully (SO_ERROR is one of the usual error codes listed
here, explaining the reason for the failure).
这里我看了下内核的代码，就是如果连接失败，比如对端不可达，内核会设置sock->sk_soft_err,而在tcp_poll中只会检测sk_err , 对应的SO_ERROR会检测这两个错误。在内核里面的注释是这样子的
* @sk_err: last error
* @sk_err_soft: errors that don’t cause failure but are the cause of a
* persistent failure not just ‘timed out’
这个按照我的理解，内核里面的sk_err 表示４层的错误，而sk_err_soft下层的错误.
在nginx中是在ngx_http_upstream_test_connect中对连接是否断开进行判断的(调用getsockopt).
然后发送数据则是调用ngx_output_chain，不过这里我们知道在ngx_output_chain中会依次调用filter链，可是upstream明显不需要调用filter链，那么nginx是怎么做的呢，是这样子的，在upstream的初始化的时候，已经讲u->output.output_filter改成ngx_chain_writer了:
1 | u->output.output_filter = ngx_chain_writer; |
最后就是一些对错误的处理，我们来看代码
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | static voidngx_http_upstream_send_request(ngx_http_request_t *r, ngx_http_upstream_t *u){ ngx_int_t rc; ngx_connection_t *c; c = u->peer.connection; ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0, "http upstream send request");//如果test connect失败，则说明连接失败，于是跳到下一个upstream，然后返回 if (!u->request_sent && ngx_http_upstream_test_connect(c) != NGX_OK) { ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR); return; } c->log->action = "sending request to upstream";//发送数据，这里的u->output.output_filter已经被修改过了 rc = ngx_output_chain(&u->output, u->request_sent ? NULL : u->request_bufs); u->request_sent = 1;........................................................//和request的处理类似，如果again，则说明数据没有发送完毕，此时挂载写事件. if (rc == NGX_AGAIN) { ngx_add_timer(c->write, u->conf->send_timeout); if (ngx_handle_write_event(c->write, u->conf->send_lowat) != NGX_OK) { ngx_http_upstream_finalize_request(r, u, NGX_HTTP_INTERNAL_SERVER_ERROR); return; } return; } /* rc == NGX_OK *///设置tcp_cork,远离和前面的keepalive部分的处理类似 if (c->tcp_nopush == NGX_TCP_NOPUSH_SET) { if (ngx_tcp_push(c->fd) == NGX_ERROR) { ngx_log_error(NGX_LOG_CRIT, c->log, ngx_socket_errno, ngx_tcp_push_n " failed"); ngx_http_upstream_finalize_request(r, u, NGX_HTTP_INTERNAL_SERVER_ERROR); return; } c->tcp_nopush = NGX_TCP_NOPUSH_UNSET; } ngx_add_timer(c->read, u->conf->read_timeout);#if 1//如果读也可以了，则开始解析头 if (c->read->ready) { /* post aio operation */ /* * TODO comment * although we can post aio operation just in the end * of ngx_http_upstream_connect() CHECK IT !!! * it's better to do here because we postpone header buffer allocation */ ngx_http_upstream_process_header(r, u); return; }#endif...........................................} |
在下一篇blog里面，我会详细的分析nginx对后端来的数据如何解析以及如何发送数据到client.
