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

Merge remote-tracking branch 'origin/3.0' into feature/3.0_wxy

This commit is contained in:
Xiaoyu Wang 2022-03-25 01:41:35 -04:00
parent 9765678e75 db62c347a4
commit 4ebc3234f2
44 changed files with 2161 additions and 309 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/common/src/tglobal.c
View File

@ -180,6 +180,10 @@ static int32_t taosSetTfsCfg(SConfig *pCfg) {
memcpy(&tsDiskCfg[index], pCfg, sizeof(SDiskCfg)); memcpy(&tsDiskCfg[index], pCfg, sizeof(SDiskCfg));
if (pCfg->level == 0 && pCfg->primary == 1) { if (pCfg->level == 0 && pCfg->primary == 1) {
tstrncpy(tsDataDir, pCfg->dir, PATH_MAX); tstrncpy(tsDataDir, pCfg->dir, PATH_MAX);
if (taosMkDir(tsDataDir) != 0) {
uError("failed to create dataDir:%s since %s", tsDataDir, terrstr());
return -1;
}
} }
if (taosMkDir(pCfg->dir) != 0) { if (taosMkDir(pCfg->dir) != 0) {
uError("failed to create tfsDir:%s since %s", tsDataDir, terrstr()); uError("failed to create tfsDir:%s since %s", tsDataDir, terrstr());

6
source/dnode/mgmt/bnode/src/bmWorker.c
View File

@ -77,7 +77,11 @@ int32_t bmStartWorker(SBnodeMgmt *pMgmt) {
return -1; return -1;
} }
dDebug("bnode workers are initialized");
return 0; return 0;
} }
void bmStopWorker(SBnodeMgmt *pMgmt) { tMultiWorkerCleanup(&pMgmt->writeWorker); } void bmStopWorker(SBnodeMgmt *pMgmt) {
tMultiWorkerCleanup(&pMgmt->writeWorker);
dDebug("bnode workers are closed");
}

22
source/dnode/mgmt/container/inc/dnd.h
View File

@ -19,7 +19,6 @@
#include "os.h" #include "os.h"
#include "cJSON.h" #include "cJSON.h"
#include "monitor.h"
#include "tcache.h" #include "tcache.h"
#include "tcrc32c.h" #include "tcrc32c.h"
#include "tdatablock.h" #include "tdatablock.h"
@ -36,8 +35,7 @@
#include "tworker.h" #include "tworker.h"
#include "dnode.h" #include "dnode.h"
#include "tfs.h" #include "monitor.h"
#include "wal.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
@ -53,7 +51,6 @@ extern "C" {
typedef enum { DNODE, VNODES, QNODE, SNODE, MNODE, BNODE, NODE_MAX } ENodeType; typedef enum { DNODE, VNODES, QNODE, SNODE, MNODE, BNODE, NODE_MAX } ENodeType;
typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EDndStatus; typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EDndStatus;
typedef enum { DND_ENV_INIT, DND_ENV_READY, DND_ENV_CLEANUP } EEnvStatus; typedef enum { DND_ENV_INIT, DND_ENV_READY, DND_ENV_CLEANUP } EEnvStatus;
typedef enum { DND_WORKER_SINGLE, DND_WORKER_MULTI } EWorkerType;
typedef enum { PROC_SINGLE, PROC_CHILD, PROC_PARENT } EProcType; typedef enum { PROC_SINGLE, PROC_CHILD, PROC_PARENT } EProcType;
typedef struct SMgmtFp SMgmtFp; typedef struct SMgmtFp SMgmtFp;
@ -127,29 +124,30 @@ typedef struct SDnode {
bool dropped; bool dropped;
EDndStatus status; EDndStatus status;
EDndEvent event; EDndEvent event;
EProcType procType;
SStartupReq startup; SStartupReq startup;
TdFilePtr pLockFile; TdFilePtr pLockFile;
STransMgmt trans; STransMgmt trans;
SMgmtWrapper wrappers[NODE_MAX]; SMgmtWrapper wrappers[NODE_MAX];
} SDnode; } SDnode;
EDndStatus dndGetStatus(SDnode *pDnode); EDndStatus dndGetStatus(SDnode *pDnode);
void dndSetStatus(SDnode *pDnode, EDndStatus stat); void dndSetStatus(SDnode *pDnode, EDndStatus stat);
SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType nodeType); void dndSetMsgHandle(SMgmtWrapper *pWrapper, int32_t msgType, NodeMsgFp nodeMsgFp, int32_t vgId);
void dndSetMsgHandle(SMgmtWrapper *pWrapper, int32_t msgType, NodeMsgFp nodeMsgFp, int32_t vgId); void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dndReportStartup(SDnode *pDnode, char *pName, char *pDesc); void dndSendMonitorReport(SDnode *pDnode);
void dndSendMonitorReport(SDnode *pDnode);
int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, SEpSet *pEpSet, SRpcMsg *pMsg); int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, SEpSet *pEpSet, SRpcMsg *pMsg);
void dndSendRsp(SMgmtWrapper *pWrapper, SRpcMsg *pRsp); void dndSendRsp(SMgmtWrapper *pWrapper, SRpcMsg *pRsp);
int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg); int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg);
int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed); int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed);
int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed); int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed);
SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType nodeType);
int32_t dndMarkWrapper(SMgmtWrapper *pWrapper);
void dndReleaseWrapper(SMgmtWrapper *pWrapper);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

6
source/dnode/mgmt/container/inc/dndInt.h
View File

@ -34,7 +34,7 @@ int32_t dndInit();
void dndCleanup(); void dndCleanup();
const char *dndStatStr(EDndStatus stat); const char *dndStatStr(EDndStatus stat);
void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup); void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup);
TdFilePtr dndCheckRunning(char *dataDir); TdFilePtr dndCheckRunning(const char *dataDir);
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg); void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg);
// dndMsg.c // dndMsg.c
@ -50,10 +50,6 @@ SDnode *dndCreate(const SDnodeOpt *pOption);
void dndClose(SDnode *pDnode); void dndClose(SDnode *pDnode);
void dndHandleEvent(SDnode *pDnode, EDndEvent event); void dndHandleEvent(SDnode *pDnode, EDndEvent event);
SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType nodeType);
int32_t dndMarkWrapper(SMgmtWrapper *pWrapper);
void dndReleaseWrapper(SMgmtWrapper *pWrapper);
// dndTransport.c // dndTransport.c
int32_t dndInitServer(SDnode *pDnode); int32_t dndInitServer(SDnode *pDnode);
void dndCleanupServer(SDnode *pDnode); void dndCleanupServer(SDnode *pDnode);

9
source/dnode/mgmt/container/src/dndExec.c
View File

@ -50,16 +50,23 @@ int32_t dndOpenNode(SMgmtWrapper *pWrapper) {
} }
void dndCloseNode(SMgmtWrapper *pWrapper) { void dndCloseNode(SMgmtWrapper *pWrapper) {
dDebug("node:%s, start to close", pWrapper->name);
taosWLockLatch(&pWrapper->latch); taosWLockLatch(&pWrapper->latch);
if (pWrapper->deployed) { if (pWrapper->deployed) {
(*pWrapper->fp.closeFp)(pWrapper); (*pWrapper->fp.closeFp)(pWrapper);
pWrapper->deployed = false; pWrapper->deployed = false;
} }
taosWUnLockLatch(&pWrapper->latch);
while (pWrapper->refCount > 0) {
taosMsleep(10);
}
if (pWrapper->pProc) { if (pWrapper->pProc) {
taosProcCleanup(pWrapper->pProc); taosProcCleanup(pWrapper->pProc);
pWrapper->pProc = NULL; pWrapper->pProc = NULL;
} }
taosWUnLockLatch(&pWrapper->latch); dDebug("node:%s, has been closed", pWrapper->name);
} }
static int32_t dndRunInSingleProcess(SDnode *pDnode) { static int32_t dndRunInSingleProcess(SDnode *pDnode) {

15
source/dnode/mgmt/container/src/dndInt.c
View File

@ -15,11 +15,12 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "dndInt.h" #include "dndInt.h"
#include "wal.h"
static int8_t once = DND_ENV_INIT; static int8_t once = DND_ENV_INIT;
int32_t dndInit() { int32_t dndInit() {
dDebug("start to init dnode env"); dInfo("start to init dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) { if (atomic_val_compare_exchange_8(&once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
terrno = TSDB_CODE_REPEAT_INIT; terrno = TSDB_CODE_REPEAT_INIT;
dError("failed to init dnode env since %s", terrstr()); dError("failed to init dnode env since %s", terrstr());
@ -52,7 +53,7 @@ int32_t dndInit() {
} }
void dndCleanup() { void dndCleanup() {
dDebug("start to cleanup dnode env"); dInfo("start to cleanup dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) { if (atomic_val_compare_exchange_8(&once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
dError("dnode env is already cleaned up"); dError("dnode env is already cleaned up");
return; return;
@ -92,7 +93,7 @@ const char *dndStatStr(EDndStatus status) {
} }
} }
void dndReportStartup(SDnode *pDnode, char *pName, char *pDesc) { void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
SStartupReq *pStartup = &pDnode->startup; SStartupReq *pStartup = &pDnode->startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN); tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN); tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);
@ -104,21 +105,21 @@ void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) {
pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING); pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING);
} }
TdFilePtr dndCheckRunning(char *dataDir) { TdFilePtr dndCheckRunning(const char *dataDir) {
char filepath[PATH_MAX] = {0}; char filepath[PATH_MAX] = {0};
snprintf(filepath, sizeof(filepath), "%s/.running", dataDir); snprintf(filepath, sizeof(filepath), "%s/.running", dataDir);
TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC); TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) { if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s, quit", filepath, terrstr()); dError("failed to lock file:%s since %s", filepath, terrstr());
return NULL; return NULL;
} }
int32_t ret = taosLockFile(pFile); int32_t ret = taosLockFile(pFile);
if (ret != 0) { if (ret != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s, quit", filepath, terrstr()); dError("failed to lock file:%s since %s", filepath, terrstr());
taosCloseFile(&pFile); taosCloseFile(&pFile);
return NULL; return NULL;
} }
@ -129,12 +130,10 @@ TdFilePtr dndCheckRunning(char *dataDir) {
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) { void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("startup req is received"); dDebug("startup req is received");
SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq)); SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq));
dndGetStartup(pDnode, pStartup); dndGetStartup(pDnode, pStartup);
dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished); dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished);
SRpcMsg rpcRsp = {.handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq)}; SRpcMsg rpcRsp = {.handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq)};
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }

33
source/dnode/mgmt/container/src/dndMonitor.c
View File

