#3534

Committed 21 Nov 2024 07:36AM UTC coverage: 60.825% (+2.0%) from 58.848%

Build # #3534

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28810 from taosdata/ehn/add-sync-heartbeat-sent-time-to-log

ehn:add-sync-heartbeat-sent-time-to-log

Run Details

120023 of 252376 branches covered (47.56%)

Branch coverage included in aggregate %.

43 of 47 new or added lines in 3 files covered. (91.49%)

2254 existing lines in 162 files now uncovered.

200876 of 275203 relevant lines covered (72.99%)

16110754.39 hits per line

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

69.37

/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        100
#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);
}

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("rsetId:%d p:%p failed to add, rsetId not valid", rsetId, p);
    return terrno = TSDB_CODE_REF_INVALID_ID;
  }

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

  pNode = taosMemoryCalloc(sizeof(SRefNode), 1);
  if (pNode == NULL) {
    taosDecRsetCount(pSet);
    uError("rsetId:%d p:%p failed to add, out of memory", rsetId, p);
    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("rsetId:%d p:%p rid:%" PRId64 " is added, count:%d, remain count:%d", rsetId, p, rid, pSet->count,
         pNode->count);

  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:%" PRId64 " failed to acquire, rsetId not valid", rsetId, rid);
    terrno = TSDB_CODE_REF_INVALID_ID;
    return NULL;
  }

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

  pSet = tsRefSetList + rsetId;
  taosIncRsetCount(pSet);
  if (pSet->state != TSDB_REF_STATE_ACTIVE) {
    uTrace("rsetId:%d rid:%" PRId64 " failed to acquire, not active", rsetId, rid);
    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("rsetId:%d p:%p rid:%" PRId64 " is acquired, remain count:%d", rsetId, pNode->p, rid, pNode->count);
    } else {
      terrno = TSDB_CODE_REF_NOT_EXIST;
      uTrace("rsetId:%d p:%p rid:%" PRId64 " is already removed, failed to acquire", rsetId, pNode->p, rid);
    }
  } else {
    terrno = TSDB_CODE_REF_NOT_EXIST;
    uTrace("rsetId:%d rid:%" PRId64 " is not there, failed to acquire", rsetId, rid);
  }

  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("rsetId:%d rid:%" PRId64 " failed to iterate, rsetId not valid", rsetId, rid);
    terrno = TSDB_CODE_REF_INVALID_ID;
    return NULL;
  }

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

  void *newP = NULL;
  pSet = tsRefSetList + rsetId;
  taosIncRsetCount(pSet);
  if (pSet->state != TSDB_REF_STATE_ACTIVE) {
    uTrace("rsetId:%d rid:%" PRId64 " failed to iterate, rset not active", rsetId, rid);
    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("rsetId:%d rid:%" PRId64 " iter:%d", rsetId, rid, iter);
      }

      if (pNode == NULL) {
        uError("rsetId:%d rid:%" PRId64 " not there, quit", rsetId, rid);
        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("rsetId:%d p:%p rid:%" PRId64 " is returned", rsetId, newP, rid);
    } 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("rsetId:%d p:%p rid:%" PRId64 "count:%d", i, pNode->p, pNode->rid, pNode->count);
        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("rsetId:%d rid:%" PRId64 " failed to remove, rsetId not valid", rsetId, rid);
    return terrno = TSDB_CODE_REF_INVALID_ID;
  }

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

  pSet = tsRefSetList + rsetId;
  if (pSet->state == TSDB_REF_STATE_EMPTY) {
    uTrace("rsetId:%d rid:%" PRId64 " failed to remove, cleaned", rsetId, rid);
    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("rsetId:%d rid:%" PRId64 " iter:%d", rsetId, rid, iter);
  }

  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("rsetId:%d p:%p rid:%" PRId64 " is released, remain count %d", rsetId, pNode->p, rid, pNode->count);
    }
  } else {
    uTrace("rsetId:%d rid:%" PRId64 " is not there, failed to release/remove", rsetId, rid);
    terrno = TSDB_CODE_REF_NOT_EXIST;
    code = terrno;
  }

  taosUnlockList(pSet->lockedBy + hash);

  if (released) {
    uTrace("rsetId:%d p:%p rid:%" PRId64 " is removed, count:%d, free mem: %p", rsetId, pNode->p, rid, pSet->count,
           pNode);
    (*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 100
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) {	131,678✔
62	SRefNode **nodeList;
63	SRefSet *pSet;
64	int64_t *lockedBy;
65	int32_t i, rsetId;
66
67	(void)taosThreadOnce(&tsRefModuleInit, taosInitRefModule);	131,678✔
68
69	nodeList = taosMemoryCalloc(sizeof(SRefNode *), (size_t)max);	131,678✔
70	if (nodeList == NULL) {	131,678!
71	return terrno;	×
72	}
73
74	lockedBy = taosMemoryCalloc(sizeof(int64_t), (size_t)max);	131,678✔
75	if (lockedBy == NULL) {	131,678!
76	taosMemoryFree(nodeList);	×
77	return terrno;	×
78	}
79
80	(void)taosThreadMutexLock(&tsRefMutex);	131,678✔
81
82	for (i = 0; i < TSDB_REF_OBJECTS; ++i) {	131,678!
83	tsNextId = (tsNextId + 1) % TSDB_REF_OBJECTS;	131,678✔
84	if (tsNextId == 0) tsNextId = 1; // dont use 0 as rsetId	131,678!
85	if (tsRefSetList[tsNextId].state == TSDB_REF_STATE_EMPTY) break;	131,678!
86	}
87
88	if (i < TSDB_REF_OBJECTS) {	131,678!
89	rsetId = tsNextId;	131,678✔
90	pSet = tsRefSetList + rsetId;	131,678✔
91	pSet->max = max;	131,678✔
92	pSet->nodeList = nodeList;	131,678✔
93	pSet->lockedBy = lockedBy;	131,678✔
94	pSet->fp = fp;	131,678✔
95	pSet->rid = 1;	131,678✔
96	pSet->rsetId = rsetId;	131,678✔
97	pSet->state = TSDB_REF_STATE_ACTIVE;	131,678✔
98	taosIncRsetCount(pSet);	131,678✔
99
100	tsRefSetNum++;	131,678✔
101	uTrace("rsetId:%d is opened, max:%d, fp:%p refSetNum:%d", rsetId, max, fp, tsRefSetNum);	131,678✔
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);	131,678✔
110
111	return rsetId;	131,678✔
112	}
113
114	void taosCloseRef(int32_t rsetId) {	114,982✔
115	SRefSet *pSet;
116	int32_t deleted = 0;	114,982✔
117
118	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	114,982!
119	uTrace("rsetId:%d is invalid, out of range", rsetId);	×
120	return;	×
121	}
122
123	pSet = tsRefSetList + rsetId;	114,982✔
124
125	(void)taosThreadMutexLock(&tsRefMutex);	114,982✔
126
127	if (pSet->state == TSDB_REF_STATE_ACTIVE) {	114,982✔
128	pSet->state = TSDB_REF_STATE_DELETED;	111,012✔
129	deleted = 1;	111,012✔
130	uTrace("rsetId:%d is closed, count:%d", rsetId, pSet->count);	111,012✔
131	} else {
132	uTrace("rsetId:%d is already closed, count:%d", rsetId, pSet->count);	3,970✔
133	}
134
135	(void)taosThreadMutexUnlock(&tsRefMutex);	114,982✔
136
137	if (deleted) taosDecRsetCount(pSet);	114,982✔
138	}
139
140	int64_t taosAddRef(int32_t rsetId, void *p) {	34,809,741✔
141	int32_t hash;
142	SRefNode *pNode;
143	SRefSet *pSet;
144	int64_t rid = 0;	34,809,741✔
145
146	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	34,809,741!
147	uTrace("rsetId:%d p:%p failed to add, rsetId not valid", rsetId, p);	×
148	return terrno = TSDB_CODE_REF_INVALID_ID;	×
149	}
150
151	pSet = tsRefSetList + rsetId;	34,853,987✔
152	taosIncRsetCount(pSet);	34,853,987✔
153	if (pSet->state != TSDB_REF_STATE_ACTIVE) {	34,923,405!
154	taosDecRsetCount(pSet);	×
155	uTrace("rsetId:%d p:%p failed to add, not active", rsetId, p);	×
156	return terrno = TSDB_CODE_REF_ID_REMOVED;	×
157	}
158
159	pNode = taosMemoryCalloc(sizeof(SRefNode), 1);	34,923,405✔
160	if (pNode == NULL) {	34,887,654!
UNCOV 161	taosDecRsetCount(pSet);	×
UNCOV 162	uError("rsetId:%d p:%p failed to add, out of memory", rsetId, p);	×
UNCOV 163	return terrno;	×
164	}
165
166	rid = atomic_add_fetch_64(&pSet->rid, 1);	34,887,654✔
167	hash = rid % pSet->max;	34,946,344✔
168	taosLockList(pSet->lockedBy + hash);	34,946,344✔
169
170	pNode->p = p;	34,935,279✔
171	pNode->rid = rid;	34,935,279✔
172	pNode->count = 1;	34,935,279✔
173
174	pNode->prev = NULL;	34,935,279✔
175	pNode->next = pSet->nodeList[hash];	34,935,279✔
176	if (pSet->nodeList[hash]) pSet->nodeList[hash]->prev = pNode;	34,935,279✔
177	pSet->nodeList[hash] = pNode;	34,935,279✔
178
179	uTrace("rsetId:%d p:%p rid:%" PRId64 " is added, count:%d, remain count:%d", rsetId, p, rid, pSet->count,	34,935,279✔
180	pNode->count);
181
182	taosUnlockList(pSet->lockedBy + hash);	34,935,284✔
183
184	return rid;	34,892,834✔
185	}
186
187	int32_t taosRemoveRef(int32_t rsetId, int64_t rid) { return taosDecRefCount(rsetId, rid, 1, NULL); }	34,842,102✔
188
189	// if rid is 0, return the first p in hash list, otherwise, return the next after current rid
190	void *taosAcquireRef(int32_t rsetId, int64_t rid) {	430,717,072✔
191	int32_t hash;
192	SRefNode *pNode;
193	SRefSet *pSet;
194	int32_t iter = 0;	430,717,072✔
195	void *p = NULL;	430,717,072✔
196
197	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	430,717,072!
198	// uTrace("rsetId:%d rid:%" PRId64 " failed to acquire, rsetId not valid", rsetId, rid);
199	terrno = TSDB_CODE_REF_INVALID_ID;	×
200	return NULL;	×
201	}
202
203	if (rid <= 0) {	431,443,628✔
204	uTrace("rsetId:%d rid:%" PRId64 " failed to acquire, rid not valid", rsetId, rid);	1,085,906✔
205	terrno = TSDB_CODE_REF_NOT_EXIST;	1,085,906✔
206	return NULL;	1,085,906✔
207	}
208
209	pSet = tsRefSetList + rsetId;	430,357,722✔
210	taosIncRsetCount(pSet);	430,357,722✔
211	if (pSet->state != TSDB_REF_STATE_ACTIVE) {	432,029,861!
212	uTrace("rsetId:%d rid:%" PRId64 " failed to acquire, not active", rsetId, rid);	×
213	taosDecRsetCount(pSet);	×
214	terrno = TSDB_CODE_REF_ID_REMOVED;	×
215	return NULL;	×
216	}
217
218	hash = rid % pSet->max;	432,029,861✔
219	taosLockList(pSet->lockedBy + hash);	432,029,861✔
220
221	pNode = pSet->nodeList[hash];	432,583,318✔
222
223	while (pNode) {	433,856,654✔
224	if (pNode->rid == rid) {	415,939,085✔
225	break;	414,665,749✔
226	}
227	iter++;	1,273,336✔
228	pNode = pNode->next;	1,273,336✔
229	}
230
231	if (iter >= TSDB_REF_ITER_THRESHOLD) {	432,583,318!
232	uWarn("rsetId:%d rid:%" PRId64 " iter:%d", rsetId, rid, iter);	×
233	}
234
235	if (pNode) {	432,850,005✔
236	if (pNode->removed == 0) {	414,537,127✔
237	pNode->count++;	413,687,115✔
238	p = pNode->p;	413,687,115✔
239	uTrace("rsetId:%d p:%p rid:%" PRId64 " is acquired, remain count:%d", rsetId, pNode->p, rid, pNode->count);	413,687,115✔
240	} else {
241	terrno = TSDB_CODE_REF_NOT_EXIST;	850,012✔
242	uTrace("rsetId:%d p:%p rid:%" PRId64 " is already removed, failed to acquire", rsetId, pNode->p, rid);	894,418✔
243	}
244	} else {
245	terrno = TSDB_CODE_REF_NOT_EXIST;	18,312,878✔
246	uTrace("rsetId:%d rid:%" PRId64 " is not there, failed to acquire", rsetId, rid);	18,293,906✔
247	}
248
249	taosUnlockList(pSet->lockedBy + hash);	432,875,858✔
250
251	taosDecRsetCount(pSet);	431,370,687✔
252
253	return p;	431,987,844✔
254	}
255
256	int32_t taosReleaseRef(int32_t rsetId, int64_t rid) { return taosDecRefCount(rsetId, rid, 0, NULL); }	412,146,493✔
257	int32_t taosReleaseRefEx(int32_t rsetId, int64_t rid, int32_t *isReleased) {	10,425,412✔
258	return taosDecRefCount(rsetId, rid, 0, isReleased);	10,425,412✔
259	}
260
261	// if rid is 0, return the first p in hash list, otherwise, return the next after current rid
262	void *taosIterateRef(int32_t rsetId, int64_t rid) {	1,196,574✔
263	SRefNode *pNode = NULL;	1,196,574✔
264	SRefSet *pSet;
265
266	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	1,196,574!
267	uTrace("rsetId:%d rid:%" PRId64 " failed to iterate, rsetId not valid", rsetId, rid);	×
268	terrno = TSDB_CODE_REF_INVALID_ID;	×
269	return NULL;	×
270	}
271
272	if (rid < 0) {	1,196,574!
273	uTrace("rsetId:%d rid:%" PRId64 " failed to iterate, rid not valid", rsetId, rid);	×
274	terrno = TSDB_CODE_REF_NOT_EXIST;	×
275	return NULL;	×
276	}
277
278	void *newP = NULL;	1,196,574✔
279	pSet = tsRefSetList + rsetId;	1,196,574✔
280	taosIncRsetCount(pSet);	1,196,574✔
281	if (pSet->state != TSDB_REF_STATE_ACTIVE) {	1,196,574!
282	uTrace("rsetId:%d rid:%" PRId64 " failed to iterate, rset not active", rsetId, rid);	×
283	terrno = TSDB_CODE_REF_ID_REMOVED;	×
284	taosDecRsetCount(pSet);	×
285	return NULL;	×
286	}
287
288	do {
289	newP = NULL;	1,196,574✔
290	int32_t hash = 0;	1,196,574✔
291	int32_t iter = 0;	1,196,574✔
292	if (rid > 0) {	1,196,574✔
293	hash = rid % pSet->max;	1,075,184✔
294	taosLockList(pSet->lockedBy + hash);	1,075,184✔
295
296	pNode = pSet->nodeList[hash];	1,075,184✔
297	while (pNode) {	2,056,793!
298	if (pNode->rid == rid) break;	2,056,793✔
299	pNode = pNode->next;	981,609✔
300	iter++;	981,609✔
301	}
302
303	if (iter >= TSDB_REF_ITER_THRESHOLD) {	1,075,184!
304	uWarn("rsetId:%d rid:%" PRId64 " iter:%d", rsetId, rid, iter);	×
305	}
306
307	if (pNode == NULL) {	1,075,184!
308	uError("rsetId:%d rid:%" PRId64 " not there, quit", rsetId, rid);	×
309	terrno = TSDB_CODE_REF_NOT_EXIST;	×
310	taosUnlockList(pSet->lockedBy + hash);	×
311	taosDecRsetCount(pSet);	×
312	return NULL;	×
313	}
314
315	// rid is there
316	pNode = pNode->next;	1,075,184✔
317	// check first place
318	while (pNode) {	1,075,184✔
319	if (!pNode->removed) break;	526,526!
320	pNode = pNode->next;	×
321	}
322	if (pNode == NULL) {	1,075,184✔
323	taosUnlockList(pSet->lockedBy + hash);	548,658✔
324	hash++;	548,658✔
325	}
326	}
327
328	if (pNode == NULL) {	1,196,574✔
329	for (; hash < pSet->max; ++hash) {	10,157,126✔
330	taosLockList(pSet->lockedBy + hash);	10,035,736✔
331	pNode = pSet->nodeList[hash];	10,035,736✔
332	if (pNode) {	10,035,736✔
333	// check first place
334	while (pNode) {	548,658!
335	if (!pNode->removed) break;	548,658!
336	pNode = pNode->next;	×
337	}
338	if (pNode) break;	548,658!
339	}
340	taosUnlockList(pSet->lockedBy + hash);	9,487,078✔
341	}
342	}
343
344	if (pNode) {	1,196,574✔
345	pNode->count++; // acquire it	1,075,184✔
346	newP = pNode->p;	1,075,184✔
347	taosUnlockList(pSet->lockedBy + hash);	1,075,184✔
348	uTrace("rsetId:%d p:%p rid:%" PRId64 " is returned", rsetId, newP, rid);	1,075,184✔
349	} else {
350	uTrace("rsetId:%d the list is over", rsetId);	121,390✔
351	}
352
353	if (rid > 0) taosReleaseRef(rsetId, rid); // release the current one	1,196,574✔
354	if (pNode) rid = pNode->rid;	1,196,574✔
355	} while (newP && pNode->removed);	1,196,574!
356
357	taosDecRsetCount(pSet);	1,196,574✔
358
359	return newP;	1,196,574✔
360	}
361
362	int32_t taosListRef() {	×
363	SRefSet *pSet;
364	SRefNode *pNode;
365	int32_t num = 0;	×
366
367	(void)taosThreadMutexLock(&tsRefMutex);	×
368
369	for (int32_t i = 0; i < TSDB_REF_OBJECTS; ++i) {	×
370	pSet = tsRefSetList + i;	×
371
372	if (pSet->state == TSDB_REF_STATE_EMPTY) continue;	×
373
374	uInfo("rsetId:%d state:%d count:%d", i, pSet->state, pSet->count);	×
375
376	for (int32_t j = 0; j < pSet->max; ++j) {	×
377	pNode = pSet->nodeList[j];	×
378
379	while (pNode) {	×
380	uInfo("rsetId:%d p:%p rid:%" PRId64 "count:%d", i, pNode->p, pNode->rid, pNode->count);	×
381	pNode = pNode->next;	×
382	num++;	×
383	}
384	}
385	}
386
387	(void)taosThreadMutexUnlock(&tsRefMutex);	×
388
389	return num;	×
390	}
391
392	static int32_t taosDecRefCount(int32_t rsetId, int64_t rid, int32_t remove, int32_t *isReleased) {	455,526,607✔
393	int32_t hash;
394	SRefSet *pSet;
395	SRefNode *pNode;
396	int32_t iter = 0;	455,526,607✔
397	int32_t released = 0;	455,526,607✔
398	int32_t code = 0;	455,526,607✔
399
400	if (rsetId < 0 \|\| rsetId >= TSDB_REF_OBJECTS) {	455,526,607!
401	uTrace("rsetId:%d rid:%" PRId64 " failed to remove, rsetId not valid", rsetId, rid);	×
402	return terrno = TSDB_CODE_REF_INVALID_ID;	×
403	}
404
405	if (rid <= 0) {	456,146,391✔
406	uTrace("rsetId:%d rid:%" PRId64 " failed to remove, rid not valid", rsetId, rid);	9,457,487✔
407	return terrno = TSDB_CODE_REF_NOT_EXIST;	9,457,487✔
408	}
409
410	pSet = tsRefSetList + rsetId;	446,688,904✔
411	if (pSet->state == TSDB_REF_STATE_EMPTY) {	446,688,904✔
412	uTrace("rsetId:%d rid:%" PRId64 " failed to remove, cleaned", rsetId, rid);	104,314✔
413	return terrno = TSDB_CODE_REF_ID_REMOVED;	104,314✔
414	}
415
416	hash = rid % pSet->max;	446,584,590✔
417	taosLockList(pSet->lockedBy + hash);	446,584,590✔
418
419	pNode = pSet->nodeList[hash];	449,063,739✔
420	while (pNode) {	451,317,583!
421	if (pNode->rid == rid) break;	451,543,868✔
422
423	pNode = pNode->next;	2,253,844✔
424	iter++;	2,253,844✔
425	}
426
427	if (iter >= TSDB_REF_ITER_THRESHOLD) {	449,063,739!
428	uWarn("rsetId:%d rid:%" PRId64 " iter:%d", rsetId, rid, iter);	×
429	}
430
431	if (pNode) {	449,065,725✔
432	pNode->count--;	449,064,012✔
433	if (remove) pNode->removed = 1;	449,064,012✔
434
435	if (pNode->count <= 0) {	449,064,012✔
436	if (pNode->prev) {	34,820,988✔
437	pNode->prev->next = pNode->next;	637✔
438	} else {
439	pSet->nodeList[hash] = pNode->next;	34,820,351✔
440	}
441
442	if (pNode->next) {	34,820,988✔
443	pNode->next->prev = pNode->prev;	40,290✔
444	}
445	released = 1;	34,820,988✔
446	} else {
447	uTrace("rsetId:%d p:%p rid:%" PRId64 " is released, remain count %d", rsetId, pNode->p, rid, pNode->count);	414,243,024✔
448	}
449	} else {
450	uTrace("rsetId:%d rid:%" PRId64 " is not there, failed to release/remove", rsetId, rid);	1,713✔
451	terrno = TSDB_CODE_REF_NOT_EXIST;	1,713✔
452	code = terrno;	1,713✔
453	}
454
455	taosUnlockList(pSet->lockedBy + hash);	449,065,779✔
456
457	if (released) {	448,199,939✔
458	uTrace("rsetId:%d p:%p rid:%" PRId64 " is removed, count:%d, free mem: %p", rsetId, pNode->p, rid, pSet->count,	34,799,441✔
459	pNode);
460	(*pSet->fp)(pNode->p);	34,799,443✔
461	taosMemoryFree(pNode);	34,824,244✔
462
463	taosDecRsetCount(pSet);	34,828,649✔
464	}
465
466	if (isReleased) {	448,243,935✔
467	*isReleased = released;	10,445,278✔
468	}
469
470	return code;	448,243,935✔
471	}
472
473	static void taosLockList(int64_t *lockedBy) {	916,699,533✔
474	int64_t tid = taosGetSelfPthreadId();	916,699,533✔
475	int32_t i = 0;	912,767,739✔
476	while (atomic_val_compare_exchange_64(lockedBy, 0, tid) != 0) {	1,303,495,328✔
477	if (++i % 100 == 0) {	390,727,907✔
478	(void)sched_yield();	3,810,552✔
479	}
480	}
481	}	924,608,800✔
482
483	static void taosUnlockList(int64_t *lockedBy) {	917,189,946✔
484	int64_t tid = taosGetSelfPthreadId();	917,189,946✔
485	(void)atomic_val_compare_exchange_64(lockedBy, tid, 0);	913,019,121✔
486	}	921,035,239✔
487
488	static void taosInitRefModule(void) { (void)taosThreadMutexInit(&tsRefMutex, NULL); }	9,000✔
489
490	static void taosIncRsetCount(SRefSet *pSet) {	465,020,767✔
491	(void)atomic_add_fetch_32(&pSet->count, 1);	465,020,767✔
492	// uTrace("rsetId:%d inc count:%d", pSet->rsetId, count);
493	}	468,078,903✔
494
495	static void taosDecRsetCount(SRefSet *pSet) {	466,391,408✔
496	int32_t count = atomic_sub_fetch_32(&pSet->count, 1);	466,391,408✔
497	// uTrace("rsetId:%d dec count:%d", pSet->rsetId, count);
498
499	if (count > 0) return;	468,138,140!
500
UNCOV 501	(void)taosThreadMutexLock(&tsRefMutex);	×
502
503	if (pSet->state != TSDB_REF_STATE_EMPTY) {	105,407!
504	pSet->state = TSDB_REF_STATE_EMPTY;	105,407✔
505	pSet->max = 0;	105,407✔
506	pSet->fp = NULL;	105,407✔
507
508	taosMemoryFreeClear(pSet->nodeList);	105,407!
509	taosMemoryFreeClear(pSet->lockedBy);	105,407!
510
511	tsRefSetNum--;	105,407✔
512	uTrace("rsetId:%d is cleaned, refSetNum:%d count:%d", pSet->rsetId, tsRefSetNum, pSet->count);	105,407✔
513	}
514
515	(void)taosThreadMutexUnlock(&tsRefMutex);	105,407✔
516	}

taosdata / TDengine / #3534

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