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

Merge pull request #12031 from taosdata/feature/dnode 

test: add unitest for sdb
This commit is contained in:
Shengliang Guan 2022-04-29 19:43:12 +08:00 committed by GitHub
parent 89495a3247 aac0fee928
commit d65a67455e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 768 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/dnode/mnode/sdb/sdb.h
View File

@ -107,10 +107,9 @@ typedef enum {
typedef enum { typedef enum {
SDB_STATUS_INIT = 0, SDB_STATUS_INIT = 0,
SDB_STATUS_CREATING = 1, SDB_STATUS_CREATING = 1,
SDB_STATUS_UPDATING = 2, SDB_STATUS_DROPPING = 2,
SDB_STATUS_DROPPING = 3, SDB_STATUS_DROPPED = 3,
SDB_STATUS_READY = 4, SDB_STATUS_READY = 4,
SDB_STATUS_DROPPED = 5
} ESdbStatus; } ESdbStatus;
typedef enum { typedef enum {

1
include/util/taoserror.h
View File

@ -87,6 +87,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0113) #define TSDB_CODE_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0113)
#define TSDB_CODE_CFG_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x0114) #define TSDB_CODE_CFG_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x0114)
#define TSDB_CODE_REPEAT_INIT TAOS_DEF_ERROR_CODE(0, 0x0115) #define TSDB_CODE_REPEAT_INIT TAOS_DEF_ERROR_CODE(0, 0x0115)
#define TSDB_CODE_DUP_KEY TAOS_DEF_ERROR_CODE(0, 0x0116)
#define TSDB_CODE_REF_NO_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0140) #define TSDB_CODE_REF_NO_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0140)
#define TSDB_CODE_REF_FULL TAOS_DEF_ERROR_CODE(0, 0x0141) #define TSDB_CODE_REF_FULL TAOS_DEF_ERROR_CODE(0, 0x0141)

58
source/dnode/mgmt/implement/src/dmTransport.c
View File

@ -72,6 +72,7 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
NodeMsgFp msgFp = NULL; NodeMsgFp msgFp = NULL;
uint16_t msgType = pRpc->msgType; uint16_t msgType = pRpc->msgType;
bool needRelease = false; bool needRelease = false;
bool isReq = msgType & 1U;
if (pEpSet && pEpSet->numOfEps > 0 && msgType == TDMT_MND_STATUS_RSP) { if (pEpSet && pEpSet->numOfEps > 0 && msgType == TDMT_MND_STATUS_RSP) {
dmSetMnodeEpSet(pWrapper->pDnode, pEpSet); dmSetMnodeEpSet(pWrapper->pDnode, pEpSet);
@ -85,13 +86,13 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
if (dmBuildMsg(pMsg, pRpc) != 0) goto _OVER; if (dmBuildMsg(pMsg, pRpc) != 0) goto _OVER;
if (pWrapper->procType == DND_PROC_SINGLE) { if (pWrapper->procType == DND_PROC_SINGLE) {
dTrace("msg:%p, is created, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, pMsg->user); dTrace("msg:%p, created, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, pMsg->user);
code = (*msgFp)(pWrapper, pMsg); code = (*msgFp)(pWrapper, pMsg);
} else if (pWrapper->procType == DND_PROC_PARENT) { } else if (pWrapper->procType == DND_PROC_PARENT) {
dTrace("msg:%p, is created and put into child queue, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), dTrace("msg:%p, created and put into child queue, type:%s handle:%p code:0x%04x user:%s contLen:%d", pMsg,
pRpc->handle, pMsg->user); TMSG_INFO(msgType), pRpc->handle, pMsg->rpcMsg.code & 0XFFFF, pMsg->user, pRpc->contLen);
code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, pRpc->handle, code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen,
pRpc->refId, PROC_FUNC_REQ); (isReq && (pMsg->rpcMsg.code == 0)) ? pRpc->handle : NULL, pRpc->refId, PROC_FUNC_REQ);
} else { } else {
dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user); dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
ASSERT(1); ASSERT(1);
@ -100,12 +101,13 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
_OVER: _OVER:
if (code == 0) { if (code == 0) {
if (pWrapper->procType == DND_PROC_PARENT) { if (pWrapper->procType == DND_PROC_PARENT) {
dTrace("msg:%p, is freed in parent process", pMsg); dTrace("msg:%p, freed in parent process", pMsg);
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
rpcFreeCont(pRpc->pCont); rpcFreeCont(pRpc->pCont);
} }
} else { } else {
dError("msg:%p, type:%s failed to process since 0x%04x:%s", pMsg, TMSG_INFO(msgType), code & 0XFFFF, terrstr()); dError("msg:%p, type:%s handle:%p failed to process since 0x%04x:%s", pMsg, TMSG_INFO(msgType), pRpc->handle,
code & 0XFFFF, terrstr());
if (msgType & 1U) { if (msgType & 1U) {
if (terrno != 0) code = terrno; if (terrno != 0) code = terrno;
if (code == TSDB_CODE_NODE_NOT_DEPLOYED || code == TSDB_CODE_NODE_OFFLINE) { if (code == TSDB_CODE_NODE_NOT_DEPLOYED || code == TSDB_CODE_NODE_OFFLINE) {
@ -359,29 +361,31 @@ static void dmConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t
static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen, static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen,
EProcFuncType ftype) { EProcFuncType ftype) {
int32_t code = pMsg->code & 0xFFFF;
pMsg->pCont = pCont; pMsg->pCont = pCont;
dTrace("msg:%p, get from parent queue, ftype:%d handle:%p code:0x%04x mtype:%d, app:%p", pMsg, ftype, pMsg->handle,
pMsg->code & 0xFFFF, pMsg->msgType, pMsg->ahandle);
switch (ftype) { if (ftype == PROC_FUNC_REQ) {
case PROC_FUNC_REGIST: dTrace("msg:%p, get from parent queue, send req:%s handle:%p code:0x%04x, app:%p", pMsg, TMSG_INFO(pMsg->msgType),
rpcRegisterBrokenLinkArg(pMsg); pMsg->handle, code, pMsg->ahandle);
break; dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
case PROC_FUNC_RELEASE: } else if (ftype == PROC_FUNC_RSP) {
taosProcRemoveHandle(pWrapper->procObj, pMsg->handle); dTrace("msg:%p, get from parent queue, rsp handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code, pMsg->ahandle);
rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code); pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
rpcFreeCont(pCont); dmSendRpcRsp(pWrapper->pDnode, pMsg);
break; } else if (ftype == PROC_FUNC_REGIST) {
case PROC_FUNC_REQ: dTrace("msg:%p, get from parent queue, regist handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code,
dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg); pMsg->ahandle);
break; rpcRegisterBrokenLinkArg(pMsg);
case PROC_FUNC_RSP: } else if (ftype == PROC_FUNC_RELEASE) {
pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle); dTrace("msg:%p, get from parent queue, release handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code,
dmSendRpcRsp(pWrapper->pDnode, pMsg); pMsg->ahandle);
break; taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
default: rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code);
break; rpcFreeCont(pCont);
} else {
dError("msg:%p, invalid ftype:%d while get from parent queue, handle:%p", pMsg, ftype, pMsg->handle);
} }
taosMemoryFree(pMsg); taosMemoryFree(pMsg);
} }

26
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -136,7 +136,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
// sync integration response // sync integration response
for (int i = 0; i < taosArrayGetSize(pArray); i++) { for (int i = 0; i < taosArrayGetSize(pArray); i++) {
SNodeMsg *pMsg; SNodeMsg *pMsg;
SRpcMsg * pRpc; SRpcMsg *pRpc;
pMsg = *(SNodeMsg **)taosArrayGet(pArray, i); pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
pRpc = &pMsg->rpcMsg; pRpc = &pMsg->rpcMsg;
@ -175,7 +175,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SVnodeObj *pVnode = pInfo->ahandle; SVnodeObj *pVnode = pInfo->ahandle;
SNodeMsg * pMsg = NULL; SNodeMsg *pMsg = NULL;
SRpcMsg rsp; SRpcMsg rsp;
for (int32_t i = 0; i < numOfMsgs; ++i) { for (int32_t i = 0; i < numOfMsgs; ++i) {
@ -218,7 +218,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SVnodeObj *pVnode = pInfo->ahandle; SVnodeObj *pVnode = pInfo->ahandle;
SNodeMsg * pMsg = NULL; SNodeMsg *pMsg = NULL;
for (int32_t i = 0; i < numOfMsgs; ++i) { for (int32_t i = 0; i < numOfMsgs; ++i) {
taosGetQitem(qall, (void **)&pMsg); taosGetQitem(qall, (void **)&pMsg);
@ -231,7 +231,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SVnodeObj *pVnode = pInfo->ahandle; SVnodeObj *pVnode = pInfo->ahandle;
SNodeMsg * pMsg = NULL; SNodeMsg *pMsg = NULL;
for (int32_t i = 0; i < numOfMsgs; ++i) { for (int32_t i = 0; i < numOfMsgs; ++i) {
taosGetQitem(qall, (void **)&pMsg); taosGetQitem(qall, (void **)&pMsg);
@ -248,7 +248,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
} }
static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueType qtype) { static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueType qtype) {
SRpcMsg * pRpc = &pMsg->rpcMsg; SRpcMsg *pRpc = &pMsg->rpcMsg;
SMsgHead *pHead = pRpc->pCont; SMsgHead *pHead = pRpc->pCont;
pHead->contLen = ntohl(pHead->contLen); pHead->contLen = ntohl(pHead->contLen);
pHead->vgId = ntohl(pHead->vgId); pHead->vgId = ntohl(pHead->vgId);
@ -262,23 +262,23 @@ static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueTyp
int32_t code = 0; int32_t code = 0;
switch (qtype) { switch (qtype) {
case QUERY_QUEUE: case QUERY_QUEUE:
dTrace("msg:%p, will be written into vnode-query queue", pMsg); dTrace("msg:%p, type:%s will be written into vnode-query queue", pMsg, TMSG_INFO(pRpc->msgType));
taosWriteQitem(pVnode->pQueryQ, pMsg); taosWriteQitem(pVnode->pQueryQ, pMsg);
break; break;
case FETCH_QUEUE: case FETCH_QUEUE:
dTrace("msg:%p, will be written into vnode-fetch queue", pMsg); dTrace("msg:%p, type:%s will be written into vnode-fetch queue", pMsg, TMSG_INFO(pRpc->msgType));
taosWriteQitem(pVnode->pFetchQ, pMsg); taosWriteQitem(pVnode->pFetchQ, pMsg);
break; break;
case WRITE_QUEUE: case WRITE_QUEUE:
dTrace("msg:%p, will be written into vnode-write queue", pMsg); dTrace("msg:%p, type:%s will be written into vnode-write queue", pMsg, TMSG_INFO(pRpc->msgType));
taosWriteQitem(pVnode->pWriteQ, pMsg); taosWriteQitem(pVnode->pWriteQ, pMsg);
break; break;
case SYNC_QUEUE: case SYNC_QUEUE:
dTrace("msg:%p, will be written into vnode-sync queue", pMsg); dTrace("msg:%p, type:%s will be written into vnode-sync queue", pMsg, TMSG_INFO(pRpc->msgType));
taosWriteQitem(pVnode->pSyncQ, pMsg); taosWriteQitem(pVnode->pSyncQ, pMsg);
break; break;
case MERGE_QUEUE: case MERGE_QUEUE:
dTrace("msg:%p, will be written into vnode-merge queue", pMsg); dTrace("msg:%p, type:%s will be written into vnode-merge queue", pMsg, TMSG_INFO(pRpc->msgType));
taosWriteQitem(pVnode->pMergeQ, pMsg); taosWriteQitem(pVnode->pMergeQ, pMsg);
break; break;
default: default:
@ -317,7 +317,7 @@ int32_t vmProcessMergeMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SVnodesMgmt * pMgmt = pWrapper->pMgmt; SVnodesMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->mgmtWorker; SSingleWorker *pWorker = &pMgmt->mgmtWorker;
dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name); dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg); taosWriteQitem(pWorker->queue, pMsg);
@ -325,7 +325,7 @@ int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SVnodesMgmt * pMgmt = pWrapper->pMgmt; SVnodesMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker; SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@ -335,7 +335,7 @@ int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) { static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt; SVnodesMgmt *pMgmt = pWrapper->pMgmt;
SMsgHead * pHead = pRpc->pCont; SMsgHead *pHead = pRpc->pCont;
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL) return -1; if (pVnode == NULL) return -1;

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

@ -625,7 +625,7 @@ static int32_t mndRetrieveMnodes(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *p
colDataAppend(pColInfo, numOfRows, b1, false); colDataAppend(pColInfo, numOfRows, b1, false);
const char *roles = syncStr(pObj->role); const char *roles = syncStr(pObj->role);
char *b2 = taosMemoryCalloc(1, strlen(roles) + VARSTR_HEADER_SIZE); char *b2 = taosMemoryCalloc(1, 12 + VARSTR_HEADER_SIZE);
STR_WITH_MAXSIZE_TO_VARSTR(b2, roles, pShow->pMeta->pSchemas[cols].bytes); STR_WITH_MAXSIZE_TO_VARSTR(b2, roles, pShow->pMeta->pSchemas[cols].bytes);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

6
source/dnode/mnode/impl/src/mndStb.c
View File

@ -679,8 +679,8 @@ static int32_t mndCreateStb(SMnode *pMnode, SNodeMsg *pReq, SMCreateStbReq *pCre
memcpy(stbObj.pAst2, pCreate->pAst2, stbObj.ast2Len); memcpy(stbObj.pAst2, pCreate->pAst2, stbObj.ast2Len);
} }
stbObj.pColumns = taosMemoryMalloc(stbObj.numOfColumns * sizeof(SSchema)); stbObj.pColumns = taosMemoryCalloc(1, stbObj.numOfColumns * sizeof(SSchema));
stbObj.pTags = taosMemoryMalloc(stbObj.numOfTags * sizeof(SSchema)); stbObj.pTags = taosMemoryCalloc(1, stbObj.numOfTags * sizeof(SSchema));
if (stbObj.pColumns == NULL || stbObj.pTags == NULL) { if (stbObj.pColumns == NULL || stbObj.pTags == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
@ -1111,7 +1111,7 @@ static int32_t mndSetAlterStbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pD
SSdbRaw *pRedoRaw = mndStbActionEncode(pStb); SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
if (pRedoRaw == NULL) return -1; if (pRedoRaw == NULL) return -1;
if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1; if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_UPDATING) != 0) return -1; if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) return -1;
return 0; return 0;
} }

722
source/dnode/mnode/impl/test/sdb/sdbTest.cpp
View File

@ -15,8 +15,31 @@
class MndTestSdb : public ::testing::Test { class MndTestSdb : public ::testing::Test {
protected: protected:
static void SetUpTestSuite() {} static void SetUpTestSuite() {
static void TearDownTestSuite() {} dDebugFlag = 143;
vDebugFlag = 0;
mDebugFlag = 143;
cDebugFlag = 0;
jniDebugFlag = 0;
tmrDebugFlag = 135;
uDebugFlag = 135;
rpcDebugFlag = 143;
qDebugFlag = 0;
wDebugFlag = 0;
sDebugFlag = 0;
tsdbDebugFlag = 0;
tsLogEmbedded = 1;
tsAsyncLog = 0;
const char *path = "/tmp/td";
taosRemoveDir(path);
taosMkDir(path);
tstrncpy(tsLogDir, path, PATH_MAX);
if (taosInitLog("taosdlog", 1) != 0) {
printf("failed to init log file\n");
}
}
static void TearDownTestSuite() { taosCloseLog(); }
public: public:
void SetUp() override {} void SetUp() override {}
@ -26,6 +49,8 @@ class MndTestSdb : public ::testing::Test {
typedef struct SMnode { typedef struct SMnode {
int32_t v100; int32_t v100;
int32_t v200; int32_t v200;
int32_t insertTimes;
int32_t deleteTimes;
SSdb *pSdb; SSdb *pSdb;
} SMnode; } SMnode;
@ -80,6 +105,48 @@ SSdbRaw *strEncode(SStrObj *pObj) {
return pRaw; return pRaw;
} }
SSdbRaw *i32Encode(SI32Obj *pObj) {
int32_t dataPos = 0;
SSdbRaw *pRaw = sdbAllocRaw(SDB_VGROUP, 2, sizeof(SI32Obj));
sdbSetRawInt32(pRaw, dataPos, pObj->key);
dataPos += sizeof(pObj->key);
sdbSetRawInt8(pRaw, dataPos, pObj->v8);
dataPos += sizeof(pObj->v8);
sdbSetRawInt16(pRaw, dataPos, pObj->v16);
dataPos += sizeof(pObj->v16);
sdbSetRawInt32(pRaw, dataPos, pObj->v32);
dataPos += sizeof(pObj->v32);
sdbSetRawInt64(pRaw, dataPos, pObj->v64);
dataPos += sizeof(pObj->v64);
sdbSetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr));
dataPos += sizeof(pObj->vstr);
sdbSetRawDataLen(pRaw, dataPos);
return pRaw;
}
SSdbRaw *i64Encode(SI64Obj *pObj) {
int32_t dataPos = 0;
SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, 3, sizeof(SI64Obj));
sdbSetRawInt64(pRaw, dataPos, pObj->key);
dataPos += sizeof(pObj->key);
sdbSetRawInt8(pRaw, dataPos, pObj->v8);
dataPos += sizeof(pObj->v8);
sdbSetRawInt16(pRaw, dataPos, pObj->v16);
dataPos += sizeof(pObj->v16);
sdbSetRawInt32(pRaw, dataPos, pObj->v32);
dataPos += sizeof(pObj->v32);
sdbSetRawInt64(pRaw, dataPos, pObj->v64);
dataPos += sizeof(pObj->v64);
sdbSetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr));
dataPos += sizeof(pObj->vstr);
sdbSetRawDataLen(pRaw, dataPos);
return pRaw;
}
SSdbRow *strDecode(SSdbRaw *pRaw) { SSdbRow *strDecode(SSdbRaw *pRaw) {
int8_t sver = 0; int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL; if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
@ -108,9 +175,103 @@ SSdbRow *strDecode(SSdbRaw *pRaw) {
return pRow; return pRow;
} }
int32_t strInsert(SSdb *pSdb, SStrObj *pObj) { return 0; } SSdbRow *i32Decode(SSdbRaw *pRaw) {
int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
if (sver != 2) return NULL;
int32_t strDelete(SSdb *pSdb, SStrObj *pObj, bool callFunc) { return 0; } SSdbRow *pRow = sdbAllocRow(sizeof(SI32Obj));
if (pRow == NULL) return NULL;
SI32Obj *pObj = (SI32Obj *)sdbGetRowObj(pRow);
if (pObj == NULL) return NULL;
int32_t dataPos = 0;
sdbGetRawInt32(pRaw, dataPos, &pObj->key);
dataPos += sizeof(pObj->key);
sdbGetRawInt8(pRaw, dataPos, &pObj->v8);
dataPos += sizeof(pObj->v8);
sdbGetRawInt16(pRaw, dataPos, &pObj->v16);
dataPos += sizeof(pObj->v16);
sdbGetRawInt32(pRaw, dataPos, &pObj->v32);
dataPos += sizeof(pObj->v32);
sdbGetRawInt64(pRaw, dataPos, &pObj->v64);
dataPos += sizeof(pObj->v64);
sdbGetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr));
dataPos += sizeof(pObj->vstr);
return pRow;
}
SSdbRow *i64Decode(SSdbRaw *pRaw) {
int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
if (sver != 3) return NULL;
SSdbRow *pRow = sdbAllocRow(sizeof(SI64Obj));
if (pRow == NULL) return NULL;
SI64Obj *pObj = (SI64Obj *)sdbGetRowObj(pRow);
if (pObj == NULL) return NULL;
int32_t dataPos = 0;
sdbGetRawInt64(pRaw, dataPos, &pObj->key);
dataPos += sizeof(pObj->key);
sdbGetRawInt8(pRaw, dataPos, &pObj->v8);
dataPos += sizeof(pObj->v8);
sdbGetRawInt16(pRaw, dataPos, &pObj->v16);
dataPos += sizeof(pObj->v16);
sdbGetRawInt32(pRaw, dataPos, &pObj->v32);
dataPos += sizeof(pObj->v32);
sdbGetRawInt64(pRaw, dataPos, &pObj->v64);
dataPos += sizeof(pObj->v64);
sdbGetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr));
dataPos += sizeof(pObj->vstr);
return pRow;
}
int32_t strInsert(SSdb *pSdb, SStrObj *pObj) {
SMnode *pMnode = pSdb->pMnode;
pMnode->insertTimes++;
return 0;
}
int32_t i32Insert(SSdb *pSdb, SI32Obj *pObj) {
SMnode *pMnode = pSdb->pMnode;
pMnode->insertTimes++;
return 0;
}
int32_t i64Insert(SSdb *pSdb, SI64Obj *pObj) {
SMnode *pMnode = pSdb->pMnode;
pMnode->insertTimes++;
return 0;
}
int32_t strDelete(SSdb *pSdb, SStrObj *pObj, bool callFunc) {
if (callFunc) {
SMnode *pMnode = pSdb->pMnode;
pMnode->deleteTimes++;
}
return 0;
}
int32_t i32Delete(SSdb *pSdb, SI32Obj *pObj, bool callFunc) {
if (callFunc) {
SMnode *pMnode = pSdb->pMnode;
pMnode->deleteTimes++;
}
return 0;
}
int32_t i64Delete(SSdb *pSdb, SI64Obj *pObj, bool callFunc) {
if (callFunc) {
SMnode *pMnode = pSdb->pMnode;
pMnode->deleteTimes++;
}
return 0;
}
int32_t strUpdate(SSdb *pSdb, SStrObj *pOld, SStrObj *pNew) { int32_t strUpdate(SSdb *pSdb, SStrObj *pOld, SStrObj *pNew) {
pOld->v8 = pNew->v8; pOld->v8 = pNew->v8;
@ -121,6 +282,24 @@ int32_t strUpdate(SSdb *pSdb, SStrObj *pOld, SStrObj *pNew) {
return 0; return 0;
} }
int32_t i32Update(SSdb *pSdb, SI32Obj *pOld, SI32Obj *pNew) {
pOld->v8 = pNew->v8;
pOld->v16 = pNew->v16;
pOld->v32 = pNew->v32;
pOld->v64 = pNew->v64;
strcpy(pOld->vstr, pNew->vstr);
return 0;
}
int32_t i64Update(SSdb *pSdb, SI64Obj *pOld, SI64Obj *pNew) {
pOld->v8 = pNew->v8;
pOld->v16 = pNew->v16;
pOld->v32 = pNew->v32;
pOld->v64 = pNew->v64;
strcpy(pOld->vstr, pNew->vstr);
return 0;
}
void strSetDefault(SStrObj *pObj, int32_t index) { void strSetDefault(SStrObj *pObj, int32_t index) {
memset(pObj, 0, sizeof(SStrObj)); memset(pObj, 0, sizeof(SStrObj));
snprintf(pObj->key, sizeof(pObj->key), "k%d", index * 1000); snprintf(pObj->key, sizeof(pObj->key), "k%d", index * 1000);
@ -131,6 +310,26 @@ void strSetDefault(SStrObj *pObj, int32_t index) {
snprintf(pObj->vstr, sizeof(pObj->vstr), "v%d", index * 1000); snprintf(pObj->vstr, sizeof(pObj->vstr), "v%d", index * 1000);
} }
void i32SetDefault(SI32Obj *pObj, int32_t index) {
memset(pObj, 0, sizeof(SI32Obj));
pObj->key = index;
pObj->v8 = index;
pObj->v16 = index;
pObj->v32 = index * 1000;
pObj->v64 = index * 1000;
snprintf(pObj->vstr, sizeof(pObj->vstr), "v%d", index * 1000);
}
void i64SetDefault(SI64Obj *pObj, int32_t index) {
memset(pObj, 0, sizeof(SI64Obj));
pObj->key = index;
pObj->v8 = index;
pObj->v16 = index;
pObj->v32 = index * 1000;
pObj->v64 = index * 1000;
snprintf(pObj->vstr, sizeof(pObj->vstr), "v%d", index * 1000);
}
int32_t strDefault(SMnode *pMnode) { int32_t strDefault(SMnode *pMnode) {
SStrObj strObj; SStrObj strObj;
SSdbRaw *pRaw = NULL; SSdbRaw *pRaw = NULL;
@ -144,6 +343,8 @@ int32_t strDefault(SMnode *pMnode) {
pRaw = strEncode(&strObj); pRaw = strEncode(&strObj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY); sdbSetRawStatus(pRaw, SDB_STATUS_READY);
if (sdbWriteWithoutFree(pMnode->pSdb, pRaw) != 0) return -1; if (sdbWriteWithoutFree(pMnode->pSdb, pRaw) != 0) return -1;
EXPECT_EQ(sdbGetRawTotalSize(pRaw), 79);
sdbFreeRaw(pRaw); sdbFreeRaw(pRaw);
return 0; return 0;
@ -151,28 +352,32 @@ int32_t strDefault(SMnode *pMnode) {
bool sdbTraverseSucc1(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) { bool sdbTraverseSucc1(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) {
if (pObj->v8 == 1) { if (pObj->v8 == 1) {
*p1 = *p2 + *p3 + pObj->v8; *p1 += *p2 + *p3 + pObj->v8;
} }
return true; return true;
} }
bool sdbTraverseSucc2(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) { bool sdbTraverseSucc2(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) {
*p1 = *p2 + *p3 + pObj->v8; *p1 += *p2 + *p3 + pObj->v8;
return true; return true;
} }
bool sdbTraverseFail(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) { bool sdbTraverseFail(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) {
*p1 = *p2 + *p3; *p1 += *p2 + *p3;
return false; return false;
} }
TEST_F(MndTestSdb, 01_Write) { TEST_F(MndTestSdb, 01_Write_Str) {
void *pIter; void *pIter = NULL;
int32_t num; int32_t num = 0;
SStrObj *pObj; SStrObj *pObj = NULL;
SMnode mnode; SMnode mnode = {0};
SSdb *pSdb; SSdb *pSdb = NULL;
SSdbOpt opt = {0}; SSdbOpt opt = {0};
SStrObj strObj = {0};
SI32Obj i32Obj = {0};
SI64Obj i64Obj = {0};
SSdbRaw *pRaw = NULL;
int32_t p1 = 0; int32_t p1 = 0;
int32_t p2 = 111; int32_t p2 = 111;
int32_t p3 = 222; int32_t p3 = 222;
@ -183,42 +388,64 @@ TEST_F(MndTestSdb, 01_Write) {
opt.path = "/tmp/mnode_test_sdb"; opt.path = "/tmp/mnode_test_sdb";
taosRemoveDir(opt.path); taosRemoveDir(opt.path);
SSdbTable strTable = { SSdbTable strTable1;
SDB_USER, memset(&strTable1, 0, sizeof(SSdbTable));
SDB_KEY_BINARY, strTable1.sdbType = SDB_USER;
(SdbDeployFp)strDefault, strTable1.keyType = SDB_KEY_BINARY;
(SdbEncodeFp)strEncode, strTable1.deployFp = (SdbDeployFp)strDefault;
(SdbDecodeFp)strDecode, strTable1.encodeFp = (SdbEncodeFp)strEncode;
(SdbInsertFp)strInsert, strTable1.decodeFp = (SdbDecodeFp)strDecode;
(SdbUpdateFp)strUpdate, strTable1.insertFp = (SdbInsertFp)strInsert;
(SdbDeleteFp)strDelete, strTable1.updateFp = (SdbUpdateFp)strUpdate;
}; strTable1.deleteFp = (SdbDeleteFp)strDelete;
SSdbTable strTable2;
memset(&strTable2, 0, sizeof(SSdbTable));
strTable2.sdbType = SDB_VGROUP;
strTable2.keyType = SDB_KEY_INT32;
strTable2.encodeFp = (SdbEncodeFp)i32Encode;
strTable2.decodeFp = (SdbDecodeFp)i32Decode;
strTable2.insertFp = (SdbInsertFp)i32Insert;
strTable2.updateFp = (SdbUpdateFp)i32Update;
strTable2.deleteFp = (SdbDeleteFp)i32Delete;
SSdbTable strTable3;
memset(&strTable3, 0, sizeof(SSdbTable));
strTable3.sdbType = SDB_CONSUMER;
strTable3.keyType = SDB_KEY_INT64;
strTable3.encodeFp = (SdbEncodeFp)i64Encode;
strTable3.decodeFp = (SdbDecodeFp)i64Decode;
strTable3.insertFp = (SdbInsertFp)i64Insert;
strTable3.updateFp = (SdbUpdateFp)i64Update;
strTable3.deleteFp = (SdbDeleteFp)i64Delete;
pSdb = sdbInit(&opt); pSdb = sdbInit(&opt);
mnode.pSdb = pSdb; mnode.pSdb = pSdb;
ASSERT_NE(pSdb, nullptr); ASSERT_NE(pSdb, nullptr);
ASSERT_EQ(sdbSetTable(pSdb, strTable), 0); ASSERT_EQ(sdbSetTable(pSdb, strTable1), 0);
ASSERT_EQ(sdbSetTable(pSdb, strTable2), 0);
ASSERT_EQ(sdbSetTable(pSdb, strTable3), 0);
ASSERT_EQ(sdbDeploy(pSdb), 0); ASSERT_EQ(sdbDeploy(pSdb), 0);
#if 0
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k1000"); pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k1000");
ASSERT_NE(pObj, nullptr); ASSERT_NE(pObj, nullptr);
EXPECT_STREQ(pObj->key, "k1000"); EXPECT_STREQ(pObj->key, "k1000");
EXPECT_STREQ(pObj->vstr, "v1000"); EXPECT_STREQ(pObj->vstr, "v1000");
EXPECT_EQ(pObj->v8, 1); ASSERT_EQ(pObj->v8, 1);
EXPECT_EQ(pObj->v16, 1); ASSERT_EQ(pObj->v16, 1);
EXPECT_EQ(pObj->v32, 1000); ASSERT_EQ(pObj->v32, 1000);
EXPECT_EQ(pObj->v64, 1000); ASSERT_EQ(pObj->v64, 1000);
sdbRelease(pSdb, pObj); sdbRelease(pSdb, pObj);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k2000"); pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k2000");
ASSERT_NE(pObj, nullptr); ASSERT_NE(pObj, nullptr);
EXPECT_STREQ(pObj->key, "k2000"); EXPECT_STREQ(pObj->key, "k2000");
EXPECT_STREQ(pObj->vstr, "v2000"); EXPECT_STREQ(pObj->vstr, "v2000");
EXPECT_EQ(pObj->v8, 2); ASSERT_EQ(pObj->v8, 2);
EXPECT_EQ(pObj->v16, 2); ASSERT_EQ(pObj->v16, 2);
EXPECT_EQ(pObj->v32, 2000); ASSERT_EQ(pObj->v32, 2000);
EXPECT_EQ(pObj->v64, 2000); ASSERT_EQ(pObj->v64, 2000);
sdbRelease(pSdb, pObj); sdbRelease(pSdb, pObj);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k200"); pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k200");
@ -233,7 +460,7 @@ TEST_F(MndTestSdb, 01_Write) {
num++; num++;
sdbRelease(pSdb, pObj); sdbRelease(pSdb, pObj);
} while (1); } while (1);
EXPECT_EQ(num, 2); ASSERT_EQ(num, 2);
do { do {
pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj); pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj);
@ -248,51 +475,254 @@ TEST_F(MndTestSdb, 01_Write) {
p1 = 0; p1 = 0;
p2 = 111; p2 = 111;
p3 = 222; p3 = 222;
sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3); sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc1, &p1, &p2, &p3);
EXPECT_EQ(p1, 334); ASSERT_EQ(p1, 334);
p1 = 0; p1 = 0;
p2 = 111; p2 = 111;
p3 = 222; p3 = 222;
sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3); sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3);
EXPECT_EQ(p1, 669); ASSERT_EQ(p1, 669);
p1 = 0; p1 = 0;
p2 = 111; p2 = 111;
p3 = 222; p3 = 222;
sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseFail, &p1, &p2, &p3); sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseFail, &p1, &p2, &p3);
EXPECT_EQ(p1, 333); ASSERT_EQ(p1, 333);
EXPECT_EQ(sdbGetSize(pSdb, SDB_USER), 2); ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 2);
EXPECT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1); ASSERT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1);
EXPECT_EQ(sdbGetTableVer(pSdb, SDB_USER), 2); ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 2 );
EXPECT_EQ(sdbUpdateVer(pSdb, 0), 2); ASSERT_EQ(sdbUpdateVer(pSdb, 0), -1);
EXPECT_EQ(sdbUpdateVer(pSdb, 1), 3); ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0);
EXPECT_EQ(sdbUpdateVer(pSdb, -1), 2); ASSERT_EQ(sdbUpdateVer(pSdb, -1), -1);
ASSERT_EQ(mnode.insertTimes, 2);
ASSERT_EQ(mnode.deleteTimes, 0);
// insert, call func {
// insert, call func
strSetDefault(&strObj, 3);
pRaw = strEncode(&strObj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k3000");
ASSERT_NE(pObj, nullptr);
EXPECT_STREQ(pObj->key, "k3000");
EXPECT_STREQ(pObj->vstr, "v3000");
ASSERT_EQ(pObj->v8, 3);
ASSERT_EQ(pObj->v16, 3);
ASSERT_EQ(pObj->v32, 3000);
ASSERT_EQ(pObj->v64, 3000);
sdbRelease(pSdb, pObj);
// update, call func ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 3);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 3);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1);
// delete, call func 2 // update, call func
strSetDefault(&strObj, 3);
strObj.v8 = 4;
pRaw = strEncode(&strObj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k3000");
ASSERT_NE(pObj, nullptr);
EXPECT_STREQ(pObj->key, "k3000");
EXPECT_STREQ(pObj->vstr, "v3000");
ASSERT_EQ(pObj->v8, 4);
ASSERT_EQ(pObj->v16, 3);
ASSERT_EQ(pObj->v32, 3000);
ASSERT_EQ(pObj->v64, 3000);
sdbRelease(pSdb, pObj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 3);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 4);
ASSERT_EQ(sdbUpdateVer(pSdb, 0), -1);
ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0);
ASSERT_EQ(sdbUpdateVer(pSdb, -1), -1);
ASSERT_EQ(mnode.insertTimes, 3);
ASSERT_EQ(mnode.deleteTimes, 0);
// delete, call func 2
strSetDefault(&strObj, 3);
strObj.v16 = 4;
pRaw = strEncode(&strObj);
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k3000");
ASSERT_EQ(pObj, nullptr);
ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 2);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 5);
ASSERT_EQ(mnode.insertTimes, 3);
ASSERT_EQ(mnode.deleteTimes, 1);
}
{
int32_t key = 4;
i32SetDefault(&i32Obj, key);
pRaw = i32Encode(&i32Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
SI32Obj *pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key);
ASSERT_NE(pI32Obj, nullptr);
ASSERT_EQ(pI32Obj->key, key);
ASSERT_EQ(pI32Obj->v8, 4);
ASSERT_EQ(pI32Obj->v16, 4);
ASSERT_EQ(pI32Obj->v32, 4000);
ASSERT_EQ(pI32Obj->v64, 4000);
sdbRelease(pSdb, pI32Obj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 1);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_VGROUP), 5);
i32SetDefault(&i32Obj, key);
i32Obj.v8 = 5;
pRaw = i32Encode(&i32Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key);
ASSERT_NE(pI32Obj, nullptr);
ASSERT_EQ(pI32Obj->key, key);
ASSERT_EQ(pI32Obj->v8, 5);
ASSERT_EQ(pI32Obj->v16, 4);
ASSERT_EQ(pI32Obj->v32, 4000);
ASSERT_EQ(pI32Obj->v64, 4000);
sdbRelease(pSdb, pI32Obj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 2);
ASSERT_EQ(mnode.insertTimes, 4);
ASSERT_EQ(mnode.deleteTimes, 1);
// delete, call func 2
key = 4;
i32SetDefault(&i32Obj, key);
pRaw = i32Encode(&i32Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key);
ASSERT_EQ(pI32Obj, nullptr);
ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 0);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 3);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_VGROUP), 5);
ASSERT_EQ(mnode.insertTimes, 4);
ASSERT_EQ(mnode.deleteTimes, 2);
key = 6;
i32SetDefault(&i32Obj, key);
pRaw = i32Encode(&i32Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key);
ASSERT_NE(pI32Obj, nullptr);
ASSERT_EQ(pI32Obj->key, key);
ASSERT_EQ(pI32Obj->v8, 6);
ASSERT_EQ(pI32Obj->v16, 6);
ASSERT_EQ(pI32Obj->v32, 6000);
ASSERT_EQ(pI32Obj->v64, 6000);
sdbRelease(pSdb, pI32Obj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 4);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_VGROUP), 7);
ASSERT_EQ(mnode.insertTimes, 5);
ASSERT_EQ(mnode.deleteTimes, 2);
}
{
int64_t key = 4;
i64SetDefault(&i64Obj, key);
pRaw = i64Encode(&i64Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
SI64Obj *pI64Obj = (SI64Obj *)sdbAcquire(pSdb, SDB_CONSUMER, &key);
ASSERT_NE(pI64Obj, nullptr);
ASSERT_EQ(pI64Obj->key, key);
ASSERT_EQ(pI64Obj->v8, 4);
ASSERT_EQ(pI64Obj->v16, 4);
ASSERT_EQ(pI64Obj->v32, 4000);
ASSERT_EQ(pI64Obj->v64, 4000);
sdbRelease(pSdb, pI64Obj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 1);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_CONSUMER), -1);
i64SetDefault(&i64Obj, key);
i64Obj.v8 = 5;
pRaw = i64Encode(&i64Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pI64Obj = (SI64Obj *)sdbAcquire(pSdb, SDB_CONSUMER, &key);
ASSERT_NE(pI64Obj, nullptr);
ASSERT_EQ(pI64Obj->key, key);
ASSERT_EQ(pI64Obj->v8, 5);
ASSERT_EQ(pI64Obj->v16, 4);
ASSERT_EQ(pI64Obj->v32, 4000);
ASSERT_EQ(pI64Obj->v64, 4000);
sdbRelease(pSdb, pI64Obj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 2);
ASSERT_EQ(mnode.insertTimes, 6);
ASSERT_EQ(mnode.deleteTimes, 2);
// delete, call func 2
key = 4;
i64SetDefault(&i64Obj, key);
pRaw = i64Encode(&i64Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_CONSUMER, &key);
ASSERT_EQ(pObj, nullptr);
ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 0);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 3);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_CONSUMER), -1);
ASSERT_EQ(mnode.insertTimes, 6);
ASSERT_EQ(mnode.deleteTimes, 3);
key = 7;
i64SetDefault(&i64Obj, key);
pRaw = i64Encode(&i64Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pI64Obj = (SI64Obj *)sdbAcquire(pSdb, SDB_CONSUMER, &key);
ASSERT_NE(pI64Obj, nullptr);
ASSERT_EQ(pI64Obj->key, key);
ASSERT_EQ(pI64Obj->v8, 7);
ASSERT_EQ(pI64Obj->v16, 7);
ASSERT_EQ(pI64Obj->v32, 7000);
ASSERT_EQ(pI64Obj->v64, 7000);
sdbRelease(pSdb, pI64Obj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 4);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_CONSUMER), -1);
ASSERT_EQ(mnode.insertTimes, 7);
ASSERT_EQ(mnode.deleteTimes, 3);
}
// write version // write version
ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0);
// sdb Write ver ASSERT_EQ(sdbUpdateVer(pSdb, 1), 1);
// sdbRead
#endif
ASSERT_EQ(sdbWriteFile(pSdb), 0); ASSERT_EQ(sdbWriteFile(pSdb), 0);
ASSERT_EQ(sdbWriteFile(pSdb), 0);
sdbCleanup(pSdb); sdbCleanup(pSdb);
ASSERT_EQ(mnode.insertTimes, 7);
ASSERT_EQ(mnode.deleteTimes, 7);
} }
TEST_F(MndTestSdb, 01_Read) { TEST_F(MndTestSdb, 01_Read_Str) {
void *pIter; void *pIter = NULL;
int32_t num; int32_t num = 0;
SStrObj *pObj; SStrObj *pObj = NULL;
SMnode mnode; SMnode mnode = {0};
SSdb *pSdb; SSdb *pSdb = NULL;
SSdbOpt opt = {0}; SSdbOpt opt = {0};
SStrObj strObj = {0};
SSdbRaw *pRaw = NULL;
int32_t p1 = 0; int32_t p1 = 0;
int32_t p2 = 111; int32_t p2 = 111;
int32_t p3 = 222; int32_t p3 = 222;
@ -301,22 +731,174 @@ TEST_F(MndTestSdb, 01_Read) {
mnode.v200 = 200; mnode.v200 = 200;
opt.pMnode = &mnode; opt.pMnode = &mnode;
opt.path = "/tmp/mnode_test_sdb"; opt.path = "/tmp/mnode_test_sdb";
taosRemoveDir(opt.path);
SSdbTable strTable = { SSdbTable strTable1;
SDB_USER, memset(&strTable1, 0, sizeof(SSdbTable));
SDB_KEY_BINARY, strTable1.sdbType = SDB_USER;
(SdbDeployFp)strDefault, strTable1.keyType = SDB_KEY_BINARY;
(SdbEncodeFp)strEncode, strTable1.deployFp = (SdbDeployFp)strDefault;
(SdbDecodeFp)strDecode, strTable1.encodeFp = (SdbEncodeFp)strEncode;
(SdbInsertFp)strInsert, strTable1.decodeFp = (SdbDecodeFp)strDecode;
(SdbUpdateFp)strUpdate, strTable1.insertFp = (SdbInsertFp)strInsert;
(SdbDeleteFp)strDelete, strTable1.updateFp = (SdbUpdateFp)strUpdate;
}; strTable1.deleteFp = (SdbDeleteFp)strDelete;
SSdbTable strTable2;
memset(&strTable2, 0, sizeof(SSdbTable));
strTable2.sdbType = SDB_VGROUP;
strTable2.keyType = SDB_KEY_INT32;
strTable2.encodeFp = (SdbEncodeFp)i32Encode;
strTable2.decodeFp = (SdbDecodeFp)i32Decode;
strTable2.insertFp = (SdbInsertFp)i32Insert;
strTable2.updateFp = (SdbUpdateFp)i32Update;
strTable2.deleteFp = (SdbDeleteFp)i32Delete;
SSdbTable strTable3;
memset(&strTable3, 0, sizeof(SSdbTable));
strTable3.sdbType = SDB_CONSUMER;
strTable3.keyType = SDB_KEY_INT64;
strTable3.encodeFp = (SdbEncodeFp)i64Encode;
strTable3.decodeFp = (SdbDecodeFp)i64Decode;
strTable3.insertFp = (SdbInsertFp)i64Insert;
strTable3.updateFp = (SdbUpdateFp)i64Update;
strTable3.deleteFp = (SdbDeleteFp)i64Delete;
pSdb = sdbInit(&opt); pSdb = sdbInit(&opt);
mnode.pSdb = pSdb; mnode.pSdb = pSdb;
ASSERT_NE(pSdb, nullptr);
ASSERT_NE(pSdb, nullptr);
ASSERT_EQ(sdbSetTable(pSdb, strTable1), 0);
ASSERT_EQ(sdbSetTable(pSdb, strTable2), 0);
ASSERT_EQ(sdbSetTable(pSdb, strTable3), 0);
ASSERT_EQ(sdbReadFile(pSdb), 0); ASSERT_EQ(sdbReadFile(pSdb), 0);
ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 2);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 5);
ASSERT_EQ(sdbUpdateVer(pSdb, 0), 1);
ASSERT_EQ(mnode.insertTimes, 4);
ASSERT_EQ(mnode.deleteTimes, 0);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k1000");
ASSERT_NE(pObj, nullptr);
EXPECT_STREQ(pObj->key, "k1000");
EXPECT_STREQ(pObj->vstr, "v1000");
ASSERT_EQ(pObj->v8, 1);
ASSERT_EQ(pObj->v16, 1);
ASSERT_EQ(pObj->v32, 1000);
ASSERT_EQ(pObj->v64, 1000);
sdbRelease(pSdb, pObj);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k2000");
ASSERT_NE(pObj, nullptr);
EXPECT_STREQ(pObj->key, "k2000");
EXPECT_STREQ(pObj->vstr, "v2000");
ASSERT_EQ(pObj->v8, 2);
ASSERT_EQ(pObj->v16, 2);
ASSERT_EQ(pObj->v32, 2000);
ASSERT_EQ(pObj->v64, 2000);
sdbRelease(pSdb, pObj);
pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k200");
ASSERT_EQ(pObj, nullptr);
pIter = NULL;
num = 0;
do {
pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj);
if (pIter == NULL) break;
ASSERT_NE(pObj, nullptr);
num++;
sdbRelease(pSdb, pObj);
} while (1);
ASSERT_EQ(num, 2);
do {
pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj);
if (pIter == NULL) break;
if (strcmp(pObj->key, "k1000") == 0) {
sdbCancelFetch(pSdb, pIter);
break;
}
} while (1);
EXPECT_STREQ(pObj->key, "k1000");
p1 = 0;
p2 = 111;
p3 = 222;
sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc1, &p1, &p2, &p3);
ASSERT_EQ(p1, 334);
p1 = 0;
p2 = 111;
p3 = 222;
sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3);
ASSERT_EQ(p1, 669);
p1 = 0;
p2 = 111;
p3 = 222;
sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseFail, &p1, &p2, &p3);
ASSERT_EQ(p1, 333);
int32_t i32key = 6;
SI32Obj *pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &i32key);
ASSERT_NE(pI32Obj, nullptr);
ASSERT_EQ(pI32Obj->key, 6);
ASSERT_EQ(pI32Obj->v8, 6);
ASSERT_EQ(pI32Obj->v16, 6);
ASSERT_EQ(pI32Obj->v32, 6000);
ASSERT_EQ(pI32Obj->v64, 6000);
sdbRelease(pSdb, pI32Obj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 4);
ASSERT_EQ(sdbGetMaxId(pSdb, SDB_VGROUP), 7);
int64_t i64key = 7;
SI64Obj *pI64Obj = (SI64Obj *)sdbAcquire(pSdb, SDB_CONSUMER, &i64key);
ASSERT_NE(pI64Obj, nullptr);
ASSERT_EQ(pI64Obj->key, 7);
ASSERT_EQ(pI64Obj->v8, 7);
ASSERT_EQ(pI64Obj->v16, 7);
ASSERT_EQ(pI64Obj->v32, 7000);
ASSERT_EQ(pI64Obj->v64, 7000);
sdbRelease(pSdb, pI64Obj);
ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 1);
ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 4);
ASSERT_EQ(mnode.insertTimes, 4);
ASSERT_EQ(mnode.deleteTimes, 0);
{
SI32Obj i32Obj = {0};
int32_t key = 6;
i32SetDefault(&i32Obj, key);
pRaw = i32Encode(&i32Obj);
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPING);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key);
ASSERT_EQ(pI32Obj, nullptr);
int32_t code = terrno;
ASSERT_EQ(code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
SI32Obj i32Obj = {0};
int32_t key = 8;
i32SetDefault(&i32Obj, key);
pRaw = i32Encode(&i32Obj);
EXPECT_NE(sdbSetRawStatus(pRaw, SDB_STATUS_INIT), 0);
sdbSetRawStatus(pRaw, SDB_STATUS_CREATING);
ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key);
ASSERT_EQ(pI32Obj, nullptr);
int32_t code = terrno;
ASSERT_EQ(code, TSDB_CODE_SDB_OBJ_CREATING);
}
sdbCleanup(pSdb); sdbCleanup(pSdb);
ASSERT_EQ(mnode.insertTimes, 5);
ASSERT_EQ(mnode.deleteTimes, 5);
} }

