Author: rgodfrey Date: Tue Mar 14 11:49:27 2017 New Revision: 1786887 URL: http://svn.apache.org/viewvc?rev=1786887&view=rev Log: QPID-7708 : Allow session window credit to be set by a context variable
Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/Session_1_0Test.java Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java?rev=1786887&r1=1786886&r2=1786887&view=diff ============================================================================== --- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java (original) +++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java Tue Mar 14 11:49:27 2017 @@ -25,6 +25,7 @@ import java.util.List; import org.apache.qpid.server.bytebuffer.QpidByteBuffer; import org.apache.qpid.server.logging.EventLoggerProvider; +import org.apache.qpid.server.model.ManagedContextDefault; import org.apache.qpid.server.model.ManagedObject; import org.apache.qpid.server.protocol.v1_0.codec.SectionDecoderRegistry; import org.apache.qpid.server.protocol.v1_0.type.FrameBody; @@ -42,6 +43,11 @@ public interface AMQPConnection_1_0<C ex ConnectionHandler, EventLoggerProvider { + + String CONNECTION_SESSION_CREDIT_WINDOW_SIZE = "connection.sessionCreditWindowSize"; + @ManagedContextDefault(name = CONNECTION_SESSION_CREDIT_WINDOW_SIZE) + int DEFAULT_CONNECTION_SESSION_CREDIT_WINDOW_SIZE = 8192; + Symbol ANONYMOUS_RELAY = Symbol.valueOf("ANONYMOUS-RELAY"); Symbol SHARED_SUBSCRIPTIONS = Symbol.valueOf("SHARED-SUBS"); Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java?rev=1786887&r1=1786886&r2=1786887&view=diff ============================================================================== --- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java (original) +++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java Tue Mar 14 11:49:27 2017 @@ -603,7 +603,11 @@ public class AMQPConnection_1_0Impl exte } else { - Session_1_0 session = new Session_1_0(this, begin, sendingChannelId, receivingChannelId); + Session_1_0 session = new Session_1_0(this, + begin, + sendingChannelId, + receivingChannelId, + getContextValue(Integer.class, AMQPConnection_1_0.CONNECTION_SESSION_CREDIT_WINDOW_SIZE)); session.create(); _receivingSessions[receivingChannelId] = session; Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java?rev=1786887&r1=1786886&r2=1786887&view=diff ============================================================================== --- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java (original) +++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java Tue Mar 14 11:49:27 2017 @@ -141,11 +141,7 @@ public class Session_1_0 extends Abstrac private final short _sendingChannel; - // has to be a power of two private static final int DEFAULT_SESSION_BUFFER_SIZE = 1 << 11; - private static final int BUFFER_SIZE_MASK = DEFAULT_SESSION_BUFFER_SIZE - 1; - - private int _nextOutgoingDeliveryId; @@ -157,7 +153,7 @@ public class Session_1_0 extends Abstrac private LinkedHashMap<UnsignedInteger,Delivery> _outgoingUnsettled = new LinkedHashMap<>(DEFAULT_SESSION_BUFFER_SIZE); private LinkedHashMap<UnsignedInteger,Delivery> _incomingUnsettled = new LinkedHashMap<>(DEFAULT_SESSION_BUFFER_SIZE); - private int _availableIncomingCredit = DEFAULT_SESSION_BUFFER_SIZE; + private final int _incomingWindowSize; private int _availableOutgoingCredit = DEFAULT_SESSION_BUFFER_SIZE; private UnsignedInteger _lastSentIncomingLimit; @@ -175,7 +171,8 @@ public class Session_1_0 extends Abstrac public Session_1_0(final AMQPConnection_1_0 connection, Begin begin, short sendingChannelId, - short receivingChannelId) + short receivingChannelId, + int incomingWindowSize) { super(connection, sendingChannelId); _sendingChannel = sendingChannelId; @@ -184,6 +181,7 @@ public class Session_1_0 extends Abstrac _nextIncomingTransferId = new SequenceNumber(begin.getNextOutgoingId().intValue()); _connection = connection; _primaryDomain = getPrimaryDomain(); + _incomingWindowSize = incomingWindowSize; AccessController.doPrivileged((new PrivilegedAction<Object>() { @@ -390,8 +388,10 @@ public class Session_1_0 extends Abstrac { UnsignedInteger clientsCredit = _lastSentIncomingLimit.subtract(UnsignedInteger.valueOf(_nextIncomingTransferId.intValue())); - int i = UnsignedInteger.valueOf(_availableIncomingCredit).subtract(clientsCredit).compareTo(clientsCredit); - if (i >= 0) + + // TODO - we should use a better metric here, and/or manage session credit across the whole connection + // send a flow if the window is at least half used up + if (UnsignedInteger.valueOf(_incomingWindowSize).subtract(clientsCredit).compareTo(clientsCredit) >= 0) { sendFlow(); } @@ -497,9 +497,9 @@ public class Session_1_0 extends Abstrac { final int nextIncomingId = _nextIncomingTransferId.intValue(); flow.setNextIncomingId(UnsignedInteger.valueOf(nextIncomingId)); - _lastSentIncomingLimit = UnsignedInteger.valueOf(nextIncomingId + _availableIncomingCredit); + _lastSentIncomingLimit = UnsignedInteger.valueOf(nextIncomingId + _incomingWindowSize); } - flow.setIncomingWindow(UnsignedInteger.valueOf(_availableIncomingCredit)); + flow.setIncomingWindow(UnsignedInteger.valueOf(_incomingWindowSize)); flow.setNextOutgoingId(UnsignedInteger.valueOf(_nextOutgoingTransferId.intValue())); flow.setOutgoingWindow(UnsignedInteger.valueOf(_availableOutgoingCredit)); @@ -653,7 +653,7 @@ public class Session_1_0 extends Abstrac UnsignedInteger getIncomingWindowSize() { - return UnsignedInteger.valueOf(_availableIncomingCredit); + return UnsignedInteger.valueOf(_incomingWindowSize); } AccessControlContext getAccessControllerContext() Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/Session_1_0Test.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/Session_1_0Test.java?rev=1786887&r1=1786886&r2=1786887&view=diff ============================================================================== --- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/Session_1_0Test.java (original) +++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/Session_1_0Test.java Tue Mar 14 11:49:27 2017 @@ -715,7 +715,7 @@ public class Session_1_0Test extends Qpi { Begin begin = mock(Begin.class); when(begin.getNextOutgoingId()).thenReturn(new UnsignedInteger(channelId)); - Session_1_0 session = new Session_1_0(connection, begin, (short) channelId, (short) channelId); + Session_1_0 session = new Session_1_0(connection, begin, (short) channelId, (short) channelId, 2048); return session; } @@ -728,4 +728,4 @@ public class Session_1_0Test extends Qpi detach.setClosed(closed); session.receiveDetach(detach); } -} \ No newline at end of file +} --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org For additional commands, e-mail: commits-h...@qpid.apache.org