From 8ba206115a18239e43453abd11186648b04d16ef Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Wed, 23 Sep 2020 14:39:14 +0800
Subject: [PATCH] [td-1575]

---
 src/util/inc/tskiplist.h |  1 +
 src/util/src/tskiplist.c | 35 ++++++++++++++++++++++++++++++-----
 2 files changed, 31 insertions(+), 5 deletions(-)

diff --git a/src/util/inc/tskiplist.h b/src/util/inc/tskiplist.h
index 4ba620dce0..a14a856561 100644
--- a/src/util/inc/tskiplist.h
+++ b/src/util/inc/tskiplist.h
@@ -136,6 +136,7 @@ typedef struct SSkipListIterator {
   SSkipListNode *cur;
   int32_t        step;          // the number of nodes that have been checked already
   int32_t        order;         // order of the iterator
+  SSkipListNode *next;          // next points to the true qualified node in skip list
 } SSkipListIterator;
 
 /**
diff --git a/src/util/src/tskiplist.c b/src/util/src/tskiplist.c
index 303c2440bf..bacdaef6c8 100644
--- a/src/util/src/tskiplist.c
+++ b/src/util/src/tskiplist.c
@@ -79,9 +79,12 @@ static SSkipListIterator* doCreateSkipListIterator(SSkipList *pSkipList, int32_t
 
 // when order is TSDB_ORDER_ASC, return the last node with key less than val
 // when order is TSDB_ORDER_DESC, return the first node with key large than val
-static SSkipListNode* getPriorNode(SSkipList* pSkipList, const char* val, int32_t order) {
+static SSkipListNode* getPriorNode(SSkipList* pSkipList, const char* val, int32_t order, SSkipListNode** pCur) {
   __compar_fn_t comparFn = pSkipList->comparFn;
   SSkipListNode *pNode = NULL;
+  if (pCur != NULL) {
+    *pCur = NULL;
+  }
 
   if (order == TSDB_ORDER_ASC) {
     pNode = pSkipList->pHead;
@@ -93,6 +96,9 @@ static SSkipListNode* getPriorNode(SSkipList* pSkipList, const char* val, int32_
           pNode = p;
           p = SL_GET_FORWARD_POINTER(p, i);
         } else {
+          if (pCur != NULL) {
+            *pCur = p;
+          }
           break;
         }
       }
@@ -107,6 +113,9 @@ static SSkipListNode* getPriorNode(SSkipList* pSkipList, const char* val, int32_
           pNode = p;
           p = SL_GET_BACKWARD_POINTER(p, i);
         } else {
+          if (pCur != NULL) {
+            *pCur = p;
+          }
           break;
         }
       }
@@ -295,7 +304,7 @@ SArray* tSkipListGet(SSkipList *pSkipList, SSkipListKey key) {
     pthread_rwlock_wrlock(pSkipList->lock);
   }
 
-  SSkipListNode* pNode = getPriorNode(pSkipList, key, TSDB_ORDER_ASC);
+  SSkipListNode* pNode = getPriorNode(pSkipList, key, TSDB_ORDER_ASC, NULL);
   while (1) {
     SSkipListNode *p = SL_GET_FORWARD_POINTER(pNode, 0);
     if (p == pSkipList->pTail) {
@@ -452,7 +461,7 @@ uint32_t tSkipListRemove(SSkipList *pSkipList, SSkipListKey key) {
     pthread_rwlock_wrlock(pSkipList->lock);
   }
 
-  SSkipListNode* pNode = getPriorNode(pSkipList, key, TSDB_ORDER_ASC);
+  SSkipListNode* pNode = getPriorNode(pSkipList, key, TSDB_ORDER_ASC, NULL);
   while (1) {
     SSkipListNode *p = SL_GET_FORWARD_POINTER(pNode, 0);
     if (p == pSkipList->pTail) {
@@ -545,7 +554,7 @@ SSkipListIterator *tSkipListCreateIterFromVal(SSkipList* pSkipList, const char*
     pthread_rwlock_rdlock(pSkipList->lock);
   }
 
-  iter->cur = getPriorNode(pSkipList, val, order);
+  iter->cur = getPriorNode(pSkipList, val, order, &iter->next);
 
   if (pSkipList->lock) {
     pthread_rwlock_unlock(pSkipList->lock);
@@ -567,8 +576,22 @@ bool tSkipListIterNext(SSkipListIterator *iter) {
   
   if (iter->order == TSDB_ORDER_ASC) {  // ascending order iterate
     iter->cur = SL_GET_FORWARD_POINTER(iter->cur, 0);
+
+    // a new node is inserted into between iter->cur and iter->next, ignore it
+    if (iter->cur != iter->next && (iter->next != NULL)) {
+      iter->cur = iter->next;
+    }
+
+    iter->next = SL_GET_FORWARD_POINTER(iter->cur, 0);
   } else { // descending order iterate
     iter->cur = SL_GET_BACKWARD_POINTER(iter->cur, 0);
+
+    // a new node is inserted into between iter->cur and iter->next, ignore it
+    if (iter->cur != iter->next && (iter->next != NULL)) {
+      iter->cur = iter->next;
+    }
+
+    iter->next = SL_GET_BACKWARD_POINTER(iter->cur, 0);
   }
   
   if (pSkipList->lock) {
@@ -715,9 +738,11 @@ SSkipListIterator* doCreateSkipListIterator(SSkipList *pSkipList, int32_t order)
   iter->order = order;
   if(order == TSDB_ORDER_ASC) {
     iter->cur = pSkipList->pHead;
+    iter->next = SL_GET_FORWARD_POINTER(iter->cur, 0);
   } else {
     iter->cur = pSkipList->pTail;
+    iter->next = SL_GET_BACKWARD_POINTER(iter->cur, 0);
   }
-  
+
   return iter;
 }
\ No newline at end of file
-- 
GitLab