Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3959 from eclipse/jetty-10.0.x-3952-server_direct_heap_bytebuffers 

Fixes #3952 - Server configuration for direct/heap ByteBuffers.
This commit is contained in:
Simone Bordet 2019-09-23 15:24:47 +02:00 committed by GitHub
parent 43449021d0 0eec727a09
commit f10325ea50
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
35 changed files with 270 additions and 203 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
jetty-fcgi/fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/HttpTransportOverFCGI.java
View File

@ -50,12 +50,6 @@ public class HttpTransportOverFCGI implements HttpTransport
this.request = request;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return false;
}
@Override
public void send(MetaData.Request request, MetaData.Response response, ByteBuffer content, boolean lastContent, Callback callback)
{

24
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Connection.java
View File

@ -56,6 +56,8 @@ public class HTTP2Connection extends AbstractConnection implements WriteFlusher.
private final ISession session;
private final int bufferSize;
private final ExecutionStrategy strategy;
private boolean useInputDirectByteBuffers;
private boolean useOutputDirectByteBuffers;
public HTTP2Connection(ByteBufferPool byteBufferPool, Executor executor, EndPoint endPoint, Parser parser, ISession session, int bufferSize)
{
@ -99,6 +101,26 @@ public class HTTP2Connection extends AbstractConnection implements WriteFlusher.
producer.setInputBuffer(buffer);
}
public boolean isUseInputDirectByteBuffers()
{
return useInputDirectByteBuffers;
}
public void setUseInputDirectByteBuffers(boolean useInputDirectByteBuffers)
{
this.useInputDirectByteBuffers = useInputDirectByteBuffers;
}
public boolean isUseOutputDirectByteBuffers()
{
return useOutputDirectByteBuffers;
}
public void setUseOutputDirectByteBuffers(boolean useOutputDirectByteBuffers)
{
this.useOutputDirectByteBuffers = useOutputDirectByteBuffers;
}
@Override
public void onOpen()
{
@ -389,7 +411,7 @@ public class HTTP2Connection extends AbstractConnection implements WriteFlusher.
{
private NetworkBuffer()
{
super(byteBufferPool, bufferSize, false);
super(byteBufferPool, bufferSize, isUseInputDirectByteBuffers());
}
private void put(ByteBuffer source)

6
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2StreamEndPoint.java
View File

@ -493,12 +493,6 @@ public abstract class HTTP2StreamEndPoint implements EndPoint
LOG.debug("onClose {}", this);
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return true;
}
@Override
public void upgrade(Connection newConnection)
{

5
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/generator/FrameGenerator.java
View File

@ -44,4 +44,9 @@ public abstract class FrameGenerator
{
return headerGenerator.getMaxFrameSize();
}
public boolean isUseDirectByteBuffers()
{
return headerGenerator.isUseDirectByteBuffers();
}
}

7
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/generator/Generator.java
View File

@ -38,10 +38,15 @@ public class Generator
}
public Generator(ByteBufferPool byteBufferPool, int maxDynamicTableSize, int maxHeaderBlockFragment)
{
this(byteBufferPool, true, maxDynamicTableSize, maxHeaderBlockFragment);
}
public Generator(ByteBufferPool byteBufferPool, boolean useDirectByteBuffers, int maxDynamicTableSize, int maxHeaderBlockFragment)
{
this.byteBufferPool = byteBufferPool;
headerGenerator = new HeaderGenerator();
headerGenerator = new HeaderGenerator(useDirectByteBuffers);
hpackEncoder = new HpackEncoder(maxDynamicTableSize);
this.generators = new FrameGenerator[FrameType.values().length];

18
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/generator/HeaderGenerator.java
View File

@ -27,10 +27,26 @@ import org.eclipse.jetty.io.ByteBufferPool;
public class HeaderGenerator
{
private int maxFrameSize = Frame.DEFAULT_MAX_LENGTH;
private final boolean useDirectByteBuffers;
public HeaderGenerator()
{
this(true);
}
public HeaderGenerator(boolean useDirectByteBuffers)
{
this.useDirectByteBuffers = useDirectByteBuffers;
}
public boolean isUseDirectByteBuffers()
{
return useDirectByteBuffers;
}
public ByteBuffer generate(ByteBufferPool.Lease lease, FrameType frameType, int capacity, int length, int flags, int streamId)
{
ByteBuffer header = lease.acquire(capacity, true);
ByteBuffer header = lease.acquire(capacity, isUseDirectByteBuffers());
header.put((byte)((length & 0x00_FF_00_00) >>> 16));
header.put((byte)((length & 0x00_00_FF_00) >>> 8));
header.put((byte)((length & 0x00_00_00_FF)));

2
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/generator/HeadersGenerator.java
View File

@ -67,7 +67,7 @@ public class HeadersGenerator extends FrameGenerator
flags = Flags.PRIORITY;
int maxFrameSize = getMaxFrameSize();
ByteBuffer hpacked = lease.acquire(maxFrameSize, false);
ByteBuffer hpacked = lease.acquire(maxFrameSize, isUseDirectByteBuffers());
BufferUtil.clearToFill(hpacked);
encoder.encode(hpacked, metaData);
int hpackedLength = hpacked.position();

2
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/generator/PushPromiseGenerator.java
View File

@ -58,7 +58,7 @@ public class PushPromiseGenerator extends FrameGenerator
int extraSpace = 4;
maxFrameSize -= extraSpace;
ByteBuffer hpacked = lease.acquire(maxFrameSize, false);
ByteBuffer hpacked = lease.acquire(maxFrameSize, isUseDirectByteBuffers());
BufferUtil.clearToFill(hpacked);
encoder.encode(hpacked, metaData);
int hpackedLength = hpacked.position();

20
jetty-http2/http2-hpack/src/main/java/org/eclipse/jetty/http2/hpack/HpackDecoder.java
View File

@ -25,7 +25,7 @@ import org.eclipse.jetty.http.HttpHeader;
import org.eclipse.jetty.http.HttpTokens;
import org.eclipse.jetty.http.MetaData;
import org.eclipse.jetty.http2.hpack.HpackContext.Entry;
import org.eclipse.jetty.util.TypeUtil;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
@ -77,13 +77,8 @@ public class HpackDecoder
while (buffer.hasRemaining())
{
if (LOG.isDebugEnabled() && buffer.hasArray())
{
int l = Math.min(buffer.remaining(), 32);
LOG.debug("decode {}{}",
TypeUtil.toHexString(buffer.array(), buffer.arrayOffset() + buffer.position(), l),
l < buffer.remaining() ? "..." : "");
}
if (LOG.isDebugEnabled())
LOG.debug("decode {}", BufferUtil.toHexString(buffer));
byte b = buffer.get();
if (b < 0)
@ -280,14 +275,9 @@ public class HpackDecoder
public static String toASCIIString(ByteBuffer buffer, int length)
{
StringBuilder builder = new StringBuilder(length);
int position = buffer.position();
int start = buffer.arrayOffset() + position;
int end = start + length;
buffer.position(position + length);
byte[] array = buffer.array();
for (int i = start; i < end; i++)
for (int i = 0; i < length; ++i)
{
builder.append((char)(0x7f & array[i]));
builder.append((char)(0x7F & buffer.get()));
}
return builder.toString();
}

4
jetty-http2/http2-hpack/src/main/java/org/eclipse/jetty/http2/hpack/HpackEncoder.java
View File

@ -35,6 +35,7 @@ import org.eclipse.jetty.http.MetaData;
import org.eclipse.jetty.http.PreEncodedHttpField;
import org.eclipse.jetty.http2.hpack.HpackContext.Entry;
import org.eclipse.jetty.http2.hpack.HpackContext.StaticEntry;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.TypeUtil;
import org.eclipse.jetty.util.log.Log;
@ -363,9 +364,8 @@ public class HpackEncoder
if (_debug)
{
int e = buffer.position();
if (LOG.isDebugEnabled())
LOG.debug("encode {}:'{}' to '{}'", encoding, field, TypeUtil.toHexString(buffer.array(), buffer.arrayOffset() + p, e - p));
LOG.debug("encode {}:'{}' to '{}'", encoding, field, BufferUtil.toHexString(buffer));
}
}

28
jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/AbstractHTTP2ServerConnectionFactory.java
View File

@ -64,6 +64,8 @@ public abstract class AbstractHTTP2ServerConnectionFactory extends AbstractConne
private RateControl rateControl = new WindowRateControl(20, Duration.ofSeconds(1));
private FlowControlStrategy.Factory flowControlStrategyFactory = () -> new BufferingFlowControlStrategy(0.5F);
private long streamIdleTimeout;
private boolean _useInputDirectByteBuffers;
private boolean _useOutputDirectByteBuffers;
public AbstractHTTP2ServerConnectionFactory(@Name("config") HttpConfiguration httpConfiguration)
{
@ -82,6 +84,8 @@ public abstract class AbstractHTTP2ServerConnectionFactory extends AbstractConne
this.httpConfiguration = Objects.requireNonNull(httpConfiguration);
addBean(httpConfiguration);
setInputBufferSize(Frame.DEFAULT_MAX_LENGTH + Frame.HEADER_LENGTH);
setUseInputDirectByteBuffers(httpConfiguration.isUseInputDirectByteBuffers());
setUseOutputDirectByteBuffers(httpConfiguration.isUseOutputDirectByteBuffers());
}
@ManagedAttribute("The HPACK dynamic table maximum size")
@ -192,6 +196,26 @@ public abstract class AbstractHTTP2ServerConnectionFactory extends AbstractConne
this.rateControl = rateControl;
}
public boolean isUseInputDirectByteBuffers()
{
return _useInputDirectByteBuffers;
}
public void setUseInputDirectByteBuffers(boolean useInputDirectByteBuffers)
{
_useInputDirectByteBuffers = useInputDirectByteBuffers;
}
public boolean isUseOutputDirectByteBuffers()
{
return _useOutputDirectByteBuffers;
}
public void setUseOutputDirectByteBuffers(boolean useOutputDirectByteBuffers)
{
_useOutputDirectByteBuffers = useOutputDirectByteBuffers;
}
public HttpConfiguration getHttpConfiguration()
{
return httpConfiguration;
@ -214,7 +238,7 @@ public abstract class AbstractHTTP2ServerConnectionFactory extends AbstractConne
{
ServerSessionListener listener = newSessionListener(connector, endPoint);
Generator generator = new Generator(connector.getByteBufferPool(), getMaxDynamicTableSize(), getMaxHeaderBlockFragment());
Generator generator = new Generator(connector.getByteBufferPool(), isUseOutputDirectByteBuffers(), getMaxDynamicTableSize(), getMaxHeaderBlockFragment());
FlowControlStrategy flowControl = getFlowControlStrategyFactory().newFlowControlStrategy();
HTTP2ServerSession session = new HTTP2ServerSession(connector.getScheduler(), endPoint, generator, listener, flowControl);
session.setMaxLocalStreams(getMaxConcurrentStreams());
@ -235,6 +259,8 @@ public abstract class AbstractHTTP2ServerConnectionFactory extends AbstractConne
HTTP2Connection connection = new HTTP2ServerConnection(connector.getByteBufferPool(), connector.getExecutor(),
endPoint, httpConfiguration, parser, session, getInputBufferSize(), listener);
connection.setUseInputDirectByteBuffers(isUseInputDirectByteBuffers());
connection.setUseOutputDirectByteBuffers(isUseOutputDirectByteBuffers());
connection.addListener(sessionContainer);
return configure(connection, connector, endPoint);
}

1
jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/HTTP2ServerConnection.java
View File

@ -289,6 +289,7 @@ public class HTTP2ServerConnection extends HTTP2Connection implements Connection
HttpTransportOverHTTP2 transport = new HttpTransportOverHTTP2(connector, this);
transport.setStream(stream);
channel = newServerHttpChannelOverHTTP2(connector, httpConfig, transport);
channel.setUseOutputDirectByteBuffers(isUseOutputDirectByteBuffers());
if (LOG.isDebugEnabled())
LOG.debug("Creating channel {} for {}", channel, this);
}

12
jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/HttpChannelOverHTTP2.java
View File

@ -57,6 +57,7 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
private boolean _expect100Continue;
private boolean _delayedUntilContent;
private boolean _useOutputDirectByteBuffers;
public HttpChannelOverHTTP2(Connector connector, HttpConfiguration configuration, EndPoint endPoint, HttpTransportOverHTTP2 transport)
{
@ -68,6 +69,17 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
return getHttpTransport().getStream();
}
@Override
public boolean isUseOutputDirectByteBuffers()
{
return _useOutputDirectByteBuffers;
}
public void setUseOutputDirectByteBuffers(boolean useOutputDirectByteBuffers)
{
_useOutputDirectByteBuffers = useOutputDirectByteBuffers;
}
@Override
public boolean isExpecting100Continue()
{

8
jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/HttpTransportOverHTTP2.java
View File

@ -62,14 +62,6 @@ public class HttpTransportOverHTTP2 implements HttpTransport
this.connection = connection;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
// Because sent buffers are passed directly to the endpoint without
// copying we can defer to the endpoint
return connection.getEndPoint().isOptimizedForDirectBuffers();
}
public IStream getStream()
{
return stream;

6
jetty-io/src/main/java/org/eclipse/jetty/io/AbstractEndPoint.java
View File

@ -327,12 +327,6 @@ public abstract class AbstractEndPoint extends IdleTimeout implements EndPoint
_connection = connection;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return false;
}
protected void reset()
{
_state.set(State.OPEN);

6
jetty-io/src/main/java/org/eclipse/jetty/io/ChannelEndPoint.java
View File

@ -175,12 +175,6 @@ public abstract class ChannelEndPoint extends AbstractEndPoint implements Manage
_gather = (channel instanceof GatheringByteChannel) ? (GatheringByteChannel)channel : null;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return true;
}
@Override
public boolean isOpen()
{

7
jetty-io/src/main/java/org/eclipse/jetty/io/EndPoint.java
View File

@ -269,13 +269,6 @@ public interface EndPoint extends Closeable
*/
void onClose(Throwable cause);
/**
* Is the endpoint optimized for DirectBuffer usage
*
* @return True if direct buffers can be used optimally.
*/
boolean isOptimizedForDirectBuffers();
/**
* Upgrade connections.
* Close the old connection, update the endpoint and open the new connection.

14
jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java
View File

@ -44,7 +44,6 @@ import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.http.MetaData;
import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.ChannelEndPoint;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.QuietException;
@ -251,12 +250,6 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
return _configuration;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return getHttpTransport().isOptimizedForDirectBuffers();
}
public Server getServer()
{
return _connector.getServer();
@ -971,12 +964,9 @@ public class HttpChannel implements Runnable, HttpOutput.Interceptor
return _connector.getScheduler();
}
/**
* @return true if the HttpChannel can efficiently use direct buffer (typically this means it is not over SSL or a multiplexed protocol)
*/
public boolean useDirectBuffers()
public boolean isUseOutputDirectByteBuffers()
{
return getEndPoint() instanceof ChannelEndPoint;
return getHttpConfiguration().isUseOutputDirectByteBuffers();
}
/**

6
jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelOverHttp.java
View File

@ -69,6 +69,12 @@ public class HttpChannelOverHttp extends HttpChannel implements HttpParser.Reque
_metadata.setURI(new HttpURI());
}
@Override
public boolean isUseOutputDirectByteBuffers()
{
return _httpConnection.isUseOutputDirectByteBuffers();
}
@Override
protected HttpInput newHttpInput(HttpChannelState state)
{

32
jetty-server/src/main/java/org/eclipse/jetty/server/HttpConfiguration.java
View File

@ -67,7 +67,8 @@ public class HttpConfiguration implements Dumpable
private boolean _delayDispatchUntilContent = true;
private boolean _persistentConnectionsEnabled = true;
private int _maxErrorDispatches = 10;
private boolean _useDirectByteBuffers = false;
private boolean _useInputDirectByteBuffers = true;
private boolean _useOutputDirectByteBuffers = true;
private long _minRequestDataRate;
private long _minResponseDataRate;
private HttpCompliance _httpCompliance = HttpCompliance.RFC7230;
@ -134,7 +135,8 @@ public class HttpConfiguration implements Dumpable
_delayDispatchUntilContent = config._delayDispatchUntilContent;
_persistentConnectionsEnabled = config._persistentConnectionsEnabled;
_maxErrorDispatches = config._maxErrorDispatches;
_useDirectByteBuffers = config._useDirectByteBuffers;
_useInputDirectByteBuffers = config._useInputDirectByteBuffers;
_useOutputDirectByteBuffers = config._useOutputDirectByteBuffers;
_minRequestDataRate = config._minRequestDataRate;
_minResponseDataRate = config._minResponseDataRate;
_httpCompliance = config._httpCompliance;
@ -327,17 +329,31 @@ public class HttpConfiguration implements Dumpable
}
/**
* @param useDirectByteBuffers if true, use direct byte buffers for requests
* @param useInputDirectByteBuffers whether to use direct ByteBuffers for reading
*/
public void setUseDirectByteBuffers(boolean useDirectByteBuffers)
public void setUseInputDirectByteBuffers(boolean useInputDirectByteBuffers)
{
_useDirectByteBuffers = useDirectByteBuffers;
_useInputDirectByteBuffers = useInputDirectByteBuffers;
}
@ManagedAttribute("Whether to use direct byte buffers for requests")
public boolean isUseDirectByteBuffers()
@ManagedAttribute("Whether to use direct ByteBuffers for reading")
public boolean isUseInputDirectByteBuffers()
{
return _useDirectByteBuffers;
return _useInputDirectByteBuffers;
}
/**
* @param useOutputDirectByteBuffers whether to use direct ByteBuffers for writing
*/
public void setUseOutputDirectByteBuffers(boolean useOutputDirectByteBuffers)
{
_useOutputDirectByteBuffers = useOutputDirectByteBuffers;
}
@ManagedAttribute("Whether to use direct ByteBuffers for writing")
public boolean isUseOutputDirectByteBuffers()
{
return _useOutputDirectByteBuffers;
}
/**

40
jetty-server/src/main/java/org/eclipse/jetty/server/HttpConnection.java
View File

@ -72,6 +72,8 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
private final boolean _recordHttpComplianceViolations;
private final LongAdder bytesIn = new LongAdder();
private final LongAdder bytesOut = new LongAdder();
private boolean _useInputDirectByteBuffers;
private boolean _useOutputDirectByteBuffers;
/**
* Get the current connection that this thread is dispatched to.
@ -163,12 +165,6 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
return _generator;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return getEndPoint().isOptimizedForDirectBuffers();
}
@Override
public long getMessagesIn()
{
@ -181,6 +177,26 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
return getHttpChannel().getRequests();
}
public boolean isUseInputDirectByteBuffers()
{
return _useInputDirectByteBuffers;
}
public void setUseInputDirectByteBuffers(boolean useInputDirectByteBuffers)
{
_useInputDirectByteBuffers = useInputDirectByteBuffers;
}
public boolean isUseOutputDirectByteBuffers()
{
return _useOutputDirectByteBuffers;
}
public void setUseOutputDirectByteBuffers(boolean useOutputDirectByteBuffers)
{
_useOutputDirectByteBuffers = useOutputDirectByteBuffers;
}
@Override
public ByteBuffer onUpgradeFrom()
{
@ -223,7 +239,10 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
public ByteBuffer getRequestBuffer()
{
if (_requestBuffer == null)
_requestBuffer = _bufferPool.acquire(getInputBufferSize(), _config.isUseDirectByteBuffers());
{
boolean useDirectByteBuffers = isUseInputDirectByteBuffers();
_requestBuffer = _bufferPool.acquire(getInputBufferSize(), useDirectByteBuffers);
}
return _requestBuffer;
}
@ -731,6 +750,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
if (_callback == null)
throw new IllegalStateException();
boolean useDirectByteBuffers = isUseOutputDirectByteBuffers();
ByteBuffer chunk = _chunk;
while (true)
{
@ -751,19 +771,19 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
case NEED_HEADER:
{
_header = _bufferPool.acquire(_config.getResponseHeaderSize(), _config.isUseDirectByteBuffers());
_header = _bufferPool.acquire(_config.getResponseHeaderSize(), useDirectByteBuffers);
continue;
}
case NEED_CHUNK:
{
chunk = _chunk = _bufferPool.acquire(HttpGenerator.CHUNK_SIZE, _config.isUseDirectByteBuffers());
chunk = _chunk = _bufferPool.acquire(HttpGenerator.CHUNK_SIZE, useDirectByteBuffers);
continue;
}
case NEED_CHUNK_TRAILER:
{
if (_chunk != null)
_bufferPool.release(_chunk);
chunk = _chunk = _bufferPool.acquire(_config.getResponseHeaderSize(), _config.isUseDirectByteBuffers());
chunk = _chunk = _bufferPool.acquire(_config.getResponseHeaderSize(), useDirectByteBuffers);
continue;
}
case FLUSH:

48
jetty-server/src/main/java/org/eclipse/jetty/server/HttpConnectionFactory.java
View File

@ -18,6 +18,8 @@
package org.eclipse.jetty.server;
import java.util.Objects;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EndPoint;
@ -32,7 +34,9 @@ import org.eclipse.jetty.util.annotation.Name;
public class HttpConnectionFactory extends AbstractConnectionFactory implements HttpConfiguration.ConnectionFactory
{
private final HttpConfiguration _config;
private boolean _recordHttpComplianceViolations = false;
private boolean _recordHttpComplianceViolations;
private boolean _useInputDirectByteBuffers;
private boolean _useOutputDirectByteBuffers;
public HttpConnectionFactory()
{
@ -42,10 +46,10 @@ public class HttpConnectionFactory extends AbstractConnectionFactory implements
public HttpConnectionFactory(@Name("config") HttpConfiguration config)
{
super(HttpVersion.HTTP_1_1.asString());
_config = config;
if (config == null)
throw new IllegalArgumentException("Null HttpConfiguration");
_config = Objects.requireNonNull(config);
addBean(_config);
setUseInputDirectByteBuffers(_config.isUseInputDirectByteBuffers());
setUseOutputDirectByteBuffers(_config.isUseOutputDirectByteBuffers());
}
@Override
@ -59,15 +63,37 @@ public class HttpConnectionFactory extends AbstractConnectionFactory implements
return _recordHttpComplianceViolations;
}
@Override
public Connection newConnection(Connector connector, EndPoint endPoint)
{
HttpConnection conn = new HttpConnection(_config, connector, endPoint, isRecordHttpComplianceViolations());
return configure(conn, connector, endPoint);
}
public void setRecordHttpComplianceViolations(boolean recordHttpComplianceViolations)
{
this._recordHttpComplianceViolations = recordHttpComplianceViolations;
}
public boolean isUseInputDirectByteBuffers()
{
return _useInputDirectByteBuffers;
}
public void setUseInputDirectByteBuffers(boolean useInputDirectByteBuffers)
{
_useInputDirectByteBuffers = useInputDirectByteBuffers;
}
public boolean isUseOutputDirectByteBuffers()
{
return _useOutputDirectByteBuffers;
}
public void setUseOutputDirectByteBuffers(boolean useOutputDirectByteBuffers)
{
_useOutputDirectByteBuffers = useOutputDirectByteBuffers;
}
@Override
public Connection newConnection(Connector connector, EndPoint endPoint)
{
HttpConnection connection = new HttpConnection(_config, connector, endPoint, isRecordHttpComplianceViolations());
connection.setUseInputDirectByteBuffers(isUseInputDirectByteBuffers());
connection.setUseOutputDirectByteBuffers(isUseOutputDirectByteBuffers());
return configure(connection, connector, endPoint);
}
}

18
jetty-server/src/main/java/org/eclipse/jetty/server/HttpOutput.java
View File

@ -127,14 +127,6 @@ public class HttpOutput extends ServletOutputStream implements Runnable
*/
Interceptor getNextInterceptor();
/**
* @return True if the Interceptor is optimized to receive direct
* {@link ByteBuffer}s in the {@link #write(ByteBuffer, boolean, Callback)}
* method. If false is returned, then passing direct buffers may cause
* inefficiencies.
*/
boolean isOptimizedForDirectBuffers();
/**
* Reset the buffers.
* <p>If the Interceptor contains buffers then reset them.
@ -417,7 +409,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
public ByteBuffer acquireBuffer()
{
if (_aggregate == null)
_aggregate = _channel.getByteBufferPool().acquire(getBufferSize(), _interceptor.isOptimizedForDirectBuffers());
_aggregate = _channel.getByteBufferPool().acquire(getBufferSize(), _channel.isUseOutputDirectByteBuffers());
return _aggregate;
}
@ -591,7 +583,6 @@ public class HttpOutput extends ServletOutputStream implements Runnable
// handle blocking write
// Should we aggregate?
int capacity = getBufferSize();
boolean last = isLastContentToWrite(len);
if (!last && len <= _commitSize)
{
@ -1065,7 +1056,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
break;
}
ByteBuffer buffer = _channel.useDirectBuffers() ? httpContent.getDirectBuffer() : null;
ByteBuffer buffer = _channel.isUseOutputDirectByteBuffers() ? httpContent.getDirectBuffer() : null;
if (buffer == null)
buffer = httpContent.getIndirectBuffer();
@ -1483,6 +1474,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{
super(callback);
_in = in;
// Reading from InputStream requires byte[], don't use direct buffers.
_buffer = _channel.getByteBufferPool().acquire(getBufferSize(), false);
}
@ -1535,7 +1527,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
* An iterating callback that will take content from a
* ReadableByteChannel and write it to the {@link HttpChannel}.
* A {@link ByteBuffer} of size {@link HttpOutput#getBufferSize()} is used that will be direct if
* {@link HttpChannel#useDirectBuffers()} is true.
* {@link HttpChannel#isUseOutputDirectByteBuffers()} is true.
* This callback is passed to the {@link HttpChannel#write(ByteBuffer, boolean, Callback)} to
* be notified as each buffer is written and only once all the input is consumed will the
* wrapped {@link Callback#succeeded()} method be called.
@ -1550,7 +1542,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
{
super(callback);
_in = in;
_buffer = _channel.getByteBufferPool().acquire(getBufferSize(), _channel.useDirectBuffers());
_buffer = _channel.getByteBufferPool().acquire(getBufferSize(), _channel.isUseOutputDirectByteBuffers());
}
@Override

7
jetty-server/src/main/java/org/eclipse/jetty/server/HttpTransport.java
View File

@ -73,11 +73,4 @@ public interface HttpTransport
* @param failure the failure that caused the abort.
*/
void abort(Throwable failure);
/**
* Is the underlying transport optimized for DirectBuffer usage
*
* @return True if direct buffers can be used optimally.
*/
boolean isOptimizedForDirectBuffers();
}

6
jetty-server/src/main/java/org/eclipse/jetty/server/ProxyConnectionFactory.java
View File

@ -601,12 +601,6 @@ public class ProxyConnectionFactory extends AbstractConnectionFactory
_local = local;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return _endp.isOptimizedForDirectBuffers();
}
@Override
public InetSocketAddress getLocalAddress()
{

6
jetty-server/src/main/java/org/eclipse/jetty/server/handler/BufferedResponseHandler.java
View File

@ -271,12 +271,6 @@ public class BufferedResponseHandler extends HandlerWrapper
return _next;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return false;
}
protected void commit(Queue<ByteBuffer> buffers, Callback callback)
{
// If only 1 buffer

6
jetty-server/src/main/java/org/eclipse/jetty/server/handler/gzip/GzipHttpOutputInterceptor.java
View File

@ -93,12 +93,6 @@ public class GzipHttpOutputInterceptor implements HttpOutput.Interceptor
return _interceptor;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return false; // No point as deflator is in user space.
}
@Override
public void write(ByteBuffer content, boolean complete, Callback callback)
{

6
jetty-server/src/test/java/org/eclipse/jetty/server/HttpOutputTest.java
View File

@ -651,12 +651,6 @@ public class HttpOutputTest
_next.write(BufferUtil.toBuffer(s), complete, callback);
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return _next.isOptimizedForDirectBuffers();
}
@Override
public Interceptor getNextInterceptor()
{

6
jetty-server/src/test/java/org/eclipse/jetty/server/ResponseTest.java
View File

@ -174,12 +174,6 @@ public class ResponseTest
{
_channelError = failure;
}
@Override
public boolean isOptimizedForDirectBuffers()
{
return false;
}
});
}

