jk_lb_worker.c

jboss与apache集成的中间件,详情请参看文档说明.

                    if (is_service_error != JK_HTTP_SERVER_BUSY) {
                        /*
                        * Error is not recoverable - break with an error.
                        */
                        jk_log(l, JK_LOG_ERROR,
                            "unrecoverable error %d, request failed."
                            " Tomcat failed in the middle of request,"
                            " we can't recover to another instance.",
                            is_service_error);
                        *is_error = is_service_error;
                        JK_TRACE_EXIT(l);
                        return JK_FALSE;
                    }
                    jk_log(l, JK_LOG_INFO,
                           "service failed, worker %s is in error state",
                           rec->s->name);
                }
                else if (service_stat == JK_CLIENT_ERROR) {
                    /*
                    * Clent error !!!
                    * Since this is bad request do not fail over.
                    */
                    rec->s->errors++;
                    rec->s->in_error_state = JK_FALSE;
                    rec->s->in_recovering = JK_FALSE;
                    rec->s->error_time = 0;
                    *is_error = is_service_error;

                    jk_log(l, JK_LOG_INFO,
                           "unrecoverable error %d, request failed."
                           " Client failed in the middle of request,"
                           " we can't recover to another instance.",
                           is_service_error);
                    JK_TRACE_EXIT(l);
                    return JK_CLIENT_ERROR;
                }
                else {
                    /* If we can not get the endpoint from the worker
                     * that does not mean that the worker is in error
                     * state. It means that the worker is busy.
                     * We will try another worker.
                     * To prevent infinite loop decrement worker count;
                     */
                }
                /*
                 * Error is recoverable by submitting the request to
                 * another worker... Lets try to do that.
                 */
                if (JK_IS_DEBUG_LEVEL(l))
                    jk_log(l, JK_LOG_DEBUG,
                           "recoverable error... will try to recover on other host");
            }
            else {
                /* NULL record, no more workers left ... */
                jk_log(l, JK_LOG_ERROR,
                       "All tomcat instances failed, no more workers left");
                JK_TRACE_EXIT(l);
                *is_error = JK_HTTP_SERVER_BUSY;
                return JK_FALSE;
            }
            --num_of_workers;
            prec = rec;
        }
        jk_log(l, JK_LOG_INFO,
               "All tomcat instances are busy or in error state");
        JK_TRACE_EXIT(l);
        /* Set error to Timeout */
        *is_error = JK_HTTP_SERVER_BUSY;
        return JK_FALSE;
    }
    if (is_error)
        *is_error = JK_HTTP_SERVER_ERROR;
    JK_LOG_NULL_PARAMS(l);
    JK_TRACE_EXIT(l);
    return JK_FALSE;
}

static int JK_METHOD done(jk_endpoint_t **e, jk_logger_t *l)
{
    JK_TRACE_ENTER(l);

    if (e && *e && (*e)->endpoint_private) {
        lb_endpoint_t *p = (*e)->endpoint_private;

        if (p->e) {
            p->e->done(&p->e, l);
        }

        free(p);
        *e = NULL;
        JK_TRACE_EXIT(l);
        return JK_TRUE;
    }

    JK_LOG_NULL_PARAMS(l);
    JK_TRACE_EXIT(l);
    return JK_FALSE;
}

