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/warning

This commit is contained in:
Shengliang Guan 2022-01-22 13:09:24 +08:00
parent 3ba3187912 963ef70e0e
commit 1c00fd6b5c
14 changed files with 230 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/common/tmsg.h
View File

@ -696,13 +696,13 @@ typedef struct SVgroupInfo {
uint32_t hashEnd; uint32_t hashEnd;
int8_t inUse; int8_t inUse;
int8_t numOfEps; int8_t numOfEps;
SEpAddr epAddr[TSDB_MAX_REPLICA]; SEpAddr epAddr[TSDB_MAX_REPLICA];
} SVgroupInfo; } SVgroupInfo;
typedef struct { typedef struct {
int32_t vgId; int32_t vgId;
int8_t numOfEps; int8_t numOfEps;
SEpAddr epAddr[TSDB_MAX_REPLICA]; SEpAddr epAddr[TSDB_MAX_REPLICA];
} SVgroupMsg; } SVgroupMsg;
typedef struct { typedef struct {

2
include/libs/planner/planner.h
View File

@ -178,7 +178,7 @@ struct SQueryNode;
* @param requestId * @param requestId
* @return * @return
*/ */
int32_t qCreateQueryDag(const struct SQueryNode* pQueryInfo, struct SQueryDag** pDag, uint64_t requestId); int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, SSchema** pResSchema, int32_t* numOfCols, SArray* pNodeList, uint64_t requestId);
// Set datasource of this subplan, multiple calls may be made to a subplan. // Set datasource of this subplan, multiple calls may be made to a subplan.
// @subplan subplan to be schedule // @subplan subplan to be schedule

1
source/client/inc/clientInt.h
View File

@ -136,6 +136,7 @@ typedef struct SReqResultInfo {
TAOS_ROW row; TAOS_ROW row;
char **pCol; char **pCol;
uint32_t numOfRows; uint32_t numOfRows;
uint64_t totalRows;
uint32_t current; uint32_t current;
bool completed; bool completed;
} SReqResultInfo; } SReqResultInfo;

66
source/client/src/clientImpl.c
View File