36
jetty-websocket/websocket-core/src/main/java/org/eclipse/jetty/websocket/core/internal/FrameFlusher.java
View File

@ -50,6 +50,8 @@ public class FrameFlusher extends IteratingCallback
private static final Logger LOG = Log.getLogger(FrameFlusher.class);
private static final Throwable CLOSED_CHANNEL = new ClosedChannelException();
private final LongAdder messagesOut = new LongAdder();
private final LongAdder bytesOut = new LongAdder();
private final ByteBufferPool bufferPool;
private final EndPoint endPoint;
private final int bufferSize;
@ -62,13 +64,12 @@ public class FrameFlusher extends IteratingCallback
private final List<Entry> previousEntries;
private final List<Entry> failedEntries;
private ByteBuffer batchBuffer = null;
private ByteBuffer batchBuffer;
private boolean canEnqueue = true;
private boolean flushed = true;
private Throwable closedCause;
private LongAdder messagesOut = new LongAdder();
private LongAdder bytesOut = new LongAdder();
private long idleTimeout = 0;
private long idleTimeout;
private boolean useDirectByteBuffers;
public FrameFlusher(ByteBufferPool bufferPool, Scheduler scheduler, Generator generator, EndPoint endPoint, int bufferSize, int maxGather)
{
@ -84,6 +85,16 @@ public class FrameFlusher extends IteratingCallback
this.timeoutScheduler = scheduler;
}
public boolean isUseDirectByteBuffers()
{
return useDirectByteBuffers;
}
public void setUseDirectByteBuffers(boolean useDirectByteBuffers)
{
this.useDirectByteBuffers = useDirectByteBuffers;
}
/**
* Enqueue a Frame to be written to the endpoint.
*
@ -225,7 +236,7 @@ public class FrameFlusher extends IteratingCallback
// Acquire a batchBuffer if we don't have one
if (batchBuffer == null)
{
batchBuffer = bufferPool.acquire(bufferSize, true);
batchBuffer = acquireBuffer(bufferSize);
buffers.add(batchBuffer);
}
@ -249,7 +260,10 @@ public class FrameFlusher extends IteratingCallback
else
{
// Add headers and payload to the list of buffers
buffers.add(entry.generateHeaderBytes());
// TODO: release this buffer.
ByteBuffer buffer = acquireBuffer(Generator.MAX_HEADER_LENGTH);
buffers.add(buffer);
entry.generateHeaderBytes(buffer);
flush = true;
ByteBuffer payload = entry.frame.getPayload();
if (BufferUtil.hasContent(payload))
@ -308,6 +322,11 @@ public class FrameFlusher extends IteratingCallback
return Action.SCHEDULED;
}
private ByteBuffer acquireBuffer(int capacity)
{
return bufferPool.acquire(capacity, isUseDirectByteBuffers());
}
private int getQueueSize()
{
synchronized (this)
@ -474,11 +493,6 @@ public class FrameFlusher extends IteratingCallback
super(frame, callback, batch);
}
private ByteBuffer generateHeaderBytes()
{
return headerBuffer = generator.generateHeaderBytes(frame);
}
private void generateHeaderBytes(ByteBuffer buffer)
{
int pos = BufferUtil.flipToFill(buffer);

34
jetty-websocket/websocket-core/src/main/java/org/eclipse/jetty/websocket/core/internal/WebSocketConnection.java
View File

@ -68,6 +68,8 @@ public class WebSocketConnection extends AbstractConnection implements Connectio
// Read / Parse variables
private RetainableByteBuffer networkBuffer;
private boolean useInputDirectByteBuffers;
private boolean useOutputDirectByteBuffers;
/**
* Create a WSConnection.
@ -132,6 +134,26 @@ public class WebSocketConnection extends AbstractConnection implements Connectio
return getEndPoint().getRemoteAddress();
}
public boolean isUseInputDirectByteBuffers()
{
return useInputDirectByteBuffers;
}
public void setUseInputDirectByteBuffers(boolean useInputDirectByteBuffers)
{
this.useInputDirectByteBuffers = useInputDirectByteBuffers;
}
public boolean isUseOutputDirectByteBuffers()
{
return useOutputDirectByteBuffers;
}
public void setUseOutputDirectByteBuffers(boolean useOutputDirectByteBuffers)
{
this.useOutputDirectByteBuffers = useOutputDirectByteBuffers;
}
/**
* Physical connection disconnect.
* <p>
@ -222,7 +244,7 @@ public class WebSocketConnection extends AbstractConnection implements Connectio
synchronized (this)
{
if (networkBuffer == null)
networkBuffer = new RetainableByteBuffer(bufferPool, getInputBufferSize());
networkBuffer = newNetworkBuffer(getInputBufferSize());
}
}
@ -237,10 +259,15 @@ public class WebSocketConnection extends AbstractConnection implements Connectio
throw new IllegalStateException();
networkBuffer.release();
networkBuffer = new RetainableByteBuffer(bufferPool, getInputBufferSize());
networkBuffer = newNetworkBuffer(getInputBufferSize());
}
}
private RetainableByteBuffer newNetworkBuffer(int capacity)
{
return new RetainableByteBuffer(bufferPool, capacity, isUseInputDirectByteBuffers());
}
private void releaseNetworkBuffer()
{
synchronized (this)
@ -445,7 +472,7 @@ public class WebSocketConnection extends AbstractConnection implements Connectio
{
synchronized (this)
{
networkBuffer = new RetainableByteBuffer(bufferPool, prefilled.remaining());
networkBuffer = newNetworkBuffer(prefilled.remaining());
}
ByteBuffer buffer = networkBuffer.getBuffer();
BufferUtil.clearToFill(buffer);
@ -572,6 +599,7 @@ public class WebSocketConnection extends AbstractConnection implements Connectio
private Flusher(Scheduler scheduler, int bufferSize, Generator generator, EndPoint endpoint)
{
super(bufferPool, scheduler, generator, endpoint, bufferSize, 8);
setUseDirectByteBuffers(isUseOutputDirectByteBuffers());
}
@Override

5
jetty-websocket/websocket-core/src/main/java/org/eclipse/jetty/websocket/core/server/internal/RFC6455Handshaker.java
View File

@ -201,6 +201,11 @@ public final class RFC6455Handshaker implements Handshaker
// Create a connection
WebSocketConnection connection = newWebSocketConnection(httpChannel.getEndPoint(), connector.getExecutor(), connector.getScheduler(), connector.getByteBufferPool(), coreSession);
// TODO: perhaps use of direct buffers should be WebSocket specific
// rather than inheriting the setting from HttpConfiguration.
HttpConfiguration httpConfig = httpChannel.getHttpConfiguration();
connection.setUseInputDirectByteBuffers(httpConfig.isUseInputDirectByteBuffers());
connection.setUseOutputDirectByteBuffers(httpChannel.isUseOutputDirectByteBuffers());
if (LOG.isDebugEnabled())
LOG.debug("connection {}", connection);
if (connection == null)

9
jetty-websocket/websocket-core/src/test/java/org/eclipse/jetty/websocket/core/extensions/ValidationExtensionTest.java
View File

@ -24,6 +24,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.websocket.core.CloseStatus;
import org.eclipse.jetty.websocket.core.ExtensionConfig;
import org.eclipse.jetty.websocket.core.Frame;
@ -88,7 +89,7 @@ public class ValidationExtensionTest extends WebSocketTester
Frame frame = serverHandler.receivedFrames.poll(5, TimeUnit.SECONDS);
assertNotNull(frame);
assertThat(frame.getOpCode(), is(OpCode.BINARY));
assertThat(frame.getPayload().array(), is(nonUtf8Payload));
assertThat(BufferUtil.toArray(frame.getPayload()), is(nonUtf8Payload));
//close normally
client.getOutputStream().write(RawFrameBuilder.buildClose(CloseStatus.NORMAL_STATUS, true));
@ -113,13 +114,13 @@ public class ValidationExtensionTest extends WebSocketTester
Frame frame = serverHandler.receivedFrames.poll(5, TimeUnit.SECONDS);
assertNotNull(frame);
assertThat(frame.getOpCode(), is(OpCode.TEXT));
assertThat(frame.getPayload().array(), is(initialPayload));
assertThat(BufferUtil.toArray(frame.getPayload()), is(initialPayload));
client.getOutputStream().write(RawFrameBuilder.buildFrame(OpCode.CONTINUATION, continuationPayload, true));
frame = serverHandler.receivedFrames.poll(5, TimeUnit.SECONDS);
assertNotNull(frame);
assertThat(frame.getOpCode(), is(OpCode.CONTINUATION));
assertThat(frame.getPayload().array(), is(continuationPayload));
assertThat(BufferUtil.toArray(frame.getPayload()), is(continuationPayload));
//close normally
client.getOutputStream().write(RawFrameBuilder.buildClose(CloseStatus.NORMAL_STATUS, true));
@ -144,7 +145,7 @@ public class ValidationExtensionTest extends WebSocketTester
Frame frame = serverHandler.receivedFrames.poll(5, TimeUnit.SECONDS);
assertNotNull(frame);
assertThat(frame.getOpCode(), is(OpCode.TEXT));
assertThat(frame.getPayload().array(), is(initialPayload));
assertThat(BufferUtil.toArray(frame.getPayload()), is(initialPayload));
client.getOutputStream().write(RawFrameBuilder.buildFrame(OpCode.CONTINUATION, incompleteContinuationPayload, true));
frame = receiveFrame(client.getInputStream());

6
jetty-websocket/websocket-core/src/test/java/org/eclipse/jetty/websocket/core/internal/MockEndpoint.java
View File

@ -164,12 +164,6 @@ public class MockEndpoint implements EndPoint
throw new UnsupportedOperationException(NOT_SUPPORTED);
}
@Override
public boolean isOptimizedForDirectBuffers()
{
throw new UnsupportedOperationException(NOT_SUPPORTED);
}
@Override
public void upgrade(Connection newConnection)
{

26
jetty-websocket/websocket-core/src/test/java/org/eclipse/jetty/websocket/core/server/WebSocketServerTest.java
View File

@ -24,7 +24,6 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.util.BlockingArrayQueue;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
@ -43,8 +42,6 @@ import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
@ -207,29 +204,6 @@ public class WebSocketServerTest extends WebSocketTester
}
assertThat(serverHandler.receivedFrames.size(), is(5));
assertThat(receivedCallbacks.size(), is(5));
byte[] first = serverHandler.receivedFrames.poll().getPayload().array();
assertThat(serverHandler.receivedFrames.poll().getPayload().array(), sameInstance(first));
assertThat(serverHandler.receivedFrames.poll().getPayload().array(), sameInstance(first));
byte[] second = serverHandler.receivedFrames.poll().getPayload().array();
assertThat(serverHandler.receivedFrames.poll().getPayload().array(), sameInstance(second));
assertThat(first, not(sameInstance(second)));
ByteBufferPool pool = server.getServer().getConnectors()[0].getByteBufferPool();
assertThat(pool.acquire(first.length, false).array(), not(sameInstance(first)));
receivedCallbacks.poll().succeeded();
assertThat(pool.acquire(first.length, false).array(), not(sameInstance(first)));
receivedCallbacks.poll().succeeded();
assertThat(pool.acquire(first.length, false).array(), not(sameInstance(first)));
receivedCallbacks.poll().succeeded();
assertThat(pool.acquire(first.length, false).array(), sameInstance(first));
assertThat(pool.acquire(second.length, false).array(), not(sameInstance(second)));
receivedCallbacks.poll().succeeded();
assertThat(pool.acquire(second.length, false).array(), not(sameInstance(second)));
receivedCallbacks.poll().succeeded();
assertThat(pool.acquire(second.length, false).array(), sameInstance(second));
}
}