ngx_http_upstream.c

Nginx是一个高性能的HTTP和反向代理服务器

                b->pos = b->start;
                b->last = b->start;
            }
        }

        size = b->end - b->last;

        if (size > u->length) {
            size = u->length;
        }

        if (size && upstream->read->ready) {

            n = upstream->recv(upstream, b->last, size);

            if (n == NGX_AGAIN) {
                break;
            }

            if (n > 0) {
                if (u->input_filter(u->input_filter_ctx, n) == NGX_ERROR) {
                    ngx_http_upstream_finalize_request(r, u, 0);
                    return;
                }
            }

            do_write = 1;

            continue;
        }

        break;
    }

    if (downstream->data == r) {
        if (ngx_handle_write_event(downstream->write, clcf->send_lowat)
            == NGX_ERROR)
        {
            ngx_http_upstream_finalize_request(r, u, 0);
            return;
        }
    }

    if (downstream->write->active && !downstream->write->ready) {
        ngx_add_timer(downstream->write, clcf->send_timeout);

    } else if (downstream->write->timer_set) {
        ngx_del_timer(downstream->write);
    }

    if (ngx_handle_read_event(upstream->read, 0) == NGX_ERROR) {
        ngx_http_upstream_finalize_request(r, u, 0);
        return;
    }

    if (upstream->read->active && !upstream->read->ready) {
        ngx_add_timer(upstream->read, u->conf->read_timeout);

    } else if (upstream->read->timer_set) {
        ngx_del_timer(upstream->read);
    }
}


static ngx_int_t
ngx_http_upstream_non_buffered_filter_init(void *data)
{
    return NGX_OK;
}


static ngx_int_t
ngx_http_upstream_non_buffered_filter(void *data, ssize_t bytes)
{
    ngx_http_request_t  *r = data;

    ngx_buf_t            *b;
    ngx_chain_t          *cl, **ll;
    ngx_http_upstream_t  *u;

    u = r->upstream;

    for (cl = u->out_bufs, ll = &u->out_bufs; cl; cl = cl->next) {
        ll = &cl->next;
    }

    cl = ngx_chain_get_free_buf(r->pool, &u->free_bufs);
    if (cl == NULL) {
        return NGX_ERROR;
    }

    *ll = cl;

    cl->buf->flush = 1;
    cl->buf->memory = 1;

    b = &u->buffer;

    cl->buf->pos = b->last;
    b->last += bytes;
    cl->buf->last = b->last;
    cl->buf->tag = u->output.tag;

    if (u->length == NGX_MAX_SIZE_T_VALUE) {
        return NGX_OK;
    }

    u->length -= bytes;

    return NGX_OK;
}


static void
ngx_http_upstream_process_downstream(ngx_http_request_t *r)
{
    ngx_http_upstream_process_body(r->connection->write);
}


