proxy_util.c

来自「linux网络服务器工具」· C语言 代码 · 共 2,062 行 · 第 1/5 页

         * network filters get flushed to the network. This is needed since
         * these buckets have been created with the bucket allocator of the
         * backend connection. This allocator either gets destroyed if
         * conn->close is set or the worker address is not reusable which
         * causes the connection to the backend to be closed or it will be used
         * again by another frontend connection that wants to recycle the
         * backend connection.
         * In this case we could run into nasty race conditions (e.g. if the
         * next user of the backend connection destroys the allocator before we
         * sent the buckets to the network).
         *
         * Remark 1: Only do this if buckets where sent down the chain before
         * that could still be buffered in the network filter. This is the case
         * if we have sent an EOS bucket or if we actually sent buckets with
         * data down the chain. In all other cases we either have not sent any
         * buckets at all down the chain or we only sent meta buckets that are
         * not EOS buckets down the chain. The only meta bucket that remains in
         * this case is the flush bucket which would have removed all possibly
         * buffered buckets in the network filter.
         * If we sent a flush bucket in the case where not ANY buckets were
         * sent down the chain, we break error handling which happens AFTER us.
         *
         * Remark 2: Doing a setaside does not help here as the buckets remain
         * created by the wrong allocator in this case.
         *
         * Remark 3: Yes, this creates a possible performance penalty in the case
         * of pipelined requests as we may send only a small amount of data over
         * the wire.
         */
        c = r->connection;
        bb = apr_brigade_create(r->pool, c->bucket_alloc);
        if (r->eos_sent) {
            /*
             * If we have already sent an EOS bucket send directly to the
             * connection based filters. We just want to flush the buckets
             * if something hasn't been sent to the network yet.
             */
            ap_fflush(c->output_filters, bb);
        }
        else {
            ap_fflush(r->output_filters, bb);
        }
        apr_brigade_destroy(bb);
        conn->r = NULL;
        conn->need_flush = 0;
    }

    /* determine if the connection need to be closed */
    if (conn->close_on_recycle || conn->close || worker->disablereuse ||
        !worker->is_address_reusable) {
        apr_pool_t *p = conn->pool;
        apr_pool_clear(p);
        conn = apr_pcalloc(p, sizeof(proxy_conn_rec));
        conn->pool = p;
        conn->worker = worker;
        apr_pool_create(&(conn->scpool), p);
        apr_pool_tag(conn->scpool, "proxy_conn_scpool");
    }
#if APR_HAS_THREADS
    if (worker->hmax && worker->cp->res) {
        conn->inreslist = 1;
        apr_reslist_release(worker->cp->res, (void *)conn);
    }
    else
#endif
    {
        worker->cp->conn = conn;
    }

    /* Always return the SUCCESS */
    return APR_SUCCESS;
}

static void socket_cleanup(proxy_conn_rec *conn)
{
    conn->sock = NULL;
    conn->connection = NULL;
    apr_pool_clear(conn->scpool);
}

PROXY_DECLARE(apr_status_t) ap_proxy_ssl_connection_cleanup(proxy_conn_rec *conn,
                                                            request_rec *r)
{
    apr_bucket_brigade *bb;
    apr_status_t rv;

    /*
     * If we have an existing SSL connection it might be possible that the
     * server sent some SSL message we have not read so far (e.g. a SSL
     * shutdown message if the server closed the keepalive connection while
     * the connection was held unused in our pool).
     * So ensure that if present (=> APR_NONBLOCK_READ) it is read and
     * processed. We don't expect any data to be in the returned brigade.
     */
    if (conn->sock && conn->connection) {
        bb = apr_brigade_create(r->pool, r->connection->bucket_alloc);
        rv = ap_get_brigade(conn->connection->input_filters, bb,
                            AP_MODE_READBYTES, APR_NONBLOCK_READ,
                            HUGE_STRING_LEN);
        if ((rv != APR_SUCCESS) && !APR_STATUS_IS_EAGAIN(rv)) {
            socket_cleanup(conn);
        }
        if (!APR_BRIGADE_EMPTY(bb)) {
            apr_off_t len;

            rv = apr_brigade_length(bb, 0, &len);
            ap_log_rerror(APLOG_MARK, APLOG_DEBUG, rv, r,
                          "proxy: SSL cleanup brigade contained %"
                          APR_OFF_T_FMT " bytes of data.", len);
        }
        apr_brigade_destroy(bb);
    }
    return APR_SUCCESS;
}

