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

pass the unit testing

This commit is contained in:
jtao1735 2020-05-02 02:36:46 +00:00
parent cdcb0daa8a
commit 27fab2cfe6
7 changed files with 80 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/cq/CMakeLists.txt
View File

@ -4,6 +4,7 @@ PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc) INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc) INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc) INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/common/inc)
INCLUDE_DIRECTORIES(inc) INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR}/src SRC) AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR}/src SRC)

90
src/cq/src/cqMain.c
View File

@ -20,6 +20,7 @@
#include <pthread.h> #include <pthread.h>
#include "taosdef.h" #include "taosdef.h"
#include "taosmsg.h" #include "taosmsg.h"
#include "tglobal.h"
#include "tlog.h" #include "tlog.h"
#include "twal.h" #include "twal.h"
#include "tcq.h" #include "tcq.h"
@ -32,7 +33,6 @@
typedef struct { typedef struct {
int vgId; int vgId;
char path[TSDB_FILENAME_LEN];
char user[TSDB_USER_LEN]; char user[TSDB_USER_LEN];
char pass[TSDB_PASSWORD_LEN]; char pass[TSDB_PASSWORD_LEN];
FCqWrite cqWrite; FCqWrite cqWrite;
@ -44,12 +44,13 @@ typedef struct {
} SCqContext; } SCqContext;
typedef struct SCqObj { typedef struct SCqObj {
int sid; // table ID int tid; // table ID
int rowSize; // bytes of a row int rowSize; // bytes of a row
char *sqlStr; // SQL string char *sqlStr; // SQL string
int columns; // number of columns int columns; // number of columns
SSchema *pSchema; // pointer to schema array SSchema *pSchema; // pointer to schema array
void *pStream; void *pStream;
struct SCqObj *prev;
struct SCqObj *next; struct SCqObj *next;
SCqContext *pContext; SCqContext *pContext;
} SCqObj; } SCqObj;
@ -65,30 +66,10 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
strcpy(pContext->user, pCfg->user); strcpy(pContext->user, pCfg->user);
strcpy(pContext->pass, pCfg->pass); strcpy(pContext->pass, pCfg->pass);
strcpy(pContext->path, pCfg->path);
pContext->vgId = pCfg->vgId; pContext->vgId = pCfg->vgId;
pContext->cqWrite = pCfg->cqWrite; pContext->cqWrite = pCfg->cqWrite;
pContext->ahandle = ahandle; pContext->ahandle = ahandle;
// open meta data file
// loop each record
while (1) {
SCqObj *pObj = calloc(sizeof(SCqObj), 1);
if (pObj == NULL) {
cError("vgId:%d, no memory", pContext->vgId);
continue;
}
pObj->next = pContext->pHead;
pContext->pHead = pObj;
// assigne each field in SCqObj
// pObj->sid =
// strcpy(pObj->sqlStr, ?? );
// schema, columns
}
pthread_mutex_init(&pContext->mutex, NULL); pthread_mutex_init(&pContext->mutex, NULL);
cTrace("vgId:%d, CQ is opened", pContext->vgId); cTrace("vgId:%d, CQ is opened", pContext->vgId);
@ -102,8 +83,6 @@ void cqClose(void *handle) {
// stop all CQs // stop all CQs
cqStop(pContext); cqStop(pContext);
// save the meta data
// free all resources // free all resources
SCqObj *pObj = pContext->pHead; SCqObj *pObj = pContext->pHead;
while (pObj) { while (pObj) {
@ -125,23 +104,23 @@ void cqStart(void *handle) {
pthread_mutex_lock(&pContext->mutex); pthread_mutex_lock(&pContext->mutex);
tscEmbedded = 1;
pContext->dbConn = taos_connect("localhost", pContext->user, pContext->pass, NULL, 0); pContext->dbConn = taos_connect("localhost", pContext->user, pContext->pass, NULL, 0);
if (pContext->dbConn) { if (pContext->dbConn == NULL) {
cError("vgId:%d, failed to connect to TDengine(%s)", pContext->vgId, tstrerror(terrno)); cError("vgId:%d, failed to connect to TDengine(%s)", pContext->vgId, tstrerror(terrno));
pthread_mutex_unlock(&pContext->mutex); pthread_mutex_unlock(&pContext->mutex);
return; return;
} }
SCqObj *pObj = pContext->pHead; SCqObj *pObj = pContext->pHead;
while (pObj) { while (pObj) {
int64_t lastKey = 0; int64_t lastKey = 0;
pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, lastKey, pObj, NULL); pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, lastKey, pObj, NULL);
if (pObj->pStream) { if (pObj->pStream) {
pContext->num++; pContext->num++;
cTrace("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->sid, pObj->sqlStr); cTrace("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->tid, pObj->sqlStr);
} else { } else {
cError("vgId:%d, id:%d CQ:%s, failed to open", pContext->vgId, pObj->sqlStr); cError("vgId:%d, id:%d CQ:%s, failed to open", pContext->vgId, pObj->tid, pObj->sqlStr);
} }
pObj = pObj->next; pObj = pObj->next;
} }
@ -158,9 +137,11 @@ void cqStop(void *handle) {
SCqObj *pObj = pContext->pHead; SCqObj *pObj = pContext->pHead;
while (pObj) { while (pObj) {
if (pObj->pStream) taos_close_stream(pObj->pStream); if (pObj->pStream) {
taos_close_stream(pObj->pStream);
pObj->pStream = NULL; pObj->pStream = NULL;
cTrace("vgId:%d, id:%d CQ:%s is closed", pContext->vgId, pObj->sid, pObj->sqlStr); cTrace("vgId:%d, id:%d CQ:%s is closed", pContext->vgId, pObj->tid, pObj->sqlStr);
}
pObj = pObj->next; pObj = pObj->next;
} }
@ -171,13 +152,13 @@ void cqStop(void *handle) {
pthread_mutex_unlock(&pContext->mutex); pthread_mutex_unlock(&pContext->mutex);
} }
void cqCreate(void *handle, int sid, char *sqlStr, SSchema *pSchema, int columns) { void *cqCreate(void *handle, int tid, char *sqlStr, SSchema *pSchema, int columns) {
SCqContext *pContext = handle; SCqContext *pContext = handle;
SCqObj *pObj = calloc(sizeof(SCqObj), 1); SCqObj *pObj = calloc(sizeof(SCqObj), 1);
if (pObj == NULL) return; if (pObj == NULL) return NULL;
pObj->sid = sid; pObj->tid = tid;
pObj->sqlStr = malloc(strlen(sqlStr)+1); pObj->sqlStr = malloc(strlen(sqlStr)+1);
strcpy(pObj->sqlStr, sqlStr); strcpy(pObj->sqlStr, sqlStr);
@ -187,11 +168,12 @@ void cqCreate(void *handle, int sid, char *sqlStr, SSchema *pSchema, int columns
pObj->pSchema = malloc(size); pObj->pSchema = malloc(size);
memcpy(pObj->pSchema, pSchema, size); memcpy(pObj->pSchema, pSchema, size);
cTrace("vgId:%d, id:%d CQ:%s is created", pContext->vgId, pObj->sid, pObj->sqlStr); cTrace("vgId:%d, id:%d CQ:%s is created", pContext->vgId, pObj->tid, pObj->sqlStr);
pthread_mutex_lock(&pContext->mutex); pthread_mutex_lock(&pContext->mutex);
pObj->next = pContext->pHead; pObj->next = pContext->pHead;
if (pContext->pHead) pContext->pHead->prev = pObj;
pContext->pHead = pObj; pContext->pHead = pObj;
if (pContext->dbConn) { if (pContext->dbConn) {
@ -199,50 +181,39 @@ void cqCreate(void *handle, int sid, char *sqlStr, SSchema *pSchema, int columns
pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, lastKey, pObj, NULL); pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, lastKey, pObj, NULL);
if (pObj->pStream) { if (pObj->pStream) {
pContext->num++; pContext->num++;
cTrace("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->sid, pObj->sqlStr); cTrace("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->tid, pObj->sqlStr);
} else { } else {
cError("vgId:%d, id:%d CQ:%s, failed to launch", pContext->vgId, pObj->sid, pObj->sqlStr); cError("vgId:%d, id:%d CQ:%s, failed to launch", pContext->vgId, pObj->tid, pObj->sqlStr);
} }
} }
pthread_mutex_unlock(&pContext->mutex); pthread_mutex_unlock(&pContext->mutex);
return pObj;
} }
void cqDrop(void *handle, int sid) { void cqDrop(void *handle) {
SCqContext *pContext = handle; SCqObj *pObj = handle;
SCqContext *pContext = pObj->pContext;
pthread_mutex_lock(&pContext->mutex); pthread_mutex_lock(&pContext->mutex);
// locate the pObj; if (pObj->prev) {
SCqObj *prev = NULL; pObj->prev->next = pObj->next;
SCqObj *pObj = pContext->pHead;
while (pObj) {
if (pObj->sid != sid) {
prev = pObj;
pObj = pObj->next;
continue;
}
// remove from the linked list
if (prev) {
prev->next = pObj->next;
} else { } else {
pContext->pHead = pObj->next; pContext->pHead = pObj->next;
} }
break; if (pObj->next) {
pObj->next->prev = pObj->prev;
} }
if (pObj) {
// update the meta data
// free the resources associated // free the resources associated
if (pObj->pStream) taos_close_stream(pObj->pStream); if (pObj->pStream) taos_close_stream(pObj->pStream);
pObj->pStream = NULL; pObj->pStream = NULL;
cTrace("vgId:%d, id:%d CQ:%s is dropped", pContext->vgId, pObj->sid, pObj->sqlStr); cTrace("vgId:%d, id:%d CQ:%s is dropped", pContext->vgId, pObj->tid, pObj->sqlStr);
free(pObj); free(pObj);
}
pthread_mutex_lock(&pContext->mutex); pthread_mutex_lock(&pContext->mutex);
} }
@ -252,7 +223,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
SCqContext *pContext = pObj->pContext; SCqContext *pContext = pObj->pContext;
if (pObj->pStream == NULL) return; if (pObj->pStream == NULL) return;
cTrace("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->sid, pObj->sqlStr); cTrace("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr);
// construct data // construct data
int size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + pObj->rowSize; int size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + pObj->rowSize;
@ -269,11 +240,10 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg)); SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg));
// to do: fill in the SSubmitBlk strucuture // to do: fill in the SSubmitBlk strucuture
pBlk->tid = pObj->sid; pBlk->tid = pObj->tid;
// write into vnode write queue // write into vnode write queue
pContext->cqWrite(pContext->ahandle, pHead, TAOS_QTYPE_CQ); pContext->cqWrite(pContext->ahandle, pHead, TAOS_QTYPE_CQ);
} }

