proxy_util.c

精通tomcat书籍原代码,希望大家共同学习

                    return rv;
                }
                /* is string LF terminated? 
                 * XXX: This check can be made more efficient by simply checking 
                 * if the last character in the 'response' buffer is an ASCII_LF.
                 * See ap_rgetline() for an example.
                 */
                if (memchr(response, APR_ASCII_LF, len)) {
                    found = 1;
                }
                /* concat strings until buff is full - then throw the data away */
                if (len > ((bufflen-1)-(pos-buff))) {
                    len = (bufflen-1)-(pos-buff);
                }
                if (len > 0) {
                    pos = apr_cpystrn(pos, response, len);
                }
            }
            APR_BUCKET_REMOVE(e);
            apr_bucket_destroy(e);
        }
    }

    return APR_SUCCESS;
}

/* unmerge an element in the table */
PROXY_DECLARE(void) ap_proxy_table_unmerge(apr_pool_t *p, apr_table_t *t, char *key)
{
    apr_off_t offset = 0;
    apr_off_t count = 0;
    char *value = NULL;

    /* get the value to unmerge */
    const char *initial = apr_table_get(t, key);
    if (!initial) {
        return;
    }
    value = apr_pstrdup(p, initial);

    /* remove the value from the headers */
    apr_table_unset(t, key);

    /* find each comma */
    while (value[count]) {
        if (value[count] == ',') {
            value[count] = 0;
            apr_table_add(t, key, value + offset);
            offset = count + 1;
        }
        count++;
    }
    apr_table_add(t, key, value + offset);
}

PROXY_DECLARE(proxy_balancer *) ap_proxy_get_balancer(apr_pool_t *p,
                                                      proxy_server_conf *conf,
                                                      const char *url)
{
    proxy_balancer *balancer;
    char *c, *uri = apr_pstrdup(p, url);
    int i;
    
    c = strchr(uri, ':');   
    if (c == NULL || c[1] != '/' || c[2] != '/' || c[3] == '\0')
       return NULL;
    /* remove path from uri */
    if ((c = strchr(c + 3, '/')))
        *c = '\0';
    balancer = (proxy_balancer *)conf->balancers->elts;
    for (i = 0; i < conf->balancers->nelts; i++) {
        if (strcasecmp(balancer->name, uri) == 0)
            return balancer;
        balancer++;
    }
    return NULL;
}

PROXY_DECLARE(const char *) ap_proxy_add_balancer(proxy_balancer **balancer,
                                                  apr_pool_t *p,
                                                  proxy_server_conf *conf,
                                                  const char *url)
{
    char *c, *q, *uri = apr_pstrdup(p, url);
    apr_status_t rc = 0;

    c = strchr(uri, ':');   
    if (c == NULL || c[1] != '/' || c[2] != '/' || c[3] == '\0')
       return "Bad syntax for a balancer name";
    /* remove path from uri */
    if ((q = strchr(c + 3, '/')))
        *q = '\0';

    ap_str_tolower(uri);
    *balancer = apr_array_push(conf->balancers);
    (*balancer)->name = uri;
    (*balancer)->workers = apr_array_make(p, 5, sizeof(proxy_runtime_worker));
    /* XXX Is this a right place to create mutex */
#if APR_HAS_THREADS
    if ((rc = apr_thread_mutex_create(&((*balancer)->mutex),
                APR_THREAD_MUTEX_DEFAULT, p)) != APR_SUCCESS) {
            /* XXX: Do we need to log something here */
            return "can not create thread mutex";
    }
#endif

    return NULL;
}

PROXY_DECLARE(proxy_worker *) ap_proxy_get_worker(apr_pool_t *p,
                                                  proxy_server_conf *conf,
                                                  const char *url)
{
    proxy_worker *worker;
    char *c, *uri = apr_pstrdup(p, url);
    int i;
    
    c = strchr(uri, ':');   
    if (c == NULL || c[1] != '/' || c[2] != '/' || c[3] == '\0')
       return NULL;
    /* remove path from uri */
    if ((c = strchr(c + 3, '/')))
        *c = '\0';

    worker = (proxy_worker *)conf->workers->elts;
    for (i = 0; i < conf->workers->nelts; i++) {
        if (strcasecmp(worker->name, uri) == 0) {
            return worker;
        }
        worker++;
    }
    return NULL;
}

static apr_status_t conn_pool_cleanup(void *thepool)
{
    proxy_conn_pool *cp = (proxy_conn_pool *)thepool;
    /* Close the socket */
    cp->addr = NULL;
    return APR_SUCCESS;
}

static void init_conn_pool(apr_pool_t *p, proxy_worker *worker)
{
    apr_pool_t *pool;
    proxy_conn_pool *cp;
    
    /* Create a connection pool's subpool. 
     * This pool is used for connection recycling.
     * Once the worker is added it is never removed but
     * it can be disabled.
     */
    apr_pool_create(&pool, p);
    /* Alloc from the same pool as worker.
     * proxy_conn_pool is permanently attached to the worker. 
     */
    cp = (proxy_conn_pool *)apr_pcalloc(p, sizeof(proxy_conn_pool));
    cp->pool = pool;    
    worker->cp = cp;
    apr_pool_cleanup_register(p, (void *)cp,
                              conn_pool_cleanup,
                              apr_pool_cleanup_null);      

}

