tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From r...@apache.org
Subject cvs commit: jakarta-tomcat-connectors/jk/java/org/apache/coyote/ajp AjpAprProcessor.java
Date Thu, 23 Jun 2005 11:50:34 GMT
remm        2005/06/23 04:50:34

  Modified:    jk/java/org/apache/coyote/ajp AjpAprProcessor.java
  Log:
  - Add buffering for output.
  - This seems to work fine, but I didn't do that much testing yet.
  - Not done yet: input buffering is still missing, and it will likely be more
    significant (at least when there's a lot of upload). It's also harder,
    unfortunately :(
  
  Revision  Changes    Path
  1.6       +179 -106  jakarta-tomcat-connectors/jk/java/org/apache/coyote/ajp/AjpAprProcessor.java
  
  Index: AjpAprProcessor.java
  ===================================================================
  RCS file: /home/cvs/jakarta-tomcat-connectors/jk/java/org/apache/coyote/ajp/AjpAprProcessor.java,v
  retrieving revision 1.5
  retrieving revision 1.6
  diff -u -r1.5 -r1.6
  --- AjpAprProcessor.java	17 Jun 2005 09:43:35 -0000	1.5
  +++ AjpAprProcessor.java	23 Jun 2005 11:50:34 -0000	1.6
  @@ -20,6 +20,7 @@
   import java.io.IOException;
   import java.io.InterruptedIOException;
   import java.net.InetAddress;
  +import java.nio.ByteBuffer;
   import java.security.cert.CertificateFactory;
   import java.security.cert.X509Certificate;
   
  @@ -84,11 +85,45 @@
           response.setOutputBuffer(new SocketOutputBuffer());
           request.setResponse(response);
   
  -        readTimeout = endpoint.getFirstReadTimeout() * 1000;
  +        if (endpoint.getFirstReadTimeout() > 0) {
  +            readTimeout = endpoint.getFirstReadTimeout() * 1000;
  +        } else {
  +            readTimeout = 100 * 1000;
  +        }
   
           // Cause loading of HexUtils
           int foo = HexUtils.DEC[0];
   
  +        // Allocate input and output buffers
  +        inputBuffer = ByteBuffer.allocateDirect(16 * 1024);
  +        outputBuffer = ByteBuffer.allocateDirect(16 * 1024);
  +
  +        // Set the get body message buffer
  +        AjpMessage getBodyMessage = new AjpMessage();
  +        getBodyMessage.reset();
  +        getBodyMessage.appendByte(Constants.JK_AJP13_GET_BODY_CHUNK);
  +        getBodyMessage.appendInt(Constants.MAX_READ_SIZE);
  +        getBodyMessage.end();
  +        getBodyMessageBuffer = ByteBuffer.allocateDirect(getBodyMessage.getLen());
  +        getBodyMessageBuffer.put(getBodyMessage.getBuffer(), 0, getBodyMessage.getLen());
  +
  +        // Set the read body message buffer
  +        AjpMessage pongMessage = new AjpMessage();
  +        pongMessage.reset();
  +        pongMessage.appendByte(Constants.JK_AJP13_CPONG_REPLY);
  +        pongMessage.end();
  +        pongMessageBuffer = ByteBuffer.allocateDirect(pongMessage.getLen());
  +        pongMessageBuffer.put(pongMessage.getBuffer(), 0, pongMessage.getLen());
  +
  +        // Allocate the end message array
  +        AjpMessage endMessage = new AjpMessage();
  +        endMessage.reset();
  +        endMessage.appendByte(Constants.JK_AJP13_END_RESPONSE);
  +        endMessage.appendByte(1);
  +        endMessage.end();
  +        endMessageArray = new byte[endMessage.getLen()];
  +        System.arraycopy(endMessage.getBuffer(), 0, endMessageArray, 0, endMessage.getLen());
  +
       }
   
   
  @@ -118,13 +153,13 @@
        * processing of the first message of a "request", so it might not be a request
        * header. It will stay unchanged during the processing of the whole request. 
        */
  -    protected AjpMessage headerMessage = new AjpMessage(); 
  +    protected AjpMessage requestHeaderMessage = new AjpMessage(); 
   
   
       /**
  -     * Message used for output. 
  +     * Message used for response header composition. 
        */
  -    protected AjpMessage outputMessage = new AjpMessage(); 
  +    protected AjpMessage responseHeaderMessage = new AjpMessage(); 
   
   
       /**
  @@ -146,18 +181,6 @@
   
       
       /**
  -     * All purpose response message.
  -     */
  -    protected AjpMessage responseMessage = new AjpMessage();
  -    
  -    
  -    /**
  -     * Read body message.
  -     */
  -    protected AjpMessage readBodyMessage = new AjpMessage();
  -
  -    
  -    /**
        * State flag.
        */
       protected boolean started = false;
  @@ -296,6 +319,39 @@
       protected boolean finished = false;
       
       
  +    /**
  +     * Direct buffer used for output.
  +     */
  +    protected ByteBuffer outputBuffer = null;
  +    
  +    
  +    /**
  +     * Direct buffer used for input.
  +     */
  +    protected ByteBuffer inputBuffer = null;
  +    
  +    
  +    /**
  +     * Direct buffer used for sending right away a get body message.
  +     * FIXME: can probably be static
  +     */
  +    protected ByteBuffer getBodyMessageBuffer = null;
  +    
  +    
  +    /**
  +     * Direct buffer used for sending right away a pong message.
  +     * FIXME: can probably be static
  +     */
  +    protected ByteBuffer pongMessageBuffer = null;
  +    
  +    
  +    /**
  +     * End message array.
  +     * FIXME: can probably be static
  +     */
  +    protected byte[] endMessageArray = null;
  +    
  +    
       // ------------------------------------------------------------- Properties
   
   
  @@ -375,14 +431,20 @@
   
           long soTimeout = endpoint.getSoTimeout();
   
  +        int limit = 0;
  +        if (endpoint.getFirstReadTimeout() > 0) {
  +            limit = endpoint.getMaxThreads() / 2;
  +        }
  +
           boolean openSocket = true;
  +        boolean keptAlive = false;
   
           while (started && !error) {
   
               // Parsing the request header
               try {
                   // Get first message of the request
  -                if (!readMessage(headerMessage, true)) {
  +                if (!readMessage(requestHeaderMessage, true)) {
                       // This means that no data is available right now
                       // (long keepalive), so that the processor should be recycled
                       // and the method should return true
  @@ -391,15 +453,17 @@
                   }
                   // Check message type, process right away and break if 
                   // not regular request processing
  -                int type = headerMessage.getByte();
  +                int type = requestHeaderMessage.getByte();
                   // FIXME: Any other types which should be checked ?
                   if (type == Constants.JK_AJP13_CPING_REQUEST) {
  -                    headerMessage.reset();
  -                    headerMessage.appendByte(Constants.JK_AJP13_CPONG_REPLY);
  -                    writeMessage(headerMessage);
  +                    if (Socket.sendb(socket, pongMessageBuffer, 0, 
  +                            pongMessageBuffer.position()) < 0) {
  +                        error = true;
  +                    }
                       continue;
                   }
   
  +                keptAlive = true;
                   request.setStartTime(System.currentTimeMillis());
               } catch (IOException e) {
                   error = true;
  @@ -515,12 +579,19 @@
   
           } else if (actionCode == ActionCode.ACTION_CLIENT_FLUSH) {
   
  -            if (response.isCommitted())
  -                return;
  +            if (!response.isCommitted()) {
  +                // Validate and write response headers
  +                try {
  +                    prepareResponse();
  +                } catch (IOException e) {
  +                    // Set error flag
  +                    error = true;
  +                    return;
  +                }
  +            }
   
  -            // Validate and write response headers
               try {
  -                prepareResponse();
  +                flushOutputBuffer();
               } catch (IOException e) {
                   // Set error flag
                   error = true;
  @@ -721,21 +792,21 @@
       protected void prepareRequest() {
   
           // Translate the HTTP method code to a String.
  -        byte methodCode = headerMessage.getByte();
  +        byte methodCode = requestHeaderMessage.getByte();
           if (methodCode != Constants.SC_M_JK_STORED) {
               String methodName = Constants.methodTransArray[(int)methodCode - 1];
               request.method().setString(methodName);
           }
   
  -        headerMessage.getBytes(request.protocol()); 
  -        headerMessage.getBytes(request.requestURI());
  +        requestHeaderMessage.getBytes(request.protocol()); 
  +        requestHeaderMessage.getBytes(request.requestURI());
   
  -        headerMessage.getBytes(request.remoteAddr());
  -        headerMessage.getBytes(request.remoteHost());
  -        headerMessage.getBytes(request.localName());
  -        request.setLocalPort(headerMessage.getInt());
  +        requestHeaderMessage.getBytes(request.remoteAddr());
  +        requestHeaderMessage.getBytes(request.remoteHost());
  +        requestHeaderMessage.getBytes(request.localName());
  +        request.setLocalPort(requestHeaderMessage.getInt());
   
  -        boolean isSSL = headerMessage.getByte() != 0;
  +        boolean isSSL = requestHeaderMessage.getByte() != 0;
           if (isSSL) {
               // XXX req.setSecure( true );
               request.scheme().setString("https");
  @@ -744,20 +815,20 @@
           // Decode headers
           MimeHeaders headers = request.getMimeHeaders();
   
  -        int hCount = headerMessage.getInt();
  +        int hCount = requestHeaderMessage.getInt();
           for(int i = 0 ; i < hCount ; i++) {
               String hName = null;
               
               // Header names are encoded as either an integer code starting
               // with 0xA0, or as a normal string (in which case the first
               // two bytes are the length).
  -            int isc = headerMessage.peekInt();
  +            int isc = requestHeaderMessage.peekInt();
               int hId = isc & 0xFF;
               
               MessageBytes vMB = null;
               isc &= 0xFF00;
               if(0xA000 == isc) {
  -                headerMessage.getInt(); // To advance the read position
  +                requestHeaderMessage.getInt(); // To advance the read position
                   hName = Constants.headerTransArray[hId - 1];
                   vMB = headers.addValue( hName );
               } else {
  @@ -768,13 +839,13 @@
                   // SC_REQ_CONTENT_LENGTH=8 - leading to unexpected
                   // behaviour.  see bug 5861 for more information.
                   hId = -1;
  -                headerMessage.getBytes(tmpMB);
  +                requestHeaderMessage.getBytes(tmpMB);
                   ByteChunk bc = tmpMB.getByteChunk();
                   vMB = headers.addValue(bc.getBuffer(),
                           bc.getStart(), bc.getLength());
               }
               
  -            headerMessage.getBytes(vMB);
  +            requestHeaderMessage.getBytes(vMB);
               
               if (hId == Constants.SC_REQ_CONTENT_LENGTH ||
                       (hId == -1 && tmpMB.equalsIgnoreCase("Content-Length"))) {
  @@ -794,7 +865,7 @@
           boolean moreAttr = true;
   
           while (moreAttr) {
  -            byte attributeCode = headerMessage.getByte();
  +            byte attributeCode = requestHeaderMessage.getByte();
               if (attributeCode == Constants.SC_A_ARE_DONE)
                   break;
   
  @@ -803,15 +874,15 @@
               if (attributeCode == Constants.SC_A_SSL_KEY_SIZE) {
                   // Bug 1326: it's an Integer.
                   request.setAttribute(SSLSupport.KEY_SIZE_KEY,
  -                                 new Integer(headerMessage.getInt()));
  +                                 new Integer(requestHeaderMessage.getInt()));
                  //Integer.toString(msg.getInt()));
               }
   
               if (attributeCode == Constants.SC_A_REQ_ATTRIBUTE ) {
                   // 2 strings ???...
  -                headerMessage.getBytes(tmpMB);
  +                requestHeaderMessage.getBytes(tmpMB);
                   String n = tmpMB.toString();
  -                headerMessage.getBytes(tmpMB);
  +                requestHeaderMessage.getBytes(tmpMB);
                   String v = tmpMB.toString();
                   request.setAttribute(n, v);
                   if (log.isTraceEnabled())
  @@ -821,63 +892,63 @@
               // 1 string attributes
               switch (attributeCode) {
               case Constants.SC_A_CONTEXT :
  -                headerMessage.getBytes(tmpMB);
  +                requestHeaderMessage.getBytes(tmpMB);
                   // nothing
                   break;
                   
               case Constants.SC_A_SERVLET_PATH :
  -                headerMessage.getBytes(tmpMB);
  +                requestHeaderMessage.getBytes(tmpMB);
                   // nothing 
                   break;
                   
               case Constants.SC_A_REMOTE_USER :
                   if (tomcatAuthentication) {
                       // ignore server
  -                    headerMessage.getBytes(tmpMB);
  +                    requestHeaderMessage.getBytes(tmpMB);
                   } else {
  -                    headerMessage.getBytes(request.getRemoteUser());
  +                    requestHeaderMessage.getBytes(request.getRemoteUser());
                   }
                   break;
                   
               case Constants.SC_A_AUTH_TYPE :
                   if (tomcatAuthentication) {
                       // ignore server
  -                    headerMessage.getBytes(tmpMB);
  +                    requestHeaderMessage.getBytes(tmpMB);
                   } else {
  -                    headerMessage.getBytes(request.getAuthType());
  +                    requestHeaderMessage.getBytes(request.getAuthType());
                   }
                   break;
                   
               case Constants.SC_A_QUERY_STRING :
  -                headerMessage.getBytes(request.queryString());
  +                requestHeaderMessage.getBytes(request.queryString());
                   break;
                   
               case Constants.SC_A_JVM_ROUTE :
  -                headerMessage.getBytes(request.instanceId());
  +                requestHeaderMessage.getBytes(request.instanceId());
                   break;
                   
               case Constants.SC_A_SSL_CERT :
                   request.scheme().setString("https");
                   // SSL certificate extraction is costy, moved to JkCoyoteHandler
  -                headerMessage.getBytes(certificates);
  +                requestHeaderMessage.getBytes(certificates);
                   break;
                   
               case Constants.SC_A_SSL_CIPHER   :
                   request.scheme().setString( "https" );
  -                headerMessage.getBytes(tmpMB);
  +                requestHeaderMessage.getBytes(tmpMB);
                   request.setAttribute(SSLSupport.CIPHER_SUITE_KEY,
                                    tmpMB.toString());
                   break;
                   
               case Constants.SC_A_SSL_SESSION  :
                   request.scheme().setString( "https" );
  -                headerMessage.getBytes(tmpMB);
  +                requestHeaderMessage.getBytes(tmpMB);
                   request.setAttribute(SSLSupport.SESSION_ID_KEY, 
                                     tmpMB.toString());
                   break;
                   
               case Constants.SC_A_SECRET  :
  -                headerMessage.getBytes(tmpMB);
  +                requestHeaderMessage.getBytes(tmpMB);
                   String secret = tmpMB.toString();
                   if(log.isInfoEnabled())
                       log.info("Secret: " + secret);
  @@ -886,7 +957,7 @@
                   break;
                   
               case Constants.SC_A_STORED_METHOD:
  -                headerMessage.getBytes(request.method()); 
  +                requestHeaderMessage.getBytes(request.method()); 
                   break;
                   
               default:
  @@ -1013,9 +1084,9 @@
   
           response.setCommitted(true);
           
  -        outputMessage.reset();
  -        outputMessage.appendByte(Constants.JK_AJP13_SEND_HEADERS);
  -        outputMessage.appendInt(response.getStatus());
  +        responseHeaderMessage.reset();
  +        responseHeaderMessage.appendByte(Constants.JK_AJP13_SEND_HEADERS);
  +        responseHeaderMessage.appendInt(response.getStatus());
           
           String message = response.getMessage();
           if (message == null){
  @@ -1024,7 +1095,7 @@
               message = message.replace('\n', ' ').replace('\r', ' ');
           }
           tmpMB.setString(message);
  -        outputMessage.appendBytes(tmpMB);
  +        responseHeaderMessage.appendBytes(tmpMB);
   
           // XXX add headers
           
  @@ -1042,14 +1113,15 @@
               headers.setValue("Content-Length").setInt(contentLength);
           }
           int numHeaders = headers.size();
  -        outputMessage.appendInt(numHeaders);
  +        responseHeaderMessage.appendInt(numHeaders);
           for (int i = 0; i < numHeaders; i++) {
               MessageBytes hN = headers.getName(i);
  -            outputMessage.appendBytes(hN);
  +            responseHeaderMessage.appendBytes(hN);
               MessageBytes hV=headers.getValue(i);
  -            outputMessage.appendBytes(hV);
  +            responseHeaderMessage.appendBytes(hV);
           }
  -        writeMessage(outputMessage);
  +        responseHeaderMessage.end();
  +        outputBuffer.put(responseHeaderMessage.getBuffer(), 0, responseHeaderMessage.getLen());
   
       }
   
  @@ -1074,10 +1146,12 @@
               return;
           
           finished = true;
  -        outputMessage.reset();
  -        outputMessage.appendByte(Constants.JK_AJP13_END_RESPONSE);
  -        outputMessage.appendByte(1);
  -        writeMessage(outputMessage);
  +        if (outputBuffer.position() + endMessageArray.length > outputBuffer.capacity())
{
  +            flushOutputBuffer();
  +        } else {
  +            outputBuffer.put(endMessageArray);
  +        }
  +        flushOutputBuffer();
       }
       
       
  @@ -1135,16 +1209,11 @@
               return false;
           }
   
  -        // Why not use outBuf??
  -        readBodyMessage.reset();
  -        readBodyMessage.appendByte(Constants.JK_AJP13_GET_BODY_CHUNK);
  -        readBodyMessage.appendInt(Constants.MAX_READ_SIZE);
  -        writeMessage(readBodyMessage);
  -
  -        // In JNI mode, response will be in bodyMsg. In TCP mode, response need to be
  -        // read
  +        // Request more data immediately
  +        Socket.sendb(socket, getBodyMessageBuffer, 0, 
  +                getBodyMessageBuffer.position());
   
  -        boolean moreData=receive();
  +        boolean moreData = receive();
           if( !moreData ) {
               endOfStream = true;
           }
  @@ -1208,20 +1277,6 @@
       
       
       /**
  -     * Send the specified AJP message.
  -     *   
  -     * @param message to send
  -     * @throws IOException IO error when writing the message
  -     */
  -    protected void writeMessage(AjpMessage message) 
  -        throws IOException {
  -        message.end();
  -        if (Socket.send(socket, message.getBuffer(), 0, message.getLen()) < 0)
  -            throw new IOException(sm.getString("iib.failedwrite"));
  -    }
  -
  -
  -    /**
        * Recycle the processor.
        */
       public void recycle() {
  @@ -1235,11 +1290,26 @@
           request.recycle();
           response.recycle();
           certificates.recycle();
  -        headerMessage.reset();
  +        requestHeaderMessage.reset();
  +        outputBuffer.clear();
   
       }
   
   
  +    /**
  +     * Callback to write data from the buffer.
  +     */
  +    protected void flushOutputBuffer()
  +        throws IOException {
  +        if (outputBuffer.position() > 0) {
  +            if (Socket.sendb(socket, outputBuffer, 0, outputBuffer.position()) < 0)
{
  +                throw new IOException(sm.getString("iib.failedwrite"));
  +            }
  +            outputBuffer.clear();
  +        }
  +    }
  +
  +
       // ------------------------------------- InputStreamInputBuffer Inner Class
   
   
  @@ -1306,25 +1376,28 @@
               }
   
               int len = chunk.getLength();
  -            byte buf[] = bodyMessage.getBuffer();
               // 4 - hardcoded, byte[] marshalling overhead 
  -            int chunkSize=buf.length - bodyMessage.getHeaderLength() - 4;
  -            int off=0;
  -            while( len > 0 ) {
  -                int thisTime=len;
  -                if( thisTime > chunkSize ) {
  -                    thisTime=chunkSize;
  +            int chunkSize = 8*1024 - 4 - 4;
  +            int off = 0;
  +            while (len > 0) {
  +                int thisTime = len;
  +                if (thisTime > chunkSize) {
  +                    thisTime = chunkSize;
                   }
                   len -= thisTime;
  -                
  -                // FIXME: Don't use a temp buffer
  -                bodyMessage.reset();
  -                bodyMessage.appendByte( AjpConstants.JK_AJP13_SEND_BODY_CHUNK);
  -                if (log.isTraceEnabled()) 
  -                    log.trace("doWrite " + off + " " + thisTime + " " + len);
  -                bodyMessage.appendBytes(chunk.getBytes(), chunk.getOffset() + off, thisTime);
  +                if (outputBuffer.position() + thisTime 
  +                        + bodyMessage.getHeaderLength() + 4 > outputBuffer.capacity())
{
  +                    flushOutputBuffer();
  +                } else {
  +                    outputBuffer.put((byte) 0x41);
  +                    outputBuffer.put((byte) 0x42);
  +                    outputBuffer.putShort((short) (thisTime + 4));
  +                    outputBuffer.put(AjpConstants.JK_AJP13_SEND_BODY_CHUNK);
  +                    outputBuffer.putShort((short) chunk.getLength());
  +                    outputBuffer.put(chunk.getBytes(), chunk.getOffset() + off, thisTime);
  +                    outputBuffer.put((byte) 0x00);
  +                }
                   off += thisTime;
  -                writeMessage(bodyMessage);
               }
               
               return chunk.getLength();
  
  
  

---------------------------------------------------------------------
To unsubscribe, e-mail: tomcat-dev-unsubscribe@jakarta.apache.org
For additional commands, e-mail: tomcat-dev-help@jakarta.apache.org


Mime
View raw message