@ -25,9 +25,9 @@
static int32_t initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSet); static int32_t initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSet);
static SMsgSendInfo* buildConnectMsg(SRequestObj *pRequest); static SMsgSendInfo* buildConnectMsg(SRequestObj *pRequest);
static void destroySendMsgInfo(SMsgSendInfo* pMsgBody); static void destroySendMsgInfo(SMsgSendInfo* pMsgBody);
static void setQueryResultByRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp); static void setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp);
static bool stringLengthCheck(const char* str, size_t maxsize) { static bool stringLengthCheck(const char* str, size_t maxsize) {
if (str == NULL) { if (str == NULL) {
return false; return false;
} }
@ -59,7 +59,7 @@ static char* getClusterKey(const char* user, const char* auth, const char* ip, i
} }
static STscObj* taosConnectImpl(const char *user, const char *auth, const char *db, uint16_t port, __taos_async_fn_t fp, void *param, SAppInstInfo* pAppInfo); static STscObj* taosConnectImpl(const char *user, const char *auth, const char *db, uint16_t port, __taos_async_fn_t fp, void *param, SAppInstInfo* pAppInfo);
static void setResSchemaInfo(SReqResultInfo* pResInfo, const SDataBlockSchema* pDataBlockSchema); static void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t numOfCols);
TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass, const char *auth, const char *db, uint16_t port) { TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass, const char *auth, const char *db, uint16_t port) {
if (taos_init() != TSDB_CODE_SUCCESS) { if (taos_init() != TSDB_CODE_SUCCESS) {
@ -202,43 +202,38 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQueryNode, SQueryDag** pDag) { int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQueryNode, SQueryDag** pDag, SArray* pNodeList) {
pRequest->type = pQueryNode->type; pRequest->type = pQueryNode->type;
SReqResultInfo* pResInfo = &pRequest->body.resInfo; SSchema* pSchema = NULL;
int32_t code = qCreateQueryDag(pQueryNode, pDag, pRequest->requestId); int32_t numOfCols = 0;
int32_t code = qCreateQueryDag(pQueryNode, pDag, &pSchema, &numOfCols, pNodeList, pRequest->requestId);
if (code != 0) { if (code != 0) {
return code; return code;
} }
if (pQueryNode->type == TSDB_SQL_SELECT) { if (pQueryNode->type == TSDB_SQL_SELECT) {
SArray* pa = taosArrayGetP((*pDag)->pSubplans, 0); setResSchemaInfo(&pRequest->body.resInfo, pSchema, numOfCols);
SSubplan* pPlan = taosArrayGetP(pa, 0);
SDataBlockSchema* pDataBlockSchema = &(pPlan->pDataSink->schema);
setResSchemaInfo(pResInfo, pDataBlockSchema);
pRequest->type = TDMT_VND_QUERY; pRequest->type = TDMT_VND_QUERY;
} }
return code; return code;
} }
void setResSchemaInfo(SReqResultInfo* pResInfo, const SDataBlockSchema* pDataBlockSchema) { void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t numOfCols) {
assert(pDataBlockSchema != NULL && pDataBlockSchema->numOfCols > 0); assert(pSchema != NULL && numOfCols > 0);
pResInfo->numOfCols = pDataBlockSchema->numOfCols; pResInfo->numOfCols = numOfCols;
pResInfo->fields = calloc(pDataBlockSchema->numOfCols, sizeof(pDataBlockSchema->pSchema[0])); pResInfo->fields = calloc(numOfCols, sizeof(pSchema[0]));
for (int32_t i = 0; i < pResInfo->numOfCols; ++i) { for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
SSchema* pSchema = &pDataBlockSchema->pSchema[i]; pResInfo->fields[i].bytes = pSchema[i].bytes;
pResInfo->fields[i].bytes = pSchema->bytes; pResInfo->fields[i].type = pSchema[i].type;
pResInfo->fields[i].type = pSchema->type; tstrncpy(pResInfo->fields[i].name, pSchema[i].name, tListLen(pResInfo->fields[i].name));
tstrncpy(pResInfo->fields[i].name, pSchema->name, tListLen(pResInfo->fields[i].name));
} }
} }
int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag) { int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, SArray* pNodeList) {
if (TSDB_SQL_INSERT == pRequest->type || TSDB_SQL_CREATE_TABLE == pRequest->type) { if (TSDB_SQL_INSERT == pRequest->type || TSDB_SQL_CREATE_TABLE == pRequest->type) {
SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf}; SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
@ -256,14 +251,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag) {
return pRequest->code; return pRequest->code;
} }
SArray *execNode = taosArrayInit(4, sizeof(SQueryNodeAddr)); return scheduleAsyncExecJob(pRequest->pTscObj->pAppInfo->pTransporter, pNodeList, pDag, &pRequest->body.pQueryJob);
SQueryNodeAddr addr = {.numOfEps = 1, .inUse = 0, .nodeId = 2};
addr.epAddr[0].port = 7100;
strcpy(addr.epAddr[0].fqdn, "localhost");
taosArrayPush(execNode, &addr);
return scheduleAsyncExecJob(pRequest->pTscObj->pAppInfo->pTransporter, execNode, pDag, &pRequest->body.pQueryJob);
} }
typedef struct tmq_t tmq_t; typedef struct tmq_t tmq_t;
@ -399,7 +387,9 @@ TAOS_RES *taos_create_topic(TAOS* taos, const char* topicName, const char* sql,
// todo check for invalid sql statement and return with error code // todo check for invalid sql statement and return with error code
CHECK_CODE_GOTO(qCreateQueryDag(pQueryNode, &pRequest->body.pDag, pRequest->requestId), _return); SSchema *schema = NULL;
int32_t numOfCols = 0;
CHECK_CODE_GOTO(qCreateQueryDag(pQueryNode, &pRequest->body.pDag, &schema, &numOfCols, NULL, pRequest->requestId), _return);
pStr = qDagToString(pRequest->body.pDag); pStr = qDagToString(pRequest->body.pDag);
if(pStr == NULL) { if(pStr == NULL) {
@ -492,6 +482,7 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
SRequestObj *pRequest = NULL; SRequestObj *pRequest = NULL;
SQueryNode *pQueryNode = NULL; SQueryNode *pQueryNode = NULL;
SArray *pNodeList = taosArrayInit(4, sizeof(struct SQueryNodeAddr));
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return); CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
@ -500,8 +491,8 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
if (qIsDdlQuery(pQueryNode)) { if (qIsDdlQuery(pQueryNode)) {
CHECK_CODE_GOTO(execDdlQuery(pRequest, pQueryNode), _return); CHECK_CODE_GOTO(execDdlQuery(pRequest, pQueryNode), _return);
} else { } else {
CHECK_CODE_GOTO(getPlan(pRequest, pQueryNode, &pRequest->body.pDag), _return); CHECK_CODE_GOTO(getPlan(pRequest, pQueryNode, &pRequest->body.pDag, pNodeList), _return);
CHECK_CODE_GOTO(scheduleQuery(pRequest, pRequest->body.pDag), _return); CHECK_CODE_GOTO(scheduleQuery(pRequest, pRequest->body.pDag, pNodeList), _return);
pRequest->code = terrno; pRequest->code = terrno;
} }
@ -719,13 +710,17 @@ void* doFetchRow(SRequestObj* pRequest) {
return NULL; return NULL;
} }
int32_t code = scheduleFetchRows(pRequest->body.pQueryJob, (void **)&pRequest->body.resInfo.pData); SReqResultInfo* pResInfo = &pRequest->body.resInfo;
int32_t code = scheduleFetchRows(pRequest->body.pQueryJob, (void **)&pResInfo->pData);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code; pRequest->code = code;
return NULL; return NULL;
} }
setQueryResultByRsp(&pRequest->body.resInfo, (SRetrieveTableRsp*)pRequest->body.resInfo.pData); setQueryResultFromRsp(&pRequest->body.resInfo, (SRetrieveTableRsp*)pResInfo->pData);
tscDebug("0x%"PRIx64 " fetch results, numOfRows:%d total Rows:%"PRId64", complete:%d, reqId:0x%"PRIx64, pRequest->self, pResInfo->numOfRows,
pResInfo->totalRows, pResInfo->completed, pRequest->requestId);
if (pResultInfo->numOfRows == 0) { if (pResultInfo->numOfRows == 0) {
return NULL; return NULL;
} }
@ -855,7 +850,7 @@ void setConnectionDB(STscObj* pTscObj, const char* db) {
pthread_mutex_unlock(&pTscObj->mutex); pthread_mutex_unlock(&pTscObj->mutex);
} }
void setQueryResultByRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp) { void setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp) {
assert(pResultInfo != NULL && pRsp != NULL); assert(pResultInfo != NULL && pRsp != NULL);
pResultInfo->pRspMsg = (const char*) pRsp; pResultInfo->pRspMsg = (const char*) pRsp;
@ -864,5 +859,6 @@ void setQueryResultByRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* p
pResultInfo->current = 0; pResultInfo->current = 0;
pResultInfo->completed = (pRsp->completed == 1); pResultInfo->completed = (pRsp->completed == 1);
pResultInfo->totalRows += pResultInfo->numOfRows;
setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows); setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows);
} }

6
source/libs/executor/src/executorimpl.c
View File