/* reslist constructor */
static apr_status_t connection_constructor(void **resource, void *params,
                                           apr_pool_t *pool)
{
    apr_pool_t *ctx;
    apr_pool_t *scpool;
    proxy_conn_rec *conn;
    proxy_worker *worker = (proxy_worker *)params;

    /*
     * Create the subpool for each connection
     * This keeps the memory consumption constant
     * when disconnecting from backend.
     */
    apr_pool_create(&ctx, pool);
    apr_pool_tag(ctx, "proxy_conn_pool");
    /*
     * Create another subpool that manages the data for the
     * socket and the connection member of the proxy_conn_rec struct as we
     * destroy this data more frequently than other data in the proxy_conn_rec
     * struct like hostname and addr (at least in the case where we have
     * keepalive connections that timed out).
     */
    apr_pool_create(&scpool, ctx);
    apr_pool_tag(scpool, "proxy_conn_scpool");
    conn = apr_pcalloc(ctx, sizeof(proxy_conn_rec));

    conn->pool   = ctx;
    conn->scpool = scpool;
    conn->worker = worker;
#if APR_HAS_THREADS
    conn->inreslist = 1;
#endif
    *resource = conn;

    return APR_SUCCESS;
}

#if APR_HAS_THREADS /* only needed when threads are used */
/* reslist destructor */
static apr_status_t connection_destructor(void *resource, void *params,
                                          apr_pool_t *pool)
{
    proxy_conn_rec *conn = (proxy_conn_rec *)resource;

    /* Destroy the pool only if not called from reslist_destroy */
    if (conn->worker->cp->pool) {
        apr_pool_destroy(conn->pool);
    }

    return APR_SUCCESS;
}
#endif

/*
 * ap_proxy_initialize_worker_share() concerns itself
 * with initializing those parts of worker which
 * are, or could be, shared. Basically worker->s
 */
PROXY_DECLARE(void) ap_proxy_initialize_worker_share(proxy_server_conf *conf,
                                                     proxy_worker *worker,
                                                     server_rec *s)
{
#if PROXY_HAS_SCOREBOARD
    lb_score *score = NULL;
#else
    void *score = NULL;
#endif

    if (PROXY_WORKER_IS_INITIALIZED(worker)) {
        /* The worker share is already initialized */
        ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
              "proxy: worker %s already initialized",
              worker->name);
        return;
    }
#if PROXY_HAS_SCOREBOARD
        /* Get scoreboard slot */
    if (ap_scoreboard_image) {
        score = ap_get_scoreboard_lb(worker->id);
        if (!score) {
            ap_log_error(APLOG_MARK, APLOG_ERR, 0, s,
                  "proxy: ap_get_scoreboard_lb(%d) failed in child %" APR_PID_T_FMT " for worker %s",
                  worker->id, getpid(), worker->name);
        }
        else {
             ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
                  "proxy: grabbed scoreboard slot %d in child %" APR_PID_T_FMT " for worker %s",
                  worker->id, getpid(), worker->name);
        }
    }
#endif
    if (!score) {
        score = apr_pcalloc(conf->pool, sizeof(proxy_worker_stat));
        ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
              "proxy: initialized plain memory in child %" APR_PID_T_FMT " for worker %s",
              getpid(), worker->name);
    }
    worker->s = (proxy_worker_stat *)score;
    /*
     * recheck to see if we've already been here. Possible
     * if proxy is using scoreboard to hold shared stats
     */
    if (PROXY_WORKER_IS_INITIALIZED(worker)) {
        /* The worker share is already initialized */
        ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
              "proxy: worker %s already initialized",
              worker->name);
        return;
    }
    if (worker->route) {
        strcpy(worker->s->route, worker->route);
    }
    else {
        *worker->s->route = '\0';
    }
    if (worker->redirect) {
        strcpy(worker->s->redirect, worker->redirect);
    }
    else {
        *worker->s->redirect = '\0';
    }

    worker->s->status |= (worker->status | PROXY_WORKER_INITIALIZED);

}

