jk_worker_ajp13.c

Tomcat 4.1与WebServer集成组件的源代码包.

                              jk_endpoint_t   *e )
{
    int err=JK_OK;
    int attempt;
    int has_post_body=JK_FALSE;
    jk_channel_t *channel= worker->channel;

    e->recoverable = JK_TRUE;
    s->is_recoverable_error = JK_TRUE;

    /*
     * Try to send the request on a valid endpoint. If one endpoint
     * fails, close the channel and try again ( maybe tomcat was restarted )
     * 
     * XXX JK_RETRIES could be replaced by the number of workers in
     * a load-balancing configuration 
     */
    for(attempt = 0 ; attempt < JK_RETRIES ;attempt++) {

        if( e->sd == -1 ) {
            err=jk2_worker_ajp13_connect(env, e);
            if( err!=JK_OK ) {
                env->l->jkLog(env, env->l, JK_LOG_ERROR,
                              "ajp13.service() failed to connect endpoint errno=%d %s\n",
                              errno, strerror( errno ));
                e->worker->in_error_state=JK_TRUE;
                return err;
            }
            if( worker->mbean->debug > 0 )
                env->l->jkLog(env, env->l, JK_LOG_DEBUG,
                              "ajp13.service() connecting to endpoint \n");
        }

        err=e->worker->channel->send( env, e->worker->channel, e,
                                      e->request );

        if( e->worker->mbean->debug > 10 )
            e->request->dump( env, e->request, "Sent" );
        
        if (err!=JK_OK ) {
            /* Can't send - bad endpoint, try again */
            env->l->jkLog(env, env->l, JK_LOG_ERROR,
                          "ajp13.service() error sending, reconnect %s %d %d %s\n",
                          e->worker->channelName, err, errno, strerror(errno));
            jk2_close_endpoint(env, e );
            continue;
        }

        /* We should have a channel now, send the post data */
        
        /* Prepare to send some post data ( ajp13 proto ). We do that after the
           request was sent ( we're receiving data from client, can be slow, no
           need to delay - we can do that in paralel. ( not very sure this is
           very usefull, and it brakes the protocol ) ! */

	/* || s->is_chunked - this can't be done here. The original protocol sends the first
	   chunk of post data ( based on Content-Length ), and that's what the java side expects.
	   Sending this data for chunked would break other ajp13 serers.

	   Note that chunking will continue to work - using the normal read.
	*/
        if (has_post_body  || s->left_bytes_to_send > 0) {
            /* We never sent any POST data and we check it we have to send at
             * least of block of data (max 8k). These data will be kept in reply
             * for resend if the remote Tomcat is down, a fact we will learn only
             * doing a read (not yet) 
             */
            if( attempt==0 )
                err=jk2_serialize_postHead( env, e->post, s, e );
            else
                err=JK_OK; /* We already have the initial body chunk */
            
            if( e->worker->mbean->debug > 10 )
                e->request->dump( env, e->request, "Post head" );
            
            if (err != JK_OK ) {
                /* the browser stop sending data, no need to recover */
                /* e->recoverable = JK_FALSE; */
                s->is_recoverable_error = JK_FALSE;
                env->l->jkLog(env, env->l, JK_LOG_ERROR,
                              "ajp13.service() Error receiving initial post %d %d %d\n", err, errno, attempt);
                return JK_ERR;
            }
            has_post_body=JK_TRUE;
            err= e->worker->channel->send( env, e->worker->channel, e,
                                           e->post );
            if( err != JK_OK ) {
                /* e->recoverable = JK_FALSE; */
                /*                 s->is_recoverable_error = JK_FALSE; */
                env->l->jkLog(env, env->l, JK_LOG_ERROR,
                              "ajp13.service() Error sending initial post %d %d %d\n", err, errno, attempt);
                jk2_close_endpoint(env, e);
                continue;
                /*  return JK_ERR; */
            }
        }

        
        err = e->worker->workerEnv->processCallbacks(env, e->worker->workerEnv,
                                                     e, s);
        
        /* 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 (err!=JK_OK && ! e->recoverable ) {
            s->is_recoverable_error = JK_FALSE;
            env->l->jkLog(env, env->l, JK_LOG_ERROR,
                          "ajp13.service() ajpGetReply unrecoverable error %d\n",
                          err);
            /* The connection is compromised, need to close it ! */
            e->worker->in_error_state = 1;
            return JK_ERR;
        }
        
        if( err != JK_OK ) {
            env->l->jkLog(env, env->l, JK_LOG_ERROR,
                          "ajp13.service() ajpGetReply recoverable error %d\n",
                          err);
            jk2_close_endpoint(env, e ); 
       }

        if( err==JK_OK )
            return err;
    }
    return err;
}

static int JK_METHOD
jk2_worker_ajp13_forwardSingleThread(jk_env_t *env, jk_worker_t *worker,
                                    jk_ws_service_t *s,
                                    jk_endpoint_t   *e )
{
    int err;

    if( e->worker->mbean->debug > 0 )
        env->l->jkLog(env, env->l, JK_LOG_DEBUG,
                      "ajp13.forwardST() Before calling native channel %s\n",
                      e->worker->channel->mbean->name);
    err=e->worker->channel->send( env, e->worker->channel, e,
                                  e->request );
    if( e->worker->mbean->debug > 0 )
        env->l->jkLog(env, env->l, JK_LOG_DEBUG,
                      "ajp13.forwardST() After %d\n",err);
    
    /* I assume no unrecoverable error can happen here - we're in a single thread,
       so things are simpler ( at least in this area ) */
    return err;
}
     