PROXY_DECLARE(const char *) ap_proxy_add_worker(proxy_worker **worker,
                                                apr_pool_t *p,
                                                proxy_server_conf *conf,
                                                const char *url)
{
    char *c, *q, *uri = apr_pstrdup(p, url);
    int port;
    
    c = strchr(uri, ':');   
    if (c == NULL || c[1] != '/' || c[2] != '/' || c[3] == '\0')
       return "Bad syntax for a remote proxy server";
    /* remove path from uri */
    if ((q = strchr(c + 3, '/')))
        *q = '\0';

    q = strchr(c + 3, ':');
    if (q != NULL) {
        if (sscanf(q + 1, "%u", &port) != 1 || port > 65535) {
            return "Bad syntax for a remote proxy server (bad port number)";
        }
    }
    else
        port = -1;
    ap_str_tolower(uri);
    *worker = apr_array_push(conf->workers);
    memset(*worker, 0, sizeof(proxy_worker));
    (*worker)->name = apr_pstrdup(p, uri);
    *c = '\0';
    (*worker)->scheme = uri;
    (*worker)->hostname = c + 3;

    if (port == -1)
        port = apr_uri_port_of_scheme((*worker)->scheme);
    (*worker)->port = port;

    init_conn_pool(p, *worker);

    return NULL;
}

PROXY_DECLARE(void) 
ap_proxy_add_worker_to_balancer(apr_pool_t *pool, proxy_balancer *balancer, proxy_worker *worker)
{
    int i;
    double median, ffactor = 0.0;
    proxy_runtime_worker *runtime, *workers;    
#if PROXY_HAS_SCOREBOARD
    lb_score *score;
#else
    void *score;
#endif

#if PROXY_HAS_SCOREBOARD
    int mpm_daemons;

    ap_mpm_query(AP_MPMQ_HARD_LIMIT_DAEMONS, &mpm_daemons);
    /* Check if we are prefork or single child */
    if (worker->hmax && mpm_daemons > 1) {
        /* Check only if workers_limit is set */
        if (lb_workers_limit && (lb_workers + 1) > lb_workers_limit) {
            ap_log_perror(APLOG_MARK, APLOG_ERR, 0, pool,
                          "proxy: Can not add worker (%s) to balancer (%s)."
                          " Dynamic limit reached.",
                          worker->name, balancer->name);
            return;
        }
        score = ap_get_scoreboard_lb(getpid(), lb_workers);
    }
    else
#endif
    {
        /* Use the plain memory */
        score = apr_pcalloc(pool, sizeof(proxy_runtime_stat));
    }
    if (!score)
        return;
    runtime = apr_array_push(balancer->workers);
    runtime->w = worker;
    runtime->s = (proxy_runtime_stat *)score;
    runtime->s->id = lb_workers;
    /* TODO: deal with the dynamic overflow */
    ++lb_workers;

    /* Recalculate lbfactors */
    workers = (proxy_runtime_worker *)balancer->workers->elts;

    for (i = 0; i < balancer->workers->nelts; i++) {
        /* Set to the original configuration */
        workers[i].s->lbfactor = workers[i].w->lbfactor;
        ffactor += workers[i].s->lbfactor;
    }
    if (ffactor < 100.0) {
        int z = 0;
        for (i = 0; i < balancer->workers->nelts; i++) {
            if (workers[i].s->lbfactor == 0.0) 
                ++z;
        }
        if (z) {
            median = (100.0 - ffactor) / z;
            for (i = 0; i < balancer->workers->nelts; i++) {
                if (workers[i].s->lbfactor == 0.0) 
                    workers[i].s->lbfactor = median;
            }
        }
        else {
            median = (100.0 - ffactor) / balancer->workers->nelts;
            for (i = 0; i < balancer->workers->nelts; i++)
                workers[i].s->lbfactor += median;
        }
    }
    else if (ffactor > 100.0) {
        median = (ffactor - 100.0) / balancer->workers->nelts;
        for (i = 0; i < balancer->workers->nelts; i++) {
            if (workers[i].s->lbfactor > median)
                workers[i].s->lbfactor -= median;
        }
    } 
    for (i = 0; i < balancer->workers->nelts; i++) {
        /* Update the status entires */
        workers[i].s->lbstatus = workers[i].s->lbfactor;
    }
}

PROXY_DECLARE(int) ap_proxy_pre_request(proxy_worker **worker,
                                        proxy_balancer **balancer,
                                        request_rec *r,
                                        proxy_server_conf *conf, char **url)
{
    int access_status;

    access_status = proxy_run_pre_request(worker, balancer, r, conf, url);
    if (access_status == DECLINED && *balancer == NULL) {
        *worker = ap_proxy_get_worker(r->pool, conf, *url);
        if (*worker) {
            *balancer = NULL;
            access_status = OK;
        }
        else
            access_status = DECLINED;
    }
    else if (access_status == DECLINED && balancer != NULL) {
        /* All the workers are busy */
        access_status = HTTP_SERVICE_UNAVAILABLE;
    }
    return access_status;
}