29
src/cq/test/cqtest.c
View File

@ -29,16 +29,16 @@ int writeToQueue(void *pVnode, void *data, int type) {
} }
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
char path[128] = "~/cq"; int num = 3;
for (int i=1; i<argc; ++i) { for (int i=1; i<argc; ++i) {
if (strcmp(argv[i], "-p")==0 && i < argc-1) { if (strcmp(argv[i], "-d")==0 && i < argc-1) {
strcpy(path, argv[++i]);
} else if (strcmp(argv[i], "-d")==0 && i < argc-1) {
ddebugFlag = atoi(argv[++i]); ddebugFlag = atoi(argv[++i]);
} else if (strcmp(argv[i], "-n") == 0 && i <argc-1) {
num = atoi(argv[++i]);
} else { } else {
printf("\nusage: %s [options] \n", argv[0]); printf("\nusage: %s [options] \n", argv[0]);
printf(" [-p path]: wal file path default is:%s\n", path); printf(" [-n num]: number of streams, default:%d\n", num);
printf(" [-d debugFlag]: debug flag, default:%d\n", ddebugFlag); printf(" [-d debugFlag]: debug flag, default:%d\n", ddebugFlag);
printf(" [-h help]: print out this help\n\n"); printf(" [-h help]: print out this help\n\n");
exit(0); exit(0);
@ -50,7 +50,6 @@ int main(int argc, char *argv[]) {
SCqCfg cqCfg; SCqCfg cqCfg;
strcpy(cqCfg.user, "root"); strcpy(cqCfg.user, "root");
strcpy(cqCfg.pass, "taosdata"); strcpy(cqCfg.pass, "taosdata");
strcpy(cqCfg.path, path);
cqCfg.vgId = 2; cqCfg.vgId = 2;
cqCfg.cqWrite = writeToQueue; cqCfg.cqWrite = writeToQueue;
@ -60,9 +59,19 @@ int main(int argc, char *argv[]) {
exit(-1); exit(-1);
} }
SSchema *pSchema = NULL; SSchema schema[2];
schema[0].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(schema[0].name, "ts");
schema[0].colId = 0;
schema[0].bytes = 8;
schema[1].type = TSDB_DATA_TYPE_INT;
strcpy(schema[1].name, "avgspeed");
schema[1].colId = 1;
schema[1].bytes = 4;
for (int sid =1; sid<10; ++sid) { for (int sid =1; sid<10; ++sid) {
cqCreate(pCq, 1, "select avg(speed) from t1 sliding(1s) interval(5s)", pSchema, 2); cqCreate(pCq, sid, "select avg(speed) from demo.t1 sliding(1s) interval(5s)", schema, 2);
} }
while (1) { while (1) {
@ -83,6 +92,8 @@ int main(int argc, char *argv[]) {
break; break;
case 'q': case 'q':
break; break;
default:
printf("invalid command:%c", c);
} }
if (c=='q') break; if (c=='q') break;
@ -90,5 +101,7 @@ int main(int argc, char *argv[]) {
cqClose(pCq); cqClose(pCq);
taosCloseLog();
return 0; return 0;
} }

14
src/inc/tcq.h
View File

@ -24,18 +24,26 @@ typedef int (*FCqWrite)(void *ahandle, void *pHead, int type);
typedef struct { typedef struct {
int vgId; int vgId;
char path[TSDB_FILENAME_LEN];
char user[TSDB_USER_LEN]; char user[TSDB_USER_LEN];
char pass[TSDB_PASSWORD_LEN]; char pass[TSDB_PASSWORD_LEN];
FCqWrite cqWrite; FCqWrite cqWrite;
} SCqCfg; } SCqCfg;
// the following API shall be called by vnode
void *cqOpen(void *ahandle, const SCqCfg *pCfg); void *cqOpen(void *ahandle, const SCqCfg *pCfg);
void cqClose(void *handle); void cqClose(void *handle);
// if vnode is master, vnode call this API to start CQ
void cqStart(void *handle); void cqStart(void *handle);
// if vnode is slave/unsynced, vnode shall call this API to stop CQ
void cqStop(void *handle); void cqStop(void *handle);
void cqCreate(void *handle, int sid, char *sqlStr, SSchema *pSchema, int columns);
void cqDrop(void *handle, int sid); // cqCreate is called by TSDB to start an instance of CQ
void *cqCreate(void *handle, int sid, char *sqlStr, SSchema *pSchema, int columns);
// cqDrop is called by TSDB to stop an instance of CQ, handle is the return value of cqCreate
void cqDrop(void *handle);
extern int cqDebugFlag; extern int cqDebugFlag;

2
src/inc/tsdb.h
View File

@ -38,9 +38,9 @@ extern "C" {
typedef struct { typedef struct {
// WAL handle // WAL handle
void *appH; void *appH;
void *cqH;
int (*walCallBack)(void *); int (*walCallBack)(void *);
int (*eventCallBack)(void *); int (*eventCallBack)(void *);
int (*cqueryCallBack)(void *);
} STsdbAppH; } STsdbAppH;
// --------- TSDB REPOSITORY CONFIGURATION DEFINITION // --------- TSDB REPOSITORY CONFIGURATION DEFINITION

13
src/vnode/src/vnodeMain.c
View File

@ -194,9 +194,16 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->wqueue = dnodeAllocateWqueue(pVnode); pVnode->wqueue = dnodeAllocateWqueue(pVnode);
pVnode->rqueue = dnodeAllocateRqueue(pVnode); pVnode->rqueue = dnodeAllocateRqueue(pVnode);
SCqCfg cqCfg;
sprintf(cqCfg.user, "root");
strcpy(cqCfg.pass, tsInternalPass);
cqCfg.cqWrite = vnodeWriteToQueue;
pVnode->cq = cqOpen(pVnode, &cqCfg);
STsdbAppH appH = {0}; STsdbAppH appH = {0};
appH.appH = (void *)pVnode; appH.appH = (void *)pVnode;
appH.walCallBack = vnodeWalCallback; appH.walCallBack = vnodeWalCallback;
appH.cqH = pVnode->cq;
sprintf(temp, "%s/tsdb", rootDir); sprintf(temp, "%s/tsdb", rootDir);
pVnode->tsdb = tsdbOpenRepo(temp, &appH); pVnode->tsdb = tsdbOpenRepo(temp, &appH);
@ -210,12 +217,6 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->wal = walOpen(temp, &pVnode->walCfg); pVnode->wal = walOpen(temp, &pVnode->walCfg);
walRestore(pVnode->wal, pVnode, vnodeWriteToQueue); walRestore(pVnode->wal, pVnode, vnodeWriteToQueue);
SCqCfg cqCfg;
sprintf(cqCfg.path, "%s/cq", rootDir);
strcpy(cqCfg.pass, tsInternalPass);
cqCfg.cqWrite = vnodeWriteToQueue;
pVnode->cq = cqOpen(pVnode, &cqCfg);
SSyncInfo syncInfo; SSyncInfo syncInfo;
syncInfo.vgId = pVnode->vgId; syncInfo.vgId = pVnode->vgId;
syncInfo.version = pVnode->version; syncInfo.version = pVnode->version;

10
src/vnode/src/vnodeWrite.c
View File

@ -114,7 +114,6 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe
int16_t numOfColumns = htons(pTable->numOfColumns); int16_t numOfColumns = htons(pTable->numOfColumns);
int16_t numOfTags = htons(pTable->numOfTags); int16_t numOfTags = htons(pTable->numOfTags);
int32_t sid = htonl(pTable->sid); int32_t sid = htonl(pTable->sid);
int32_t sqlDataLen = htonl(pTable->sqlDataLen);
uint64_t uid = htobe64(pTable->uid); uint64_t uid = htobe64(pTable->uid);
SSchema *pSchema = (SSchema *) pTable->data; SSchema *pSchema = (SSchema *) pTable->data;
@ -151,14 +150,6 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe
} }
code = tsdbCreateTable(pVnode->tsdb, &tCfg); code = tsdbCreateTable(pVnode->tsdb, &tCfg);
if (code == 0 && sqlDataLen >0) {
char *sqlStr = NULL;
// to do: get the sqlStr
cqCreate(pVnode->cq, sid, sqlStr, pSchema, numOfColumns);
}
tfree(pDestSchema); tfree(pDestSchema);
dTrace("pVnode:%p vgId:%d, table:%s is created, result:%x", pVnode, pVnode->vgId, pTable->tableId, code); dTrace("pVnode:%p vgId:%d, table:%s is created, result:%x", pVnode, pVnode->vgId, pTable->tableId, code);
@ -176,7 +167,6 @@ static int32_t vnodeProcessDropTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
}; };
code = tsdbDropTable(pVnode->tsdb, tableId); code = tsdbDropTable(pVnode->tsdb, tableId);
cqDrop(pVnode->cq, tableId.tid);
return code; return code;
} }