channel block/unblock now async, remove unnecessary selector bumps

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-6262-JavaBrokerNIO@1663708 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 24 insertions, 60 deletions

diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index d7e5857924..9631530f90 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -209,6 +209,8 @@ public class AMQChannel
     private final List<StoredMessage<MessageMetaData>> _uncommittedMessages = new ArrayList<>();
     private long _maxUncommittedInMemorySize;
 
+    private boolean _wireBlockingState;
+
     public AMQChannel(AMQProtocolEngine connection, int channelId, final MessageStore messageStore)
     {
         _creditManager = new Pre0_10CreditManager(0l,0l, connection);
@@ -1611,12 +1613,14 @@ public class AMQChannel
     {
         if(_blockingEntities.add(this))
         {
+
             if(_blocking.compareAndSet(false,true))
             {
                 getVirtualHost().getEventLogger().message(_logSubject,
                                                           ChannelMessages.FLOW_ENFORCED("** All Queues **"));
-                flow(false);
-                _blockTime = System.currentTimeMillis();
+
+
+                getConnection().notifyWork();
             }
         }
     }
@@ -1628,8 +1632,7 @@ public class AMQChannel
             if(_blockingEntities.isEmpty() && _blocking.compareAndSet(true,false))
             {
                 getVirtualHost().getEventLogger().message(_logSubject, ChannelMessages.FLOW_REMOVED());
-
-                flow(true);
+                getConnection().notifyWork();
             }
         }
     }
@@ -1643,8 +1646,7 @@ public class AMQChannel
             if(_blocking.compareAndSet(false,true))
             {
                 getVirtualHost().getEventLogger().message(_logSubject, ChannelMessages.FLOW_ENFORCED(queue.getName()));
-                flow(false);
-                _blockTime = System.currentTimeMillis();
+                getConnection().notifyWork();
 
             }
         }
@@ -1657,7 +1659,7 @@ public class AMQChannel
             if(_blockingEntities.isEmpty() && _blocking.compareAndSet(true,false) && !isClosing())
             {
                 getVirtualHost().getEventLogger().message(_logSubject, ChannelMessages.FLOW_REMOVED());
-                flow(true);
+                getConnection().notifyWork();
             }
         }
     }
@@ -2262,7 +2264,7 @@ public class AMQChannel
     private boolean blockingTimeoutExceeded()
     {
 
-        return _blocking.get() && (System.currentTimeMillis() - _blockTime) > _blockingTimeout;
+        return _wireBlockingState && (System.currentTimeMillis() - _blockTime) > _blockingTimeout;
     }
 
     @Override
@@ -3598,9 +3600,17 @@ public class AMQChannel
     }
 
     @Override
-    public void processPendingMessages()
+    public void processPending()
     {
 
+        boolean desiredBlockingState = _blocking.get();
+        if (desiredBlockingState != _wireBlockingState)
+        {
+            _wireBlockingState = desiredBlockingState;
+            flow(!desiredBlockingState);
+            _blockTime = desiredBlockingState ? System.currentTimeMillis() : 0;
+        }
+
         for(ConsumerTarget target : _tag2SubscriptionTargetMap.values())
         {
             target.processPending();
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
index cb65424b67..659207d9e8 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
@@ -44,8 +44,6 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
 import javax.security.auth.Subject;
 import javax.security.sasl.SaslException;
@@ -148,11 +146,8 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
      * The channels that the latest call to {@link #received(ByteBuffer)} applied to.
      * Used so we know which channels we need to call {@link AMQChannel#receivedComplete()}
      * on after handling the frames.
-     *
-     * Thread-safety: guarded by {@link #_receivedLock}.
      */
-    private final Set<AMQChannel> _channelsForCurrentMessage =
-            new HashSet<>();
+    private final Set<AMQChannel> _channelsForCurrentMessage = new HashSet<>();
 
     private AMQDecoder _decoder;
 
@@ -197,7 +192,6 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
     private long _lastReceivedTime = System.currentTimeMillis();  // TODO consider if this is what we want?
     private boolean _blocking;
 
-    private final ReentrantLock _receivedLock;
     private AtomicLong _lastWriteTime = new AtomicLong(System.currentTimeMillis());
     private final Broker<?> _broker;
     private final Transport _transport;
@@ -251,7 +245,6 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
         _port = port;
         _transport = transport;
         _maxNoOfChannels = broker.getConnection_sessionCountLimit();
-        _receivedLock = new ReentrantLock();
         _decoder = new BrokerDecoder(this);
         _connectionID = connectionId;
         _logSubject = new ConnectionLogSubject(this);
@@ -545,43 +538,8 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
 
 
     private final byte[] _reusableBytes = new byte[REUSABLE_BYTE_BUFFER_CAPACITY];
-    private final ByteBuffer _reusableByteBuffer = ByteBuffer.wrap(_reusableBytes);
     private final BytesDataOutput _reusableDataOutput = new BytesDataOutput(_reusableBytes);
 
-    private ByteBuffer asByteBuffer(AMQDataBlock block)
-    {
-        final int size = (int) block.getSize();
-
-        final byte[] data;
-
-
-        if(size > REUSABLE_BYTE_BUFFER_CAPACITY)
-        {
-            data= new byte[size];
-        }
-        else
-        {
-
-            data = _reusableBytes;
-        }
-        _reusableDataOutput.setBuffer(data);
-
-        try
-        {
-            block.writePayload(_reusableDataOutput);
-        }
-        catch (IOException e)
-        {
-            throw new ServerScopedRuntimeException(e);
-        }
-
-        final ByteBuffer copy = ByteBuffer.allocate(_reusableDataOutput.length());
-        copy.put(data, 0, _reusableDataOutput.length());
-        copy.flip();
-        return copy;
-    }
-
-
     /**
      * Convenience method that writes a frame to the protocol session. Equivalent to calling
      * getProtocolSession().write().
@@ -1969,11 +1927,6 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
         return _reference;
     }
 
-    public Lock getReceivedLock()
-    {
-        return _receivedLock;
-    }
-
     @Override
     public long getLastReadTime()
     {
@@ -2095,6 +2048,8 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
     @Override
     public void processPending()
     {
+
+
         while(_asyncTaskList.peek() != null)
         {
             Action<? super AMQProtocolEngine> asyncAction = _asyncTaskList.poll();
@@ -2103,7 +2058,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
 
         for (AMQSessionModel session : getSessionModels())
         {
-            session.processPendingMessages();
+            session.processPending();
         }
     }
 
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
index 2bf2fc6d27..a2113de8ea 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
@@ -515,6 +515,7 @@ public abstract class ConsumerTarget_0_8 extends AbstractConsumerTarget implemen
         if (isAutoClose())
         {
             _needToClose.set(true);
+            getChannel().getConnection().notifyWork();
         }
     }
 
@@ -531,8 +532,6 @@ public abstract class ConsumerTarget_0_8 extends AbstractConsumerTarget implemen
     public void flushBatched()
     {
         _channel.getConnection().setDeferFlush(false);
-
-        _channel.getConnection().notifyWork();
     }
 
     protected void addUnacknowledgedMessage(MessageInstance entry)