static int JK_METHOD validate(jk_worker_t *pThis,
                              jk_map_t *props,
                              jk_worker_env_t *we, jk_logger_t *l)
{
    JK_TRACE_ENTER(l);

    if (pThis && pThis->worker_private) {
        lb_worker_t *p = pThis->worker_private;
        char **worker_names;
        unsigned int num_of_workers;
        const char *secret;

        p->s->sticky_session = jk_get_is_sticky_session(props, p->s->name);
        p->s->sticky_session_force = jk_get_is_sticky_session_force(props, p->s->name);
        secret = jk_get_worker_secret(props, p->s->name);

        if (jk_get_lb_worker_list(props,
                                  p->s->name,
                                  &worker_names,
                                  &num_of_workers) && num_of_workers) {
            unsigned int i = 0;

            p->lb_workers = jk_pool_alloc(&p->p,
                                          num_of_workers *
                                          sizeof(worker_record_t));
            if (!p->lb_workers) {
                JK_TRACE_EXIT(l);
                return JK_FALSE;
            }

            for (i = 0; i < num_of_workers; i++) {
                p->lb_workers[i].s = jk_shm_alloc_worker(&p->p);
                if (p->lb_workers[i].s == NULL) {
                    jk_log(l, JK_LOG_ERROR,
                           "allocating worker record from shared memory");
                    JK_TRACE_EXIT(l);
                    return JK_FALSE;
                }
            }
            for (i = 0; i < num_of_workers; i++) {
                const char *s;
                strncpy(p->lb_workers[i].s->name, worker_names[i],
                        JK_SHM_STR_SIZ);
                p->lb_workers[i].s->lb_factor =
                    jk_get_lb_factor(props, worker_names[i]);
                if (p->lb_workers[i].s->lb_factor < 1) {
                    p->lb_workers[i].s->lb_factor = 1;
                }
                if ((s = jk_get_worker_domain(props, worker_names[i], NULL)))
                    strncpy(p->lb_workers[i].s->domain, s, JK_SHM_STR_SIZ);
                if ((s = jk_get_worker_redirect(props, worker_names[i], NULL)))
                    strncpy(p->lb_workers[i].s->redirect, s, JK_SHM_STR_SIZ);

                p->lb_workers[i].s->lb_value = p->lb_workers[i].s->lb_factor;
                p->lb_workers[i].s->in_error_state = JK_FALSE;
                p->lb_workers[i].s->in_recovering = JK_FALSE;
                p->lb_workers[i].s->is_busy = JK_FALSE;
                p->lb_workers[i].s->error_time = 0;
                /* Worker can be initaly disabled as hot standby */
                p->lb_workers[i].s->is_disabled = jk_get_is_worker_disabled(props, worker_names[i]);
                /* Worker can be initaly deactive as cold standby */
                p->lb_workers[i].s->is_stopped = jk_get_is_worker_stopped(props, worker_names[i]);
                if (!wc_create_worker(p->lb_workers[i].s->name, 0,
                                      props,
                                      &(p->lb_workers[i].w),
                                      we, l) || !p->lb_workers[i].w) {
                    break;
                }
                if (secret && (p->lb_workers[i].w->type == JK_AJP13_WORKER_TYPE ||
                    p->lb_workers[i].w->type == JK_AJP14_WORKER_TYPE)) {
                    ajp_worker_t *aw = (ajp_worker_t *)p->lb_workers[i].w->worker_private; 
                    if (!aw->secret)
                        aw->secret = secret;
                }
            }

            if (i != num_of_workers) {
                jk_log(l, JK_LOG_ERROR,
                       "Failed creating worker %s",
                       p->lb_workers[i].s->name);
                close_workers(p, i, l);
            }
            else {
                if (JK_IS_DEBUG_LEVEL(l)) {
                    for (i = 0; i < num_of_workers; i++) {
                        jk_log(l, JK_LOG_DEBUG,
                               "Balanced worker %i has name %s in domain %s",
                               i, p->lb_workers[i].s->name, p->lb_workers[i].s->domain);
                    }
                }
                p->num_of_workers = num_of_workers;
                JK_TRACE_EXIT(l);
                return JK_TRUE;
            }
        }
    }

    JK_LOG_NULL_PARAMS(l);
    JK_TRACE_EXIT(l);
    return JK_FALSE;
}

