ngx_http_copy_filter_module分析 · Nginx開發從入門到精通

#### ngx_http_copy_filter_module分析[](http://tengine.taobao.org/book/chapter_12.html#ngx-http-copy-filter-module "永久鏈接至標題") ngx_http_copy_filter_module是響應體過濾鏈（body filter）中非常重要的一個模塊，這個filter模塊主要是來將一些需要復制的buf（可能在文件中，也可能在內存中）重新復制一份交給后面的filter模塊處理。先來看它的初始化函數： [](http:// "點擊提交Issue,反饋你的意見...") static ngx_int_t ngx_http_copy_filter_init(ngx_conf_t *cf) { ngx_http_next_body_filter = ngx_http_top_body_filter; ngx_http_top_body_filter = ngx_http_copy_filter; return NGX_OK; } 可以看到，它只注冊了body filter，而沒有注冊header filter，也就是說只有body filter鏈中才有這個模塊。該模塊有一個命令，命令名為output_buffers，用來配置可用的buffer數和buffer大小，它的值保存在copy filter的loc conf的bufs字段，默認數量為1，大小為32768字節。這個參數具體的作用后面會做介紹。 Nginx中，一般filter模塊可以header filter函數中根據請求響應頭設置一個模塊上下文（context），用來保存相關的信息，在body filter函數中使用這個上下文。而copy filter沒有header filter，因此它的context的初始化也是放在body filter中的，而它的ctx就是ngx_output_chain_ctx_t，為什么名字是output_chain呢，這是因為copy filter的主要邏輯的處理都放在ngx_output_chain模塊中，另外這個模塊在core目錄下，而不是屬于http目錄。接下來看一下上面說到的context結構： [](http:// "點擊提交Issue,反饋你的意見...") struct ngx_output_chain_ctx_s { ngx_buf_t *buf; /* 保存臨時的buf */ ngx_chain_t *in; /* 保存了將要發送的chain */ ngx_chain_t *free; /* 保存了已經發送完畢的chain，以便于重復利用 */ ngx_chain_t *busy; /* 保存了還未發送的chain */ unsigned sendfile:1; /* sendfile標記 */ unsigned directio:1; /* directio標記 */ #if (NGX_HAVE_ALIGNED_DIRECTIO) unsigned unaligned:1; #endif unsigned need_in_memory:1; /* 是否需要在內存中保存一份(使用sendfile的話，內存中沒有文件的拷貝的，而我們有時需要處理文件，此時就需要設置這個標記) */ unsigned need_in_temp:1; /* 是否需要在內存中重新復制一份，不管buf是在內存還是文件, 這樣的話，后續模塊可以直接修改這塊內存 */ #if (NGX_HAVE_FILE_AIO) unsigned aio:1; ngx_output_chain_aio_pt aio_handler; #endif off_t alignment; ngx_pool_t *pool; ngx_int_t allocated; /* 已經分別的buf個數 */ ngx_bufs_t bufs; /* 對應loc conf中設置的bufs */ ngx_buf_tag_t tag; /* 模塊標記，主要用于buf回收 */ ngx_output_chain_filter_pt output_filter; /* 一般是ngx_http_next_filter,也就是繼續調用filter鏈 */ void *filter_ctx; /* 當前filter的上下文，這里是由于upstream也會調用output_chain */ }; 為了更好的理解context結構每個域的具體含義，接下來分析filter的具體實現： [](http:// "點擊提交Issue,反饋你的意見...") static ngx_int_t ngx_http_copy_filter(ngx_http_request_t *r, ngx_chain_t *in) { ngx_int_t rc; ngx_connection_t *c; ngx_output_chain_ctx_t *ctx; ngx_http_core_loc_conf_t *clcf; ngx_http_copy_filter_conf_t *conf; c = r->connection; ngx_log_debug2(NGX_LOG_DEBUG_HTTP, c->log, 0, "http copy filter: \"%V?%V\"", &r->uri, &r->args); /* 獲取ctx */ ctx = ngx_http_get_module_ctx(r, ngx_http_copy_filter_module); /* 如果為空，則說明需要初始化ctx */ if (ctx == NULL) { ctx = ngx_pcalloc(r->pool, sizeof(ngx_output_chain_ctx_t)); if (ctx == NULL) { return NGX_ERROR; } ngx_http_set_ctx(r, ctx, ngx_http_copy_filter_module); conf = ngx_http_get_module_loc_conf(r, ngx_http_copy_filter_module); clcf = ngx_http_get_module_loc_conf(r, ngx_http_core_module); /* 設置sendfile */ ctx->sendfile = c->sendfile; /* 如果request設置了filter_need_in_memory的話，ctx的這個域就會被設置 */ ctx->need_in_memory = r->main_filter_need_in_memory || r->filter_need_in_memory; /* 和上面類似 */ ctx->need_in_temp = r->filter_need_temporary; ctx->alignment = clcf->directio_alignment; ctx->pool = r->pool; ctx->bufs = conf->bufs; ctx->tag = (ngx_buf_tag_t) &ngx_http_copy_filter_module; /* 可以看到output_filter就是下一個body filter節點 */ ctx->output_filter = (ngx_output_chain_filter_pt) ngx_http_next_body_filter; /* 此時filter ctx為當前的請求 */ ctx->filter_ctx = r; ... if (in && in->buf && ngx_buf_size(in->buf)) { r->request_output = 1; } } ... for ( ;; ) { /* 最關鍵的函數，下面會詳細分析 */ rc = ngx_output_chain(ctx, in); if (ctx->in == NULL) { r->buffered &= ~NGX_HTTP_COPY_BUFFERED; } else { r->buffered |= NGX_HTTP_COPY_BUFFERED; } ... return rc; } } 上面的代碼去掉了AIO相關的部分，函數首先設置并初始化context，接著調用ngx_output_chain函數，這個函數實際上包含了copy filter模塊的主要邏輯，它的原型為： ngx_int_t ngx_output_chain(ngx_output_chain_ctx_t?[*](http://tengine.taobao.org/book/chapter_12.html#id12)ctx, ngx_chain_t?[*](http://tengine.taobao.org/book/chapter_12.html#id14)in) 分段來看它的代碼，下面這段代碼是一個快捷路徑（short path），也就是說當能直接確定所有的in chain都不需要復制的時，可以直接調用output_filter來交給剩下的filter去處理： [](http:// "點擊提交Issue,反饋你的意見...") if (ctx->in == NULL && ctx->busy == NULL) { /* * the short path for the case when the ctx->in and ctx->busy chains * are empty, the incoming chain is empty too or has the single buf * that does not require the copy */ if (in == NULL) { return ctx->output_filter(ctx->filter_ctx, in); } if (in->next == NULL #if (NGX_SENDFILE_LIMIT) && !(in->buf->in_file && in->buf->file_last > NGX_SENDFILE_LIMIT) #endif && ngx_output_chain_as_is(ctx, in->buf)) { return ctx->output_filter(ctx->filter_ctx, in); } } 上面可以看到了一個函數ngx_output_chain_as_is，這個函數很關鍵，下面還會再次被調用，這個函數主要用來判斷是否需要復制buf。返回1,表示不需要拷貝，否則為需要拷貝： [](http:// "點擊提交Issue,反饋你的意見...") static ngx_inline ngx_int_t ngx_output_chain_as_is(ngx_output_chain_ctx_t *ctx, ngx_buf_t *buf) { ngx_uint_t sendfile; /* 是否為特殊buf（special buf），是的話返回1，也就是不用拷貝 */ if (ngx_buf_special(buf)) { return 1; } /* 如果buf在文件中，并且使用了directio的話，需要拷貝buf */ if (buf->in_file && buf->file->directio) { return 0; } /* sendfile標記 */ sendfile = ctx->sendfile; #if (NGX_SENDFILE_LIMIT) /* 如果pos大于sendfile的限制，設置標記為0 */ if (buf->in_file && buf->file_pos >= NGX_SENDFILE_LIMIT) { sendfile = 0; } #endif if (!sendfile) { /* 如果不走sendfile，而且buf不在內存中，則我們就需要復制到內存一份 */ if (!ngx_buf_in_memory(buf)) { return 0; } buf->in_file = 0; } /* 如果需要內存中有一份拷貝，而并不在內存中，此時返回0，表示需要拷貝 */ if (ctx->need_in_memory && !ngx_buf_in_memory(buf)) { return 0; } /* 如果需要內存中有可修改的拷貝，并且buf存在于只讀的內存中或者mmap中，則返回0 */ if (ctx->need_in_temp && (buf->memory || buf->mmap)) { return 0; } return 1; } 上面有兩個標記要注意，一個是need_in_memory ，這個主要是用于當使用sendfile的時候，Nginx并不會將請求文件拷貝到內存中，而有時需要操作文件的內容，此時就需要設置這個標記。然后后面的body filter就能操作內容了。第二個是need_in_temp，這個主要是用于把本來就存在于內存中的buf復制一份可修改的拷貝出來，這里有用到的模塊有charset，也就是編解碼 filter。然后接下來這段是復制in chain到ctx->in的結尾，它是通過調用ngx_output_chain_add_copy來進行add copy的，這個函數比較簡單，這里就不分析了，不過只有一個要注意的地方，那就是如果buf是存在于文件中，并且file_pos超過了sendfile limit，此時就會切割buf為兩個buf，然后保存在兩個chain中，最終連接起來： [](http:// "點擊提交Issue,反饋你的意見...") /* add the incoming buf to the chain ctx->in */ if (in) { if (ngx_output_chain_add_copy(ctx->pool, &ctx->in, in) == NGX_ERROR) { return NGX_ERROR; } } 然后就是主要的邏輯處理階段。這里nginx做的非常巧妙也非常復雜，首先是chain的重用，然后是buf的重用。先來看chain的重用。關鍵的幾個結構以及域：ctx的free，busy以及ctx->pool的chain域。其中每次發送沒有發完的chain就放到busy中，而已經發送完畢的就放到free中，而最后會調用 ngx_free_chain來將free的chain放入到pool->chain中,而在ngx_alloc_chain_link中，如果pool->chain中存在chain的話，就不用malloc了，而是直接返回pool->chain，相關的代碼如下： [](http:// "點擊提交Issue,反饋你的意見...") /* 鏈接cl到pool->chain中 */ #define ngx_free_chain(pool, cl) \ cl->next = pool->chain; \ pool->chain = cl /* 從pool中分配chain */ ngx_chain_t * ngx_alloc_chain_link(ngx_pool_t *pool) { ngx_chain_t *cl; cl = pool->chain; /* 如果cl存在，則直接返回cl */ if (cl) { pool->chain = cl->next; return cl; } /* 否則才會malloc chain */ cl = ngx_palloc(pool, sizeof(ngx_chain_t)); if (cl == NULL) { return NULL; } return cl; } 然后是buf的重用，嚴格意義上來說buf的重用是從free中的chain中取得的，當free中的buf被重用，則這個buf對應的chain就會被鏈接到ctx->pool中，從而這個chain就會被重用。也就是說首先考慮的是buf的重用，只有當這個chain的buf確定不需要被重用(或者說已經被重用)的時候，chain才會被鏈接到ctx->pool中被重用。還有一個就是ctx的allocated域，這個域表示了當前的上下文中已經分配了多少個buf，output_buffer命令用來設置output的buf大小以及buf的個數。而allocated如果比output_buffer大的話，則需要先發送完已經存在的buf，然后才能再次重新分配buf。來看代碼，上面所說的重用以及buf的控制，代碼里面都可以看的比較清晰。下面這段主要是拷貝buf前所做的一些工作，比如判斷是否拷貝，以及給buf分貝內存等： [](http:// "點擊提交Issue,反饋你的意見...") /* out為最終需要傳輸的chain，也就是交給剩下的filter處理的chain */ out = NULL; /* last_out為out的最后一個chain */ last_out = &out; last = NGX_NONE; for ( ;; ) { /* 開始遍歷chain */ while (ctx->in) { /* 取得當前chain的buf大小 */ bsize = ngx_buf_size(ctx->in->buf); /* 跳過bsize為0的buf */ if (bsize == 0 && !ngx_buf_special(ctx->in->buf)) { ngx_debug_point(); ctx->in = ctx->in->next; continue; } /* 判斷是否需要復制buf */ if (ngx_output_chain_as_is(ctx, ctx->in->buf)) { /* move the chain link to the output chain */ /* 如果不需要復制，則直接鏈接chain到out，然后繼續循環 */ cl = ctx->in; ctx->in = cl->next; *last_out = cl; last_out = &cl->next; cl->next = NULL; continue; } /* 到達這里，說明我們需要拷貝buf，這里buf最終都會被拷貝進ctx->buf中，因此這里先判斷ctx->buf是否為空 */ if (ctx->buf == NULL) { /* 如果為空，則取得buf，這里要注意，一般來說如果沒有開啟directio的話，這個函數都會返回NGX_DECLINED */ rc = ngx_output_chain_align_file_buf(ctx, bsize); if (rc == NGX_ERROR) { return NGX_ERROR; } /* 大部分情況下，都會落入這個分支 */ if (rc != NGX_OK) { /* 準備分配buf，首先在free中尋找可以重用的buf */ if (ctx->free) { /* get the free buf */ /* 得到free buf */ cl = ctx->free; ctx->buf = cl->buf; ctx->free = cl->next; /* 將要重用的chain鏈接到ctx->poll中，以便于chain的重用 */ ngx_free_chain(ctx->pool, cl); } else if (out || ctx->allocated == ctx->bufs.num) { /* 如果已經等于buf的個數限制，則跳出循環，發送已經存在的buf。這里可以看到如果out存在的話，nginx會跳出循環，然后發送out，等發送完會再次處理，這里很好的體現了nginx的流式處理 */ break; } else if (ngx_output_chain_get_buf(ctx, bsize) != NGX_OK) { /* 上面這個函數也比較關鍵，它用來取得buf。接下來會詳細看這個函數 */ return NGX_ERROR; } } } /* 從原來的buf中拷貝內容或者從文件中讀取內容 */ rc = ngx_output_chain_copy_buf(ctx); if (rc == NGX_ERROR) { return rc; } if (rc == NGX_AGAIN) { if (out) { break; } return rc; } /* delete the completed buf from the ctx->in chain */ if (ngx_buf_size(ctx->in->buf) == 0) { ctx->in = ctx->in->next; } /* 分配新的chain節點 */ cl = ngx_alloc_chain_link(ctx->pool); if (cl == NULL) { return NGX_ERROR; } cl->buf = ctx->buf; cl->next = NULL; *last_out = cl; last_out = &cl->next; ctx->buf = NULL; } ... } 上面的代碼分析的時候有個很關鍵的函數，那就是ngx_output_chain_get_buf，這個函數當沒有可重用的buf時用來分配buf。如果當前的buf位于最后一個chain，則需要特殊處理，一是buf的recycled域，另外是將要分配的buf的大小。先來說recycled域，這個域表示當前的buf需要被回收。而一般情況下Nginx(比如在非last buf)會緩存一部分buf(默認是1460字節)，然后再發送，而設置了recycled的話，就不會讓它緩存buf，也就是盡量發送出去，然后以供回收使用。因此如果是最后一個buf，則不需要設置recycled域的，否則的話，需要設置recycled域。然后就是buf的大小。這里會有兩個大小，一個是需要復制的buf的大小，一個是配置文件中設置的大小。如果不是最后一個buf，則只需要分配配置中設置的buf的大小就行了。如果是最后一個buf，則就處理不太一樣，下面的代碼會看到： [](http:// "點擊提交Issue,反饋你的意見...") static ngx_int_t ngx_output_chain_get_buf(ngx_output_chain_ctx_t *ctx, off_t bsize) { size_t size; ngx_buf_t *b, *in; ngx_uint_t recycled; in = ctx->in->buf; /* 可以看到這里分配的buf，每個buf的大小是配置文件中設置的size */ size = ctx->bufs.size; /* 默認有設置recycled域 */ recycled = 1; /* 如果當前的buf是屬于最后一個chain的時候，需要特殊處理 */ if (in->last_in_chain) { /* 如果buf大小小于配置指定的大小，則直接按實際大小分配，不設置回收標記 */ if (bsize < (off_t) size) { /* * allocate a small temp buf for a small last buf * or its small last part */ size = (size_t) bsize; recycled = 0; } else if (!ctx->directio && ctx->bufs.num == 1 && (bsize < (off_t) (size + size / 4))) { /* * allocate a temp buf that equals to a last buf, * if there is no directio, the last buf size is lesser * than 1.25 of bufs.size and the temp buf is single */ size = (size_t) bsize; recycled = 0; } } /* 開始分配buf內存 */ b = ngx_calloc_buf(ctx->pool); if (b == NULL) { return NGX_ERROR; } if (ctx->directio) { /* directio需要對齊 */ b->start = ngx_pmemalign(ctx->pool, size, (size_t) ctx->alignment); if (b->start == NULL) { return NGX_ERROR; } } else { /* 大部分情況會走到這里 */ b->start = ngx_palloc(ctx->pool, size); if (b->start == NULL) { return NGX_ERROR; } } b->pos = b->start; b->last = b->start; b->end = b->last + size; /* 設置temporary */ b->temporary = 1; b->tag = ctx->tag; b->recycled = recycled; ctx->buf = b; /* 更新allocated,可以看到每分配一個就加1 */ ctx->allocated++; return NGX_OK; } 分配新的buf和chain，并調用ngx_output_chain_copy_buf拷貝完數據之后，Nginx就將新的chain鏈表交給下一個body filter繼續處理： [](http:// "點擊提交Issue,反饋你的意見...") if (out == NULL && last != NGX_NONE) { if (ctx->in) { return NGX_AGAIN; } return last; } last = ctx->output_filter(ctx->filter_ctx, out); if (last == NGX_ERROR || last == NGX_DONE) { return last; } ngx_chain_update_chains(ctx->pool, &ctx->free, &ctx->busy, &out, ctx->tag); last_out = &out; 在其他body filter處理完之后，ngx_output_chain函數還需要更新chain鏈表，以便回收利用，ngx_chain_update_chains函數主要是將處理完畢的chain節點放入到free鏈表，沒有處理完畢的放到busy鏈表中，另外這個函數用到了tag，它只回收copy filter產生的chain節點。