2
source/dnode/mnode/sdb/inc/sdbInt.h
View File

@ -31,8 +31,6 @@ extern "C" {
#define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }} #define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }}
#define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }} #define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }}
#define SDB_MAX_SIZE (32 * 1024)
typedef struct SSdbRaw { typedef struct SSdbRaw {
int8_t type; int8_t type;
int8_t status; int8_t status;

12
source/dnode/mnode/sdb/src/sdb.c
View File

@ -28,12 +28,12 @@ SSdb *sdbInit(SSdbOpt *pOption) {
return NULL; return NULL;
} }
char path[PATH_MAX + 100]; char path[PATH_MAX + 100] = {0};
snprintf(path, PATH_MAX + 100, "%s%sdata", pOption->path, TD_DIRSEP); snprintf(path, sizeof(path), "%s%sdata", pOption->path, TD_DIRSEP);
pSdb->currDir = strdup(path); pSdb->currDir = strdup(path);
snprintf(path, PATH_MAX + 100, "%s%ssync", pOption->path, TD_DIRSEP); snprintf(path, sizeof(path), "%s%ssync", pOption->path, TD_DIRSEP);
pSdb->syncDir = strdup(path); pSdb->syncDir = strdup(path);
snprintf(path, PATH_MAX + 100, "%s%stmp", pOption->path, TD_DIRSEP); snprintf(path, sizeof(path), "%s%stmp", pOption->path, TD_DIRSEP);
pSdb->tmpDir = strdup(path); pSdb->tmpDir = strdup(path);
if (pSdb->currDir == NULL || pSdb->currDir == NULL || pSdb->currDir == NULL) { if (pSdb->currDir == NULL || pSdb->currDir == NULL || pSdb->currDir == NULL) {
sdbCleanup(pSdb); sdbCleanup(pSdb);
@ -50,7 +50,7 @@ SSdb *sdbInit(SSdbOpt *pOption) {
for (ESdbType i = 0; i < SDB_MAX; ++i) { for (ESdbType i = 0; i < SDB_MAX; ++i) {
taosInitRWLatch(&pSdb->locks[i]); taosInitRWLatch(&pSdb->locks[i]);
pSdb->maxId[i] = 0; pSdb->maxId[i] = 0;
pSdb->tableVer[i] = -1; pSdb->tableVer[i] = 0;
pSdb->keyTypes[i] = SDB_KEY_INT32; pSdb->keyTypes[i] = SDB_KEY_INT32;
} }
@ -99,7 +99,7 @@ void sdbCleanup(SSdb *pSdb) {
taosHashClear(hash); taosHashClear(hash);
taosHashCleanup(hash); taosHashCleanup(hash);
pSdb->hashObjs[i] = NULL; pSdb->hashObjs[i] = NULL;
mDebug("sdb table:%d is cleaned up", i); mDebug("sdb table:%s is cleaned up", sdbTableName(i));
} }
taosMemoryFree(pSdb); taosMemoryFree(pSdb);

26
source/dnode/mnode/sdb/src/sdbFile.c
View File

@ -18,7 +18,7 @@
#include "tchecksum.h" #include "tchecksum.h"
#include "wal.h" #include "wal.h"
#define SDB_TABLE_SIZE 24 #define SDB_TABLE_SIZE 24
#define SDB_RESERVE_SIZE 512 #define SDB_RESERVE_SIZE 512
static int32_t sdbRunDeployFp(SSdb *pSdb) { static int32_t sdbRunDeployFp(SSdb *pSdb) {
@ -50,7 +50,7 @@ static int32_t sdbReadFileHead(SSdb *pSdb, TdFilePtr pFile) {
} }
for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) { for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {
int64_t maxId = -1; int64_t maxId = 0;
ret = taosReadFile(pFile, &maxId, sizeof(int64_t)); ret = taosReadFile(pFile, &maxId, sizeof(int64_t));
if (ret < 0) { if (ret < 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
@ -66,7 +66,7 @@ static int32_t sdbReadFileHead(SSdb *pSdb, TdFilePtr pFile) {
} }
for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) { for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {
int64_t ver = -1; int64_t ver = 0;
ret = taosReadFile(pFile, &ver, sizeof(int64_t)); ret = taosReadFile(pFile, &ver, sizeof(int64_t));
if (ret < 0) { if (ret < 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
@ -102,7 +102,7 @@ static int32_t sdbWriteFileHead(SSdb *pSdb, TdFilePtr pFile) {
} }
for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) { for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {
int64_t maxId = -1; int64_t maxId = 0;
if (i < SDB_MAX) { if (i < SDB_MAX) {
maxId = pSdb->maxId[i]; maxId = pSdb->maxId[i];
} }
@ -113,7 +113,7 @@ static int32_t sdbWriteFileHead(SSdb *pSdb, TdFilePtr pFile) {
} }
for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) { for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {
int64_t ver = -1; int64_t ver = 0;
if (i < SDB_MAX) { if (i < SDB_MAX) {
ver = pSdb->tableVer[i]; ver = pSdb->tableVer[i];
} }
@ -165,6 +165,9 @@ int32_t sdbReadFile(SSdb *pSdb) {
return -1; return -1;
} }
int64_t tableVer[SDB_MAX] = {0};
memcpy(tableVer, pSdb->tableVer, sizeof(tableVer));
while (1) { while (1) {
readLen = sizeof(SSdbRaw); readLen = sizeof(SSdbRaw);
ret = taosReadFile(pFile, pRaw, readLen); ret = taosReadFile(pFile, pRaw, readLen);
@ -206,15 +209,16 @@ int32_t sdbReadFile(SSdb *pSdb) {
code = sdbWriteWithoutFree(pSdb, pRaw); code = sdbWriteWithoutFree(pSdb, pRaw);
if (code != 0) { if (code != 0) {
mError("failed to read file:%s since %s", file, terrstr()); mError("failed to read file:%s since %s", file, terrstr());
goto PARSE_SDB_DATA_ERROR; goto _OVER;
} }
} }
code = 0; code = 0;
pSdb->lastCommitVer = pSdb->curVer; pSdb->lastCommitVer = pSdb->curVer;
memcpy(pSdb->tableVer, tableVer, sizeof(tableVer));
mDebug("read file:%s successfully, ver:%" PRId64, file, pSdb->lastCommitVer); mDebug("read file:%s successfully, ver:%" PRId64, file, pSdb->lastCommitVer);
PARSE_SDB_DATA_ERROR: _OVER:
taosCloseFile(&pFile); taosCloseFile(&pFile);
sdbFreeRaw(pRaw); sdbFreeRaw(pRaw);
@ -259,7 +263,13 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
SSdbRow **ppRow = taosHashIterate(hash, NULL); SSdbRow **ppRow = taosHashIterate(hash, NULL);
while (ppRow != NULL) { while (ppRow != NULL) {
SSdbRow *pRow = *ppRow; SSdbRow *pRow = *ppRow;
if (pRow == NULL || pRow->status != SDB_STATUS_READY) { if (pRow == NULL) {
ppRow = taosHashIterate(hash, ppRow);
continue;
}
if (pRow->status != SDB_STATUS_READY && pRow->status != SDB_STATUS_DROPPING) {
sdbPrintOper(pSdb, pRow, "not-write");
ppRow = taosHashIterate(hash, ppRow); ppRow = taosHashIterate(hash, ppRow);
continue; continue;
} }

47
source/dnode/mnode/sdb/src/sdbHash.c
View File

@ -16,7 +16,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "sdbInt.h" #include "sdbInt.h"
static void sdbCheck(SSdb *pSdb, SSdbRow *pRow); static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow);
const char *sdbTableName(ESdbType type) { const char *sdbTableName(ESdbType type) {
switch (type) { switch (type) {
@ -65,12 +65,10 @@ const char *sdbTableName(ESdbType type) {
} }
} }
static const char *sdbStatusStr(ESdbStatus status) { static const char *sdbStatusName(ESdbStatus status) {
switch (status) { switch (status) {
case SDB_STATUS_CREATING: case SDB_STATUS_CREATING:
return "creating"; return "creating";
case SDB_STATUS_UPDATING:
return "updating";
case SDB_STATUS_DROPPING: case SDB_STATUS_DROPPING:
return "dropping"; return "dropping";
case SDB_STATUS_READY: case SDB_STATUS_READY:
@ -89,13 +87,13 @@ void sdbPrintOper(SSdb *pSdb, SSdbRow *pRow, const char *oper) {
if (keyType == SDB_KEY_BINARY) { if (keyType == SDB_KEY_BINARY) {
mTrace("%s:%s, ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), (char *)pRow->pObj, pRow->refCount, oper, mTrace("%s:%s, ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), (char *)pRow->pObj, pRow->refCount, oper,
pRow->pObj, sdbStatusStr(pRow->status)); pRow->pObj, sdbStatusName(pRow->status));
} else if (keyType == SDB_KEY_INT32) { } else if (keyType == SDB_KEY_INT32) {
mTrace("%s:%d, ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), *(int32_t *)pRow->pObj, pRow->refCount, mTrace("%s:%d, ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), *(int32_t *)pRow->pObj, pRow->refCount,
oper, pRow->pObj, sdbStatusStr(pRow->status)); oper, pRow->pObj, sdbStatusName(pRow->status));
} else if (keyType == SDB_KEY_INT64) { } else if (keyType == SDB_KEY_INT64) {
mTrace("%s:%" PRId64 ", ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), *(int64_t *)pRow->pObj, mTrace("%s:%" PRId64 ", ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), *(int64_t *)pRow->pObj,
pRow->refCount, oper, pRow->pObj, sdbStatusStr(pRow->status)); pRow->refCount, oper, pRow->pObj, sdbStatusName(pRow->status));
} else { } else {
} }
} }
@ -116,7 +114,7 @@ static SHashObj *sdbGetHash(SSdb *pSdb, int32_t type) {
} }
static int32_t sdbGetkeySize(SSdb *pSdb, ESdbType type, const void *pKey) { static int32_t sdbGetkeySize(SSdb *pSdb, ESdbType type, const void *pKey) {
int32_t keySize; int32_t keySize = 0;
EKeyType keyType = pSdb->keyTypes[type]; EKeyType keyType = pSdb->keyTypes[type];
if (keyType == SDB_KEY_INT32) { if (keyType == SDB_KEY_INT32) {
@ -149,7 +147,7 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
if (taosHashPut(hash, pRow->pObj, keySize, &pRow, sizeof(void *)) != 0) { if (taosHashPut(hash, pRow->pObj, keySize, &pRow, sizeof(void *)) != 0) {
taosWUnLockLatch(pLock); taosWUnLockLatch(pLock);
sdbFreeRow(pSdb, pRow, false); sdbFreeRow(pSdb, pRow, false);
terrno = TSDB_CODE_SDB_OBJ_ALREADY_THERE; terrno = TSDB_CODE_OUT_OF_MEMORY;
return terrno; return terrno;
} }
@ -183,18 +181,18 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pNewRow, int32_t keySize) { static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pNewRow, int32_t keySize) {
SRWLatch *pLock = &pSdb->locks[pNewRow->type]; SRWLatch *pLock = &pSdb->locks[pNewRow->type];
taosRLockLatch(pLock); taosWLockLatch(pLock);
SSdbRow **ppOldRow = taosHashGet(hash, pNewRow->pObj, keySize); SSdbRow **ppOldRow = taosHashGet(hash, pNewRow->pObj, keySize);
if (ppOldRow == NULL || *ppOldRow == NULL) { if (ppOldRow == NULL || *ppOldRow == NULL) {
taosRUnLockLatch(pLock); taosWUnLockLatch(pLock);
return sdbInsertRow(pSdb, hash, pRaw, pNewRow, keySize); return sdbInsertRow(pSdb, hash, pRaw, pNewRow, keySize);
} }
SSdbRow *pOldRow = *ppOldRow; SSdbRow *pOldRow = *ppOldRow;
pOldRow->status = pRaw->status; pOldRow->status = pRaw->status;
sdbPrintOper(pSdb, pOldRow, "update"); sdbPrintOper(pSdb, pOldRow, "update");
taosRUnLockLatch(pLock); taosWUnLockLatch(pLock);
int32_t code = 0; int32_t code = 0;
SdbUpdateFp updateFp = pSdb->updateFps[pNewRow->type]; SdbUpdateFp updateFp = pSdb->updateFps[pNewRow->type];
@ -230,8 +228,7 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
pSdb->tableVer[pOldRow->type]++; pSdb->tableVer[pOldRow->type]++;
sdbFreeRow(pSdb, pRow, false); sdbFreeRow(pSdb, pRow, false);
sdbCheck(pSdb, pOldRow); sdbCheckRow(pSdb, pOldRow);
// sdbRelease(pSdb, pOldRow->pObj);
return 0; return 0;
} }
@ -254,7 +251,6 @@ int32_t sdbWriteWithoutFree(SSdb *pSdb, SSdbRaw *pRaw) {
case SDB_STATUS_CREATING: case SDB_STATUS_CREATING:
code = sdbInsertRow(pSdb, hash, pRaw, pRow, keySize); code = sdbInsertRow(pSdb, hash, pRaw, pRow, keySize);
break; break;
case SDB_STATUS_UPDATING:
case SDB_STATUS_READY: case SDB_STATUS_READY:
case SDB_STATUS_DROPPING: case SDB_STATUS_DROPPING:
code = sdbUpdateRow(pSdb, hash, pRaw, pRow, keySize); code = sdbUpdateRow(pSdb, hash, pRaw, pRow, keySize);
@ -295,7 +291,6 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
SSdbRow *pRow = *ppRow; SSdbRow *pRow = *ppRow;
switch (pRow->status) { switch (pRow->status) {
case SDB_STATUS_READY: case SDB_STATUS_READY:
case SDB_STATUS_UPDATING:
atomic_add_fetch_32(&pRow->refCount, 1); atomic_add_fetch_32(&pRow->refCount, 1);
pRet = pRow->pObj; pRet = pRow->pObj;
sdbPrintOper(pSdb, pRow, "acquire"); sdbPrintOper(pSdb, pRow, "acquire");
@ -315,9 +310,9 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
return pRet; return pRet;
} }
static void sdbCheck(SSdb *pSdb, SSdbRow *pRow) { static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) {
SRWLatch *pLock = &pSdb->locks[pRow->type]; SRWLatch *pLock = &pSdb->locks[pRow->type];
taosRLockLatch(pLock); taosWLockLatch(pLock);
int32_t ref = atomic_load_32(&pRow->refCount); int32_t ref = atomic_load_32(&pRow->refCount);
sdbPrintOper(pSdb, pRow, "check"); sdbPrintOper(pSdb, pRow, "check");
@ -325,7 +320,7 @@ static void sdbCheck(SSdb *pSdb, SSdbRow *pRow) {
sdbFreeRow(pSdb, pRow, true); sdbFreeRow(pSdb, pRow, true);
} }
taosRUnLockLatch(pLock); taosWUnLockLatch(pLock);
} }
void sdbRelease(SSdb *pSdb, void *pObj) { void sdbRelease(SSdb *pSdb, void *pObj) {
@ -335,7 +330,7 @@ void sdbRelease(SSdb *pSdb, void *pObj) {
if (pRow->type >= SDB_MAX) return; if (pRow->type >= SDB_MAX) return;
SRWLatch *pLock = &pSdb->locks[pRow->type]; SRWLatch *pLock = &pSdb->locks[pRow->type];
taosRLockLatch(pLock); taosWLockLatch(pLock);
int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1); int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1);
sdbPrintOper(pSdb, pRow, "release"); sdbPrintOper(pSdb, pRow, "release");
@ -343,7 +338,7 @@ void sdbRelease(SSdb *pSdb, void *pObj) {
sdbFreeRow(pSdb, pRow, true); sdbFreeRow(pSdb, pRow, true);
} }
taosRUnLockLatch(pLock); taosWUnLockLatch(pLock);
} }
void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) { void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
@ -355,16 +350,6 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
SRWLatch *pLock = &pSdb->locks[type]; SRWLatch *pLock = &pSdb->locks[type];
taosRLockLatch(pLock); taosRLockLatch(pLock);
#if 0
if (pIter != NULL) {
SSdbRow *pLastRow = *(SSdbRow **)pIter;
int32_t ref = atomic_load_32(&pLastRow->refCount);
if (ref <= 0 && pLastRow->status == SDB_STATUS_DROPPED) {
sdbFreeRow(pSdb, pLastRow);
}
}
#endif
SSdbRow **ppRow = taosHashIterate(hash, pIter); SSdbRow **ppRow = taosHashIterate(hash, pIter);
while (ppRow != NULL) { while (ppRow != NULL) {
SSdbRow *pRow = *ppRow; SSdbRow *pRow = *ppRow;

5
source/dnode/mnode/sdb/src/sdbRaw.c
View File

@ -134,6 +134,11 @@ int32_t sdbSetRawStatus(SSdbRaw *pRaw, ESdbStatus status) {
return -1; return -1;
} }
if (status == SDB_STATUS_INIT) {
terrno = TSDB_CODE_INVALID_PARA;
return -1;
}
pRaw->status = status; pRaw->status = status;
return 0; return 0;
} }

2
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -190,7 +190,7 @@ int vnodeSyncCommit(SVnode *pVnode) {
} }
int vnodeCommit(SVnode *pVnode) { int vnodeCommit(SVnode *pVnode) {
SVnodeInfo info; SVnodeInfo info = {0};
char dir[TSDB_FILENAME_LEN]; char dir[TSDB_FILENAME_LEN];
vInfo("vgId:%d start to commit, version: %" PRId64, TD_VID(pVnode), pVnode->state.applied); vInfo("vgId:%d start to commit, version: %" PRId64, TD_VID(pVnode), pVnode->state.applied);

2
source/libs/transport/src/transSrv.c
View File

@ -207,7 +207,7 @@ static bool addHandleToAcceptloop(void* arg);
SExHandle* exh2 = uvAcquireExHandle(refId); \ SExHandle* exh2 = uvAcquireExHandle(refId); \
if (exh2 == NULL || refId != exh2->refId) { \ if (exh2 == NULL || refId != exh2->refId) { \
tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \ tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
exh1->refId, refId); \ exh2 ? exh2->refId : 0, refId); \
goto _return1; \ goto _return1; \
} \ } \
} else if (refId == 0) { \ } else if (refId == 0) { \

1
source/util/src/terror.c
View File

@ -93,6 +93,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_OPS_NOT_SUPPORT, "Operation not support
TAOS_DEFINE_ERROR(TSDB_CODE_MSG_NOT_PROCESSED, "Message not processed") TAOS_DEFINE_ERROR(TSDB_CODE_MSG_NOT_PROCESSED, "Message not processed")
TAOS_DEFINE_ERROR(TSDB_CODE_CFG_NOT_FOUND, "Config not found") TAOS_DEFINE_ERROR(TSDB_CODE_CFG_NOT_FOUND, "Config not found")
TAOS_DEFINE_ERROR(TSDB_CODE_REPEAT_INIT, "Repeat initialization") TAOS_DEFINE_ERROR(TSDB_CODE_REPEAT_INIT, "Repeat initialization")
TAOS_DEFINE_ERROR(TSDB_CODE_DUP_KEY, "Cannot add duplicate keys to hash")
TAOS_DEFINE_ERROR(TSDB_CODE_REF_NO_MEMORY, "Ref out of memory") TAOS_DEFINE_ERROR(TSDB_CODE_REF_NO_MEMORY, "Ref out of memory")
TAOS_DEFINE_ERROR(TSDB_CODE_REF_FULL, "too many Ref Objs") TAOS_DEFINE_ERROR(TSDB_CODE_REF_FULL, "too many Ref Objs")

3
source/util/src/thash.c
View File

@ -310,6 +310,7 @@ int32_t taosHashGetSize(const SHashObj *pHashObj) {
int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, const void *data, size_t size) { int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, const void *data, size_t size) {
if (pHashObj == NULL || key == NULL || keyLen == 0) { if (pHashObj == NULL || key == NULL || keyLen == 0) {
terrno = TSDB_CODE_INVALID_PTR;
return -1; return -1;
} }
@ -378,6 +379,8 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, const vo
} }
doUpdateHashNode(pHashObj, pe, prev, pNode, pNewNode); doUpdateHashNode(pHashObj, pe, prev, pNode, pNewNode);
} else {
terrno = TSDB_CODE_DUP_KEY;
} }
taosHashEntryWUnlock(pHashObj, pe); taosHashEntryWUnlock(pHashObj, pe);

20
source/util/src/tprocess.c
View File

@ -175,7 +175,6 @@ static int32_t taosProcQueuePush(SProcObj *pProc, SProcQueue *pQueue, const char
if (handle != 0 && ftype == PROC_FUNC_REQ) { if (handle != 0 && ftype == PROC_FUNC_REQ) {
if (taosHashPut(pProc->hash, &handle, sizeof(int64_t), &handleRef, sizeof(int64_t)) != 0) { if (taosHashPut(pProc->hash, &handle, sizeof(int64_t), &handleRef, sizeof(int64_t)) != 0) {
taosThreadMutexUnlock(&pQueue->mutex); taosThreadMutexUnlock(&pQueue->mutex);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
} }
@ -227,8 +226,8 @@ static int32_t taosProcQueuePush(SProcObj *pProc, SProcQueue *pQueue, const char
taosThreadMutexUnlock(&pQueue->mutex); taosThreadMutexUnlock(&pQueue->mutex);
tsem_post(&pQueue->sem); tsem_post(&pQueue->sem);
uTrace("proc:%s, push msg at pos:%d ftype:%d remain:%d, head:%d %p body:%d %p", pQueue->name, pos, ftype, uTrace("proc:%s, push msg at pos:%d ftype:%d remain:%d handle:%p ref:%" PRId64 ", head:%d %p body:%d %p",
pQueue->items, headLen, pHead, bodyLen, pBody); pQueue->name, pos, ftype, pQueue->items, (void *)handle, handleRef, headLen, pHead, bodyLen, pBody);
return 0; return 0;
} }
@ -455,25 +454,28 @@ void taosProcCleanup(SProcObj *pProc) {
} }
int32_t taosProcPutToChildQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen, int32_t taosProcPutToChildQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
void *handle, int64_t handleRef, EProcFuncType ftype) { void *handle, int64_t ref, EProcFuncType ftype) {
if (ftype != PROC_FUNC_REQ) { if (ftype != PROC_FUNC_REQ) {
terrno = TSDB_CODE_INVALID_PARA; terrno = TSDB_CODE_INVALID_PARA;
return -1; return -1;
} }
return taosProcQueuePush(pProc, pProc->pChildQueue, pHead, headLen, pBody, bodyLen, (int64_t)handle, handleRef, return taosProcQueuePush(pProc, pProc->pChildQueue, pHead, headLen, pBody, bodyLen, (int64_t)handle, ref, ftype);
ftype);
} }
int64_t taosProcRemoveHandle(SProcObj *pProc, void *handle) { int64_t taosProcRemoveHandle(SProcObj *pProc, void *handle) {
int64_t h = (int64_t)handle; int64_t h = (int64_t)handle;
taosThreadMutexLock(&pProc->pChildQueue->mutex); taosThreadMutexLock(&pProc->pChildQueue->mutex);
int64_t *handleRef = taosHashGet(pProc->hash, &h, sizeof(int64_t)); int64_t *pRef = taosHashGet(pProc->hash, &h, sizeof(int64_t));
int64_t ref = 0;
if (pRef != NULL) {
ref = *pRef;
}
taosHashRemove(pProc->hash, &h, sizeof(int64_t)); taosHashRemove(pProc->hash, &h, sizeof(int64_t));
taosThreadMutexUnlock(&pProc->pChildQueue->mutex); taosThreadMutexUnlock(&pProc->pChildQueue->mutex);
if (handleRef == NULL) return 0; return ref;
return *handleRef;
} }
void taosProcCloseHandles(SProcObj *pProc, void (*HandleFp)(void *handle)) { void taosProcCloseHandles(SProcObj *pProc, void (*HandleFp)(void *handle)) {

6
tests/script/jenkins/basic.txt
View File

@ -66,10 +66,10 @@
# --- stable # --- stable
./test.sh -f tsim/stable/disk.sim ./test.sh -f tsim/stable/disk.sim
#./test.sh -f tsim/stable/dnode3.sim ./test.sh -f tsim/stable/dnode3.sim
./test.sh -f tsim/stable/metrics.sim ./test.sh -f tsim/stable/metrics.sim
./test.sh -f tsim/stable/refcount.sim ./test.sh -f tsim/stable/refcount.sim
# ./test.sh -f tsim/stable/show.sim #./test.sh -f tsim/stable/show.sim
./test.sh -f tsim/stable/values.sim ./test.sh -f tsim/stable/values.sim
./test.sh -f tsim/stable/vnode3.sim ./test.sh -f tsim/stable/vnode3.sim
@ -81,7 +81,7 @@
./test.sh -f tsim/insert/backquote.sim -m ./test.sh -f tsim/insert/backquote.sim -m
./test.sh -f tsim/parser/fourArithmetic-basic.sim -m ./test.sh -f tsim/parser/fourArithmetic-basic.sim -m
./test.sh -f tsim/query/interval-offset.sim -m ./test.sh -f tsim/query/interval-offset.sim -m
#./test.sh -f tsim/tmq/basic3.sim -m ./test.sh -f tsim/tmq/basic3.sim -m
./test.sh -f tsim/stable/vnode3.sim -m ./test.sh -f tsim/stable/vnode3.sim -m
./test.sh -f tsim/qnode/basic1.sim -m ./test.sh -f tsim/qnode/basic1.sim -m
./test.sh -f tsim/mnode/basic1.sim -m ./test.sh -f tsim/mnode/basic1.sim -m

3
tests/script/tsim/show/basic.sim
View File

@ -3,9 +3,6 @@ system sh/deploy.sh -n dnode1 -i 1
system sh/deploy.sh -n dnode2 -i 2 system sh/deploy.sh -n dnode2 -i 2
system sh/exec.sh -n dnode1 -s start system sh/exec.sh -n dnode1 -s start
system sh/exec.sh -n dnode2 -s start system sh/exec.sh -n dnode2 -s start
# after mnode support, del sleep 2000, and check dnode status
sleep 2000
sql connect sql connect
#$loop_cnt = 0 #$loop_cnt = 0