@ -22,7 +22,12 @@ static int32_t dndGetMonitorDiskInfo(SDnode *pDnode, SMonDiskInfo *pInfo) {
tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name)); tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name));
pInfo->tempdir.size = tsTempSpace.size; pInfo->tempdir.size = tsTempSpace.size;
return vmMonitorTfsInfo(dndAcquireWrapper(pDnode, VNODES), pInfo); SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, VNODES);
if (pWrapper != NULL) {
vmMonitorTfsInfo(pWrapper, pInfo);
dndReleaseWrapper(pWrapper);
}
return 0;
} }
static void dndGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) { static void dndGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
@ -45,8 +50,17 @@ static void dndGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) {
taosGetCardInfo(&pInfo->net_in, &pInfo->net_out); taosGetCardInfo(&pInfo->net_in, &pInfo->net_out);
taosGetProcIO(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk); taosGetProcIO(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk);
vmMonitorVnodeReqs(dndAcquireWrapper(pDnode, VNODES), pInfo); SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, VNODES);
pInfo->has_mnode = (dndAcquireWrapper(pDnode, MNODE)->required); if (pWrapper != NULL) {
vmMonitorVnodeReqs(pWrapper, pInfo);
dndReleaseWrapper(pWrapper);
}
pWrapper = dndAcquireWrapper(pDnode, MNODE);
if (pWrapper != NULL) {
pInfo->has_mnode = pWrapper->required;
dndReleaseWrapper(pWrapper);
}
} }
void dndSendMonitorReport(SDnode *pDnode) { void dndSendMonitorReport(SDnode *pDnode) {
@ -63,10 +77,15 @@ void dndSendMonitorReport(SDnode *pDnode) {
SMonClusterInfo clusterInfo = {0}; SMonClusterInfo clusterInfo = {0};
SMonVgroupInfo vgroupInfo = {0}; SMonVgroupInfo vgroupInfo = {0};
SMonGrantInfo grantInfo = {0}; SMonGrantInfo grantInfo = {0};
if (mmMonitorMnodeInfo(dndAcquireWrapper(pDnode, MNODE), &clusterInfo, &vgroupInfo, &grantInfo) == 0) {
monSetClusterInfo(pMonitor, &clusterInfo); SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, MNODE);
monSetVgroupInfo(pMonitor, &vgroupInfo); if (pWrapper != NULL) {
monSetGrantInfo(pMonitor, &grantInfo); if (mmMonitorMnodeInfo(pWrapper, &clusterInfo, &vgroupInfo, &grantInfo) == 0) {
monSetClusterInfo(pMonitor, &clusterInfo);
monSetVgroupInfo(pMonitor, &vgroupInfo);
monSetGrantInfo(pMonitor, &grantInfo);
}
dndReleaseWrapper(pWrapper);
} }
SMonDnodeInfo dnodeInfo = {0}; SMonDnodeInfo dnodeInfo = {0};

2
source/dnode/mgmt/container/src/dndMsg.c
View File

@ -20,8 +20,8 @@ static void dndUpdateMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, DNODE); SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, DNODE);
if (pWrapper != NULL) { if (pWrapper != NULL) {
dmUpdateMnodeEpSet(pWrapper->pMgmt, pEpSet); dmUpdateMnodeEpSet(pWrapper->pMgmt, pEpSet);
dndReleaseWrapper(pWrapper);
} }
dndReleaseWrapper(pWrapper);
} }
static inline NodeMsgFp dndGetMsgFp(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) { static inline NodeMsgFp dndGetMsgFp(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {

2
source/dnode/mgmt/container/src/dndObj.c
View File

@ -58,7 +58,7 @@ static void dndClearMemory(SDnode *pDnode) {
SDnode *dndCreate(const SDnodeOpt *pOption) { SDnode *dndCreate(const SDnodeOpt *pOption) {
dInfo("start to create dnode object"); dInfo("start to create dnode object");
int32_t code = -1; int32_t code = -1;
char path[PATH_MAX]; char path[PATH_MAX] = {0};
SDnode *pDnode = NULL; SDnode *pDnode = NULL;
pDnode = calloc(1, sizeof(SDnode)); pDnode = calloc(1, sizeof(SDnode));

36
source/dnode/mgmt/container/src/dndTransport.c
View File

@ -146,9 +146,14 @@ static void dndProcessRequest(void *param, SRpcMsg *pReq, SEpSet *pEpSet) {
static void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pRpcMsg, SRpcMsg *pRpcRsp) { static void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pRpcMsg, SRpcMsg *pRpcRsp) {
STransMgmt *pMgmt = &pDnode->trans; STransMgmt *pMgmt = &pDnode->trans;
SEpSet epSet = {0};
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, DNODE);
if (pWrapper != NULL) {
dmGetMnodeEpSet(pWrapper->pMgmt, &epSet);
dndReleaseWrapper(pWrapper);
}
SEpSet epSet = {0};
dmGetMnodeEpSet(dndAcquireWrapper(pDnode, DNODE)->pMgmt, &epSet);
rpcSendRecv(pMgmt->clientRpc, &epSet, pRpcMsg, pRpcRsp); rpcSendRecv(pMgmt->clientRpc, &epSet, pRpcMsg, pRpcRsp);
} }
@ -182,9 +187,14 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char
return 0; return 0;
} }
if (mmGetUserAuth(dndAcquireWrapper(pDnode, MNODE), user, spi, encrypt, secret, ckey) == 0) { SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, MNODE);
dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt); if (pWrapper != NULL) {
return 0; if (mmGetUserAuth(pWrapper, user, spi, encrypt, secret, ckey) == 0) {
dndReleaseWrapper(pWrapper);
dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
return 0;
}
dndReleaseWrapper(pWrapper);
} }
if (terrno != TSDB_CODE_APP_NOT_READY) { if (terrno != TSDB_CODE_APP_NOT_READY) {
@ -271,7 +281,7 @@ int32_t dndInitMsgHandle(SDnode *pDnode) {
int32_t vgId = pWrapper->msgVgIds[msgIndex]; int32_t vgId = pWrapper->msgVgIds[msgIndex];
if (msgFp == NULL) continue; if (msgFp == NULL) continue;
dTrace("msg:%s will be processed by %s, vgId:%d", tMsgInfo[msgIndex], pWrapper->name, vgId); // dTrace("msg:%s will be processed by %s, vgId:%d", tMsgInfo[msgIndex], pWrapper->name, vgId);
SMsgHandle *pHandle = &pMgmt->msgHandles[msgIndex]; SMsgHandle *pHandle = &pMgmt->msgHandles[msgIndex];
if (vgId == QND_VGID) { if (vgId == QND_VGID) {
@ -328,7 +338,12 @@ int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pReq) {
SDnode *pDnode = pWrapper->pDnode; SDnode *pDnode = pWrapper->pDnode;
STransMgmt *pTrans = &pDnode->trans; STransMgmt *pTrans = &pDnode->trans;
SEpSet epSet = {0}; SEpSet epSet = {0};
dmGetMnodeEpSet(dndAcquireWrapper(pDnode, DNODE)->pMgmt, &epSet);
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, DNODE);
if (pWrapper != NULL) {
dmGetMnodeEpSet(pWrapper->pMgmt, &epSet);
dndReleaseWrapper(pWrapper);
}
return dndSendRpcReq(pTrans, &epSet, pReq); return dndSendRpcReq(pTrans, &epSet, pReq);
} }
} }
@ -336,7 +351,12 @@ int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pReq) {
void dndSendRpcRsp(SMgmtWrapper *pWrapper, SRpcMsg *pRsp) { void dndSendRpcRsp(SMgmtWrapper *pWrapper, SRpcMsg *pRsp) {
if (pRsp->code == TSDB_CODE_APP_NOT_READY) { if (pRsp->code == TSDB_CODE_APP_NOT_READY) {
SMgmtWrapper *pDnodeWrapper = dndAcquireWrapper(pWrapper->pDnode, DNODE); SMgmtWrapper *pDnodeWrapper = dndAcquireWrapper(pWrapper->pDnode, DNODE);
dmSendRedirectRsp(pDnodeWrapper->pMgmt, pRsp); if (pDnodeWrapper != NULL) {
dmSendRedirectRsp(pDnodeWrapper->pMgmt, pRsp);
dndReleaseWrapper(pDnodeWrapper);
} else {
rpcSendResponse(pRsp);
}
} else { } else {
rpcSendResponse(pRsp); rpcSendResponse(pRsp);
} }

2
source/dnode/mgmt/dnode/src/dmFile.c
View File

@ -209,7 +209,7 @@ int32_t dmWriteFile(SDnodeMgmt *pMgmt) {
} }
pMgmt->updateTime = taosGetTimestampMs(); pMgmt->updateTime = taosGetTimestampMs();
dDebug("successed to write %s", file); dDebug("successed to write %s", realfile);
return 0; return 0;
} }

4
source/dnode/mgmt/dnode/src/dmInt.c
View File

@ -74,14 +74,14 @@ void dmSendRedirectRsp(SDnodeMgmt *pMgmt, SRpcMsg *pReq) {
} }
static int32_t dmStart(SMgmtWrapper *pWrapper) { static int32_t dmStart(SMgmtWrapper *pWrapper) {
dDebug("dnode mgmt start to run"); dDebug("dnode-mgmt start to run");
return dmStartThread(pWrapper->pMgmt); return dmStartThread(pWrapper->pMgmt);
} }
int32_t dmInit(SMgmtWrapper *pWrapper) { int32_t dmInit(SMgmtWrapper *pWrapper) {
SDnode *pDnode = pWrapper->pDnode; SDnode *pDnode = pWrapper->pDnode;
SDnodeMgmt *pMgmt = calloc(1, sizeof(SDnodeMgmt)); SDnodeMgmt *pMgmt = calloc(1, sizeof(SDnodeMgmt));
dInfo("dnode-mgmt is initialized"); dInfo("dnode-mgmt start to init");
pDnode->dnodeId = 0; pDnode->dnodeId = 0;
pDnode->dropped = 0; pDnode->dropped = 0;

8
source/dnode/mgmt/dnode/src/dmMsg.c
View File

@ -41,8 +41,12 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN); memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
taosRUnLockLatch(&pMgmt->latch); taosRUnLockLatch(&pMgmt->latch);
req.pVloads = taosArrayInit(TSDB_MAX_VNODES, sizeof(SVnodeLoad)); SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, VNODES);
vmMonitorVnodeLoads(dndAcquireWrapper(pDnode, VNODES), req.pVloads); if (pWrapper != NULL) {
req.pVloads = taosArrayInit(TSDB_MAX_VNODES, sizeof(SVnodeLoad));
vmMonitorVnodeLoads(pWrapper, req.pVloads);
dndReleaseWrapper(pWrapper);
}
int32_t contLen = tSerializeSStatusReq(NULL, 0, &req); int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
void *pHead = rpcMallocCont(contLen); void *pHead = rpcMallocCont(contLen);

