| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.coyote.http11; |
| |
| import java.nio.channels.SelectionKey; |
| import java.nio.channels.SocketChannel; |
| import java.security.AccessController; |
| import java.security.PrivilegedAction; |
| import java.util.Iterator; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentLinkedQueue; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import javax.management.ObjectName; |
| |
| import org.apache.coyote.RequestGroupInfo; |
| import org.apache.coyote.RequestInfo; |
| import org.apache.juli.logging.Log; |
| import org.apache.juli.logging.LogFactory; |
| import org.apache.tomcat.util.ExceptionUtils; |
| import org.apache.tomcat.util.modeler.Registry; |
| import org.apache.tomcat.util.net.AbstractEndpoint; |
| import org.apache.tomcat.util.net.NioChannel; |
| import org.apache.tomcat.util.net.NioEndpoint; |
| import org.apache.tomcat.util.net.NioEndpoint.Handler; |
| import org.apache.tomcat.util.net.SSLImplementation; |
| import org.apache.tomcat.util.net.SecureNioChannel; |
| import org.apache.tomcat.util.net.SocketStatus; |
| |
| |
| /** |
| * Abstract the protocol implementation, including threading, etc. |
| * Processor is single threaded and specific to stream-based protocols, |
| * will not fit Jk protocols like JNI. |
| * |
| * @author Remy Maucherat |
| * @author Costin Manolache |
| * @author Filip Hanik |
| */ |
| public class Http11NioProtocol extends AbstractHttp11JsseProtocol { |
| |
| private static final Log log = LogFactory.getLog(Http11NioProtocol.class); |
| |
| |
| @Override |
| protected Log getLog() { return log; } |
| |
| |
| @Override |
| protected AbstractEndpoint.Handler getHandler() { |
| return cHandler; |
| } |
| |
| |
| public Http11NioProtocol() { |
| endpoint=new NioEndpoint(); |
| cHandler = new Http11ConnectionHandler(this); |
| ((NioEndpoint) endpoint).setHandler(cHandler); |
| setSoLinger(Constants.DEFAULT_CONNECTION_LINGER); |
| setSoTimeout(Constants.DEFAULT_CONNECTION_TIMEOUT); |
| setTcpNoDelay(Constants.DEFAULT_TCP_NO_DELAY); |
| setProcessorCache(200); |
| } |
| |
| |
| public NioEndpoint getEndpoint() { |
| return ((NioEndpoint)endpoint); |
| } |
| |
| |
| // -------------------- Properties-------------------- |
| |
| |
| private int socketCloseDelay=-1; |
| |
| private Http11ConnectionHandler cHandler; |
| |
| // -------------------- Pool setup -------------------- |
| |
| public void setPollerThreadCount(int count) { |
| ((NioEndpoint)endpoint).setPollerThreadCount(count); |
| } |
| |
| public int getPollerThreadCount() { |
| return ((NioEndpoint)endpoint).getPollerThreadCount(); |
| } |
| |
| public void setSelectorTimeout(long timeout) { |
| ((NioEndpoint)endpoint).setSelectorTimeout(timeout); |
| } |
| |
| public long getSelectorTimeout() { |
| return ((NioEndpoint)endpoint).getSelectorTimeout(); |
| } |
| // * |
| |
| |
| public void setAcceptorThreadPriority(int threadPriority) { |
| ((NioEndpoint)endpoint).setAcceptorThreadPriority(threadPriority); |
| } |
| |
| public void setPollerThreadPriority(int threadPriority) { |
| ((NioEndpoint)endpoint).setPollerThreadPriority(threadPriority); |
| } |
| |
| public int getAcceptorThreadPriority() { |
| return ((NioEndpoint)endpoint).getAcceptorThreadPriority(); |
| } |
| |
| public int getPollerThreadPriority() { |
| return ((NioEndpoint)endpoint).getThreadPriority(); |
| } |
| |
| |
| public boolean getUseSendfile() { |
| return ((NioEndpoint)endpoint).getUseSendfile(); |
| } |
| |
| public void setUseSendfile(boolean useSendfile) { |
| ((NioEndpoint)endpoint).setUseSendfile(useSendfile); |
| } |
| |
| // -------------------- Tcp setup -------------------- |
| public int getSocketCloseDelay() { |
| return socketCloseDelay; |
| } |
| |
| public void setSocketCloseDelay( int d ) { |
| socketCloseDelay=d; |
| } |
| |
| public void setOomParachute(int oomParachute) { |
| ((NioEndpoint)endpoint).setOomParachute(oomParachute); |
| } |
| |
| // ----------------------------------------------------- JMX related methods |
| |
| @Override |
| protected String getNamePrefix() { |
| return ("http-nio"); |
| } |
| |
| |
| // -------------------- Connection handler -------------------- |
| |
| static class Http11ConnectionHandler implements Handler { |
| |
| protected Http11NioProtocol proto; |
| protected static int count = 0; |
| protected RequestGroupInfo global = new RequestGroupInfo(); |
| |
| protected ConcurrentHashMap<NioChannel, Http11NioProcessor> connections = |
| new ConcurrentHashMap<NioChannel, Http11NioProcessor>(); |
| |
| protected ConcurrentLinkedQueue<Http11NioProcessor> recycledProcessors = new ConcurrentLinkedQueue<Http11NioProcessor>() { |
| private static final long serialVersionUID = 1L; |
| protected AtomicInteger size = new AtomicInteger(0); |
| @Override |
| public boolean offer(Http11NioProcessor processor) { |
| boolean offer = proto.getProcessorCache()==-1?true:size.get() < proto.getProcessorCache(); |
| //avoid over growing our cache or add after we have stopped |
| boolean result = false; |
| if ( offer ) { |
| result = super.offer(processor); |
| if ( result ) { |
| size.incrementAndGet(); |
| } |
| } |
| if (!result) deregister(processor); |
| return result; |
| } |
| |
| @Override |
| public Http11NioProcessor poll() { |
| Http11NioProcessor result = super.poll(); |
| if ( result != null ) { |
| size.decrementAndGet(); |
| } |
| return result; |
| } |
| |
| @Override |
| public void clear() { |
| Http11NioProcessor next = poll(); |
| while ( next != null ) { |
| deregister(next); |
| next = poll(); |
| } |
| super.clear(); |
| size.set(0); |
| } |
| }; |
| |
| Http11ConnectionHandler(Http11NioProtocol proto) { |
| this.proto = proto; |
| } |
| |
| @Override |
| public Object getGlobal() { |
| return global; |
| } |
| |
| @Override |
| public void recycle() { |
| recycledProcessors.clear(); |
| } |
| |
| @Override |
| public SSLImplementation getSslImplementation() { |
| return proto.sslImplementation; |
| } |
| |
| @Override |
| public void release(SocketChannel socket) { |
| if (log.isDebugEnabled()) |
| log.debug("Iterating through our connections to release a socket channel:"+socket); |
| boolean released = false; |
| Iterator<java.util.Map.Entry<NioChannel, Http11NioProcessor>> it = connections.entrySet().iterator(); |
| while (it.hasNext()) { |
| java.util.Map.Entry<NioChannel, Http11NioProcessor> entry = it.next(); |
| if (entry.getKey().getIOChannel()==socket) { |
| it.remove(); |
| Http11NioProcessor result = entry.getValue(); |
| result.recycle(); |
| deregister(result); |
| released = true; |
| break; |
| } |
| } |
| if (log.isDebugEnabled()) |
| log.debug("Done iterating through our connections to release a socket channel:"+socket +" released:"+released); |
| } |
| |
| /** |
| * Use this only if the processor is not available, otherwise use |
| * {@link #release(NioChannel, Http11NioProcessor). |
| */ |
| @Override |
| public void release(NioChannel socket) { |
| Http11NioProcessor processor = connections.remove(socket); |
| if (processor != null) { |
| processor.recycle(); |
| recycledProcessors.offer(processor); |
| } |
| } |
| |
| |
| public void release(NioChannel socket, Http11NioProcessor processor) { |
| connections.remove(socket); |
| processor.recycle(); |
| recycledProcessors.offer(processor); |
| } |
| |
| |
| @Override |
| public SocketState event(NioChannel socket, SocketStatus status) { |
| Http11NioProcessor processor = connections.get(socket); |
| NioEndpoint.KeyAttachment att = (NioEndpoint.KeyAttachment)socket.getAttachment(false); |
| att.setAsync(false); //no longer check for timeout |
| SocketState state = SocketState.CLOSED; |
| if (processor != null) { |
| if (log.isDebugEnabled()) log.debug("Http11NioProcessor.error="+processor.error); |
| // Call the appropriate event |
| try { |
| if (processor.comet) { |
| state = processor.event(status); |
| } else { |
| state = processor.asyncDispatch(status); |
| } |
| } catch (java.net.SocketException e) { |
| // SocketExceptions are normal |
| Http11NioProtocol.log.debug |
| (sm.getString |
| ("http11protocol.proto.socketexception.debug"), e); |
| } catch (java.io.IOException e) { |
| // IOExceptions are normal |
| Http11NioProtocol.log.debug |
| (sm.getString |
| ("http11protocol.proto.ioexception.debug"), e); |
| } |
| // Future developers: if you discover any other |
| // rare-but-nonfatal exceptions, catch them here, and log as |
| // above. |
| catch (Throwable e) { |
| ExceptionUtils.handleThrowable(e); |
| // any other exception or error is odd. Here we log it |
| // with "ERROR" level, so it will show up even on |
| // less-than-verbose logs. |
| Http11NioProtocol.log.error |
| (sm.getString("http11protocol.proto.error"), e); |
| } finally { |
| if (processor.isAsync()) { |
| state = processor.asyncPostProcess(); |
| } |
| if (state == SocketState.OPEN || state == SocketState.CLOSED) { |
| release(socket, processor); |
| if (state == SocketState.OPEN) { |
| socket.getPoller().add(socket); |
| } |
| } else if (state == SocketState.LONG) { |
| if (processor.isAsync()) { |
| att.setAsync(true); // Re-enable timeouts |
| } else { |
| // Comet |
| if (log.isDebugEnabled()) log.debug("Keeping processor["+processor); |
| // May receive more data from client |
| SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector()); |
| key.interestOps(SelectionKey.OP_READ); |
| att.interestOps(SelectionKey.OP_READ); |
| } |
| } else { |
| // state == SocketState.ASYNC_END |
| // No further work required |
| } |
| } |
| } |
| return state; |
| } |
| |
| @Override |
| public SocketState process(NioChannel socket) { |
| Http11NioProcessor processor = connections.remove(socket); |
| try { |
| if (processor == null) { |
| processor = recycledProcessors.poll(); |
| } |
| if (processor == null) { |
| processor = createProcessor(); |
| } |
| |
| if (proto.isSSLEnabled() && |
| (proto.sslImplementation != null) |
| && (socket instanceof SecureNioChannel)) { |
| SecureNioChannel ch = (SecureNioChannel)socket; |
| processor.setSslSupport( |
| proto.sslImplementation.getSSLSupport( |
| ch.getSslEngine().getSession())); |
| } else { |
| processor.setSslSupport(null); |
| } |
| |
| SocketState state = processor.process(socket); |
| if (state == SocketState.LONG) { |
| // In the middle of processing a request/response. Keep the |
| // socket associated with the processor. |
| connections.put(socket, processor); |
| |
| if (processor.isAsync()) { |
| NioEndpoint.KeyAttachment att = (NioEndpoint.KeyAttachment)socket.getAttachment(false); |
| att.setAsync(true); |
| // longPoll may change socket state (e.g. to trigger a |
| // complete or dispatch) |
| state = processor.asyncPostProcess(); |
| } else { |
| // Either: |
| // - this is comet request |
| // - the request line/headers have not been completely |
| // read |
| SelectionKey key = socket.getIOChannel().keyFor( |
| socket.getPoller().getSelector()); |
| NioEndpoint.KeyAttachment att = |
| (NioEndpoint.KeyAttachment)socket.getAttachment(false); |
| key.interestOps(SelectionKey.OP_READ); |
| att.interestOps(SelectionKey.OP_READ); |
| } |
| } |
| if (state == SocketState.LONG || state == SocketState.ASYNC_END) { |
| // Already done all we need to do. |
| } else if (state == SocketState.OPEN){ |
| // In keep-alive but between requests. OK to recycle |
| // processor. Continue to poll for the next request. |
| release(socket, processor); |
| socket.getPoller().add(socket); |
| } else { |
| // Connection closed. OK to recycle the processor. |
| release(socket, processor); |
| } |
| return state; |
| |
| } catch (java.net.SocketException e) { |
| // SocketExceptions are normal |
| log.debug(sm.getString( |
| "http11protocol.proto.socketexception.debug"), e); |
| } catch (java.io.IOException e) { |
| // IOExceptions are normal |
| log.debug(sm.getString( |
| "http11protocol.proto.ioexception.debug"), e); |
| } |
| // Future developers: if you discover any other |
| // rare-but-nonfatal exceptions, catch them here, and log as |
| // above. |
| catch (Throwable e) { |
| ExceptionUtils.handleThrowable(e); |
| // any other exception or error is odd. Here we log it |
| // with "ERROR" level, so it will show up even on |
| // less-than-verbose logs. |
| log.error(sm.getString("http11protocol.proto.error"), e); |
| } |
| release(socket, processor); |
| return SocketState.CLOSED; |
| } |
| |
| public Http11NioProcessor createProcessor() { |
| Http11NioProcessor processor = new Http11NioProcessor( |
| proto.getMaxHttpHeaderSize(), (NioEndpoint)proto.endpoint, |
| proto.getMaxTrailerSize()); |
| processor.setAdapter(proto.adapter); |
| processor.setMaxKeepAliveRequests(proto.getMaxKeepAliveRequests()); |
| processor.setConnectionUploadTimeout( |
| proto.getConnectionUploadTimeout()); |
| processor.setDisableUploadTimeout(proto.getDisableUploadTimeout()); |
| processor.setCompressionMinSize(proto.getCompressionMinSize()); |
| processor.setCompression(proto.getCompression()); |
| processor.setNoCompressionUserAgents(proto.getNoCompressionUserAgents()); |
| processor.setCompressableMimeTypes(proto.getCompressableMimeTypes()); |
| processor.setRestrictedUserAgents(proto.getRestrictedUserAgents()); |
| processor.setSocketBuffer(proto.getSocketBuffer()); |
| processor.setMaxSavePostSize(proto.getMaxSavePostSize()); |
| processor.setServer(proto.getServer()); |
| register(processor); |
| return processor; |
| } |
| AtomicInteger registerCount = new AtomicInteger(0); |
| public void register(Http11NioProcessor processor) { |
| if (proto.getDomain() != null) { |
| synchronized (this) { |
| try { |
| registerCount.addAndGet(1); |
| if (log.isDebugEnabled()) log.debug("Register ["+processor+"] count="+registerCount.get()); |
| final RequestInfo rp = processor.getRequest().getRequestProcessor(); |
| rp.setGlobalProcessor(global); |
| final ObjectName rpName = new ObjectName |
| (proto.getDomain() + ":type=RequestProcessor,worker=" |
| + proto.getName() + ",name=HttpRequest" + count++); |
| if (Constants.IS_SECURITY_ENABLED) { |
| AccessController.doPrivileged(new PrivilegedAction<Void>() { |
| @Override |
| public Void run() { |
| try { |
| Registry.getRegistry(null, null).registerComponent(rp, rpName, null); |
| } catch (Exception e) { |
| log.warn("Error registering request"); |
| } |
| return null; |
| } |
| }); |
| } else { |
| Registry.getRegistry(null, null).registerComponent(rp, rpName, null); |
| } |
| rp.setRpName(rpName); |
| } catch (Exception e) { |
| log.warn("Error registering request"); |
| } |
| } |
| } |
| } |
| |
| public void deregister(Http11NioProcessor processor) { |
| if (proto.getDomain() != null) { |
| synchronized (this) { |
| try { |
| registerCount.addAndGet(-1); |
| if (log.isDebugEnabled()) log.debug("Deregister ["+processor+"] count="+registerCount.get()); |
| RequestInfo rp = processor.getRequest().getRequestProcessor(); |
| rp.setGlobalProcessor(null); |
| ObjectName rpName = rp.getRpName(); |
| Registry.getRegistry(null, null).unregisterComponent(rpName); |
| rp.setRpName(null); |
| } catch (Exception e) { |
| log.warn("Error unregistering request", e); |
| } |
| } |
| } |
| } |
| |
| } |
| |
| |
| } |
