QPID-3886 Committing a patch by Kevin Conner.

The purpose of the patch is to improve the efficiency of processing
known-complete (command id ranges).

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1298536 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 72 insertions, 11 deletions

diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Range.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Range.java
index 8380744024..413ec8e8fd 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Range.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Range.java
@@ -122,6 +122,11 @@ public abstract class Range implements RangeSet
         throw new UnsupportedOperationException();
     }
 
+    public void subtract(RangeSet rangeSet)
+    {
+        throw new UnsupportedOperationException();
+    }
+
     public RangeSet copy()
     {
         RangeSet rangeSet = RangeSetFactory.createRangeSet();
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/RangeSet.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/RangeSet.java
index 8737df88d6..19990a4610 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/RangeSet.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/RangeSet.java
@@ -49,6 +49,8 @@ public interface RangeSet extends Iterable<Range>
 
     void add(int value);
 
+    void subtract(final RangeSet other);
+
     void clear();
 
     RangeSet copy();
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/RangeSetImpl.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/RangeSetImpl.java
index 3e24c10a06..adf18e2920 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/RangeSetImpl.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/RangeSetImpl.java
@@ -150,6 +150,68 @@ public class RangeSetImpl implements RangeSet
         ranges.clear();
     }
 
+    public void subtract(final RangeSet other)
+    {
+        final Iterator<Range> otherIter = other.iterator() ;
+        if (otherIter.hasNext())
+        {
+            Range otherRange = otherIter.next();
+            final ListIterator<Range> iter = ranges.listIterator() ;
+            if (iter.hasNext())
+            {
+                Range range = iter.next();
+                do
+                {
+                    if (otherRange.getUpper() < range.getLower())
+                    {
+                        otherRange = nextRange(otherIter) ;
+                    }
+                    else if (range.getUpper() < otherRange.getLower())
+                    {
+                        range = nextRange(iter) ;
+                    }
+                    else
+                    {
+                        final boolean first = range.getLower() < otherRange.getLower() ;
+                        final boolean second = otherRange.getUpper() < range.getUpper() ;
+
+                        if (first)
+                        {
+                            iter.set(Range.newInstance(range.getLower(), otherRange.getLower()-1)) ;
+                            if (second)
+                            {
+                                iter.add(Range.newInstance(otherRange.getUpper()+1, range.getUpper())) ;
+                                iter.previous() ;
+                                range = iter.next() ;
+                            }
+                            else
+                            {
+                                range = nextRange(iter) ;
+                            }
+                        }
+                        else if (second)
+                        {
+                            range = Range.newInstance(otherRange.getUpper()+1, range.getUpper()) ;
+                            iter.set(range) ;
+                            otherRange = nextRange(otherIter) ;
+                        }
+                        else
+                        {
+                            iter.remove() ;
+                            range = nextRange(iter) ;
+                        }
+                    }
+                }
+                while ((otherRange != null) && (range != null)) ;
+            }
+        }
+    }
+
+    private Range nextRange(final Iterator<Range> iter)
+    {
+        return (iter.hasNext() ? iter.next() : null) ;
+    }
+
     public RangeSet copy()
     {
         return new org.apache.qpid.transport.RangeSetImpl(this);
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
index 0cdffeee80..110c73f718 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
@@ -513,20 +513,12 @@ public class Session extends SessionInvoker
 
     void knownComplete(RangeSet kc)
     {
-        synchronized (processedLock)
+        if (kc.size() > 0)
         {
-            RangeSet newProcessed = RangeSetFactory.createRangeSet();
-            for (Range pr : processed)
+            synchronized (processedLock)
             {
-                for (Range kr : kc)
-                {
-                    for (Range r : pr.subtract(kr))
-                    {
-                        newProcessed.add(r);
-                    }
-                }
+                processed.subtract(kc) ;
             }
-            this.processed = newProcessed;
         }
     }