@ -5408,7 +5408,7 @@ SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbReadHandle, STaskRunt
return pOperator; return pOperator;
} }
SOperatorInfo* createStreamScanOperatorInfo(void *streamReadHandle, SArray* pExprInfo, SExecTaskInfo* pTaskInfo) { SOperatorInfo* createStreamScanOperatorInfo(void *streamReadHandle, SArray* pExprInfo, uint64_t uid, SExecTaskInfo* pTaskInfo) {
SStreamBlockScanInfo* pInfo = calloc(1, sizeof(SStreamBlockScanInfo)); SStreamBlockScanInfo* pInfo = calloc(1, sizeof(SStreamBlockScanInfo));
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
@ -5428,6 +5428,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void *streamReadHandle, SArray* pExp
// set the extract column id to streamHandle // set the extract column id to streamHandle
tqReadHandleSetColIdList((STqReadHandle* )streamReadHandle, pColList); tqReadHandleSetColIdList((STqReadHandle* )streamReadHandle, pColList);
tqReadHandleSetTbUid(streamReadHandle, uid);
pInfo->readerHandle = streamReadHandle; pInfo->readerHandle = streamReadHandle;
@ -7719,7 +7720,8 @@ SOperatorInfo* doCreateOperatorTreeNode(SPhyNode* pPhyNode, SExecTaskInfo* pTask
return createExchangeOperatorInfo(pEx->pSrcEndPoints, pEx->node.pTargets, pTaskInfo); return createExchangeOperatorInfo(pEx->pSrcEndPoints, pEx->node.pTargets, pTaskInfo);
} else if (pPhyNode->info.type == OP_StreamScan) { } else if (pPhyNode->info.type == OP_StreamScan) {
size_t numOfCols = taosArrayGetSize(pPhyNode->pTargets); size_t numOfCols = taosArrayGetSize(pPhyNode->pTargets);
return createStreamScanOperatorInfo(readerHandle, pPhyNode->pTargets, pTaskInfo); SScanPhyNode* pScanPhyNode = (SScanPhyNode*)pPhyNode; // simple child table.
return createStreamScanOperatorInfo(readerHandle, pPhyNode->pTargets, pScanPhyNode->uid, pTaskInfo);
} }
} }

2
source/libs/planner/inc/plannerInt.h
View File

@ -106,7 +106,7 @@ int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str);
*/ */
int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql); int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql);
int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag, uint64_t requestId); int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag, SArray* pNodeList, uint64_t requestId);
void setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SDownstreamSource* pSource); void setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SDownstreamSource* pSource);
int32_t subPlanToString(const SSubplan *pPhyNode, char** str, int32_t* len); int32_t subPlanToString(const SSubplan *pPhyNode, char** str, int32_t* len);
int32_t stringToSubplan(const char* str, SSubplan** subplan); int32_t stringToSubplan(const char* str, SSubplan** subplan);

26
source/libs/planner/src/physicalPlan.c
View File