4
source/dnode/mgmt/dnode/src/dmWorker.c
View File

@ -114,6 +114,7 @@ int32_t dmStartWorker(SDnodeMgmt *pMgmt) {
return -1; return -1;
} }
dDebug("dnode workers are initialized");
return 0; return 0;
} }
@ -136,6 +137,7 @@ void dmStopWorker(SDnodeMgmt *pMgmt) {
taosDestoryThread(pMgmt->threadId); taosDestoryThread(pMgmt->threadId);
pMgmt->threadId = NULL; pMgmt->threadId = NULL;
} }
dDebug("dnode workers are closed");
} }
int32_t dmProcessMgmtMsg(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t dmProcessMgmtMsg(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
@ -144,6 +146,6 @@ int32_t dmProcessMgmtMsg(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
pWorker = &pMgmt->statusWorker; pWorker = &pMgmt->statusWorker;
} }
dTrace("msg:%p, will be written to worker %s", pMsg, pWorker->name); dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
return taosWriteQitem(pWorker->queue, pMsg); return taosWriteQitem(pWorker->queue, pMsg);
} }

3
source/dnode/mgmt/mnode/src/mmInt.c
View File

@ -15,6 +15,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "mmInt.h" #include "mmInt.h"
#include "wal.h"
static bool mmDeployRequired(SDnode *pDnode) { static bool mmDeployRequired(SDnode *pDnode) {
if (pDnode->dnodeId > 0) return false; if (pDnode->dnodeId > 0) return false;
@ -226,7 +227,7 @@ static int32_t mmOpen(SMgmtWrapper *pWrapper) {
} }
static int32_t mmStart(SMgmtWrapper *pWrapper) { static int32_t mmStart(SMgmtWrapper *pWrapper) {
dDebug("mnode mgmt start to run"); dDebug("mnode-mgmt start to run");
SMnodeMgmt *pMgmt = pWrapper->pMgmt; SMnodeMgmt *pMgmt = pWrapper->pMgmt;
return mndStart(pMgmt->pMnode); return mndStart(pMgmt->pMnode);
} }

2
source/dnode/mgmt/mnode/src/mmWorker.c
View File

@ -108,6 +108,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
return -1; return -1;
} }
dDebug("mnode workers are initialized");
return 0; return 0;
} }
@ -115,4 +116,5 @@ void mmStopWorker(SMnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->readWorker); tSingleWorkerCleanup(&pMgmt->readWorker);
tSingleWorkerCleanup(&pMgmt->writeWorker); tSingleWorkerCleanup(&pMgmt->writeWorker);
tSingleWorkerCleanup(&pMgmt->syncWorker); tSingleWorkerCleanup(&pMgmt->syncWorker);
dDebug("mnode workers are closed");
} }

2
source/dnode/mgmt/qnode/src/qmWorker.c
View File

@ -132,10 +132,12 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt) {
return -1; return -1;
} }
dDebug("qnode workers are initialized");
return 0; return 0;
} }
void qmStopWorker(SQnodeMgmt *pMgmt) { void qmStopWorker(SQnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->queryWorker); tSingleWorkerCleanup(&pMgmt->queryWorker);
tSingleWorkerCleanup(&pMgmt->fetchWorker); tSingleWorkerCleanup(&pMgmt->fetchWorker);
dDebug("qnode workers are closed");
} }

2
source/dnode/mgmt/snode/src/smWorker.c
View File

@ -80,6 +80,7 @@ int32_t smStartWorker(SSnodeMgmt *pMgmt) {
return -1; return -1;
} }
dDebug("snode workers are initialized");
return 0; return 0;
} }
@ -90,6 +91,7 @@ void smStopWorker(SSnodeMgmt *pMgmt) {
} }
taosArrayDestroy(pMgmt->uniqueWorkers); taosArrayDestroy(pMgmt->uniqueWorkers);
tSingleWorkerCleanup(&pMgmt->sharedWorker); tSingleWorkerCleanup(&pMgmt->sharedWorker);
dDebug("snode workers are closed");
} }
static FORCE_INLINE int32_t smGetSWIdFromMsg(SRpcMsg *pMsg) { static FORCE_INLINE int32_t smGetSWIdFromMsg(SRpcMsg *pMsg) {

8
source/dnode/mgmt/vnode/src/vmInt.c
View File

@ -257,14 +257,14 @@ static void vmCleanup(SMgmtWrapper *pWrapper) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt; SVnodesMgmt *pMgmt = pWrapper->pMgmt;
if (pMgmt == NULL) return; if (pMgmt == NULL) return;
dInfo("vnodes-mgmt start to cleanup"); dInfo("vnode-mgmt start to cleanup");
vmCloseVnodes(pMgmt); vmCloseVnodes(pMgmt);
vmStopWorker(pMgmt); vmStopWorker(pMgmt);
vnodeCleanup(); vnodeCleanup();
// walCleanUp(); // walCleanUp();
free(pMgmt); free(pMgmt);
pWrapper->pMgmt = NULL; pWrapper->pMgmt = NULL;
dInfo("vnodes-mgmt is cleaned up"); dInfo("vnode-mgmt is cleaned up");
} }
static int32_t vmInit(SMgmtWrapper *pWrapper) { static int32_t vmInit(SMgmtWrapper *pWrapper) {
@ -272,7 +272,7 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
SVnodesMgmt *pMgmt = calloc(1, sizeof(SVnodesMgmt)); SVnodesMgmt *pMgmt = calloc(1, sizeof(SVnodesMgmt));
int32_t code = -1; int32_t code = -1;
dInfo("vnodes-mgmt start to init"); dInfo("vnode-mgmt start to init");
if (pMgmt == NULL) goto _OVER; if (pMgmt == NULL) goto _OVER;
pMgmt->path = pWrapper->path; pMgmt->path = pWrapper->path;
@ -312,7 +312,7 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
} }
if (vmOpenVnodes(pMgmt) != 0) { if (vmOpenVnodes(pMgmt) != 0) {
dError("failed to open vnodes since %s", terrstr()); dError("failed to open vnode since %s", terrstr());
return -1; return -1;
} }

4
source/dnode/mgmt/vnode/src/vmWorker.c
View File

@ -356,7 +356,7 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
return -1; return -1;
} }
dDebug("vnode workers is initialized"); dDebug("vnode workers are initialized");
return 0; return 0;
} }
@ -366,5 +366,5 @@ void vmStopWorker(SVnodesMgmt *pMgmt) {
tQWorkerCleanup(&pMgmt->queryPool); tQWorkerCleanup(&pMgmt->queryPool);
tWWorkerCleanup(&pMgmt->writePool); tWWorkerCleanup(&pMgmt->writePool);
tWWorkerCleanup(&pMgmt->syncPool); tWWorkerCleanup(&pMgmt->syncPool);
dDebug("vnode workers is closed"); dDebug("vnode workers are closed");
} }

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

@ -69,7 +69,7 @@ void mndCleanupSma(SMnode *pMnode) {}
static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma) { static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
int32_t size = sizeof(SSmaObj) + pSma->exprLen + pSma->tagsFilterLen + TSDB_SMA_RESERVE_SIZE; int32_t size = sizeof(SSmaObj) + pSma->exprLen + pSma->tagsFilterLen + pSma->sqlLen + pSma->astLen + TSDB_SMA_RESERVE_SIZE;
SSdbRaw *pRaw = sdbAllocRaw(SDB_SMA, TSDB_SMA_VER_NUMBER, size); SSdbRaw *pRaw = sdbAllocRaw(SDB_SMA, TSDB_SMA_VER_NUMBER, size);
if (pRaw == NULL) goto _OVER; if (pRaw == NULL) goto _OVER;

3
source/dnode/mnode/impl/src/mndStream.c
View File

@ -269,6 +269,9 @@ static int32_t mndStreamGetPlanString(const char *ast, char **pStr) {
int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, STrans *pTrans) { int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, STrans *pTrans) {
SNode *pAst = NULL; SNode *pAst = NULL;
#if 1 // TODO: remove debug info later
printf("ast = %s\n", ast);
#endif
if (nodesStringToNode(ast, &pAst) < 0) { if (nodesStringToNode(ast, &pAst) < 0) {
return -1; return -1;
} }

47
source/libs/function/CMakeLists.txt
View File

@ -1,12 +1,55 @@
aux_source_directory(src FUNCTION_SRC) aux_source_directory(src FUNCTION_SRC)
list(REMOVE_ITEM FUNCTION_SRC src/udfd.c)
add_library(function STATIC ${FUNCTION_SRC}) add_library(function STATIC ${FUNCTION_SRC})
target_include_directories( target_include_directories(
function function
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/function" PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/function"
"${CMAKE_SOURCE_DIR}/contrib/libuv/include"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc" PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
) )
target_link_libraries( target_link_libraries(
function function
PUBLIC uv_a
PRIVATE os util common nodes PRIVATE os util common nodes
) )
add_executable(runUdf test/runUdf.c)
target_include_directories(
runUdf
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/function"
"${CMAKE_SOURCE_DIR}/contrib/libuv/include"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
runUdf
PUBLIC uv_a
PRIVATE os util common nodes function
)
add_library(udf1 MODULE test/udf1.c)
target_include_directories(
udf1
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/function"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
#SET(EXECUTABLE_OUTPUT_PATH ${CMAKE_BINARY_DIR}/build/bin)
add_executable(udfd src/udfd.c)
target_include_directories(
udfd
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/function"
"${CMAKE_SOURCE_DIR}/contrib/libuv/include"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
udfd
PUBLIC uv_a
PRIVATE os util common nodes function
)

144
source/libs/function/inc/tudf.h
View File

@ -20,68 +20,116 @@
extern "C" { extern "C" {
#endif #endif
#include "os.h" //======================================================================================
#include "taoserror.h" //begin API to taosd and qworker
/**
* start udf dameon service
* @return error code
*/
int32_t startUdfService();
/**
* stop udf dameon service
* @return error code
*/
int32_t stopUdfService();
enum { enum {
TSDB_UDF_FUNC_NORMAL = 0, TSDB_UDF_TYPE_SCALAR = 0,
TSDB_UDF_FUNC_INIT, TSDB_UDF_TYPE_AGGREGATE = 1
TSDB_UDF_FUNC_FINALIZE,
TSDB_UDF_FUNC_MERGE,
TSDB_UDF_FUNC_DESTROY,
TSDB_UDF_FUNC_MAX_NUM
}; };
typedef struct SUdfInit { enum {
int32_t maybe_null; /* 1 if function can return NULL */ TSDB_UDF_SCRIPT_BIN_LIB = 0,
uint32_t decimals; /* for real functions */ TSDB_UDF_SCRIPT_LUA = 1,
uint64_t length; /* For string functions */ };
char* ptr; /* free pointer for function data */
int32_t const_item; /* 0 if result is independent of arguments */
// script like lua/javascript
void* script_ctx;
void (*destroyCtxFunc)(void* script_ctx);
} SUdfInit;
typedef struct SUdfInfo { typedef struct SUdfInfo {
int32_t functionId; // system assigned function id char *udfName; // function name
int32_t funcType; // scalar function or aggregate function int32_t udfType; // scalar function or aggregate function
int8_t resType; // result type int8_t scriptType;
int16_t resBytes; // result byte char *path;
int32_t contLen; // content length
int32_t bufSize; // interbuf size
char* name; // function name
void* handle; // handle loaded in mem
void* funcs[TSDB_UDF_FUNC_MAX_NUM]; // function ptr
// for script like lua/javascript only int8_t resType; // result type
int isScript; int16_t resBytes; // result byte
void* pScriptCtx; int32_t bufSize; //interbuf size
SUdfInit init;
char* content;
char* path;
} SUdfInfo; } SUdfInfo;
typedef void *UdfHandle;
/**
* setup udf
* @param udf, in
* @param handle, out
* @return error code
*/
int32_t setupUdf(SUdfInfo* udf, UdfHandle *handle);
enum {
TSDB_UDF_STEP_NORMAL = 0,
TSDB_UDF_STEP_MERGE,
TSDb_UDF_STEP_FINALIZE,
TSDB_UDF_STEP_MAX_NUM
};
/**
* call udf
* @param handle udf handle
* @param step
* @param state
* @param stateSize
* @param input
* @param newstate
* @param newStateSize
* @param output
* @return error code
*/
//TODO: must change the following after metadata flow and data flow between qworker and udfd is well defined
typedef struct SUdfDataBlock {
char* data;
int32_t size;
} SUdfDataBlock;
int32_t callUdf(UdfHandle handle, int8_t step, char *state, int32_t stateSize, SUdfDataBlock input, char **newstate,
int32_t *newStateSize, SUdfDataBlock *output);
/**
* tearn down udf
* @param handle
* @return
*/
int32_t teardownUdf(UdfHandle handle);
// end API to taosd and qworker
//=============================================================================================================================
// TODO: Must change
// begin API to UDF writer.
// script // script
typedef int32_t (*scriptInitFunc)(void* pCtx); //typedef int32_t (*scriptInitFunc)(void* pCtx);
typedef void (*scriptNormalFunc)(void* pCtx, char* data, int16_t iType, int16_t iBytes, int32_t numOfRows, //typedef void (*scriptNormalFunc)(void* pCtx, char* data, int16_t iType, int16_t iBytes, int32_t numOfRows,
int64_t* ptList, int64_t key, char* dataOutput, char* tsOutput, int32_t* numOfOutput, // int64_t* ptList, int64_t key, char* dataOutput, char* tsOutput, int32_t* numOfOutput,
int16_t oType, int16_t oBytes); // int16_t oType, int16_t oBytes);
typedef void (*scriptFinalizeFunc)(void* pCtx, int64_t key, char* dataOutput, int32_t* numOfOutput); //typedef void (*scriptFinalizeFunc)(void* pCtx, int64_t key, char* dataOutput, int32_t* numOfOutput);
typedef void (*scriptMergeFunc)(void* pCtx, char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput); //typedef void (*scriptMergeFunc)(void* pCtx, char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput);
typedef void (*scriptDestroyFunc)(void* pCtx); //typedef void (*scriptDestroyFunc)(void* pCtx);
// dynamic lib // dynamic lib
typedef void (*udfNormalFunc)(char* data, int16_t itype, int16_t iBytes, int32_t numOfRows, int64_t* ts, typedef int32_t (*TUdfInitFunc)();
char* dataOutput, char* interBuf, char* tsOutput, int32_t* numOfOutput, int16_t oType, typedef void (*TUdfDestroyFunc)();
int16_t oBytes, SUdfInit* buf);
typedef int32_t (*udfInitFunc)(SUdfInit* data); typedef void (*TUdfFunc)(int8_t step,
typedef void (*udfFinalizeFunc)(char* dataOutput, char* interBuf, int32_t* numOfOutput, SUdfInit* buf); char *state, int32_t stateSize, SUdfDataBlock input,
typedef void (*udfMergeFunc)(char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput, SUdfInit* buf); char **newstate, int32_t *newStateSize, SUdfDataBlock *output);
typedef void (*udfDestroyFunc)(SUdfInit* buf);
//typedef void (*udfMergeFunc)(char *data, int32_t numOfRows, char *dataOutput, int32_t* numOfOutput);
//typedef void (*udfFinalizeFunc)(char* state, int32_t stateSize, SUdfDataBlock *output);
// end API to UDF writer
//=======================================================================================================================
#ifdef __cplusplus #ifdef __cplusplus
} }

100
source/libs/function/inc/tudfInt.h Normal file
View File

@ -0,0 +1,100 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_TUDF_INT_H
#define TDENGINE_TUDF_INT_H
#ifdef __cplusplus
extern "C" {
#endif
//TODO replaces them with fnDebug
//#define debugPrint(...) taosPrintLog("Function", DEBUG_INFO, 135, __VA_ARGS__)
#define debugPrint(...) {fprintf(stderr, __VA_ARGS__);fprintf(stderr, "\n");}
enum {
UDF_TASK_SETUP = 0,
UDF_TASK_CALL = 1,
UDF_TASK_TEARDOWN = 2
};
typedef struct SUdfSetupRequest {
char udfName[16]; //
int8_t scriptType; // 0:c, 1: lua, 2:js
int8_t udfType; //udaf, udf
int16_t pathSize;
char *path;
} SUdfSetupRequest;
typedef struct SUdfSetupResponse {
int64_t udfHandle;
} SUdfSetupResponse;
typedef struct SUdfCallRequest {
int64_t udfHandle;
int8_t step;
int32_t inputBytes;
char *input;
int32_t stateBytes;
char *state;
} SUdfCallRequest;
typedef struct SUdfCallResponse {
int32_t outputBytes;
char *output;
int32_t newStateBytes;
char *newState;
} SUdfCallResponse;
typedef struct SUdfTeardownRequest {
int64_t udfHandle;
} SUdfTeardownRequest;
typedef struct SUdfTeardownResponse {
} SUdfTeardownResponse;
typedef struct SUdfRequest {
int32_t msgLen;
int64_t seqNum;
int8_t type;
void *subReq;
} SUdfRequest;
typedef struct SUdfResponse {
int32_t msgLen;
int64_t seqNum;
int8_t type;
int32_t code;
void *subRsp;
} SUdfResponse;
int32_t decodeRequest(char *buf, int32_t bufLen, SUdfRequest **pRequest);
int32_t encodeResponse(char **buf, int32_t *bufLen, SUdfResponse *response);
int32_t encodeRequest(char **buf, int32_t *bufLen, SUdfRequest *request);
int32_t decodeResponse(char *buf, int32_t bufLen, SUdfResponse **pResponse);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_TUDF_INT_H

110
source/libs/function/inc/udfc.h Normal file
View File

@ -0,0 +1,110 @@
//
// Created by shenglian on 28/02/22.
//
#ifndef UDF_UDF_H
#define UDF_UDF_H
#include <stdlib.h>
#define DEBUG
#ifdef DEBUG
#define debugPrint(...) fprintf(__VA_ARGS__)
#else
#define debugPrint(...) /**/
#endif
enum {
UDF_TASK_SETUP = 0,
UDF_TASK_CALL = 1,
UDF_TASK_TEARDOWN = 2
};
typedef struct SSDataBlock{
char *data;
int32_t size;
} SSDataBlock;
typedef struct SUdfInfo {
char *udfName;
char *path;
} SUdfInfo;
typedef void *UdfHandle;
int32_t startUdfService();
int32_t stopUdfService();
//int32_t setupUdf(SUdfInfo *udf, int32_t numOfUdfs, UdfHandle *handles);
int32_t setupUdf(SUdfInfo* udf, UdfHandle* handle);
int32_t callUdf(UdfHandle handle, int8_t step, char *state, int32_t stateSize, SSDataBlock input, char **newstate,
int32_t *newStateSize, SSDataBlock *output);
int32_t teardownUdf(UdfHandle handle);
typedef struct SUdfSetupRequest {
char udfName[16]; //
int8_t scriptType; // 0:c, 1: lua, 2:js
int8_t udfType; //udaf, udf, udtf
int16_t pathSize;
char *path;
} SUdfSetupRequest;
typedef struct SUdfSetupResponse {
int64_t udfHandle;
} SUdfSetupResponse;
typedef struct SUdfCallRequest {
int64_t udfHandle;
int8_t step;
int32_t inputBytes;
char *input;
int32_t stateBytes;
char *state;
} SUdfCallRequest;
typedef struct SUdfCallResponse {
int32_t outputBytes;
char *output;
int32_t newStateBytes;
char *newState;
} SUdfCallResponse;
typedef struct SUdfTeardownRequest {
int64_t udfHandle;
} SUdfTeardownRequest;
typedef struct SUdfTeardownResponse {
} SUdfTeardownResponse;
typedef struct SUdfRequest {
int32_t msgLen;
int64_t seqNum;
int8_t type;
void *subReq;
} SUdfRequest;
typedef struct SUdfResponse {
int32_t msgLen;
int64_t seqNum;
int8_t type;
int32_t code;
void *subRsp;
} SUdfResponse;
int32_t decodeRequest(char *buf, int32_t bufLen, SUdfRequest **pRequest);
int32_t encodeResponse(char **buf, int32_t *bufLen, SUdfResponse *response);
int32_t encodeRequest(char **buf, int32_t *bufLen, SUdfRequest *request);
int32_t decodeResponse(char *buf, int32_t bufLen, SUdfResponse **pResponse);
#endif //UDF_UDF_H

1038
source/libs/function/src/tudf.c
View File

File diff suppressed because it is too large Load Diff

356
source/libs/function/src/udfd.c Normal file
View File

@ -0,0 +1,356 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "uv.h"
#include "os.h"
#include "tlog.h"
#include "tudf.h"
#include "tudfInt.h"
static uv_loop_t *loop;
typedef struct SUdfdUvConn {
uv_stream_t *client;
char *inputBuf;
int32_t inputLen;
int32_t inputCap;
int32_t inputTotal;
} SUdfdUvConn;
typedef struct SUvUdfWork {
uv_stream_t *client;
uv_buf_t input;
uv_buf_t output;
} SUvUdfWork;
typedef struct SUdf {
int32_t refCount;
char name[16];
int8_t type;
uv_lib_t lib;
TUdfFunc normalFunc;
} SUdf;
//TODO: low priority: change name onxxx to xxxCb, and udfc or udfd as prefix
//TODO: add private udf structure.
typedef struct SUdfHandle {
SUdf *udf;
} SUdfHandle;
void udfdProcessRequest(uv_work_t *req) {
SUvUdfWork *uvUdf = (SUvUdfWork *) (req->data);
SUdfRequest *request = NULL;
decodeRequest(uvUdf->input.base, uvUdf->input.len, &request);
switch (request->type) {
case UDF_TASK_SETUP: {
debugPrint("%s", "process setup request");
SUdf *udf = malloc(sizeof(SUdf));
udf->refCount = 0;
SUdfSetupRequest *setup = request->subReq;
strcpy(udf->name, setup->udfName);
int err = uv_dlopen(setup->path, &udf->lib);
if (err != 0) {
debugPrint("can not load library %s. error: %s", setup->path, uv_strerror(err));
//TODO set error
}
char normalFuncName[32] = {0};
strcpy(normalFuncName, setup->udfName);
//TODO error,
//TODO find all functions normal, init, destroy, normal, merge, finalize
uv_dlsym(&udf->lib, normalFuncName, (void **) (&udf->normalFunc));
SUdfHandle *handle = malloc(sizeof(SUdfHandle));
handle->udf = udf;
udf->refCount++;
//TODO: allocate private structure and call init function and set it to handle
SUdfResponse *rsp = malloc(sizeof(SUdfResponse));
rsp->seqNum = request->seqNum;
rsp->type = request->type;
rsp->code = 0;
SUdfSetupResponse *subRsp = malloc(sizeof(SUdfSetupResponse));
subRsp->udfHandle = (int64_t) (handle);
rsp->subRsp = subRsp;
char *buf;
int32_t len;
encodeResponse(&buf, &len, rsp);
uvUdf->output = uv_buf_init(buf, len);
free(rsp->subRsp);
free(rsp);
free(request->subReq);
free(request);
free(uvUdf->input.base);
break;
}
case UDF_TASK_CALL: {
debugPrint("%s", "process call request");
SUdfCallRequest *call = request->subReq;
SUdfHandle *handle = (SUdfHandle *) (call->udfHandle);
SUdf *udf = handle->udf;
char *newState;
int32_t newStateSize;
SUdfDataBlock input = {.data = call->input, .size= call->inputBytes};
SUdfDataBlock output;
//TODO: call different functions according to the step
udf->normalFunc(call->step, call->state, call->stateBytes, input, &newState, &newStateSize, &output);
SUdfResponse *rsp = malloc(sizeof(SUdfResponse));
rsp->seqNum = request->seqNum;
rsp->type = request->type;
rsp->code = 0;
SUdfCallResponse *subRsp = malloc(sizeof(SUdfCallResponse));
subRsp->outputBytes = output.size;
subRsp->output = output.data;
subRsp->newStateBytes = newStateSize;
subRsp->newState = newState;
rsp->subRsp = subRsp;
char *buf;
int32_t len;
encodeResponse(&buf, &len, rsp);
uvUdf->output = uv_buf_init(buf, len);
free(rsp->subRsp);
free(rsp);
free(newState);
free(output.data);
free(request->subReq);
free(request);
free(uvUdf->input.base);
break;
}
case UDF_TASK_TEARDOWN: {
debugPrint("%s", "process teardown request");
SUdfTeardownRequest *teardown = request->subReq;
SUdfHandle *handle = (SUdfHandle *) (teardown->udfHandle);
SUdf *udf = handle->udf;
udf->refCount--;
if (udf->refCount == 0) {
uv_dlclose(&udf->lib);
}
free(udf);
//TODO: call destroy and free udf private
free(handle);
SUdfResponse *rsp = malloc(sizeof(SUdfResponse));
rsp->seqNum = request->seqNum;
rsp->type = request->type;
rsp->code = 0;
SUdfTeardownResponse *subRsp = malloc(sizeof(SUdfTeardownResponse));
rsp->subRsp = subRsp;
char *buf;
int32_t len;
encodeResponse(&buf, &len, rsp);
uvUdf->output = uv_buf_init(buf, len);
free(rsp->subRsp);
free(rsp);
free(request->subReq);
free(request);
free(uvUdf->input.base);
break;
}
default: {
break;
}
}
}
void udfdOnWrite(uv_write_t *req, int status) {
debugPrint("%s", "after writing to pipe");
if (status < 0) {
debugPrint("Write error %s", uv_err_name(status));
}
SUvUdfWork *work = (SUvUdfWork *) req->data;
debugPrint("\tlength: %zu", work->output.len);
free(work->output.base);
free(work);
free(req);
}
void udfdSendResponse(uv_work_t *work, int status) {
debugPrint("%s", "send response");
SUvUdfWork *udfWork = (SUvUdfWork *) (work->data);
uv_write_t *write_req = malloc(sizeof(uv_write_t));
write_req->data = udfWork;
uv_write(write_req, udfWork->client, &udfWork->output, 1, udfdOnWrite);
free(work);
}
void udfdAllocBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
debugPrint("%s", "allocate buffer for read");
SUdfdUvConn *ctx = handle->data;
int32_t msgHeadSize = sizeof(int32_t) + sizeof(int64_t);
if (ctx->inputCap == 0) {
ctx->inputBuf = malloc(msgHeadSize);
if (ctx->inputBuf) {
ctx->inputLen = 0;
ctx->inputCap = msgHeadSize;
ctx->inputTotal = -1;
buf->base = ctx->inputBuf;
buf->len = ctx->inputCap;
} else {
//TODO: log error
buf->base = NULL;
buf->len = 0;
}
} else {
ctx->inputCap = ctx->inputTotal > ctx->inputCap ? ctx->inputTotal : ctx->inputCap;
void *inputBuf = realloc(ctx->inputBuf, ctx->inputCap);
if (inputBuf) {
ctx->inputBuf = inputBuf;
buf->base = ctx->inputBuf + ctx->inputLen;
buf->len = ctx->inputCap - ctx->inputLen;
} else {
//TODO: log error
buf->base = NULL;
buf->len = 0;
}
}
debugPrint("\tinput buf cap - len - total : %d - %d - %d", ctx->inputCap, ctx->inputLen, ctx->inputTotal);
}
bool isUdfdUvMsgComplete(SUdfdUvConn *pipe) {
if (pipe->inputTotal == -1 && pipe->inputLen >= sizeof(int32_t)) {
pipe->inputTotal = *(int32_t *) (pipe->inputBuf);
}
if (pipe->inputLen == pipe->inputCap && pipe->inputTotal == pipe->inputCap) {
return true;
}
return false;
}
void udfdHandleRequest(SUdfdUvConn *conn) {
uv_work_t *work = malloc(sizeof(uv_work_t));
SUvUdfWork *udfWork = malloc(sizeof(SUvUdfWork));
udfWork->client = conn->client;
udfWork->input = uv_buf_init(conn->inputBuf, conn->inputLen);
conn->inputBuf = NULL;
conn->inputLen = 0;
conn->inputCap = 0;
conn->inputTotal = -1;
work->data = udfWork;
uv_queue_work(loop, work, udfdProcessRequest, udfdSendResponse);
}
void udfdPipeCloseCb(uv_handle_t *pipe) {
SUdfdUvConn *conn = pipe->data;
free(conn->client);
free(conn->inputBuf);
free(conn);
}
void udfdUvHandleError(SUdfdUvConn *conn) {
uv_close((uv_handle_t *) conn->client, udfdPipeCloseCb);
}
void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
debugPrint("%s, nread: %zd", "read from pipe", nread);
if (nread == 0) return;
SUdfdUvConn *conn = client->data;
if (nread > 0) {
conn->inputLen += nread;
if (isUdfdUvMsgComplete(conn)) {
udfdHandleRequest(conn);
} else {
//log error or continue;
}
return;
}
if (nread < 0) {
debugPrint("Read error %s", uv_err_name(nread));
if (nread == UV_EOF) {
//TODO check more when close
} else {
}
udfdUvHandleError(conn);
}
}
void udfdOnNewConnection(uv_stream_t *server, int status) {
debugPrint("%s", "on new connection");
if (status < 0) {
// TODO
return;
}
uv_pipe_t *client = (uv_pipe_t *) malloc(sizeof(uv_pipe_t));
uv_pipe_init(loop, client, 0);
if (uv_accept(server, (uv_stream_t *) client) == 0) {
SUdfdUvConn *ctx = malloc(sizeof(SUdfdUvConn));
ctx->client = (uv_stream_t *) client;
ctx->inputBuf = 0;
ctx->inputLen = 0;
ctx->inputCap = 0;
client->data = ctx;
ctx->client = (uv_stream_t *) client;
uv_read_start((uv_stream_t *) client, udfdAllocBuffer, udfdPipeRead);
} else {
uv_close((uv_handle_t *) client, NULL);
}
}
void removeListeningPipe(int sig) {
uv_fs_t req;
uv_fs_unlink(loop, &req, "udf.sock", NULL);
exit(0);
}
int main() {
debugPrint("libuv version: %x", UV_VERSION_HEX);
loop = uv_default_loop();
uv_fs_t req;
uv_fs_unlink(loop, &req, "udf.sock", NULL);
uv_pipe_t server;
uv_pipe_init(loop, &server, 0);
signal(SIGINT, removeListeningPipe);
int r;
if ((r = uv_pipe_bind(&server, "udf.sock"))) {
debugPrint("Bind error %s\n", uv_err_name(r));
removeListeningPipe(0);
return 1;
}
if ((r = uv_listen((uv_stream_t *) &server, 128, udfdOnNewConnection))) {
debugPrint("Listen error %s", uv_err_name(r));
return 2;
}
uv_run(loop, UV_RUN_DEFAULT);
uv_loop_close(loop);
}

45
source/libs/function/test/runUdf.c Normal file
View File

@ -0,0 +1,45 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "uv.h"
#include "tudf.h"
int main(int argc, char *argv[]) {
startUdfService();
uv_sleep(1000);
char path[256] = {0};
size_t cwdSize = 256;
int err = uv_cwd(path, &cwdSize);
if (err != 0) {
fprintf(stderr, "err cwd: %s\n", uv_strerror(err));
return err;
}
fprintf(stdout, "current working directory:%s\n", path);
strcat(path, "/libudf1.so");
SUdfInfo udfInfo = {.udfName="udf1", .path=path};
UdfHandle handle;
setupUdf(&udfInfo, &handle);
//char state[5000000] = "state";
//char input[5000000] = "input";
int dataSize = 500;
int callCount = 2;
if (argc > 1) dataSize = atoi(argv[1]);
if (argc > 2) callCount = atoi(argv[2]);
char *state = malloc(dataSize);
char *input = malloc(dataSize);
SUdfDataBlock blockInput = {.data = input, .size = dataSize};
SUdfDataBlock blockOutput;
char* newState;
int32_t newStateSize;
for (int l = 0; l < callCount; ++l) {
callUdf(handle, 0, state, dataSize, blockInput, &newState, &newStateSize, &blockOutput);
}
free(state);
free(input);
teardownUdf(handle);
stopUdfService();
}

20
source/libs/function/test/udf1.c Normal file
View File

@ -0,0 +1,20 @@
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include "tudf.h"
void udf1(int8_t step, char *state, int32_t stateSize, SUdfDataBlock input,
char **newState, int32_t *newStateSize, SUdfDataBlock *output) {
fprintf(stdout, "%s, step:%d\n", "udf function called", step);
char *newStateBuf = malloc(stateSize);
memcpy(newStateBuf, state, stateSize);
*newState = newStateBuf;
*newStateSize = stateSize;
char *outputBuf = malloc(input.size);
memcpy(outputBuf, input.data, input.size);
output->data = outputBuf;
output->size = input.size;
return;
}

2
source/libs/parser/src/parTranslater.c
View File

@ -1519,7 +1519,7 @@ static int32_t translateCreateSmaIndex(STranslateContext* pCxt, SCreateIndexStmt
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
pCxt->pCmdMsg->epSet = pCxt->pParseCxt->mgmtEpSet; pCxt->pCmdMsg->epSet = pCxt->pParseCxt->mgmtEpSet;
pCxt->pCmdMsg->msgType = TDMT_VND_CREATE_SMA; pCxt->pCmdMsg->msgType = TDMT_MND_CREATE_SMA;
pCxt->pCmdMsg->msgLen = tSerializeSMCreateSmaReq(NULL, 0, &createSmaReq); pCxt->pCmdMsg->msgLen = tSerializeSMCreateSmaReq(NULL, 0, &createSmaReq);
pCxt->pCmdMsg->pMsg = malloc(pCxt->pCmdMsg->msgLen); pCxt->pCmdMsg->pMsg = malloc(pCxt->pCmdMsg->msgLen);
if (NULL == pCxt->pCmdMsg->pMsg) { if (NULL == pCxt->pCmdMsg->pMsg) {

9
source/libs/sync/inc/syncRaftEntry.h
View File

@ -27,6 +27,12 @@ extern "C" {
#include "syncMessage.h" #include "syncMessage.h"
#include "taosdef.h" #include "taosdef.h"
typedef enum EntryType {
SYNC_RAFT_ENTRY_NOOP = 0,
SYNC_RAFT_ENTRY_DATA = 1,
SYNC_RAFT_ENTRY_CONFIG = 2,
} EntryType;
typedef struct SSyncRaftEntry { typedef struct SSyncRaftEntry {
uint32_t bytes; uint32_t bytes;
uint32_t msgType; uint32_t msgType;
@ -35,12 +41,15 @@ typedef struct SSyncRaftEntry {
bool isWeak; bool isWeak;
SyncTerm term; SyncTerm term;
SyncIndex index; SyncIndex index;
EntryType entryType;
uint32_t dataLen; uint32_t dataLen;
char data[]; char data[];
} SSyncRaftEntry; } SSyncRaftEntry;
SSyncRaftEntry* syncEntryBuild(uint32_t dataLen); SSyncRaftEntry* syncEntryBuild(uint32_t dataLen);
SSyncRaftEntry* syncEntryBuild2(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index); // step 4 SSyncRaftEntry* syncEntryBuild2(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index); // step 4
SSyncRaftEntry* syncEntryBuild3(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index, EntryType entryType);
SSyncRaftEntry* syncEntryBuildNoop(SyncTerm term, SyncIndex index);
void syncEntryDestory(SSyncRaftEntry* pEntry); void syncEntryDestory(SSyncRaftEntry* pEntry);
char* syncEntrySerialize(const SSyncRaftEntry* pEntry, uint32_t* len); // step 5 char* syncEntrySerialize(const SSyncRaftEntry* pEntry, uint32_t* len); // step 5
SSyncRaftEntry* syncEntryDeserialize(const char* buf, uint32_t len); // step 6 SSyncRaftEntry* syncEntryDeserialize(const char* buf, uint32_t len); // step 6

9
source/libs/sync/src/syncAppendEntries.c
View File

@ -199,6 +199,9 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
SSyncRaftEntry* pRollBackEntry = logStoreGetEntry(ths->pLogStore, index); SSyncRaftEntry* pRollBackEntry = logStoreGetEntry(ths->pLogStore, index);
assert(pRollBackEntry != NULL); assert(pRollBackEntry != NULL);
// maybe is a NOOP ENTRY
// assert(pRollBackEntry->entryType == SYNC_RAFT_ENTRY_DATA);
SRpcMsg rpcMsg; SRpcMsg rpcMsg;
syncEntry2OriginalRpc(pRollBackEntry, &rpcMsg); syncEntry2OriginalRpc(pRollBackEntry, &rpcMsg);
ths->pFsm->FpRollBackCb(ths->pFsm, &rpcMsg, pRollBackEntry->index, pRollBackEntry->isWeak, 0, ths->state); ths->pFsm->FpRollBackCb(ths->pFsm, &rpcMsg, pRollBackEntry->index, pRollBackEntry->isWeak, 0, ths->state);
@ -217,7 +220,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
SRpcMsg rpcMsg; SRpcMsg rpcMsg;
syncEntry2OriginalRpc(pAppendEntry, &rpcMsg); syncEntry2OriginalRpc(pAppendEntry, &rpcMsg);
if (ths->pFsm != NULL) { if (ths->pFsm != NULL) {
if (ths->pFsm->FpPreCommitCb != NULL) { if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pAppendEntry->index, pAppendEntry->isWeak, 2, ths->state); ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pAppendEntry->index, pAppendEntry->isWeak, 2, ths->state);
} }
} }
@ -242,7 +245,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
SRpcMsg rpcMsg; SRpcMsg rpcMsg;
syncEntry2OriginalRpc(pAppendEntry, &rpcMsg); syncEntry2OriginalRpc(pAppendEntry, &rpcMsg);
if (ths->pFsm != NULL) { if (ths->pFsm != NULL) {
if (ths->pFsm->FpPreCommitCb != NULL) { if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pAppendEntry->index, pAppendEntry->isWeak, 3, ths->state); ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pAppendEntry->index, pAppendEntry->isWeak, 3, ths->state);
} }
} }
@ -298,7 +301,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
SRpcMsg rpcMsg; SRpcMsg rpcMsg;
syncEntry2OriginalRpc(pEntry, &rpcMsg); syncEntry2OriginalRpc(pEntry, &rpcMsg);
if (ths->pFsm->FpCommitCb != NULL) { if (ths->pFsm->FpCommitCb != NULL && pEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, ths->state); ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, ths->state);
} }

