#5052

Committed 13 May 2026 12:00PM UTC coverage: 73.338% (-0.02%) from 73.358%

Build # #5052

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat: taosdump support stream backup/restore (#35326)

Coverage Stats

139 of 170 new or added lines in 3 files covered. (81.76%)

761 existing lines in 163 files now uncovered.

281469 of 383795 relevant lines covered (73.34%)

134502812.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.3

/source/util/src/tref.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#define _DEFAULT_SOURCE
#include "tref.h"
#include "taoserror.h"
#include "tlog.h"
#include "tutil.h"

#define TSDB_REF_OBJECTS        2000
#define TSDB_REF_STATE_EMPTY    0
#define TSDB_REF_STATE_ACTIVE   1
#define TSDB_REF_STATE_DELETED  2
#define TSDB_REF_ITER_THRESHOLD 100

typedef struct SRefNode {
  struct SRefNode *prev;     // previous node
  struct SRefNode *next;     // next node
  void            *p;        // pointer to resource protected,
  int64_t          rid;      // reference ID
  int32_t          count;    // number of references
  int32_t          removed;  // 1: removed
} SRefNode;

typedef struct {
  SRefNode **nodeList;  // array of SRefNode linked list
  int32_t    state;     // 0: empty, 1: active;  2: deleted
  int32_t    rsetId;    // refSet ID, global unique
  int64_t    rid;       // increase by one for each new reference
  int32_t    max;       // mod
  int32_t    count;     // total number of SRefNodes in this set
  int64_t   *lockedBy;
  void (*fp)(void *);
} SRefSet;

static SRefSet       tsRefSetList[TSDB_REF_OBJECTS];
static TdThreadOnce  tsRefModuleInit = PTHREAD_ONCE_INIT;
static TdThreadMutex tsRefMutex;
static int32_t       tsRefSetNum = 0;
static int32_t       tsNextId = 0;

static void    taosInitRefModule(void);
static void    taosLockList(int64_t *lockedBy);
static void    taosUnlockList(int64_t *lockedBy);
static void    taosIncRsetCount(SRefSet *pSet);
static void    taosDecRsetCount(SRefSet *pSet);
static int32_t taosDecRefCount(int32_t rsetId, int64_t rid, int32_t remove, int32_t *isReleased);

int32_t taosOpenRef(int32_t max, RefFp fp) {
  SRefNode **nodeList;
  SRefSet   *pSet;
  int64_t   *lockedBy;
  int32_t    i, rsetId;

  (void)taosThreadOnce(&tsRefModuleInit, taosInitRefModule);

  nodeList = taosMemoryCalloc(sizeof(SRefNode *), (size_t)max);
  if (nodeList == NULL) {
    return terrno;
  }

  lockedBy = taosMemoryCalloc(sizeof(int64_t), (size_t)max);
  if (lockedBy == NULL) {
    taosMemoryFree(nodeList);
    return terrno;
  }

  (void)taosThreadMutexLock(&tsRefMutex);

  for (i = 0; i < TSDB_REF_OBJECTS; ++i) {
    tsNextId = (tsNextId + 1) % TSDB_REF_OBJECTS;
    if (tsNextId == 0) tsNextId = 1;  // dont use 0 as rsetId
    if (tsRefSetList[tsNextId].state == TSDB_REF_STATE_EMPTY) break;
  }

  if (i < TSDB_REF_OBJECTS) {
    rsetId = tsNextId;
    pSet = tsRefSetList + rsetId;
    pSet->max = max;
    pSet->nodeList = nodeList;
    pSet->lockedBy = lockedBy;
    pSet->fp = fp;
    pSet->rid = 1;
    pSet->rsetId = rsetId;
    pSet->state = TSDB_REF_STATE_ACTIVE;
    taosIncRsetCount(pSet);

    tsRefSetNum++;
    uTrace("rsetId:%d, is opened, max:%d, fp:%p refSetNum:%d", rsetId, max, fp, tsRefSetNum);
  } else {
    rsetId = TSDB_CODE_REF_FULL;
    taosMemoryFree(nodeList);
    taosMemoryFree(lockedBy);
    uTrace("run out of Ref ID, maximum:%d refSetNum:%d", TSDB_REF_OBJECTS, tsRefSetNum);
  }

  (void)taosThreadMutexUnlock(&tsRefMutex);

  return rsetId;
}

void taosCloseRef(int32_t rsetId) {
  SRefSet *pSet;
  int32_t  deleted = 0;

  if (rsetId < 0 || rsetId >= TSDB_REF_OBJECTS) {
    uTrace("rsetId:%d, is invalid, out of range", rsetId);
    return;
  }

  pSet = tsRefSetList + rsetId;

  (void)taosThreadMutexLock(&tsRefMutex);

  if (pSet->state == TSDB_REF_STATE_ACTIVE) {
    pSet->state = TSDB_REF_STATE_DELETED;
    deleted = 1;
    uTrace("rsetId:%d, is closed, count:%d", rsetId, pSet->count);
  } else {
    uTrace("rsetId:%d, is already closed, count:%d", rsetId, pSet->count);
  }

  (void)taosThreadMutexUnlock(&tsRefMutex);

  if (deleted) taosDecRsetCount(pSet);
}

int32_t taosGetRefSetCount(int32_t rsetId, int32_t *pCount) {
  SRefSet *pSet;

  if (pCount == NULL) {
    return terrno = TSDB_CODE_INVALID_PARA;
  }

  if (rsetId < 0 || rsetId >= TSDB_REF_OBJECTS) {
    uTrace("rsetId:%d, failed to get state/count, out of range", rsetId);
    return terrno = TSDB_CODE_REF_INVALID_ID;
  }

  (void)taosThreadOnce(&tsRefModuleInit, taosInitRefModule);

  pSet = tsRefSetList + rsetId;

  (void)taosThreadMutexLock(&tsRefMutex);
  *pCount = atomic_load_32(&pSet->count);
  (void)taosThreadMutexUnlock(&tsRefMutex);

  return TSDB_CODE_SUCCESS;
}

int64_t taosAddRef(int32_t rsetId, void *p) {
  int32_t   hash;
  SRefNode *pNode;
  SRefSet  *pSet;
  int64_t   rid = 0;

  if (rsetId < 0 || rsetId >= TSDB_REF_OBJECTS) {
    uTrace("p:%p, failed to add, rsetId not valid, rsetId:%d", p, rsetId);
    return terrno = TSDB_CODE_REF_INVALID_ID;
  }

  pSet = tsRefSetList + rsetId;
  taosIncRsetCount(pSet);
  if (pSet->state != TSDB_REF_STATE_ACTIVE) {
    taosDecRsetCount(pSet);
    uTrace("p:%p, failed to add, not active, rsetId:%d", p, rsetId);
    return terrno = TSDB_CODE_REF_ID_REMOVED;
  }

  pNode = taosMemoryCalloc(sizeof(SRefNode), 1);
  if (pNode == NULL) {
    taosDecRsetCount(pSet);
    uError("p:%p, failed to add, out of memory, rsetId:%d", p, rsetId);
    return terrno;
  }

  rid = atomic_add_fetch_64(&pSet->rid, 1);
  hash = rid % pSet->max;
  taosLockList(pSet->lockedBy + hash);

  pNode->p = p;
  pNode->rid = rid;
  pNode->count = 1;

  pNode->prev = NULL;
  pNode->next = pSet->nodeList[hash];
  if (pSet->nodeList[hash]) pSet->nodeList[hash]->prev = pNode;
  pSet->nodeList[hash] = pNode;

  uTrace("p:%p, rid:0x%" PRIx64 " is added, count:%d, remain count:%d rsetId:%d", p, rid, pSet->count, pNode->count,
         rsetId);

  taosUnlockList(pSet->lockedBy + hash);

  return rid;
}

int32_t taosRemoveRef(int32_t rsetId, int64_t rid) { return taosDecRefCount(rsetId, rid, 1, NULL); }

// if rid is 0, return the first p in hash list, otherwise, return the next after current rid
void *taosAcquireRef(int32_t rsetId, int64_t rid) {
  int32_t   hash;
  SRefNode *pNode;
  SRefSet  *pSet;
  int32_t   iter = 0;
  void     *p = NULL;

  if (rsetId < 0 || rsetId >= TSDB_REF_OBJECTS) {
    // uTrace("rsetId:%d, rid:0x%" PRIx64 " failed to acquire, rsetId not valid", rsetId, rid);
    terrno = TSDB_CODE_REF_INVALID_ID;
    return NULL;
  }

  if (rid <= 0) {
    uTrace("rid:0x%" PRIx64 ", failed to acquire, rid not valid, rsetId:%d", rid, rsetId);
    terrno = TSDB_CODE_REF_NOT_EXIST;
    return NULL;
  }

  pSet = tsRefSetList + rsetId;
  taosIncRsetCount(pSet);
  if (pSet->state != TSDB_REF_STATE_ACTIVE) {
    uTrace("rid:0x%" PRIx64 ", failed to acquire, not active, rsetId:%d", rid, rsetId);
    taosDecRsetCount(pSet);
    terrno = TSDB_CODE_REF_ID_REMOVED;
    return NULL;
  }

  hash = rid % pSet->max;
  taosLockList(pSet->lockedBy + hash);

  pNode = pSet->nodeList[hash];

  while (pNode) {
    if (pNode->rid == rid) {
      break;
    }
    iter++;
    pNode = pNode->next;
  }
 
 if (iter >= TSDB_REF_ITER_THRESHOLD) {
    uWarn("rsetId:%d rid:%" PRId64 " iter:%d", rsetId, rid, iter);
  }

  if (pNode) {
    if (pNode->removed == 0) {
      pNode->count++;
      p = pNode->p;
      uTrace("p:%p, rid:0x%" PRIx64 " is acquired, remain count:%d, rsetId:%d", pNode->p, rid, pNode->count, rsetId);
    } else {
      terrno = TSDB_CODE_REF_NOT_EXIST;
      uTrace("p:%p, rid:0x%" PRIx64 " is already removed, failed to acquire, rsetId:%d", pNode->p, rid, rsetId);
    }
  } else {
    terrno = TSDB_CODE_REF_NOT_EXIST;
    uTrace("rid:0x%" PRIx64 ", is not there, failed to acquire, rsetId:%d", rid, rsetId);
  }

  taosUnlockList(pSet->lockedBy + hash);

  taosDecRsetCount(pSet);

  return p;
}

int32_t taosReleaseRef(int32_t rsetId, int64_t rid) { return taosDecRefCount(rsetId, rid, 0, NULL); }
int32_t taosReleaseRefEx(int32_t rsetId, int64_t rid, int32_t *isReleased) {
  return taosDecRefCount(rsetId, rid, 0, isReleased);
}

// if rid is 0, return the first p in hash list, otherwise, return the next after current rid
void *taosIterateRef(int32_t rsetId, int64_t rid) {
  SRefNode *pNode = NULL;
  SRefSet  *pSet;

  if (rsetId < 0 || rsetId >= TSDB_REF_OBJECTS) {
    uTrace("rid:0x%" PRIx64 ", failed to iterate, rsetId not valid, rsetId:%d", rid, rsetId);
    terrno = TSDB_CODE_REF_INVALID_ID;
    return NULL;
  }

  if (rid < 0) {
    uTrace("rid:0x%" PRIx64 ", failed to iterate, rid not valid, rsetId:%d", rid, rsetId);
    terrno = TSDB_CODE_REF_NOT_EXIST;
    return NULL;
  }

  void *newP = NULL;
  pSet = tsRefSetList + rsetId;
  taosIncRsetCount(pSet);
  if (pSet->state != TSDB_REF_STATE_ACTIVE) {
    uTrace("rid:0x%" PRIx64 ", failed to iterate, rset not active, rsetId:%d", rid, rsetId);
    terrno = TSDB_CODE_REF_ID_REMOVED;
    taosDecRsetCount(pSet);
    return NULL;
  }

  do {
    newP = NULL;
    int32_t hash = 0;
    int32_t iter = 0;
    if (rid > 0) {
      hash = rid % pSet->max;
      taosLockList(pSet->lockedBy + hash);

      pNode = pSet->nodeList[hash];
      while (pNode) {
        if (pNode->rid == rid) break;
        pNode = pNode->next;
        iter++;
      }

      if (iter >= TSDB_REF_ITER_THRESHOLD) {
        uWarn("rid:0x%" PRIx64 ", iter:%d, rsetId:%d", rid, iter, rsetId);
      }

      if (pNode == NULL) {
        uError("rid:0x%" PRIx64 ", not there and quit, rsetId:%d", rid, rsetId);
        terrno = TSDB_CODE_REF_NOT_EXIST;
        taosUnlockList(pSet->lockedBy + hash);
        taosDecRsetCount(pSet);
        return NULL;
      }

      // rid is there
      pNode = pNode->next;
      // check first place
      while (pNode) {
        if (!pNode->removed) break;
        pNode = pNode->next;
      }
      if (pNode == NULL) {
        taosUnlockList(pSet->lockedBy + hash);
        hash++;
      }
    }

    if (pNode == NULL) {
      for (; hash < pSet->max; ++hash) {
        taosLockList(pSet->lockedBy + hash);
        pNode = pSet->nodeList[hash];
        if (pNode) {
          // check first place
          while (pNode) {
            if (!pNode->removed) break;
            pNode = pNode->next;
          }
          if (pNode) break;
        }
        taosUnlockList(pSet->lockedBy + hash);
      }
    }

    if (pNode) {
      pNode->count++;  // acquire it
      newP = pNode->p;
      taosUnlockList(pSet->lockedBy + hash);
      uTrace("p:%p, rid:0x%" PRIx64 " is returned, rsetId:%d", newP, rid, rsetId);
    } else {
      uTrace("rsetId:%d, the list is over", rsetId);
    }

    if (rid > 0) taosReleaseRef(rsetId, rid);  // release the current one
    if (pNode) rid = pNode->rid;
  } while (newP && pNode->removed);

  taosDecRsetCount(pSet);

  return newP;
}

int32_t taosListRef() {
  SRefSet  *pSet;
  SRefNode *pNode;
  int32_t   num = 0;

  (void)taosThreadMutexLock(&tsRefMutex);

  for (int32_t i = 0; i < TSDB_REF_OBJECTS; ++i) {
    pSet = tsRefSetList + i;

    if (pSet->state == TSDB_REF_STATE_EMPTY) continue;

    uInfo("rsetId:%d, state:%d count:%d", i, pSet->state, pSet->count);

    for (int32_t j = 0; j < pSet->max; ++j) {
      pNode = pSet->nodeList[j];

      while (pNode) {
        uInfo("p:%p, rid:0x%" PRIx64 " count:%d, rsetId:%d", pNode->p, pNode->rid, pNode->count, i);
        pNode = pNode->next;
        num++;
      }
    }
  }

  (void)taosThreadMutexUnlock(&tsRefMutex);

  return num;
}

static int32_t taosDecRefCount(int32_t rsetId, int64_t rid, int32_t remove, int32_t *isReleased) {
  int32_t   hash;
  SRefSet  *pSet;
  SRefNode *pNode;
  int32_t   iter = 0;
  int32_t   released = 0;
  int32_t   code = 0;

  if (rsetId < 0 || rsetId >= TSDB_REF_OBJECTS) {
    uTrace("rid:0x%" PRIx64 ", failed to remove, rsetId not valid, rsetId:%d", rid, rsetId);
    return terrno = TSDB_CODE_REF_INVALID_ID;
  }

  if (rid <= 0) {
    uTrace("rid:0x%" PRIx64 ", failed to remove, rid not valid, rsetId:%d", rid, rsetId);
    return terrno = TSDB_CODE_REF_NOT_EXIST;
  }

  pSet = tsRefSetList + rsetId;
  if (pSet->state == TSDB_REF_STATE_EMPTY) {
    uTrace("rid:0x%" PRIx64 ", failed to remove, cleaned, rsetId:%d", rid, rsetId);
    return terrno = TSDB_CODE_REF_ID_REMOVED;
  }

  hash = rid % pSet->max;
  taosLockList(pSet->lockedBy + hash);

  pNode = pSet->nodeList[hash];
  while (pNode) {
    if (pNode->rid == rid) break;

    pNode = pNode->next;
    iter++;
  }

  if (iter >= TSDB_REF_ITER_THRESHOLD) {
    uWarn("rid:0x%" PRIx64 ", iter:%d, rsetId:%d", rid, iter, rsetId);
  }

  if (pNode) {
    pNode->count--;
    if (remove) pNode->removed = 1;

    if (pNode->count <= 0) {
      if (pNode->prev) {
        pNode->prev->next = pNode->next;
      } else {
        pSet->nodeList[hash] = pNode->next;
      }

      if (pNode->next) {
        pNode->next->prev = pNode->prev;
      }
      released = 1;
    } else {
      uTrace("p:%p, rid:0x%" PRIx64 " is released, remain count:%d, rsetId:%d", pNode->p, rid, pNode->count, rsetId);
    }
  } else {
    uTrace("rid:0x%" PRIx64 ", is not there, failed to release/remove, rsetId:%d", rid, rsetId);
    terrno = TSDB_CODE_REF_NOT_EXIST;
    code = terrno;
  }

  taosUnlockList(pSet->lockedBy + hash);

  if (released) {
    uTrace("p:%p, rid:0x%" PRIx64 " is removed, count:%d, free mem:%p, rsetId:%d", pNode->p, rid, pSet->count, pNode,
           rsetId);
    (*pSet->fp)(pNode->p);
    taosMemoryFree(pNode);

    taosDecRsetCount(pSet);
  }

  if (isReleased) {
    *isReleased = released;
  }

  return code;
}

static void taosLockList(int64_t *lockedBy) {
  int64_t tid = taosGetSelfPthreadId();
  int32_t i = 0;
  while (atomic_val_compare_exchange_64(lockedBy, 0, tid) != 0) {
    if (++i % 100 == 0) {
      (void)sched_yield();
    }

  }
}

static void taosUnlockList(int64_t *lockedBy) {
  int64_t tid = taosGetSelfPthreadId();
  (void)atomic_val_compare_exchange_64(lockedBy, tid, 0);
}

static void taosInitRefModule(void) { (void)taosThreadMutexInit(&tsRefMutex, NULL); }

static void taosIncRsetCount(SRefSet *pSet) {
  (void)atomic_add_fetch_32(&pSet->count, 1);
  // uTrace("rsetId:%d, inc count:%d", pSet->rsetId, count);
}

static void taosDecRsetCount(SRefSet *pSet) {
  int32_t count = atomic_sub_fetch_32(&pSet->count, 1);
  // uTrace("rsetId:%d, dec count:%d", pSet->rsetId, count);

  if (count > 0) return;

  (void)taosThreadMutexLock(&tsRefMutex);

  if (pSet->state != TSDB_REF_STATE_EMPTY) {
    pSet->state = TSDB_REF_STATE_EMPTY;
    pSet->max = 0;
    pSet->fp = NULL;

    taosMemoryFreeClear(pSet->nodeList);
    taosMemoryFreeClear(pSet->lockedBy);

    tsRefSetNum--;
    uTrace("rsetId:%d, is cleaned, refSetNum:%d count:%d", pSet->rsetId, tsRefSetNum, pSet->count);
  }

  (void)taosThreadMutexUnlock(&tsRefMutex);
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#define _DEFAULT_SOURCE
17	#include "tref.h"
18	#include "taoserror.h"
19	#include "tlog.h"
20	#include "tutil.h"
21
22	#define TSDB_REF_OBJECTS 2000
23	#define TSDB_REF_STATE_EMPTY 0
24	#define TSDB_REF_STATE_ACTIVE 1
25	#define TSDB_REF_STATE_DELETED 2
26	#define TSDB_REF_ITER_THRESHOLD 100
27
28	typedef struct SRefNode {
29	struct SRefNode *prev; // previous node
30	struct SRefNode *next; // next node
31	void *p; // pointer to resource protected,
32	int64_t rid; // reference ID
33	int32_t count; // number of references
34	int32_t removed; // 1: removed
35	} SRefNode;
36
37	typedef struct {
38	SRefNode **nodeList; // array of SRefNode linked list
39	int32_t state; // 0: empty, 1: active; 2: deleted
40	int32_t rsetId; // refSet ID, global unique
41	int64_t rid; // increase by one for each new reference
42	int32_t max; // mod
43	int32_t count; // total number of SRefNodes in this set
44	int64_t *lockedBy;
45	void (fp)(void );
46	} SRefSet;
47
48	static SRefSet tsRefSetList[TSDB_REF_OBJECTS];
49	static TdThreadOnce tsRefModuleInit = PTHREAD_ONCE_INIT;
50	static TdThreadMutex tsRefMutex;
51	static int32_t tsRefSetNum = 0;
52	static int32_t tsNextId = 0;
53
54	static void taosInitRefModule(void);
55	static void taosLockList(int64_t *lockedBy);
56	static void taosUnlockList(int64_t *lockedBy);
57	static void taosIncRsetCount(SRefSet *pSet);
58	static void taosDecRsetCount(SRefSet *pSet);
59	static int32_t taosDecRefCount(int32_t rsetId, int64_t rid, int32_t remove, int32_t *isReleased);
60
61	int32_t taosOpenRef(int32_t max, RefFp fp) {	40,593,683✔
62	SRefNode **nodeList;
63	SRefSet *pSet;
64	int64_t *lockedBy;
65	int32_t i, rsetId;
66
67	(void)taosThreadOnce(&tsRefModuleInit, taosInitRefModule);	40,593,683✔
68
69	nodeList = taosMemoryCalloc(sizeof(SRefNode *), (size_t)max);	40,593,683✔
70	if (nodeList == NULL) {	40,593,683✔
71	return terrno;	×
72	}
73
74	lockedBy = taosMemoryCalloc(sizeof(int64_t), (size_t)max);	40,593,683✔
75	if (lockedBy == NULL) {	40,593,683✔
76	taosMemoryFree(nodeList);	×
77	return terrno;	×
78	}
79
80	(void)taosThreadMutexLock(&tsRefMutex);	40,593,683✔
81
82	for (i = 0; i < TSDB_REF_OBJECTS; ++i) {	40,593,683✔
83	tsNextId = (tsNextId + 1) % TSDB_REF_OBJECTS;	40,593,683✔
84	if (tsNextId == 0) tsNextId = 1; // dont use 0 as rsetId	40,593,683✔
85	if (tsRefSetList[tsNextId].state == TSDB_REF_STATE_EMPTY) break;	40,593,683✔
86	}
87
88	if (i < TSDB_REF_OBJECTS) {	40,593,683✔
89	rsetId = tsNextId;	40,593,683✔
90	pSet = tsRefSetList + rsetId;	40,593,683✔
91	pSet->max = max;	40,593,683✔
92	pSet->nodeList = nodeList;	40,593,683✔
93	pSet->lockedBy = lockedBy;	40,593,683✔
94	pSet->fp = fp;	40,593,683✔
95	pSet->rid = 1;	40,593,683✔
96	pSet->rsetId = rsetId;	40,593,683✔
97	pSet->state = TSDB_REF_STATE_ACTIVE;	40,593,683✔
98	taosIncRsetCount(pSet);	40,593,683✔
99
100	tsRefSetNum++;	40,593,683✔
101	uTrace("rsetId:%d, is opened, max:%d, fp:%p refSetNum:%d", rsetId, max, fp, tsRefSetNum);	40,593,683✔
102	} else {
103	rsetId = TSDB_CODE_REF_FULL;	×
104	taosMemoryFree(nodeList);	×
105	taosMemoryFree(lockedBy);	×
106	uTrace("run out of Ref ID, maximum:%d refSetNum:%d", TSDB_REF_OBJECTS, tsRefSetNum);	×
107	}
108
109	(void)taosThreadMutexUnlock(&tsRefMutex);	40,593,683✔
110
111	return rsetId;	40,593,683✔
112	}
113
114	void taosCloseRef(int32_t rsetId) {	36,774,118✔
115	SRefSet *pSet;
116	int32_t deleted = 0;	36,774,118✔
117
118	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	36,774,118✔
119	uTrace("rsetId:%d, is invalid, out of range", rsetId);	130✔
120	return;	130✔
121	}
122
123	pSet = tsRefSetList + rsetId;	36,773,988✔
124
125	(void)taosThreadMutexLock(&tsRefMutex);	36,773,988✔
126
127	if (pSet->state == TSDB_REF_STATE_ACTIVE) {	36,773,988✔
128	pSet->state = TSDB_REF_STATE_DELETED;	34,324,899✔
129	deleted = 1;	34,324,899✔
130	uTrace("rsetId:%d, is closed, count:%d", rsetId, pSet->count);	34,324,899✔
131	} else {
132	uTrace("rsetId:%d, is already closed, count:%d", rsetId, pSet->count);	2,449,089✔
133	}
134
135	(void)taosThreadMutexUnlock(&tsRefMutex);	36,773,988✔
136
137	if (deleted) taosDecRsetCount(pSet);	36,773,988✔
138	}
139
140	int32_t taosGetRefSetCount(int32_t rsetId, int32_t *pCount) {	1,052,785,197✔
141	SRefSet *pSet;
142
143	if (pCount == NULL) {	1,052,785,197✔
144	return terrno = TSDB_CODE_INVALID_PARA;	×
145	}
146
147	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	1,052,785,197✔
148	uTrace("rsetId:%d, failed to get state/count, out of range", rsetId);	×
149	return terrno = TSDB_CODE_REF_INVALID_ID;	×
150	}
151
152	(void)taosThreadOnce(&tsRefModuleInit, taosInitRefModule);	1,052,785,197✔
153
154	pSet = tsRefSetList + rsetId;	1,052,785,197✔
155
156	(void)taosThreadMutexLock(&tsRefMutex);	1,052,785,197✔
157	*pCount = atomic_load_32(&pSet->count);	1,052,785,197✔
158	(void)taosThreadMutexUnlock(&tsRefMutex);	1,052,785,197✔
159
160	return TSDB_CODE_SUCCESS;	1,052,785,197✔
161	}
162
163	int64_t taosAddRef(int32_t rsetId, void *p) {	2,147,483,647✔
164	int32_t hash;
165	SRefNode *pNode;
166	SRefSet *pSet;
167	int64_t rid = 0;	2,147,483,647✔
168
169	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	2,147,483,647✔
UNCOV 170	uTrace("p:%p, failed to add, rsetId not valid, rsetId:%d", p, rsetId);	×
UNCOV 171	return terrno = TSDB_CODE_REF_INVALID_ID;	×
172	}
173
174	pSet = tsRefSetList + rsetId;	2,147,483,647✔
175	taosIncRsetCount(pSet);	2,147,483,647✔
176	if (pSet->state != TSDB_REF_STATE_ACTIVE) {	2,147,483,647✔
177	taosDecRsetCount(pSet);	×
178	uTrace("p:%p, failed to add, not active, rsetId:%d", p, rsetId);	×
179	return terrno = TSDB_CODE_REF_ID_REMOVED;	×
180	}
181
182	pNode = taosMemoryCalloc(sizeof(SRefNode), 1);	2,147,483,647✔
183	if (pNode == NULL) {	2,147,483,647✔
184	taosDecRsetCount(pSet);	×
185	uError("p:%p, failed to add, out of memory, rsetId:%d", p, rsetId);	×
186	return terrno;	×
187	}
188
189	rid = atomic_add_fetch_64(&pSet->rid, 1);	2,147,483,647✔
190	hash = rid % pSet->max;	2,147,483,647✔
191	taosLockList(pSet->lockedBy + hash);	2,147,483,647✔
192
193	pNode->p = p;	2,147,483,647✔
194	pNode->rid = rid;	2,147,483,647✔
195	pNode->count = 1;	2,147,483,647✔
196
197	pNode->prev = NULL;	2,147,483,647✔
198	pNode->next = pSet->nodeList[hash];	2,147,483,647✔
199	if (pSet->nodeList[hash]) pSet->nodeList[hash]->prev = pNode;	2,147,483,647✔
200	pSet->nodeList[hash] = pNode;	2,147,483,647✔
201
202	uTrace("p:%p, rid:0x%" PRIx64 " is added, count:%d, remain count:%d rsetId:%d", p, rid, pSet->count, pNode->count,	2,147,483,647✔
203	rsetId);
204
205	taosUnlockList(pSet->lockedBy + hash);	2,147,483,647✔
206
207	return rid;	2,147,483,647✔
208	}
209
210	int32_t taosRemoveRef(int32_t rsetId, int64_t rid) { return taosDecRefCount(rsetId, rid, 1, NULL); }	2,147,483,647✔
211
212	// if rid is 0, return the first p in hash list, otherwise, return the next after current rid
213	void *taosAcquireRef(int32_t rsetId, int64_t rid) {	2,147,483,647✔
214	int32_t hash;
215	SRefNode *pNode;
216	SRefSet *pSet;
217	int32_t iter = 0;	2,147,483,647✔
218	void *p = NULL;	2,147,483,647✔
219
220	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	2,147,483,647✔
221	// uTrace("rsetId:%d, rid:0x%" PRIx64 " failed to acquire, rsetId not valid", rsetId, rid);
222	terrno = TSDB_CODE_REF_INVALID_ID;	325✔
223	return NULL;	58,187✔
224	}
225
226	if (rid <= 0) {	2,147,483,647✔
227	uTrace("rid:0x%" PRIx64 ", failed to acquire, rid not valid, rsetId:%d", rid, rsetId);	273,228,104✔
228	terrno = TSDB_CODE_REF_NOT_EXIST;	273,228,104✔
229	return NULL;	273,228,104✔
230	}
231
232	pSet = tsRefSetList + rsetId;	2,147,483,647✔
233	taosIncRsetCount(pSet);	2,147,483,647✔
234	if (pSet->state != TSDB_REF_STATE_ACTIVE) {	2,147,483,647✔
235	uTrace("rid:0x%" PRIx64 ", failed to acquire, not active, rsetId:%d", rid, rsetId);	×
236	taosDecRsetCount(pSet);	×
237	terrno = TSDB_CODE_REF_ID_REMOVED;	×
238	return NULL;	×
239	}
240
241	hash = rid % pSet->max;	2,147,483,647✔
242	taosLockList(pSet->lockedBy + hash);	2,147,483,647✔
243
244	pNode = pSet->nodeList[hash];	2,147,483,647✔
245
246	while (pNode) {	2,147,483,647✔
247	if (pNode->rid == rid) {	2,147,483,647✔
248	break;	2,147,483,647✔
249	}
250	iter++;	16,047,737✔
251	pNode = pNode->next;	16,047,737✔
252	}
253
254	if (iter >= TSDB_REF_ITER_THRESHOLD) {	2,147,483,647✔
255	uWarn("rsetId:%d rid:%" PRId64 " iter:%d", rsetId, rid, iter);	×
256	}
257
258	if (pNode) {	2,147,483,647✔
259	if (pNode->removed == 0) {	2,147,483,647✔
260	pNode->count++;	2,147,483,647✔
261	p = pNode->p;	2,147,483,647✔
262	uTrace("p:%p, rid:0x%" PRIx64 " is acquired, remain count:%d, rsetId:%d", pNode->p, rid, pNode->count, rsetId);	2,147,483,647✔
263	} else {
264	terrno = TSDB_CODE_REF_NOT_EXIST;	10,097,613✔
265	uTrace("p:%p, rid:0x%" PRIx64 " is already removed, failed to acquire, rsetId:%d", pNode->p, rid, rsetId);	10,318,149✔
266	}
267	} else {
268	terrno = TSDB_CODE_REF_NOT_EXIST;	1,562,611,063✔
269	uTrace("rid:0x%" PRIx64 ", is not there, failed to acquire, rsetId:%d", rid, rsetId);	1,562,608,668✔
270	}
271
272	taosUnlockList(pSet->lockedBy + hash);	2,147,483,647✔
273
274	taosDecRsetCount(pSet);	2,147,483,647✔
275
276	return p;	2,147,483,647✔
277	}
278
279	int32_t taosReleaseRef(int32_t rsetId, int64_t rid) { return taosDecRefCount(rsetId, rid, 0, NULL); }	2,147,483,647✔
280	int32_t taosReleaseRefEx(int32_t rsetId, int64_t rid, int32_t *isReleased) {	996,314,401✔
281	return taosDecRefCount(rsetId, rid, 0, isReleased);	996,314,401✔
282	}
283
284	// if rid is 0, return the first p in hash list, otherwise, return the next after current rid
285	void *taosIterateRef(int32_t rsetId, int64_t rid) {	278,526,881✔
286	SRefNode *pNode = NULL;	278,526,881✔
287	SRefSet *pSet;
288
289	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	278,526,881✔
290	uTrace("rid:0x%" PRIx64 ", failed to iterate, rsetId not valid, rsetId:%d", rid, rsetId);	×
291	terrno = TSDB_CODE_REF_INVALID_ID;	×
292	return NULL;	×
293	}
294
295	if (rid < 0) {	278,526,881✔
296	uTrace("rid:0x%" PRIx64 ", failed to iterate, rid not valid, rsetId:%d", rid, rsetId);	×
297	terrno = TSDB_CODE_REF_NOT_EXIST;	×
298	return NULL;	×
299	}
300
301	void *newP = NULL;	278,526,881✔
302	pSet = tsRefSetList + rsetId;	278,526,881✔
303	taosIncRsetCount(pSet);	278,526,881✔
304	if (pSet->state != TSDB_REF_STATE_ACTIVE) {	278,526,881✔
305	uTrace("rid:0x%" PRIx64 ", failed to iterate, rset not active, rsetId:%d", rid, rsetId);	×
306	terrno = TSDB_CODE_REF_ID_REMOVED;	×
307	taosDecRsetCount(pSet);	×
308	return NULL;	×
309	}
310
311	do {
312	newP = NULL;	278,526,881✔
313	int32_t hash = 0;	278,526,881✔
314	int32_t iter = 0;	278,526,881✔
315	if (rid > 0) {	278,526,881✔
316	hash = rid % pSet->max;	214,228,149✔
317	taosLockList(pSet->lockedBy + hash);	214,228,149✔
318
319	pNode = pSet->nodeList[hash];	214,228,149✔
320	while (pNode) {	218,089,088✔
321	if (pNode->rid == rid) break;	218,089,088✔
322	pNode = pNode->next;	3,860,939✔
323	iter++;	3,860,939✔
324	}
325
326	if (iter >= TSDB_REF_ITER_THRESHOLD) {	214,228,149✔
327	uWarn("rid:0x%" PRIx64 ", iter:%d, rsetId:%d", rid, iter, rsetId);	×
328	}
329
330	if (pNode == NULL) {	214,228,149✔
331	uError("rid:0x%" PRIx64 ", not there and quit, rsetId:%d", rid, rsetId);	×
332	terrno = TSDB_CODE_REF_NOT_EXIST;	×
333	taosUnlockList(pSet->lockedBy + hash);	×
334	taosDecRsetCount(pSet);	×
335	return NULL;	×
336	}
337
338	// rid is there
339	pNode = pNode->next;	214,228,149✔
340	// check first place
341	while (pNode) {	214,228,149✔
342	if (!pNode->removed) break;	3,595,280✔
343	pNode = pNode->next;	×
344	}
345	if (pNode == NULL) {	214,228,149✔
346	taosUnlockList(pSet->lockedBy + hash);	210,632,869✔
347	hash++;	210,632,869✔
348	}
349	}
350
351	if (pNode == NULL) {	278,526,881✔
352	for (; hash < pSet->max; ++hash) {	2,147,483,647✔
353	taosLockList(pSet->lockedBy + hash);	2,147,483,647✔
354	pNode = pSet->nodeList[hash];	2,147,483,647✔
355	if (pNode) {	2,147,483,647✔
356	// check first place
357	while (pNode) {	210,632,869✔
358	if (!pNode->removed) break;	210,632,869✔
359	pNode = pNode->next;	×
360	}
361	if (pNode) break;	210,632,869✔
362	}
363	taosUnlockList(pSet->lockedBy + hash);	2,147,483,647✔
364	}
365	}
366
367	if (pNode) {	278,526,881✔
368	pNode->count++; // acquire it	214,228,149✔
369	newP = pNode->p;	214,228,149✔
370	taosUnlockList(pSet->lockedBy + hash);	214,228,149✔
371	uTrace("p:%p, rid:0x%" PRIx64 " is returned, rsetId:%d", newP, rid, rsetId);	214,228,149✔
372	} else {
373	uTrace("rsetId:%d, the list is over", rsetId);	64,298,732✔
374	}
375
376	if (rid > 0) taosReleaseRef(rsetId, rid); // release the current one	278,526,881✔
377	if (pNode) rid = pNode->rid;	278,526,881✔
378	} while (newP && pNode->removed);	278,526,881✔
379
380	taosDecRsetCount(pSet);	278,526,881✔
381
382	return newP;	278,526,881✔
383	}
384
385	int32_t taosListRef() {	×
386	SRefSet *pSet;
387	SRefNode *pNode;
388	int32_t num = 0;	×
389
390	(void)taosThreadMutexLock(&tsRefMutex);	×
391
392	for (int32_t i = 0; i < TSDB_REF_OBJECTS; ++i) {	×
393	pSet = tsRefSetList + i;	×
394
395	if (pSet->state == TSDB_REF_STATE_EMPTY) continue;	×
396
397	uInfo("rsetId:%d, state:%d count:%d", i, pSet->state, pSet->count);	×
398
399	for (int32_t j = 0; j < pSet->max; ++j) {	×
400	pNode = pSet->nodeList[j];	×
401
402	while (pNode) {	×
403	uInfo("p:%p, rid:0x%" PRIx64 " count:%d, rsetId:%d", pNode->p, pNode->rid, pNode->count, i);	×
404	pNode = pNode->next;	×
405	num++;	×
406	}
407	}
408	}
409
410	(void)taosThreadMutexUnlock(&tsRefMutex);	×
411
412	return num;	×
413	}
414
415	static int32_t taosDecRefCount(int32_t rsetId, int64_t rid, int32_t remove, int32_t *isReleased) {	2,147,483,647✔
416	int32_t hash;
417	SRefSet *pSet;
418	SRefNode *pNode;
419	int32_t iter = 0;	2,147,483,647✔
420	int32_t released = 0;	2,147,483,647✔
421	int32_t code = 0;	2,147,483,647✔
422
423	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	2,147,483,647✔
424	uTrace("rid:0x%" PRIx64 ", failed to remove, rsetId not valid, rsetId:%d", rid, rsetId);	349✔
425	return terrno = TSDB_CODE_REF_INVALID_ID;	349✔
426	}
427
428	if (rid <= 0) {	2,147,483,647✔
429	uTrace("rid:0x%" PRIx64 ", failed to remove, rid not valid, rsetId:%d", rid, rsetId);	714,683,918✔
430	return terrno = TSDB_CODE_REF_NOT_EXIST;	714,683,918✔
431	}
432
433	pSet = tsRefSetList + rsetId;	2,147,483,647✔
434	if (pSet->state == TSDB_REF_STATE_EMPTY) {	2,147,483,647✔
435	uTrace("rid:0x%" PRIx64 ", failed to remove, cleaned, rsetId:%d", rid, rsetId);	×
436	return terrno = TSDB_CODE_REF_ID_REMOVED;	×
437	}
438
439	hash = rid % pSet->max;	2,147,483,647✔
440	taosLockList(pSet->lockedBy + hash);	2,147,483,647✔
441
442	pNode = pSet->nodeList[hash];	2,147,483,647✔
443	while (pNode) {	2,147,483,647✔
444	if (pNode->rid == rid) break;	2,147,483,647✔
445
446	pNode = pNode->next;	32,530,608✔
447	iter++;	32,530,608✔
448	}
449
450	if (iter >= TSDB_REF_ITER_THRESHOLD) {	2,147,483,647✔
451	uWarn("rid:0x%" PRIx64 ", iter:%d, rsetId:%d", rid, iter, rsetId);	×
452	}
453
454	if (pNode) {	2,147,483,647✔
455	pNode->count--;	2,147,483,647✔
456	if (remove) pNode->removed = 1;	2,147,483,647✔
457
458	if (pNode->count <= 0) {	2,147,483,647✔
459	if (pNode->prev) {	2,147,483,647✔
460	pNode->prev->next = pNode->next;	5,173,806✔
461	} else {
462	pSet->nodeList[hash] = pNode->next;	2,147,483,647✔
463	}
464
465	if (pNode->next) {	2,147,483,647✔
466	pNode->next->prev = pNode->prev;	1,641,651✔
467	}
468	released = 1;	2,147,483,647✔
469	} else {
470	uTrace("p:%p, rid:0x%" PRIx64 " is released, remain count:%d, rsetId:%d", pNode->p, rid, pNode->count, rsetId);	2,147,483,647✔
471	}
472	} else {
473	uTrace("rid:0x%" PRIx64 ", is not there, failed to release/remove, rsetId:%d", rid, rsetId);	1,303,514✔
474	terrno = TSDB_CODE_REF_NOT_EXIST;	1,303,514✔
475	code = terrno;	1,303,514✔
476	}
477
478	taosUnlockList(pSet->lockedBy + hash);	2,147,483,647✔
479
480	if (released) {	2,147,483,647✔
481	uTrace("p:%p, rid:0x%" PRIx64 " is removed, count:%d, free mem:%p, rsetId:%d", pNode->p, rid, pSet->count, pNode,	2,147,483,647✔
482	rsetId);
483	(*pSet->fp)(pNode->p);	2,147,483,647✔
484	taosMemoryFree(pNode);	2,147,483,647✔
485
486	taosDecRsetCount(pSet);	2,147,483,647✔
487	}
488
489	if (isReleased) {	2,147,483,647✔
490	*isReleased = released;	996,355,985✔
491	}
492
493	return code;	2,147,483,647✔
494	}
495
496	static void taosLockList(int64_t *lockedBy) {	2,147,483,647✔
497	int64_t tid = taosGetSelfPthreadId();	2,147,483,647✔
498	int32_t i = 0;	2,147,483,647✔
499	while (atomic_val_compare_exchange_64(lockedBy, 0, tid) != 0) {	2,147,483,647✔
500	if (++i % 100 == 0) {	2,147,483,647✔
501	(void)sched_yield();	351,715,868✔
502	}
503
504	}
505	}	2,147,483,647✔
506
507	static void taosUnlockList(int64_t *lockedBy) {	2,147,483,647✔
508	int64_t tid = taosGetSelfPthreadId();	2,147,483,647✔
509	(void)atomic_val_compare_exchange_64(lockedBy, tid, 0);	2,147,483,647✔
510	}	2,147,483,647✔
511
512	static void taosInitRefModule(void) { (void)taosThreadMutexInit(&tsRefMutex, NULL); }	3,109,571✔
513
514	static void taosIncRsetCount(SRefSet *pSet) {	2,147,483,647✔
515	(void)atomic_add_fetch_32(&pSet->count, 1);	2,147,483,647✔
516	// uTrace("rsetId:%d, inc count:%d", pSet->rsetId, count);
517	}	2,147,483,647✔
518
519	static void taosDecRsetCount(SRefSet *pSet) {	2,147,483,647✔
520	int32_t count = atomic_sub_fetch_32(&pSet->count, 1);	2,147,483,647✔
521	// uTrace("rsetId:%d, dec count:%d", pSet->rsetId, count);
522
523	if (count > 0) return;	2,147,483,647✔
524
525	(void)taosThreadMutexLock(&tsRefMutex);	33,943,113✔
526
527	if (pSet->state != TSDB_REF_STATE_EMPTY) {	34,071,694✔
528	pSet->state = TSDB_REF_STATE_EMPTY;	34,071,694✔
529	pSet->max = 0;	34,071,694✔
530	pSet->fp = NULL;	34,071,694✔
531
532	taosMemoryFreeClear(pSet->nodeList);	34,071,694✔
533	taosMemoryFreeClear(pSet->lockedBy);	34,071,694✔
534
535	tsRefSetNum--;	34,071,694✔
536	uTrace("rsetId:%d, is cleaned, refSetNum:%d count:%d", pSet->rsetId, tsRefSetNum, pSet->count);	34,071,694✔
537	}
538
539	(void)taosThreadMutexUnlock(&tsRefMutex);	34,071,694✔
540	}

taosdata / TDengine / #5052

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous