jk_ajp_common.c

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

    ajp_operation_t *op = &oper;

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

    if(e && e->endpoint_private && s && is_recoverable_error) {
        ajp_endpoint_t *p = e->endpoint_private;
            op->request = jk_b_new(&(p->pool));
        jk_b_set_buffer_size(op->request, DEF_BUFFER_SZ); 
        jk_b_reset(op->request);
       
        op->reply = jk_b_new(&(p->pool));
        jk_b_set_buffer_size(op->reply, DEF_BUFFER_SZ);
        jk_b_reset(op->reply); 
        
        op->post = jk_b_new(&(p->pool));
        jk_b_set_buffer_size(op->post, DEF_BUFFER_SZ);
        jk_b_reset(op->post); 
        
        op->recoverable = JK_TRUE;
        op->uploadfd     = -1;      /* not yet used, later ;) */

        p->left_bytes_to_send = s->content_length;
        p->reuse = JK_FALSE;
        *is_recoverable_error = JK_TRUE;

                s->secret = p->worker->secret;
        /* 
         * We get here initial request (in reqmsg)
         */
        if (!ajp_marshal_into_msgb(op->request, s, l, p)) {
            *is_recoverable_error = JK_FALSE;                
            return JK_FALSE;
        }

        /* 
         * JK_RETRIES could be replaced by the number of workers in
         * a load-balancing configuration 
         */
        for (i = 0; i < JK_RETRIES; i++)
        {
            /*
             * We're using reqmsg which hold initial request
             * if Tomcat is stopped or restarted, we will pass reqmsg
             * to next valid tomcat. 
             */
            if (ajp_send_request(e, s, l, p, op)) {

                /* If we have the no recoverable error, it's probably because the sender (browser)
                 * stop sending data before the end (certainly in a big post)
                 */
                if (! op->recoverable) {
                    *is_recoverable_error = JK_FALSE;
                    jk_log(l, JK_LOG_ERROR, "ERROR: sending request to tomcat failed without recovery in send loop %d\n", i);
                    return JK_FALSE;
                }

                /* Up to there we can recover */
                *is_recoverable_error = JK_TRUE;
                op->recoverable = JK_TRUE;

                if (ajp_get_reply(e, s, l, p, op))
                    return (JK_TRUE);

                /* if we can't get reply, check if no recover flag was set 
                 * if is_recoverable_error is cleared, we have started received 
                 * upload data and we must consider that operation is no more recoverable
                 */
                if (! op->recoverable) {
                    *is_recoverable_error = JK_FALSE;
                    jk_log(l, JK_LOG_ERROR, "ERROR: receiving reply from tomcat failed without recovery in send loop %d\n", i);
                    return JK_FALSE;
                }
                
                jk_log(l, JK_LOG_ERROR, "ERROR: Receiving from tomcat failed, recoverable operation. err=%d\n", i);
            }
            else
                jk_log(l, JK_LOG_INFO, "sending request to tomcat failed in send loop. err=%d\n", i);
        
            jk_close_socket(p->sd);
            p->sd = -1;
            ajp_reuse_connection(p, l);
        }
        
        /* Log the error only once per failed request. */
        jk_log(l, JK_LOG_ERROR, "Error connecting to tomcat. Tomcat is probably not started or is listenning on the wrong port. Failed errno = %d\n", errno);

    } else {
            jk_log(l, JK_LOG_ERROR, "In jk_endpoint_t::service, NULL parameters\n");
    }

    return JK_FALSE;
}
/*
 * Validate the worker (ajp13/ajp14)
 */

