jk_lb_worker.c

来自「Tomcat 4.1与WebServer集成组件的源代码包.」· C语言 代码 · 共 627 行 · 第 1/2 页

    jk_log(l, JK_LOG_DEBUG, 
           "Into jk_endpoint_t::service\n");

    if(e && e->endpoint_private && s && is_recoverable_error) {
        lb_endpoint_t *p = e->endpoint_private;
        jk_endpoint_t *end = NULL;
        int attempt=0;

        /* you can not recover on another load balancer */
        *is_recoverable_error = JK_FALSE;

        
        while(1) {
            worker_record_t *rec = get_most_suitable_worker(p->worker, s, attempt++);
            int rc;

            if(rec) {
                int is_recoverable = JK_FALSE;
                
                s->jvm_route = jk_pool_strdup(s->pool,  rec->name);

                rc = rec->w->get_endpoint(rec->w, &end, l);
                if(rc && end) {
                    int src = end->service(end, s, l, &is_recoverable);
                    end->done(&end, l);
                    if(src ) {                        
                        if(rec->in_recovering && rec->lb_value != 0) {
                            rec->lb_value = get_max_lb(p->worker) + ADDITINAL_WAIT_LOAD;
                        }
                        rec->in_error_state = JK_FALSE;
                        rec->in_recovering  = JK_FALSE;
                        rec->error_time     = 0;                        
                        return JK_TRUE;
                    } 
                }

                /*
                 * Service failed !!!
                 *
                 * Time for fault tolerance (if possible)...
                 */

                rec->in_error_state = JK_TRUE;
                rec->in_recovering  = JK_FALSE;
                rec->error_time     = time(0);

                if(!is_recoverable) {
                    /*
                     * Error is not recoverable - break with an error.
                     */
                    jk_log(l, JK_LOG_ERROR, 
                           "lb: unrecoverable error, request failed. Tomcat failed in the middle of request, we can't recover to another instance.\n");
                    break;
                }

                /* 
                 * Error is recoverable by submitting the request to
                 * another worker... Lets try to do that.
                 */
                 jk_log(l, JK_LOG_DEBUG, 
                        "In jk_endpoint_t::service, recoverable error... will try to recover on other host\n");
            } else {
                /* NULL record, no more workers left ... */
                 jk_log(l, JK_LOG_ERROR, 
                        "lb: All tomcat instances failed, no more workers left.\n");
                break;
            }
        }
    }

    jk_log(l, JK_LOG_ERROR, 
           "In jk_endpoint_t::service: NULL Parameters\n");

    return JK_FALSE;
}

