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

feat(tmq): add push mode

This commit is contained in:
Liu Jicong 2022-04-22 19:47:00 +08:00
parent 3035ecd58e
commit 32e9887664
14 changed files with 441 additions and 223 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
example/src/tmq.c
View File

@ -101,8 +101,6 @@ int32_t create_topic() {
}
taos_free_result(pRes);
/*const char* sql = "select * from tu1";*/
/*pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));*/
/*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/
pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");
if (taos_errno(pRes) != 0) {

2
include/client/taos.h
View File

@ -245,6 +245,8 @@ typedef void(tmq_commit_cb(tmq_t *, tmq_resp_err_t, tmq_topic_vgroup_list_t *));
DLL_EXPORT tmq_list_t *tmq_list_new();
DLL_EXPORT int32_t tmq_list_append(tmq_list_t *, const char *);
DLL_EXPORT void tmq_list_destroy(tmq_list_t *);
DLL_EXPORT int32_t tmq_list_get_size(const tmq_list_t *);
DLL_EXPORT char **tmq_list_to_c_array(const tmq_list_t *);
#if 0
DLL_EXPORT tmq_t *tmq_consumer_new(void *conn, tmq_conf_t *conf, char *errstr, int32_t errstrLen);

7
include/os/osSemaphore.h
View File

@ -23,17 +23,24 @@ extern "C" {
#include <semaphore.h>
#if defined(_TD_DARWIN_64)
typedef struct tsem_s *tsem_t;
int tsem_init(tsem_t *sem, int pshared, unsigned int value);
int tsem_wait(tsem_t *sem);
int tsem_timewait(tsem_t *sim, int64_t nanosecs);
int tsem_post(tsem_t *sem);
int tsem_destroy(tsem_t *sem);
#else
#define tsem_t sem_t
#define tsem_init sem_init
int tsem_wait(tsem_t *sem);
int tsem_timewait(tsem_t *sim, int64_t nanosecs);
#define tsem_post sem_post
#define tsem_destroy sem_destroy
#endif
#if defined(_TD_DARWIN_64)

4
include/util/tcache.h
View File

@ -47,13 +47,13 @@ typedef struct STrashElem STrashElem;
/**
* initialize the cache object
* @param keyType key type
* @param refreshTimeInSeconds refresh operation interval time, the maximum survival time when one element is expired
* @param refreshTimeInMs refresh operation interval time, the maximum survival time when one element is expired
* and not referenced by other objects
* @param extendLifespan auto extend lifespan, if accessed
* @param fn free resource callback function
* @return
*/
SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInSeconds, bool extendLifespan, __cache_free_fn_t fn,
SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInMs, bool extendLifespan, __cache_free_fn_t fn,
const char *cacheName);
/**

12
source/client/src/tmq.c
View File

@ -260,6 +260,16 @@ void tmq_list_destroy(tmq_list_t* list) {
taosArrayDestroy(container);
}
int32_t tmq_list_get_size(const tmq_list_t* list) {
const SArray* container = &list->container;
return taosArrayGetSize(container);
}
char** tmq_list_to_c_array(const tmq_list_t* list) {
const SArray* container = &list->container;
return container->pData;
}
static int32_t tmqMakeTopicVgKey(char* dst, const char* topicName, int32_t vg) {
return sprintf(dst, "%s:%d", topicName, vg);
}
@ -387,7 +397,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
pTmq->commit_cb = conf->commit_cb;
pTmq->resetOffsetCfg = conf->resetOffset;
pTmq->consumerId = generateRequestId() & (((uint64_t)-1) >> 1);
pTmq->consumerId = tGenIdPI64();
pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
if (pTmq->clientTopics == NULL) {
taosMemoryFree(pTmq);

2
source/dnode/mnode/impl/src/mndPerfSchema.c
View File

@ -49,6 +49,7 @@ static const SPerfsTableSchema topicSchema[] = {
static const SPerfsTableSchema consumerSchema[] = {
{.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "app_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "status", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
@ -61,6 +62,7 @@ static const SPerfsTableSchema subscribeSchema[] = {
{.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
{.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
};

9
source/dnode/mnode/impl/src/mndProfile.c
View File

@ -58,7 +58,8 @@ static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter);
int32_t mndInitProfile(SMnode *pMnode) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
int32_t connCheckTime = tsShellActivityTimer * 2;
// in ms
int32_t connCheckTime = tsShellActivityTimer * 2 * 1000;
pMgmt->cache = taosCacheInit(TSDB_DATA_TYPE_INT, connCheckTime, true, (__cache_free_fn_t)mndFreeConn, "conn");
if (pMgmt->cache == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -324,7 +325,8 @@ static SClientHbRsp *mndMqHbBuildRsp(SMnode *pMnode, SClientHbReq *pReq) {
return NULL;
}
static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHbReq *pHbReq, SClientHbBatchRsp *pBatchRsp) {
static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHbReq *pHbReq,
SClientHbBatchRsp *pBatchRsp) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SClientHbRsp hbRsp = {.connKey = pHbReq->connKey, .status = 0, .info = NULL, .query = NULL};
@ -336,7 +338,8 @@ static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHb
SConnObj *pConn = mndAcquireConn(pMnode, pBasic->connId);
if (pConn == NULL) {
pConn = mndCreateConn(pMnode, connInfo.user, CONN_TYPE__QUERY, connInfo.clientIp, connInfo.clientPort, pBasic->pid, pBasic->app, 0);
pConn = mndCreateConn(pMnode, connInfo.user, CONN_TYPE__QUERY, connInfo.clientIp, connInfo.clientPort,
pBasic->pid, pBasic->app, 0);
if (pConn == NULL) {
mError("user:%s, conn:%u is freed and failed to create new since %s", connInfo.user, pBasic->connId, terrstr());
return -1;

2
source/dnode/mnode/impl/src/mndShow.c
View File

@ -28,7 +28,7 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq);
int32_t mndInitShow(SMnode *pMnode) {
SShowMgmt *pMgmt = &pMnode->showMgmt;
pMgmt->cache = taosCacheInit(TSDB_DATA_TYPE_INT, 5, true, (__cache_free_fn_t)mndFreeShowObj, "show");
pMgmt->cache = taosCacheInit(TSDB_DATA_TYPE_INT, 5000, true, (__cache_free_fn_t)mndFreeShowObj, "show");
if (pMgmt->cache == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("failed to alloc show cache since %s", terrstr());

4
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -534,8 +534,8 @@ static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB
colDataAppend(pColInfo, numOfRows, (const char *)&pTopic->createTime, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
char *sql = taosMemoryCalloc(1, strlen(pTopic->sql) + 1 + VARSTR_HEADER_SIZE);
strcpy(&sql[VARSTR_HEADER_SIZE], pTopic->sql);
char sql[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0};
tstrncpy(&sql[VARSTR_HEADER_SIZE], pTopic->sql, TSDB_SHOW_SQL_LEN);
varDataSetLen(sql, strlen(&sql[VARSTR_HEADER_SIZE]));
colDataAppend(pColInfo, numOfRows, (const char *)sql, false);

21
source/dnode/vnode/src/inc/tq.h
View File

@ -18,8 +18,10 @@
#include "executor.h"
#include "os.h"
#include "tcache.h"
#include "thash.h"
#include "tmsg.h"
#include "trpc.h"
#include "ttimer.h"
#include "wal.h"
@ -141,6 +143,15 @@ typedef struct {
FTqDelete pDeleter;
} STqMetaStore;
typedef struct {
int64_t consumerId;
int32_t epoch;
int32_t skipLogNum;
int64_t reqOffset;
SRWLatch lock;
SRpcMsg* handle;
} STqPushHandle;
typedef struct {
char subKey[TSDB_SUBSCRIBE_KEY_LEN];
int64_t consumerId;
@ -151,17 +162,19 @@ typedef struct {
int8_t withTag;
int8_t withTagSchema;
char* qmsg;
STqPushHandle pushHandle;
// SRWLatch lock;
SWalReadHandle* pWalReader;
// number should be identical to fetch thread num
STqReadHandle* pStreamReader[4];
qTaskInfo_t task[4];
// task number should be the same with fetch thread
STqReadHandle* pExecReader[5];
qTaskInfo_t task[5];
} STqExec;
struct STQ {
char* path;
// STqMetaStore* tqMeta;
SHashObj* execs; // subKey -> tqExec
SHashObj* pushMgr; // consumerId -> STqExec*
SHashObj* execs; // subKey -> STqExec
SHashObj* pStreamTasks;
SVnode* pVnode;
SWal* pWal;

167
source/dnode/vnode/src/tq/tq.c
View File

@ -41,6 +41,8 @@ STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) {
pTq->pStreamTasks = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
pTq->pushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
return pTq;
}
@ -52,8 +54,139 @@ void tqClose(STQ* pTq) {
// TODO
}
int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
if (msgType != TDMT_VND_SUBMIT) return 0;
void* pIter = NULL;
STqExec* pExec = NULL;
SSubmitReq* pReq = (SSubmitReq*)msg;
int32_t workerId = 4;
int64_t fetchOffset = ver;
while (1) {
pIter = taosHashIterate(pTq->pushMgr, pIter);
if (pIter == NULL) break;
pExec = (STqExec**)pIter;
taosWLockLatch(&pExec->pushHandle.lock);
SRpcMsg* pMsg = atomic_load_ptr(&pExec->pushHandle.handle);
ASSERT(pMsg);
SMqDataBlkRsp rsp = {0};
rsp.reqOffset = pExec->pushHandle.reqOffset;
rsp.blockData = taosArrayInit(0, sizeof(int32_t));
rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
qTaskInfo_t task = pExec->task[workerId];
ASSERT(task);
qSetStreamInput(task, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK);
while (1) {
SSDataBlock* pDataBlock = NULL;
uint64_t ts = 0;
if (qExecTask(task, &pDataBlock, &ts) < 0) {
ASSERT(0);
}
if (pDataBlock == NULL) break;
ASSERT(pDataBlock->info.rows != 0);
ASSERT(pDataBlock->info.numOfCols != 0);
int32_t dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pDataBlock);
void* buf = taosMemoryCalloc(1, dataStrLen);
SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
pRetrieve->useconds = ts;
pRetrieve->precision = TSDB_DEFAULT_PRECISION;
pRetrieve->compressed = 0;
pRetrieve->completed = 1;
pRetrieve->numOfRows = htonl(pDataBlock->info.rows);
// TODO enable compress
int32_t actualLen = 0;
blockCompressEncode(pDataBlock, pRetrieve->data, &actualLen, pDataBlock->info.numOfCols, false);
actualLen += sizeof(SRetrieveTableRsp);
ASSERT(actualLen <= dataStrLen);
taosArrayPush(rsp.blockDataLen, &actualLen);
taosArrayPush(rsp.blockData, &buf);
rsp.blockNum++;
}
} else if (pExec->subType == TOPIC_SUB_TYPE__DB) {
STqReadHandle* pReader = pExec->pExecReader[workerId];
tqReadHandleSetMsg(pReader, pReq, 0);
while (tqNextDataBlock(pReader)) {
SSDataBlock block = {0};
if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.rows,
&block.info.numOfCols) < 0) {
ASSERT(0);
}
int32_t dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(&block);
void* buf = taosMemoryCalloc(1, dataStrLen);
SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
/*pRetrieve->useconds = 0;*/
pRetrieve->precision = TSDB_DEFAULT_PRECISION;
pRetrieve->compressed = 0;
pRetrieve->completed = 1;
pRetrieve->numOfRows = htonl(block.info.rows);
// TODO enable compress
int32_t actualLen = 0;
blockCompressEncode(&block, pRetrieve->data, &actualLen, block.info.numOfCols, false);
actualLen += sizeof(SRetrieveTableRsp);
ASSERT(actualLen <= dataStrLen);
taosArrayPush(rsp.blockDataLen, &actualLen);
taosArrayPush(rsp.blockData, &buf);
rsp.blockNum++;
}
} else {
ASSERT(0);
}
if (rsp.blockNum == 0) {
taosWUnLockLatch(&pExec->pushHandle.lock);
continue;
}
ASSERT(taosArrayGetSize(rsp.blockData) == rsp.blockNum);
ASSERT(taosArrayGetSize(rsp.blockDataLen) == rsp.blockNum);
rsp.rspOffset = fetchOffset;
int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqDataBlkRsp(NULL, &rsp);
void* buf = rpcMallocCont(tlen);
if (buf == NULL) {
pMsg->code = -1;
return -1;
}
((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
((SMqRspHead*)buf)->epoch = pExec->pushHandle.epoch;
((SMqRspHead*)buf)->consumerId = pExec->pushHandle.consumerId;
void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
tEncodeSMqDataBlkRsp(&abuf, &rsp);
pMsg->pCont = buf;
pMsg->contLen = tlen;
pMsg->code = 0;
tmsgSendRsp(pMsg);
atomic_store_ptr(&pExec->pushHandle.handle, NULL);
taosWUnLockLatch(&pExec->pushHandle.lock);
vDebug("vg %d offset %ld from consumer %ld (epoch %d) send rsp, block num: %d, reqOffset: %ld, rspOffset: %ld",
TD_VID(pTq->pVnode), fetchOffset, pExec->pushHandle.consumerId, pExec->pushHandle.epoch, rsp.blockNum,
rsp.reqOffset, rsp.rspOffset);
// TODO destroy
taosArrayDestroy(rsp.blockData);
taosArrayDestroy(rsp.blockDataLen);
}
return 0;
}
int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t version) {
if (msgType != TDMT_VND_SUBMIT) return 0;
void* data = taosMemoryMalloc(msgLen);
if (data == NULL) {
return -1;
@ -71,6 +204,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t versi
.pCont = data,
.contLen = msgLen,
};
tmsgPutToQueue(&pTq->pVnode->msgCb, FETCH_QUEUE, &req);
#if 0
@ -240,6 +374,7 @@ int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsu
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
SMqPollReqV2* pReq = pMsg->pCont;
int64_t consumerId = pReq->consumerId;
int64_t waitTime = pReq->blockingTime;
int32_t reqEpoch = pReq->epoch;
int64_t fetchOffset;
@ -265,8 +400,8 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
SMqDataBlkRsp rsp = {0};
rsp.reqOffset = pReq->currentOffset;
rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
rsp.blockData = taosArrayInit(0, sizeof(void*));
rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
while (1) {
consumerEpoch = atomic_load_32(&pExec->epoch);
@ -283,6 +418,28 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
// response to user
vDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, no more log to return", consumerId, pReq->epoch,
TD_VID(pTq->pVnode), fetchOffset);
#if 0
// add to pushMgr
taosWLockLatch(&pExec->pushHandle.lock);
pExec->pushHandle.consumerId = consumerId;
pExec->pushHandle.epoch = reqEpoch;
pExec->pushHandle.reqOffset = rsp.reqOffset;
pExec->pushHandle.skipLogNum = rsp.skipLogNum;
pExec->pushHandle.handle = pMsg;
taosWUnLockLatch(&pExec->pushHandle.lock);
// TODO add timer
// TODO: the pointer will always be valid?
taosHashPut(pTq->pushMgr, &consumerId, sizeof(int64_t), &pExec, sizeof(void*));
taosArrayDestroy(rsp.blockData);
taosArrayDestroy(rsp.blockDataLen);
return 0;
#endif
break;
}
@ -325,7 +482,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
rsp.blockNum++;
}
} else if (pExec->subType == TOPIC_SUB_TYPE__DB) {
STqReadHandle* pReader = pExec->pStreamReader[workerId];
STqReadHandle* pReader = pExec->pExecReader[workerId];
tqReadHandleSetMsg(pReader, pCont, 0);
while (tqNextDataBlock(pReader)) {
SSDataBlock block = {0};
@ -635,10 +792,10 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
req.qmsg = NULL;
pExec->pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
for (int32_t i = 0; i < 4; i++) {
pExec->pStreamReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
for (int32_t i = 0; i < 5; i++) {
pExec->pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
SReadHandle handle = {
.reader = pExec->pStreamReader[i],
.reader = pExec->pExecReader[i],
.meta = pTq->pVnode->pMeta,
};
pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);

64
source/os/src/osSemaphore.c
View File

@ -140,7 +140,8 @@ struct tsem_s {
int tsem_init(tsem_t *sem, int pshared, unsigned int value) {
// fprintf(stderr, "==%s[%d]%s():[%p]==creating\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
if (*sem) {
fprintf(stderr, "==%s[%d]%s():[%p]==already initialized\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==already initialized\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
struct tsem_s *p = (struct tsem_s *)taosMemoryCalloc(1, sizeof(*p));
@ -180,20 +181,22 @@ int tsem_init(tsem_t *sem, int pshared, unsigned int value) {
int e = errno;
if (e == EEXIST) continue;
if (e == EINTR) continue;
fprintf(stderr, "==%s[%d]%s():[%p]==not created[%d]%s\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem, e,
strerror(e));
fprintf(stderr, "==%s[%d]%s():[%p]==not created[%d]%s\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem,
e, strerror(e));
abort();
} while (p->sem == SEM_FAILED);
#elif defined(SEM_USE_SEM)
taosThreadOnce(&sem_once, once_init);
if (sem_inited != 1) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal resource init failed\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal resource init failed\n", taosDirEntryBaseName(__FILE__), __LINE__,
__func__, sem);
errno = ENOMEM;
return -1;
}
kern_return_t ret = semaphore_create(sem_port, &p->sem, SYNC_POLICY_FIFO, value);
if (ret != KERN_SUCCESS) {
fprintf(stderr, "==%s[%d]%s():[%p]==semophore_create failed\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==semophore_create failed\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
// we fail-fast here, because we have less-doc about semaphore_create for the moment
abort();
}
@ -224,18 +227,21 @@ int tsem_wait(tsem_t *sem) {
}
#ifdef SEM_USE_PTHREAD
if (taosThreadMutexLock(&p->lock)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
p->val -= 1;
if (p->val < 0) {
if (taosThreadCondWait(&p->cond, &p->lock)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
}
if (taosThreadMutexUnlock(&p->lock)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
return 0;
@ -260,18 +266,21 @@ int tsem_post(tsem_t *sem) {
}
#ifdef SEM_USE_PTHREAD
if (taosThreadMutexLock(&p->lock)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
p->val += 1;
if (p->val <= 0) {
if (taosThreadCondSignal(&p->cond)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
}
if (taosThreadMutexUnlock(&p->lock)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
return 0;
@ -293,26 +302,30 @@ int tsem_destroy(tsem_t *sem) {
}
struct tsem_s *p = *sem;
if (!p->valid) {
// fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
// abort();
// fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
// sem); abort();
return 0;
}
#ifdef SEM_USE_PTHREAD
if (taosThreadMutexLock(&p->lock)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
p->valid = 0;
if (taosThreadCondDestroy(&p->cond)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
if (taosThreadMutexUnlock(&p->lock)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
if (taosThreadMutexDestroy(&p->lock)) {
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
sem);
abort();
}
#elif defined(SEM_USE_POSIX)
@ -321,8 +334,8 @@ int tsem_destroy(tsem_t *sem) {
int r = sem_unlink(name);
if (r) {
int e = errno;
fprintf(stderr, "==%s[%d]%s():[%p]==unlink failed[%d]%s\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem, e,
strerror(e));
fprintf(stderr, "==%s[%d]%s():[%p]==unlink failed[%d]%s\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem,
e, strerror(e));
abort();
}
#elif defined(SEM_USE_SEM)
@ -424,4 +437,17 @@ int32_t tsem_wait(tsem_t* sem) {
return ret;
}
int32_t tsem_timewait(tsem_t* sem, int64_t nanosecs) {
int ret = 0;
struct timespec tv = {
.tv_sec = 0,
.tv_nsec = nanosecs,
};
while ((ret = sem_timedwait(sem, &tv)) == -1 && errno == EINTR) continue;
return ret;
}
#endif

14
source/util/src/tcache.c
View File

@ -285,8 +285,8 @@ static FORCE_INLINE SCacheEntry* doFindEntry(SCacheObj* pCacheObj, const void* k
return &pCacheObj->pEntryList[slot];
}
static FORCE_INLINE SCacheNode *
doSearchInEntryList(SCacheEntry *pe, const void *key, size_t keyLen, SCacheNode** prev) {
static FORCE_INLINE SCacheNode *doSearchInEntryList(SCacheEntry *pe, const void *key, size_t keyLen,
SCacheNode **prev) {
SCacheNode *pNode = pe->next;
while (pNode) {
if ((pNode->keyLen == keyLen) && memcmp(pNode->key, key, keyLen) == 0) {
@ -350,11 +350,11 @@ static FORCE_INLINE int32_t getCacheCapacity(int32_t length) {
return len > CACHE_MAX_CAPACITY ? CACHE_MAX_CAPACITY : len;
}
SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInSeconds, bool extendLifespan, __cache_free_fn_t fn,
SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInMs, bool extendLifespan, __cache_free_fn_t fn,
const char *cacheName) {
const int32_t SLEEP_DURATION = 500; // 500 ms
if (refreshTimeInSeconds <= 0) {
if (refreshTimeInMs <= 0) {
return NULL;
}
@ -376,7 +376,7 @@ SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInSeconds, bool ext
// set free cache node callback function
pCacheObj->hashFp = taosGetDefaultHashFunction(keyType);
pCacheObj->freeFp = fn;
pCacheObj->refreshTime = refreshTimeInSeconds * 1000;
pCacheObj->refreshTime = refreshTimeInMs;
pCacheObj->checkTick = pCacheObj->refreshTime / SLEEP_DURATION;
pCacheObj->extendLifespan = extendLifespan; // the TTL after the last access
@ -914,9 +914,7 @@ void taosStopCacheRefreshWorker(void) {
taosArrayDestroy(pCacheArrayList);
}
size_t taosCacheGetNumOfObj(const SCacheObj* pCacheObj) {
return pCacheObj->numOfElems + pCacheObj->numOfElemsInTrash;
}
size_t taosCacheGetNumOfObj(const SCacheObj *pCacheObj) { return pCacheObj->numOfElems + pCacheObj->numOfElemsInTrash; }
SCacheIter *taosCacheCreateIter(const SCacheObj *pCacheObj) {
ASSERT(pCacheObj != NULL);

12
source/util/test/cacheTest.cpp
View File

@ -1,5 +1,5 @@
#include <iostream>
#include <gtest/gtest.h>
#include <iostream>
#include "os.h"
#include "taos.h"
@ -8,7 +8,7 @@
// test cache
TEST(cacheTest, client_cache_test) {
const int32_t REFRESH_TIME_IN_SEC = 2;
SCacheObj* tscMetaCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, REFRESH_TIME_IN_SEC, 0, NULL, "test");
SCacheObj* tscMetaCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, REFRESH_TIME_IN_SEC * 1000, 0, NULL, "test");
const char* key1 = "test1";
char data1[] = "test11";
@ -94,7 +94,7 @@ TEST(cacheTest, client_cache_test) {
TEST(cacheTest, cache_iter_test) {
const int32_t REFRESH_TIME_IN_SEC = 2;
auto* pCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, REFRESH_TIME_IN_SEC, false, NULL, "test");
auto* pCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, REFRESH_TIME_IN_SEC * 1000, false, NULL, "test");
char key[256] = {0};
char data[1024] = "abcdefghijk";
@ -110,7 +110,8 @@ TEST(cacheTest, cache_iter_test) {
uint64_t endTime = taosGetTimestampUs();
printf("add %d object cost:%" PRIu64 " us, avg:%f us\n", num, endTime - startTime, (endTime-startTime)/(double)num);
printf("add %d object cost:%" PRIu64 " us, avg:%f us\n", num, endTime - startTime,
(endTime - startTime) / (double)num);
startTime = taosGetTimestampUs();
for (int32_t i = 0; i < num; ++i) {
@ -119,7 +120,8 @@ TEST(cacheTest, cache_iter_test) {
assert(k != 0);
}
endTime = taosGetTimestampUs();
printf("retrieve %d object cost:%" PRIu64 " us,avg:%f\n", num, endTime - startTime, (endTime - startTime)/(double)num);
printf("retrieve %d object cost:%" PRIu64 " us,avg:%f\n", num, endTime - startTime,
(endTime - startTime) / (double)num);
int32_t count = 0;
SCacheIter* pIter = taosCacheCreateIter(pCache);