optimize range by score

src/datastruct/sortedset/border.go

@@ -24,6 +24,8 @@ type ScoreBorder struct {
     Exclude bool
 }
 
+// if max.greater(score) then the score is within the upper border
+// do not use min.greater()
 func (border *ScoreBorder)greater(value float64)bool {
     if border.Inf == negativeInf {
         return false

src/datastruct/sortedset/skiplist.go

@@ -131,7 +131,7 @@ func (skiplist *skiplist)insert(member string, score float64)*Node {
 
 /*
  * param node: node to delete
- * param update: backward node (of target) or last node of each level
+ * param update: backward node (of target)
  */
 func (skiplist *skiplist) removeNode(node *Node, update []*Node) {
     for i := int16(0); i < skiplist.level; i++ {
@@ -223,6 +223,62 @@ func (skiplist *skiplist) getByRank(rank int64)*Node {
     return nil
 }
 
+func (skiplist *skiplist) hasInRange(min *ScoreBorder, max *ScoreBorder) bool {
+    // min & max = empty
+    if min.Value > max.Value || (min.Value == max.Value && (min.Exclude || max.Exclude)) {
+        return false
+    }
+    // min > tail
+    n := skiplist.tail
+    if n == nil || !min.less(n.Score) {
+        return false
+    }
+    // max < head
+    n = skiplist.header.level[0].forward
+    if n == nil || !max.greater(n.Score) {
+        return false
+    }
+    return true
+}
+
+func (skiplist *skiplist) getFirstInScoreRange(min *ScoreBorder, max *ScoreBorder) *Node {
+    if !skiplist.hasInRange(min, max) {
+        return nil
+    }
+    n := skiplist.header
+    // scan from top level
+    for level := skiplist.level - 1; level >= 0; level-- {
+        // if forward is not in range than move forward
+        for n.level[level].forward != nil && !min.less(n.level[level].forward.Score) {
+            n = n.level[level].forward
+        }
+    }
+    /* This is an inner range, so the next node cannot be NULL. */
+    n = n.level[0].forward
+    if !max.greater(n.Score) {
+        return nil
+    }
+    return n
+}
+
+func (skiplist *skiplist) getLastInScoreRange(min *ScoreBorder, max *ScoreBorder) *Node {
+    if !skiplist.hasInRange(min, max) {
+        return nil
+    }
+    n := skiplist.header
+    // scan from top level
+    for level := skiplist.level - 1; level >= 0; level-- {
+        for n.level[level].forward != nil && max.greater(n.level[level].forward.Score) {
+            n = n.level[level].forward
+        }
+    }
+    n = n.level[0].forward
+    if !min.less(n.Score) {
+        return nil
+    }
+    return n
+}
+
 /*
  * return removed elements
  */

src/datastruct/sortedset/sortedset.go

@@ -152,6 +152,42 @@ func (sortedSet *SortedSet) Count(min *ScoreBorder, max *ScoreBorder)int64 {
     return i
 }
 
+func (sortedSet *SortedSet) ForEachByScore(min *ScoreBorder, max *ScoreBorder, offset int64, limit int64, desc bool, consumer func(element *Element) bool) {
+    // find start node
+    var node *Node
+    if desc {
+        node = sortedSet.skiplist.getLastInScoreRange(min, max)
+    } else {
+        node = sortedSet.skiplist.getFirstInScoreRange(min, max)
+    }
+
+    for node != nil && offset > 0 {
+        if desc {
+            node = node.backward
+        } else {
+            node = node.level[0].forward
+        }
+        offset--
+    }
+
+    // A negative limit returns all elements from the offset
+    for i := 0; (i < int(limit) || limit < 0) && node != nil; i++ {
+        if !consumer(&node.Element) {
+            break
+        }
+        if desc {
+            node = node.backward
+        } else {
+            node = node.level[0].forward
+        }
+        gtMin := min.less(node.Element.Score) // greater than min
+        ltMax := max.greater(node.Element.Score)
+        if !gtMin || !ltMax {
+            break // break through score border
+        }
+    }
+}
+
 /*
  * param limit: <0 means no limit
  */
@@ -160,23 +196,8 @@ func (sortedSet *SortedSet) RangeByScore(min *ScoreBorder, max *ScoreBorder, off
         return make([]*Element, 0)
     }
     slice := make([]*Element, 0)
-    var skipped int64 = 0
+    sortedSet.ForEachByScore(min, max, offset, limit, desc, func(element *Element) bool {
-    sortedSet.ForEach(0, sortedSet.Len(), desc, func(element *Element)bool {
+        slice = append(slice, element)
-        gtMin := min.less(element.Score) // greater than min
-        ltMax := max.greater(element.Score) // less than max
-        if gtMin && ltMax { // in score range
-            if skipped < offset {
-                skipped++
-                return true
-            }
-            slice = append(slice, element)
-            if len(slice) == int(limit) { // reach limit
-                return false
-            }
-        }
-        if (desc && !gtMin) || (!desc && !ltMax) {
-            return false // break through score border
-        }
         return true
     })
     return slice

src/db/db.go

@@ -24,7 +24,7 @@ const (
 	dataDictSize = 1 << 16
 	ttlDictSize  = 1 << 10
 	lockerSize   = 128
-	aofQueueSize = 2 << 10
+	aofQueueSize = 1 << 16
 )
 
 type extra struct {