Commit 3355d900 authored by Jim Jagielski's avatar Jim Jagielski
Browse files

Remove the thread mutex from the worker... it really should be 

in the balancer. Thus we have global and thread for the balancer.
Use global when updating the full, shm list of workers; use
thread when being local.

git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/trunk@1068581 13f79535-47bb-0310-9956-ffa450edef68
parent c9eac71c

modules/proxy/mod_proxy.h

+8 −6
Changes for modules/proxy/mod_proxy.h: 8 added lines, 6 removed lines.
Original line number Diff line number Diff line
@@ -161,6 +161,7 @@ typedef struct { 
    apr_sockaddr_t *source_address;

    apr_global_mutex_t  *mutex; /* global lock (needed??) */

    ap_slotmem_instance_t *slot;  /* balancers shm data - runtime */

    ap_slotmem_provider_t *storage;



    int req_set:1;

    int viaopt_set:1;
@@ -375,7 +376,6 @@ struct proxy_worker { 
    proxy_conn_pool     *cp;    /* Connection pool to use */

    proxy_worker_shared   *s;   /* Shared data */

    proxy_balancer  *balancer;  /* which balancer am I in? */

    apr_thread_mutex_t  *mutex; /* Thread lock for updating address cache */

    void            *context;   /* general purpose storage */

};
@@ -406,12 +406,14 @@ struct proxy_balancer { 
    apr_array_header_t *workers;  /* initially configured workers */

    apr_array_header_t *errstatuses;  /* statuses to force members into error */

    ap_slotmem_instance_t *slot;  /* worker shm data - runtime */

    ap_slotmem_provider_t *storage;

    int growth;                   /* number of post-config workers can added */

    int max_workers;              /* maximum number of allowed workers */

    const char *name;             /* name of the load balancer */

    const char *sname;            /* filesystem safe balancer name */

    apr_time_t      wupdated;    /* timestamp of last change to workers list */

    apr_global_mutex_t  *mutex; /* global lock for updating lb params */

    apr_global_mutex_t  *gmutex; /* global lock for updating list of workers */

    apr_thread_mutex_t  *tmutex; /* Thread lock for updating address cache and worker selection*/

    void            *context;    /* general purpose storage */

    proxy_balancer_shared *s;    /* Shared data */

};
@@ -426,11 +428,11 @@ struct proxy_balancer_method { 
    apr_status_t (*updatelbstatus)(proxy_balancer *balancer, proxy_worker *elected, server_rec *s);

};



#define PROXY_THREAD_LOCK(x)      apr_thread_mutex_lock((x)->mutex)

#define PROXY_THREAD_UNLOCK(x)    apr_thread_mutex_unlock((x)->mutex)

#define PROXY_THREAD_LOCK(x)      ( (x) && (x)->tmutex ? apr_thread_mutex_lock((x)->tmutex) : APR_SUCCESS)

#define PROXY_THREAD_UNLOCK(x)    ( (x) && (x)->tmutex ? apr_thread_mutex_unlock((x)->tmutex) : APR_SUCCESS)



#define PROXY_GLOBAL_LOCK(x)      apr_global_mutex_lock((x)->mutex)

#define PROXY_GLOBAL_UNLOCK(x)    apr_global_mutex_unlock((x)->mutex)

#define PROXY_GLOBAL_LOCK(x)      ( (x) && (x)->gmutex ? apr_global_mutex_lock((x)->gmutex) : APR_SUCCESS)

#define PROXY_GLOBAL_UNLOCK(x)    ( (x) && (x)->gmutex ? apr_global_mutex_unlock((x)->gmutex) : APR_SUCCESS)



/* hooks */

modules/proxy/mod_proxy_balancer.c

+25 −53
Changes for modules/proxy/mod_proxy_balancer.c: 25 added lines, 53 removed lines.
Original line number Diff line number Diff line
@@ -104,7 +104,7 @@ static int proxy_balancer_canon(request_rec *r, char *url) 
    return OK;

}