2
source/libs/sync/src/syncCommit.c
View File

@ -97,7 +97,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
SRpcMsg rpcMsg; SRpcMsg rpcMsg;
syncEntry2OriginalRpc(pEntry, &rpcMsg); syncEntry2OriginalRpc(pEntry, &rpcMsg);
if (pSyncNode->pFsm->FpCommitCb != NULL) { if (pSyncNode->pFsm->FpCommitCb != NULL && pEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, pSyncNode->state); pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, pSyncNode->state);
} }

33
source/libs/sync/src/syncIO.c
View File

@ -29,7 +29,7 @@ static int32_t syncIODestroy(SSyncIO *io);
static int32_t syncIOStartInternal(SSyncIO *io); static int32_t syncIOStartInternal(SSyncIO *io);
static int32_t syncIOStopInternal(SSyncIO *io); static int32_t syncIOStopInternal(SSyncIO *io);
static void * syncIOConsumerFunc(void *param); static void *syncIOConsumerFunc(void *param);
static void syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet); static void syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
static void syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet); static void syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
static int32_t syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey); static int32_t syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey);
@ -75,6 +75,7 @@ int32_t syncIOSendMsg(void *clientRpc, const SEpSet *pEpSet, SRpcMsg *pMsg) {
syncRpcMsgPrint2(logBuf, pMsg); syncRpcMsgPrint2(logBuf, pMsg);
pMsg->handle = NULL; pMsg->handle = NULL;
pMsg->noResp = 1;
rpcSendRequest(clientRpc, pEpSet, pMsg, NULL); rpcSendRequest(clientRpc, pEpSet, pMsg, NULL);
return ret; return ret;
} }
@ -234,9 +235,9 @@ static int32_t syncIOStopInternal(SSyncIO *io) {
} }
static void *syncIOConsumerFunc(void *param) { static void *syncIOConsumerFunc(void *param) {
SSyncIO * io = param; SSyncIO *io = param;
STaosQall *qall; STaosQall *qall;
SRpcMsg * pRpcMsg, rpcMsg; SRpcMsg *pRpcMsg, rpcMsg;
qall = taosAllocateQall(); qall = taosAllocateQall();
while (1) { while (1) {
@ -324,19 +325,21 @@ static void *syncIOConsumerFunc(void *param) {
taosGetQitem(qall, (void **)&pRpcMsg); taosGetQitem(qall, (void **)&pRpcMsg);
rpcFreeCont(pRpcMsg->pCont); rpcFreeCont(pRpcMsg->pCont);
if (pRpcMsg->handle != NULL) { /*
int msgSize = 32; if (pRpcMsg->handle != NULL) {
memset(&rpcMsg, 0, sizeof(rpcMsg)); int msgSize = 32;
rpcMsg.msgType = SYNC_RESPONSE; memset(&rpcMsg, 0, sizeof(rpcMsg));
rpcMsg.pCont = rpcMallocCont(msgSize); rpcMsg.msgType = SYNC_RESPONSE;
rpcMsg.contLen = msgSize; rpcMsg.pCont = rpcMallocCont(msgSize);
snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "give a reply"); rpcMsg.contLen = msgSize;
rpcMsg.handle = pRpcMsg->handle; snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "give a reply");
rpcMsg.code = 0; rpcMsg.handle = pRpcMsg->handle;
rpcMsg.code = 0;
syncRpcMsgPrint2((char *)"syncIOConsumerFunc rpcSendResponse --> ", &rpcMsg); syncRpcMsgPrint2((char *)"syncIOConsumerFunc rpcSendResponse --> ", &rpcMsg);
rpcSendResponse(&rpcMsg); rpcSendResponse(&rpcMsg);
} }
*/
taosFreeQitem(pRpcMsg); taosFreeQitem(pRpcMsg);
} }

59
source/libs/sync/src/syncMain.c
View File

@ -37,11 +37,13 @@ static int32_t tsNodeRefId = -1;
// ------ local funciton --------- // ------ local funciton ---------
// enqueue message ---- // enqueue message ----
static void syncNodeEqPingTimer(void* param, void* tmrId); static void syncNodeEqPingTimer(void* param, void* tmrId);
static void syncNodeEqElectTimer(void* param, void* tmrId); static void syncNodeEqElectTimer(void* param, void* tmrId);
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId); static void syncNodeEqHeartbeatTimer(void* param, void* tmrId);
static int32_t syncNodeEqNoop(SSyncNode* ths);
static int32_t syncNodeAppendNoop(SSyncNode* ths);
// on message ---- // process message ----
static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg); static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg); static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);
static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg); static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg);
@ -669,6 +671,10 @@ void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {
assert(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE); assert(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
assert(voteGrantedMajority(pSyncNode->pVotesGranted)); assert(voteGrantedMajority(pSyncNode->pVotesGranted));
syncNodeBecomeLeader(pSyncNode); syncNodeBecomeLeader(pSyncNode);
// Raft 3.6.2 Committing entries from previous terms
syncNodeAppendNoop(pSyncNode);
// syncNodeEqNoop(pSyncNode);
} }
void syncNodeFollower2Candidate(SSyncNode* pSyncNode) { void syncNodeFollower2Candidate(SSyncNode* pSyncNode) {
@ -803,6 +809,47 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
} }
} }
static int32_t syncNodeEqNoop(SSyncNode* ths) {
int32_t ret = 0;
assert(ths->state == TAOS_SYNC_STATE_LEADER);
SyncIndex index = ths->pLogStore->getLastIndex(ths->pLogStore) + 1;
SyncTerm term = ths->pRaftStore->currentTerm;
SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index);
assert(pEntry != NULL);
uint32_t entryLen;
char* serialized = syncEntrySerialize(pEntry, &entryLen);
SyncClientRequest* pSyncMsg = syncClientRequestBuild(entryLen);
assert(pSyncMsg->dataLen == entryLen);
memcpy(pSyncMsg->data, serialized, entryLen);
SRpcMsg rpcMsg;
syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
ths->FpEqMsg(ths->queue, &rpcMsg);
free(serialized);
syncClientRequestDestroy(pSyncMsg);
return ret;
}
static int32_t syncNodeAppendNoop(SSyncNode* ths) {
int32_t ret = 0;
SyncIndex index = ths->pLogStore->getLastIndex(ths->pLogStore) + 1;
SyncTerm term = ths->pRaftStore->currentTerm;
SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index);
assert(pEntry != NULL);
if (ths->state == TAOS_SYNC_STATE_LEADER) {
ths->pLogStore->appendEntry(ths->pLogStore, pEntry);
syncNodeReplicate(ths);
}
return ret;
}
// on message ---- // on message ----
static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) { static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) {
int32_t ret = 0; int32_t ret = 0;
@ -851,7 +898,7 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg
syncEntry2OriginalRpc(pEntry, &rpcMsg); syncEntry2OriginalRpc(pEntry, &rpcMsg);
if (ths->pFsm != NULL) { if (ths->pFsm != NULL) {
if (ths->pFsm->FpPreCommitCb != NULL) { if (ths->pFsm->FpPreCommitCb != NULL && pEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, ths->state); ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, ths->state);
} }
} }
@ -866,7 +913,7 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg
syncEntry2OriginalRpc(pEntry, &rpcMsg); syncEntry2OriginalRpc(pEntry, &rpcMsg);
if (ths->pFsm != NULL) { if (ths->pFsm != NULL) {
if (ths->pFsm->FpPreCommitCb != NULL) { if (ths->pFsm->FpPreCommitCb != NULL && pEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 1, ths->state); ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 1, ths->state);
} }
} }

