Make some synchronisation changes and explain them.

Author: Simon MacMullen

src/com/rabbitmq/client/impl/ChannelN.java

@@ -1188,14 +1188,19 @@ public AMQCommand rpc(Method method) throws IOException {
 
     @Override
     public void enqueueRpc(RpcContinuation k) {
-        super.enqueueRpc(k);
-        dispatcher.setLimited(false);
+        synchronized (_channelMutex) {
+            super.enqueueRpc(k);
+            dispatcher.setLimited(false);
+        }
     }
 
     @Override
     public RpcContinuation nextOutstandingRpc() {
-        dispatcher.setLimited(true);
-        return super.nextOutstandingRpc();
+        synchronized (_channelMutex) {
+            RpcContinuation res = super.nextOutstandingRpc();
+            if (res != null) dispatcher.setLimited(true);
+            return res;
+        }
     }
 
     private void handleAckNack(long seqNo, boolean multiple, boolean nack) {

src/com/rabbitmq/client/impl/WorkPool.java

@@ -71,7 +71,19 @@ public class WorkPool<K, W> {
 
     // This is like a LinkedBlockingQueue of limited length except you can turn the limit
     // on and off. And it only has the methods we need.
-    // TODO synchronised how?
+    //
+    // This class is partly synchronised because:
+    //
+    // a) we cannot make put(T) synchronised as it may block indefinitely. Therefore we
+    //    only lock before modifying the list.
+    // b) we don't want to make setLimited() synchronised as it is called frequently by
+    //    the channel.
+    // c) anyway the issue with setLimited() is not that it be synchronised itself but
+    //    that calls to it should alternate between false and true. We assert this, but
+    //    it should not be able to go wrong because the RPC calls in AMQChannel and
+    //    ChannelN are all protected by the _channelMutex; we can't have more than one
+    //    outstanding RPC or finish the same RPC twice.
+
     private static class WorkQueue<T> {
         private Semaphore semaphore;
         private LinkedList<T> list;
@@ -90,10 +102,12 @@ public void put(T t) throws InterruptedException {
                 assert !semaphore.hasQueuedThreads();
                 semaphore.acquire();
             }
-            list.add(t);
+            synchronized (this) {
+                list.add(t);
+            }
         }
 
-        public T poll() {
+        public synchronized T poll() {
             T res = list.poll();
 
             if (list.size() <= maxLengthWhenLimited && semaphore.hasQueuedThreads()) {
@@ -104,6 +118,7 @@ public T poll() {
         }
 
         public void setLimited(boolean limited) {
+            assert this.limited != limited;
             this.limited = limited;
             if (!limited && semaphore.hasQueuedThreads()) {
                 semaphore.release();