static int JK_METHOD done(jk_endpoint_t **e,
                          jk_logger_t *l)
{
    jk_log(l, JK_LOG_DEBUG, 
           "Into jk_endpoint_t::done\n");

    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;
        return JK_TRUE;
    }

    jk_log(l, JK_LOG_ERROR, 
           "In jk_endpoint_t::done: NULL Parameters\n");

    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_log(l, JK_LOG_DEBUG, 
           "Into jk_worker_t::validate\n");

    if(pThis && pThis->worker_private) {        
        lb_worker_t *p = pThis->worker_private;
        char **worker_names;
        unsigned num_of_workers;
        p->in_local_worker_mode = JK_FALSE;
        p->local_worker_only = jk_get_local_worker_only_flag(props, p->name);
        p->sticky_session = jk_get_is_sticky_session(props, p->name);
        
        if(jk_get_lb_worker_list(props,
                                 p->name,
                                 &worker_names, 
                                 &num_of_workers) && num_of_workers) {
            unsigned i = 0;
            unsigned j = 0;

            p->lb_workers = jk_pool_alloc(&p->p, 
                                          num_of_workers * sizeof(worker_record_t));

            if(!p->lb_workers) {
                return JK_FALSE;
            }

            for(i = 0 ; i < num_of_workers ; i++) {
                p->lb_workers[i].name = jk_pool_strdup(&p->p, worker_names[i]);
                p->lb_workers[i].lb_factor = jk_get_lb_factor(props, 
                                                               worker_names[i]);
                if(p->lb_workers[i].lb_factor < 0) {
                    p->lb_workers[i].lb_factor = -1 * p->lb_workers[i].lb_factor;
                }
                if (0 < p->lb_workers[i].lb_factor) {
                    p->lb_workers[i].lb_factor = 1/p->lb_workers[i].lb_factor;
                }

                p->lb_workers[i].is_local_worker = jk_get_is_local_worker(props, worker_names[i]);
                if (p->lb_workers[i].is_local_worker) p->in_local_worker_mode = JK_TRUE;
                /* 
                 * Allow using lb in fault-tolerant mode.
                 * A value of 0 means the worker will be used for all requests without
                 * sessions
                 */
                p->lb_workers[i].lb_value = p->lb_workers[i].lb_factor;
                p->lb_workers[i].in_error_state = JK_FALSE;
                p->lb_workers[i].in_recovering  = JK_FALSE;
                if(!wc_create_worker(p->lb_workers[i].name, 
                                     props, 
                                     &(p->lb_workers[i].w), 
                                     we,
                                     l) || !p->lb_workers[i].w) {
                    break;
                } else if (p->lb_workers[i].is_local_worker) {
                    /*
                     * If lb_value is 0 than move it at the beginning of the list
                     */
                    if (i != j) {
                        worker_record_t tmp_worker;
                        tmp_worker = p->lb_workers[j];
                        p->lb_workers[j] = p->lb_workers[i];
                        p->lb_workers[i] = tmp_worker;
                    }
                    j++;
                }
            }
            
            if (!p->in_local_worker_mode) {
                p->local_worker_only = JK_FALSE;
            }
            
            if(i != num_of_workers) {
                close_workers(p, i, l);
                jk_log(l, JK_LOG_DEBUG, 
                       "In jk_worker_t::validate: Failed to create worker %s\n",
                       p->lb_workers[i].name);

            } else {
                for (i = 0; i < num_of_workers; i++) {
                    jk_log(l, JK_LOG_DEBUG,
                        "Balanced worker %i has name %s\n",
                        i, p->lb_workers[i].name);
                }
                jk_log(l, JK_LOG_DEBUG,
                        "in_local_worker_mode: %s\n",
                        (p->in_local_worker_mode ? "true" : "false"));
                jk_log(l, JK_LOG_DEBUG,
                        "local_worker_only: %s\n",
                        (p->local_worker_only ? "true" : "false"));
                p->num_of_workers = num_of_workers;
                return JK_TRUE;
            }
        }        
    }

    jk_log(l, JK_LOG_ERROR, 
           "In jk_worker_t::validate: NULL Parameters\n");

    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)
{
    /* Nothing to do for now */
    return JK_TRUE;
}

static int JK_METHOD get_endpoint(jk_worker_t *pThis,
                                  jk_endpoint_t **pend,
                                  jk_logger_t *l)
{
    jk_log(l, JK_LOG_DEBUG, 
           "Into jk_worker_t::get_endpoint\n");

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

            return JK_TRUE;
        }
        jk_log(l, JK_LOG_ERROR, 
               "In jk_worker_t::get_endpoint, malloc failed\n");
    } else {
        jk_log(l, JK_LOG_ERROR, 
               "In jk_worker_t::get_endpoint, NULL parameters\n");
    }

    return JK_FALSE;
}

static int JK_METHOD destroy(jk_worker_t **pThis,
                             jk_logger_t *l)
{
    jk_log(l, JK_LOG_DEBUG, 
           "Into jk_worker_t::destroy\n");

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

        close_workers(private_data, 
                      private_data->num_of_workers,
                      l);

        jk_close_pool(&private_data->p);
        free(private_data);

        return JK_TRUE;
    }

    jk_log(l, JK_LOG_ERROR, 
           "In jk_worker_t::destroy, NULL parameters\n");
    return JK_FALSE;
}

int JK_METHOD lb_worker_factory(jk_worker_t **w,
                                const char *name,
                                jk_logger_t *l)
{
    jk_log(l, JK_LOG_DEBUG, 
           "Into lb_worker_factory\n");

    if(NULL != name && NULL != w) {
        lb_worker_t *private_data = 
            (lb_worker_t *)malloc(sizeof(lb_worker_t));

        if(private_data) {

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

            private_data->name = jk_pool_strdup(&private_data->p, name);          

            if(private_data->name) {
                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;

                *w = &private_data->worker;
                return JK_TRUE;
            }

            jk_close_pool(&private_data->p);
            free(private_data);
        }
        jk_log(l, JK_LOG_ERROR, 
               "In lb_worker_factory, malloc failed\n");
    } else {
        jk_log(l, JK_LOG_ERROR, 
               "In lb_worker_factory, NULL parameters\n");
    }

    return JK_FALSE;
}