static void init_balancer_members(proxy_server_conf *conf, server_rec *s,

static void init_balancer_members(apr_pool_t *p, server_rec *s,

                                 proxy_balancer *balancer)

{

    int i;
@@ -119,7 +119,7 @@ static void init_balancer_members(proxy_server_conf *conf, server_rec *s, 
                     "Looking at %s -> %s initialized?", balancer->name, worker->s->name);

        worker_is_initialized = PROXY_WORKER_IS_INITIALIZED(worker);

        if (!worker_is_initialized) {

            ap_proxy_initialize_worker(worker, s, conf->pool);

            ap_proxy_initialize_worker(worker, s, p);

        }

        ++workers;

    }
@@ -325,7 +325,7 @@ static proxy_worker *find_best_worker(proxy_balancer *balancer, 
    proxy_worker *candidate = NULL;

    apr_status_t rv;



    if ((rv = PROXY_GLOBAL_LOCK(balancer)) != APR_SUCCESS) {

    if ((rv = PROXY_THREAD_LOCK(balancer)) != APR_SUCCESS) {

        ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

        "proxy: BALANCER: (%s). Lock failed for find_best_worker()", balancer->name);

        return NULL;
@@ -336,12 +336,7 @@ static proxy_worker *find_best_worker(proxy_balancer *balancer, 
    if (candidate)

        candidate->s->elected++;



/*

        PROXY_GLOBAL_UNLOCK(conf);

        return NULL;

*/



    if ((rv = PROXY_GLOBAL_UNLOCK(balancer)) != APR_SUCCESS) {

    if ((rv = PROXY_THREAD_UNLOCK(balancer)) != APR_SUCCESS) {

        ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

        "proxy: BALANCER: (%s). Unlock failed for find_best_worker()", balancer->name);

    }
@@ -463,7 +458,7 @@ static int proxy_balancer_pre_request(proxy_worker **worker, 
    /* Step 2: Lock the LoadBalancer

     * XXX: perhaps we need the process lock here

     */

    if ((rv = PROXY_GLOBAL_LOCK(*balancer)) != APR_SUCCESS) {

    if ((rv = PROXY_THREAD_LOCK(*balancer)) != APR_SUCCESS) {

        ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

                     "proxy: BALANCER: (%s). Lock failed for pre_request",

                     (*balancer)->name);
@@ -529,7 +524,7 @@ static int proxy_balancer_pre_request(proxy_worker **worker, 
            ap_log_error(APLOG_MARK, APLOG_ERR, 0, r->server,

                         "proxy: BALANCER: (%s). All workers are in error state for route (%s)",

                         (*balancer)->name, route);

            if ((rv = PROXY_GLOBAL_UNLOCK(*balancer)) != APR_SUCCESS) {

            if ((rv = PROXY_THREAD_UNLOCK(*balancer)) != APR_SUCCESS) {

                ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

                             "proxy: BALANCER: (%s). Unlock failed for pre_request",

                             (*balancer)->name);
@@ -538,7 +533,7 @@ static int proxy_balancer_pre_request(proxy_worker **worker, 
        }

    }



    if ((rv = PROXY_GLOBAL_UNLOCK(*balancer)) != APR_SUCCESS) {

    if ((rv = PROXY_THREAD_UNLOCK(*balancer)) != APR_SUCCESS) {

        ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

                     "proxy: BALANCER: (%s). Unlock failed for pre_request",

                     (*balancer)->name);
@@ -614,7 +609,7 @@ static int proxy_balancer_post_request(proxy_worker *worker, 


    apr_status_t rv;



    if ((rv = PROXY_GLOBAL_LOCK(balancer)) != APR_SUCCESS) {

    if ((rv = PROXY_THREAD_UNLOCK(balancer)) != APR_SUCCESS) {

        ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

            "proxy: BALANCER: (%s). Lock failed for post_request",

            balancer->name);
@@ -636,7 +631,7 @@ static int proxy_balancer_post_request(proxy_worker *worker, 
        }

    }



    if ((rv = PROXY_GLOBAL_UNLOCK(balancer)) != APR_SUCCESS) {

    if ((rv = PROXY_THREAD_UNLOCK(balancer)) != APR_SUCCESS) {

        ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

            "proxy: BALANCER: (%s). Unlock failed for post_request",

            balancer->name);
@@ -682,9 +677,9 @@ static apr_status_t lock_remove(void *data) 


    balancer = (proxy_balancer *)conf->balancers->elts;

    for (i = 0; i < conf->balancers->nelts; i++, balancer++) {

        if (balancer->mutex) {

            apr_global_mutex_destroy(balancer->mutex);

            balancer->mutex = NULL;

        if (balancer->gmutex) {

            apr_global_mutex_destroy(balancer->gmutex);

            balancer->gmutex = NULL;

        }

    }

    return(0);
@@ -746,6 +741,7 @@ static int balancer_post_config(apr_pool_t *pconf, apr_pool_t *plog, 
            }

            conf->slot = new;

        }

        conf->storage = storage;



        /* Initialize shared scoreboard data */

        balancer = (proxy_balancer *)conf->balancers->elts;
@@ -762,9 +758,9 @@ static int balancer_post_config(apr_pool_t *pconf, apr_pool_t *plog, 
            balancer->sname = apr_pstrcat(pconf, conf->id, "_", balancer->sname, NULL);



            /* Create global mutex */

            rv = ap_global_mutex_create(&(balancer->mutex), NULL, balancer_mutex_type,

            rv = ap_global_mutex_create(&(balancer->gmutex), NULL, balancer_mutex_type,

                                        balancer->sname, s, pconf, 0);

            if (rv != APR_SUCCESS || !balancer->mutex) {

            if (rv != APR_SUCCESS || !balancer->gmutex) {

                ap_log_error(APLOG_MARK, APLOG_EMERG, rv, s,

                             "mutex creation of %s : %s failed", balancer_mutex_type,

                             balancer->sname);
@@ -803,6 +799,7 @@ static int balancer_post_config(apr_pool_t *pconf, apr_pool_t *plog, 
                return !OK;

            }

            balancer->slot = new;

            balancer->storage = storage;



            /* sync all timestamps */

            balancer->wupdated = balancer->s->wupdated = tstamp;
@@ -878,13 +875,13 @@ static int balancer_handler(request_rec *r) 
    balancer = (proxy_balancer *)conf->balancers->elts;

    for (i = 0; i < conf->balancers->nelts; i++, balancer++) {

        apr_status_t rv;

        if ((rv = PROXY_GLOBAL_LOCK(balancer)) != APR_SUCCESS) {

        if ((rv = PROXY_THREAD_LOCK(balancer)) != APR_SUCCESS) {

            ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

                         "proxy: BALANCER: (%s). Lock failed for balancer_handler",

                         balancer->name);

        }

        ap_proxy_update_members(balancer, r->server, conf);

        if ((rv = PROXY_GLOBAL_UNLOCK(balancer)) != APR_SUCCESS) {

        if ((rv = PROXY_THREAD_UNLOCK(balancer)) != APR_SUCCESS) {

            ap_log_error(APLOG_MARK, APLOG_ERR, rv, r->server,

                         "proxy: BALANCER: (%s). Unlock failed for balancer_handler",

                         balancer->name);
@@ -1301,11 +1298,11 @@ static void balancer_child_init(apr_pool_t *p, server_rec *s) 
        int i;

        void *sconf = s->module_config;

        proxy_server_conf *conf = (proxy_server_conf *)ap_get_module_config(sconf, &proxy_module);

        apr_size_t size;

        unsigned int num;

        apr_status_t rv;



        if (conf->balancers->nelts) {

            apr_size_t size;

            unsigned int num;

            storage->attach(&(conf->slot), conf->id, &size, &num, p);

            if (!conf->slot) {

                ap_log_error(APLOG_MARK, APLOG_NOERRNO|APLOG_EMERG, 0, s, "slotmem_attach failed");
@@ -1314,40 +1311,16 @@ static void balancer_child_init(apr_pool_t *p, server_rec *s) 
        }



        balancer = (proxy_balancer *)conf->balancers->elts;

        for (i = 0; i < conf->balancers->nelts; i++) {



            /*

             * for each balancer we need to init the global

             * mutex and then attach to the shared worker shm

             */

            if (!balancer->mutex) {

                ap_log_error(APLOG_MARK, APLOG_INFO, 0, s,

                             "no mutex %s: %s", balancer->name,

                             balancer_mutex_type);

                return;

            }

        for (i = 0; i < conf->balancers->nelts; i++, balancer++) {

            rv = ap_proxy_initialize_balancer(balancer, s, p);



            /* Re-open the mutex for the child. */

            rv = apr_global_mutex_child_init(&(balancer->mutex),

                                             apr_global_mutex_lockfile(balancer->mutex),

                                             p);

            if (rv != APR_SUCCESS) {

                ap_log_error(APLOG_MARK, APLOG_CRIT, rv, s,

                             "Failed to reopen mutex %s: %s in child",

                             balancer->name, balancer_mutex_type);

                exit(1); /* Ugly, but what else? */

            }



            /* now attach */

            storage->attach(&(balancer->slot), balancer->sname, &size, &num, p);

            if (!balancer->slot) {

                ap_log_error(APLOG_MARK, APLOG_NOERRNO|APLOG_EMERG, 0, s, "slotmem_attach failed");

                             "Failed to init balancer %s in child",

                             balancer->name);

                exit(1); /* Ugly, but what else? */

            }

            if (balancer->s->lbmethod && balancer->s->lbmethod->reset)

               balancer->s->lbmethod->reset(balancer, s);

            init_balancer_members(conf, s, balancer);

            balancer++;

            init_balancer_members(conf->pool, s, balancer);

        }

        s = s->next;

    }
@@ -1395,7 +1368,6 @@ PROXY_DECLARE(apr_status_t) ap_proxy_update_members(proxy_balancer *b, server_re 
            (*runtime)->hash = shm->hash;

            (*runtime)->context = NULL;

            (*runtime)->cp = NULL;

            (*runtime)->mutex = NULL;

            (*runtime)->balancer = b;

            (*runtime)->s = shm;

            if ((rv = ap_proxy_initialize_worker(*runtime, s, conf->pool)) != APR_SUCCESS) {

modules/proxy/mod_proxy_ftp.c

+2 −2
Changes for modules/proxy/mod_proxy_ftp.c: 2 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -1039,7 +1039,7 @@ static int proxy_ftp_handler(request_rec *r, proxy_worker *worker, 


    if (worker->s->is_address_reusable) {

        if (!worker->cp->addr) {

            if ((err = PROXY_THREAD_LOCK(worker)) != APR_SUCCESS) {

            if ((err = PROXY_THREAD_LOCK(worker->balancer)) != APR_SUCCESS) {

                ap_log_error(APLOG_MARK, APLOG_ERR, err, r->server,

                             "proxy: FTP: lock");

                return HTTP_INTERNAL_SERVER_ERROR;
@@ -1059,7 +1059,7 @@ static int proxy_ftp_handler(request_rec *r, proxy_worker *worker, 
                                    address_pool);

    if (worker->s->is_address_reusable && !worker->cp->addr) {

        worker->cp->addr = connect_addr;

        if ((uerr = PROXY_THREAD_UNLOCK(worker)) != APR_SUCCESS) {

        if ((uerr = PROXY_THREAD_UNLOCK(worker->balancer)) != APR_SUCCESS) {

            ap_log_error(APLOG_MARK, APLOG_ERR, uerr, r->server,

                         "proxy: FTP: unlock");

        }

modules/proxy/proxy_util.c

+51 −13
Changes for modules/proxy/proxy_util.c: 51 added lines, 13 removed lines.
Original line number Diff line number Diff line
@@ -1379,7 +1379,8 @@ PROXY_DECLARE(char *) ap_proxy_define_balancer(apr_pool_t *p, 


    (*balancer)->name = uri;

    (*balancer)->workers = apr_array_make(p, 5, sizeof(proxy_worker *));

    (*balancer)->mutex = NULL;

    (*balancer)->gmutex = NULL;

    (*balancer)->tmutex = NULL;



    if (do_malloc)

        bshared = malloc(sizeof(proxy_balancer_shared));
@@ -1423,8 +1424,55 @@ PROXY_DECLARE(apr_status_t) ap_proxy_share_balancer(proxy_balancer *balancer, 
PROXY_DECLARE(apr_status_t) ap_proxy_initialize_balancer(proxy_balancer *balancer, server_rec *s, apr_pool_t *p)

{

    apr_status_t rv = APR_SUCCESS;

    ap_slotmem_provider_t *storage = balancer->storage;

    apr_size_t size;

    unsigned int num;



    if (!storage) {

        ap_log_error(APLOG_MARK, APLOG_INFO, 0, s,

                     "no provider for %s", balancer->name);

        return APR_EGENERAL;

    }

    /*

     * for each balancer we need to init the global

     * mutex and then attach to the shared worker shm

     */

    if (!balancer->gmutex) {

        ap_log_error(APLOG_MARK, APLOG_INFO, 0, s,

                     "no mutex %s", balancer->name);

        return APR_EGENERAL;

    }

    

    /* Re-open the mutex for the child. */

    rv = apr_global_mutex_child_init(&(balancer->gmutex),

                                     apr_global_mutex_lockfile(balancer->gmutex),

                                     p);

    if (rv != APR_SUCCESS) {

        ap_log_error(APLOG_MARK, APLOG_CRIT, rv, s,

                     "Failed to reopen mutex %s in child",

                     balancer->name);

        return rv;

    }

    

    /* now attach */

    storage->attach(&(balancer->slot), balancer->sname, &size, &num, p);

    if (!balancer->slot) {

        ap_log_error(APLOG_MARK, APLOG_NOERRNO|APLOG_EMERG, 0, s, "slotmem_attach failed");

        return APR_EGENERAL;

    }

    if (balancer->s->lbmethod && balancer->s->lbmethod->reset)

        balancer->s->lbmethod->reset(balancer, s);

    

    if (balancer->tmutex == NULL) {

        rv = apr_thread_mutex_create(&(balancer->tmutex), APR_THREAD_MUTEX_DEFAULT, p);

        if (rv != APR_SUCCESS) {

            ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,

                         "can not create thread mutex");

            return rv;

        }

    }    

    return APR_SUCCESS;

}



/*

 * CONNECTION related...
@@ -1765,7 +1813,6 @@ PROXY_DECLARE(char *) ap_proxy_define_worker(apr_pool_t *p, 
    (*worker)->hash = wshared->hash;

    (*worker)->context = NULL;

    (*worker)->cp = NULL;

    (*worker)->mutex = NULL;

    (*worker)->balancer = balancer;

    (*worker)->s = wshared;
@@ -1851,15 +1898,6 @@ PROXY_DECLARE(apr_status_t) ap_proxy_initialize_worker(proxy_worker *worker, ser 
            return APR_EGENERAL;

        }



        if (worker->mutex == NULL) {

            rv = apr_thread_mutex_create(&(worker->mutex), APR_THREAD_MUTEX_DEFAULT, p);

            if (rv != APR_SUCCESS) {

                ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, s,

                    "can not create thread mutex");

                return rv;

            }

        }



        if (worker->s->hmax) {

            rv = apr_reslist_create(&(worker->cp->res),

                                    worker->s->min, worker->s->smax,
@@ -2255,7 +2293,7 @@ ap_proxy_determine_connection(apr_pool_t *p, request_rec *r, 
                                    conn->pool);

    }

    else if (!worker->cp->addr) {

        if ((err = PROXY_THREAD_LOCK(worker)) != APR_SUCCESS) {

        if ((err = PROXY_THREAD_LOCK(worker->balancer)) != APR_SUCCESS) {

            ap_log_error(APLOG_MARK, APLOG_ERR, err, r->server,

                         "proxy: lock");

            return HTTP_INTERNAL_SERVER_ERROR;
@@ -2272,7 +2310,7 @@ ap_proxy_determine_connection(apr_pool_t *p, request_rec *r, 
                                    conn->port, 0,

                                    worker->cp->pool);

        conn->addr = worker->cp->addr;

        if ((uerr = PROXY_THREAD_UNLOCK(worker)) != APR_SUCCESS) {

        if ((uerr = PROXY_THREAD_UNLOCK(worker->balancer)) != APR_SUCCESS) {

            ap_log_error(APLOG_MARK, APLOG_ERR, uerr, r->server,

                         "proxy: unlock");

        }