static void
ngx_http_upstream_process_body(ngx_event_t *ev)
{
    ngx_temp_file_t      *tf;
    ngx_event_pipe_t     *p;
    ngx_connection_t     *c, *downstream;
    ngx_http_log_ctx_t   *ctx;
    ngx_http_request_t   *r;
    ngx_http_upstream_t  *u;

    c = ev->data;
    r = c->data;
    u = r->upstream;
    downstream = r->connection;

    if (ev->write) {
        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,
                       "http upstream process downstream");
        c->log->action = "sending to client";

    } else {
        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,
                       "http upstream process upstream");
        c->log->action = "reading upstream";

        ctx = c->log->data;
        ctx->current_request = r;
    }

    p = u->pipe;

    if (ev->timedout) {
        if (ev->write) {
            if (ev->delayed) {

                ev->timedout = 0;
                ev->delayed = 0;

                if (!ev->ready) {
                    ngx_add_timer(ev, p->send_timeout);

                    if (ngx_handle_write_event(ev, p->send_lowat) == NGX_ERROR)
                    {
                        ngx_http_upstream_finalize_request(r, u, 0);
                        return;
                    }

                    return;
                }

                if (ngx_event_pipe(p, ev->write) == NGX_ABORT) {

                    if (downstream->destroyed) {
                        return;
                    }

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

            } else {
                p->downstream_error = 1;
                c->timedout = 1;
                ngx_connection_error(c, NGX_ETIMEDOUT, "client timed out");
            }

        } else {
            p->upstream_error = 1;
            ngx_connection_error(c, NGX_ETIMEDOUT, "upstream timed out");
        }

    } else {
        if (ev->write && ev->delayed) {
            ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,
                           "http downstream delayed");

            if (ngx_handle_write_event(ev, p->send_lowat) == NGX_ERROR) {
                return;
            }

            return;
        }

        if (ngx_event_pipe(p, ev->write) == NGX_ABORT) {

            if (downstream->destroyed) {
                return;
            }

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

    if (u->peer.connection) {

        if (u->store) {

            tf = u->pipe->temp_file;

            if (p->upstream_eof
                && u->headers_in.status_n == NGX_HTTP_OK
                && (u->headers_in.content_length_n == -1
                    || (u->headers_in.content_length_n == tf->offset)))
            {
                ngx_http_upstream_store(r, u);

            } else if ((p->upstream_error
                        || (p->upstream_eof
                            && u->headers_in.status_n != NGX_HTTP_OK))
                       && tf->file.fd != NGX_INVALID_FILE)
            {
                if (ngx_delete_file(tf->file.name.data) == NGX_FILE_ERROR) {

                    ngx_log_error(NGX_LOG_CRIT, r->connection->log, ngx_errno,
                                  ngx_delete_file_n " \"%s\" failed",
                                  u->pipe->temp_file->file.name.data);
                }
            }
        }

#if (NGX_HTTP_FILE_CACHE)

        if (p->upstream_done && u->cacheable) {
            if (ngx_http_cache_update(r) == NGX_ERROR) {
                ngx_http_busy_unlock(u->conf->busy_lock, &u->busy_lock);
                ngx_http_upstream_finalize_request(r, u, 0);
                return;
            }

        } else if (p->upstream_eof && u->cacheable) {

            /* TODO: check length & update cache */

            if (ngx_http_cache_update(r) == NGX_ERROR) {
                ngx_http_busy_unlock(u->conf->busy_lock, &u->busy_lock);
                ngx_http_upstream_finalize_request(r, u, 0);
                return;
            }
        }

#endif

        if (p->upstream_done || p->upstream_eof || p->upstream_error) {
            ngx_log_debug1(NGX_LOG_DEBUG_HTTP, c->log, 0,
                           "http upstream exit: %p", p->out);
#if 0
            ngx_http_busy_unlock(u->conf->busy_lock, &u->busy_lock);
#endif
            ngx_http_upstream_finalize_request(r, u, 0);
            return;
        }
    }

    if (p->downstream_error) {
        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0,
                       "http upstream downstream error");

        if (!u->cacheable && u->peer.connection) {
            ngx_http_upstream_finalize_request(r, u, 0);
        }
    }
}


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) {

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

    ext.access = u->conf->store_access;
    ext.time = -1;
    ext.create_path = 1;
    ext.delete_file = 1;
    ext.log = r->connection->log;

    if (u->headers_in.last_modified) {

        lm = ngx_http_parse_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) {

        ngx_http_map_uri_to_path(r, &path, &root, 0);

    } else {
        if (ngx_http_script_run(r, &path, u->conf->store_lengths->elts, 0,
                                u->conf->store_values->elts)
            == NULL)
        {
            return;
        }
    }

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


static void
ngx_http_upstream_dummy_handler(ngx_event_t *wev)
{
    ngx_log_debug0(NGX_LOG_DEBUG_HTTP, wev->log, 0,
                   "http upstream dummy handler");
}


static void
ngx_http_upstream_next(ngx_http_request_t *r, ngx_http_upstream_t *u,
    ngx_uint_t ft_type)
{
    ngx_uint_t  status, state;

    ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
                   "http next upstream, %xi", ft_type);

#if 0
    ngx_http_busy_unlock(u->conf->busy_lock, &u->busy_lock);
#endif

    if (ft_type == NGX_HTTP_UPSTREAM_FT_HTTP_404) {
        state = NGX_PEER_NEXT;
    } else {
        state = NGX_PEER_FAILED;
    }

    if (ft_type != NGX_HTTP_UPSTREAM_FT_NOLIVE) {
        u->peer.free(&u->peer, u->peer.data, state);
    }

    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) {
        status = 0;

    } else {
        switch(ft_type) {

        case NGX_HTTP_UPSTREAM_FT_TIMEOUT:
            status = NGX_HTTP_GATEWAY_TIME_OUT;
            break;

        case NGX_HTTP_UPSTREAM_FT_HTTP_500:
            status = NGX_HTTP_INTERNAL_SERVER_ERROR;
            break;

        case NGX_HTTP_UPSTREAM_FT_HTTP_404:
            status = NGX_HTTP_NOT_FOUND;
            break;

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

    if (status) {
        u->state->status = status;

        if (u->peer.tries == 0 || !(u->conf->next_upstream & ft_type)) {

#if (NGX_HTTP_CACHE)

            if (u->stale && (u->conf->use_stale & ft_type)) {
                ngx_http_upstream_finalize_request(r, u,
                                             ngx_http_send_cached_response(r));
                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.c