19
source/libs/sync/src/syncRaftEntry.c
View File

@ -28,6 +28,13 @@ SSyncRaftEntry* syncEntryBuild(uint32_t dataLen) {
// step 4. SyncClientRequest => SSyncRaftEntry, add term, index // step 4. SyncClientRequest => SSyncRaftEntry, add term, index
SSyncRaftEntry* syncEntryBuild2(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index) { SSyncRaftEntry* syncEntryBuild2(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index) {
SSyncRaftEntry* pEntry = syncEntryBuild3(pMsg, term, index, SYNC_RAFT_ENTRY_DATA);
assert(pEntry != NULL);
return pEntry;
}
SSyncRaftEntry* syncEntryBuild3(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index, EntryType entryType) {
SSyncRaftEntry* pEntry = syncEntryBuild(pMsg->dataLen); SSyncRaftEntry* pEntry = syncEntryBuild(pMsg->dataLen);
assert(pEntry != NULL); assert(pEntry != NULL);
@ -37,12 +44,23 @@ SSyncRaftEntry* syncEntryBuild2(SyncClientRequest* pMsg, SyncTerm term, SyncInde
pEntry->isWeak = pMsg->isWeak; pEntry->isWeak = pMsg->isWeak;
pEntry->term = term; pEntry->term = term;
pEntry->index = index; pEntry->index = index;
pEntry->entryType = entryType;
pEntry->dataLen = pMsg->dataLen; pEntry->dataLen = pMsg->dataLen;
memcpy(pEntry->data, pMsg->data, pMsg->dataLen); memcpy(pEntry->data, pMsg->data, pMsg->dataLen);
return pEntry; return pEntry;
} }
SSyncRaftEntry* syncEntryBuildNoop(SyncTerm term, SyncIndex index) {
SSyncRaftEntry* pEntry = syncEntryBuild(0);
assert(pEntry != NULL);
pEntry->term = term;
pEntry->index = index;
pEntry->entryType = SYNC_RAFT_ENTRY_NOOP;
return pEntry;
}
void syncEntryDestory(SSyncRaftEntry* pEntry) { void syncEntryDestory(SSyncRaftEntry* pEntry) {
if (pEntry != NULL) { if (pEntry != NULL) {
free(pEntry); free(pEntry);
@ -85,6 +103,7 @@ cJSON* syncEntry2Json(const SSyncRaftEntry* pEntry) {
cJSON_AddStringToObject(pRoot, "term", u64buf); cJSON_AddStringToObject(pRoot, "term", u64buf);
snprintf(u64buf, sizeof(u64buf), "%lu", pEntry->index); snprintf(u64buf, sizeof(u64buf), "%lu", pEntry->index);
cJSON_AddStringToObject(pRoot, "index", u64buf); cJSON_AddStringToObject(pRoot, "index", u64buf);
cJSON_AddNumberToObject(pRoot, "entryType", pEntry->entryType);
cJSON_AddNumberToObject(pRoot, "dataLen", pEntry->dataLen); cJSON_AddNumberToObject(pRoot, "dataLen", pEntry->dataLen);
char* s; char* s;

22
source/libs/sync/src/syncRaftLog.c
View File

@ -34,7 +34,7 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
pLogStore->getLastTerm = logStoreLastTerm; pLogStore->getLastTerm = logStoreLastTerm;
pLogStore->updateCommitIndex = logStoreUpdateCommitIndex; pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
pLogStore->getCommitIndex = logStoreGetCommitIndex; pLogStore->getCommitIndex = logStoreGetCommitIndex;
return pLogStore; // to avoid compiler error return pLogStore;
} }
void logStoreDestory(SSyncLogStore* pLogStore) { void logStoreDestory(SSyncLogStore* pLogStore) {
@ -48,18 +48,22 @@ int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
SSyncLogStoreData* pData = pLogStore->data; SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal; SWal* pWal = pData->pWal;
assert(pEntry->index == logStoreLastIndex(pLogStore) + 1); SyncIndex lastIndex = logStoreLastIndex(pLogStore);
assert(pEntry->index == lastIndex + 1);
uint32_t len; uint32_t len;
char* serialized = syncEntrySerialize(pEntry, &len); char* serialized = syncEntrySerialize(pEntry, &len);
assert(serialized != NULL); assert(serialized != NULL);
int code; int code = 0;
code = walWrite(pWal, pEntry->index, pEntry->msgType, serialized, len); /*
assert(code == 0); code = walWrite(pWal, pEntry->index, pEntry->entryType, serialized, len);
assert(code == 0);
*/
assert(walWrite(pWal, pEntry->index, pEntry->entryType, serialized, len) == 0);
walFsync(pWal, true); walFsync(pWal, true);
free(serialized); free(serialized);
return code; // to avoid compiler error return code;
} }
SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) { SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
@ -69,7 +73,7 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
if (index >= SYNC_INDEX_BEGIN && index <= logStoreLastIndex(pLogStore)) { if (index >= SYNC_INDEX_BEGIN && index <= logStoreLastIndex(pLogStore)) {
SWalReadHandle* pWalHandle = walOpenReadHandle(pWal); SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
walReadWithHandle(pWalHandle, index); assert(walReadWithHandle(pWalHandle, index) == 0);
pEntry = syncEntryDeserialize(pWalHandle->pHead->head.body, pWalHandle->pHead->head.len); pEntry = syncEntryDeserialize(pWalHandle->pHead->head.body, pWalHandle->pHead->head.len);
assert(pEntry != NULL); assert(pEntry != NULL);
@ -83,7 +87,7 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex) { int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex) {
SSyncLogStoreData* pData = pLogStore->data; SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal; SWal* pWal = pData->pWal;
walRollback(pWal, fromIndex); assert(walRollback(pWal, fromIndex) == 0);
return 0; // to avoid compiler error return 0; // to avoid compiler error
} }
@ -107,7 +111,7 @@ SyncTerm logStoreLastTerm(SSyncLogStore* pLogStore) {
int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) { int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
SSyncLogStoreData* pData = pLogStore->data; SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal; SWal* pWal = pData->pWal;
walCommit(pWal, index); assert(walCommit(pWal, index) == 0);
return 0; // to avoid compiler error return 0; // to avoid compiler error
} }

14
source/libs/sync/test/CMakeLists.txt
View File

@ -34,6 +34,7 @@ add_executable(syncEncodeTest "")
add_executable(syncWriteTest "") add_executable(syncWriteTest "")
add_executable(syncReplicateTest "") add_executable(syncReplicateTest "")
add_executable(syncReplicateTest2 "") add_executable(syncReplicateTest2 "")
add_executable(syncReplicateTest3 "")
add_executable(syncReplicateLoadTest "") add_executable(syncReplicateLoadTest "")
add_executable(syncRefTest "") add_executable(syncRefTest "")
add_executable(syncLogStoreCheck "") add_executable(syncLogStoreCheck "")
@ -183,6 +184,10 @@ target_sources(syncReplicateTest2
PRIVATE PRIVATE
"syncReplicateTest2.cpp" "syncReplicateTest2.cpp"
) )
target_sources(syncReplicateTest3
PRIVATE
"syncReplicateTest3.cpp"
)
target_sources(syncReplicateLoadTest target_sources(syncReplicateLoadTest
PRIVATE PRIVATE
"syncReplicateLoadTest.cpp" "syncReplicateLoadTest.cpp"
@ -377,6 +382,11 @@ target_include_directories(syncReplicateTest2
"${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc" "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
) )
target_include_directories(syncReplicateTest3
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_include_directories(syncReplicateLoadTest target_include_directories(syncReplicateLoadTest
PUBLIC PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_SOURCE_DIR}/include/libs/sync"
@ -538,6 +548,10 @@ target_link_libraries(syncReplicateTest2
sync sync
gtest_main gtest_main
) )
target_link_libraries(syncReplicateTest3
sync
gtest_main
)
target_link_libraries(syncReplicateLoadTest target_link_libraries(syncReplicateLoadTest
sync sync
gtest_main gtest_main

18
source/libs/sync/test/syncEntryTest.cpp
View File

@ -46,6 +46,20 @@ void test2() {
} }
void test3() { void test3() {
SyncClientRequest* pSyncMsg = syncClientRequestBuild(10);
pSyncMsg->originalRpcType = 33;
pSyncMsg->seqNum = 11;
pSyncMsg->isWeak = 1;
strcpy(pSyncMsg->data, "test3");
SSyncRaftEntry* pEntry = syncEntryBuild3(pSyncMsg, 100, 200, SYNC_RAFT_ENTRY_NOOP);
syncEntryPrint(pEntry);
syncClientRequestDestroy(pSyncMsg);
syncEntryDestory(pEntry);
}
void test4() {
SSyncRaftEntry* pEntry = syncEntryBuild(10); SSyncRaftEntry* pEntry = syncEntryBuild(10);
assert(pEntry != NULL); assert(pEntry != NULL);
pEntry->msgType = 11; pEntry->msgType = 11;
@ -54,7 +68,8 @@ void test3() {
pEntry->isWeak = true; pEntry->isWeak = true;
pEntry->term = 44; pEntry->term = 44;
pEntry->index = 55; pEntry->index = 55;
strcpy(pEntry->data, "test3"); pEntry->entryType = SYNC_RAFT_ENTRY_CONFIG;
strcpy(pEntry->data, "test4");
syncEntryPrint(pEntry); syncEntryPrint(pEntry);
uint32_t len; uint32_t len;
@ -76,6 +91,7 @@ int main(int argc, char** argv) {
test1(); test1();
test2(); test2();
test3(); test3();
test4();
return 0; return 0;
} }

5
source/libs/sync/test/syncReplicateLoadTest.cpp
View File

@ -178,9 +178,10 @@ int main(int argc, char **argv) {
while (1) { while (1) {
sTrace( sTrace(
"replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d", "electTimerMS:%d, commitIndex:%ld",
pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm, pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS); pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
pSyncNode->commitIndex);
taosMsleep(1000); taosMsleep(1000);
} }

12
source/libs/sync/test/syncReplicateTest.cpp
View File

@ -183,18 +183,20 @@ int main(int argc, char **argv) {
taosMsleep(1000); taosMsleep(1000);
sTrace( sTrace(
"replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " "syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d", "electTimerMS:%d, commitIndex:%ld",
gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm,
gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS,
gSyncNode->commitIndex);
} }
while (1) { while (1) {
sTrace( sTrace(
"replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d", "electTimerMS:%d, commitIndex:%ld",
gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm,
gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS,
gSyncNode->commitIndex);
taosMsleep(1000); taosMsleep(1000);
} }

10
source/libs/sync/test/syncReplicateTest2.cpp
View File

@ -185,17 +185,19 @@ int main(int argc, char **argv) {
sTrace( sTrace(
"syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " "syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d", "electTimerMS:%d, commitIndex:%ld",
pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm, pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS); pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
pSyncNode->commitIndex);
} }
while (1) { while (1) {
sTrace( sTrace(
"replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d", "electTimerMS:%d, commitIndex:%ld",
pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm, pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS); pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
pSyncNode->commitIndex);
taosMsleep(1000); taosMsleep(1000);
} }

217
source/libs/sync/test/syncReplicateTest3.cpp Normal file
View File

@ -0,0 +1,217 @@
#define ALLOW_FORBID_FUNC
#include <gtest/gtest.h>
#include <stdio.h>
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncMessage.h"
#include "syncRaftEntry.h"
#include "syncRaftLog.h"
#include "syncRaftStore.h"
#include "syncUtil.h"
void logTest() {
sTrace("--- sync log test: trace");
sDebug("--- sync log test: debug");
sInfo("--- sync log test: info");
sWarn("--- sync log test: warn");
sError("--- sync log test: error");
sFatal("--- sync log test: fatal");
}
uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
int32_t replicaNum = 3;
int32_t myIndex = 0;
SRaftId ids[TSDB_MAX_REPLICA];
SSyncInfo syncInfo;
SSyncFSM *pFsm;
SWal * pWal;
void CommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code,
ESyncState state) {
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n",
pFsm, index, isWeak, code, state, syncUtilState2String(state));
syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg);
}
void PreCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code,
ESyncState state) {
char logBuf[256];
snprintf(logBuf, sizeof(logBuf),
"==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", pFsm, index, isWeak,
code, state, syncUtilState2String(state));
syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg);
}
void RollBackCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code,
ESyncState state) {
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n",
pFsm, index, isWeak, code, state, syncUtilState2String(state));
syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg);
}
void initFsm() {
pFsm = (SSyncFSM *)malloc(sizeof(SSyncFSM));
pFsm->FpCommitCb = CommitCb;
pFsm->FpPreCommitCb = PreCommitCb;
pFsm->FpRollBackCb = RollBackCb;
}
int64_t syncNodeInit() {
syncInfo.vgId = 1234;
syncInfo.rpcClient = gSyncIO->clientRpc;
syncInfo.FpSendMsg = syncIOSendMsg;
syncInfo.queue = gSyncIO->pMsgQ;
syncInfo.FpEqMsg = syncIOEqMsg;
syncInfo.pFsm = pFsm;
snprintf(syncInfo.path, sizeof(syncInfo.path), "./replicate2_test_%d", myIndex);
int code = walInit();
assert(code == 0);
SWalCfg walCfg;
memset(&walCfg, 0, sizeof(SWalCfg));
walCfg.vgId = syncInfo.vgId;
walCfg.fsyncPeriod = 1000;
walCfg.retentionPeriod = 1000;
walCfg.rollPeriod = 1000;
walCfg.retentionSize = 1000;
walCfg.segSize = 1000;
walCfg.level = TAOS_WAL_FSYNC;
char tmpdir[128];
snprintf(tmpdir, sizeof(tmpdir), "./replicate2_test_wal_%d", myIndex);
pWal = walOpen(tmpdir, &walCfg);
assert(pWal != NULL);
syncInfo.pWal = pWal;
SSyncCfg *pCfg = &syncInfo.syncCfg;
pCfg->myIndex = myIndex;
pCfg->replicaNum = replicaNum;
for (int i = 0; i < replicaNum; ++i) {
pCfg->nodeInfo[i].nodePort = ports[i];
snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
// taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
}
int64_t rid = syncStart(&syncInfo);
assert(rid > 0);
SSyncNode *pSyncNode = (SSyncNode *)syncNodeAcquire(rid);
assert(pSyncNode != NULL);
// pSyncNode->hbBaseLine = 500;
// pSyncNode->electBaseLine = 1500;
gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;
gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest;
gSyncIO->pSyncNode = pSyncNode;
syncNodeRelease(pSyncNode);
return rid;
}
void initRaftId(SSyncNode *pSyncNode) {
for (int i = 0; i < replicaNum; ++i) {
ids[i] = pSyncNode->replicasId[i];
char *s = syncUtilRaftId2Str(&ids[i]);
printf("raftId[%d] : %s\n", i, s);
free(s);
}
}
SRpcMsg *step0(int i) {
SRpcMsg *pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg));
memset(pMsg, 0, sizeof(SRpcMsg));
pMsg->msgType = 9999;
pMsg->contLen = 128;
pMsg->pCont = malloc(pMsg->contLen);
snprintf((char *)(pMsg->pCont), pMsg->contLen, "value-%u-%d", ports[myIndex], i);
return pMsg;
}
SyncClientRequest *step1(const SRpcMsg *pMsg) {
SyncClientRequest *pRetMsg = syncClientRequestBuild2(pMsg, 123, true);
return pRetMsg;
}
int main(int argc, char **argv) {
// taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog = 0;
sDebugFlag = 143 + 64;
void logTest();
myIndex = 0;
if (argc >= 2) {
myIndex = atoi(argv[1]);
}
int recordCount = 100;
if (argc >= 3) {
recordCount = atoi(argv[2]);
}
int sleepMS = 10;
if (argc >= 4) {
sleepMS = atoi(argv[3]);
}
int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]);
assert(ret == 0);
initFsm();
ret = syncInit();
assert(ret == 0);
int64_t rid = syncNodeInit();
assert(rid > 0);
SSyncNode *pSyncNode = (SSyncNode *)syncNodeAcquire(rid);
assert(pSyncNode != NULL);
syncNodePrint2((char *)"", pSyncNode);
initRaftId(pSyncNode);
for (int i = 0; i < recordCount; ++i) {
// step0
SRpcMsg *pMsg0 = step0(i);
syncRpcMsgPrint2((char *)"==step0==", pMsg0);
syncPropose(rid, pMsg0, true);
taosMsleep(sleepMS);
free(pMsg0);
sTrace(
"syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d, commitIndex:%ld",
pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
pSyncNode->commitIndex);
}
while (1) {
sTrace(
"replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d, commitIndex:%ld",
pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
pSyncNode->commitIndex);
taosMsleep(1000);
}
return 0;
}