xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

minor changes

This commit is contained in:
Shengliang Guan 2022-02-28 14:21:18 +08:00
parent 7747ab2324
commit 99bc4379f0
4 changed files with 101 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
include/util/tsched.h
View File

@ -13,8 +13,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#ifndef _TD_UTIL_SCHED_H #ifndef _TD_UTIL_SCHED_H_
#define _TD_UTIL_SCHED_H #define _TD_UTIL_SCHED_H_
#include "os.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
@ -36,7 +38,7 @@ typedef struct SSchedMsg {
* @param label the label of the queue * @param label the label of the queue
* @return the created queue scheduler * @return the created queue scheduler
*/ */
void *taosInitScheduler(int capacity, int numOfThreads, const char *label); void *taosInitScheduler(int32_t capacity, int32_t numOfThreads, const char *label);
/** /**
* Create a thread-safe ring-buffer based task queue and return the instance. * Create a thread-safe ring-buffer based task queue and return the instance.
@ -47,7 +49,7 @@ void *taosInitScheduler(int capacity, int numOfThreads, const char *label);
* @param tmrCtrl the timer controller, tmr_ctrl_t* * @param tmrCtrl the timer controller, tmr_ctrl_t*
* @return the created queue scheduler * @return the created queue scheduler
*/ */
void *taosInitSchedulerWithInfo(int capacity, int numOfThreads, const char *label, void *tmrCtrl); void *taosInitSchedulerWithInfo(int32_t capacity, int32_t numOfThreads, const char *label, void *tmrCtrl);
/** /**
* Clean up the queue scheduler instance and free the memory. * Clean up the queue scheduler instance and free the memory.
@ -68,4 +70,4 @@ void taosScheduleTask(void *queueScheduler, SSchedMsg *pMsg);
} }
#endif #endif
#endif /*_TD_UTIL_SCHED_H*/ #endif /*_TD_UTIL_SCHED_H_*/

16
include/util/tskiplist.h
View File

@ -16,15 +16,15 @@
#ifndef _TD_UTIL_SKILIST_H #ifndef _TD_UTIL_SKILIST_H
#define _TD_UTIL_SKILIST_H #define _TD_UTIL_SKILIST_H
#ifdef __cplusplus
extern "C" {
#endif
#include "os.h" #include "os.h"
#include "taos.h" #include "taos.h"
#include "tarray.h" #include "tarray.h"
#include "tfunctional.h" #include "tfunctional.h"
#ifdef __cplusplus
extern "C" {
#endif
#define MAX_SKIP_LIST_LEVEL 15 #define MAX_SKIP_LIST_LEVEL 15
#define SKIP_LIST_RECORD_PERFORMANCE 0 #define SKIP_LIST_RECORD_PERFORMANCE 0
@ -100,14 +100,10 @@ typedef struct tSkipListState {
uint64_t nTotalElapsedTimeForInsert; uint64_t nTotalElapsedTimeForInsert;
} tSkipListState; } tSkipListState;
typedef enum { typedef enum { SSkipListPutSuccess = 0, SSkipListPutEarlyStop = 1, SSkipListPutSkipOne = 2 } SSkipListPutStatus;
SSkipListPutSuccess = 0,
SSkipListPutEarlyStop = 1,
SSkipListPutSkipOne = 2
} SSkipListPutStatus;
typedef struct SSkipList { typedef struct SSkipList {
unsigned int seed; uint32_t seed;
__compar_fn_t comparFn; __compar_fn_t comparFn;
__sl_key_fn_t keyFn; __sl_key_fn_t keyFn;
pthread_rwlock_t *lock; pthread_rwlock_t *lock;

33
source/util/src/tsched.c
View File

@ -13,11 +13,12 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "tdef.h" #define _DEFAULT_SOURCE
#include "tutil.h"
#include "tlog.h"
#include "tsched.h" #include "tsched.h"
#include "tdef.h"
#include "tlog.h"
#include "ttimer.h" #include "ttimer.h"
#include "tutil.h"
#define DUMP_SCHEDULER_TIME_WINDOW 30000 // every 30sec, take a snap shot of task queue. #define DUMP_SCHEDULER_TIME_WINDOW 30000 // every 30sec, take a snap shot of task queue.
@ -26,10 +27,10 @@ typedef struct {
tsem_t emptySem; tsem_t emptySem;
tsem_t fullSem; tsem_t fullSem;
pthread_mutex_t queueMutex; pthread_mutex_t queueMutex;
int fullSlot; int32_t fullSlot;
int emptySlot; int32_t emptySlot;
int queueSize; int32_t queueSize;
int numOfThreads; int32_t numOfThreads;
pthread_t *qthread; pthread_t *qthread;
SSchedMsg *queue; SSchedMsg *queue;
bool stop; bool stop;
@ -40,7 +41,7 @@ typedef struct {
static void *taosProcessSchedQueue(void *param); static void *taosProcessSchedQueue(void *param);
static void taosDumpSchedulerStatus(void *qhandle, void *tmrId); static void taosDumpSchedulerStatus(void *qhandle, void *tmrId);
void *taosInitScheduler(int queueSize, int numOfThreads, const char *label) { void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *label) {
SSchedQueue *pSched = (SSchedQueue *)calloc(sizeof(SSchedQueue), 1); SSchedQueue *pSched = (SSchedQueue *)calloc(sizeof(SSchedQueue), 1);
if (pSched == NULL) { if (pSched == NULL) {
uError("%s: no enough memory for pSched", label); uError("%s: no enough memory for pSched", label);
@ -73,7 +74,7 @@ void *taosInitScheduler(int queueSize, int numOfThreads, const char *label) {
return NULL; return NULL;
} }
if (tsem_init(&pSched->emptySem, 0, (unsigned int)pSched->queueSize) != 0) { if (tsem_init(&pSched->emptySem, 0, (uint32_t)pSched->queueSize) != 0) {
uError("init %s:empty semaphore failed(%s)", label, strerror(errno)); uError("init %s:empty semaphore failed(%s)", label, strerror(errno));
taosCleanUpScheduler(pSched); taosCleanUpScheduler(pSched);
return NULL; return NULL;
@ -86,11 +87,11 @@ void *taosInitScheduler(int queueSize, int numOfThreads, const char *label) {
} }
pSched->stop = false; pSched->stop = false;
for (int i = 0; i < numOfThreads; ++i) { for (int32_t i = 0; i < numOfThreads; ++i) {
pthread_attr_t attr; pthread_attr_t attr;
pthread_attr_init(&attr); pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE); pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
int code = pthread_create(pSched->qthread + i, &attr, taosProcessSchedQueue, (void *)pSched); int32_t code = pthread_create(pSched->qthread + i, &attr, taosProcessSchedQueue, (void *)pSched);
pthread_attr_destroy(&attr); pthread_attr_destroy(&attr);
if (code != 0) { if (code != 0) {
uError("%s: failed to create rpc thread(%s)", label, strerror(errno)); uError("%s: failed to create rpc thread(%s)", label, strerror(errno));
@ -105,7 +106,7 @@ void *taosInitScheduler(int queueSize, int numOfThreads, const char *label) {
return (void *)pSched; return (void *)pSched;
} }
void *taosInitSchedulerWithInfo(int queueSize, int numOfThreads, const char *label, void *tmrCtrl) { void *taosInitSchedulerWithInfo(int32_t queueSize, int32_t numOfThreads, const char *label, void *tmrCtrl) {
SSchedQueue *pSched = taosInitScheduler(queueSize, numOfThreads, label); SSchedQueue *pSched = taosInitScheduler(queueSize, numOfThreads, label);
if (tmrCtrl != NULL && pSched != NULL) { if (tmrCtrl != NULL && pSched != NULL) {
@ -119,7 +120,7 @@ void *taosInitSchedulerWithInfo(int queueSize, int numOfThreads, const char *lab
void *taosProcessSchedQueue(void *scheduler) { void *taosProcessSchedQueue(void *scheduler) {
SSchedMsg msg; SSchedMsg msg;
SSchedQueue *pSched = (SSchedQueue *)scheduler; SSchedQueue *pSched = (SSchedQueue *)scheduler;
int ret = 0; int32_t ret = 0;
char name[16] = {0}; char name[16] = {0};
snprintf(name, tListLen(name), "%s-taskQ", pSched->label); snprintf(name, tListLen(name), "%s-taskQ", pSched->label);
@ -164,7 +165,7 @@ void *taosProcessSchedQueue(void *scheduler) {
void taosScheduleTask(void *queueScheduler, SSchedMsg *pMsg) { void taosScheduleTask(void *queueScheduler, SSchedMsg *pMsg) {
SSchedQueue *pSched = (SSchedQueue *)queueScheduler; SSchedQueue *pSched = (SSchedQueue *)queueScheduler;
int ret = 0; int32_t ret = 0;
if (pSched == NULL) { if (pSched == NULL) {
uError("sched is not ready, msg:%p is dropped", pMsg); uError("sched is not ready, msg:%p is dropped", pMsg);
@ -200,12 +201,12 @@ void taosCleanUpScheduler(void *param) {
if (pSched == NULL) return; if (pSched == NULL) return;
pSched->stop = true; pSched->stop = true;
for (int i = 0; i < pSched->numOfThreads; ++i) { for (int32_t i = 0; i < pSched->numOfThreads; ++i) {
if (taosCheckPthreadValid(pSched->qthread[i])) { if (taosCheckPthreadValid(pSched->qthread[i])) {
tsem_post(&pSched->fullSem); tsem_post(&pSched->fullSem);
} }
} }
for (int i = 0; i < pSched->numOfThreads; ++i) { for (int32_t i = 0; i < pSched->numOfThreads; ++i) {
if (taosCheckPthreadValid(pSched->qthread[i])) { if (taosCheckPthreadValid(pSched->qthread[i])) {
pthread_join(pSched->qthread[i], NULL); pthread_join(pSched->qthread[i], NULL);
} }

43
source/util/src/tskiplist.c
View File

@ -14,12 +14,13 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "tcompare.h" #define _DEFAULT_SOURCE
#include "tskiplist.h" #include "tskiplist.h"
#include "tutil.h" #include "tcompare.h"
#include "tlog.h" #include "tlog.h"
#include "tutil.h"
static int initForwardBackwardPtr(SSkipList *pSkipList); static int32_t initForwardBackwardPtr(SSkipList *pSkipList);
static SSkipListNode *getPriorNode(SSkipList *pSkipList, const char *val, int32_t order, SSkipListNode **pCur); static SSkipListNode *getPriorNode(SSkipList *pSkipList, const char *val, int32_t order, SSkipListNode **pCur);
static void tSkipListRemoveNodeImpl(SSkipList *pSkipList, SSkipListNode *pNode); static void tSkipListRemoveNodeImpl(SSkipList *pSkipList, SSkipListNode *pNode);
static void tSkipListCorrectLevel(SSkipList *pSkipList); static void tSkipListCorrectLevel(SSkipList *pSkipList);
@ -31,10 +32,9 @@ static SSkipListNode *tSkipListNewNode(uint8_t level);
static SSkipListNode *tSkipListPutImpl(SSkipList *pSkipList, void *pData, SSkipListNode **direction, bool isForward, static SSkipListNode *tSkipListPutImpl(SSkipList *pSkipList, void *pData, SSkipListNode **direction, bool isForward,
bool hasDup); bool hasDup);
static FORCE_INLINE int32_t tSkipListWLock(SSkipList *pSkipList);
static FORCE_INLINE int tSkipListWLock(SSkipList *pSkipList); static FORCE_INLINE int32_t tSkipListRLock(SSkipList *pSkipList);
static FORCE_INLINE int tSkipListRLock(SSkipList *pSkipList); static FORCE_INLINE int32_t tSkipListUnlock(SSkipList *pSkipList);
static FORCE_INLINE int tSkipListUnlock(SSkipList *pSkipList);
static FORCE_INLINE int32_t getSkipListRandLevel(SSkipList *pSkipList); static FORCE_INLINE int32_t getSkipListRandLevel(SSkipList *pSkipList);
SSkipList *tSkipListCreate(uint8_t maxLevel, uint8_t keyType, uint16_t keyLen, __compar_fn_t comparFn, uint8_t flags, SSkipList *tSkipListCreate(uint8_t maxLevel, uint8_t keyType, uint16_t keyLen, __compar_fn_t comparFn, uint8_t flags,
@ -140,7 +140,7 @@ void tSkipListPutBatchByIter(SSkipList *pSkipList, void *iter, iter_next_fn_t it
bool hasDup = false; bool hasDup = false;
char *pKey = NULL; char *pKey = NULL;
char *pDataKey = NULL; char *pDataKey = NULL;
int compare = 0; int32_t compare = 0;
tSkipListWLock(pSkipList); tSkipListWLock(pSkipList);
@ -152,7 +152,7 @@ void tSkipListPutBatchByIter(SSkipList *pSkipList, void *iter, iter_next_fn_t it
tSkipListPutImpl(pSkipList, pData, backward, false, hasDup); tSkipListPutImpl(pSkipList, pData, backward, false, hasDup);
for (int level = 0; level < pSkipList->maxLevel; level++) { for (int32_t level = 0; level < pSkipList->maxLevel; level++) {
forward[level] = SL_NODE_GET_BACKWARD_POINTER(backward[level], level); forward[level] = SL_NODE_GET_BACKWARD_POINTER(backward[level], level);
} }
@ -165,12 +165,12 @@ void tSkipListPutBatchByIter(SSkipList *pSkipList, void *iter, iter_next_fn_t it
pKey = SL_GET_MAX_KEY(pSkipList); pKey = SL_GET_MAX_KEY(pSkipList);
compare = pSkipList->comparFn(pDataKey, pKey); compare = pSkipList->comparFn(pDataKey, pKey);
if (compare > 0) { if (compare > 0) {
for (int i = 0; i < pSkipList->maxLevel; i++) { for (int32_t i = 0; i < pSkipList->maxLevel; i++) {
forward[i] = SL_NODE_GET_BACKWARD_POINTER(pSkipList->pTail, i); forward[i] = SL_NODE_GET_BACKWARD_POINTER(pSkipList->pTail, i);
} }
} else { } else {
SSkipListNode *px = pSkipList->pHead; SSkipListNode *px = pSkipList->pHead;
for (int i = pSkipList->maxLevel - 1; i >= 0; --i) { for (int32_t i = pSkipList->maxLevel - 1; i >= 0; --i) {
if (i < pSkipList->level) { if (i < pSkipList->level) {
// set new px // set new px
if (forward[i] != pSkipList->pHead) { if (forward[i] != pSkipList->pHead) {
@ -433,21 +433,21 @@ static SSkipListIterator *doCreateSkipListIterator(SSkipList *pSkipList, int32_t
return iter; return iter;
} }
static FORCE_INLINE int tSkipListWLock(SSkipList *pSkipList) { static FORCE_INLINE int32_t tSkipListWLock(SSkipList *pSkipList) {
if (pSkipList->lock) { if (pSkipList->lock) {
return pthread_rwlock_wrlock(pSkipList->lock); return pthread_rwlock_wrlock(pSkipList->lock);
} }
return 0; return 0;
} }
static FORCE_INLINE int tSkipListRLock(SSkipList *pSkipList) { static FORCE_INLINE int32_t tSkipListRLock(SSkipList *pSkipList) {
if (pSkipList->lock) { if (pSkipList->lock) {
return pthread_rwlock_rdlock(pSkipList->lock); return pthread_rwlock_rdlock(pSkipList->lock);
} }
return 0; return 0;
} }
static FORCE_INLINE int tSkipListUnlock(SSkipList *pSkipList) { static FORCE_INLINE int32_t tSkipListUnlock(SSkipList *pSkipList) {
if (pSkipList->lock) { if (pSkipList->lock) {
return pthread_rwlock_unlock(pSkipList->lock); return pthread_rwlock_unlock(pSkipList->lock);
} }
@ -455,12 +455,12 @@ static FORCE_INLINE int tSkipListUnlock(SSkipList *pSkipList) {
} }
static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward, void *pData) { static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward, void *pData) {
int compare = 0; int32_t compare = 0;
bool hasDupKey = false; bool hasDupKey = false;
char *pDataKey = pSkipList->keyFn(pData); char *pDataKey = pSkipList->keyFn(pData);
if (pSkipList->size == 0) { if (pSkipList->size == 0) {
for (int i = 0; i < pSkipList->maxLevel; i++) { for (int32_t i = 0; i < pSkipList->maxLevel; i++) {
backward[i] = pSkipList->pTail; backward[i] = pSkipList->pTail;
} }
} else { } else {
@ -470,7 +470,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
pKey = SL_GET_MAX_KEY(pSkipList); pKey = SL_GET_MAX_KEY(pSkipList);
compare = pSkipList->comparFn(pDataKey, pKey); compare = pSkipList->comparFn(pDataKey, pKey);
if (compare >= 0) { if (compare >= 0) {
for (int i = 0; i < pSkipList->maxLevel; i++) { for (int32_t i = 0; i < pSkipList->maxLevel; i++) {
backward[i] = pSkipList->pTail; backward[i] = pSkipList->pTail;
} }
@ -481,7 +481,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
pKey = SL_GET_MIN_KEY(pSkipList); pKey = SL_GET_MIN_KEY(pSkipList);
compare = pSkipList->comparFn(pDataKey, pKey); compare = pSkipList->comparFn(pDataKey, pKey);
if (compare < 0) { if (compare < 0) {
for (int i = 0; i < pSkipList->maxLevel; i++) { for (int32_t i = 0; i < pSkipList->maxLevel; i++) {
backward[i] = SL_NODE_GET_FORWARD_POINTER(pSkipList->pHead, i); backward[i] = SL_NODE_GET_FORWARD_POINTER(pSkipList->pHead, i);
} }
@ -489,7 +489,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
} }
SSkipListNode *px = pSkipList->pTail; SSkipListNode *px = pSkipList->pTail;
for (int i = pSkipList->maxLevel - 1; i >= 0; --i) { for (int32_t i = pSkipList->maxLevel - 1; i >= 0; --i) {
if (i < pSkipList->level) { if (i < pSkipList->level) {
SSkipListNode *p = SL_NODE_GET_BACKWARD_POINTER(px, i); SSkipListNode *p = SL_NODE_GET_BACKWARD_POINTER(px, i);
while (p != pSkipList->pHead) { while (p != pSkipList->pHead) {
@ -532,7 +532,8 @@ static void tSkipListRemoveNodeImpl(SSkipList *pSkipList, SSkipListNode *pNode)
// Function must be called after calling tSkipListRemoveNodeImpl() function // Function must be called after calling tSkipListRemoveNodeImpl() function
static void tSkipListCorrectLevel(SSkipList *pSkipList) { static void tSkipListCorrectLevel(SSkipList *pSkipList) {
while (pSkipList->level > 0 && SL_NODE_GET_FORWARD_POINTER(pSkipList->pHead, pSkipList->level - 1) == pSkipList->pTail) { while (pSkipList->level > 0 &&
SL_NODE_GET_FORWARD_POINTER(pSkipList->pHead, pSkipList->level - 1) == pSkipList->pTail) {
pSkipList->level -= 1; pSkipList->level -= 1;
} }
} }
@ -636,7 +637,7 @@ static SSkipListNode *getPriorNode(SSkipList *pSkipList, const char *val, int32_
return pNode; return pNode;
} }
static int initForwardBackwardPtr(SSkipList *pSkipList) { static int32_t initForwardBackwardPtr(SSkipList *pSkipList) {
uint32_t maxLevel = pSkipList->maxLevel; uint32_t maxLevel = pSkipList->maxLevel;
// head info // head info