int ajp_validate(jk_worker_t *pThis,
                 jk_map_t    *props,
                 jk_worker_env_t *we,
                 jk_logger_t *l,
                 int          proto)
{
    int    port;
    char * host;

    jk_log(l, JK_LOG_DEBUG, "Into jk_worker_t::validate\n");

    if (proto == AJP13_PROTO) {
        port = AJP13_DEF_PORT;
        host = AJP13_DEF_HOST;
    }
    else if (proto == AJP14_PROTO) {
        port = AJP14_DEF_PORT;
        host = AJP14_DEF_HOST;
    }
    else {
        jk_log(l, JK_LOG_DEBUG, "In jk_worker_t::validate unknown protocol %d\n", proto);
        return JK_FALSE;
    } 

    if (pThis && pThis->worker_private) {
        ajp_worker_t *p = pThis->worker_private;
        port = jk_get_worker_port(props, p->name, port);
        host = jk_get_worker_host(props, p->name, host);

        jk_log(l, JK_LOG_DEBUG, "In jk_worker_t::validate for worker %s contact is %s:%d\n", p->name, host, port);

        if(port > 1024 && host) {
            if(jk_resolve(host, (short)port, &p->worker_inet_addr)) {
                return JK_TRUE;
            }
            jk_log(l, JK_LOG_ERROR, "ERROR: can't resolve tomcat address %s\n", host);
        }
        jk_log(l, JK_LOG_ERROR, "ERROR: invalid host and port %s %d\n", (( host==NULL ) ? "NULL" : host ), port);
    } else {
        jk_log(l, JK_LOG_ERROR, "In jk_worker_t::validate, NULL parameters\n");
    }

    return JK_FALSE;
}


int ajp_init(jk_worker_t *pThis,
             jk_map_t    *props,
             jk_worker_env_t *we,
             jk_logger_t *l,
             int          proto)
{
    int cache;
    
    /*
     * start the connection cache
     */
    jk_log(l, JK_LOG_DEBUG, "Into jk_worker_t::init\n");

    if (proto == AJP13_PROTO) {
        cache = AJP13_DEF_CACHE_SZ;
    }
    else if (proto == AJP14_PROTO) {
        cache = AJP13_DEF_CACHE_SZ;
    }
    else {
        jk_log(l, JK_LOG_DEBUG, "In jk_worker_t::init, unknown protocol %d\n", proto);
        return JK_FALSE;
    }

    if (pThis && pThis->worker_private) {
        ajp_worker_t *p = pThis->worker_private;
        int cache_sz = jk_get_worker_cache_size(props, p->name, cache);
        int socket_timeout = jk_get_worker_socket_timeout(props, p->name, AJP13_DEF_TIMEOUT);
        int socket_keepalive = jk_get_worker_socket_keepalive(props, p->name, JK_FALSE);
        int cache_timeout = jk_get_worker_cache_timeout(props, p->name, AJP_DEF_CACHE_TIMEOUT);

        jk_log(l, JK_LOG_DEBUG, "In jk_worker_t::init, setting socket timeout to %d\n", socket_timeout);
        p->socket_timeout = socket_timeout;
        p->keepalive = socket_keepalive;
        p->cache_timeout = cache_timeout;
        /* 
         *  Need to initialize secret here since we could return from inside
         *  of the following loop
         */
           
        p->secret = jk_get_worker_secret(props, p->name );
        p->ep_cache_sz = 0;
        p->ep_mincache_sz = 0;
        if (cache_sz > 0) {
            p->ep_cache = (ajp_endpoint_t **)malloc(sizeof(ajp_endpoint_t *) * cache_sz);
            if(p->ep_cache) {
                int i;
                p->ep_cache_sz = cache_sz;
                for(i = 0 ; i < cache_sz ; i++) {
                    p->ep_cache[i] = NULL;
                }
                JK_INIT_CS(&(p->cs), i);
                if (i) {
                    return JK_TRUE;
                }
            }
        }
    } else {
        jk_log(l, JK_LOG_ERROR, "In jk_worker_t::init, NULL parameters\n");
    }

    return JK_FALSE;
}

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

    if (pThis && *pThis && (*pThis)->worker_private) {
        ajp_worker_t *aw = (*pThis)->worker_private;

        free(aw->name);

        jk_log(l, JK_LOG_DEBUG, "Into jk_worker_t::destroy up to %d endpoint to close\n", aw->ep_cache_sz);

        if(aw->ep_cache_sz) {
            unsigned i;
            for(i = 0 ; i < aw->ep_cache_sz ; i++) {
                if(aw->ep_cache[i]) {
                    ajp_close_endpoint(aw->ep_cache[i], l);
                }
            }
            free(aw->ep_cache);
            JK_DELETE_CS(&(aw->cs), i);
        }

        if (aw->login) {
            free(aw->login);
            aw->login = NULL;
        }

        free(aw);
        return JK_TRUE;
    }

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