@ -261,14 +261,14 @@ static void vgroupMsgToEpSet(const SVgroupMsg* vg, SQueryNodeAddr* execNode) {
return; return;
} }
static uint64_t splitSubplanByTable(SPlanContext* pCxt, SQueryPlanNode* pPlanNode, SQueryTableInfo* pTable) { static uint64_t splitSubplanByTable(SPlanContext* pCxt, SQueryPlanNode* pPlanNode, SQueryTableInfo* pTableInfo) {
SVgroupsInfo* pVgroupList = pTable->pMeta->vgroupList; SVgroupsInfo* pVgroupList = pTableInfo->pMeta->vgroupList;
for (int32_t i = 0; i < pVgroupList->numOfVgroups; ++i) { for (int32_t i = 0; i < pVgroupList->numOfVgroups; ++i) {
STORE_CURRENT_SUBPLAN(pCxt); STORE_CURRENT_SUBPLAN(pCxt);
SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_SCAN); SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_SCAN);
subplan->msgType = TDMT_VND_QUERY; subplan->msgType = TDMT_VND_QUERY;
vgroupMsgToEpSet(&(pTable->pMeta->vgroupList->vgroups[i]), &subplan->execNode); vgroupMsgToEpSet(&(pTableInfo->pMeta->vgroupList->vgroups[i]), &subplan->execNode);
subplan->pNode = createMultiTableScanNode(pPlanNode, pTable); subplan->pNode = createMultiTableScanNode(pPlanNode, pTableInfo);
subplan->pDataSink = createDataDispatcher(pCxt, pPlanNode, subplan->pNode); subplan->pDataSink = createDataDispatcher(pCxt, pPlanNode, subplan->pNode);
RECOVERY_CURRENT_SUBPLAN(pCxt); RECOVERY_CURRENT_SUBPLAN(pCxt);
} }
@ -384,18 +384,19 @@ static void createSubplanByLevel(SPlanContext* pCxt, SQueryPlanNode* pRoot) {
subplan->msgType = TDMT_VND_QUERY; subplan->msgType = TDMT_VND_QUERY;
subplan->pNode = createPhyNode(pCxt, pRoot); subplan->pNode = createPhyNode(pCxt, pRoot);
subplan->pDataSink = createDataDispatcher(pCxt, pRoot, subplan->pNode); subplan->pDataSink = createDataDispatcher(pCxt, pRoot, subplan->pNode);
} }
// todo deal subquery // todo deal subquery
} }
int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag, uint64_t requestId) { int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag, SArray* pNodeList, uint64_t requestId) {
TRY(TSDB_MAX_TAG_CONDITIONS) { TRY(TSDB_MAX_TAG_CONDITIONS) {
SPlanContext context = { SPlanContext context = {
.pCatalog = pCatalog, .pCatalog = pCatalog,
.pDag = validPointer(calloc(1, sizeof(SQueryDag))), .pDag = validPointer(calloc(1, sizeof(SQueryDag))),
.pCurrentSubplan = NULL, .pCurrentSubplan = NULL,
.nextId = {.queryId = requestId}, //The unsigned Id starting from 1 would be better
.nextId = {.queryId = requestId, .subplanId = 1, .templateId = 1},
}; };
*pDag = context.pDag; *pDag = context.pDag;
@ -408,6 +409,17 @@ int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryD
terrno = code; terrno = code;
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} END_TRY } END_TRY
// traverse the dag again to acquire the execution node.
if (pNodeList != NULL) {
SArray** pSubLevel = taosArrayGetLast((*pDag)->pSubplans);
size_t num = taosArrayGetSize(*pSubLevel);
for (int32_t j = 0; j < num; ++j) {
SSubplan* pPlan = taosArrayGetP(*pSubLevel, j);
taosArrayPush(pNodeList, &pPlan->execNode);
}
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

20
source/libs/planner/src/planner.c
View File

@ -16,6 +16,8 @@
#include "parser.h" #include "parser.h"
#include "plannerInt.h" #include "plannerInt.h"
static void extractResSchema(struct SQueryDag* const* pDag, SSchema** pResSchema, int32_t* numOfCols);
static void destroyDataSinkNode(SDataSink* pSinkNode) { static void destroyDataSinkNode(SDataSink* pSinkNode) {
if (pSinkNode == NULL) { if (pSinkNode == NULL) {
return; return;
@ -56,7 +58,7 @@ void qDestroyQueryDag(struct SQueryDag* pDag) {
tfree(pDag); tfree(pDag);
} }
int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, uint64_t requestId) { int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, SSchema** pResSchema, int32_t* numOfCols, SArray* pNodeList, uint64_t requestId) {
SQueryPlanNode* pLogicPlan; SQueryPlanNode* pLogicPlan;
int32_t code = createQueryPlan(pNode, &pLogicPlan); int32_t code = createQueryPlan(pNode, &pLogicPlan);
if (TSDB_CODE_SUCCESS != code) { if (TSDB_CODE_SUCCESS != code) {
@ -76,17 +78,31 @@ int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag,
return code; return code;
} }
code = createDag(pLogicPlan, NULL, pDag, requestId); code = createDag(pLogicPlan, NULL, pDag, pNodeList, requestId);
if (TSDB_CODE_SUCCESS != code) { if (TSDB_CODE_SUCCESS != code) {
destroyQueryPlan(pLogicPlan); destroyQueryPlan(pLogicPlan);
qDestroyQueryDag(*pDag); qDestroyQueryDag(*pDag);
return code; return code;
} }
extractResSchema(pDag, pResSchema, numOfCols);
destroyQueryPlan(pLogicPlan); destroyQueryPlan(pLogicPlan);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void extractResSchema(struct SQueryDag* const* pDag, SSchema** pResSchema,
int32_t* numOfCols) { // extract the final result schema
SArray* pTopSubplan = taosArrayGetP((*pDag)->pSubplans, 0);
SSubplan* pPlan = taosArrayGetP(pTopSubplan, 0);
SDataBlockSchema* pDataBlockSchema = &(pPlan->pDataSink->schema);
*numOfCols = pDataBlockSchema->numOfCols;
*pResSchema = calloc(pDataBlockSchema->numOfCols, sizeof(SSchema));
memcpy((*pResSchema), pDataBlockSchema->pSchema, pDataBlockSchema->numOfCols * sizeof(SSchema));
}
void qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SDownstreamSource* pSource) { void qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SDownstreamSource* pSource) {
setSubplanExecutionNode(subplan, templateId, pSource); setSubplanExecutionNode(subplan, templateId, pSource);
} }

6
source/libs/planner/test/phyPlanTests.cpp
View File

@ -61,7 +61,7 @@ protected:
int32_t run() { int32_t run() {
SQueryDag* dag = nullptr; SQueryDag* dag = nullptr;
uint64_t requestId = 20; uint64_t requestId = 20;
int32_t code = createDag(logicPlan_.get(), nullptr, &dag, requestId); int32_t code = createDag(logicPlan_.get(), nullptr, &dag, NULL, requestId);
dag_.reset(dag); dag_.reset(dag);
return code; return code;
} }
@ -78,9 +78,9 @@ protected:
SQueryDag* dag = nullptr; SQueryDag* dag = nullptr;
uint64_t requestId = 20; uint64_t requestId = 20;
SSchema *schema = NULL; SSchema *schema = NULL;
uint32_t numOfOutput = 0; int32_t numOfOutput = 0;
code = qCreateQueryDag(query, &dag, requestId); code = qCreateQueryDag(query, &dag, &schema, &numOfOutput, nullptr, requestId);
dag_.reset(dag); dag_.reset(dag);
return code; return code;
} }

18
source/libs/qworker/inc/qworkerInt.h
View File

@ -171,17 +171,17 @@ typedef struct SQWorkerMgmt {
#define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__) #define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__)
#define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__) #define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__)
#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__) #define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__) #define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__) #define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_ELOG_E(param) qError("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId) #define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId) #define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId) #define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
#define QW_SCH_TASK_ELOG(param, ...) qError("QW:%p SID:%"PRIx64",QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__) #define QW_SCH_TASK_ELOG(param, ...) qError("QW:%p SID:%"PRIx64",QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_WLOG(param, ...) qWarn("QW:%p SID:%"PRIx64",QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__) #define QW_SCH_TASK_WLOG(param, ...) qWarn("QW:%p SID:%"PRIx64",QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64",QID:%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__) #define QW_SCH_TASK_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64",QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_LOCK_DEBUG(...) do { if (gQWDebug.lockDebug) { qDebug(__VA_ARGS__); } } while (0) #define QW_LOCK_DEBUG(...) do { if (gQWDebug.lockDebug) { qDebug(__VA_ARGS__); } } while (0)

10
source/libs/scheduler/inc/schedulerInt.h
View File

@ -146,15 +146,15 @@ typedef struct SSchJob {
#define SCH_SET_JOB_TYPE(pAttr, type) (pAttr)->queryJob = ((type) != QUERY_TYPE_MODIFY) #define SCH_SET_JOB_TYPE(pAttr, type) (pAttr)->queryJob = ((type) != QUERY_TYPE_MODIFY)
#define SCH_JOB_NEED_FETCH(pAttr) ((pAttr)->queryJob) #define SCH_JOB_NEED_FETCH(pAttr) ((pAttr)->queryJob)
#define SCH_JOB_ELOG(param, ...) qError("QID:%" PRIx64 " " param, pJob->queryId, __VA_ARGS__) #define SCH_JOB_ELOG(param, ...) qError("QID:0x%" PRIx64 " " param, pJob->queryId, __VA_ARGS__)
#define SCH_JOB_DLOG(param, ...) qDebug("QID:%" PRIx64 " " param, pJob->queryId, __VA_ARGS__) #define SCH_JOB_DLOG(param, ...) qDebug("QID:0x%" PRIx64 " " param, pJob->queryId, __VA_ARGS__)
#define SCH_TASK_ELOG(param, ...) \ #define SCH_TASK_ELOG(param, ...) \
qError("QID:%" PRIx64 ",TID:%" PRId64 " " param, pJob->queryId, pTask->taskId, __VA_ARGS__) qError("QID:0x%" PRIx64 ",TID:%" PRId64 " " param, pJob->queryId, pTask->taskId, __VA_ARGS__)
#define SCH_TASK_DLOG(param, ...) \ #define SCH_TASK_DLOG(param, ...) \
qDebug("QID:%" PRIx64 ",TID:%" PRId64 " " param, pJob->queryId, pTask->taskId, __VA_ARGS__) qDebug("QID:0x%" PRIx64 ",TID:%" PRId64 " " param, pJob->queryId, pTask->taskId, __VA_ARGS__)
#define SCH_TASK_WLOG(param, ...) \ #define SCH_TASK_WLOG(param, ...) \
qWarn("QID:%" PRIx64 ",TID:%" PRId64 " " param, pJob->queryId, pTask->taskId, __VA_ARGS__) qWarn("QID:0x%" PRIx64 ",TID:%" PRId64 " " param, pJob->queryId, pTask->taskId, __VA_ARGS__)
#define SCH_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0) #define SCH_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define SCH_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0) #define SCH_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)

15
source/libs/scheduler/src/scheduler.c
View File

@ -650,15 +650,11 @@ _return:
SCH_RET(code); SCH_RET(code);
} }
int32_t schProcessOnDataFetched(SSchJob *job) { int32_t schProcessOnDataFetched(SSchJob *job) {
atomic_val_compare_exchange_32(&job->remoteFetch, 1, 0); atomic_val_compare_exchange_32(&job->remoteFetch, 1, 0);
tsem_post(&job->rspSem); tsem_post(&job->rspSem);
} }
// Note: no more error processing, handled in function internal // Note: no more error processing, handled in function internal
int32_t schProcessOnTaskFailure(SSchJob *pJob, SSchTask *pTask, int32_t errCode) { int32_t schProcessOnTaskFailure(SSchJob *pJob, SSchTask *pTask, int32_t errCode) {
bool needRetry = false; bool needRetry = false;
@ -882,7 +878,6 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
} }
SCH_ERR_JRET(schProcessOnDataFetched(pJob)); SCH_ERR_JRET(schProcessOnDataFetched(pJob));
break; break;
} }
case TDMT_VND_DROP_TASK: { case TDMT_VND_DROP_TASK: {
@ -892,7 +887,6 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
} }
default: default:
SCH_TASK_ELOG("unknown rsp msg, type:%d, status:%d", msgType, SCH_GET_TASK_STATUS(pTask)); SCH_TASK_ELOG("unknown rsp msg, type:%d, status:%d", msgType, SCH_GET_TASK_STATUS(pTask));
SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT); SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
} }
@ -935,8 +929,7 @@ int32_t schHandleCallback(void* param, const SDataBuf* pMsg, int32_t msgType, in
} }
pTask = *task; pTask = *task;
SCH_TASK_DLOG("rsp msg received, type:%s, code:%s", TMSG_INFO(msgType), tstrerror(rspCode));
SCH_TASK_DLOG("rsp msg received, type:%d, code:%x", msgType, rspCode);
SCH_ERR_JRET(schHandleResponseMsg(pJob, pTask, msgType, pMsg->pData, pMsg->len, rspCode)); SCH_ERR_JRET(schHandleResponseMsg(pJob, pTask, msgType, pMsg->pData, pMsg->len, rspCode));
@ -1562,8 +1555,8 @@ int32_t scheduleFetchRows(SSchJob *pJob, void** pData) {
if (NULL == pJob || NULL == pData) { if (NULL == pJob || NULL == pData) {
SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
} }
int32_t code = 0;
int32_t code = 0;
atomic_add_fetch_32(&pJob->ref, 1); atomic_add_fetch_32(&pJob->ref, 1);
int8_t status = SCH_GET_JOB_STATUS(pJob); int8_t status = SCH_GET_JOB_STATUS(pJob);
@ -1609,7 +1602,6 @@ _return:
while (true) { while (true) {
*pData = atomic_load_ptr(&pJob->res); *pData = atomic_load_ptr(&pJob->res);
if (*pData != atomic_val_compare_exchange_ptr(&pJob->res, *pData, NULL)) { if (*pData != atomic_val_compare_exchange_ptr(&pJob->res, *pData, NULL)) {
continue; continue;
} }
@ -1628,8 +1620,7 @@ _return:
atomic_val_compare_exchange_8(&pJob->userFetch, 1, 0); atomic_val_compare_exchange_8(&pJob->userFetch, 1, 0);
SCH_JOB_DLOG("fetch done, code:%x", code); SCH_JOB_DLOG("fetch done, code:%s", tstrerror(code));
atomic_sub_fetch_32(&pJob->ref, 1); atomic_sub_fetch_32(&pJob->ref, 1);
SCH_RET(code); SCH_RET(code);

131
source/libs/transport/src/transCli.c
View File

@ -17,6 +17,8 @@
#include "transComm.h" #include "transComm.h"
#define CONN_PERSIST_TIME(para) (para * 1000 * 100)
typedef struct SCliConn { typedef struct SCliConn {
uv_connect_t connReq; uv_connect_t connReq;
uv_stream_t* stream; uv_stream_t* stream;
@ -42,7 +44,7 @@ typedef struct SCliThrdObj {
uv_loop_t* loop; uv_loop_t* loop;
uv_async_t* cliAsync; // uv_async_t* cliAsync; //
uv_timer_t* pTimer; uv_timer_t* pTimer;
void* cache; // conn pool void* pool; // conn pool
queue msg; queue msg;
pthread_mutex_t msgMtx; pthread_mutex_t msgMtx;
uint64_t nextTimeout; // next timeout uint64_t nextTimeout; // next timeout
@ -63,10 +65,10 @@ typedef struct SConnList {
// conn pool // conn pool
// add expire timeout and capacity limit // add expire timeout and capacity limit
static void* connCacheCreate(int size); static void* connPoolCreate(int size);
static void* connCacheDestroy(void* cache); static void* connPoolDestroy(void* pool);
static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port); static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port);
static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn); static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn);
// register timer in each thread to clear expire conn // register timer in each thread to clear expire conn
static void clientTimeoutCb(uv_timer_t* handle); static void clientTimeoutCb(uv_timer_t* handle);
@ -88,8 +90,14 @@ static void clientConnDestroy(SCliConn* pConn);
static void clientMsgDestroy(SCliMsg* pMsg); static void clientMsgDestroy(SCliMsg* pMsg);
// thread obj
static SCliThrdObj* createThrdObj();
static void destroyThrdObj(SCliThrdObj* pThrd);
// thread
static void* clientThread(void* arg); static void* clientThread(void* arg);
static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void clientProcessData(SCliConn* conn) { static void clientProcessData(SCliConn* conn) {
STransConnCtx* pCtx = ((SCliMsg*)conn->data)->ctx; STransConnCtx* pCtx = ((SCliMsg*)conn->data)->ctx;
SRpcInfo* pRpc = pCtx->pRpc; SRpcInfo* pRpc = pCtx->pRpc;
@ -101,19 +109,22 @@ static void clientProcessData(SCliConn* conn) {
(pRpc->cfp)(NULL, &rpcMsg, NULL); (pRpc->cfp)(NULL, &rpcMsg, NULL);
SCliThrdObj* pThrd = conn->hostThrd; SCliThrdObj* pThrd = conn->hostThrd;
addConnToCache(pThrd->cache, pCtx->ip, pCtx->port, conn); addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn);
if (!uv_is_active((uv_handle_t*)pThrd->pTimer) && pRpc->idleTime > 0) {
uv_timer_start((uv_timer_t*)pThrd->pTimer, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
}
free(pCtx->ip); free(pCtx->ip);
free(pCtx); free(pCtx);
// impl // impl
} }
static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void clientTimeoutCb(uv_timer_t* handle) { static void clientTimeoutCb(uv_timer_t* handle) {
SCliThrdObj* pThrd = handle->data; SCliThrdObj* pThrd = handle->data;
SRpcInfo* pRpc = pThrd->pTransInst; SRpcInfo* pRpc = pThrd->pTransInst;
int64_t currentTime = pThrd->nextTimeout; int64_t currentTime = pThrd->nextTimeout;
tDebug("timeout, try to remove expire conn from conn pool");
SConnList* p = taosHashIterate((SHashObj*)pThrd->cache, NULL); SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL);
while (p != NULL) { while (p != NULL) {
while (!QUEUE_IS_EMPTY(&p->conn)) { while (!QUEUE_IS_EMPTY(&p->conn)) {
queue* h = QUEUE_HEAD(&p->conn); queue* h = QUEUE_HEAD(&p->conn);
@ -125,18 +136,18 @@ static void clientTimeoutCb(uv_timer_t* handle) {
break; break;
} }
} }
p = taosHashIterate((SHashObj*)pThrd->cache, p); p = taosHashIterate((SHashObj*)pThrd->pool, p);
} }
pThrd->nextTimeout = taosGetTimestampMs() + pRpc->idleTime * 1000 * 10; pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
uv_timer_start(handle, clientTimeoutCb, pRpc->idleTime * 10, 0); uv_timer_start(handle, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
} }
static void* connCacheCreate(int size) { static void* connPoolCreate(int size) {
SHashObj* cache = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK); SHashObj* pool = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
return cache; return pool;
} }
static void* connCacheDestroy(void* cache) { static void* connPoolDestroy(void* pool) {
SConnList* connList = taosHashIterate((SHashObj*)cache, NULL); SConnList* connList = taosHashIterate((SHashObj*)pool, NULL);
while (connList != NULL) { while (connList != NULL) {
while (!QUEUE_IS_EMPTY(&connList->conn)) { while (!QUEUE_IS_EMPTY(&connList->conn)) {
queue* h = QUEUE_HEAD(&connList->conn); queue* h = QUEUE_HEAD(&connList->conn);
@ -144,23 +155,22 @@ static void* connCacheDestroy(void* cache) {
SCliConn* c = QUEUE_DATA(h, SCliConn, conn); SCliConn* c = QUEUE_DATA(h, SCliConn, conn);
clientConnDestroy(c); clientConnDestroy(c);
} }
connList = taosHashIterate((SHashObj*)cache, connList); connList = taosHashIterate((SHashObj*)pool, connList);
} }
taosHashClear(cache); taosHashClear(pool);
} }
static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port) { static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
char key[128] = {0}; char key[128] = {0};
tstrncpy(key, ip, strlen(ip)); tstrncpy(key, ip, strlen(ip));
tstrncpy(key + strlen(key), (char*)(&port), sizeof(port)); tstrncpy(key + strlen(key), (char*)(&port), sizeof(port));
SHashObj* pCache = cache; SHashObj* pPool = pool;
SConnList* plist = taosHashGet(pCache, key, strlen(key)); SConnList* plist = taosHashGet(pPool, key, strlen(key));
if (plist == NULL) { if (plist == NULL) {
SConnList list; SConnList list;
plist = &list; taosHashPut(pPool, key, strlen(key), (void*)&list, sizeof(list));
taosHashPut(pCache, key, strlen(key), plist, sizeof(*plist)); plist = taosHashGet(pPool, key, strlen(key));
plist = taosHashGet(pCache, key, strlen(key));
QUEUE_INIT(&plist->conn); QUEUE_INIT(&plist->conn);
} }
@ -171,14 +181,14 @@ static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port) {
QUEUE_REMOVE(h); QUEUE_REMOVE(h);
return QUEUE_DATA(h, SCliConn, conn); return QUEUE_DATA(h, SCliConn, conn);
} }
static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn) { static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn) {
char key[128] = {0}; char key[128] = {0};
tstrncpy(key, ip, strlen(ip)); tstrncpy(key, ip, strlen(ip));
tstrncpy(key + strlen(key), (char*)(&port), sizeof(port)); tstrncpy(key + strlen(key), (char*)(&port), sizeof(port));
SRpcInfo* pRpc = ((SCliThrdObj*)conn->hostThrd)->pTransInst; SRpcInfo* pRpc = ((SCliThrdObj*)conn->hostThrd)->pTransInst;
conn->expireTime = taosGetTimestampMs() + pRpc->idleTime * 1000 * 10; conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
SConnList* plist = taosHashGet((SHashObj*)cache, key, strlen(key)); SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key));
// list already create before // list already create before
assert(plist != NULL); assert(plist != NULL);
QUEUE_PUSH(&plist->conn, &conn->conn); QUEUE_PUSH(&plist->conn, &conn->conn);
@ -322,7 +332,7 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
et = taosGetTimestampUs(); et = taosGetTimestampUs();
STransConnCtx* pCtx = pMsg->ctx; STransConnCtx* pCtx = pMsg->ctx;
SCliConn* conn = getConnFromCache(pThrd->cache, pCtx->ip, pCtx->port); SCliConn* conn = getConnFromPool(pThrd->pool, pCtx->ip, pCtx->port);
if (conn != NULL) { if (conn != NULL) {
// impl later // impl later
conn->data = pMsg; conn->data = pMsg;
@ -373,15 +383,14 @@ static void clientAsyncCb(uv_async_t* handle) {
SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q); SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q);
clientHandleReq(pMsg, pThrd); clientHandleReq(pMsg, pThrd);
count++; count++;
if (count >= 2) { }
tError("send batch size: %d", count); if (count >= 2) {
} tDebug("already process batch size: %d", count);
} }
} }
static void* clientThread(void* arg) { static void* clientThread(void* arg) {
SCliThrdObj* pThrd = (SCliThrdObj*)arg; SCliThrdObj* pThrd = (SCliThrdObj*)arg;
uv_run(pThrd->loop, UV_RUN_DEFAULT); uv_run(pThrd->loop, UV_RUN_DEFAULT);
} }
@ -394,24 +403,8 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
cli->pThreadObj = (SCliThrdObj**)calloc(cli->numOfThreads, sizeof(SCliThrdObj*)); cli->pThreadObj = (SCliThrdObj**)calloc(cli->numOfThreads, sizeof(SCliThrdObj*));
for (int i = 0; i < cli->numOfThreads; i++) { for (int i = 0; i < cli->numOfThreads; i++) {
SCliThrdObj* pThrd = (SCliThrdObj*)calloc(1, sizeof(SCliThrdObj)); SCliThrdObj* pThrd = createThrdObj();
pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
QUEUE_INIT(&pThrd->msg);
pthread_mutex_init(&pThrd->msgMtx, NULL);
pThrd->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t));
uv_loop_init(pThrd->loop);
pThrd->cliAsync = malloc(sizeof(uv_async_t));
uv_async_init(pThrd->loop, pThrd->cliAsync, clientAsyncCb);
pThrd->cliAsync->data = pThrd;
pThrd->pTimer = malloc(sizeof(uv_timer_t));
uv_timer_init(pThrd->loop, pThrd->pTimer);
pThrd->pTimer->data = pThrd;
pThrd->nextTimeout = taosGetTimestampMs() + pRpc->idleTime * 1000 * 10;
pThrd->cache = connCacheCreate(1);
pThrd->pTransInst = shandle; pThrd->pTransInst = shandle;
int err = pthread_create(&pThrd->thread, NULL, clientThread, (void*)(pThrd)); int err = pthread_create(&pThrd->thread, NULL, clientThread, (void*)(pThrd));
@ -426,16 +419,42 @@ static void clientMsgDestroy(SCliMsg* pMsg) {
// impl later // impl later
free(pMsg); free(pMsg);
} }
static SCliThrdObj* createThrdObj() {
SCliThrdObj* pThrd = (SCliThrdObj*)calloc(1, sizeof(SCliThrdObj));
QUEUE_INIT(&pThrd->msg);
pthread_mutex_init(&pThrd->msgMtx, NULL);
pThrd->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t));
uv_loop_init(pThrd->loop);
pThrd->cliAsync = malloc(sizeof(uv_async_t));
uv_async_init(pThrd->loop, pThrd->cliAsync, clientAsyncCb);
pThrd->cliAsync->data = pThrd;
pThrd->pTimer = malloc(sizeof(uv_timer_t));
uv_timer_init(pThrd->loop, pThrd->pTimer);
pThrd->pTimer->data = pThrd;
pThrd->pool = connPoolCreate(1);
return pThrd;
}
static void destroyThrdObj(SCliThrdObj* pThrd) {
if (pThrd == NULL) {
return;
}
pthread_join(pThrd->thread, NULL);
pthread_mutex_destroy(&pThrd->msgMtx);
free(pThrd->cliAsync);
free(pThrd->loop);
free(pThrd);
}
//
void taosCloseClient(void* arg) { void taosCloseClient(void* arg) {
// impl later // impl later
SClientObj* cli = arg; SClientObj* cli = arg;
for (int i = 0; i < cli->numOfThreads; i++) { for (int i = 0; i < cli->numOfThreads; i++) {
SCliThrdObj* pThrd = cli->pThreadObj[i]; destroyThrdObj(cli->pThreadObj[i]);
pthread_join(pThrd->thread, NULL);
pthread_mutex_destroy(&pThrd->msgMtx);
free(pThrd->cliAsync);
free(pThrd->loop);
free(pThrd);
} }
free(cli->pThreadObj); free(cli->pThreadObj);
free(cli); free(cli);

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

@ -91,10 +91,14 @@ static SConn* connCreate();
static void connDestroy(SConn* conn); static void connDestroy(SConn* conn);
static void uvConnDestroy(uv_handle_t* handle); static void uvConnDestroy(uv_handle_t* handle);
// server worke thread // server and worker thread
static void* workerThread(void* arg); static void* workerThread(void* arg);
static void* acceptThread(void* arg); static void* acceptThread(void* arg);
// add handle loop
static bool addHandleToWorkloop(void* arg);
static bool addHandleToAcceptloop(void* arg);
void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
/* /*
* formate of data buffer: * formate of data buffer:
@ -268,7 +272,7 @@ static void uvProcessData(SConn* pConn) {
rpcMsg.handle = pConn; rpcMsg.handle = pConn;
(*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL); (*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL);
uv_timer_start(pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0); // uv_timer_start(pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0);
// auth // auth
// validate msg type // validate msg type
} }
@ -451,22 +455,14 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
void* acceptThread(void* arg) { void* acceptThread(void* arg) {
// opt // opt
SServerObj* srv = (SServerObj*)arg; SServerObj* srv = (SServerObj*)arg;
uv_tcp_init(srv->loop, &srv->server);
struct sockaddr_in bind_addr;
uv_ip4_addr("0.0.0.0", srv->port, &bind_addr);
uv_tcp_bind(&srv->server, (const struct sockaddr*)&bind_addr, 0);
int err = 0;
if ((err = uv_listen((uv_stream_t*)&srv->server, 128, uvOnAcceptCb)) != 0) {
tError("Listen error %s\n", uv_err_name(err));
return NULL;
}
uv_run(srv->loop, UV_RUN_DEFAULT); uv_run(srv->loop, UV_RUN_DEFAULT);
} }
static void initWorkThrdObj(SWorkThrdObj* pThrd) { static bool addHandleToWorkloop(void* arg) {
SWorkThrdObj* pThrd = arg;
pThrd->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t)); pThrd->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t));
uv_loop_init(pThrd->loop); if (0 != uv_loop_init(pThrd->loop)) {
return false;
}
// SRpcInfo* pRpc = pThrd->shandle; // SRpcInfo* pRpc = pThrd->shandle;
uv_pipe_init(pThrd->loop, pThrd->pipe, 1); uv_pipe_init(pThrd->loop, pThrd->pipe, 1);
@ -482,6 +478,31 @@ static void initWorkThrdObj(SWorkThrdObj* pThrd) {
pThrd->workerAsync->data = pThrd; pThrd->workerAsync->data = pThrd;
uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb); uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
return true;
}
static bool addHandleToAcceptloop(void* arg) {
// impl later
SServerObj* srv = arg;
int err = 0;
if ((err = uv_tcp_init(srv->loop, &srv->server)) != 0) {
tError("failed to init accept server: %s", uv_err_name(err));
return false;
}
struct sockaddr_in bind_addr;
uv_ip4_addr("0.0.0.0", srv->port, &bind_addr);
if ((err = uv_tcp_bind(&srv->server, (const struct sockaddr*)&bind_addr, 0)) != 0) {
tError("failed to bind: %s", uv_err_name(err));
return false;
}
if ((err = uv_listen((uv_stream_t*)&srv->server, 128, uvOnAcceptCb)) != 0) {
tError("failed to listen: %s", uv_err_name(err));
return false;
}
return true;
} }
void* workerThread(void* arg) { void* workerThread(void* arg) {
SWorkThrdObj* pThrd = (SWorkThrdObj*)arg; SWorkThrdObj* pThrd = (SWorkThrdObj*)arg;
@ -546,11 +567,12 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
for (int i = 0; i < srv->numOfThreads; i++) { for (int i = 0; i < srv->numOfThreads; i++) {
SWorkThrdObj* thrd = (SWorkThrdObj*)calloc(1, sizeof(SWorkThrdObj)); SWorkThrdObj* thrd = (SWorkThrdObj*)calloc(1, sizeof(SWorkThrdObj));
srv->pThreadObj[i] = thrd;
srv->pipe[i] = (uv_pipe_t*)calloc(2, sizeof(uv_pipe_t)); srv->pipe[i] = (uv_pipe_t*)calloc(2, sizeof(uv_pipe_t));
int fds[2]; int fds[2];
if (uv_socketpair(AF_UNIX, SOCK_STREAM, fds, UV_NONBLOCK_PIPE, UV_NONBLOCK_PIPE) != 0) { if (uv_socketpair(AF_UNIX, SOCK_STREAM, fds, UV_NONBLOCK_PIPE, UV_NONBLOCK_PIPE) != 0) {
return NULL; goto End;
} }
uv_pipe_init(srv->loop, &(srv->pipe[i][0]), 1); uv_pipe_init(srv->loop, &(srv->pipe[i][0]), 1);
uv_pipe_open(&(srv->pipe[i][0]), fds[1]); // init write uv_pipe_open(&(srv->pipe[i][0]), fds[1]); // init write
@ -559,7 +581,9 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
thrd->fd = fds[0]; thrd->fd = fds[0];
thrd->pipe = &(srv->pipe[i][1]); // init read thrd->pipe = &(srv->pipe[i][1]); // init read
initWorkThrdObj(thrd); if (false == addHandleToWorkloop(thrd)) {
goto End;
}
int err = pthread_create(&(thrd->thread), NULL, workerThread, (void*)(thrd)); int err = pthread_create(&(thrd->thread), NULL, workerThread, (void*)(thrd));
if (err == 0) { if (err == 0) {
tDebug("sucess to create worker-thread %d", i); tDebug("sucess to create worker-thread %d", i);
@ -568,9 +592,10 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
// TODO: clear all other resource later // TODO: clear all other resource later
tError("failed to create worker-thread %d", i); tError("failed to create worker-thread %d", i);
} }
srv->pThreadObj[i] = thrd;
} }
if (false == addHandleToAcceptloop(srv)) {
goto End;
}
int err = pthread_create(&srv->thread, NULL, acceptThread, (void*)srv); int err = pthread_create(&srv->thread, NULL, acceptThread, (void*)srv);
if (err == 0) { if (err == 0) {
tDebug("success to create accept-thread"); tDebug("success to create accept-thread");
@ -579,16 +604,25 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
} }
return srv; return srv;
End:
taosCloseServer(srv);
return NULL;
}
void destroyWorkThrd(SWorkThrdObj* pThrd) {
if (pThrd == NULL) {
return;
}
pthread_join(pThrd->thread, NULL);
// free(srv->pipe[i]);
free(pThrd->loop);
free(pThrd);
} }
void taosCloseServer(void* arg) { void taosCloseServer(void* arg) {
// impl later // impl later
SServerObj* srv = arg; SServerObj* srv = arg;
for (int i = 0; i < srv->numOfThreads; i++) { for (int i = 0; i < srv->numOfThreads; i++) {
SWorkThrdObj* pThrd = srv->pThreadObj[i]; destroyWorkThrd(srv->pThreadObj[i]);
pthread_join(pThrd->thread, NULL);
free(srv->pipe[i]);
free(pThrd->loop);
free(pThrd);
} }
free(srv->loop); free(srv->loop);
free(srv->pipe); free(srv->pipe);