标签：http   io   os   使用   java   ar   for   文件   数据   

函数定义
ngx_int_t ngx_output_chain(ngx_output_chain_ctx_t ctx, ngx_chain_t in)

函数目的是发送 in 中的数据，ctx 用来保存发送的上下文，因为发送通常情况下，不能一次完成。nginx 因为使用了 ET 模式，在网络编程事件管理上简单了，但是编程中处理事件复杂了，需要不停的循环做处理；事件的函数回掉，次数也不确定，因此需要使用 context 上下文对象来保存发送到什么环节了。

http proxy 模块，依赖 http upstream 模块，ngx_http_upstream_send_request 函数会调用 ngx_output_chain 函数发送 client 请求的数据给后端的 server，调用如下：

rc = ngx_output_chain(&u->output, u->request_sent ? NULL : u->request_bufs);

u->request_bufs 是 client 请求的数据

下面详细分析 ngx_output_chain 中的函数和实现机制

ngx_int_t
ngx_output_chain(ngx_output_chain_ctx_t ctx, ngx_chain_t in)
{

off_t         bsize;
ngx_int_t     rc, last;
ngx_chain_t  *cl, *out, **last_out;

if (ctx->in == NULL && ctx->busy == NULL) {
    if (in == NULL) {
        return ctx->output_filter(ctx->filter_ctx, in);
    }

    // 要发送的 buf 只有一个，不需要复制
    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);
    }
}

// 把输出 in 追加到 ctx->in chain 列表后，chain 对象是新建的，buf 对象还是复用 in 中的
if (in) {
    if (ngx_output_chain_add_copy(ctx->pool, &ctx->in, in) == NGX_ERROR) {
        return NGX_ERROR;
    }
}

out = NULL;
last_out = &out;
last = NGX_NONE;

