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 16:31:55 GMT
remm        2005/06/23 09:31:55

  Modified:    jk/java/org/apache/coyote/ajp AjpAprProcessor.java
  Log:
  - Redo input, but somehow with no additional performance increase :(
  - My measurements seem to indicate that this is faster than regular AJP (which
    could probably use a more aggressive output buffering like I did), but with
    everything running on localhost, it's not exactly that great a benchmark.
    Poller only keepalives (firstReadTimeout="0" pollTime="2000") seems to be
    about as fast as regular AJP, which will mean minimum thread usage in Tomcat
    at no cost (provided previous performance levels were enough). Another
    fiding is that about 50% of the CPU usage is in Apache for some reason.
  - Hopefully this is reasonably bug free (don't expect that much though).
  
  Revision  Changes    Path
  1.7       +143 -87   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.6
  retrieving revision 1.7
  diff -u -r1.6 -r1.7
  --- AjpAprProcessor.java	23 Jun 2005 11:50:34 -0000	1.6
  +++ AjpAprProcessor.java	23 Jun 2005 16:31:55 -0000	1.7
  @@ -96,34 +96,9 @@
   
           // Allocate input and output buffers
           inputBuffer = ByteBuffer.allocateDirect(16 * 1024);
  +        inputBuffer.limit(0);
           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());
  -
       }
   
   
  @@ -333,23 +308,58 @@
       
       /**
        * Direct buffer used for sending right away a get body message.
  -     * FIXME: can probably be static
        */
  -    protected ByteBuffer getBodyMessageBuffer = null;
  +    protected static final ByteBuffer getBodyMessageBuffer;
       
       
       /**
        * Direct buffer used for sending right away a pong message.
  -     * FIXME: can probably be static
        */
  -    protected ByteBuffer pongMessageBuffer = null;
  +    protected static final ByteBuffer pongMessageBuffer;
       
       
       /**
        * End message array.
  -     * FIXME: can probably be static
        */
  -    protected byte[] endMessageArray = null;
  +    protected static final byte[] endMessageArray;
  +    
  +    
  +    // ----------------------------------------------------- Static Initializer
  +
  +
  +    static {
  +
  +        // 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());
  +
  +    }
       
       
       // ------------------------------------------------------------- Properties
  @@ -444,7 +454,8 @@
               // Parsing the request header
               try {
                   // Get first message of the request
  -                if (!readMessage(requestHeaderMessage, true)) {
  +                if (!readMessage(requestHeaderMessage, true, 
  +                        keptAlive && (endpoint.getCurrentThreadsBusy() > limit)))
{
                       // This means that no data is available right now
                       // (long keepalive), so that the processor should be recycled
                       // and the method should return true
  @@ -507,7 +518,7 @@
               // Finish the response if not done yet
               if (!finished) {
                   try {
  -                    endRequest();
  +                    finish();
                   } catch (Throwable t) {
                       error = true;
                   }
  @@ -591,7 +602,7 @@
               }
   
               try {
  -                flushOutputBuffer();
  +                flush();
               } catch (IOException e) {
                   // Set error flag
                   error = true;
  @@ -604,7 +615,7 @@
               // transactions with the client
   
               try {
  -                endRequest();
  +                finish();
               } catch (IOException e) {
                   // Set error flag
                   error = true;
  @@ -1129,7 +1140,7 @@
       /**
        * Finish AJP response.
        */
  -    protected void endRequest()
  +    protected void finish()
           throws IOException {
   
           if (!response.isCommitted()) {
  @@ -1147,11 +1158,75 @@
           
           finished = true;
           if (outputBuffer.position() + endMessageArray.length > outputBuffer.capacity())
{
  -            flushOutputBuffer();
  -        } else {
  -            outputBuffer.put(endMessageArray);
  +            flush();
           }
  -        flushOutputBuffer();
  +        outputBuffer.put(endMessageArray);
  +        flush();
  +    }
  +    
  +    
  +    /**
  +     * Read at least the specified amount of bytes, and place them 
  +     * in the input buffer.
  +     */
  +    protected boolean read(int n)
  +        throws IOException {
  +
  +        if (inputBuffer.capacity() - inputBuffer.limit() <= 
  +                n - inputBuffer.remaining()) {
  +            inputBuffer.compact();
  +            inputBuffer.limit(inputBuffer.position());
  +            inputBuffer.position(0);
  +        }
  +        while (inputBuffer.remaining() < n) {
  +            int nRead = Socket.recvb
  +                (socket, inputBuffer, inputBuffer.limit(), 
  +                        inputBuffer.capacity() - inputBuffer.limit());
  +            if (nRead > 0) {
  +                inputBuffer.limit(inputBuffer.limit() + nRead);
  +            } else {
  +                throw new IOException(sm.getString("iib.failedread"));
  +            }
  +        }
  +        
  +        return true;
  +        
  +    }
  +    
  +    
  +    /**
  +     * Read at least the specified amount of bytes, and place them 
  +     * in the input buffer.
  +     */
  +    protected boolean readt(int n, boolean useAvailableData)
  +        throws IOException {
  +        
  +        if (useAvailableData && inputBuffer.remaining() == 0) {
  +            return false;
  +        }
  +        if (inputBuffer.capacity() - inputBuffer.limit() <= 
  +                n - inputBuffer.remaining()) {
  +            inputBuffer.compact();
  +            inputBuffer.limit(inputBuffer.position());
  +            inputBuffer.position(0);
  +        }
  +        while (inputBuffer.remaining() < n) {
  +            int nRead = Socket.recvbt
  +                (socket, inputBuffer, inputBuffer.limit(), 
  +                        inputBuffer.capacity() - inputBuffer.limit(), readTimeout);
  +            if (nRead > 0) {
  +                inputBuffer.limit(inputBuffer.limit() + nRead);
  +            } else {
  +                if ((-nRead) == Status.ETIMEDOUT || (-nRead) == Status.TIMEUP) {
  +                    return false;
  +                } else {
  +                    throw new IOException(sm.getString("iib.failedread"));
  +                }
  +            }
  +        }
  +        
  +        return true;
  +        
       }
       
       
  @@ -1162,7 +1237,7 @@
       public boolean receive() throws IOException {
           first = false;
           bodyMessage.reset();
  -        readMessage(bodyMessage, false);
  +        readMessage(bodyMessage, false, false);
           if( log.isDebugEnabled() )
               log.info( "Receiving: getting request body chunk " + bodyMessage.getLen() );
           
  @@ -1173,7 +1248,6 @@
               return false;
           }
           int blen = bodyMessage.peekInt();
  -
           if( blen == 0 ) {
               return false;
           }
  @@ -1188,15 +1262,14 @@
           empty = false;
           return true;
       }
  -    
  +
       /**
        * Get more request body data from the web server and store it in the 
        * internal buffer.
        *
        * @return true if there is more data, false if not.    
        */
  -    private boolean refillReadBuffer() throws IOException 
  -    {
  +    private boolean refillReadBuffer() throws IOException {
           // If the server returns an empty packet, assume that that end of
           // the stream has been reached (yuck -- fix protocol??).
           // FORM support
  @@ -1230,52 +1303,30 @@
        *         didn't return anything
        * @throws IOException any other failure, including incomplete reads
        */
  -    protected boolean readMessage(AjpMessage message, boolean first)
  +    protected boolean readMessage(AjpMessage message, boolean first, 
  +            boolean useAvailableData)
           throws IOException {
           
           byte[] buf = message.getBuffer();
           int headerLength = message.getHeaderLength();
   
  -        // Read the message header
  -        // FIXME: better buffering to avoid doing two reads !!!!
           if (first) {
  -            int nRead = Socket.recvt
  -                (socket, buf, 0, headerLength, readTimeout);
  -            // Note: Unlike before, I assume it is not acceptable to do more than
  -            // one read for the first four bytes
  -            if (nRead == headerLength) {
  -                message.processHeader();
  -            } else {
  -                if ((-nRead) == Status.ETIMEDOUT || (-nRead) == Status.TIMEUP) {
  -                    return false;
  -                } else {
  -                    throw new IOException(sm.getString("iib.failedread"));
  -                }
  +            if (!readt(headerLength, useAvailableData)) {
  +                return false;
               }
           } else {
  -            int nRead = Socket.recv(socket, buf, 0, headerLength);
  -            if (nRead == headerLength) {
  -                message.processHeader();
  -            } else {
  -                throw new IOException(sm.getString("iib.failedread"));
  -            }
  +            read(headerLength);
           }
  +        inputBuffer.get(message.getBuffer(), 0, headerLength);
  +        message.processHeader();
  +        read(message.getLen());
  +        inputBuffer.get(message.getBuffer(), headerLength, message.getLen());
           
  -        // Read the full message body; incomplete reads is a protocol error
  -        int messageLength = message.getLen();
  -        int pos = headerLength;
  -        while (pos < headerLength + messageLength) {
  -            int nRead = Socket.recv(socket, buf, pos, headerLength + messageLength - pos);
  -            if (nRead > 0) {
  -                pos += nRead;
  -            } else {
  -                throw new IOException(sm.getString("iib.failedread"));
  -            }
  -        }
           return true;
  +        
       }
  -    
  -    
  +
  +
       /**
        * Recycle the processor.
        */
  @@ -1290,7 +1341,9 @@
           request.recycle();
           response.recycle();
           certificates.recycle();
  -        requestHeaderMessage.reset();
  +
  +        inputBuffer.clear();
  +        inputBuffer.limit(0);
           outputBuffer.clear();
   
       }
  @@ -1299,7 +1352,7 @@
       /**
        * Callback to write data from the buffer.
        */
  -    protected void flushOutputBuffer()
  +    protected void flush()
           throws IOException {
           if (outputBuffer.position() > 0) {
               if (Socket.sendb(socket, outputBuffer, 0, outputBuffer.position()) < 0)
{
  @@ -1369,10 +1422,13 @@
               throws IOException {
   
               if (!response.isCommitted()) {
  -                // Send the connector a request for commit. The connector should
  -                // then validate the headers, send them (using sendHeader) and 
  -                // set the filters accordingly.
  -                response.action(ActionCode.ACTION_COMMIT, null);
  +                // Validate and write response headers
  +                try {
  +                    prepareResponse();
  +                } catch (IOException e) {
  +                    // Set error flag
  +                    error = true;
  +                }
               }
   
               int len = chunk.getLength();
  @@ -1387,7 +1443,7 @@
                   len -= thisTime;
                   if (outputBuffer.position() + thisTime 
                           + bodyMessage.getHeaderLength() + 4 > outputBuffer.capacity())
{
  -                    flushOutputBuffer();
  +                    flush();
                   } else {
                       outputBuffer.put((byte) 0x41);
                       outputBuffer.put((byte) 0x42);
  
  
  

---------------------------------------------------------------------
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