static int JK_METHOD
jk2_worker_ajp13_service1(jk_env_t *env, jk_worker_t *w,
                        jk_ws_service_t *s,
                        jk_endpoint_t   *e )
{
    int err;

    if( ( e== NULL ) || ( s == NULL ) ) {
	env->l->jkLog(env, env->l, JK_LOG_ERROR,
                      "ajp13.service() NullPointerException\n");
	return JK_ERR;
    }
    
    if( w->channel==NULL ) {
        env->l->jkLog(env, env->l, JK_LOG_INFO,
                      "ajp13.service() no channel defined, error in init\n", w->mbean->name);
        return JK_ERR;
    }
    
    e->currentRequest=s;

    /* XXX configurable ? */
    strncpy( e->stats->active, s->req_uri, 64);
    
    /* Prepare the messages we'll use.*/ 
    e->request->reset( env, e->request );
    e->reply->reset( env, e->reply );
    e->post->reset( env, e->post );
    
    s->is_recoverable_error = JK_TRUE;
    /* Up to now, we can recover */
    e->recoverable = JK_TRUE;

    s->left_bytes_to_send = s->content_length;
    s->content_read=0;

    /* 
     * We get here initial request (in reqmsg)
     */
    err=jk2_serialize_request13(env, e->request, s, e);
    if (err!=JK_OK) {
	s->is_recoverable_error = JK_FALSE;                
	env->l->jkLog(env, env->l, JK_LOG_ERROR,
                      "ajp13.service(): error marshaling\n");
	return JK_ERR;
    }

    if( w->mbean->debug > 0 ) 
        env->l->jkLog(env, env->l, JK_LOG_DEBUG,
                      "ajp13.service() %s\n", w->channelName);

    if( w->channel->beforeRequest != NULL ) {
        w->channel->beforeRequest( env, w->channel, w, e, s );
    }
    
    /* First message for this request */
    if( w->channel->is_stream == JK_TRUE ) {
        err=jk2_worker_ajp13_forwardStream( env, w, s, e );
    } else {
        err=jk2_worker_ajp13_forwardSingleThread( env, w, s, e );
    }
    if (err != JK_OK){
        env->l->jkLog(env, env->l, JK_LOG_ERROR,
              "ajp13.service() Error  forwarding %s %d %d\n", e->worker->mbean->name,
                      e->recoverable, e->worker->in_error_state);
    }

    if( w->mbean->debug > 0 ) 
        env->l->jkLog(env, env->l, JK_LOG_DEBUG,
                      "ajp13.service() done %s\n", e->worker->mbean->name);

    if( w->channel->afterRequest != NULL ) {
        w->channel->afterRequest( env, w->channel, w, e, s );
    }

    e->currentRequest=NULL;
    
    return err;
}




static int JK_METHOD
jk2_worker_ajp13_done(jk_env_t *env, jk_worker_t *we, jk_endpoint_t *e)
{
    jk_worker_t *w;
    int rc=JK_OK;
    
    w= e->worker;
    
    if( e->cPool != NULL ) 
        e->cPool->reset(env, e->cPool);

    if( w->endpointCache == NULL ) {
        env->l->jkLog(env, env->l, JK_LOG_ERROR, "ajp13.done() No pool\n");
        return JK_ERR;
    }
    
    if(  w->in_error_state ) {
        jk2_close_endpoint(env, e);
        /*     if( w->mbean->debug > 0 )  */
        env->l->jkLog(env, env->l, JK_LOG_INFO,
                      "ajp13.done() close endpoint %s error_state %d\n",
                      w->mbean->name, w->in_error_state );
    }
        
    rc=w->endpointCache->put( env, w->endpointCache, e );
    if( rc!=JK_OK ) {
        env->l->jkLog(env, env->l, JK_LOG_ERROR, "ajp13.done() Error recycling ep\n");
        return rc;        
    }
    if( w->mbean->debug > 0 ) 
        env->l->jkLog(env, env->l, JK_LOG_DEBUG,
                      "ajp13.done() return to pool %s\n",
                      w->mbean->name );
    return JK_OK;
}

static int JK_METHOD
jk2_worker_ajp13_getEndpoint(jk_env_t *env,
                            jk_worker_t *ajp13,
                            jk_endpoint_t **eP)
{
    jk_endpoint_t *e = NULL;
    jk_bean_t *jkb;
    int ret;
    
    if( ajp13->secret ==NULL ) {
    }

    if( ajp13->endpointCache == NULL ) {
        env->l->jkLog(env, env->l, JK_LOG_ERROR, "ajp13.getEndpoint() No pool\n");
        return JK_ERR;
    }

    e=ajp13->endpointCache->get( env, ajp13->endpointCache );

    if (e!=NULL) {
        *eP = e;
        return JK_OK;
    }

    if( ajp13->cs != NULL ) 
        ajp13->cs->lock( env, ajp13->cs );

    {
        if (ajp13->maxEndpoints && 
            ajp13->maxEndpoints <= ajp13->endpointMap->size(env, ajp13->endpointMap)) {