int JK_METHOD ajp_done(jk_endpoint_t **e,
                       jk_logger_t    *l)
{
    if (e && *e && (*e)->endpoint_private) {
        ajp_endpoint_t *p = (*e)->endpoint_private;
        int reuse_ep = p->reuse;

        ajp_reset_endpoint(p);

        if(reuse_ep) {
            ajp_worker_t *w = p->worker;
            if(w->ep_cache_sz) {
                int rc;
                JK_ENTER_CS(&w->cs, rc);
                if(rc) {
                    unsigned i;

                    for(i = 0 ; i < w->ep_cache_sz ; i++) {
                        if(!w->ep_cache[i]) {
                            w->ep_cache[i] = p;
                            break;
                        }
                    }
                    JK_LEAVE_CS(&w->cs, rc);
                    if(i < w->ep_cache_sz) {
                        jk_log(l, JK_LOG_DEBUG, "Into jk_endpoint_t::done, recycling connection\n");
                        return JK_TRUE;
                    }
                }
            }
        }
        jk_log(l, JK_LOG_DEBUG, "Into jk_endpoint_t::done, closing connection %d\n", reuse_ep);
        ajp_close_endpoint(p, l);
        *e = NULL;

        return JK_TRUE;
    }

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

int ajp_get_endpoint(jk_worker_t    *pThis,
                     jk_endpoint_t **je,
                     jk_logger_t    *l,
                     int             proto)
{
    jk_log(l, JK_LOG_DEBUG, "Into jk_worker_t::get_endpoint\n");

    if (pThis && pThis->worker_private && je) {
        ajp_worker_t   *aw = pThis->worker_private;
        ajp_endpoint_t *ae = NULL;

        if (aw->ep_cache_sz) {
            int rc;
            JK_ENTER_CS(&aw->cs, rc);
            if (rc) {
                unsigned i;
                time_t now;
                if (aw->socket_timeout > 0 || aw->cache_timeout) {
                    now = time(NULL);
                }
                for (i = 0 ; i < aw->ep_cache_sz ; i++) {
                    if (aw->ep_cache[i]) {
                        ae = aw->ep_cache[i];
                        aw->ep_cache[i] = NULL;
                        break;
                    }
                }
                /* Handle enpoint cache timeouts */
                if (aw->cache_timeout) {
                    for ( ; i < aw->ep_cache_sz ; i++) {
                        if (aw->ep_cache[i]) {
                            unsigned elapsed = (unsigned)(now - ae->last_access);
                            if (elapsed > aw->cache_timeout) {
                                jk_log(l, JK_LOG_DEBUG, 
                                    "In jk_endpoint_t::ajp_get_endpoint," \
                                    " cleaning cache slot = %d elapsed %d\n",
                                     i, elapsed);
                                ajp_close_endpoint(aw->ep_cache[i], l);
                                aw->ep_cache[i] = NULL;
                            }
                        }
                    }
                }
                JK_LEAVE_CS(&aw->cs, rc);
                if (ae) {
                    if (ae->sd > 0) {
                        /* Handle timeouts for open sockets */
                        unsigned elapsed = (unsigned)(now - ae->last_access);
                        ae->last_access = now;
                        jk_log(l, JK_LOG_DEBUG, "In jk_endpoint_t::ajp_get_endpoint, time elapsed since last request = %d seconds\n", elapsed);
                        if (aw->socket_timeout > 0 && elapsed > aw->socket_timeout) {
                            jk_close_socket(ae->sd);
                            jk_log(l, JK_LOG_DEBUG, 
                                   "In jk_endpoint_t::ajp_get_endpoint, reached socket timeout, closed sd = %d\n", ae->sd);
                            ae->sd = -1; /* just to avoid twice close */
                        }
                    }
                    *je = &ae->endpoint;
                    return JK_TRUE;
                }
            }
        }

        ae = (ajp_endpoint_t *)malloc(sizeof(ajp_endpoint_t));
        if (ae) {
            ae->sd = -1;
            ae->reuse = JK_FALSE;
            ae->last_access = time(NULL);
            jk_open_pool(&ae->pool, ae->buf, sizeof(ae->buf));
            ae->worker = pThis->worker_private;
            ae->endpoint.endpoint_private = ae;
            ae->proto = proto;
            ae->endpoint.service = ajp_service;
            ae->endpoint.done = ajp_done;
            *je = &ae->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;
}