Author: remy.maucherat(a)jboss.com Date: 2013-10-15 13:30:15 -0400 (Tue, 15 Oct 2013) New Revision: 2279 Modified: branches/7.4.x/src/main/java/org/apache/catalina/core/StandardWrapperValve.java branches/7.4.x/src/main/java/org/apache/coyote/http11/AbstractInternalOutputBuffer.java branches/7.4.x/src/main/java/org/apache/coyote/http11/Constants.java branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11NioProcessor.java branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11NioProtocol.java branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalNioInputBuffer.java branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalNioOutputBuffer.java Log: - Refactor the non blocking mode of the NIO2 connector. - Some debug code remains. Modified: branches/7.4.x/src/main/java/org/apache/catalina/core/StandardWrapperValve.java =================================================================== --- branches/7.4.x/src/main/java/org/apache/catalina/core/StandardWrapperValve.java 2013-10-14 13:50:24 UTC (rev 2278) +++ branches/7.4.x/src/main/java/org/apache/catalina/core/StandardWrapperValve.java 2013-10-15 17:30:15 UTC (rev 2279) @@ -21,6 +21,7 @@ import static org.jboss.web.CatalinaMessages.MESSAGES; +import java.io.EOFException; import java.io.IOException; import java.util.Iterator; @@ -545,7 +546,7 @@ if (error) { Throwable throwable = asyncContext.getError(); if (throwable == null) { - throwable = new Exception(); + throwable = new EOFException(); } if (request.getReadListener() != null) { request.getReadListener().onError(throwable); Modified: branches/7.4.x/src/main/java/org/apache/coyote/http11/AbstractInternalOutputBuffer.java =================================================================== --- branches/7.4.x/src/main/java/org/apache/coyote/http11/AbstractInternalOutputBuffer.java 2013-10-14 13:50:24 UTC (rev 2278) +++ branches/7.4.x/src/main/java/org/apache/coyote/http11/AbstractInternalOutputBuffer.java 2013-10-15 17:30:15 UTC (rev 2279) @@ -41,76 +41,6 @@ public abstract class AbstractInternalOutputBuffer implements OutputBuffer { /** - * Associated Coyote response. - */ - protected Response response; - - /** - * Headers of the associated request. - */ - protected MimeHeaders headers; - - /** - * Committed flag. - */ - protected boolean committed; - - /** - * Finished flag. - */ - protected boolean finished; - - /** - * Pointer to the current write buffer. - */ - protected byte[] buf; - - /** - * Position in the buffer. - */ - protected int pos; - - /** - * Underlying output buffer. - */ - protected OutputBuffer outputBuffer; - - /** - * Filter library. Note: Filter[0] is always the "chunked" filter. - */ - protected OutputFilter[] filterLibrary; - - /** - * Active filter (which is actually the top of the pipeline). - */ - protected OutputFilter[] activeFilters; - - /** - * Index of the last active filter. - */ - protected int lastActiveFilter; - - /** - * Direct byte buffer used for writing. - */ - protected ByteBuffer bbuf = null; - - /** - * Leftover bytes which could not be written during a non blocking write. - */ - protected ByteChunk leftover = null; - - /** - * Non blocking mode. - */ - protected boolean nonBlocking = false; - - /** - * Write timeout - */ - protected int writeTimeout = -1; - - /** * Create a new instance of {@code AbstractInternalOutputBuffer} * * @param response @@ -118,553 +48,6 @@ */ public AbstractInternalOutputBuffer(Response response, int headerBufferSize) { - this.response = response; - this.headers = response.getMimeHeaders(); - buf = new byte[headerBufferSize]; - if (headerBufferSize < Constants.DEFAULT_HTTP_HEADER_BUFFER_SIZE) { - bbuf = ByteBuffer.allocateDirect(6 * 1500); - } else { - bbuf = ByteBuffer.allocateDirect((headerBufferSize / 1500 + 1) * 1500); - } - - outputBuffer = new OutputBufferImpl(); - filterLibrary = new OutputFilter[0]; - activeFilters = new OutputFilter[0]; - lastActiveFilter = -1; - - committed = false; - finished = false; - - leftover = new ByteChunk(); - nonBlocking = false; - - // Cause loading of HttpMessages - HttpMessages.getMessage(200); } - /** - * Initialize the internal output buffer - */ - protected abstract void init(); - - /** - * Set the non blocking flag. - * - * @param nonBlocking - */ - public void setNonBlocking(boolean nonBlocking) { - this.nonBlocking = nonBlocking; - } - - /** - * Get the non blocking flag value. - * - * @return non blocking - */ - public boolean getNonBlocking() { - return nonBlocking; - } - - /** - * Add an output filter to the filter library. - * - * @param filter - */ - public void addFilter(OutputFilter filter) { - - OutputFilter[] newFilterLibrary = new OutputFilter[filterLibrary.length + 1]; - for (int i = 0; i < filterLibrary.length; i++) { - newFilterLibrary[i] = filterLibrary[i]; - } - newFilterLibrary[filterLibrary.length] = filter; - filterLibrary = newFilterLibrary; - activeFilters = new OutputFilter[filterLibrary.length]; - } - - /** - * Get filters. - * - * @return the list of filters - */ - public OutputFilter[] getFilters() { - return filterLibrary; - } - - /** - * Clear filters. - */ - public void clearFilters() { - filterLibrary = new OutputFilter[0]; - lastActiveFilter = -1; - } - - /** - * Perform a write operation. The operation may be blocking or non-blocking - * depending on the value of {@code nonBlocking} flag. - * - * @param timeout - * a timeout for the operation - * @param unit - * The time unit of the timeout - * @return - */ - protected abstract int write(final long timeout, final TimeUnit unit); - - /** - * Add an output filter to the filter library. - * - * @param filter - */ - public void addActiveFilter(OutputFilter filter) { - - if (lastActiveFilter == -1) { - filter.setBuffer(outputBuffer); - } else { - for (int i = 0; i <= lastActiveFilter; i++) { - if (activeFilters[i] == filter) - return; - } - filter.setBuffer(activeFilters[lastActiveFilter]); - } - - activeFilters[++lastActiveFilter] = filter; - filter.setResponse(response); - } - - public void removeActiveFilters() { - // Recycle filters - for (int i = 0; i <= lastActiveFilter; i++) { - activeFilters[i].recycle(); - } - lastActiveFilter = -1; - } - - /** - * Flush the response. - * - * @throws IOException - * an undelying I/O error occured - */ - public void flush() throws IOException { - if (!committed) { - - // 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); - } - - // Flush the current buffer - flushBuffer(); - } - - /** - * - */ - protected void clearBuffer() { - synchronized (this.bbuf) { - this.bbuf.clear(); - } - } - - /** - * Recycle this object - */ - public void recycle() { - // Recycle Request object - response.recycle(); - this.clearBuffer(); - pos = 0; - lastActiveFilter = -1; - committed = false; - finished = false; - } - - /** - * End processing of current HTTP request. Note: All bytes of the current - * request should have been already consumed. This method only resets all - * the pointers so that we are ready to parse the next HTTP request. - */ - public void nextRequest() { - // Recycle Request object - response.recycle(); - - // Recycle filters - for (int i = 0; i <= lastActiveFilter; i++) { - activeFilters[i].recycle(); - } - - // Reset pointers - leftover.recycle(); - pos = 0; - lastActiveFilter = -1; - committed = false; - finished = false; - nonBlocking = false; - } - - /** - * End request. - * - * @throws IOException - * an undelying I/O error occured - */ - public void endRequest() throws IOException { - - if (!committed) { - // 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); - } - - if (finished) { - return; - } - - if (lastActiveFilter != -1) { - activeFilters[lastActiveFilter].end(); - } - - flushBuffer(); - finished = true; - } - - // ------------------------------------------------ HTTP/1.1 Output Methods - - /** - * Send an acknowledgment. - * - * @throws Exception - */ - public abstract void sendAck() throws Exception; - - /** - * Send the response status line. - */ - public void sendStatus() { - - // Write protocol name - write(Constants.HTTP_11_BYTES); - buf[pos++] = Constants.SP; - - // Write status code - int status = response.getStatus(); - switch (status) { - case 200: - write(Constants._200_BYTES); - break; - case 400: - write(Constants._400_BYTES); - break; - case 404: - write(Constants._404_BYTES); - break; - default: - write(status); - } - - buf[pos++] = Constants.SP; - - // Write message - String message = null; - if (org.apache.coyote.Constants.USE_CUSTOM_STATUS_MSG_IN_HEADER) { - message = response.getMessage(); - } - if (message == null) { - write(HttpMessages.getMessage(status)); - } else { - write(message.replace('\n', ' ').replace('\r', ' ')); - } - - // End the response status line - buf[pos++] = Constants.CR; - buf[pos++] = Constants.LF; - } - - /** - * Send a header. - * - * @param name - * Header name - * @param value - * Header value - */ - public void sendHeader(MessageBytes name, MessageBytes value) { - if (name.getLength() > 0 && !value.isNull()) { - write(name); - buf[pos++] = Constants.COLON; - buf[pos++] = Constants.SP; - write(value); - buf[pos++] = Constants.CR; - buf[pos++] = Constants.LF; - } - } - - /** - * Send a header. - * - * @param name - * Header name - * @param value - * Header value - */ - public void sendHeader(ByteChunk name, ByteChunk value) { - write(name); - buf[pos++] = Constants.COLON; - buf[pos++] = Constants.SP; - write(value); - buf[pos++] = Constants.CR; - buf[pos++] = Constants.LF; - } - - /** - * Send a header. - * - * @param name - * Header name - * @param value - * Header value - */ - public void sendHeader(String name, String value) { - write(name); - buf[pos++] = Constants.COLON; - buf[pos++] = Constants.SP; - write(value); - buf[pos++] = Constants.CR; - buf[pos++] = Constants.LF; - } - - /** - * End the header block. - */ - public void endHeaders() { - buf[pos++] = Constants.CR; - buf[pos++] = Constants.LF; - } - - /** - * Write the contents of a byte chunk. - * - * @param chunk - * byte chunk - * @return number of bytes written - * @throws IOException - * an undelying I/O error occured - */ - public abstract int doWrite(ByteChunk chunk, Response res) throws IOException; - - /** - * Commit the response. - * - * @throws IOException - * an undelying I/O error occured - */ - protected void commit() throws IOException { - - // The response is now committed - committed = true; - response.setCommitted(true); - - if (pos > 0) { - // Sending the response header buffer - bbuf.put(buf, 0, pos); - } - } - - /** - * This method will write the contents of the specyfied message bytes buffer - * to the output stream, without filtering. This method is meant to be used - * to write the response header. - * - * @param mb - * data to be written - */ - protected void write(MessageBytes mb) { - if (mb == null) { - return; - } - - switch (mb.getType()) { - case MessageBytes.T_BYTES: - write(mb.getByteChunk()); - break; - case MessageBytes.T_CHARS: - write(mb.getCharChunk()); - break; - default: - write(mb.toString()); - break; - } - } - - /** - * This method will write the contents of the specyfied message bytes buffer - * to the output stream, without filtering. This method is meant to be used - * to write the response header. - * - * @param bc - * data to be written - */ - protected void write(ByteChunk bc) { - // Writing the byte chunk to the output buffer - int length = bc.getLength(); - System.arraycopy(bc.getBytes(), bc.getStart(), buf, pos, length); - pos = pos + length; - } - - /** - * This method will write the contents of the specyfied char buffer to the - * output stream, without filtering. This method is meant to be used to - * write the response header. - * - * @param cc - * data to be written - */ - protected void write(CharChunk cc) { - int start = cc.getStart(); - int end = cc.getEnd(); - char[] cbuf = cc.getBuffer(); - for (int i = start; i < end; i++) { - char c = cbuf[i]; - // Note: This is clearly incorrect for many strings, - // but is the only consistent approach within the current - // servlet framework. It must suffice until servlet output - // streams properly encode their output. - if (((c <= 31) && (c != 9)) || c == 127 || c > 255) { - c = ' '; - } - buf[pos++] = (byte) c; - } - - } - - /** - * This method will write the contents of the specyfied byte buffer to the - * output stream, without filtering. This method is meant to be used to - * write the response header. - * - * @param b - * data to be written - */ - public void write(byte[] b) { - // Writing the byte chunk to the output buffer - System.arraycopy(b, 0, buf, pos, b.length); - pos = pos + b.length; - } - - /** - * This method will write the contents of the specyfied String to the output - * stream, without filtering. This method is meant to be used to write the - * response header. - * - * @param s - * data to be written - */ - protected void write(String s) { - if (s == null) { - return; - } - - // From the Tomcat 3.3 HTTP/1.0 connector - int len = s.length(); - for (int i = 0; i < len; i++) { - char c = s.charAt(i); - // Note: This is clearly incorrect for many strings, - // but is the only consistent approach within the current - // servlet framework. It must suffice until servlet output - // streams properly encode their output. - if (((c <= 31) && (c != 9)) || c == 127 || c > 255) { - c = ' '; - } - - buf[pos++] = (byte) c; - } - } - - /** - * This method will print the specified integer to the output stream, - * without filtering. This method is meant to be used to write the response - * header. - * - * @param i - * data to be written - */ - protected void write(int i) { - write(String.valueOf(i)); - } - - /** - * Callback to write data from the buffer. - */ - protected abstract void flushBuffer() throws IOException; - - /** - * Flush leftover bytes. - * - * @return true if all leftover bytes have been flushed - * @throws IOException - */ - public abstract boolean flushLeftover() throws IOException; - - // ----------------------------------- OutputBufferImpl Inner Class - - /** - * {@code OutputBufferImpl} This class is an output buffer which will write - * data to an output stream/channel. - * - * Created on Jan 10, 2012 at 12:20:15 PM - * - * @author <a href="mailto:nbenothm@redhat.com">Nabil Benothman</a> - */ - class OutputBufferImpl implements OutputBuffer { - - /** - * Write chunk. - */ - public int doWrite(ByteChunk chunk, Response res) throws IOException { - // If non blocking (event) and there are leftover bytes, - // put all remaining bytes in the leftover buffer (they are - // part of the same write operation) - if (leftover.getLength() > 0) { - leftover.append(chunk); - return chunk.getLength(); - } - - int len = chunk.getLength(); - int start = chunk.getStart(); - byte[] b = chunk.getBuffer(); - - while (len > 0) { - int thisTime = len; - // if (bbuf.position() == bbuf.capacity()) { - if (!bbuf.hasRemaining()) { - flushBuffer(); - if (leftover.getLength() > 0) { - // If non blocking (event) and there are leftover bytes, - // put all remaining bytes in the leftover buffer (they - // are - // part of the same write operation) - int oldStart = chunk.getOffset(); - chunk.setOffset(start); - leftover.append(chunk); - chunk.setOffset(oldStart); - // After that, all content has been "written" - return chunk.getLength(); - } - } - // if (thisTime > bbuf.capacity() - bbuf.position()) { - if (thisTime > bbuf.remaining()) { - // thisTime = bbuf.capacity() - bbuf.position(); - thisTime = bbuf.remaining(); - } - - bbuf.put(b, start, thisTime); - len = len - thisTime; - start = start + thisTime; - } - return chunk.getLength(); - } - } - } Modified: branches/7.4.x/src/main/java/org/apache/coyote/http11/Constants.java =================================================================== --- branches/7.4.x/src/main/java/org/apache/coyote/http11/Constants.java 2013-10-14 13:50:24 UTC (rev 2278) +++ branches/7.4.x/src/main/java/org/apache/coyote/http11/Constants.java 2013-10-15 17:30:15 UTC (rev 2279) @@ -132,6 +132,13 @@ Integer.valueOf(System.getProperty("org.apache.coyote.http11.DEFAULT_HTTP_HEADER_BUFFER_SIZE", "8192")).intValue(); + /** + * Async buffer size. + */ + public static final int ASYNC_BUFFER_SIZE = + Integer.valueOf(System.getProperty("org.apache.coyote.http11.ASYNC_BUFFER_SIZE", "32768")).intValue(); + + /* Various constant "strings" */ public static final byte[] CRLF_BYTES = ByteChunk.convertToBytes(CRLF); public static final byte[] COLON_BYTES = ByteChunk.convertToBytes(": "); Modified: branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11NioProcessor.java =================================================================== --- branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11NioProcessor.java 2013-10-14 13:50:24 UTC (rev 2278) +++ branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11NioProcessor.java 2013-10-15 17:30:15 UTC (rev 2279) @@ -214,12 +214,6 @@ try { // If processing a write event, must flush any leftover bytes first if (status == SocketStatus.OPEN_WRITE) { - // If the flush does not manage to flush all leftover bytes, the - // socket should - // go back to the poller. - if (!outputBuffer.flushLeftover()) { - return SocketState.LONG; - } // The write notification is now done writeNotification = false; // Allow convenient synchronous blocking writes Modified: branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11NioProtocol.java =================================================================== --- branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11NioProtocol.java 2013-10-14 13:50:24 UTC (rev 2278) +++ branches/7.4.x/src/main/java/org/apache/coyote/http11/Http11NioProtocol.java 2013-10-15 17:30:15 UTC (rev 2279) @@ -806,11 +806,10 @@ * .AsynchronousSocketChannel, org.apache.tomcat.util.net.ChannelStatus) */ @Override - public SocketState event(NioChannel channel, SocketStatus status) { + public synchronized SocketState event(NioChannel channel, SocketStatus status) { Http11NioProcessor processor = connections.get(channel.getId()); SocketState state = SocketState.CLOSED; - if (processor != null) { processor.startProcessing(); // Call the appropriate event @@ -868,7 +867,7 @@ state = event(channel, SocketStatus.OPEN_READ); } else if (proto.endpoint.isRunning()) { proto.endpoint.addEventChannel(channel, processor.getTimeout(), - processor.getReadNotifications(), + false, processor.getWriteNotification(), processor.getResumeNotification(), false); } Modified: branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalNioInputBuffer.java =================================================================== --- branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalNioInputBuffer.java 2013-10-14 13:50:24 UTC (rev 2278) +++ branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalNioInputBuffer.java 2013-10-15 17:30:15 UTC (rev 2279) @@ -80,11 +80,11 @@ */ private CompletionHandler<Integer, NioChannel> completionHandler; - /** - * Latch to wait for non blocking operations to run the completion handler. - */ + /** + * Latch used for auto blocking. + */ private CountDownLatch latch = null; - + /** * Create a new instance of {@code InternalNioInputBuffer} * @@ -111,36 +111,30 @@ this.completionHandler = new CompletionHandler<Integer, NioChannel>() { @Override - public void completed(Integer nBytes, NioChannel attachment) { + public synchronized void completed(Integer nBytes, NioChannel attachment) { if (nBytes < 0) { failed(new ClosedChannelException(), attachment); return; } - try { - if (nBytes > 0) { - bbuf.flip(); - bbuf.get(buf, pos, nBytes); - lastValid = pos + nBytes; - if (!processor.isProcessing() && processor.getReadNotifications()) { - endpoint.processChannel(attachment, SocketStatus.OPEN_READ); - } + if (nBytes > 0) { + bbuf.flip(); + bbuf.get(buf, pos, nBytes); + lastValid = pos + nBytes; + latch.countDown(); + if (!processor.isProcessing() && processor.getReadNotifications()) { + endpoint.processChannel(attachment, SocketStatus.OPEN_READ); } - } finally { - latch.countDown(); } } @Override public void failed(Throwable exc, NioChannel attachment) { - try { - endpoint.removeEventChannel(attachment); - if (!processor.isProcessing()) { - endpoint.processChannel(attachment, SocketStatus.ERROR); - } - } finally { - latch.countDown(); - } + exc.printStackTrace(); + endpoint.removeEventChannel(attachment); + if (!processor.isProcessing()) { + endpoint.processChannel(attachment, SocketStatus.ERROR); + } } }; } @@ -420,13 +414,25 @@ private int fill0() throws IOException { int nRead = 0; - // Prepare the internal input buffer for reading - this.prepare(); // Reading from client if (nonBlocking) { - nonBlockingRead(bbuf, readTimeout, unit); - nRead = lastValid - pos; + synchronized (completionHandler) { + // Prepare the internal input buffer for reading + this.prepare(); + nonBlockingRead(bbuf, readTimeout, unit); + nRead = lastValid - pos; + } + if (nRead == 0 && !available) { + try { + latch.await(readTimeout, unit); + } catch (InterruptedException e) { + // Ignore + } + nRead = lastValid - pos; + } } else { + // Prepare the internal input buffer for reading + this.prepare(); nRead = blockingRead(bbuf, readTimeout, unit); if (nRead > 0) { bbuf.flip(); @@ -446,7 +452,6 @@ throw new SocketTimeoutException(MESSAGES.failedRead()); } } - return nRead; } @@ -485,7 +490,6 @@ try { latch = new CountDownLatch(1); ch.read(bb, ch, this.completionHandler); - latch.await(); } catch (Throwable t) { if (CoyoteLogger.HTTP_LOGGER.isDebugEnabled()) { CoyoteLogger.HTTP_LOGGER.errorWithNonBlockingRead(t); @@ -527,20 +531,28 @@ */ public int doRead(ByteChunk chunk, Request req) throws IOException { - if (pos >= lastValid) { - int nRead = fill0(); - if (nRead < 0) { - return -1; - } else if (nRead == 0) { - return 0; - } - } + if (pos >= lastValid) { + int nRead = fill0(); + if (nRead < 0) { + return -1; + } else if (nRead == 0) { + return 0; + } + } - int length = lastValid - pos; - chunk.setBytes(buf, pos, length); - pos = lastValid; - - return (length); + if (nonBlocking) { + synchronized (completionHandler) { + int length = lastValid - pos; + chunk.setBytes(buf, pos, length); + pos = lastValid; + return (length); + } + } else { + int length = lastValid - pos; + chunk.setBytes(buf, pos, length); + pos = lastValid; + return (length); + } } } } Modified: branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalNioOutputBuffer.java =================================================================== --- branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalNioOutputBuffer.java 2013-10-14 13:50:24 UTC (rev 2278) +++ branches/7.4.x/src/main/java/org/apache/coyote/http11/InternalNioOutputBuffer.java 2013-10-15 17:30:15 UTC (rev 2279) @@ -1,6 +1,6 @@ /* * JBoss, Home of Professional Open Source. - * Copyright 2012 Red Hat, Inc., and individual contributors + * Copyright 2013 Red Hat, Inc., and individual contributors * as indicated by the @author tags. * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -21,15 +21,19 @@ import static org.jboss.web.CoyoteMessages.MESSAGES; import java.io.IOException; -import java.nio.channels.ClosedChannelException; +import java.nio.ByteBuffer; import java.nio.channels.CompletionHandler; -import java.nio.channels.WritePendingException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.apache.coyote.ActionCode; +import org.apache.coyote.OutputBuffer; import org.apache.coyote.Response; import org.apache.tomcat.util.buf.ByteChunk; +import org.apache.tomcat.util.buf.CharChunk; +import org.apache.tomcat.util.buf.MessageBytes; +import org.apache.tomcat.util.http.HttpMessages; +import org.apache.tomcat.util.http.MimeHeaders; import org.apache.tomcat.util.net.NioChannel; import org.apache.tomcat.util.net.NioEndpoint; import org.apache.tomcat.util.net.SocketStatus; @@ -42,8 +46,78 @@ * * @author <a href="mailto:nbenothm@redhat.com">Nabil Benothman</a> */ -public class InternalNioOutputBuffer extends AbstractInternalOutputBuffer { +public class InternalNioOutputBuffer implements OutputBuffer { + /** + * Associated Coyote response. + */ + protected Response response; + + /** + * Headers of the associated request. + */ + protected MimeHeaders headers; + + /** + * Committed flag. + */ + protected boolean committed; + + /** + * Finished flag. + */ + protected boolean finished; + + /** + * Pointer to the current write buffer. + */ + protected byte[] buf; + + /** + * Position in the buffer. + */ + protected int pos; + + /** + * Underlying output buffer. + */ + protected OutputBuffer outputBuffer; + + /** + * Filter library. Note: Filter[0] is always the "chunked" filter. + */ + protected OutputFilter[] filterLibrary; + + /** + * Active filter (which is actually the top of the pipeline). + */ + protected OutputFilter[] activeFilters; + + /** + * Index of the last active filter. + */ + protected int lastActiveFilter; + + /** + * Direct byte buffer used for writing. + */ + protected ByteBuffer bbuf = null; + + /** + * Leftover bytes which could not be written during a non blocking write. + */ + protected ByteChunk leftover = null; + + /** + * Non blocking mode. + */ + protected boolean nonBlocking = false; + + /** + * Write timeout + */ + protected int writeTimeout = -1; + /** * Underlying channel. */ @@ -64,12 +138,12 @@ */ private CompletionHandler<Integer, NioChannel> completionHandler; - /** - * Latch to wait for non blocking operations to run the completion handler. - */ - private CountDownLatch latch = null; - /** + * Latch used for auto blocking. + */ + private CountDownLatch latch = new CountDownLatch(0); + + /** * Create a new instance of {@code InternalNioOutputBuffer} * * @param response @@ -77,11 +151,35 @@ * @param endpoint */ public InternalNioOutputBuffer(Http11NioProcessor processor, Response response, int headerBufferSize, NioEndpoint endpoint) { - super(response, headerBufferSize); - this.endpoint = endpoint; - this.processor = processor; - // Initialize the input buffer - this.init(); + + this.response = response; + this.headers = response.getMimeHeaders(); + buf = new byte[headerBufferSize]; + if (headerBufferSize < Constants.DEFAULT_HTTP_HEADER_BUFFER_SIZE) { + bbuf = ByteBuffer.allocateDirect(6 * 1500); + } else { + bbuf = ByteBuffer.allocateDirect((headerBufferSize / 1500 + 1) * 1500); + } + + outputBuffer = new OutputBufferImpl(); + filterLibrary = new OutputFilter[0]; + activeFilters = new OutputFilter[0]; + lastActiveFilter = -1; + + committed = false; + finished = false; + + leftover = new ByteChunk(); + nonBlocking = false; + + this.endpoint = endpoint; + this.processor = processor; + // Initialize the input buffer + this.init(); + + // Cause loading of HttpMessages + HttpMessages.getMessage(200); + } /* @@ -97,34 +195,34 @@ this.completionHandler = new CompletionHandler<Integer, NioChannel>() { @Override - public void completed(Integer nBytes, NioChannel attachment) { - try { - if (nBytes < 0) { - failed(new ClosedChannelException(), attachment); - return; - } - - if (bbuf.hasRemaining()) { - try { - attachment.write(bbuf, writeTimeout, TimeUnit.MILLISECONDS, attachment, this); - } catch (WritePendingException e) { - response.setLastWrite(0); - } + public synchronized void completed(Integer nBytes, NioChannel attachment) { + if (nBytes < 0) { + failed(new IOException(MESSAGES.failedWrite()), attachment); + return; + } + if (!bbuf.hasRemaining()) { + bbuf.clear(); + if (leftover.getLength() > 0) { + int n = Math.min(leftover.getLength(), bbuf.remaining()); + bbuf.put(leftover.getBuffer(), leftover.getOffset(), n).flip(); + leftover.setOffset(leftover.getOffset() + n); } else { - // Clear the buffer when all bytes are written - clearBuffer(); response.setLastWrite(nBytes); + leftover.recycle(); + latch.countDown(); if (!processor.isProcessing() && processor.getWriteNotification()) { endpoint.processChannel(attachment, SocketStatus.OPEN_WRITE); } + return; } - } finally { - latch.countDown(); } + // Write the remaining bytes + attachment.write(bbuf, writeTimeout, TimeUnit.MILLISECONDS, attachment, this); } @Override public void failed(Throwable exc, NioChannel attachment) { + exc.printStackTrace(); endpoint.removeEventChannel(attachment); if (!processor.isProcessing()) { endpoint.processChannel(attachment, SocketStatus.ERROR); @@ -152,15 +250,6 @@ } /** - * Recycle the output buffer. This should be called when closing the - * connection. - */ - public void recycle() { - super.recycle(); - channel = null; - } - - /** * Close the channel * * @param channel @@ -209,10 +298,15 @@ */ private void nonBlockingWrite(final long timeout, final TimeUnit unit) { try { + // Calculate the number of bytes that fit in the buffer + int n = Math.min(leftover.getLength(), bbuf.capacity() - bbuf.position()); + // put bytes in the buffer + bbuf.put(leftover.getBuffer(), leftover.getOffset(), n).flip(); + // Update the offset + leftover.setOffset(leftover.getOffset() + n); + // Perform the write operation latch = new CountDownLatch(1); - // Perform the write operation - this.channel.write(this.bbuf, timeout, unit, this.channel, this.completionHandler); - latch.await(); + this.channel.write(this.bbuf, timeout, unit, this.channel, this.completionHandler); } catch (Throwable t) { if (CoyoteLogger.HTTP_LOGGER.isDebugEnabled()) { CoyoteLogger.HTTP_LOGGER.errorWithNonBlockingWrite(t); @@ -220,28 +314,25 @@ } } - /* - * (non-Javadoc) - * - * @see - * org.apache.coyote.http11.AbstractInternalOutputBuffer#write(java.nio. - * ByteBuffer, long, java.util.concurrent.TimeUnit) - */ - @Override + /** + * Perform a write operation. The operation may be blocking or non-blocking + * depending on the value of {@code nonBlocking} flag. + * + * @param timeout + * a timeout for the operation + * @param unit + * The time unit of the timeout + * @return + */ protected int write(final long timeout, final TimeUnit unit) { - if (nonBlocking) { - nonBlockingWrite(timeout, unit); - return 0; - } - return blockingWrite(timeout, unit); } - /** - * Send an acknowledgment. - * - * @throws Exception - */ + /** + * Send an acknowledgment. + * + * @throws Exception + */ public void sendAck() throws Exception { if (!committed) { @@ -253,13 +344,15 @@ } } - /* - * (non-Javadoc) - * - * @see - * org.apache.coyote.http11.AbstractInternalOutputBuffer#doWrite(org.apache - * .tomcat.util.buf.ByteChunk, org.apache.coyote.Response) - */ + /** + * Write the contents of a byte chunk. + * + * @param chunk + * byte chunk + * @return number of bytes written + * @throws IOException + * an undelying I/O error occured + */ public int doWrite(ByteChunk chunk, Response res) throws IOException { if (!committed) { @@ -269,12 +362,6 @@ response.action(ActionCode.ACTION_COMMIT, null); } - // If non blocking (event) and there are leftover bytes, - // and lastWrite was 0 -> error - if (leftover.getLength() > 0 && !(Http11NioProcessor.containerThread.get() == Boolean.TRUE)) { - throw new IOException(MESSAGES.invalidBacklog()); - } - if (lastActiveFilter == -1) { return outputBuffer.doWrite(chunk, res); } else { @@ -282,66 +369,24 @@ } } - /* - * (non-Javadoc) - * - * @see org.apache.coyote.http11.AbstractInternalOutputBuffer#flushBuffer() - */ + /** + * Callback to write data from the buffer. + */ protected void flushBuffer() throws IOException { int res = 0; - // If there are still leftover bytes here, this means the user did a - // direct flush: - // - If the call is asynchronous, throw an exception - // - If the call is synchronous, make regular blocking writes to flush - // the data - if (leftover.getLength() > 0) { - if (Http11NioProcessor.containerThread.get() == Boolean.TRUE) { - // Send leftover bytes - while (leftover.getLength() > 0) { - // Calculate the maximum number of bytes that can fit in the - // buffer - int n = Math.min(bbuf.capacity() - bbuf.position(), leftover.getLength()); - int off = leftover.getOffset(); - // Put bytes into the buffer - bbuf.put(leftover.getBuffer(), off, n).flip(); - // Update the offset of the leftover ByteChunck - leftover.setOffset(off + n); - while (bbuf.hasRemaining()) { - res = blockingWrite(writeTimeout, TimeUnit.MILLISECONDS); - if (res < 0) { - break; - } - } - bbuf.clear(); - if (res < 0) { - throw new IOException(MESSAGES.failedWrite()); - } - } - leftover.recycle(); - } else { - throw new IOException(MESSAGES.invalidBacklog()); - } - } - - if (bbuf.position() > 0) { + if (!nonBlocking && bbuf.position() > 0) { bbuf.flip(); - if (nonBlocking) { - // Perform non blocking writes until all data is written, or the - // result of the write is 0 - nonBlockingWrite(writeTimeout, TimeUnit.MILLISECONDS); - } else { - while (bbuf.hasRemaining()) { - res = blockingWrite(writeTimeout, TimeUnit.MILLISECONDS); - if (res <= 0) { - break; - } - } - response.setLastWrite(res); - // bbuf.clear(); - clearBuffer(); + while (bbuf.hasRemaining()) { + res = blockingWrite(writeTimeout, TimeUnit.MILLISECONDS); + if (res <= 0) { + break; + } } + response.setLastWrite(res); + // bbuf.clear(); + clearBuffer(); if (res < 0) { throw new IOException(MESSAGES.failedWrite()); @@ -349,57 +394,478 @@ } } - /* - * (non-Javadoc) - * - * @see - * org.apache.coyote.http11.AbstractInternalOutputBuffer#flushLeftover() - */ - @Override - public boolean flushLeftover() throws IOException { - if (leftover.getLength() == 0) { - return true; - } - // Calculate the number of bytes that fit in the buffer - int n = Math.min(leftover.getLength(), bbuf.capacity() - bbuf.position()); - // put bytes in the buffer - bbuf.put(leftover.getBuffer(), leftover.getOffset(), n).flip(); - // Update the offset - leftover.setOffset(leftover.getOffset() + n); - final NioChannel ch = channel; + /** + * Set the non blocking flag. + * + * @param nonBlocking + */ + public void setNonBlocking(boolean nonBlocking) { + this.nonBlocking = nonBlocking; + } - ch.write(bbuf, writeTimeout, TimeUnit.MILLISECONDS, null, - new CompletionHandler<Integer, Void>() { + /** + * Get the non blocking flag value. + * + * @return non blocking + */ + public boolean getNonBlocking() { + return nonBlocking; + } - @Override - public void completed(Integer result, Void attachment) { - if (result < 0) { - failed(new IOException(MESSAGES.failedWrite()), attachment); - return; - } - response.setLastWrite(result); - if (!bbuf.hasRemaining()) { - bbuf.clear(); - if (leftover.getLength() > 0) { - int n = Math.min(leftover.getLength(), bbuf.remaining()); - bbuf.put(leftover.getBuffer(), leftover.getOffset(), n).flip(); - leftover.setOffset(leftover.getOffset() + n); - } else { - leftover.recycle(); - return; - } - } - // Write the remaining bytes - ch.write(bbuf, writeTimeout, TimeUnit.MILLISECONDS, null, this); - } + /** + * Add an output filter to the filter library. + * + * @param filter + */ + public void addFilter(OutputFilter filter) { - @Override - public void failed(Throwable exc, Void attachment) { - close(ch); - } - }); + OutputFilter[] newFilterLibrary = new OutputFilter[filterLibrary.length + 1]; + for (int i = 0; i < filterLibrary.length; i++) { + newFilterLibrary[i] = filterLibrary[i]; + } + newFilterLibrary[filterLibrary.length] = filter; + filterLibrary = newFilterLibrary; + activeFilters = new OutputFilter[filterLibrary.length]; + } - return true; - } + /** + * Get filters. + * + * @return the list of filters + */ + public OutputFilter[] getFilters() { + return filterLibrary; + } + /** + * Clear filters. + */ + public void clearFilters() { + filterLibrary = new OutputFilter[0]; + lastActiveFilter = -1; + } + + /** + * Add an output filter to the filter library. + * + * @param filter + */ + public void addActiveFilter(OutputFilter filter) { + + if (lastActiveFilter == -1) { + filter.setBuffer(outputBuffer); + } else { + for (int i = 0; i <= lastActiveFilter; i++) { + if (activeFilters[i] == filter) + return; + } + filter.setBuffer(activeFilters[lastActiveFilter]); + } + + activeFilters[++lastActiveFilter] = filter; + filter.setResponse(response); + } + + public void removeActiveFilters() { + // Recycle filters + for (int i = 0; i <= lastActiveFilter; i++) { + activeFilters[i].recycle(); + } + lastActiveFilter = -1; + } + + /** + * Flush the response. + * + * @throws IOException + * an undelying I/O error occured + */ + public void flush() throws IOException { + if (!committed) { + + // 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); + } + + // Flush the current buffer + flushBuffer(); + } + + /** + * + */ + protected void clearBuffer() { + this.bbuf.clear(); + } + + /** + * Recycle this object + */ + public void recycle() { + channel = null; + // Recycle Request object + response.recycle(); + this.clearBuffer(); + pos = 0; + lastActiveFilter = -1; + committed = false; + finished = false; + } + + /** + * End processing of current HTTP request. Note: All bytes of the current + * request should have been already consumed. This method only resets all + * the pointers so that we are ready to parse the next HTTP request. + */ + public void nextRequest() { + // Recycle Request object + response.recycle(); + + // Recycle filters + for (int i = 0; i <= lastActiveFilter; i++) { + activeFilters[i].recycle(); + } + + // Reset pointers + leftover.recycle(); + pos = 0; + lastActiveFilter = -1; + committed = false; + finished = false; + nonBlocking = false; + } + + /** + * End request. + * + * @throws IOException + * an undelying I/O error occured + */ + public void endRequest() throws IOException { + + if (!committed) { + // 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); + } + + if (finished) { + return; + } + + if (lastActiveFilter != -1) { + activeFilters[lastActiveFilter].end(); + } + + flushBuffer(); + finished = true; + } + + // ------------------------------------------------ HTTP/1.1 Output Methods + + /** + * Send the response status line. + */ + public void sendStatus() { + + // Write protocol name + write(Constants.HTTP_11_BYTES); + buf[pos++] = Constants.SP; + + // Write status code + int status = response.getStatus(); + switch (status) { + case 200: + write(Constants._200_BYTES); + break; + case 400: + write(Constants._400_BYTES); + break; + case 404: + write(Constants._404_BYTES); + break; + default: + write(status); + } + + buf[pos++] = Constants.SP; + + // Write message + String message = null; + if (org.apache.coyote.Constants.USE_CUSTOM_STATUS_MSG_IN_HEADER) { + message = response.getMessage(); + } + if (message == null) { + write(HttpMessages.getMessage(status)); + } else { + write(message.replace('\n', ' ').replace('\r', ' ')); + } + + // End the response status line + buf[pos++] = Constants.CR; + buf[pos++] = Constants.LF; + } + + /** + * Send a header. + * + * @param name + * Header name + * @param value + * Header value + */ + public void sendHeader(MessageBytes name, MessageBytes value) { + if (name.getLength() > 0 && !value.isNull()) { + write(name); + buf[pos++] = Constants.COLON; + buf[pos++] = Constants.SP; + write(value); + buf[pos++] = Constants.CR; + buf[pos++] = Constants.LF; + } + } + + /** + * Send a header. + * + * @param name + * Header name + * @param value + * Header value + */ + public void sendHeader(ByteChunk name, ByteChunk value) { + write(name); + buf[pos++] = Constants.COLON; + buf[pos++] = Constants.SP; + write(value); + buf[pos++] = Constants.CR; + buf[pos++] = Constants.LF; + } + + /** + * Send a header. + * + * @param name + * Header name + * @param value + * Header value + */ + public void sendHeader(String name, String value) { + write(name); + buf[pos++] = Constants.COLON; + buf[pos++] = Constants.SP; + write(value); + buf[pos++] = Constants.CR; + buf[pos++] = Constants.LF; + } + + /** + * End the header block. + */ + public void endHeaders() { + buf[pos++] = Constants.CR; + buf[pos++] = Constants.LF; + } + + /** + * Commit the response. + * + * @throws IOException + * an undelying I/O error occured + */ + protected void commit() throws IOException { + + // The response is now committed + committed = true; + response.setCommitted(true); + + if (pos > 0) { + // Sending the response header buffer + bbuf.put(buf, 0, pos); + } + } + + /** + * This method will write the contents of the specyfied message bytes buffer + * to the output stream, without filtering. This method is meant to be used + * to write the response header. + * + * @param mb + * data to be written + */ + protected void write(MessageBytes mb) { + if (mb == null) { + return; + } + + switch (mb.getType()) { + case MessageBytes.T_BYTES: + write(mb.getByteChunk()); + break; + case MessageBytes.T_CHARS: + write(mb.getCharChunk()); + break; + default: + write(mb.toString()); + break; + } + } + + /** + * This method will write the contents of the specyfied message bytes buffer + * to the output stream, without filtering. This method is meant to be used + * to write the response header. + * + * @param bc + * data to be written + */ + protected void write(ByteChunk bc) { + // Writing the byte chunk to the output buffer + int length = bc.getLength(); + System.arraycopy(bc.getBytes(), bc.getStart(), buf, pos, length); + pos = pos + length; + } + + /** + * This method will write the contents of the specyfied char buffer to the + * output stream, without filtering. This method is meant to be used to + * write the response header. + * + * @param cc + * data to be written + */ + protected void write(CharChunk cc) { + int start = cc.getStart(); + int end = cc.getEnd(); + char[] cbuf = cc.getBuffer(); + for (int i = start; i < end; i++) { + char c = cbuf[i]; + // Note: This is clearly incorrect for many strings, + // but is the only consistent approach within the current + // servlet framework. It must suffice until servlet output + // streams properly encode their output. + if (((c <= 31) && (c != 9)) || c == 127 || c > 255) { + c = ' '; + } + buf[pos++] = (byte) c; + } + + } + + /** + * This method will write the contents of the specyfied byte buffer to the + * output stream, without filtering. This method is meant to be used to + * write the response header. + * + * @param b + * data to be written + */ + public void write(byte[] b) { + // Writing the byte chunk to the output buffer + System.arraycopy(b, 0, buf, pos, b.length); + pos = pos + b.length; + } + + /** + * This method will write the contents of the specyfied String to the output + * stream, without filtering. This method is meant to be used to write the + * response header. + * + * @param s + * data to be written + */ + protected void write(String s) { + if (s == null) { + return; + } + + // From the Tomcat 3.3 HTTP/1.0 connector + int len = s.length(); + for (int i = 0; i < len; i++) { + char c = s.charAt(i); + // Note: This is clearly incorrect for many strings, + // but is the only consistent approach within the current + // servlet framework. It must suffice until servlet output + // streams properly encode their output. + if (((c <= 31) && (c != 9)) || c == 127 || c > 255) { + c = ' '; + } + + buf[pos++] = (byte) c; + } + } + + /** + * This method will print the specified integer to the output stream, + * without filtering. This method is meant to be used to write the response + * header. + * + * @param i + * data to be written + */ + protected void write(int i) { + write(String.valueOf(i)); + } + + // ----------------------------------- OutputBufferImpl Inner Class + + /** + * {@code OutputBufferImpl} This class is an output buffer which will write + * data to an output stream/channel. + * + * Created on Jan 10, 2012 at 12:20:15 PM + * + * @author <a href="mailto:nbenothm@redhat.com">Nabil Benothman</a> + */ + class OutputBufferImpl implements OutputBuffer { + + /** + * Write chunk. + */ + public int doWrite(ByteChunk chunk, Response res) throws IOException { + if (nonBlocking) { + // Autoblocking if the buffer is already full + if (leftover.getLength() > Constants.ASYNC_BUFFER_SIZE && response.getFlushLeftovers() + && Http11AbstractProcessor.containerThread.get() == Boolean.TRUE) { + try { + latch.await(writeTimeout, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + // Ignore + } + // FIXME? throw new IOException(MESSAGES.invalidBacklog()); + } + synchronized (completionHandler) { + leftover.append(chunk); + if (leftover.getLength() > Constants.ASYNC_BUFFER_SIZE) { + response.setLastWrite(0); + } + if (latch.getCount() == 0) { + nonBlockingWrite(writeTimeout, TimeUnit.MILLISECONDS); + } + } + } else { + int len = chunk.getLength(); + int start = chunk.getStart(); + byte[] b = chunk.getBuffer(); + while (len > 0) { + int thisTime = len; + if (!bbuf.hasRemaining()) { + flushBuffer(); + } + if (thisTime > bbuf.remaining()) { + thisTime = bbuf.remaining(); + } + + bbuf.put(b, start, thisTime); + len = len - thisTime; + start = start + thisTime; + } + } + return chunk.getLength(); + } + } + }