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

5 files changed, 186 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;
         }
     }
 
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/RangeSetTest.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/RangeSetTest.java
index 56dbaf5535..14589eb541 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/transport/RangeSetTest.java
+++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/RangeSetTest.java
@@ -27,6 +27,7 @@ import static org.apache.qpid.util.Serial.eq;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 
 /**
@@ -236,4 +237,117 @@ public class RangeSetTest extends TestCase
         assertEquals(d.getUpper(), 10);
     }
 
+    public void testSetSubtract1()
+    {
+        final RangeSet orig = createRangeSet(0, 10) ;
+        final RangeSet update = createRangeSet(3, 15) ;
+        orig.subtract(update) ;
+        checkRange(orig, 0, 2) ;
+    }
+
+    public void testSetSubtract2()
+    {
+        final RangeSet orig = createRangeSet(0, 10) ;
+        final RangeSet update = createRangeSet(3, 10) ;
+        orig.subtract(update) ;
+        checkRange(orig, 0, 2) ;
+    }
+
+    public void testSetSubtract3()
+    {
+        final RangeSet orig = createRangeSet(0, 10) ;
+        final RangeSet update = createRangeSet(3, 4) ;
+        orig.subtract(update) ;
+        checkRange(orig, 0, 2, 5, 10) ;
+    }
+
+    public void testSetSubtract4()
+    {
+        final RangeSet orig = createRangeSet(3, 15) ;
+        final RangeSet update = createRangeSet(0, 10) ;
+        orig.subtract(update) ;
+        checkRange(orig, 11, 15) ;
+    }
+
+    public void testSetSubtract5()
+    {
+        final RangeSet orig = createRangeSet(3, 10) ;
+        final RangeSet update = createRangeSet(0, 10) ;
+        orig.subtract(update) ;
+        checkRange(orig) ;
+    }
+
+    public void testSetSubtract6()
+    {
+        final RangeSet orig = createRangeSet(3, 10) ;
+        final RangeSet update = createRangeSet(0, 15) ;
+        orig.subtract(update) ;
+        checkRange(orig) ;
+    }
+
+    public void testSetSubtract7()
+    {
+        final RangeSet orig = createRangeSet(0, 10) ;
+        final RangeSet update = createRangeSet(0, 15) ;
+        orig.subtract(update) ;
+        checkRange(orig) ;
+    }
+
+    public void testSetSubtract8()
+    {
+        final RangeSet orig = createRangeSet(0, 15) ;
+        final RangeSet update = createRangeSet(0, 10) ;
+        orig.subtract(update) ;
+        checkRange(orig, 11, 15) ;
+    }
+
+    public void testSetSubtract9()
+    {
+        final RangeSet orig = createRangeSet(0, 15, 20, 30) ;
+        final RangeSet update = createRangeSet(2, 3, 5, 6, 8, 9, 22, 23, 27, 28) ;
+        orig.subtract(update) ;
+        checkRange(orig, 0, 1, 4, 4, 7, 7, 10, 15, 20, 21, 24, 26, 29, 30) ;
+    }
+
+    public void testSetSubtract10()
+    {
+        final RangeSet orig = createRangeSet(0, 15, 20, 30) ;
+        final RangeSet update = createRangeSet(0, 2, 4, 6, 10, 22, 24, 24, 27, 30) ;
+        orig.subtract(update) ;
+        checkRange(orig, 3, 3, 7, 9, 23, 23, 25, 26) ;
+    }
+
+    public void testSetSubtract11()
+    {
+        final RangeSet orig = createRangeSet(0, 2, 4, 6, 10, 22, 24, 24, 27, 30) ;
+        final RangeSet update = createRangeSet(0, 2, 4, 6, 10, 22, 24, 24, 27, 30) ;
+        orig.subtract(update) ;
+        checkRange(orig) ;
+    }
+    
+    private RangeSet createRangeSet(int ... bounds)
+    {
+        RangeSet set = RangeSetFactory.createRangeSet();
+        final int length = (bounds == null ? 0 : bounds.length) ;
+        int count = 0 ;
+        while(count < length)
+        {
+            set.add(bounds[count++], bounds[count++]) ;
+        }
+        return set ;
+    }
+    
+    private void checkRange(final RangeSet rangeSet, int ... bounds)
+    {
+        final int length = (bounds == null ? 0 : bounds.length) ;
+        assertEquals("Range count", length/2, rangeSet.size()) ;
+        final Iterator<Range> iter = rangeSet.iterator() ;
+        int count = 0 ;
+        while(count < length)
+        {
+            final Range range = iter.next() ;
+            assertEquals("Range lower", bounds[count++], range.getLower()) ;
+            assertEquals("Range upper", bounds[count++], range.getUpper()) ;
+        }
+    }
 }