upstream 是nginx作為代理及快取的核心結構 並且請求上游  傳送至下游都能由相關聯的模組進行干預處理

upstream 模組流程處理如下

建立upstream

ngx_int_t
ngx_http_upstream_create(ngx_http_request_t *r)
{
    ngx_http_upstream_t  *u;

    u = r->upstream;

    if (u && u->cleanup) {    //upstream變數存在 並且配置有cleanup 進行清除
        r->main->count++;
        ngx_http_upstream_cleanup(r);
    }
    ...  //建立upstream 必要初始化
    return NGX_OK;
}
 

upstream初始化

void
ngx_http_upstream_init(ngx_http_request_t *r)
{
    ngx_connection_t     *c;

    c = r->connection;

    ngx_log_debug1(NGX_LOG_DEBUG_HTTP, c->log, 0,
                   "http init upstream, client timer: %d", c->read->timer_set);

#if (NGX_HTTP_V2)   //http 2.0初始化處理
    if (r->stream) { 
        ngx_http_upstream_init_request(r);
        return;
    }
#endif

    if (c->read->timer_set) {      //刪除超時定時器
        ngx_del_timer(c->read);
    }
    //upstream 初始化
    ngx_http_upstream_init_request(r);
}
 

ngx_http_upstream_init_request具體實現

static void
ngx_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;
            }
        }
    }

#endif

    u->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;
#endif

        host = &u->resolved->host;

        umcf = ngx_http_get_module_main_conf(r, ngx_http_upstream_module);

        uscfp = umcf->upstreams.elts;

        for (i = 0; i < umcf->upstreams.nelts; 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;
#endif

    if (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->peer.tries > u->conf->next_upstream_tries)
    {
        u->peer.tries = u->conf->next_upstream_tries; //重試次數
    }

    ngx_http_upstream_connect(r, u);     //連線到上游伺服器
}
 

上游伺服器連線ngx_http_upstream_connect處理

static void
ngx_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->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;

    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->writer.connection = c;
    u->writer.limit = 0;

    if (u->request_sent) {
        if (ngx_http_upstream_reinit(r, u) != NGX_OK) {   //重新初始化upstream 
            ngx_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;
    }

#endif

    ngx_http_upstream_send_request(r, u, 1);  //向上遊伺服器傳送請求
}

ngx_http_upstream_send_request 向上遊伺服器傳送請求處理

static void
ngx_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->peer.connection;

    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;
    }
}

ngx_http_upstream_process_header處理上游伺服器響應頭

static void
ngx_http_upstream_process_header(ngx_http_request_t *r, ngx_http_upstream_t *u)
{
    ...

    c = u->peer.connection;

    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);  //讀事件超時 轉移到下一個server
        return;
    }

    if (!u->request_sent && ngx_http_upstream_test_connect(c) != NGX_OK) {  //請求未傳送到上游伺服器 同時連線測試不通過 轉移到下一個serveer
        ngx_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->buffer.end - u->buffer.last);  

        if (n == NGX_AGAIN) {
#if 0
            ngx_add_timer(rev, u->read_timeout);
#endif

            if (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) {           //出錯 嘗試到下一個server
            ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR);
            return;
        }

        u->state->bytes_received += n;        //更新接收到的位元組數

        u->buffer.last += n; //更新buffer last位置

#if 0
        u->valid_header_in = 0;

        u->peer.cached = 0;
#endif

        rc = u->process_header(r);      //process_header介入處理

        if (rc == NGX_AGAIN) {             //頭部資料讀取不完整 

            if (u->buffer.last == u->buffer.end) {   //頭部過大 標記為錯誤 轉向下一臺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鏈中處理的
}

upstream包體處理 ngx_http_upstream_process_body_in_memory

static void
ngx_http_upstream_process_body_in_memory(ngx_http_request_t *r,
    ngx_http_upstream_t *u)
{
    ...

    c = u->peer.connection;
    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對於傳送響應到請求端處理

ngx_http_upstream_send_response 處理流程

static void
ngx_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_nodelay
            ngx_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->peer.connection->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->headers_in.etag) {              //更新快取的etag值
                    r->cache->etag = u->headers_in.etag->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->peer.connection;
    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;
    }
#endif

    p->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->buffer.recycled = 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);  //上游包體處理
}