static int JK_METHOD init(jk_worker_t *pThis,
                          jk_map_t *props,
                          jk_worker_env_t *we, jk_logger_t *log)
{
    int i;

    lb_worker_t *p = (lb_worker_t *)pThis->worker_private;
    JK_TRACE_ENTER(log);

    pThis->retries = jk_get_worker_retries(props, p->s->name,
                                           JK_RETRIES);
    p->s->retries = pThis->retries;
    p->s->recover_wait_time = jk_get_worker_recover_timeout(props, p->s->name,
                                                            WAIT_BEFORE_RECOVER);
    if (p->s->recover_wait_time < WAIT_BEFORE_RECOVER)
        p->s->recover_wait_time = WAIT_BEFORE_RECOVER;

    p->lbmethod = jk_get_lb_method(props, p->s->name);
    p->lblock   = jk_get_lb_lock(props, p->s->name);

    JK_INIT_CS(&(p->cs), i);
    if (i == JK_FALSE) {
        jk_log(log, JK_LOG_ERROR,
               "creating thread lock errno=%d",
               errno);
        JK_TRACE_EXIT(log);
        return JK_FALSE;
    }

    JK_TRACE_EXIT(log);
    return JK_TRUE;
}

static int JK_METHOD get_endpoint(jk_worker_t *pThis,
                                  jk_endpoint_t **pend, jk_logger_t *l)
{
    JK_TRACE_ENTER(l);

    if (pThis && pThis->worker_private && pend) {
        lb_endpoint_t *p = (lb_endpoint_t *) malloc(sizeof(lb_endpoint_t));
        p->e = NULL;
        p->worker = pThis->worker_private;
        p->endpoint.endpoint_private = p;
        p->endpoint.service = service;
        p->endpoint.done = done;
        *pend = &p->endpoint;

        JK_TRACE_EXIT(l);
        return JK_TRUE;
    }
    else {
        JK_LOG_NULL_PARAMS(l);
    }

    JK_TRACE_EXIT(l);
    return JK_FALSE;
}

static int JK_METHOD destroy(jk_worker_t **pThis, jk_logger_t *l)
{
    JK_TRACE_ENTER(l);

    if (pThis && *pThis && (*pThis)->worker_private) {
        unsigned int i;
        lb_worker_t *private_data = (*pThis)->worker_private;

        close_workers(private_data, private_data->num_of_workers, l);
        JK_DELETE_CS(&(private_data->cs), i);
        jk_close_pool(&private_data->p);
        free(private_data);

        JK_TRACE_EXIT(l);
        return JK_TRUE;
    }

    JK_LOG_NULL_PARAMS(l);
    JK_TRACE_EXIT(l);
    return JK_FALSE;
}

int JK_METHOD lb_worker_factory(jk_worker_t **w,
                                const char *name, jk_logger_t *l)
{
    JK_TRACE_ENTER(l);

    if (NULL != name && NULL != w) {
        lb_worker_t *private_data =
            (lb_worker_t *) calloc(1, sizeof(lb_worker_t));


        jk_open_pool(&private_data->p,
                        private_data->buf,
                        sizeof(jk_pool_atom_t) * TINY_POOL_SIZE);

        private_data->s = jk_shm_alloc_worker(&private_data->p);
        if (!private_data->s) {
            free(private_data);
            JK_TRACE_EXIT(l);
            return 0;
        }
        strncpy(private_data->s->name, name, JK_SHM_STR_SIZ);
        private_data->lb_workers = NULL;
        private_data->num_of_workers = 0;
        private_data->worker.worker_private = private_data;
        private_data->worker.validate = validate;
        private_data->worker.init = init;
        private_data->worker.get_endpoint = get_endpoint;
        private_data->worker.destroy = destroy;
        private_data->worker.maintain = maintain_workers;
        private_data->worker.retries = JK_RETRIES;
        private_data->s->recover_wait_time = WAIT_BEFORE_RECOVER;
        *w = &private_data->worker;
        JK_TRACE_EXIT(l);
        return JK_LB_WORKER_TYPE;
    }
    else {
        JK_LOG_NULL_PARAMS(l);
    }

    JK_TRACE_EXIT(l);
    return 0;
}