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: [EMAIL PROTECTED] For additional commands, e-mail: [EMAIL PROTECTED]