ngx_http_upstream_process_upstream對上游包體進行處理

static void
ngx_http_upstream_process_upstream(ngx_http_request_t *r,
    ngx_http_upstream_t *u)
{
    ...
    c = u->peer.connection;
    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處理結果後續處理
}

ngx_http_upstream_process_request處理流程

static void
ngx_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;
    }

#endif

    if (u->peer.connection) {       //與上游伺服器連線存在

        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->headers_in.content_length_n == -1
                        || u->headers_in.content_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->headers_in.content_length_n == -1
                        || u->headers_in.content_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);
            }
        }

#endif

        if (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->peer.connection) {
            ngx_http_upstream_finalize_request(r, u, NGX_ERROR);
        }
    }
}

ngx_http_upstream_store 上游響應資料儲存

static void
ngx_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;
}

ngx_http_upstream_next 轉移請求到下一個server流程處理

static void
ngx_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->peer.tries++;
    }

    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->peer.tries == 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;
        }
#endif

        ngx_http_upstream_finalize_request(r, u, status);
        return;
    }

    if (u->peer.connection) {  //與上游伺服器的連線依舊存在 釋放連線
        ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
                       "close http upstream connection: %d",
                       u->peer.connection->fd);
#if (NGX_HTTP_SSL)

        if (u->peer.connection->ssl) {
            u->peer.connection->ssl->no_wait_shutdown = 1;
            u->peer.connection->ssl->no_send_shutdown = 1;

            (void) ngx_ssl_shutdown(u->peer.connection);
        }
#endif

        if (u->peer.connection->pool) {    //連線使用的記憶體池釋放
            ngx_destroy_pool(u->peer.connection->pool);
        }

        ngx_close_connection(u->peer.connection); //關閉與上游伺服器的連線
        u->peer.connection = NULL;
    }

    ngx_http_upstream_connect(r, u);  //連線至上游伺服器
}

ngx_http_upstream_upgrade 服務端遵從客戶端協議轉換 (HTTP1.1協議新增)處理 狀態碼101

static void
ngx_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->peer.connection) != 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->peer.connection->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上游伺服器
}

upstream 對於ngx_http_upstream_process_upgraded 實際的處理過程 upgrade操作http頭部

static void
ngx_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->peer.connection;

    if (downstream->write->timedout) {     //請求端寫事件超時 終止upgrade
        c->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) { //上游伺服器讀/寫超時  終止upgrage
        ngx_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上游伺服器upgrage

                if (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 接收 請求端upgrage 

            if (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 void
ngx_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->peer.connection) {

#if (NGX_HTTP_SSL)

      ...//ssl釋放
#endif

        ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
                       "close http upstream connection: %d",
                       u->peer.connection->fd);

        if (u->peer.connection->pool) {
            ngx_destroy_pool(u->peer.connection->pool);     //記憶體池釋放
        }

        ngx_close_connection(u->peer.connection);  //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;
                }
            }
        }

       ...
    }

#endif

    if (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;
    }  ...

1."Set-Cookie" 設定到請求端的cookie  在對請求端的輸出頭生成cookie hash表 如果 upstream配置有rewrite_cookie處理 就執行

2.rewrite_redirect 處理 上游資料返回http頭部有Refresh屬性  rewrite_redirect 存在 就執行

3.上游資料返回的http頭部包含cache_control   no-cache no-store private 條件下 upstream cacheable為假 也就是不進行快取

s-maxage= max-age=屬性 會設定快取的有效時間

stale-while-revalidate= 屬性 upstream將不進行快取 設定快取updating狀態時長 

stale-if-error= upstream將不進行快取 設定error_sec時長

4.Expire頭部 設定快取的有效時間

5.Connection頭部 設定上游伺服器返回的頭部connection狀態資訊 是否為close狀態

6.Vary 頭部 設定快取的Vary值

7.X-Accel-Expires 設定快取的有效時間

8.X-Accel-Limit-Rate設定請求的速度

9.X-Accel-Buffering設定是否進行buffering 

10.X-Accel-Charset頭 設定字符集

11.Transfer-Encoding頭 主要是chunked傳輸判斷

upstream處理流程圖解