ngx_http_upstream_init_round_robin 中初始化us->peer.init = ngx_http_upstream_init_round_robin_peer，
然后us->peer.data为解析的upstream信息，backup和非backup等分开解析。
ngx_http_upstream_init_round_robin_peer，完成了对upstream中服务器的初始化工作。
代码中有些不易理解的地方，主要是对tried变量值的使用。tried中记录了服务器当前是否被尝试连接过。
他是一个位图。如果服务器数量小于32，则只需在一个int中即可记录下所有服务器状态。如果服务器数量大于32，则需在内存池中申请内存来存储。
ngx_http_upstream_get_round_robin_peer/ngx_http_upstream_free_round_robin_peer/ngx_http_upstream_tries/ngx_http_upstream_set_round_robin_peer_session/
ngx_http_upstream_save_round_robin_peer_session等在ngx_http_upstream_init_round_robin_peer中挂载，其中get完成获取后端服务器的功能，其代码主要是逻辑处理。
需要注意的地方是最后有个对backup服务器的处理：当正常服务器都失败时，会落到backup组中。
ngx_http_upstream_session_sticky_set_sid函数把session_sticky cookie=OLWAFRSID mode=insert option=indirect maxidle=3600;
配置里的cookie即OLWAFRSID转换成md5赋给sscf->server[i].sid.data。
GitHub - openresty/lua-upstream-nginx-module: Nginx C module to expose Lua API to ngx_lua for Nginx upstreams
lua-upstream-nginx-module介绍：主要功能是通过lua api获取get_upstreams名，获取指定上游组中所有服务器的配置，获取指定上游组中所有主（非backuop）server配置（权重 effective_weight fail_timeout 失败ID等）。
get_backup_peers获取指定上游组中所有备份对等体的配置。，set_peer_down设置指定对等体的“down”（布尔）属性。需要指定上游名称。
current_upstream_name返回当前请求的代理上游的名称。
static voidngx_http_upstream_init_request(ngx_http_request_t *r){//处于aio状态 则直接返回if (r->aio) {return;}u = r->upstream;#if (NGX_HTTP_CACHE)if (u->conf->cache) {ngx_int_t rc;rc = ngx_http_upstream_cache(r, u);if (rc == NGX_BUSY) { //来不及处理 返回等待事件触发 NGX_BUSY用于并发请求miss 在获取数据之前 除第一个请求以外的其他请求均被阻塞return;}r->write_event_handler = ngx_http_request_empty_handler; //写事件触发不做处理if (rc == NGX_ERROR) { //缓存信息获取出错 直接将错误发送给请求端ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);return;}if (rc == NGX_OK) { //缓存命中 直接发送rc = ngx_http_upstream_cache_send(r, u);if (rc == NGX_DONE) { //头部处理出错return;}if (rc == NGX_HTTP_UPSTREAM_INVALID_HEADER) { //无效头部判定 会回源rc = NGX_DECLINED;r->cached = 0;u->buffer.start = NULL;u->cache_status = NGX_HTTP_CACHE_MISS;u->request_sent = 1;}if (ngx_http_upstream_cache_background_update(r, u) != NGX_OK) {rc = NGX_ERROR;}}}#endifu->store = u->conf->store;if (!u->store && !r->post_action && !u->conf->ignore_client_abort) { //开启上下游连接检测r->read_event_handler = ngx_http_upstream_rd_check_broken_connection;r->write_event_handler = ngx_http_upstream_wr_check_broken_connection;}if (r->request_body) { //存在http请求包体 包体信息存入request_bufs中u->request_bufs = r->request_body->bufs;}if (u->create_request(r) != NGX_OK) { //创建请求干预过程出错 直接响应服务器内部错误到请求端ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);return;}if (ngx_http_upstream_set_local(r, u, u->conf->local) != NGX_OK) {//本地upstream地址配置ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);return;}clcf = ngx_http_get_module_loc_conf(r, ngx_http_core_module);u->output.alignment = clcf->directio_alignment;u->output.pool = r->pool;u->output.bufs.num = 1;u->output.bufs.size = clcf->client_body_buffer_size;if (u->output.output_filter == NULL) {u->output.output_filter = ngx_chain_writer;u->output.filter_ctx = &u->writer;}u->writer.pool = r->pool;//upstream状态信息初始化if (r->upstream_states == NULL) {...} else {...}... //生命周期结束清理处理设置if (u->resolved == NULL) { //还未开始解析uscf = u->conf->upstream;} else {#if (NGX_HTTP_SSL)u->ssl_name = u->resolved->host;#endifhost = &u->resolved->host;umcf = ngx_http_get_module_main_conf(r, ngx_http_upstream_module);uscfp = umcf->upstreams.elts;for (i = 0; i < umcf-<s; i++) { //从配置的upstream地址信息中搜索（优先处理）uscf = uscfp[i];if (uscf->host.len == host->len&& ((uscf->port == 0 && u->resolved->no_port)|| uscf->port == u->resolved->port)&& ngx_strncasecmp(uscf->host.data, host->data, host->len) == 0){goto found;}}if (u->resolved->sockaddr) {... //已经解析地址信息检查if (ngx_http_upstream_create_round_robin_peer(r, u->resolved) //检查通过 创建轮询peer(根据返回的主机数量) 具有负载效果!= NGX_OK){ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}ngx_http_upstream_connect(r, u); //连接至上游服务器return;}if (u->resolved->port == 0) { //无效port返回服务器错误...return;}temp.name = *host;ctx = ngx_resolve_start(clcf->resolver, &temp); //开始域名解析 域名解析器由core_module配置 未配置对于非IP host处理会失败if (ctx == NULL) {ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}if (ctx == NGX_NO_RESOLVER) { //未配置域名解析器 响应错误ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,"no resolver defined to resolve %V", host);ngx_http_upstream_finalize_request(r, u, NGX_HTTP_BAD_GATEWAY);return;}... //upstream域名解析器信息设置 解析完成会在ngx_http_upstream_resolve_handler中触发if (ngx_resolve_name(ctx) != NGX_OK) { //域名解析出错 响应错误请求（注意 域名解析过程是异步的）u->resolved->ctx = NULL;ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}return;}found:if (uscf == NULL) {ngx_log_error(NGX_LOG_ALERT, r->connection->log, 0,"no upstream configuration");ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}u->upstream = uscf;#if (NGX_HTTP_SSL)u->ssl_name = uscf->host;#endifif (uscf->peer.init(r, uscf) != NGX_OK) { //连接信息初始化失败 响应服务器错误ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}u->peer.start_time = ngx_current_msec; //设置连接时间if (u->conf->next_upstream_tries&& u-&ies > u->conf->next_upstream_tries){u-&ies = u->conf->next_upstream_tries; //重试次数}ngx_http_upstream_connect(r, u); //连接到上游服务器}static voidngx_http_upstream_connect(ngx_http_request_t *r, ngx_http_upstream_t *u){ngx_int_t rc;ngx_connection_t *c;r->connection->log->action = "connecting to upstream";if (u->state && u->state->response_time) {u->state->response_time = ngx_current_msec - u->state->response_time; //更新uptream state响应时间}... //upstream state信息初始化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;}u->state->peer = u->peer.name;if (rc == NGX_BUSY) { //来不及处理 尝试连接到另一个server （如果有）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) { //连接被拒绝 尝试连接到另一个server (如果有)ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR);return;}/* rc == NGX_OK || rc == NGX_AGAIN || rc == NGX_DONE */c = u-&tion;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;if (c->pool == NULL) {/* we need separate pool here to be able to cache SSL connections */c->pool = ngx_create_pool(128, r->connection->log);... //创建连接内存池}c->log = r->connection->log;c->pool->log = c->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-&tion = c;u->writer.limit = 0;if (u->request_sent) {if (ngx_http_upstream_reinit(r, u) != NGX_OK) { //重新初始化upstreamngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}}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 temporary bufs*///请求包体buf循环利用 挂载到upstream 输出空闲buf中（对于将请求包体放入临时文件的请求）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;}u->output.free->buf = r->request_body->buf;u->output.free->next = NULL;u->output.allocated = 1;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;}u->request_sent = 0;u->request_body_sent = 0;if (rc == NGX_AGAIN) { //还无法判断是否成功连接 加入到定时器中ngx_add_timer(c->write, u->conf->connect_timeout);return;}#if (NGX_HTTP_SSL)if (u->ssl && c->ssl == NULL) { //初始化ssl连接ngx_http_upstream_ssl_init_connection(r, u, c);return;}#endifngx_http_upstream_send_request(r, u, 1); //向上游服务器发送请求}static voidngx_http_upstream_send_request(ngx_http_request_t *r, ngx_http_upstream_t *u,ngx_uint_t do_write){ngx_int_t rc;ngx_connection_t *c;c = u-&tion;ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,"http upstream send request");if (u->state->connect_time == (ngx_msec_t) -1) { //更新upstream 连接时间耗费状态信息u->state->connect_time = ngx_current_msec - u->state->response_time;}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";rc = ngx_http_upstream_send_request_body(r, u, do_write); //发送请求包体 这里是写事件if (rc == NGX_ERROR) {ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR); //发送出错 转移到下一个server请求return;}if (rc >= NGX_HTTP_SPECIAL_RESPONSE) { //满足http错误状态码条件 结束请求 同时响应结果到请求端ngx_http_upstream_finalize_request(r, u, rc);return;}if (rc == NGX_AGAIN) { //无数据可发送if (!c->write->ready) { //写事件没有准备号 将写事件加入定时器ngx_add_timer(c->write, u->conf->send_timeout);} else if (c->write->timer_set) {ngx_del_timer(c->write);}if (ngx_handle_write_event(c->write, u->conf->send_lowat) != NGX_OK) { //lowat低潮值发送至上游服务器ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}return;}/* rc == NGX_OK */u->request_body_sent = 1;if (c->write->timer_set) {ngx_del_timer(c->write);}if (c->tcp_nopush == NGX_TCP_NOPUSH_SET) { //默认no_push方式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;}u->write_event_handler = ngx_http_upstream_dummy_handler; //不对写事件进行处理if (ngx_handle_write_event(c->write, 0) != NGX_OK) { //写事件处理出错 响应至请求端ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}ngx_add_timer(c->read, u->conf->read_timeout);if (c->read->ready) {ngx_http_upstream_process_header(r, u); //读事件触发 准备处理http头部信息return;}}static voidngx_http_upstream_process_header(ngx_http_request_t *r, ngx_http_upstream_t *u){...c = u-&tion;ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,"http upstream process header");c->log->action = "reading response header from upstream";if (c->read->timedout) {ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_TIMEOUT); //读事件超时 转移到下一个serverreturn;}if (!u->request_sent && ngx_http_upstream_test_connect(c) != NGX_OK) { //请求未发送到上游服务器 同时连接测试不通过 转移到下一个serveerngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR);return;}if (u->buffer.start == NULL) {u->buffer.start = ngx_palloc(r->pool, u->conf->buffer_size); //根据配置的buffer大小创建缓冲区... //缓冲信息初始化if (ngx_list_init(&u->headers_in.headers, r->pool, 8,sizeof(ngx_table_elt_t)) //初始化upstream 头部hash表 默认8个!= NGX_OK){ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}#if (NGX_HTTP_CACHE)if (r->cache) {u->buffer.pos += r->cache->header_start; //http头部位置偏移设置u->buffer.last = u->buffer.pos;}#endif}for ( ;; ) {n = c->recv(c, u->buffer.last, u-&d - u->buffer.last);if (n == NGX_AGAIN) {#if 0ngx_add_timer(rev, u->read_timeout);#endifif (ngx_handle_read_event(c->read, 0) != NGX_OK) { //再尝试读取一次ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}return;}if (n == 0) { //读取不到任何数据ngx_log_error(NGX_LOG_ERR, c->log, 0,"upstream prematurely closed connection");}if (n == NGX_ERROR || n == 0) { //出错 尝试到下一个serverngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR);return;}u->state->bytes_received += n; //更新接收到的字节数u->buffer.last += n; //更新buffer last位置#if 0u->valid_header_in = 0;u->peer.cached = 0;#endifrc = u->process_header(r); //process_header介入处理if (rc == NGX_AGAIN) { //头部数据读取不完整if (u->buffer.last == u-&d) { //头部过大 标记为错误 转向下一台server请求ngx_log_error(NGX_LOG_ERR, c->log, 0,"upstream sent too big header");ngx_http_upstream_next(r, u,NGX_HTTP_UPSTREAM_FT_INVALID_HEADER);return;}continue;}break; //头部数据读取完成}if (rc == NGX_HTTP_UPSTREAM_INVALID_HEADER) { //发现头部无效 转向下一台ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_INVALID_HEADER);return;}if (rc == NGX_ERROR) { //错误 一般是服务器逻辑处理错误 此时不再转向下一个server 直接终止请求 并响应server错误到请求端ngx_http_upstream_finalize_request(r, u,NGX_HTTP_INTERNAL_SERVER_ERROR);return;}/* rc == NGX_OK */u->state->header_time = ngx_current_msec - u->state->response_time; //处理成功 更新header处理时间if (u->headers_in.status_n >= NGX_HTTP_SPECIAL_RESPONSE) {if (ngx_http_upstream_test_next(r, u) == NGX_OK) {return;}if (ngx_http_upstream_intercept_errors(r, u) == NGX_OK) { //error_pages机制跳转return;}}if (ngx_http_upstream_process_headers(r, u) != NGX_OK) { //upstream自身http头部信息处理return;}if (!r->subrequest_in_memory) { //subrequest_in_memory标记意味着上游响应直接处理后直接发送给客户端ngx_http_upstream_send_response(r, u);return;}/* subrequest content in memory */ //处理在内存中的子请求数据 包含头部和包体数据处理 不会直接发送到请求端if (u->input_filter == NULL) { //upstream上游input_filter设置u->input_filter_init = ngx_http_upstream_non_buffered_filter_init;u->input_filter = ngx_http_upstream_non_buffered_filter;u->input_filter_ctx = r;}if (u->input_filter_init(u->input_filter_ctx) == NGX_ERROR) { //input_filter信息初始化ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}n = u->buffer.last - u->buffer.pos; //计算获取字节数if (n) {u->buffer.last = u->buffer.pos; //更新last位置u->state->response_length += n;if (u->input_filter(u->input_filter_ctx, n) == NGX_ERROR) { //input_filter处理ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}}if (u->length == 0) { //没有body数据 结束ngx_http_upstream_finalize_request(r, u, 0);return;}u->read_event_handler = ngx_http_upstream_process_body_in_memory;ngx_http_upstream_process_body_in_memory(r, u); //处理upstream返回包体数据 这里的包体数据是在buffer链中处理的}static voidngx_http_upstream_process_body_in_memory(ngx_http_request_t *r,ngx_http_upstream_t *u){...c = u-&tion;rev = c->read;ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,"http upstream process body on memory");if (rev->timedout) { //包体数据读取超时 结束请求 错误响应到请求端ngx_connection_error(c, NGX_ETIMEDOUT, "upstream timed out");ngx_http_upstream_finalize_request(r, u, NGX_HTTP_GATEWAY_TIME_OUT);return;}b = &u->buffer;for ( ;; ) {size = b->end - b->last;if (size == 0) { //缓冲区过小ngx_log_error(NGX_LOG_ALERT, c->log, 0,"upstream buffer is too small to read response");ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}n = c->recv(c, b->last, size); //读取包体数据if (n == NGX_AGAIN) { //读事件没有准备好break;}if (n == 0 || n == NGX_ERROR) { //读取错误ngx_http_upstream_finalize_request(r, u, n);return;}u->state->response_length += n; //更新响应长度if (u->input_filter(u->input_filter_ctx, n) == NGX_ERROR) { //input_filter包体过滤处理ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}if (!rev->ready) { //读事件未准备好break;}}if (u->length == 0) { //包体读取完毕 结束ngx_http_upstream_finalize_request(r, u, 0);return;}if (ngx_handle_read_event(rev, 0) != NGX_OK) { //尝试读取ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}if (rev->active) { //读事件是激活状态 加入超时处理ngx_add_timer(rev, u->conf->read_timeout);} else if (rev->timer_set) { //已设置超时删除 定时器ngx_del_timer(rev);}}以上是upstream 对上游包体数据处理流程 接下来是upstream对于发送响应到请求端处理
static voidngx_http_upstream_send_response(ngx_http_request_t *r, ngx_http_upstream_t *u){...rc = ngx_http_send_header(r); //发送头部数据到请求端if (rc == NGX_ERROR || rc > NGX_OK || r->post_action) { //满足条件 由ngx_http_upstream_finalize_request处理ngx_http_upstream_finalize_request(r, u, rc);return;}u->header_sent = 1; //标记头部发送if (u->upgrade) { //upgrage升级处理ngx_http_upstream_upgrade(r, u);return;}c = r->connection;if (r->header_only) { //只有请求头if (!u->buffering) { //不进行缓冲ngx_http_upstream_finalize_request(r, u, rc);return;}if (!u->cacheable && !u->store) { //不可缓存 并且不保存ngx_http_upstream_finalize_request(r, u, rc);return;}u->pipe->downstream_error = 1; //只有请求头部 便标记为downstream_error}if (r->request_body && r->request_body->temp_file) { //对在临时文件的请求包体进行清理ngx_pool_run_cleanup_file(r->pool, r->request_body->temp_file->file.fd);r->request_body->temp_file->file.fd = NGX_INVALID_FILE;}clcf = ngx_http_get_module_loc_conf(r, ngx_http_core_module);if (!u->buffering) { //不进行缓冲处理if (u->input_filter == NULL) {u->input_filter_init = ngx_http_upstream_non_buffered_filter_init;u->input_filter = ngx_http_upstream_non_buffered_filter;u->input_filter_ctx = r;}u->read_event_handler = ngx_http_upstream_process_non_buffered_upstream;r->write_event_handler =ngx_http_upstream_process_non_buffered_downstream;r->limit_rate = 0;if (u->input_filter_init(u->input_filter_ctx) == NGX_ERROR) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}if (clcf->tcp_nodelay && c->tcp_nodelay == NGX_TCP_NODELAY_UNSET) { //设置tcp_nodelayngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0, "tcp_nodelay");tcp_nodelay = 1;if (setsockopt(c->fd, IPPROTO_TCP, TCP_NODELAY,(const void *) &tcp_nodelay, sizeof(int)) == -1){ngx_connection_error(c, ngx_socket_errno,"setsockopt(TCP_NODELAY) failed");ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}c->tcp_nodelay = NGX_TCP_NODELAY_SET;}n = u->buffer.last - u->buffer.pos;if (n) {u->buffer.last = u->buffer.pos;u->state->response_length += n;if (u->input_filter(u->input_filter_ctx, n) == NGX_ERROR) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}ngx_http_upstream_process_non_buffered_downstream(r);} else {u->buffer.pos = u->buffer.start;u->buffer.last = u->buffer.start;if (ngx_http_send_special(r, NGX_HTTP_FLUSH) == NGX_ERROR) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}if (u-&tion->read->ready || u->length == 0) {ngx_http_upstream_process_non_buffered_upstream(r, u);}}return;}/* TODO: preallocate event_pipe bufs, look "Content-Length" */#if (NGX_HTTP_CACHE)if (r->cache && r->cache->file.fd != NGX_INVALID_FILE) { //缓存文件无效 进行文件清理ngx_pool_run_cleanup_file(r->pool, r->cache->file.fd);r->cache->file.fd = NGX_INVALID_FILE;}switch (ngx_http_test_predicates(r, u->conf->no_cache)) { //no_cache检测case NGX_ERROR:ngx_http_upstream_finalize_request(r, u, NGX_ERROR); //检测出错return;case NGX_DECLINED: //满足不缓存条件 设置不缓存标记u->cacheable = 0;break;default: /* NGX_OK */if (u->cache_status == NGX_HTTP_CACHE_BYPASS) { //可以进行缓存/* create cache if previously bypassed */if (ngx_http_file_cache_create(r) != NGX_OK) { //创建缓存文件ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}}break;}if (u->cacheable) {time_t now, valid;now = ngx_time();valid = r->cache->valid_sec;if (valid == 0) {valid = ngx_http_file_cache_valid(u->conf->cache_valid,u->headers_in.status_n); //获取缓存有效期if (valid) {r->cache->valid_sec = now + valid; //更新缓存有效期}}if (valid) {r->cache->date = now;r->cache->body_start = (u_short) (u->buffer.pos - u->buffer.start); //body_start位置获取if (u->headers_in.status_n == NGX_HTTP_OK|| u->headers_in.status_n == NGX_HTTP_PARTIAL_CONTENT) //缓存满足200/206状态值 必要参数更新{r->cache->last_modified = u->headers_in.last_modified_time; //更新缓存的last_modified_time值if (u->ag) { //更新缓存的etag值r->cache->etag = u->ag->value;} else {ngx_str_null(&r->cache->etag); //etag赋空值}} else {r->cache->last_modified = -1;ngx_str_null(&r->cache->etag);}if (ngx_http_file_cache_set_header(r, u->buffer.start) != NGX_OK) { //设置缓存头部信息ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}} else { //无valid_sec 即没有有效时间 不可缓存u->cacheable = 0;}}ngx_log_debug1(NGX_LOG_DEBUG_HTTP, c->log, 0,"http cacheable: %d", u->cacheable);if (u->cacheable == 0 && r->cache) { //临时文件清理ngx_http_file_cache_free(r, u->pipe->temp_file);}if (r->header_only && !u->cacheable && !u->store) { //满足只有head请求或者body数据返回条件ngx_http_upstream_finalize_request(r, u, 0);return;}#endif //以下为event_pipe对上游服务器返回的包体处理p = u->pipe;p->output_filter = ngx_http_upstream_output_filter; //event_pipe对响应请求包体处理p->output_ctx = r;p->tag = u->output.tag;p->bufs = u->conf->bufs;p->busy_size = u->conf->busy_buffers_size;p->upstream = u-&tion;p->downstream = c;p->pool = r->pool;p->log = c->log;p->limit_rate = u->conf->limit_rate;p->start_sec = ngx_time();p->cacheable = u->cacheable || u->store; //传入upstream标记值p->temp_file = ngx_pcalloc(r->pool, sizeof(ngx_temp_file_t)); //创建临时文件 以用于缓存包体信息if (p->temp_file == NULL) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}p->temp_file->file.fd = NGX_INVALID_FILE;p->temp_file->file.log = c->log;p->temp_file->path = u->conf->temp_path;p->temp_file->pool = r->pool;if (p->cacheable) {p->temp_file->persistent = 1; //临时文件不会在close调用之后进行销毁#if (NGX_HTTP_CACHE)if (r->cache && r->cache->file_cache->temp_path) {p->temp_file->path = r->cache->file_cache->temp_path; //传入临时文件目录}#endif} else {p->temp_file->log_level = NGX_LOG_WARN;p->temp_file->warn = "an upstream response is buffered ""to a temporary file";}p->max_temp_file_size = u->conf->max_temp_file_size; //传入临时文件配置的大小限制p->temp_file_write_size = u->conf->temp_file_write_size;#if (NGX_THREADS)if (clcf->aio == NGX_HTTP_AIO_THREADS && clcf->aio_write) { //任务提交到线程池p->thread_handler = ngx_http_upstream_thread_handler;p->thread_ctx = r;}#endifp->preread_bufs = ngx_alloc_chain_link(r->pool);if (p->preread_bufs == NULL) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}p->preread_bufs->buf = &u->buffer;p->preread_bufs->next = NULL;u-&led = 1;p->preread_size = u->buffer.last - u->buffer.pos;if (u->cacheable) {p->buf_to_file = ngx_calloc_buf(r->pool); //创建写入到文件的buffer缓冲if (p->buf_to_file == NULL) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}p->buf_to_file->start = u->buffer.start;p->buf_to_file->pos = u->buffer.start;p->buf_to_file->last = u->buffer.pos;p->buf_to_file->temporary = 1;}if (ngx_event_flags & NGX_USE_IOCP_EVENT) {/* the posted aio operation may corrupt a shadow buffer */p->single_buf = 1;}/* TODO: p->free_bufs = 0 if use ngx_create_chain_of_bufs() */p->free_bufs = 1;/** event_pipe would do u->buffer.last += p->preread_size* as though these bytes were read*/u->buffer.last = u->buffer.pos;if (u->conf->cyclic_temp_file) { //临时文件复用标记/** we need to disable the use of sendfile() if we use cyclic temp file* because the writing a new data may interfere with sendfile()* that uses the same kernel file pages (at least on FreeBSD)*/p->cyclic_temp_file = 1;c->sendfile = 0;} else {p->cyclic_temp_file = 0;}p->read_timeout = u->conf->read_timeout;p->send_timeout = clcf->send_timeout;p->send_lowat = clcf->send_lowat;p->length = -1;if (u->input_filter_init&& u->input_filter_init(p->input_ctx) != NGX_OK) //上游服务器返回包体filter初始化{ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}u->read_event_handler = ngx_http_upstream_process_upstream; //上游包体数据接收处理r->write_event_handler = ngx_http_upstream_process_downstream; //下游包体数据传送处理ngx_http_upstream_process_upstream(r, u); //上游包体处理}static voidngx_http_upstream_process_upstream(ngx_http_request_t *r,ngx_http_upstream_t *u){...c = u-&tion;p = u->pipe;rev = c->read;ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,"http upstream process upstream");c->log->action = "reading upstream";if (rev->timedout) { //upstream包体读取超时p->upstream_error = 1;ngx_connection_error(c, NGX_ETIMEDOUT, "upstream timed out");} else {if (rev->delayed) { //upstream包体延时接收ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,"http upstream delayed");if (ngx_handle_read_event(rev, 0) != NGX_OK) { //upstream包体处理ngx_http_upstream_finalize_request(r, u, NGX_ERROR);}return;}if (ngx_event_pipe(p, 0) == NGX_ABORT) { //转入event_pipe对包体处理ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}}ngx_http_upstream_process_request(r, u); //event_pipe处理结果后续处理}static voidngx_http_upstream_process_request(ngx_http_request_t *r,ngx_http_upstream_t *u){...#if (NGX_THREADS)if (p->writing && !p->aio) {/** make sure to call ngx_event_pipe()* if there is an incomplete aio write*/if (ngx_event_pipe(p, 1) == NGX_ABORT) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}}if (p->writing) { //event_pipe向下游请求端协处理return;}#endifif (u-&tion) { //与上游服务器连接存在if (u->store) {if (p->upstream_eof || p->upstream_done) { //上游包体数据处理完成tf = p->temp_file;if (u->headers_in.status_n == NGX_HTTP_OK&& (p->upstream_done || p->length == -1)&& (u->t_length_n == -1|| u->t_length_n == tf->offset)){ngx_http_upstream_store(r, u); //保存数据}}}#if (NGX_HTTP_CACHE)if (u->cacheable) {if (p->upstream_done) { //upstream包体接收完成 更新缓存ngx_http_file_cache_update(r, p->temp_file);} else if (p->upstream_eof) { //对应上游服务器网络缓冲无数据可读tf = p->temp_file;if (p->length == -1&& (u->t_length_n == -1|| u->t_length_n== tf->offset - (off_t) r->cache->body_start)) //确认upstream包体完全接收 更新缓存{ngx_http_file_cache_update(r, tf);} else {ngx_http_file_cache_free(r->cache, tf); //upstream包体接收不完整 不可进行缓存 直接清理}} else if (p->upstream_error) { //upstream包体接收 处理过程出错 不可缓存 进行清理ngx_http_file_cache_free(r->cache, p->temp_file);}}#endifif (p->upstream_done || p->upstream_eof || p->upstream_error) { //满足upsream终止条件ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,"http upstream exit: %p", p->out);if (p->upstream_done|| (p->upstream_eof && p->length == -1)) //确认upstream包体完整接收 正常终止{ngx_http_upstream_finalize_request(r, u, 0);return;}if (p->upstream_eof) { //网络缓冲区无数据可读 可确认与上游服务器非正常断开 提示错误ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,"upstream prematurely closed connection");}ngx_http_upstream_finalize_request(r, u, NGX_HTTP_BAD_GATEWAY);return;}}if (p->downstream_error) { //下游请求端写入出错 终止处理ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,"http upstream downstream error");if (!u->cacheable && !u->store && u-&tion) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);}}}static voidngx_http_upstream_store(ngx_http_request_t *r, ngx_http_upstream_t *u){size_t root;time_t lm;ngx_str_t path;ngx_temp_file_t *tf;ngx_ext_rename_file_t ext;tf = u->pipe->temp_file;if (tf->file.fd == NGX_INVALID_FILE) { //空的200响应状态码/* create file for empty 200 response */tf = ngx_pcalloc(r->pool, sizeof(ngx_temp_file_t));if (tf == NULL) {return;}tf->file.fd = NGX_INVALID_FILE;tf->file.log = r->connection->log;tf->path = u->conf->temp_path;tf->pool = r->pool;tf->persistent = 1;if (ngx_create_temp_file(&tf->file, tf->path, tf->pool,tf->persistent, tf->clean, tf->access)!= NGX_OK){return;}u->pipe->temp_file = tf;}... //目标文件属性设置if (u->headers_in.last_modified) {lm = ngx_parse_http_time(u->headers_in.last_modified->value.data,u->headers_in.last_modified->value.len);if (lm != NGX_ERROR) {ext.time = lm;ext.fd = tf->file.fd;}}if (u->conf->store_lengths == NULL) {if (ngx_http_map_uri_to_path(r, &path, &root, 0) == NULL) {return;}} else {if (ngx_http_script_run(r, &path, u->conf->store_lengths->elts, 0,u->conf->store_values->elts) //生成完成路径== NULL){return;}}path.len--;ngx_log_debug2(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,"upstream stores "%s" to "%s"",tf->file.name.data, path.data);(void) ngx_ext_rename_file(&tf->file.name, &path, &ext); //event_pipe 临时文件替换成指定全路径文件u->store = 0;}static voidngx_http_upstream_next(ngx_http_request_t *r, ngx_http_upstream_t *u,ngx_uint_t ft_type){...ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,"http next upstream, %xi", ft_type);if (u->peer.sockaddr) { //原有的peer数据释放回收if (ft_type == NGX_HTTP_UPSTREAM_FT_HTTP_403|| ft_type == NGX_HTTP_UPSTREAM_FT_HTTP_404){state = NGX_PEER_NEXT;} else {state = NGX_PEER_FAILED;}u->peer.free(&u->peer, u->peer.data, state);u->peer.sockaddr = NULL;}if (ft_type == NGX_HTTP_UPSTREAM_FT_TIMEOUT) { //超时错误提示ngx_log_error(NGX_LOG_ERR, r->connection->log, NGX_ETIMEDOUT,"upstream timed out");}if (u->peer.cached && ft_type == NGX_HTTP_UPSTREAM_FT_ERROR) {/* TODO: inform balancer instead */u-&ies++;}switch (ft_type) {... //根据fail_type进行状态设置/** NGX_HTTP_UPSTREAM_FT_BUSY_LOCK and NGX_HTTP_UPSTREAM_FT_MAX_WAITING* never reach here*/default:status = NGX_HTTP_BAD_GATEWAY;}if (r->connection->error) { //发现客户端已经断开连接 终止处理ngx_http_upstream_finalize_request(r, u,NGX_HTTP_CLIENT_CLOSED_REQUEST);return;}u->state->status = status;timeout = u->conf->next_upstream_timeout;if (u->request_sent&& (r->method & (NGX_HTTP_POST|NGX_HTTP_LOCK|NGX_HTTP_PATCH))) //请求头部已经发送并且请求method是非幂等的{ft_type |= NGX_HTTP_UPSTREAM_FT_NON_IDEMPOTENT;}if (u-&ies == 0 //达到重试次数上限|| ((u->conf->next_upstream & ft_type) != ft_type)|| (u->request_sent && r->request_body_no_buffering)|| (timeout && ngx_current_msec - u->peer.start_time >= timeout)){#if (NGX_HTTP_CACHE)if (u->cache_status == NGX_HTTP_CACHE_EXPIRED&& ((u->conf->cache_use_stale & ft_type) || r->cache->stale_error)) //用原有的缓存替代错误状态信息{ngx_int_t rc;rc = u->reinit_request(r); //重新初始化upstream请求if (rc != NGX_OK) {ngx_http_upstream_finalize_request(r, u, rc);return;}u->cache_status = NGX_HTTP_CACHE_STALE;rc = ngx_http_upstream_cache_send(r, u);if (rc == NGX_DONE) {return;}if (rc == NGX_HTTP_UPSTREAM_INVALID_HEADER) {//发现缓存头部是无效的rc = NGX_HTTP_INTERNAL_SERVER_ERROR;}ngx_http_upstream_finalize_request(r, u, rc);return;}#endifngx_http_upstream_finalize_request(r, u, status);return;}if (u-&tion) { //与上游服务器的连接依旧存在 释放连接ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,"close http upstream connection: %d",u-&tion->fd);#if (NGX_HTTP_SSL)if (u-&tion->ssl) {u-&tion->ssl->no_wait_shutdown = 1;u-&tion->ssl->no_send_shutdown = 1;(void) ngx_ssl_shutdown(u-&tion);}#endifif (u-&tion->pool) { //连接使用的内存池释放ngx_destroy_pool(u-&tion->pool);}ngx_close_connection(u-&tion); //关闭与上游服务器的连接u-&tion = NULL;}ngx_http_upstream_connect(r, u); //连接至上游服务器}static voidngx_http_upstream_upgrade(ngx_http_request_t *r, ngx_http_upstream_t *u){.../* TODO: prevent upgrade if not requested or not possible */if (r != r->main) { //不允许子请求进行协议转换ngx_log_error(NGX_LOG_ERR, c->log, 0,"connection upgrade in subrequest");ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}r->keepalive = 0;c->log->action = "proxying upgraded connection";...//事件处理设置if (clcf->tcp_nodelay) { //对请求端连接tcp_nodelay设置 和对上游服务器进行tcp_nodelay设置if (ngx_tcp_nodelay(c) != NGX_OK) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}if (ngx_tcp_nodelay(u-&tion) != NGX_OK) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}}if (ngx_http_send_special(r, NGX_HTTP_FLUSH) == NGX_ERROR) { //响应到请求端的buffer缓存刷新标记ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}if (u-&tion->read->ready|| u->buffer.pos != u->buffer.last){ngx_post_event(c->read, &ngx_posted_events);ngx_http_upstream_process_upgraded(r, 1, 1); //upgrage请求端return;}ngx_http_upstream_process_upgraded(r, 0, 1); //upgrage上游服务器}static voidngx_http_upstream_process_upgraded(ngx_http_request_t *r,ngx_uint_t from_upstream, ngx_uint_t do_write){...c = r->connection;u = r->upstream;ngx_log_debug1(NGX_LOG_DEBUG_HTTP, c->log, 0,"http upstream process upgraded, fu:%ui", from_upstream);downstream = c;upstream = u-&tion;if (downstream->write->timedout) { //请求端写事件超时 终止upgradec->timedout = 1;ngx_connection_error(c, NGX_ETIMEDOUT, "client timed out");ngx_http_upstream_finalize_request(r, u, NGX_HTTP_REQUEST_TIME_OUT);return;}if (upstream->read->timedout || upstream->write->timedout) { //上游服务器读/写超时 终止upgragengx_connection_error(c, NGX_ETIMEDOUT, "upstream timed out");ngx_http_upstream_finalize_request(r, u, NGX_HTTP_GATEWAY_TIME_OUT);return;}//先从请求端获取upgrade数据流 之后将upgrade流发送到上游服务器if (from_upstream) { //从upstream上游服务器到请求端更新src = upstream;dst = downstream;b = &u->buffer;} else { //和前面相反src = downstream;dst = upstream;b = &u->from_client;if (r->header_in->last > r->header_in->pos) { //请求头部缓冲还在 do_write标记 (注意到复用已有的buffer)b = r->header_in;b->end = b->last;do_write = 1;}if (b->start == NULL) {b->start = ngx_palloc(r->pool, u->conf->buffer_size);if (b->start == NULL) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}b->pos = b->start;b->last = b->start;b->end = b->start + u->conf->buffer_size;b->temporary = 1;b->tag = u->output.tag;}}for ( ;; ) {if (do_write) {size = b->last - b->pos;if (size && dst->write->ready) {n = dst->send(dst, b->pos, size); //dst upgrade发送 upstream上游服务器upgrageif (n == NGX_ERROR) {ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}if (n > 0) {b->pos += n;if (b->pos == b->last) {b->pos = b->start;b->last = b->start;}}}}size = b->end - b->last;if (size && src->read->ready) {n = src->recv(src, b->last, size); //src 接收 请求端upgrageif (n == NGX_AGAIN || n == 0) {break;}if (n > 0) {do_write = 1;b->last += n;if (from_upstream) {u->state->bytes_received += n;}continue;}if (n == NGX_ERROR) {src->read->eof = 1;}}break;}if ((upstream->read->eof && u->buffer.pos == u->buffer.last)|| (downstream->read->eof && u->from_client.pos == u->from_client.last)|| (downstream->read->eof && upstream->read->eof)){ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,"http upstream upgraded done");ngx_http_upstream_finalize_request(r, u, 0);return;}clcf = ngx_http_get_module_loc_conf(r, ngx_http_core_module);if (ngx_handle_write_event(upstream->write, u->conf->send_lowat)!= NGX_OK){ngx_http_upstream_finalize_request(r, u, NGX_ERROR);return;}...}upstream终止请求ngx_http_upstream_finalize_request 处理流程:static voidngx_http_upstream_finalize_request(ngx_http_request_t *r,ngx_http_upstream_t *u, ngx_int_t rc){ngx_uint_t flush;ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,"finalize http upstream request: %i", rc);if (u->cleanup == NULL) {...//请求已经终止}*u->cleanup = NULL;u->cleanup = NULL;...//释放域名解析相关if (u->state && u->state->response_time) {... //更新state状态}u->finalize_request(r, rc); //介入接口处理... //释放连接信息if (u-&tion) {#if (NGX_HTTP_SSL)...//ssl释放#endifngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,"close http upstream connection: %d",u-&tion->fd);if (u-&tion->pool) {ngx_destroy_pool(u-&tion->pool); //内存池释放}ngx_close_connection(u-&tion); //tcp连接关闭}...#if (NGX_HTTP_CACHE)if (r->cache) {if (u->cacheable) {if (rc == NGX_HTTP_BAD_GATEWAY || rc == NGX_HTTP_GATEWAY_TIME_OUT) { //回源获取数据出错 只刷新缓存有效时间time_t valid;valid = ngx_http_file_cache_valid(u->conf->cache_valid, rc);if (valid) {r->cache->valid_sec = ngx_time() + valid;r->cache->error = rc;}}}...}#endifif (r->subrequest_in_memory //标记子请求得到的上游数据响应只在内存中进行处理&& u->headers_in.status_n >= NGX_HTTP_SPECIAL_RESPONSE){u->buffer.last = u->buffer.pos; //有子请求在内存中 重置buffer 以供给子请求使用 重置只更新内存使用位置标记 不会释放内存}r->read_event_handler = ngx_http_block_reading;if (rc == NGX_DECLINED) {return;}r->connection->log->action = "sending to client";...flush = 0;if (rc >= NGX_HTTP_SPECIAL_RESPONSE) {rc = NGX_ERROR;flush = 1;} ...本文发布于:2024-01-28 08:37:53，感谢您对本站的认可！
本文链接：https://www.4u4v.net/it/17064022796162.html
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系，我们将在24小时内删除。
| 留言与评论（共有 0 条评论） |
