You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2016/11/16 15:58:24 UTC
svn commit: r1770006 -
/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/
Author: rgodfrey
Date: Wed Nov 16 15:58:24 2016
New Revision: 1770006
URL: http://svn.apache.org/viewvc?rev=1770006&view=rev
Log:
QPID-7518 : Keep track of buffered size to improve performance
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionUndecidedDelegate.java
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java?rev=1770006&r1=1770005&r2=1770006&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java Wed Nov 16 15:58:24 2016
@@ -44,7 +44,6 @@ import org.apache.qpid.server.model.port
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.transport.ByteBufferSender;
-import org.apache.qpid.transport.network.Ticker;
import org.apache.qpid.transport.network.TransportEncryption;
import org.apache.qpid.util.SystemUtils;
@@ -77,6 +76,7 @@ public class NonBlockingConnection imple
private final AtomicLong _maxReadIdleMillis = new AtomicLong();
private final List<SchedulingDelayNotificationListener> _schedulingDelayNotificationListeners = new CopyOnWriteArrayList<>();
private final AtomicBoolean _hasShutdown = new AtomicBoolean();
+ private long _bufferedSize;
public NonBlockingConnection(SocketChannel socketChannel,
ProtocolEngine protocolEngine,
@@ -368,17 +368,7 @@ public class NonBlockingConnection imple
private long getBufferedSize()
{
- // Avoids iterator garbage if empty
- if (_buffers.isEmpty())
- {
- return 0L;
- }
- long totalSize = 0L;
- for(QpidByteBuffer buf : _buffers)
- {
- totalSize += buf.remaining();
- }
- return totalSize;
+ return _bufferedSize;
}
private void shutdown()
@@ -515,7 +505,9 @@ public class NonBlockingConnection imple
private boolean doWrite() throws IOException
{
- _fullyWritten = _delegate.doWrite(_buffers);
+ final NonBlockingConnectionDelegate.WriteResult result = _delegate.doWrite(_buffers);
+ _bufferedSize -= result.getBytesConsumed();
+ _fullyWritten = result.isComplete();
while(!_buffers.isEmpty())
{
QpidByteBuffer buf = _buffers.peek();
@@ -565,6 +557,7 @@ public class NonBlockingConnection imple
else if (msg.remaining() > 0)
{
_buffers.add(msg.duplicate());
+ _bufferedSize += msg.remaining();
}
msg.position(msg.limit());
}
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java?rev=1770006&r1=1770005&r2=1770006&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java Wed Nov 16 15:58:24 2016
@@ -28,7 +28,28 @@ import org.apache.qpid.bytebuffer.QpidBy
interface NonBlockingConnectionDelegate
{
- boolean doWrite(Collection<QpidByteBuffer> bufferArray) throws IOException;
+ class WriteResult
+ {
+ final boolean _complete;
+ final long _bytesConsumed;
+
+ public WriteResult(final boolean complete, final long bytesConsumed)
+ {
+ _complete = complete;
+ _bytesConsumed = bytesConsumed;
+ }
+
+ public boolean isComplete()
+ {
+ return _complete;
+ }
+
+ public long getBytesConsumed()
+ {
+ return _bytesConsumed;
+ }
+ }
+ WriteResult doWrite(Collection<QpidByteBuffer> bufferArray) throws IOException;
boolean readyForRead();
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java?rev=1770006&r1=1770005&r2=1770006&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java Wed Nov 16 15:58:24 2016
@@ -24,11 +24,11 @@ import java.security.Principal;
import java.security.cert.Certificate;
import java.util.Collection;
-import org.apache.qpid.server.model.port.AmqpPort;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.model.port.AmqpPort;
public class NonBlockingConnectionPlainDelegate implements NonBlockingConnectionDelegate
{
@@ -99,9 +99,9 @@ public class NonBlockingConnectionPlainD
@Override
- public boolean doWrite(Collection<QpidByteBuffer> bufferArray) throws IOException
+ public WriteResult doWrite(Collection<QpidByteBuffer> bufferArray) throws IOException
{
- long bytesToWrite = 0l;
+ long bytesToWrite = 0L;
if(!bufferArray.isEmpty())
{
for (QpidByteBuffer buf : bufferArray)
@@ -109,7 +109,16 @@ public class NonBlockingConnectionPlainD
bytesToWrite += buf.remaining();
}
}
- return bytesToWrite == 0l || _parent.writeToTransport(bufferArray) >= bytesToWrite;
+ if(bytesToWrite == 0L)
+ {
+ return new WriteResult(true, 0);
+ }
+ else
+ {
+
+ long bytesWritten = _parent.writeToTransport(bufferArray);
+ return new WriteResult(bytesWritten >= bytesToWrite, bytesWritten);
+ }
}
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java?rev=1770006&r1=1770005&r2=1770006&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java Wed Nov 16 15:58:24 2016
@@ -19,19 +19,6 @@
package org.apache.qpid.server.transport;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.server.model.port.AmqpPort;
-import org.apache.qpid.server.util.ServerScopedRuntimeException;
-import org.apache.qpid.transport.network.security.ssl.SSLUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.net.ssl.SSLEngine;
-import javax.net.ssl.SSLEngineResult;
-import javax.net.ssl.SSLException;
-import javax.net.ssl.SSLPeerUnverifiedException;
-import javax.net.ssl.SSLSession;
-
import java.io.IOException;
import java.security.Principal;
import java.security.cert.Certificate;
@@ -41,6 +28,20 @@ import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLSession;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.util.ServerScopedRuntimeException;
+import org.apache.qpid.transport.network.security.ssl.SSLUtil;
+
public class NonBlockingConnectionTLSDelegate implements NonBlockingConnectionDelegate
{
private static final Logger LOGGER = LoggerFactory.getLogger(NonBlockingConnectionTLSDelegate.class);
@@ -129,11 +130,11 @@ public class NonBlockingConnectionTLSDel
}
@Override
- public boolean doWrite(Collection<QpidByteBuffer> bufferArray) throws IOException
+ public WriteResult doWrite(Collection<QpidByteBuffer> bufferArray) throws IOException
{
final int bufCount = bufferArray.size();
- wrapBufferArray(bufferArray);
+ int totalConsumed = wrapBufferArray(bufferArray);
boolean bufsSent = true;
final Iterator<QpidByteBuffer> itr = bufferArray.iterator();
@@ -163,7 +164,7 @@ public class NonBlockingConnectionTLSDel
}
}
}
- return bufsSent && _encryptedOutput.isEmpty();
+ return new WriteResult(bufsSent && _encryptedOutput.isEmpty(), totalConsumed);
}
protected void restoreApplicationBufferForWrite()
@@ -200,8 +201,9 @@ public class NonBlockingConnectionTLSDel
}
- private void wrapBufferArray(Collection<QpidByteBuffer> bufferArray) throws SSLException
+ private int wrapBufferArray(Collection<QpidByteBuffer> bufferArray) throws SSLException
{
+ int totalConsumed = 0;
boolean encrypted;
do
{
@@ -223,6 +225,7 @@ public class NonBlockingConnectionTLSDel
_status = QpidByteBuffer.encryptSSL(_sslEngine, bufferArray, _netOutputBuffer);
encrypted = _status.bytesProduced() > 0;
+ totalConsumed += _status.bytesConsumed();
runSSLEngineTasks(_status);
if(encrypted && _netOutputBuffer.remaining() < _sslEngine.getSession().getPacketBufferSize())
{
@@ -250,7 +253,7 @@ public class NonBlockingConnectionTLSDel
_encryptedOutput.add(outputBuffer);
}
-
+ return totalConsumed;
}
private boolean runSSLEngineTasks(final SSLEngineResult status)
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionUndecidedDelegate.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionUndecidedDelegate.java?rev=1770006&r1=1770005&r2=1770006&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionUndecidedDelegate.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionUndecidedDelegate.java Wed Nov 16 15:58:24 2016
@@ -19,14 +19,14 @@
package org.apache.qpid.server.transport;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.transport.network.TransportEncryption;
-
import java.io.IOException;
import java.security.Principal;
import java.security.cert.Certificate;
import java.util.Collection;
+import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.transport.network.TransportEncryption;
+
public class NonBlockingConnectionUndecidedDelegate implements NonBlockingConnectionDelegate
{
private static final int NUMBER_OF_BYTES_FOR_TLS_CHECK = 6;
@@ -73,9 +73,9 @@ public class NonBlockingConnectionUndeci
}
@Override
- public boolean doWrite(Collection<QpidByteBuffer> buffers) throws IOException
+ public WriteResult doWrite(Collection<QpidByteBuffer> buffers) throws IOException
{
- return true;
+ return new WriteResult(true, 0L);
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org