PROXY_DECLARE(int) ap_proxy_post_request(proxy_worker *worker,
                                         proxy_balancer *balancer,
                                         request_rec *r,
                                         proxy_server_conf *conf)
{
    int access_status;
    if (balancer)
        access_status = proxy_run_post_request(worker, balancer, r, conf);
    else { 
        

        access_status = OK;
    }

    return access_status;
}

/* DEPRECATED */
PROXY_DECLARE(int) ap_proxy_connect_to_backend(apr_socket_t **newsock,
                                               const char *proxy_function,
                                               apr_sockaddr_t *backend_addr,
                                               const char *backend_name,
                                               proxy_server_conf *conf,
                                               server_rec *s,
                                               apr_pool_t *p)
{
    apr_status_t rv;
    int connected = 0;
    int loglevel;
    
    while (backend_addr && !connected) {
        if ((rv = apr_socket_create(newsock, backend_addr->family,
                                    SOCK_STREAM, 0, p)) != APR_SUCCESS) {
            loglevel = backend_addr->next ? APLOG_DEBUG : APLOG_ERR;
            ap_log_error(APLOG_MARK, loglevel, rv, s,
                         "proxy: %s: error creating fam %d socket for target %s",
                         proxy_function,
                         backend_addr->family,
                         backend_name);
            /* this could be an IPv6 address from the DNS but the
             * local machine won't give us an IPv6 socket; hopefully the
             * DNS returned an additional address to try
             */
            backend_addr = backend_addr->next;
            continue;
        }

#if !defined(TPF) && !defined(BEOS)
        if (conf->recv_buffer_size > 0 &&
            (rv = apr_socket_opt_set(*newsock, APR_SO_RCVBUF,
                                     conf->recv_buffer_size))) {
            ap_log_error(APLOG_MARK, APLOG_ERR, rv, s,
                         "apr_socket_opt_set(SO_RCVBUF): Failed to set "
                         "ProxyReceiveBufferSize, using default");
        }
#endif

        /* Set a timeout on the socket */
        if (conf->timeout_set == 1) {
            apr_socket_timeout_set(*newsock, conf->timeout);
        }
        else {
             apr_socket_timeout_set(*newsock, s->timeout);
        }

        ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
                     "proxy: %s: fam %d socket created to connect to %s",
                     proxy_function, backend_addr->family, backend_name);

        /* make the connection out of the socket */
        rv = apr_socket_connect(*newsock, backend_addr);

        /* if an error occurred, loop round and try again */
        if (rv != APR_SUCCESS) {
            apr_socket_close(*newsock);
            loglevel = backend_addr->next ? APLOG_DEBUG : APLOG_ERR;
            ap_log_error(APLOG_MARK, loglevel, rv, s,
                         "proxy: %s: attempt to connect to %pI (%s) failed",
                         proxy_function,
                         backend_addr,
                         backend_name);
            backend_addr = backend_addr->next;
            continue;
        }
        connected = 1;
    }
    return connected ? 0 : 1;
}

static apr_status_t proxy_conn_cleanup(void *theconn)
{
    proxy_conn_rec *conn = (proxy_conn_rec *)theconn;
    /* Close the socket */
    if (conn->sock)
        apr_socket_close(conn->sock);
    conn->sock = NULL;
    conn->pool = NULL;
    return APR_SUCCESS;
}

static apr_status_t connection_cleanup(void *theconn)
{
    proxy_conn_rec *conn = (proxy_conn_rec *)theconn;
    proxy_worker *worker = conn->worker;
    
    /* deterimine if the connection need to be closed */
    if (conn->close_on_recycle) {
        if (conn->sock)
            apr_socket_close(conn->sock);
        conn->sock = NULL;
    }
#if APR_HAS_THREADS
    if (worker->hmax && worker->cp->res) {
        apr_reslist_release(worker->cp->res, (void *)conn);
    }
    else
#endif
    {
        worker->cp->conn = conn;
    }

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

/* reslist constructor */
static apr_status_t connection_constructor(void **resource, void *params,
                                           apr_pool_t *pool)
{
    apr_pool_t *ctx;
    proxy_conn_rec *conn;
    server_rec *s = (server_rec *)params;
    
    /* Create the subpool for each connection
     * This keeps the memory consumption constant
     * when disconnecting from backend.
     */
    apr_pool_create(&ctx, pool);
    conn = apr_pcalloc(ctx, sizeof(proxy_conn_rec));

    conn->pool = ctx;
    *resource = conn;
    /* register the pool cleanup */
    apr_pool_cleanup_register(ctx, (void *)conn,
                              proxy_conn_cleanup,
                              apr_pool_cleanup_null);      

    ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,
                 "proxy: socket is constructed");

    return APR_SUCCESS;
}

/* reslist destructor */
static apr_status_t connection_destructor(void *resource, void *params,
                                          apr_pool_t *pool)
{
    proxy_conn_rec *conn = (proxy_conn_rec *)resource;