PROXY_DECLARE(apr_status_t) ap_proxy_initialize_worker(proxy_worker *worker, server_rec *s)
{
    apr_status_t rv;

#if APR_HAS_THREADS
    int mpm_threads;
#endif

    if (worker->status & PROXY_WORKER_INITIALIZED) {
        /* The worker is already initialized */
        return APR_SUCCESS;
    }

    /* Set default parameters */
    if (!worker->retry_set) {
        worker->retry = apr_time_from_sec(PROXY_WORKER_DEFAULT_RETRY);
    }
    /* By default address is reusable unless DisableReuse is set */
    if (worker->disablereuse) {
        worker->is_address_reusable = 0;
    }
    else {
        worker->is_address_reusable = 1;
    }

#if APR_HAS_THREADS
    ap_mpm_query(AP_MPMQ_MAX_THREADS, &mpm_threads);
    if (mpm_threads > 1) {
        /* Set hard max to no more then mpm_threads */
        if (worker->hmax == 0 || worker->hmax > mpm_threads) {
            worker->hmax = mpm_threads;
        }
        if (worker->smax == 0 || worker->smax > worker->hmax) {
            worker->smax = worker->hmax;
        }
        /* Set min to be lower then smax */
        if (worker->min > worker->smax) {
            worker->min = worker->smax;
        }
    }
    else {
        /* This will supress the apr_reslist creation */
        worker->min = worker->smax = worker->hmax = 0;
    }
    if (worker->hmax) {
        rv = apr_reslist_create(&(worker->cp->res),
                                worker->min, worker->smax,
                                worker->hmax, worker->ttl,
                                connection_constructor, connection_destructor,
                                worker, worker->cp->pool);

        apr_pool_cleanup_register(worker->cp->pool, (void *)worker,
                                  conn_pool_cleanup,
                                  apr_pool_cleanup_null);

        ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
            "proxy: initialized worker %d in child %" APR_PID_T_FMT " for (%s) min=%d max=%d smax=%d",
             worker->id, getpid(), worker->hostname, worker->min,
             worker->hmax, worker->smax);

#if (APR_MAJOR_VERSION > 0)
        /* Set the acquire timeout */
        if (rv == APR_SUCCESS && worker->acquire_set) {
            apr_reslist_timeout_set(worker->cp->res, worker->acquire);
        }
#endif
    }
    else
#endif
    {

        rv = connection_constructor((void **)&(worker->cp->conn), worker, worker->cp->pool);
        ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
             "proxy: initialized single connection worker %d in child %" APR_PID_T_FMT " for (%s)",
             worker->id, getpid(), worker->hostname);
    }
    if (rv == APR_SUCCESS) {
        worker->status |= (PROXY_WORKER_INITIALIZED);
    }
    return rv;
}

PROXY_DECLARE(int) ap_proxy_retry_worker(const char *proxy_function,
                                         proxy_worker *worker,
                                         server_rec *s)
{
    if (worker->s->status & PROXY_WORKER_IN_ERROR) {
        ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
                    "proxy: %s: retrying the worker for (%s)",
                     proxy_function, worker->hostname);
        if (apr_time_now() > worker->s->error_time + worker->retry) {
            ++worker->s->retries;
            worker->s->status &= ~PROXY_WORKER_IN_ERROR;
            ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
                         "proxy: %s: worker for (%s) has been marked for retry",
                         proxy_function, worker->hostname);
            return OK;
        }
        else {
            return DECLINED;
        }
    }
    else {
        return OK;
    }
}

PROXY_DECLARE(int) ap_proxy_acquire_connection(const char *proxy_function,
                                               proxy_conn_rec **conn,
                                               proxy_worker *worker,
                                               server_rec *s)
{
    apr_status_t rv;

    if (!PROXY_WORKER_IS_USABLE(worker)) {
        /* Retry the worker */
        ap_proxy_retry_worker(proxy_function, worker, s);

        if (!PROXY_WORKER_IS_USABLE(worker)) {
            ap_log_error(APLOG_MARK, APLOG_ERR, 0, s,
                         "proxy: %s: disabled connection for (%s)",
                         proxy_function, worker->hostname);
            return HTTP_SERVICE_UNAVAILABLE;
        }
    }
#if APR_HAS_THREADS
    if (worker->hmax && worker->cp->res) {
        rv = apr_reslist_acquire(worker->cp->res, (void **)conn);
    }
    else
#endif
    {
        /* create the new connection if the previous was destroyed */
        if (!worker->cp->conn) {
            connection_constructor((void **)conn, worker, worker->cp->pool);
        }
        else {
            *conn = worker->cp->conn;
            worker->cp->conn = NULL;
        }
        rv = APR_SUCCESS;
    }

    if (rv != APR_SUCCESS) {
        ap_log_error(APLOG_MARK, APLOG_ERR, rv, s,
                     "proxy: %s: failed to acquire connection for (%s)",
                     proxy_function, worker->hostname);
        return HTTP_SERVICE_UNAVAILABLE;
    }
    ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
                 "proxy: %s: has acquired connection for (%s)",
                 proxy_function, worker->hostname);

    (*conn)->worker = worker;
    (*conn)->close  = 0;
    (*conn)->close_on_recycle = 0;
#if APR_HAS_THREADS
    (*conn)->inreslist = 0;
#endif

    return OK;
}

PROXY_DECLARE(int) ap_proxy_release_connection(const char *proxy_function,
                                               proxy_conn_rec *conn,
                                               server_rec *s)
{
    ap_log_error(APLOG_MARK, APLOG_DEBUG, 0,