| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| /*************************************************************************** |
| * Description: Load balancer worker, knows how to load balance among * |
| * several workers. * |
| * Author: Gal Shachor <shachor@il.ibm.com> * |
| * Author: Mladen Turk <mturk@apache.org> * |
| * Author: Rainer Jung <rjung@apache.org> * |
| * Based on: * |
| * Version: $Revision$ * |
| ***************************************************************************/ |
| |
| #include "jk_pool.h" |
| #include "jk_service.h" |
| #include "jk_util.h" |
| #include "jk_worker.h" |
| #include "jk_lb_worker.h" |
| #include "jk_ajp13.h" |
| #include "jk_ajp13_worker.h" |
| #include "jk_ajp14_worker.h" |
| #include "jk_mt.h" |
| #include "jk_shm.h" |
| |
| /* |
| * The load balancing code in this |
| */ |
| |
| #define JK_WORKER_USABLE(w) ((w)->state != JK_LB_STATE_ERROR && (w)->state != JK_LB_STATE_PROBE && (w)->state != JK_LB_STATE_BUSY && (w)->activation != JK_LB_ACTIVATION_STOPPED && (w)->activation != JK_LB_ACTIVATION_DISABLED) |
| #define JK_WORKER_USABLE_STICKY(w) ((w)->state != JK_LB_STATE_ERROR && (w)->state != JK_LB_STATE_PROBE && (w)->activation != JK_LB_ACTIVATION_STOPPED) |
| |
| static const char *lb_locking_type[] = { |
| JK_LB_LOCK_TEXT_OPTIMISTIC, |
| JK_LB_LOCK_TEXT_PESSIMISTIC, |
| "unknown", |
| NULL |
| }; |
| |
| static const char *lb_method_type[] = { |
| JK_LB_METHOD_TEXT_REQUESTS, |
| JK_LB_METHOD_TEXT_TRAFFIC, |
| JK_LB_METHOD_TEXT_BUSYNESS, |
| JK_LB_METHOD_TEXT_SESSIONS, |
| "unknown", |
| NULL |
| }; |
| |
| static const char *lb_state_type[] = { |
| JK_LB_STATE_TEXT_NA, |
| JK_LB_STATE_TEXT_OK, |
| JK_LB_STATE_TEXT_RECOVER, |
| JK_LB_STATE_TEXT_BUSY, |
| JK_LB_STATE_TEXT_ERROR, |
| JK_LB_STATE_TEXT_FORCE, |
| JK_LB_STATE_TEXT_PROBE, |
| "unknown", |
| NULL |
| }; |
| |
| static const char *lb_activation_type[] = { |
| JK_LB_ACTIVATION_TEXT_ACTIVE, |
| JK_LB_ACTIVATION_TEXT_DISABLED, |
| JK_LB_ACTIVATION_TEXT_STOPPED, |
| "unknown", |
| NULL |
| }; |
| |
| static const char *lb_first_log_names[] = { |
| JK_NOTE_LB_FIRST_NAME, |
| JK_NOTE_LB_FIRST_VALUE, |
| JK_NOTE_LB_FIRST_ACCESSED, |
| JK_NOTE_LB_FIRST_READ, |
| JK_NOTE_LB_FIRST_TRANSFERRED, |
| JK_NOTE_LB_FIRST_ERRORS, |
| JK_NOTE_LB_FIRST_BUSY, |
| JK_NOTE_LB_FIRST_ACTIVATION, |
| JK_NOTE_LB_FIRST_STATE, |
| NULL |
| }; |
| |
| static const char *lb_last_log_names[] = { |
| JK_NOTE_LB_LAST_NAME, |
| JK_NOTE_LB_LAST_VALUE, |
| JK_NOTE_LB_LAST_ACCESSED, |
| JK_NOTE_LB_LAST_READ, |
| JK_NOTE_LB_LAST_TRANSFERRED, |
| JK_NOTE_LB_LAST_ERRORS, |
| JK_NOTE_LB_LAST_BUSY, |
| JK_NOTE_LB_LAST_ACTIVATION, |
| JK_NOTE_LB_LAST_STATE, |
| NULL |
| }; |
| |
| struct lb_endpoint |
| { |
| lb_worker_t *worker; |
| |
| jk_endpoint_t endpoint; |
| }; |
| typedef struct lb_endpoint lb_endpoint_t; |
| |
| |
| /* Calculate the greatest common divisor of two positive integers */ |
| static jk_uint64_t gcd(jk_uint64_t a, jk_uint64_t b) |
| { |
| jk_uint64_t r; |
| if (b > a) { |
| r = a; |
| a = b; |
| b = r; |
| } |
| while (b > 0) { |
| r = a % b; |
| a = b; |
| b = r; |
| } |
| return a; |
| } |
| |
| /* Calculate the smallest common multiple of two positive integers */ |
| static jk_uint64_t scm(jk_uint64_t a, jk_uint64_t b) |
| { |
| return a * b / gcd(a, b); |
| } |
| |
| /* Return the string representation of the lb lock type */ |
| const char *jk_lb_get_lock(lb_worker_t *p, jk_logger_t *l) |
| { |
| return lb_locking_type[p->lblock]; |
| } |
| |
| /* Return the int representation of the lb lock type */ |
| int jk_lb_get_lock_code(const char *v) |
| { |
| if (!v) |
| return JK_LB_LOCK_DEF; |
| else if (*v == 'o' || *v == 'O' || *v == '0') |
| return JK_LB_LOCK_OPTIMISTIC; |
| else if (*v == 'p' || *v == 'P' || *v == '1') |
| return JK_LB_LOCK_PESSIMISTIC; |
| else |
| return JK_LB_LOCK_DEF; |
| } |
| |
| /* Return the string representation of the lb method type */ |
| const char *jk_lb_get_method(lb_worker_t *p, jk_logger_t *l) |
| { |
| return lb_method_type[p->lbmethod]; |
| } |
| |
| /* Return the int representation of the lb lock type */ |
| int jk_lb_get_method_code(const char *v) |
| { |
| if (!v) |
| return JK_LB_METHOD_DEF; |
| else if (*v == 'r' || *v == 'R' || *v == '0') |
| return JK_LB_METHOD_REQUESTS; |
| else if (*v == 't' || *v == 'T' || *v == '1') |
| return JK_LB_METHOD_TRAFFIC; |
| else if (*v == 'b' || *v == 'B' || *v == '2') |
| return JK_LB_METHOD_BUSYNESS; |
| else if (*v == 's' || *v == 'S' || *v == '3') |
| return JK_LB_METHOD_SESSIONS; |
| else |
| return JK_LB_METHOD_DEF; |
| } |
| |
| /* Return the string representation of the balance worker state */ |
| const char *jk_lb_get_state(worker_record_t *p, jk_logger_t *l) |
| { |
| return lb_state_type[p->s->state]; |
| } |
| |
| /* Return the int representation of the lb lock type */ |
| int jk_lb_get_state_code(const char *v) |
| { |
| if (!v) |
| return JK_LB_STATE_DEF; |
| else if (*v == 'n' || *v == 'N' || *v == '0') |
| return JK_LB_STATE_NA; |
| else if (*v == 'o' || *v == 'O' || *v == '1') |
| return JK_LB_STATE_OK; |
| else if (*v == 'r' || *v == 'R' || *v == '2') |
| return JK_LB_STATE_RECOVER; |
| else if (*v == 'b' || *v == 'B' || *v == '3') |
| return JK_LB_STATE_BUSY; |
| else if (*v == 'e' || *v == 'E' || *v == '4') |
| return JK_LB_STATE_ERROR; |
| else if (*v == 'f' || *v == 'F' || *v == '5') |
| return JK_LB_STATE_FORCE; |
| else if (*v == 'p' || *v == 'P' || *v == '6') |
| return JK_LB_STATE_PROBE; |
| else |
| return JK_LB_STATE_DEF; |
| } |
| |
| /* Return the string representation of the balance worker activation */ |
| const char *jk_lb_get_activation(worker_record_t *p, jk_logger_t *l) |
| { |
| return lb_activation_type[p->s->activation]; |
| } |
| |
| int jk_lb_get_activation_code(const char *v) |
| { |
| if (!v) |
| return JK_LB_ACTIVATION_DEF; |
| else if (*v == 'a' || *v == 'A' || *v == '0') |
| return JK_LB_ACTIVATION_ACTIVE; |
| else if (*v == 'd' || *v == 'D' || *v == '1') |
| return JK_LB_ACTIVATION_DISABLED; |
| else if (*v == 's' || *v == 'S' || *v == '2') |
| return JK_LB_ACTIVATION_STOPPED; |
| else |
| return JK_LB_ACTIVATION_DEF; |
| } |
| |
| /* Update the load multipliers wrt. lb_factor */ |
| void update_mult(lb_worker_t *p, jk_logger_t *l) |
| { |
| unsigned int i = 0; |
| jk_uint64_t s = 1; |
| JK_TRACE_ENTER(l); |
| for (i = 0; i < p->num_of_workers; i++) { |
| s = scm(s, p->lb_workers[i].s->lb_factor); |
| } |
| for (i = 0; i < p->num_of_workers; i++) { |
| p->lb_workers[i].s->lb_mult = s / p->lb_workers[i].s->lb_factor; |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "worker %s gets multiplicity %" |
| JK_UINT64_T_FMT, |
| p->lb_workers[i].s->name, |
| p->lb_workers[i].s->lb_mult); |
| } |
| JK_TRACE_EXIT(l); |
| } |
| |
| /* Reset all lb values. |
| */ |
| void reset_lb_values(lb_worker_t *p, jk_logger_t *l) |
| { |
| unsigned int i = 0; |
| JK_TRACE_ENTER(l); |
| if (p->lbmethod != JK_LB_METHOD_BUSYNESS) { |
| for (i = 0; i < p->num_of_workers; i++) { |
| p->lb_workers[i].s->lb_value = 0; |
| } |
| } |
| JK_TRACE_EXIT(l); |
| } |
| |
| /* Syncing config values from shm */ |
| void jk_lb_pull(lb_worker_t * p, jk_logger_t *l) { |
| JK_TRACE_ENTER(l); |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "syncing mem for lb '%s' from shm", |
| p->s->name); |
| p->sticky_session = p->s->sticky_session; |
| p->sticky_session_force = p->s->sticky_session_force; |
| p->recover_wait_time = p->s->recover_wait_time; |
| p->retries = p->s->retries; |
| p->lbmethod = p->s->lbmethod; |
| p->lblock = p->s->lblock; |
| p->sequence = p->s->sequence; |
| JK_TRACE_EXIT(l); |
| } |
| |
| /* Syncing config values from shm */ |
| void jk_lb_push(lb_worker_t * p, jk_logger_t *l) { |
| JK_TRACE_ENTER(l); |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "syncing shm for lb '%s' from mem", |
| p->s->name); |
| p->s->sticky_session = p->sticky_session; |
| p->s->sticky_session_force = p->sticky_session_force; |
| p->s->recover_wait_time = p->recover_wait_time; |
| p->s->retries = p->retries; |
| p->s->lbmethod = p->lbmethod; |
| p->s->lblock = p->lblock; |
| p->s->sequence = p->sequence; |
| JK_TRACE_EXIT(l); |
| } |
| |
| /* Retrieve the parameter with the given name */ |
| static char *get_path_param(jk_ws_service_t *s, const char *name) |
| { |
| char *id_start = NULL; |
| for (id_start = strstr(s->req_uri, name); |
| id_start; id_start = strstr(id_start + 1, name)) { |
| if (id_start[strlen(name)] == '=') { |
| /* |
| * Session path-cookie was found, get it's value |
| */ |
| id_start += (1 + strlen(name)); |
| if (strlen(id_start)) { |
| char *id_end; |
| id_start = jk_pool_strdup(s->pool, id_start); |
| /* |
| * The query string is not part of req_uri, however |
| * to be on the safe side lets remove the trailing query |
| * string if appended... |
| */ |
| if ((id_end = strchr(id_start, '?')) != NULL) { |
| *id_end = '\0'; |
| } |
| /* |
| * Remove any trailing path element. |
| */ |
| if ((id_end = strchr(id_start, ';')) != NULL) { |
| *id_end = '\0'; |
| } |
| return id_start; |
| } |
| } |
| } |
| |
| return NULL; |
| } |
| |
| /* Retrieve the cookie with the given name */ |
| static char *get_cookie(jk_ws_service_t *s, const char *name) |
| { |
| unsigned i; |
| char *result = NULL; |
| |
| for (i = 0; i < s->num_headers; i++) { |
| if (strcasecmp(s->headers_names[i], "cookie") == 0) { |
| |
| char *id_start; |
| for (id_start = strstr(s->headers_values[i], name); |
| id_start; id_start = strstr(id_start + 1, name)) { |
| if (id_start == s->headers_values[i] || |
| id_start[-1] == ';' || |
| id_start[-1] == ',' || isspace(id_start[-1])) { |
| id_start += strlen(name); |
| while (*id_start && isspace(*id_start)) |
| ++id_start; |
| if (*id_start == '=' && id_start[1]) { |
| /* |
| * Session cookie was found, get it's value |
| */ |
| char *id_end; |
| ++id_start; |
| id_start = jk_pool_strdup(s->pool, id_start); |
| if ((id_end = strchr(id_start, ';')) != NULL) { |
| *id_end = '\0'; |
| } |
| if ((id_end = strchr(id_start, ',')) != NULL) { |
| *id_end = '\0'; |
| } |
| if (result == NULL) { |
| result = id_start; |
| } |
| else { |
| size_t osz = strlen(result) + 1; |
| size_t sz = osz + strlen(id_start) + 1; |
| result = |
| jk_pool_realloc(s->pool, sz, result, osz); |
| strcat(result, ";"); |
| strcat(result, id_start); |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| return result; |
| } |
| |
| |
| /* Retrieve session id from the cookie or the parameter |
| * (parameter first) |
| */ |
| static char *get_sessionid(jk_ws_service_t *s) |
| { |
| char *val; |
| val = get_path_param(s, JK_PATH_SESSION_IDENTIFIER); |
| if (!val) { |
| val = get_cookie(s, JK_SESSION_IDENTIFIER); |
| } |
| return val; |
| } |
| |
| static void close_workers(lb_worker_t * p, int num_of_workers, jk_logger_t *l) |
| { |
| int i = 0; |
| for (i = 0; i < num_of_workers; i++) { |
| p->lb_workers[i].w->destroy(&(p->lb_workers[i].w), l); |
| } |
| } |
| |
| /* If the worker is in error state run |
| * retry on that worker. It will be marked as |
| * operational if the retry timeout is elapsed. |
| * The worker might still be unusable, but we try |
| * anyway. |
| * If the worker is in ok state and got no requests |
| * since the last global maintenance, we mark its |
| * state as not available. |
| * Return the number of workers not in error state. |
| */ |
| static int recover_workers(lb_worker_t *p, |
| jk_uint64_t curmax, |
| time_t now, |
| jk_logger_t *l) |
| { |
| unsigned int i; |
| int non_error = 0; |
| int elapsed; |
| worker_record_t *w = NULL; |
| JK_TRACE_ENTER(l); |
| |
| if (p->sequence != p->s->sequence) |
| jk_lb_pull(p, l); |
| for (i = 0; i < p->num_of_workers; i++) { |
| w = &p->lb_workers[i]; |
| if (w->s->state == JK_LB_STATE_ERROR) { |
| elapsed = (int)difftime(now, w->s->error_time); |
| if (elapsed <= p->s->recover_wait_time) { |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "worker %s will recover in %d seconds", |
| w->s->name, p->s->recover_wait_time - elapsed); |
| } |
| else { |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "worker %s is marked for recovery", |
| w->s->name); |
| if (p->lbmethod != JK_LB_METHOD_BUSYNESS) |
| w->s->lb_value = curmax; |
| w->s->state = JK_LB_STATE_RECOVER; |
| non_error++; |
| } |
| } |
| else { |
| non_error++; |
| if (w->s->state == JK_LB_STATE_OK && |
| w->s->elected == w->s->elected_snapshot) |
| w->s->state = JK_LB_STATE_NA; |
| } |
| w->s->elected_snapshot = w->s->elected; |
| } |
| |
| JK_TRACE_EXIT(l); |
| return non_error; |
| } |
| |
| static int force_recovery(lb_worker_t *p, |
| jk_logger_t *l) |
| { |
| unsigned int i; |
| int forced = 0; |
| worker_record_t *w = NULL; |
| JK_TRACE_ENTER(l); |
| |
| for (i = 0; i < p->num_of_workers; i++) { |
| w = &p->lb_workers[i]; |
| if (w->s->state == JK_LB_STATE_ERROR) { |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_INFO, |
| "worker %s is marked for recovery", |
| w->s->name); |
| w->s->state = JK_LB_STATE_FORCE; |
| forced++; |
| } |
| } |
| |
| JK_TRACE_EXIT(l); |
| return forced; |
| } |
| |
| /* Divide old load values by the decay factor, |
| * such that older values get less important |
| * for the routing decisions. |
| */ |
| static jk_uint64_t decay_load(lb_worker_t *p, |
| time_t exponent, |
| jk_logger_t *l) |
| { |
| unsigned int i; |
| jk_uint64_t curmax = 0; |
| |
| JK_TRACE_ENTER(l); |
| if (p->lbmethod != JK_LB_METHOD_BUSYNESS) { |
| for (i = 0; i < p->num_of_workers; i++) { |
| p->lb_workers[i].s->lb_value >>= exponent; |
| if (p->lb_workers[i].s->lb_value > curmax) { |
| curmax = p->lb_workers[i].s->lb_value; |
| } |
| } |
| } |
| JK_TRACE_EXIT(l); |
| return curmax; |
| } |
| |
| static int JK_METHOD maintain_workers(jk_worker_t *p, time_t now, jk_logger_t *l) |
| { |
| unsigned int i = 0; |
| jk_uint64_t curmax = 0; |
| long delta; |
| |
| JK_TRACE_ENTER(l); |
| if (p && p->worker_private) { |
| lb_worker_t *lb = (lb_worker_t *)p->worker_private; |
| |
| for (i = 0; i < lb->num_of_workers; i++) { |
| if (lb->lb_workers[i].w->maintain) { |
| lb->lb_workers[i].w->maintain(lb->lb_workers[i].w, now, l); |
| } |
| } |
| |
| jk_shm_lock(); |
| |
| /* Now we check for global maintenance (once for all processes). |
| * Checking workers for recovery and applying decay to the |
| * load values should not be done by each process individually. |
| * Therefore we globally sync and we use a global timestamp. |
| * Since it's possible that we come here a few milliseconds |
| * before the interval has passed, we allow a little tolerance. |
| */ |
| delta = (long)difftime(now, lb->s->last_maintain_time) + JK_LB_MAINTAIN_TOLERANCE; |
| if (delta >= lb->maintain_time) { |
| lb->s->last_maintain_time = now; |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "decay with 2^%d", |
| JK_LB_DECAY_MULT * delta / lb->maintain_time); |
| curmax = decay_load(lb, JK_LB_DECAY_MULT * delta / lb->maintain_time, l); |
| if (!recover_workers(lb, curmax, now, l)) { |
| force_recovery(lb, l); |
| } |
| } |
| |
| jk_shm_unlock(); |
| |
| } |
| else { |
| JK_LOG_NULL_PARAMS(l); |
| } |
| |
| JK_TRACE_EXIT(l); |
| return JK_TRUE; |
| } |
| |
| static worker_record_t *find_by_session(lb_worker_t *p, |
| const char *name, |
| jk_logger_t *l) |
| { |
| |
| worker_record_t *rc = NULL; |
| unsigned int i; |
| |
| for (i = 0; i < p->num_of_workers; i++) { |
| if (strcmp(p->lb_workers[i].s->route, name) == 0) { |
| rc = &p->lb_workers[i]; |
| rc->r = &(rc->s->route[0]); |
| break; |
| } |
| } |
| return rc; |
| } |
| |
| static worker_record_t *find_best_bydomain(lb_worker_t *p, |
| const char *domain, |
| jk_logger_t *l) |
| { |
| unsigned int i; |
| int d = 0; |
| jk_uint64_t curmin = 0; |
| |
| worker_record_t *candidate = NULL; |
| |
| /* First try to see if we have available candidate */ |
| for (i = 0; i < p->num_of_workers; i++) { |
| /* Skip all workers that are not member of domain */ |
| if (strlen(p->lb_workers[i].s->domain) == 0 || |
| strcmp(p->lb_workers[i].s->domain, domain)) |
| continue; |
| /* Take into calculation only the workers that are |
| * not in error state, stopped, disabled or busy. |
| */ |
| if (JK_WORKER_USABLE(p->lb_workers[i].s)) { |
| if (!candidate || p->lb_workers[i].s->distance < d || |
| (p->lb_workers[i].s->lb_value < curmin && |
| p->lb_workers[i].s->distance == d)) { |
| candidate = &p->lb_workers[i]; |
| curmin = p->lb_workers[i].s->lb_value; |
| d = p->lb_workers[i].s->distance; |
| } |
| } |
| } |
| |
| if (candidate) { |
| candidate->r = &(candidate->s->domain[0]); |
| } |
| |
| return candidate; |
| } |
| |
| |
| static worker_record_t *find_best_byvalue(lb_worker_t *p, |
| jk_logger_t *l) |
| { |
| static unsigned int next_offset = 0; |
| unsigned int i; |
| unsigned int j; |
| unsigned int offset; |
| int d = 0; |
| jk_uint64_t curmin = 0; |
| |
| /* find the least busy worker */ |
| worker_record_t *candidate = NULL; |
| |
| offset = next_offset; |
| |
| /* First try to see if we have available candidate */ |
| for (j = offset; j < p->num_of_workers + offset; j++) { |
| i = j % p->num_of_workers; |
| |
| /* Take into calculation only the workers that are |
| * not in error state, stopped, disabled or busy. |
| */ |
| if (JK_WORKER_USABLE(p->lb_workers[i].s)) { |
| if (!candidate || p->lb_workers[i].s->distance < d || |
| (p->lb_workers[i].s->lb_value < curmin && |
| p->lb_workers[i].s->distance == d)) { |
| candidate = &p->lb_workers[i]; |
| curmin = p->lb_workers[i].s->lb_value; |
| d = p->lb_workers[i].s->distance; |
| next_offset = i + 1; |
| } |
| } |
| } |
| return candidate; |
| } |
| |
| static worker_record_t *find_bysession_route(lb_worker_t *p, |
| const char *name, |
| jk_logger_t *l) |
| { |
| int uses_domain = 0; |
| worker_record_t *candidate = NULL; |
| |
| candidate = find_by_session(p, name, l); |
| if (!candidate) { |
| uses_domain = 1; |
| candidate = find_best_bydomain(p, name, l); |
| } |
| if (candidate) { |
| if (!JK_WORKER_USABLE_STICKY(candidate->s)) { |
| /* We have a worker that is error state or stopped. |
| * If it has a redirection set use that redirection worker. |
| * This enables to safely remove the member from the |
| * balancer. Of course you will need a some kind of |
| * session replication between those two remote. |
| */ |
| if (p->sticky_session_force) |
| candidate = NULL; |
| else if (*candidate->s->redirect) |
| candidate = find_by_session(p, candidate->s->redirect, l); |
| else if (*candidate->s->domain && !uses_domain) { |
| uses_domain = 1; |
| candidate = find_best_bydomain(p, candidate->s->domain, l); |
| } |
| if (candidate && !JK_WORKER_USABLE_STICKY(candidate->s)) |
| candidate = NULL; |
| } |
| } |
| return candidate; |
| } |
| |
| static worker_record_t *find_failover_worker(lb_worker_t * p, |
| jk_logger_t *l) |
| { |
| worker_record_t *rc = NULL; |
| unsigned int i; |
| const char *redirect = NULL; |
| |
| for (i = 0; i < p->num_of_workers; i++) { |
| if (strlen(p->lb_workers[i].s->redirect)) { |
| redirect = &(p->lb_workers[i].s->redirect[0]); |
| break; |
| } |
| } |
| if (redirect) |
| rc = find_bysession_route(p, redirect, l); |
| return rc; |
| } |
| |
| static worker_record_t *find_best_worker(lb_worker_t * p, |
| jk_logger_t *l) |
| { |
| worker_record_t *rc = NULL; |
| |
| rc = find_best_byvalue(p, l); |
| /* By default use worker route as session route */ |
| if (rc) |
| rc->r = &(rc->s->route[0]); |
| else |
| rc = find_failover_worker(p, l); |
| return rc; |
| } |
| |
| static worker_record_t *get_most_suitable_worker(lb_worker_t * p, |
| char *sessionid, |
| jk_ws_service_t *s, |
| jk_logger_t *l) |
| { |
| worker_record_t *rc = NULL; |
| int r; |
| |
| JK_TRACE_ENTER(l); |
| if (p->num_of_workers == 1) { |
| /* No need to find the best worker |
| * if there is a single one |
| */ |
| if (JK_WORKER_USABLE_STICKY(p->lb_workers[0].s)) { |
| if (p->lb_workers[0].s->activation != JK_LB_ACTIVATION_DISABLED) { |
| p->lb_workers[0].r = &(p->lb_workers[0].s->route[0]); |
| JK_TRACE_EXIT(l); |
| return &p->lb_workers[0]; |
| } |
| } |
| else { |
| JK_TRACE_EXIT(l); |
| return NULL; |
| } |
| } |
| if (p->lblock == JK_LB_LOCK_PESSIMISTIC) |
| r = jk_shm_lock(); |
| else { |
| JK_ENTER_CS(&(p->cs), r); |
| } |
| if (!r) { |
| jk_log(l, JK_LOG_ERROR, |
| "locking failed (errno=%d)", |
| errno); |
| JK_TRACE_EXIT(l); |
| return NULL; |
| } |
| if (sessionid) { |
| char *session = sessionid; |
| while (sessionid) { |
| char *next = strchr(sessionid, ';'); |
| char *session_route = NULL; |
| if (next) |
| *next++ = '\0'; |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "searching worker for partial sessionid %s", |
| sessionid); |
| session_route = strchr(sessionid, '.'); |
| if (session_route) { |
| ++session_route; |
| |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "searching worker for session route %s", |
| session_route); |
| |
| /* We have a session route. Whow! */ |
| rc = find_bysession_route(p, session_route, l); |
| if (rc) { |
| if (p->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_unlock(); |
| else { |
| JK_LEAVE_CS(&(p->cs), r); |
| } |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "found worker %s (%s) for route %s and partial sessionid %s", |
| rc->s->name, rc->s->route, session_route, sessionid); |
| JK_TRACE_EXIT(l); |
| return rc; |
| } |
| } |
| /* Try next partial sessionid if present */ |
| sessionid = next; |
| rc = NULL; |
| } |
| if (!rc && p->sticky_session_force) { |
| if (p->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_unlock(); |
| else { |
| JK_LEAVE_CS(&(p->cs), r); |
| } |
| jk_log(l, JK_LOG_INFO, |
| "all workers are in error state for session %s", |
| session); |
| JK_TRACE_EXIT(l); |
| return NULL; |
| } |
| } |
| rc = find_best_worker(p, l); |
| if (p->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_unlock(); |
| else { |
| JK_LEAVE_CS(&(p->cs), r); |
| } |
| if (rc && JK_IS_DEBUG_LEVEL(l)) { |
| jk_log(l, JK_LOG_DEBUG, |
| "found best worker %s (%s) using method '%s'", |
| rc->s->name, rc->s->route, jk_lb_get_method(p, l)); |
| } |
| JK_TRACE_EXIT(l); |
| return rc; |
| } |
| |
| static void lb_add_log_items(jk_ws_service_t *s, |
| const char *const *log_names, |
| worker_record_t *w, |
| jk_logger_t *l) |
| { |
| const char **log_values = jk_pool_alloc(s->pool, sizeof(char *) * JK_LB_NOTES_COUNT); |
| char *buf = jk_pool_alloc(s->pool, sizeof(char *) * JK_LB_NOTES_COUNT * JK_LB_UINT64_STR_SZ); |
| if (log_values && buf) { |
| /* JK_NOTE_LB_FIRST/LAST_NAME */ |
| log_values[0] = w->s->name; |
| snprintf(buf, JK_LB_UINT64_STR_SZ, "%" JK_UINT64_T_FMT, w->s->lb_value); |
| /* JK_NOTE_LB_FIRST/LAST_VALUE */ |
| log_values[1] = buf; |
| buf += JK_LB_UINT64_STR_SZ; |
| snprintf(buf, JK_LB_UINT64_STR_SZ, "%" JK_UINT64_T_FMT, w->s->elected); |
| /* JK_NOTE_LB_FIRST/LAST_ACCESSED */ |
| log_values[2] = buf; |
| buf += JK_LB_UINT64_STR_SZ; |
| snprintf(buf, JK_LB_UINT64_STR_SZ, "%" JK_UINT64_T_FMT, w->s->readed); |
| /* JK_NOTE_LB_FIRST/LAST_READ */ |
| log_values[3] = buf; |
| buf += JK_LB_UINT64_STR_SZ; |
| snprintf(buf, JK_LB_UINT64_STR_SZ, "%" JK_UINT64_T_FMT, w->s->transferred); |
| /* JK_NOTE_LB_FIRST/LAST_TRANSFERRED */ |
| log_values[4] = buf; |
| buf += JK_LB_UINT64_STR_SZ; |
| snprintf(buf, JK_LB_UINT64_STR_SZ, "%" JK_UINT32_T_FMT, w->s->errors); |
| /* JK_NOTE_LB_FIRST/LAST_ERRORS */ |
| log_values[5] = buf; |
| buf += JK_LB_UINT64_STR_SZ; |
| snprintf(buf, JK_LB_UINT64_STR_SZ, "%d", w->s->busy); |
| /* JK_NOTE_LB_FIRST/LAST_BUSY */ |
| log_values[6] = buf; |
| /* JK_NOTE_LB_FIRST/LAST_ACTIVATION */ |
| log_values[7] = jk_lb_get_activation(w, l); |
| /* JK_NOTE_LB_FIRST/LAST_STATE */ |
| log_values[8] = jk_lb_get_state(w, l); |
| s->add_log_items(s, log_names, log_values, JK_LB_NOTES_COUNT); |
| } |
| } |
| |
| static int JK_METHOD service(jk_endpoint_t *e, |
| jk_ws_service_t *s, |
| jk_logger_t *l, int *is_error) |
| { |
| lb_endpoint_t *p; |
| int attempt = 1; |
| worker_record_t *prec = NULL; |
| int num_of_workers; |
| int first = 1; |
| int was_forced = 0; |
| int rc = -1; |
| char *sessionid = NULL; |
| |
| JK_TRACE_ENTER(l); |
| |
| if (is_error) |
| *is_error = JK_HTTP_SERVER_ERROR; |
| if (!e || !e->endpoint_private || !s || !is_error) { |
| JK_LOG_NULL_PARAMS(l); |
| JK_TRACE_EXIT(l); |
| return JK_FALSE; |
| } |
| |
| p = e->endpoint_private; |
| num_of_workers = p->worker->num_of_workers; |
| |
| /* Set returned error to OK */ |
| *is_error = JK_HTTP_OK; |
| |
| /* set the recovery post, for LB mode */ |
| s->reco_buf = jk_b_new(s->pool); |
| jk_b_set_buffer_size(s->reco_buf, p->worker->max_packet_size); |
| jk_b_reset(s->reco_buf); |
| s->reco_status = RECO_INITED; |
| |
| jk_shm_lock(); |
| if (p->worker->sequence != p->worker->s->sequence) |
| jk_lb_pull(p->worker, l); |
| jk_shm_unlock(); |
| |
| if (p->worker->sticky_session) { |
| /* Use sessionid only if sticky_session is |
| * defined for this load balancer |
| */ |
| sessionid = get_sessionid(s); |
| } |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "service sticky_session=%d id='%s'", |
| p->worker->sticky_session, sessionid ? sessionid : "empty"); |
| |
| while (attempt <= num_of_workers && rc == -1) { |
| worker_record_t *rec = |
| get_most_suitable_worker(p->worker, sessionid, s, l); |
| /* Do not reuse previous worker, because |
| * that worker already failed. |
| */ |
| if (rec) { |
| int r; |
| int is_service_error = JK_HTTP_OK; |
| jk_endpoint_t *end = NULL; |
| int retry = 0; |
| int retry_wait = JK_LB_MIN_RETRY_WAIT; |
| s->route = rec->r; |
| prec = rec; |
| |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "service worker=%s route=%s", |
| rec->s->name, s->route); |
| |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_lock(); |
| if (rec->s->state == JK_LB_STATE_RECOVER) |
| rec->s->state = JK_LB_STATE_PROBE; |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_unlock(); |
| |
| while ((!(r=rec->w->get_endpoint(rec->w, &end, l)) || !end) && (retry < p->worker->s->retries)) { |
| retry++; |
| retry_wait *=2; |
| |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_lock(); |
| if (retry_wait > JK_LB_MAX_RETRY_WAIT) |
| retry_wait = JK_LB_MAX_RETRY_WAIT; |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_unlock(); |
| |
| if (JK_IS_DEBUG_LEVEL(l)) |
| jk_log(l, JK_LOG_DEBUG, |
| "could not get free endpoint for worker" |
| " (retry %d, sleeping for %d ms)", |
| retry, retry_wait); |
| jk_sleep(retry_wait); |
| } |
| if (!r || !end) { |
| /* If we can not get the endpoint |
| * mark the worker as busy rather then |
| * as in error if the retry number is |
| * greater then the number of retries. |
| */ |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_lock(); |
| if (rec->s->state != JK_LB_STATE_ERROR) |
| rec->s->state = JK_LB_STATE_BUSY; |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_unlock(); |
| jk_log(l, JK_LOG_INFO, |
| "could not get free endpoint for worker %s (%d retries)", |
| rec->s->name, retry); |
| } |
| else { |
| int service_stat = -1; |
| size_t rd = 0; |
| size_t wr = 0; |
| /* Reset endpoint read and write sizes for |
| * this request. |
| */ |
| end->rd = end->wr = 0; |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_lock(); |
| |
| rec->s->elected++; |
| /* Increment the number of workers serving request */ |
| p->worker->s->busy++; |
| if (p->worker->s->busy > p->worker->s->max_busy) |
| p->worker->s->max_busy = p->worker->s->busy; |
| rec->s->busy++; |
| if (rec->s->busy > rec->s->max_busy) |
| rec->s->max_busy = rec->s->busy; |
| if ( (p->worker->lbmethod == JK_LB_METHOD_REQUESTS) || |
| (p->worker->lbmethod == JK_LB_METHOD_BUSYNESS) || |
| (p->worker->lbmethod == JK_LB_METHOD_SESSIONS && |
| !sessionid) ) |
| rec->s->lb_value += rec->s->lb_mult; |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_unlock(); |
| |
| service_stat = end->service(end, s, l, &is_service_error); |
| rd = end->rd; |
| wr = end->wr; |
| end->done(&end, l); |
| |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_lock(); |
| |
| /* Update partial reads and writes if any */ |
| rec->s->readed += rd; |
| rec->s->transferred += wr; |
| if (p->worker->lbmethod == JK_LB_METHOD_TRAFFIC) { |
| rec->s->lb_value += (rd+wr)*rec->s->lb_mult; |
| } |
| else if (p->worker->lbmethod == JK_LB_METHOD_BUSYNESS) { |
| if (rec->s->lb_value >= rec->s->lb_mult) { |
| rec->s->lb_value -= rec->s->lb_mult; |
| } |
| else { |
| rec->s->lb_value = 0; |
| if (JK_IS_DEBUG_LEVEL(l)) { |
| jk_log(l, JK_LOG_DEBUG, |
| "worker %s has load value to low (%" |
| JK_UINT64_T_FMT |
| " < %" |
| JK_UINT64_T_FMT |
| ") ", |
| "- correcting to 0", |
| rec->s->name, |
| rec->s->lb_value, |
| rec->s->lb_mult); |
| } |
| } |
| } |
| |
| /* When returning the endpoint mark the worker as not busy. |
| * We have at least one endpoint free |
| */ |
| if (rec->s->state == JK_LB_STATE_BUSY) |
| rec->s->state = JK_LB_STATE_OK; |
| /* Decrement the busy worker count. |
| * Check if the busy was reset to zero by graceful |
| * restart of the server. |
| */ |
| if (rec->s->busy) |
| rec->s->busy--; |
| if (p->worker->s->busy) |
| p->worker->s->busy--; |
| if (service_stat == JK_TRUE) { |
| rec->s->state = JK_LB_STATE_OK; |
| rec->s->error_time = 0; |
| rc = JK_TRUE; |
| } |
| else if (service_stat == JK_CLIENT_ERROR) { |
| /* |
| * Client error !!! |
| * Since this is bad request do not fail over. |
| */ |
| rec->s->client_errors++; |
| rec->s->state = JK_LB_STATE_OK; |
| rec->s->error_time = 0; |
| 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); |
| *is_error = is_service_error; |
| rc = JK_CLIENT_ERROR; |
| } |
| else { |
| /* |
| * Service failed !!! |
| * Time for fault tolerance (if possible)... |
| */ |
| |
| rec->s->errors++; |
| rec->s->state = JK_LB_STATE_ERROR; |
| rec->s->error_time = time(NULL); |
| 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; |
| rc = JK_FALSE; |
| } |
| else |
| jk_log(l, JK_LOG_INFO, |
| "service failed, worker %s is in error state", |
| rec->s->name); |
| } |
| if (p->worker->lblock == JK_LB_LOCK_PESSIMISTIC) |
| jk_shm_unlock(); |
| } |
| if ( rc == -1 ) { |
| /* |
| * 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 worker"); |
| } |
| if (first == 1 && s->add_log_items) { |
| first = 0; |
| lb_add_log_items(s, lb_first_log_names, prec, l); |
| } |
| } |
| else { |
| /* NULL record, no more workers left ... */ |
| if (!was_forced) { |
| int nf; |
| /* Force recovery only once. |
| * If it still fails, Tomcat is still disconnected. |
| */ |
| jk_shm_lock(); |
| nf = force_recovery(p->worker, l); |
| jk_shm_unlock(); |
| was_forced = 1; |
| if (nf) { |
| /* We have forced recovery. |
| * Reset the service loop and go again |
| */ |
| prec = NULL; |
| rc = -1; |
| jk_log(l, JK_LOG_INFO, |
| "Forcing recovery once for %d workers", nf); |
| continue; |
| } |
| else { |
| /* No workers in error state. |
| * Somebody set them all to disabled? |
| */ |
| jk_log(l, JK_LOG_ERROR, |
| "All tomcat instances failed, no more workers left for recovery"); |
| *is_error = JK_HTTP_SERVER_BUSY; |
| rc = JK_FALSE; |
| } |
| } |
| else { |
| jk_log(l, JK_LOG_ERROR, |
| "All tomcat instances failed, no more workers left"); |
| *is_error = JK_HTTP_SERVER_BUSY; |
| rc = JK_FALSE; |
| } |
| } |
| attempt++; |
| } |
| if ( rc == -1 ) { |
| jk_log(l, JK_LOG_INFO, |
| "All tomcat instances are busy or in error state"); |
| /* Set error to Timeout */ |
| *is_error = JK_HTTP_SERVER_BUSY; |
| rc = JK_FALSE; |
| } |
| if (prec && s->add_log_items) { |
| lb_add_log_items(s, lb_last_log_names, prec, l); |
| } |
| |
| JK_TRACE_EXIT(l); |
| return rc; |
| } |
| |
| 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; |
| |
| 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->sticky_session = jk_get_is_sticky_session(props, p->s->name); |
| p->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; |
| unsigned int j = 0; |
| p->max_packet_size = DEF_BUFFER_SZ; |
| 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; |
| unsigned int ms; |
| 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; |
| } |
| /* Calculate the maximum packet size from all workers |
| * for the recovery buffer. |
| */ |
| ms = jk_get_max_packet_size(props, worker_names[i]); |
| if (ms > p->max_packet_size) |
| p->max_packet_size = ms; |
| p->lb_workers[i].s->distance = |
| jk_get_distance(props, worker_names[i]); |
| if ((s = jk_get_worker_route(props, worker_names[i], NULL))) |
| strncpy(p->lb_workers[i].s->route, s, JK_SHM_STR_SIZ); |
| else |
| strncpy(p->lb_workers[i].s->route, worker_names[i], JK_SHM_STR_SIZ); |
| 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 = 0; |
| p->lb_workers[i].s->state = JK_LB_STATE_NA; |
| p->lb_workers[i].s->error_time = 0; |
| p->lb_workers[i].s->activation = |
| jk_get_worker_activation(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 { |
| /* Update domain names if route contains period '.' */ |
| for (i = 0; i < num_of_workers; i++) { |
| if (!p->lb_workers[i].s->domain[0]) { |
| char * id_domain = strchr(p->lb_workers[i].s->route, '.'); |
| if (id_domain) { |
| *id_domain = '\0'; |
| strcpy(p->lb_workers[i].s->domain, p->lb_workers[i].s->route); |
| *id_domain = '.'; |
| } |
| } |
| if (JK_IS_DEBUG_LEVEL(l)) { |
| jk_log(l, JK_LOG_DEBUG, |
| "Balanced worker %i has name %s and route %s in domain %s", |
| i, |
| p->lb_workers[i].s->name, |
| p->lb_workers[i].s->route, |
| p->lb_workers[i].s->domain); |
| } |
| } |
| p->num_of_workers = num_of_workers; |
| update_mult(p, l); |
| for (i = 0; i < num_of_workers; i++) { |
| for (j = 0; j < i; j++) { |
| if (strcmp(p->lb_workers[i].s->route, p->lb_workers[j].s->route) == 0) { |
| jk_log(l, JK_LOG_ERROR, |
| "Balanced workers number %i (%s) and %i (%s) share the same route %s - aborting configuration!", |
| i, |
| p->lb_workers[i].s->name, |
| j, |
| p->lb_workers[j].s->name, |
| p->lb_workers[i].s->route); |
| JK_TRACE_EXIT(l); |
| return JK_FALSE; |
| } |
| } |
| } |
| 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->retries = pThis->retries; |
| p->recover_wait_time = jk_get_worker_recover_timeout(props, p->s->name, |
| WAIT_BEFORE_RECOVER); |
| if (p->recover_wait_time < 1) |
| p->recover_wait_time = 1; |
| p->maintain_time = jk_get_worker_maintain_time(props); |
| if(p->maintain_time < 0) |
| p->maintain_time = 0; |
| p->s->last_maintain_time = time(NULL); |
| |
| 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_shm_lock(); |
| p->sequence++; |
| jk_lb_push(p, log); |
| jk_shm_unlock(); |
| |
| 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->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->recover_wait_time = WAIT_BEFORE_RECOVER; |
| private_data->sequence = 0; |
| *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; |
| } |