for ( ;; ) {

if (NGX_HAVE_FILE_AIO)

if (ctx->aio) {
        return NGX_AGAIN;
    }

endif

while (ctx->in) { // 遍历 ctx->in chain 列表，处理 in，只会处理一次，如果发送不完数据，下次再进入函数，ctx->in 就是空

        /*
         * cycle while there are the ctx->in bufs
         * and there are the free output bufs to copy in
         */

        bsize = ngx_buf_size(ctx->in->buf);

        if (bsize == 0 && !ngx_buf_special(ctx->in->buf)) {

            ngx_log_error(NGX_LOG_ALERT, ctx->pool->log, 0,
                          "zero size buf in output "
                          "t:%d r:%d f:%d %p %p-%p %p %O-%O",
                          ctx->in->buf->temporary,
                          ctx->in->buf->recycled,
                          ctx->in->buf->in_file,
                          ctx->in->buf->start,
                          ctx->in->buf->pos,
                          ctx->in->buf->last,
                          ctx->in->buf->file,
                          ctx->in->buf->file_pos,
                          ctx->in->buf->file_last);

            ngx_debug_point();

            ctx->in = ctx->in->next;

            continue;
        }

        if (ngx_output_chain_as_is(ctx, ctx->in->buf)) {
            // buf 不需要复制
            // 结束后，out 指向 ctx->in 指向的 chain
            // ctx->in 指向下一个 chain

            cl = ctx->in;
            ctx->in = cl->next;

            *last_out = cl;
            last_out = &cl->next;
            cl->next = NULL;

            continue;
        }

        if (ctx->buf == NULL) {

            rc = ngx_output_chain_align_file_buf(ctx, bsize);

            if (rc == NGX_ERROR) {
                return NGX_ERROR;
            }

            if (rc != NGX_OK) {

                if (ctx->free) {

                    /* get the free buf */

                    cl = ctx->free;
                    ctx->buf = cl->buf;
                    ctx->free = cl->next;

                    ngx_free_chain(ctx->pool, cl);

                } else if (out || ctx->allocated == ctx->bufs.num) {

                    break;

                } else if (ngx_output_chain_get_buf(ctx, bsize) != NGX_OK) {
                    return NGX_ERROR;
                }
            }
        }

        rc = ngx_output_chain_copy_buf(ctx); // buf 复制

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

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

    if (out == NULL && last != NGX_NONE) {

        if (ctx->in) {
            return NGX_AGAIN;
        }

        return last;
    }

    last = ctx->output_filter(ctx->filter_ctx, out); // 发送数据，filter_ctx 是一个 ngx_chain_writer_ctx_t 对象；nginx 中各种 context 对象都很重要，因为它们需要记录很多状态

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

}

详细分析 ngx_output_chain_add_copy 函数

函数调用为 ngx_output_chain_add_copy(ctx->pool, &ctx->in, in)

参数 ctx->pool 就是请求中的 pool，参数 ctx->in ，参数 in 是要处理的数据，也就是 upstream 中的 request_bufs

此函数的最终效果：创建新的 chain 对象追加到 ctx->in 列表中，这些对象指向输入 in 中的 buf 对象

static ngx_int_t
ngx_output_chain_add_copy(ngx_pool_t pool, ngx_chain_t *chain,

ngx_chain_t *in)

{

ngx_chain_t  *cl, **ll;

if (NGX_SENDFILE_LIMIT)

ngx_buf_t    *b, *buf;

endif

ll = chain; // chain 指向 ctx 中的 in，in 初始为 null

// 循环接收后，ll 指向 ctx 中 in chain 的最后一个 chain 的 next 变量
for (cl = *chain; cl; cl = cl->next) {
    ll = &cl->next;
}

while (in) { // 遍历要处理的 chain

    cl = ngx_alloc_chain_link(pool);     // 从池中获取一个 chain对象
    if (cl == NULL) {
        return NGX_ERROR;
    }

if (NGX_SENDFILE_LIMIT)

buf = in->buf;

    if (buf->in_file
        && buf->file_pos < NGX_SENDFILE_LIMIT
        && buf->file_last > NGX_SENDFILE_LIMIT)
    { // buf 分裂，此分之先不看
        /* split a file buf on two bufs by the sendfile limit */

        b = ngx_calloc_buf(pool);
        if (b == NULL) {
            return NGX_ERROR;
        }

        ngx_memcpy(b, buf, sizeof(ngx_buf_t));

        if (ngx_buf_in_memory(buf)) {
            buf->pos += (ssize_t) (NGX_SENDFILE_LIMIT - buf->file_pos);
            b->last = buf->pos;
        }

        buf->file_pos = NGX_SENDFILE_LIMIT;
        b->file_last = NGX_SENDFILE_LIMIT;

        cl->buf = b;

    } else {
        cl->buf = buf;
        in = in->next;
    }

else

cl->buf = in->buf; // buf 对象还是原来的，chain 对象是新创建的
    in = in->next;

endif

cl->next = NULL;
    *ll = cl;     // 串联到 ll 上，ll 就是 ctx->in 最后一个 chain 的 next 指针
    ll = &cl->next;
}

return NGX_OK;

}

函数 ngx_alloc_chain_link，从 pool 中拿到一个 chain，此函数会复用池中的 chain

ngx_chain_t *
ngx_alloc_chain_link(ngx_pool_t *pool)
{

ngx_chain_t  *cl;

cl = pool->chain;

if (cl) {     // 池中存在空闲 chain
    pool->chain = cl->next;
    return cl;
}

cl = ngx_palloc(pool, sizeof(ngx_chain_t));
if (cl == NULL) {
    return NULL;
}

return cl;

}

此函数判断 chain 需不需要复制，条件复杂，碰到问题再分析；返回 1 不需要复制，返回 0 需要复制

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;

if (ngx_buf_special(buf)) {     // 特殊 buf 不需要复制
    return 1;
}

if (buf->in_file && buf->file->directio) {
    return 0;
}

sendfile = ctx->sendfile;

if (NGX_SENDFILE_LIMIT)

if (buf->in_file && buf->file_pos >= NGX_SENDFILE_LIMIT) {
    sendfile = 0;
}

endif

if (!sendfile) {

    if (!ngx_buf_in_memory(buf)) {
        return 0;
    }

    buf->in_file = 0;
}

if (ctx->need_in_memory && !ngx_buf_in_memory(buf)) { // 需要在 内存中，但是目前不在内存中的，需要复制
    return 0;
}

if (ctx->need_in_temp && (buf->memory || buf->mmap)) {
    return 0;
}

return 1;

}

创建临时 buf

