change applications

src/query/src/qResultbuf.c

@@ -420,9 +420,8 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf) {
   unlink(pResultBuf->path);
   tfree(pResultBuf->path);
 
-  SHashMutableIterator* iter = taosHashCreateIter(pResultBuf->groupSet);
-  while(taosHashIterNext(iter)) {
-    SArray** p = (SArray**) taosHashIterGet(iter);
+  SArray** p = taosHashIterate(pResultBuf->groupSet, NULL);
+  while(p) {
     size_t n = taosArrayGetSize(*p);
     for(int32_t i = 0; i < n; ++i) {
       SPageInfo* pi = taosArrayGetP(*p, i);
@@ -431,10 +430,9 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf) {
     }
 
     taosArrayDestroy(*p);
+    p = taosHashIterate(pResultBuf->groupSet, p);
   }
 
-  taosHashDestroyIter(iter);
-
   tdListFree(pResultBuf->lruList);
   taosArrayDestroy(pResultBuf->emptyDummyIdList);
   taosHashCleanup(pResultBuf->groupSet);

src/util/inc/hash.h

@@ -31,16 +31,17 @@ extern "C" {
 typedef void (*_hash_free_fn_t)(void *param);
 
 typedef struct SHashNode {
-//  char             *key;
   struct SHashNode *next;
   uint32_t          hashVal;  // the hash value of key
   uint32_t          keyLen;   // length of the key
-//  char             *data;
+  uint32_t          dataLen;  // length of data
+  int8_t            count;    // reference count
+  char              data[];
 } SHashNode;
 
-#define GET_HASH_NODE_KEY(_n)  ((char*)(_n) + sizeof(SHashNode))
-#define GET_HASH_NODE_DATA(_n) ((char*)(_n) + sizeof(SHashNode) + (_n)->keyLen)
-
+#define GET_HASH_NODE_KEY(_n)  ((char*)(_n) + sizeof(SHashNode) + (_n)->dataLen)
+#define GET_HASH_NODE_DATA(_n) ((char*)(_n) + sizeof(SHashNode))
+#define GET_HASH_PNODE(_n) ((char*)(_n) - sizeof(SHashNode));
 typedef enum SHashLockTypeE {
   HASH_NO_LOCK     = 0,
   HASH_ENTRY_LOCK  = 1,
@@ -65,15 +66,6 @@ typedef struct SHashObj {
   SArray         *pMemBlock;    // memory block allocated for SHashEntry
 } SHashObj;
 
-typedef struct SHashMutableIterator {
-  SHashObj  *pHashObj;
-  int32_t    entryIndex;
-  SHashNode *pCur;
-  SHashNode *pNext;           // current node can be deleted for mutable iterator, so keep the next one before return current
-  size_t     numOfChecked;    // already check number of elements in hash table
-  size_t     numOfEntries;    // number of entries while the iterator is created
-} SHashMutableIterator;
-
 /**
  * init the hash table
  *
@@ -142,33 +134,9 @@ int32_t taosHashCondTraverse(SHashObj *pHashObj, bool (*fp)(void *, void *), voi
  */
 void taosHashCleanup(SHashObj *pHashObj);
 
-/**
- *
- * @param pHashObj
- * @return
- */
-SHashMutableIterator* taosHashCreateIter(SHashObj *pHashObj);
-
-/**
- *
- * @param iter
- * @return
- */
-bool taosHashIterNext(SHashMutableIterator *iter);
-
-/**
- *
- * @param iter
- * @return
- */
-void *taosHashIterGet(SHashMutableIterator *iter);
-
-/**
- *
- * @param iter
- * @return
- */
-void* taosHashDestroyIter(SHashMutableIterator* iter);
+/*
+void *SHashMutableIterator* taosHashCreateIter(SHashObj *pHashObj, void *);
+*/
 
 /**
  *
@@ -179,6 +147,9 @@ int32_t taosHashGetMaxOverflowLinkLength(const SHashObj *pHashObj);
 
 size_t taosHashGetMemSize(const SHashObj *pHashObj);
 
+void *taosHashIterate(SHashObj *pHashObj, void *p);
+void  taosHashCancelIterate(SHashObj *pHashObj, void *p);
+
 #ifdef __cplusplus
 }
 #endif

src/util/src/hash.c

@@ -114,15 +114,25 @@ static SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *p
  * @param dsize   size of actual data
  * @return        hash node
  */
-static FORCE_INLINE SHashNode *doUpdateHashNode(SHashEntry* pe, SHashNode* prev, SHashNode *pNode, SHashNode *pNewNode) {
+static FORCE_INLINE SHashNode *doUpdateHashNode(SHashObj *pHashObj, SHashEntry* pe, SHashNode* prev, SHashNode *pNode, SHashNode *pNewNode) {
   assert(pNode->keyLen == pNewNode->keyLen);
+
+  pNode->count--;
   if (prev != NULL) {
     prev->next = pNewNode;
   } else {
     pe->next = pNewNode;
   }
 
-  pNewNode->next = pNode->next;
+  if (pNode->count <= 0) {
+    pNewNode->next = pNode->next;
+    DO_FREE_HASH_NODE(pNode);
+  } else {
+    pNewNode->next = pNode;    
+    pe->num++;
+    atomic_add_fetch_64(&pHashObj->size, 1);
+  }
+  
   return pNewNode;
 }
 
@@ -139,7 +149,6 @@ static void pushfrontNodeInEntryList(SHashEntry *pEntry, SHashNode *pNode);
  * @param pIter
  * @return
  */
-static SHashNode *getNextHashNode(SHashMutableIterator *pIter);
 
 SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTypeE type) {
   if (capacity == 0 || fn == NULL) {
@@ -244,8 +253,7 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *da
   } else {
     // not support the update operation, return error
     if (pHashObj->enableUpdate) {
-      doUpdateHashNode(pe, prev, pNode, pNewNode);
-      DO_FREE_HASH_NODE(pNode);
+      doUpdateHashNode(pHashObj, pe, prev, pNode, pNewNode);
     } else {
       DO_FREE_HASH_NODE(pNewNode);
     }
@@ -335,22 +343,10 @@ int32_t taosHashRemoveWithData(SHashObj *pHashObj, const void *key, size_t keyLe
   int32_t     slot = HASH_INDEX(hashVal, pHashObj->capacity);
   SHashEntry *pe = pHashObj->hashList[slot];
 
-  // no data, return directly
-  if (pe->num == 0) {
-    __rd_unlock(&pHashObj->lock, pHashObj->type);
-    return -1;
-  }
-
   if (pHashObj->type == HASH_ENTRY_LOCK) {
     taosWLockLatch(&pe->latch);
   }
 
-  if (pe->num == 0) {
-    assert(pe->next == NULL);
-  } else {
-    assert(pe->next != NULL);
-  }
-
   // double check after locked
   if (pe->num == 0) {
     assert(pe->next == NULL);
@@ -360,37 +356,36 @@ int32_t taosHashRemoveWithData(SHashObj *pHashObj, const void *key, size_t keyLe
     return -1;
   }
 
+  int code = -1;
   SHashNode *pNode = pe->next;
-  SHashNode *pRes = NULL;
+  SHashNode *prevNode = NULL;
 
-  // remove it
-  if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0)) {
-    pe->num -= 1;
-    pRes = pNode;
-    pe->next = pNode->next;
-  } else {
-    while (pNode->next != NULL) {
-      if (((pNode->next)->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY((pNode->next)), key, keyLen) == 0)) {
-        assert((pNode->next)->hashVal == hashVal);
-        break;
-      }
+  while (pNode) {
+    if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0)) 
+      break;
 
-      pNode = pNode->next;
-    }
+    prevNode = pNode;
+    pNode = pNode->next;
+  }
 
+  if (pNode) {
+    code = 0;  // it is found
 
-    if (pNode->next != NULL) {
-      pe->num -= 1;
-      pRes = pNode->next;
-      pNode->next = pNode->next->next;
-    }
-  }
+    pNode->count--;
+    if (pNode->count <= 0) {
+      if (prevNode) {
+        prevNode->next = pNode->next;
+      } else {
+        pe->next = pNode->next;
+      }
+   
+      if (data) memcpy(data, GET_HASH_NODE_DATA(pNode), dsize);
 
-  if (pe->num == 0) {
-    assert(pe->next == NULL);
-  } else {
-    assert(pe->next != NULL);
-  }
+      pe->num--;
+      atomic_sub_fetch_64(&pHashObj->size, 1);
+      FREE_HASH_NODE(pHashObj, pNode);
+    }
+  } 
 
   if (pHashObj->type == HASH_ENTRY_LOCK) {
     taosWUnLockLatch(&pe->latch);
@@ -398,17 +393,7 @@ int32_t taosHashRemoveWithData(SHashObj *pHashObj, const void *key, size_t keyLe
 
   __rd_unlock(&pHashObj->lock, pHashObj->type);
 
-  if (data != NULL && pRes != NULL) {
-    memcpy(data, GET_HASH_NODE_DATA(pRes), dsize);
-  }
-
-  if (pRes != NULL) {
-    atomic_sub_fetch_64(&pHashObj->size, 1);
-    FREE_HASH_NODE(pHashObj, pRes);
-    return 0;
-  } else {
-    return -1;
-  }
+  return code;
 }
 
 int32_t taosHashCondTraverse(SHashObj *pHashObj, bool (*fp)(void *, void *), void *param) {
@@ -531,98 +516,6 @@ void taosHashCleanup(SHashObj *pHashObj) {
   free(pHashObj);
 }
 
-SHashMutableIterator *taosHashCreateIter(SHashObj *pHashObj) {
-  SHashMutableIterator *pIter = calloc(1, sizeof(SHashMutableIterator));
-  if (pIter == NULL) {
-    return NULL;
-  }
-
-  pIter->pHashObj = pHashObj;
-
-  // keep it in local variable, in case the resize operation expand the size
-  pIter->numOfEntries = pHashObj->capacity;
-  return pIter;
-}
-
-bool taosHashIterNext(SHashMutableIterator *pIter) {
-  if (pIter == NULL) {
-    return false;
-  }
-
-  size_t size = taosHashGetSize(pIter->pHashObj);
-  if (size == 0) {
-    return false;
-  }
-
-  // check the first one
-  if (pIter->numOfChecked == 0) {
-    assert(pIter->pCur == NULL && pIter->pNext == NULL);
-
-    while (1) {
-      SHashEntry *pEntry = pIter->pHashObj->hashList[pIter->entryIndex];
-      if (pEntry->num == 0) {
-        assert(pEntry->next == NULL);
-
-        pIter->entryIndex++;
-        continue;
-      }
-
-      if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
-        taosRLockLatch(&pEntry->latch);
-      }
-
-      pIter->pCur = pEntry->next;
-
-      if (pIter->pCur->next) {
-        pIter->pNext = pIter->pCur->next;
-
-        if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
-          taosRUnLockLatch(&pEntry->latch);
-        }
-      } else {
-        if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
-          taosRUnLockLatch(&pEntry->latch);
-        }
-
-        pIter->pNext = getNextHashNode(pIter);
-      }
-
-      break;
-    }
-
-    pIter->numOfChecked++;
-    return true;
-  } else {
-    assert(pIter->pCur != NULL);
-    if (pIter->pNext) {
-      pIter->pCur = pIter->pNext;
-    } else {  // no more data in the hash list
-      return false;
-    }
-
-    pIter->numOfChecked++;
-
-    if (pIter->pCur->next) {
-      pIter->pNext = pIter->pCur->next;
-    } else {
-      pIter->pNext = getNextHashNode(pIter);
-    }
-
-    return true;
-  }
-}
-
-void *taosHashIterGet(SHashMutableIterator *iter) { return (iter == NULL) ? NULL : GET_HASH_NODE_DATA(iter->pCur); }
-
-void *taosHashDestroyIter(SHashMutableIterator *iter) {
-  if (iter == NULL) {
-    return NULL;
-  }
-
-  free(iter);
-  return NULL;
-}
-
 // for profile only
 int32_t taosHashGetMaxOverflowLinkLength(const SHashObj *pHashObj) {
   if (pHashObj == NULL || pHashObj->size == 0) {
@@ -759,6 +652,8 @@ SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *pData, s
 
   pNewNode->keyLen = (uint32_t)keyLen;
   pNewNode->hashVal = hashVal;
+  pNewNode->dataLen = dsize;
+  pNewNode->count = 1;
 
   memcpy(GET_HASH_NODE_DATA(pNewNode), pData, dsize);
   memcpy(GET_HASH_NODE_KEY(pNewNode), key, keyLen);
@@ -775,39 +670,126 @@ void pushfrontNodeInEntryList(SHashEntry *pEntry, SHashNode *pNode) {
   pEntry->num += 1;
 }
 
-SHashNode *getNextHashNode(SHashMutableIterator *pIter) {
-  assert(pIter != NULL);
+size_t taosHashGetMemSize(const SHashObj *pHashObj) {
+  if (pHashObj == NULL) {
+    return 0;
+  }
 
-  pIter->entryIndex++;
-  SHashNode *p = NULL;
+  return (pHashObj->capacity * (sizeof(SHashEntry) + POINTER_BYTES)) + sizeof(SHashNode) * taosHashGetSize(pHashObj) + sizeof(SHashObj);
+}
 
-  while (pIter->entryIndex < pIter->numOfEntries) {
-    SHashEntry *pEntry = pIter->pHashObj->hashList[pIter->entryIndex];
-    if (pEntry->num == 0) {
-      pIter->entryIndex++;
-      continue;
-    }
+// release the pNode, return next pNode, and lock the current entry
+static void *taosHashReleaseNode(SHashObj *pHashObj, void *p, int *slot) {
 
-    if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
-      taosRLockLatch(&pEntry->latch);
+  SHashNode *pOld = (SHashNode *)GET_HASH_PNODE(p);
+  SHashNode *prevNode = NULL;
+
+  *slot = HASH_INDEX(pOld->hashVal, pHashObj->capacity);
+  SHashEntry *pe = pHashObj->hashList[*slot];
+
+  // lock entry
+  if (pHashObj->type == HASH_ENTRY_LOCK) {
+    taosWLockLatch(&pe->latch);
+  }
+
+  SHashNode *pNode = pe->next;
+
+  while (pNode) {
+    if (pNode == pOld) 
+      break;
+
+    prevNode = pNode;
+    pNode = pNode->next;
+  }
+
+  if (pNode) {
+    pNode = pNode->next;
+    pOld->count--;
+    if (pOld->count <=0) {
+      if (prevNode) {
+        prevNode->next = pOld->next;
+      } else {
+        pe->next = pOld->next;
+      }
+   
+      pe->num--;
+      atomic_sub_fetch_64(&pHashObj->size, 1);
+      FREE_HASH_NODE(pHashObj, pOld);
+    } 
+  } else {
+    uError("pNode:%p data:%p is not there!!!", pNode, p);
+  }
+
+  return pNode;
+}
+
+void *taosHashIterate(SHashObj *pHashObj, void *p) {
+  if (pHashObj == NULL) return NULL; 
+
+  int  slot = 0;
+  char *data = NULL;
+
+  // only add the read lock to disable the resize process
+  __rd_lock(&pHashObj->lock, pHashObj->type);
+
+  SHashNode *pNode = NULL;
+  if (p) {
+    pNode = taosHashReleaseNode(pHashObj, p, &slot);
+    if (pNode == NULL) {
+      SHashEntry *pe = pHashObj->hashList[slot];
+      if (pHashObj->type == HASH_ENTRY_LOCK) {
+        taosWUnLockLatch(&pe->latch);
+      } 
+
+      slot = slot + 1;
     }
+  }
 
-    p = pEntry->next;
+  if (pNode == NULL) {
+    for (; slot < pHashObj->capacity; ++slot) {
+      SHashEntry *pe = pHashObj->hashList[slot];
 
-    if (pIter->pHashObj->type == HASH_ENTRY_LOCK) {
-      taosRUnLockLatch(&pEntry->latch);
+      // lock entry
+      if (pHashObj->type == HASH_ENTRY_LOCK) {
+        taosWLockLatch(&pe->latch);
+      }
+
+      pNode = pe->next;
+      if (pNode) break;
+
+      if (pHashObj->type == HASH_ENTRY_LOCK) {
+        taosWUnLockLatch(&pe->latch);
+      } 
     }
+  }
 
-    return p;
+  if (pNode) {
+    SHashEntry *pe = pHashObj->hashList[slot];
+    pNode->count++;
+    data = GET_HASH_NODE_DATA(pNode);
+    if (pHashObj->type == HASH_ENTRY_LOCK) {
+      taosWUnLockLatch(&pe->latch);
+    } 
   }
 
-  return NULL;
+  __rd_unlock(&pHashObj->lock, pHashObj->type);
+  return data;
+
 }
 
-size_t taosHashGetMemSize(const SHashObj *pHashObj) {
-  if (pHashObj == NULL) {
-    return 0;
-  }
+void taosHashCancelIterate(SHashObj *pHashObj, void *p) {
+  if (pHashObj == NULL || p == NULL) return;
 
-  return (pHashObj->capacity * (sizeof(SHashEntry) + POINTER_BYTES)) + sizeof(SHashNode) * taosHashGetSize(pHashObj) + sizeof(SHashObj);
+  // only add the read lock to disable the resize process
+  __rd_lock(&pHashObj->lock, pHashObj->type);
+
+  int slot;
+  taosHashReleaseNode(pHashObj, p, &slot);
+
+  SHashEntry *pe = pHashObj->hashList[slot];
+  if (pHashObj->type == HASH_ENTRY_LOCK) {
+    taosWUnLockLatch(&pe->latch);
+  } 
+
+  __rd_unlock(&pHashObj->lock, pHashObj->type);
 }

src/util/src/tkvstore.c

@@ -529,7 +529,7 @@ static int tdRestoreKVStore(SKVStore *pStore) {
   void *                buf = NULL;
   int64_t               maxBufSize = 0;
   SKVRecord             rInfo = {0};
-  SHashMutableIterator *pIter = NULL;
+  SKVRecord            *pRecord = NULL;
 
   ASSERT(TD_KVSTORE_HEADER_SIZE == lseek(pStore->fd, 0, SEEK_CUR));
   ASSERT(pStore->info.size == TD_KVSTORE_HEADER_SIZE);
@@ -582,16 +582,8 @@ static int tdRestoreKVStore(SKVStore *pStore) {
     goto _err;
   }
 
-  pIter = taosHashCreateIter(pStore->map);
-  if (pIter == NULL) {
-    uError("failed to create hash iter while opening KV store %s", pStore->fname);
-    terrno = TSDB_CODE_COM_OUT_OF_MEMORY;
-    goto _err;
-  }
-
-  while (taosHashIterNext(pIter)) {
-    SKVRecord *pRecord = taosHashIterGet(pIter);
-
+  pRecord = taosHashIterate(pStore->map, NULL);
+  while (pRecord) {
     if (lseek(pStore->fd, (off_t)(pRecord->offset + sizeof(SKVRecord)), SEEK_SET) < 0) {
       uError("failed to lseek file %s since %s, offset %" PRId64, pStore->fname, strerror(errno), pRecord->offset);
       terrno = TAOS_SYSTEM_ERROR(errno);
@@ -613,16 +605,17 @@ static int tdRestoreKVStore(SKVStore *pStore) {
         goto _err;
       }
     }
+
+    pRecord = taosHashIterate(pStore->map, pRecord);
   }
 
   if (pStore->aFunc) (*pStore->aFunc)(pStore->appH);
 
-  taosHashDestroyIter(pIter);
   tfree(buf);
   return 0;
 
 _err:
-  taosHashDestroyIter(pIter);
+  taosHashCancelIterate(pStore->map, pRecord);
   tfree(buf);
   return -1;
 }

src/vnode/src/vnodeMain.c

@@ -517,11 +517,10 @@ static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) {
 }
 
 int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {
-  SHashMutableIterator *pIter = taosHashCreateIter(tsVnodesHash);
-  while (taosHashIterNext(pIter)) {
-    SVnodeObj **pVnode = taosHashIterGet(pIter);
-    if (pVnode == NULL) continue;
-    if (*pVnode == NULL) continue;
+  void *pIter = taosHashIterate(tsVnodesHash, NULL);
+  while (pIter) {
+    SVnodeObj **pVnode = pIter;
+    if (*pVnode) {
 
     (*numOfVnodes)++;
     if (*numOfVnodes >= TSDB_MAX_VNODES) {
@@ -530,25 +529,24 @@ int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {
     } else {
       vnodeList[*numOfVnodes - 1] = (*pVnode)->vgId;
     }
-  }
 
-  taosHashDestroyIter(pIter);
+    }
+
+    pIter = taosHashIterate(tsVnodesHash, pIter);    
+  }
   return TSDB_CODE_SUCCESS;
 }
 
 void vnodeBuildStatusMsg(void *param) {
   SStatusMsg *pStatus = param;
-  SHashMutableIterator *pIter = taosHashCreateIter(tsVnodesHash);
-
-  while (taosHashIterNext(pIter)) {
-    SVnodeObj **pVnode = taosHashIterGet(pIter);
-    if (pVnode == NULL) continue;
-    if (*pVnode == NULL) continue;
 
+  void *pIter = taosHashIterate(tsVnodesHash, NULL);
+  while (pIter) {
+    SVnodeObj **pVnode = pIter; 
+    if (*pVnode) { 
     vnodeBuildVloadMsg(*pVnode, pStatus);
+    }
   }
-
-  taosHashDestroyIter(pIter);
 }
 
 void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes) {