static ngx_int_t
ngx_output_chain_align_file_buf(ngx_output_chain_ctx_t *ctx, off_t bsize)
{

size_t      size;
ngx_buf_t  *in;

in = ctx->in->buf;

if (in->file == NULL || !in->file->directio) { // ？此标志位以后看，感觉是 nginx 兼容多种情况做的处理
    return NGX_DECLINED;
}

ctx->directio = 1;

size = (size_t) (in->file_pos - (in->file_pos & ~(ctx->alignment - 1)));

if (size == 0) {

    if (bsize >= (off_t) ctx->bufs.size) {
        return NGX_DECLINED;
    }

    size = (size_t) bsize;

} else {
    size = (size_t) ctx->alignment - size;

    if ((off_t) size > bsize) {
        size = (size_t) bsize;
    }
}

ctx->buf = ngx_create_temp_buf(ctx->pool, size);
if (ctx->buf == NULL) {
    return NGX_ERROR;
}

/*
 * we do not set ctx->buf->tag, because we do not want
 * to reuse the buf via ctx->free list
 */

if (NGX_HAVE_ALIGNED_DIRECTIO)

ctx->unaligned = 1;

endif

return NGX_OK;

}

buf 复制，从 ctx->in 到 ctx->buf

可能内存复制，可能从文件中读取数据

static ngx_int_t
ngx_output_chain_copy_buf(ngx_output_chain_ctx_t *ctx)
{

off_t        size;
ssize_t      n;
ngx_buf_t   *src, *dst;
ngx_uint_t   sendfile;

src = ctx->in->buf;
dst = ctx->buf;

size = ngx_buf_size(src);
size = ngx_min(size, dst->end - dst->pos);

sendfile = ctx->sendfile & !ctx->directio;

if (NGX_SENDFILE_LIMIT)

if (src->in_file && src->file_pos >= NGX_SENDFILE_LIMIT) {
    sendfile = 0;
}

endif

if (ngx_buf_in_memory(src)) {
    ngx_memcpy(dst->pos, src->pos, (size_t) size);
    src->pos += (size_t) size;
    dst->last += (size_t) size;

    if (src->in_file) {

        if (sendfile) {
            dst->in_file = 1;
            dst->file = src->file;
            dst->file_pos = src->file_pos;
            dst->file_last = src->file_pos + size;

        } else {
            dst->in_file = 0;
        }

        src->file_pos += size;

    } else {
        dst->in_file = 0;
    }

    if (src->pos == src->last) {
        dst->flush = src->flush;
        dst->last_buf = src->last_buf;
        dst->last_in_chain = src->last_in_chain;
    }

} else {

if (NGX_HAVE_ALIGNED_DIRECTIO)

if (ctx->unaligned) {
        if (ngx_directio_off(src->file->fd) == NGX_FILE_ERROR) {
            ngx_log_error(NGX_LOG_ALERT, ctx->pool->log, ngx_errno,
                          ngx_directio_off_n " \"%s\" failed",
                          src->file->name.data);
        }
    }

endif

if (NGX_HAVE_FILE_AIO)

if (ctx->aio_handler) {
        n = ngx_file_aio_read(src->file, dst->pos, (size_t) size,
                              src->file_pos, ctx->pool);
        if (n == NGX_AGAIN) {
            ctx->aio_handler(ctx, src->file);
            return NGX_AGAIN;
        }

    } else {
        n = ngx_read_file(src->file, dst->pos, (size_t) size,
                          src->file_pos);
    }

else

n = ngx_read_file(src->file, dst->pos, (size_t) size, src->file_pos);

endif

if (NGX_HAVE_ALIGNED_DIRECTIO)

if (ctx->unaligned) {
        ngx_err_t  err;

        err = ngx_errno;

        if (ngx_directio_on(src->file->fd) == NGX_FILE_ERROR) {
            ngx_log_error(NGX_LOG_ALERT, ctx->pool->log, ngx_errno,
                          ngx_directio_on_n " \"%s\" failed",
                          src->file->name.data);
        }

        ngx_set_errno(err);

        ctx->unaligned = 0;
    }

endif

if (n == NGX_ERROR) {
        return (ngx_int_t) n;
    }

    if (n != size) {
        ngx_log_error(NGX_LOG_ALERT, ctx->pool->log, 0,
                      ngx_read_file_n " read only %z of %O from \"%s\"",
                      n, size, src->file->name.data);
        return NGX_ERROR;
    }

    dst->last += n;

    if (sendfile) {
        dst->in_file = 1;
        dst->file = src->file;
        dst->file_pos = src->file_pos;
        dst->file_last = src->file_pos + n;

    } else {
        dst->in_file = 0;
    }

    src->file_pos += n;

    if (src->file_pos == src->file_last) {
        dst->flush = src->flush;
        dst->last_buf = src->last_buf;
        dst->last_in_chain = src->last_in_chain;
    }
}

return NGX_OK;

}

ngx_output_chain 函数分析

标签：http   io   os   使用   java   ar   for   文件   数据   

原文地址：http://my.oschina.net/astute/blog/316954