diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index 52fff978bc82594e39911f10c98d06a7824fb854..2ff4d58e0b77f60a51a62b759d6509b525d32bfd 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -172,7 +172,6 @@ enum { TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES, "vnode-show-tables", SVShowTablesReq, SVShowTablesRsp) TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES_FETCH, "vnode-show-tables-fetch", SVShowTablesFetchReq, SVShowTablesFetchRsp) TD_DEF_MSG_TYPE(TDMT_VND_QUERY_CONTINUE, "vnode-query-continue", NULL, NULL) - TD_DEF_MSG_TYPE(TDMT_VND_SCHEDULE_DATA_SINK, "vnode-schedule-data-sink", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp) TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp) diff --git a/include/libs/executor/dataSinkMgt.h b/include/libs/executor/dataSinkMgt.h index 371cb124056b3d7b4d2175a785f9bd139e61531e..19438b5dd4bfcdbe5997314c87f4be980ebe9534 100644 --- a/include/libs/executor/dataSinkMgt.h +++ b/include/libs/executor/dataSinkMgt.h @@ -48,7 +48,6 @@ typedef struct SOutputData { int8_t compressed; char* pData; bool queryEnd; - int32_t scheduleJobNo; int32_t bufStatus; int64_t useconds; int8_t precision; diff --git a/source/dnode/vnode/src/vnd/vnodeQuery.c b/source/dnode/vnode/src/vnd/vnodeQuery.c index a01960b77371d6226748318ce507a1a0dc08de89..dd1e5ba9ae302746d4d29f198deaf7705f694f65 100644 --- a/source/dnode/vnode/src/vnd/vnodeQuery.c +++ b/source/dnode/vnode/src/vnd/vnodeQuery.c @@ -29,8 +29,6 @@ int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { return qWorkerProcessQueryMsg(pVnode->pTsdb, pVnode->pQuery, pMsg); case TDMT_VND_QUERY_CONTINUE: return qWorkerProcessCQueryMsg(pVnode->pTsdb, pVnode->pQuery, pMsg); - case TDMT_VND_SCHEDULE_DATA_SINK: - return qWorkerProcessDataSinkMsg(pVnode->pTsdb, pVnode->pQuery, pMsg); default: vError("unknown msg type:%d in query queue", pMsg->msgType); return TSDB_CODE_VND_APP_ERROR; diff --git a/source/libs/executor/src/dataDispatcher.c b/source/libs/executor/src/dataDispatcher.c index edba4fc97dbe95aaf48f097627b6b6f2cd9b15ea..f9e61f91de8a30b60ba8131e8a29ba3dfd9e7aca 100644 --- a/source/libs/executor/src/dataDispatcher.c +++ b/source/libs/executor/src/dataDispatcher.c @@ -196,7 +196,6 @@ static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) { pOutput->bufStatus = updateStatus(pDispatcher); pthread_mutex_lock(&pDispatcher->mutex); pOutput->queryEnd = pDispatcher->queryEnd; - pOutput->scheduleJobNo = 0; pOutput->useconds = pDispatcher->useconds; pOutput->precision = pDispatcher->schema.precision; pthread_mutex_unlock(&pDispatcher->mutex); diff --git a/source/libs/qworker/inc/qworkerInt.h b/source/libs/qworker/inc/qworkerInt.h index 913057787ec30b9ab32ae36174bcd06516983f5c..7d043a0e024b09de3bbea4085f1717f30edcffec 100644 --- a/source/libs/qworker/inc/qworkerInt.h +++ b/source/libs/qworker/inc/qworkerInt.h @@ -31,8 +31,6 @@ enum { QW_PHASE_POST_QUERY, QW_PHASE_PRE_CQUERY, QW_PHASE_POST_CQUERY, - QW_PHASE_PRE_SINK, - QW_PHASE_POST_SINK, QW_PHASE_PRE_FETCH, QW_PHASE_POST_FETCH, }; @@ -105,10 +103,12 @@ typedef struct SQWTaskStatus { typedef struct SQWTaskCtx { SRWLatch lock; - int32_t phase; - - int32_t sinkId; - int32_t readyCode; + int8_t phase; + + bool emptyRes; + int8_t queryContinue; + int8_t inQueue; + int32_t rspCode; int8_t events[QW_EVENT_MAX]; @@ -144,7 +144,11 @@ typedef struct SQWorkerMgmt { #define QW_SET_EVENT_RECEIVED(ctx, event) atomic_store_8(&(ctx)->events[event], QW_EVENT_RECEIVED) #define QW_SET_EVENT_PROCESSED(ctx, event) atomic_store_8(&(ctx)->events[event], QW_EVENT_PROCESSED) -#define QW_IN_EXECUTOR(ctx) ((ctx)->phase == QW_PHASE_PRE_QUERY || (ctx)->phase == QW_PHASE_PRE_CQUERY || (ctx)->phase == QW_PHASE_PRE_FETCH || (ctx)->phase == QW_PHASE_PRE_SINK) +#define QW_GET_PHASE(ctx) atomic_load_8(&(ctx)->phase) + +#define QW_SET_RSP_CODE(ctx, code) atomic_val_compare_exchange_32(&(ctx)->rspCode, 0, code) + +#define QW_IN_EXECUTOR(ctx) (QW_GET_PHASE(ctx) == QW_PHASE_PRE_QUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_CQUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_FETCH) #define QW_TASK_NOT_EXIST(code) (TSDB_CODE_QRY_SCH_NOT_EXIST == (code) || TSDB_CODE_QRY_TASK_NOT_EXIST == (code)) #define QW_TASK_ALREADY_EXIST(code) (TSDB_CODE_QRY_TASK_ALREADY_EXIST == (code)) diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index d01f4f4e5231a0a02328a54ed0bde5ee96654752..c6e2bad4210673f818aac37cb69e57ac844c93e3 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -254,6 +254,19 @@ int32_t qwAcquireTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_ return TSDB_CODE_SUCCESS; } +int32_t qwGetTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx **ctx) { + char id[sizeof(qId) + sizeof(tId)] = {0}; + QW_SET_QTID(id, qId, tId); + + *ctx = taosHashGet(mgmt->ctxHash, id, sizeof(id)); + if (NULL == (*ctx)) { + QW_TASK_ELOG("ctx not in ctxHash, id:%s", id); + QW_ERR_RET(TSDB_CODE_QRY_RES_CACHE_NOT_EXIST); + } + + return TSDB_CODE_SUCCESS; +} + int32_t qwAddTaskCtxImpl(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int32_t rwType, int32_t status, SQWTaskCtx **ctx) { char id[sizeof(qId) + sizeof(tId)] = {0}; QW_SET_QTID(id, qId, tId); @@ -270,6 +283,8 @@ int32_t qwAddTaskCtxImpl(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_ if (HASH_NODE_EXIST(code)) { if (rwType && ctx) { QW_RET(qwAcquireTaskCtx(QW_FPARAMS(), rwType, ctx)); + } else if (ctx) { + QW_RET(qwGetTaskCtx(QW_FPARAMS(), ctx)); } else { QW_TASK_ELOG("task ctx already exist, id:%s", id); QW_ERR_RET(TSDB_CODE_QRY_TASK_ALREADY_EXIST); @@ -283,6 +298,8 @@ int32_t qwAddTaskCtxImpl(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_ if (rwType && ctx) { QW_RET(qwAcquireTaskCtx(QW_FPARAMS(), rwType, ctx)); + } else if (ctx) { + QW_RET(qwGetTaskCtx(QW_FPARAMS(), ctx)); } return TSDB_CODE_SUCCESS; @@ -292,39 +309,48 @@ int32_t qwAddTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tI QW_RET(qwAddTaskCtxImpl(QW_FPARAMS(), 0, 0, NULL)); } -int32_t qwGetTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx **ctx) { - char id[sizeof(qId) + sizeof(tId)] = {0}; - QW_SET_QTID(id, qId, tId); - - *ctx = taosHashGet(mgmt->ctxHash, id, sizeof(id)); - if (NULL == (*ctx)) { - QW_TASK_ELOG("ctx not in ctxHash, id:%s", id); - QW_ERR_RET(TSDB_CODE_QRY_RES_CACHE_NOT_EXIST); - } - - return TSDB_CODE_SUCCESS; - -} int32_t qwAddAcquireTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int32_t rwType, SQWTaskCtx **ctx) { return qwAddTaskCtxImpl(QW_FPARAMS(), rwType, 0, ctx); } +int32_t qwAddGetTaskCtx(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx **ctx) { + return qwAddTaskCtxImpl(QW_FPARAMS(), 0, 0, ctx); +} + + void qwReleaseTaskCtx(int32_t rwType, SQWorkerMgmt *mgmt) { QW_UNLOCK(rwType, &mgmt->ctxLock); } +void qwFreeTaskHandle(QW_FPARAMS_DEF, SQWTaskCtx *ctx) { + // RC WARNING + qTaskInfo_t taskHandle = atomic_load_ptr(&ctx->taskHandle); + if (taskHandle && atomic_val_compare_exchange_ptr(&ctx->taskHandle, taskHandle, NULL)) { + qDestroyTask(taskHandle); + } +} -void qwFreeTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx *ctx) { - if (ctx->taskHandle) { - qDestroyTask(ctx->taskHandle); - ctx->taskHandle = NULL; +int32_t qwKillTaskHandle(QW_FPARAMS_DEF, SQWTaskCtx *ctx) { + int32_t code = 0; + // RC WARNING + qTaskInfo_t taskHandle = atomic_load_ptr(&ctx->taskHandle); + if (taskHandle && atomic_val_compare_exchange_ptr(&ctx->taskHandle, taskHandle, NULL)) { + code = qKillTask(taskHandle); + atomic_store_ptr(&ctx->taskHandle, taskHandle); } - // TODO - if (ctx->sinkHandle) { + QW_RET(code); +} + +void qwFreeTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) { + qwFreeTaskHandle(QW_FPARAMS(), ctx); + + if (ctx->sinkHandle) { + dsDestroyDataSinker(ctx->sinkHandle); + ctx->sinkHandle = NULL; } } @@ -418,7 +444,7 @@ int32_t qwDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWTaskCtx *ctx = NULL; bool locked = false; - QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx)); + QW_ERR_JRET(qwAddGetTaskCtx(QW_FPARAMS(), &ctx)); QW_LOCK(QW_WRITE, &ctx->lock); @@ -430,9 +456,7 @@ int32_t qwDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, } if (QW_IN_EXECUTOR(ctx)) { - if (ctx->taskHandle) { - QW_ERR_JRET(qKillTask(ctx->taskHandle)); - } + QW_ERR_JRET(qwKillTaskHandle(QW_FPARAMS(), ctx)); QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_DROPPING)); } else if (ctx->phase > 0) { @@ -449,12 +473,12 @@ int32_t qwDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, _return: - if (locked) { - QW_UNLOCK(QW_WRITE, &ctx->lock); + if (code) { + QW_SET_RSP_CODE(ctx, code); } - if (ctx) { - qwReleaseTaskCtx(QW_READ, mgmt); + if (locked) { + QW_UNLOCK(QW_WRITE, &ctx->lock); } QW_RET(code); @@ -465,8 +489,11 @@ int32_t qwExecTask(QW_FPARAMS_DEF, qTaskInfo_t taskHandle, DataSinkHandle sinkHa bool qcontinue = true; SSDataBlock* pRes = NULL; uint64_t useconds = 0; - - while (qcontinue) { + int32_t i = 0; + + while (true) { + QW_TASK_DLOG("start to execTask in executor, loopIdx:%d", i++); + code = qExecTask(taskHandle, &pRes, &useconds); if (code) { QW_TASK_ELOG("qExecTask failed, code:%x", code); @@ -474,7 +501,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, qTaskInfo_t taskHandle, DataSinkHandle sinkHa } if (NULL == pRes) { - QW_TASK_DLOG("query done, useconds:%"PRIu64, useconds); + QW_TASK_DLOG("task query done, useconds:%"PRIu64, useconds); dsEndPut(sinkHandle, useconds); break; } @@ -485,6 +512,12 @@ int32_t qwExecTask(QW_FPARAMS_DEF, qTaskInfo_t taskHandle, DataSinkHandle sinkHa QW_TASK_ELOG("dsPutDataBlock failed, code:%x", code); QW_ERR_JRET(code); } + + QW_TASK_DLOG("data put into sink, rows:%d, continueExecTask:%d", pRes->info.rows, qcontinue); + + if (!qcontinue) { + break; + } } _return: @@ -499,6 +532,20 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void bool queryEnd = false; int32_t code = 0; + if (ctx->emptyRes) { + QW_TASK_DLOG("query empty result, query end, phase:%d", ctx->phase); + + QW_ERR_RET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCCEED)); + + QW_ERR_RET(qwMallocFetchRsp(len, &rsp)); + + *rspMsg = rsp; + *dataLen = 0; + pOutput->queryEnd = true; + + return TSDB_CODE_SUCCESS; + } + dsGetDataLength(ctx->sinkHandle, &len, &queryEnd); if (len < 0) { @@ -519,9 +566,7 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void QW_ERR_RET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCCEED)); QW_ERR_RET(qwMallocFetchRsp(len, &rsp)); - *rspMsg = rsp; - *dataLen = 0; return TSDB_CODE_SUCCESS; @@ -540,15 +585,12 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void QW_TASK_DLOG("task got data in sink, dataLength:%d", len); QW_ERR_RET(qwMallocFetchRsp(len, &rsp)); - *rspMsg = rsp; pOutput->pData = rsp->data; - code = dsGetDataBlock(ctx->sinkHandle, pOutput); if (code) { QW_TASK_ELOG("dsGetDataBlock failed, code:%x", code); - qwFreeFetchRsp(rsp); QW_ERR_RET(code); } @@ -576,7 +618,7 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S switch (phase) { case QW_PHASE_PRE_QUERY: { - QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx)); + QW_ERR_JRET(qwAddGetTaskCtx(QW_FPARAMS(), &ctx)); ctx->phase = phase; @@ -596,7 +638,6 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S output->needStop = true; QW_ERR_JRET(qwAddTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_CANCELLED)); - qwFreeTask(QW_FPARAMS(), ctx); QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL); @@ -609,7 +650,7 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S break; } case QW_PHASE_POST_QUERY: { - QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx)); + QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); QW_LOCK(QW_WRITE, &ctx->lock); @@ -617,8 +658,14 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S ctx->taskHandle = input->taskHandle; ctx->sinkHandle = input->sinkHandle; + + if (NULL == ctx->taskHandle && NULL == ctx->sinkHandle) { + ctx->emptyRes = true; + } - ctx->readyCode = input->code; + if (input->code) { + QW_SET_RSP_CODE(ctx, input->code); + } assert(!QW_IS_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL)); @@ -654,8 +701,78 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S } break; } + case QW_PHASE_PRE_CQUERY: { + QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); + + QW_LOCK(QW_WRITE, &ctx->lock); + + locked = true; + + ctx->phase = phase; + + if (QW_IS_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL)) { + QW_TASK_WLOG("task already cancelled, phase:%d", phase); + output->needStop = true; + output->rspCode = TSDB_CODE_QRY_TASK_CANCELLED; + QW_ERR_JRET(TSDB_CODE_QRY_TASK_CANCELLED); + } + + if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) { + QW_TASK_WLOG("task is dropping, phase:%d", phase); + output->needStop = true; + output->rspCode = TSDB_CODE_QRY_TASK_DROPPING; + } else if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_CANCEL)) { + QW_TASK_WLOG("task is cancelling, phase:%d", phase); + output->needStop = true; + output->rspCode = TSDB_CODE_QRY_TASK_CANCELLING; + } + + if (ctx->rspCode) { + QW_TASK_ELOG("task already failed, code:%x, phase:%d", ctx->rspCode, phase); + output->needStop = true; + output->rspCode = ctx->rspCode; + QW_ERR_JRET(output->rspCode); + } + break; + } + case QW_PHASE_POST_CQUERY: { + QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); + + QW_LOCK(QW_WRITE, &ctx->lock); + + locked = true; + + if (input->code) { + QW_SET_RSP_CODE(ctx, input->code); + } + + if (QW_IS_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL)) { + QW_TASK_WLOG("task already cancelled, phase:%d", phase); + output->needStop = true; + output->rspCode = TSDB_CODE_QRY_TASK_CANCELLED; + QW_ERR_JRET(TSDB_CODE_QRY_TASK_CANCELLED); + } + + if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) { + QW_TASK_WLOG("task is dropping, phase:%d", phase); + output->needStop = true; + output->rspCode = TSDB_CODE_QRY_TASK_DROPPING; + } else if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_CANCEL)) { + QW_TASK_WLOG("task is cancelling, phase:%d", phase); + output->needStop = true; + output->rspCode = TSDB_CODE_QRY_TASK_CANCELLING; + } + + if (ctx->rspCode) { + QW_TASK_ELOG("task failed, code:%x, phase:%d", ctx->rspCode, phase); + output->needStop = true; + output->rspCode = ctx->rspCode; + QW_ERR_JRET(output->rspCode); + } + break; + } case QW_PHASE_PRE_FETCH: { - QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx)); + QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); QW_LOCK(QW_WRITE, &ctx->lock); @@ -693,15 +810,26 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S output->rspCode = TSDB_CODE_QRY_TASK_MSG_ERROR; QW_ERR_JRET(TSDB_CODE_QRY_TASK_MSG_ERROR); } + + if (ctx->rspCode) { + QW_TASK_ELOG("task already failed, code:%x, phase:%d", ctx->rspCode, phase); + output->needStop = true; + output->rspCode = ctx->rspCode; + QW_ERR_JRET(output->rspCode); + } break; } case QW_PHASE_POST_FETCH: { - QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx)); + QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); QW_LOCK(QW_WRITE, &ctx->lock); locked = true; + if (input->code) { + QW_SET_RSP_CODE(ctx, input->code); + } + if (QW_IS_EVENT_PROCESSED(ctx, QW_EVENT_CANCEL)) { QW_TASK_WLOG("task already cancelled, phase:%d", phase); output->needStop = true; @@ -718,12 +846,17 @@ int32_t qwHandleTaskEvent(QW_FPARAMS_DEF, int32_t phase, SQWPhaseInput *input, S output->needStop = true; output->rspCode = TSDB_CODE_QRY_TASK_CANCELLING; } + + if (ctx->rspCode) { + QW_TASK_ELOG("task failed, code:%x, phase:%d", ctx->rspCode, phase); + output->needStop = true; + output->rspCode = ctx->rspCode; + QW_ERR_JRET(output->rspCode); + } break; } - } - _return: if (locked) { @@ -732,10 +865,6 @@ _return: QW_UNLOCK(QW_WRITE, &ctx->lock); } - if (ctx) { - qwReleaseTaskCtx(QW_READ, mgmt); - } - QW_RET(code); } @@ -774,12 +903,14 @@ int32_t qwProcessQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t QW_ERR_JRET(code); } - QW_ERR_JRET(qwBuildAndSendQueryRsp(qwMsg->connection, TSDB_CODE_SUCCESS)); + QW_ERR_JRET(qwBuildAndSendQueryRsp(qwMsg->connection, code)); + QW_TASK_DLOG("query msg rsped, code:%d", code); queryRsped = true; - QW_ERR_JRET(qwExecTask(QW_FPARAMS(), pTaskInfo, sinkHandle)); - + if (pTaskInfo && sinkHandle) { + QW_ERR_JRET(qwExecTask(QW_FPARAMS(), pTaskInfo, sinkHandle)); + } _return: if (code) { @@ -788,6 +919,10 @@ _return: if (!queryRsped) { code = qwBuildAndSendQueryRsp(qwMsg->connection, rspCode); + if (TSDB_CODE_SUCCESS == code) { + QW_TASK_DLOG("query msg rsped, code:%d", rspCode); + } + if (TSDB_CODE_SUCCESS == rspCode && code) { rspCode = code; } @@ -810,7 +945,8 @@ _return: QW_ERR_RET(qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_POST_QUERY, &input, &output)); if (queryRsped && output.needRsp) { - qwBuildAndSendReadyRsp(qwMsg->connection, output.rspCode); + qwBuildAndSendReadyRsp(qwMsg->connection, output.rspCode); + QW_TASK_DLOG("ready msg rsped, code:%x", output.rspCode); } QW_RET(rspCode); @@ -819,24 +955,34 @@ _return: int32_t qwProcessReady(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg) { int32_t code = 0; SQWTaskCtx *ctx = NULL; - - QW_ERR_JRET(qwAddAcquireTaskCtx(QW_FPARAMS(), QW_READ, &ctx)); + int8_t phase = 0; + QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); + QW_LOCK(QW_WRITE, &ctx->lock); - if (ctx->phase == QW_PHASE_PRE_QUERY) { + phase = QW_GET_PHASE(ctx); + + if (phase == QW_PHASE_PRE_QUERY) { QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_READY); - } else if (ctx->phase == QW_PHASE_POST_QUERY) { + QW_TASK_DLOG("ready msg not rsped, phase:%d", phase); + } else if (phase == QW_PHASE_POST_QUERY) { QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_READY); - QW_ERR_JRET(qwBuildAndSendReadyRsp(qwMsg->connection, ctx->readyCode)); + QW_ERR_JRET(qwBuildAndSendReadyRsp(qwMsg->connection, ctx->rspCode)); + QW_TASK_DLOG("ready msg rsped, code:%x", ctx->rspCode); + } else { + QW_TASK_ELOG("invalid phase when got ready msg, phase:%d", phase); + assert(0); } _return: + if (code) { + QW_SET_RSP_CODE(ctx, code); + } + if (ctx) { QW_UNLOCK(QW_WRITE, &ctx->lock); - - qwReleaseTaskCtx(QW_READ, mgmt); } QW_RET(code); @@ -844,84 +990,77 @@ _return: int32_t qwProcessCQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg) { + SQWTaskCtx *ctx = NULL; int32_t code = 0; bool queryRsped = false; bool needStop = false; struct SSubplan *plan = NULL; - int32_t rspCode = 0; SQWPhaseInput input = {0}; SQWPhaseOutput output = {0}; - SQWTaskCtx *ctx = NULL; void *rsp = NULL; int32_t dataLen = 0; + + do { + QW_ERR_JRET(qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_PRE_CQUERY, &input, &output)); - QW_ERR_JRET(qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_PRE_CQUERY, &input, &output)); + needStop = output.needStop; + code = output.rspCode; + + if (needStop) { + QW_TASK_DLOG("task need stop, phase:%d", QW_PHASE_PRE_CQUERY); + QW_ERR_JRET(code); + } - needStop = output.needStop; - code = output.rspCode; - - if (needStop) { - QW_TASK_DLOG("task need stop, phase:%d", QW_PHASE_PRE_CQUERY); - QW_ERR_JRET(code); - } + QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); - QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); - - qTaskInfo_t taskHandle = ctx->taskHandle; - DataSinkHandle sinkHandle = ctx->sinkHandle; + atomic_store_8(&ctx->inQueue, 0); - QW_ERR_JRET(qwExecTask(QW_FPARAMS(), taskHandle, sinkHandle)); + qTaskInfo_t taskHandle = ctx->taskHandle; + DataSinkHandle sinkHandle = ctx->sinkHandle; - QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_CQUERY); + QW_ERR_JRET(qwExecTask(QW_FPARAMS(), taskHandle, sinkHandle)); - if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) { - SOutputData sOutput = {0}; - QW_ERR_JRET(qwGetResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput)); - - // Note: schedule data sink firstly and will schedule query after it's done - if (sOutput.scheduleJobNo) { - if (sOutput.scheduleJobNo > ctx->sinkId) { - QW_TASK_DLOG("sink need schedule, scheduleJobNo:%d", sOutput.scheduleJobNo); - - ctx->sinkId = sOutput.scheduleJobNo; - QW_ERR_JRET(qwBuildAndSendSchSinkMsg(QW_FPARAMS(), qwMsg->connection)); + if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) { + SOutputData sOutput = {0}; + QW_ERR_JRET(qwGetResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput)); + + if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) { + QW_TASK_DLOG("task not end, need to continue, bufStatus:%d", sOutput.bufStatus); + + // RC WARNING + atomic_store_8(&ctx->queryContinue, 1); } - } else if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) { - QW_TASK_DLOG("task not end, need to continue, bufStatus:%d", sOutput.bufStatus); - - if (!QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_CQUERY)) { - QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_CQUERY); - - QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_EXECUTING)); + + if (rsp) { + qwBuildFetchRsp(rsp, &sOutput, dataLen); + + QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH); - QW_ERR_RET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), qwMsg->connection)); + qwBuildAndSendFetchRsp(qwMsg->connection, rsp, dataLen, code); + QW_TASK_DLOG("fetch msg rsped, code:%x, dataLen:%d", code, dataLen); + } else { + atomic_store_8(&ctx->queryContinue, 1); } } - - if (rsp) { - qwBuildFetchRsp(rsp, &sOutput, dataLen); - } - - } - -_return: - qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_POST_CQUERY, &input, &output); + _return: - if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) { - if (code) { + if (code && QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) { QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH); qwFreeFetchRsp(rsp); rsp = NULL; qwBuildAndSendFetchRsp(qwMsg->connection, rsp, 0, code); - } else if (rsp) { - QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH); - - qwBuildAndSendFetchRsp(qwMsg->connection, rsp, dataLen, code); + QW_TASK_DLOG("fetch msg rsped, code:%x, dataLen:%d", code, 0); } - } - - QW_RET(rspCode); + + input.code = code; + qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_POST_CQUERY, &input, &output); + + needStop = output.needStop; + code = output.rspCode; + } while ((!needStop) && (0 == code) && atomic_val_compare_exchange_8(&ctx->queryContinue, 1, 0)); + + QW_RET(code); } @@ -958,46 +1097,49 @@ int32_t qwProcessFetch(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t if (NULL == rsp) { QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_FETCH); + } else { + qwBuildFetchRsp(rsp, &sOutput, dataLen); } - // Note: schedule data sink firstly and will schedule query after it's done - if (sOutput.scheduleJobNo) { - if (sOutput.scheduleJobNo > ctx->sinkId) { - QW_TASK_DLOG("sink need schedule, scheduleJobNo:%d", sOutput.scheduleJobNo); - - ctx->sinkId = sOutput.scheduleJobNo; - - QW_ERR_JRET(qwBuildAndSendSchSinkMsg(QW_FPARAMS(), qwMsg->connection)); - } - } else if ((!sOutput.queryEnd) && (/* DS_BUF_LOW == sOutput.bufStatus || */ DS_BUF_EMPTY == sOutput.bufStatus)) { + if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) { QW_TASK_DLOG("task not end, need to continue, bufStatus:%d", sOutput.bufStatus); - if (!QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_CQUERY)) { - QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_CQUERY); + QW_LOCK(QW_WRITE, &ctx->lock); + locked = true; + // RC WARNING + if (QW_IN_EXECUTOR(ctx)) { + atomic_store_8(&ctx->queryContinue, 1); + } else if (0 == atomic_load_8(&ctx->inQueue)) { QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_EXECUTING)); + + atomic_store_8(&ctx->inQueue, 1); - QW_ERR_RET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), qwMsg->connection)); + QW_ERR_JRET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), qwMsg->connection)); } } - - if (rsp) { - qwBuildFetchRsp(rsp, &sOutput, dataLen); - } _return: + if (locked) { + QW_UNLOCK(QW_WRITE, &ctx->lock); + } + + input.code = code; + qwHandleTaskEvent(QW_FPARAMS(), QW_PHASE_POST_FETCH, &input, &output); if (code) { qwFreeFetchRsp(rsp); rsp = NULL; - qwBuildAndSendFetchRsp(qwMsg->connection, rsp, 0, code); + dataLen = 0; + qwBuildAndSendFetchRsp(qwMsg->connection, rsp, dataLen, code); + QW_TASK_DLOG("fetch msg rsped, code:%x, dataLen:%d", code, dataLen); } else if (rsp) { qwBuildAndSendFetchRsp(qwMsg->connection, rsp, dataLen, code); + QW_TASK_DLOG("fetch msg rsped, code:%x, dataLen:%d", code, dataLen); } - QW_RET(code); } diff --git a/source/libs/qworker/src/qworkerMsg.c b/source/libs/qworker/src/qworkerMsg.c index b9fd8e78b60e9affd231ae98db01c9e39f870c3e..dad3a5d73b7be66d91dda8391c0a02c75163afbf 100644 --- a/source/libs/qworker/src/qworkerMsg.c +++ b/source/libs/qworker/src/qworkerMsg.c @@ -229,42 +229,6 @@ int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchRe return TSDB_CODE_SUCCESS; } - -int32_t qwBuildAndSendSchSinkMsg(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, void *connection) { - SRpcMsg *pMsg = (SRpcMsg *)connection; - SSinkDataReq * req = (SSinkDataReq *)rpcMallocCont(sizeof(SSinkDataReq)); - if (NULL == req) { - qError("rpcMallocCont %d failed", (int32_t)sizeof(SSinkDataReq)); - QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); - } - - req->header.vgId = mgmt->nodeId; - req->sId = sId; - req->queryId = qId; - req->taskId = tId; - - SRpcMsg pNewMsg = { - .handle = pMsg->handle, - .ahandle = pMsg->ahandle, - .msgType = TDMT_VND_SCHEDULE_DATA_SINK, - .pCont = req, - .contLen = sizeof(SSinkDataReq), - .code = 0, - }; - - int32_t code = (*mgmt->putToQueueFp)(mgmt->nodeObj, &pNewMsg); - if (TSDB_CODE_SUCCESS != code) { - qError("put data sink schedule msg to queue failed, code:%x", code); - rpcFreeCont(req); - QW_ERR_RET(code); - } - - qDebug("put data sink schedule msg to query queue"); - - return TSDB_CODE_SUCCESS; -} - - int32_t qwBuildAndSendCQueryMsg(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, void *connection) { SRpcMsg *pMsg = (SRpcMsg *)connection; SQueryContinueReq * req = (SQueryContinueReq *)rpcMallocCont(sizeof(SQueryContinueReq)); @@ -366,25 +330,6 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { return TSDB_CODE_SUCCESS; } - - -int32_t qWorkerProcessDataSinkMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){ - if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) { - return TSDB_CODE_QRY_INVALID_INPUT; - } - - SSinkDataReq *msg = pMsg->pCont; - if (NULL == msg || pMsg->contLen < sizeof(*msg)) { - qError("invalid sink data msg"); - QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); - } - - //dsScheduleProcess(); - //TODO - - return TSDB_CODE_SUCCESS; -} - int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){ if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) { return TSDB_CODE_QRY_INVALID_INPUT; diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index 4b14f9f2c792e7727a50679609d5d9832a022c5e..c760acd52e80443d2a4bb7b5874ce7258e687798 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -102,38 +102,110 @@ typedef void* queue[2]; #define QUEUE_DATA(e, type, field) ((type*)((void*)((char*)(e)-offsetof(type, field)))) typedef struct { - SRpcInfo* pRpc; // associated SRpcInfo - SEpSet epSet; // ip list provided by app - void* ahandle; // handle provided by app - struct SRpcConn* pConn; // pConn allocated - tmsg_t msgType; // message type - uint8_t* pCont; // content provided by app - int32_t contLen; // content length - int32_t code; // error code - int16_t numOfTry; // number of try for different servers - int8_t oldInUse; // server EP inUse passed by app - int8_t redirect; // flag to indicate redirect - int8_t connType; // connection type - int64_t rid; // refId returned by taosAddRef - SRpcMsg* pRsp; // for synchronous API - tsem_t* pSem; // for synchronous API - SEpSet* pSet; // for synchronous API - char msg[0]; // RpcHead starts from here + SRpcInfo* pRpc; // associated SRpcInfo + SEpSet epSet; // ip list provided by app + void* ahandle; // handle provided by app + // struct SRpcConn* pConn; // pConn allocated + tmsg_t msgType; // message type + uint8_t* pCont; // content provided by app + int32_t contLen; // content length + // int32_t code; // error code + // int16_t numOfTry; // number of try for different servers + // int8_t oldInUse; // server EP inUse passed by app + // int8_t redirect; // flag to indicate redirect + int8_t connType; // connection type + int64_t rid; // refId returned by taosAddRef + SRpcMsg* pRsp; // for synchronous API + tsem_t* pSem; // for synchronous API + char* ip; + uint32_t port; + // SEpSet* pSet; // for synchronous API } SRpcReqContext; +typedef struct { + SRpcInfo* pRpc; // associated SRpcInfo + SEpSet epSet; // ip list provided by app + void* ahandle; // handle provided by app + // struct SRpcConn* pConn; // pConn allocated + tmsg_t msgType; // message type + uint8_t* pCont; // content provided by app + int32_t contLen; // content length + // int32_t code; // error code + // int16_t numOfTry; // number of try for different servers + // int8_t oldInUse; // server EP inUse passed by app + // int8_t redirect; // flag to indicate redirect + int8_t connType; // connection type + int64_t rid; // refId returned by taosAddRef + SRpcMsg* pRsp; // for synchronous API + tsem_t* pSem; // for synchronous API + char* ip; + uint32_t port; + // SEpSet* pSet; // for synchronous API +} STransConnCtx; + +#pragma pack(push, 1) + +typedef struct { + char version : 4; // RPC version + char comp : 4; // compression algorithm, 0:no compression 1:lz4 + char resflag : 2; // reserved bits + char spi : 3; // security parameter index + char encrypt : 3; // encrypt algorithm, 0: no encryption + + uint32_t code; // del later + uint32_t msgType; + int32_t msgLen; + uint8_t content[0]; // message body starts from here +} STransMsgHead; + +typedef struct { + int32_t reserved; + int32_t contLen; +} STransCompMsg; + +typedef struct { + uint32_t timeStamp; + uint8_t auth[TSDB_AUTH_LEN]; +} STransDigestMsg; + +#pragma pack(pop) + #define container_of(ptr, type, member) ((type*)((char*)(ptr)-offsetof(type, member))) -#define RPC_RESERVE_SIZE (sizeof(SRpcReqContext)) +#define RPC_RESERVE_SIZE (sizeof(STranConnCtx)) -#define RPC_MSG_OVERHEAD (sizeof(SRpcReqContext) + sizeof(SRpcHead) + sizeof(SRpcDigest)) +#define RPC_MSG_OVERHEAD (sizeof(SRpcHead) + sizeof(SRpcDigest)) #define rpcHeadFromCont(cont) ((SRpcHead*)((char*)cont - sizeof(SRpcHead))) #define rpcContFromHead(msg) (msg + sizeof(SRpcHead)) #define rpcMsgLenFromCont(contLen) (contLen + sizeof(SRpcHead)) #define rpcContLenFromMsg(msgLen) (msgLen - sizeof(SRpcHead)) #define rpcIsReq(type) (type & 1U) +#define TRANS_RESERVE_SIZE (sizeof(STranConnCtx)) + +#define TRANS_MSG_OVERHEAD (sizeof(STransMsgHead) + sizeof(STransDigestMsg)) +#define transHeadFromCont(cont) ((STransMsgHead*)((char*)cont - sizeof(STransMsgHead))) +#define transContFromHead(msg) (msg + sizeof(STransMsgHead)) +#define transMsgLenFromCont(contLen) (contLen + sizeof(STransMsgHead)) +#define transContLenFromMsg(msgLen) (msgLen - sizeof(STransMsgHead)); +#define transIsReq(type) (type & 1U) + int rpcAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey); void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey); int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen); SRpcHead* rpcDecompressRpcMsg(SRpcHead* pHead); +int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey); +void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey); +bool transCompressMsg(char* msg, int32_t len, int32_t* flen); +bool transDecompressMsg(char* msg, int32_t len, int32_t* flen); + +void transConnCtxDestroy(STransConnCtx* ctx); + +typedef struct SConnBuffer { + char* buf; + int len; + int cap; + int left; +} SConnBuffer; + #endif diff --git a/source/libs/transport/inc/transportInt.h b/source/libs/transport/inc/transportInt.h index e39e0d9273b214f30296ece89aa836cbce31d30f..3c8c922d83cd3f6014935c6dd9d0aa70e2332d62 100644 --- a/source/libs/transport/inc/transportInt.h +++ b/source/libs/transport/inc/transportInt.h @@ -45,6 +45,9 @@ extern "C" { void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle); void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle); +void taosCloseServer(void* arg); +void taosCloseClient(void* arg); + typedef struct { int sessions; // number of sessions allowed int numOfThreads; // number of threads to process incoming messages diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c index cb8ef87b48acd9bae0ae8acf7cfab43ccdd61942..cf1e1539654347933898860548cd984cd188d1a9 100644 --- a/source/libs/transport/src/trans.c +++ b/source/libs/transport/src/trans.c @@ -17,15 +17,9 @@ #include "transComm.h" -typedef struct SConnBuffer { - char* buf; - int len; - int cap; - int left; -} SConnBuffer; - -void* (*taosHandle[])(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) = { +void* (*taosInitHandle[])(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) = { taosInitServer, taosInitClient}; +void (*taosCloseHandle[])(void* arg) = {taosCloseServer, taosCloseClient}; void* rpcOpen(const SRpcInit* pInit) { SRpcInfo* pRpc = calloc(1, sizeof(SRpcInfo)); @@ -38,13 +32,18 @@ void* rpcOpen(const SRpcInit* pInit) { pRpc->cfp = pInit->cfp; pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads; pRpc->connType = pInit->connType; - pRpc->tcphandle = (*taosHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc); + pRpc->tcphandle = (*taosInitHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc); return pRpc; } -void rpcClose(void* arg) { return; } +void rpcClose(void* arg) { + SRpcInfo* pRpc = (SRpcInfo*)arg; + (*taosCloseHandle[pRpc->connType])(pRpc->tcphandle); + free(pRpc); + return; +} void* rpcMallocCont(int contLen) { - int size = contLen + RPC_MSG_OVERHEAD; + int size = contLen + TRANS_MSG_OVERHEAD; char* start = (char*)calloc(1, (size_t)size); if (start == NULL) { @@ -53,7 +52,7 @@ void* rpcMallocCont(int contLen) { } else { tTrace("malloc mem:%p size:%d", start, size); } - return start + sizeof(SRpcReqContext) + sizeof(SRpcHead); + return start + sizeof(STransMsgHead); } void rpcFreeCont(void* cont) { return; } void* rpcReallocCont(void* ptr, int contLen) { return NULL; } diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index f197e72ec5e475f49e355cb1bc100b8144937d2e..86e9c05ccb541f17261d4aec80bd007e36cd34db 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -21,15 +21,18 @@ typedef struct SCliConn { uv_connect_t connReq; uv_stream_t* stream; uv_write_t* writeReq; + SConnBuffer readBuf; void* data; queue conn; char spi; char secured; } SCliConn; + typedef struct SCliMsg { - SRpcReqContext* context; - queue q; - uint64_t st; + STransConnCtx* ctx; + SRpcMsg msg; + queue q; + uint64_t st; } SCliMsg; typedef struct SCliThrdObj { @@ -53,27 +56,92 @@ typedef struct SClientObj { static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port); static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn); -static void clientAllocrReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf); +// process data read from server, auth/decompress etc +static void clientProcessData(SCliConn* conn); +// check whether already read complete packet from server +static bool clientReadComplete(SConnBuffer* pBuf); +// alloc buf for read +static void clientAllocBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf); +// callback after read nbytes from socket static void clientReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf); +// callback after write data to socket static void clientWriteCb(uv_write_t* req, int status); +// callback after conn to server static void clientConnCb(uv_connect_t* req, int status); static void clientAsyncCb(uv_async_t* handle); static void clientDestroy(uv_handle_t* handle); static void clientConnDestroy(SCliConn* pConn); +static void clientMsgDestroy(SCliMsg* pMsg); + static void* clientThread(void* arg); +static void clientProcessData(SCliConn* conn) { + // impl +} static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd); -static void clientAllocrReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { +static bool clientReadComplete(SConnBuffer* data) { + STransMsgHead head; + int32_t headLen = sizeof(head); + if (data->len >= headLen) { + memcpy((char*)&head, data->buf, headLen); + int32_t msgLen = (int32_t)htonl((uint32_t)head.msgLen); + if (msgLen > data->len) { + data->left = msgLen - data->len; + return false; + } else { + return true; + } + } else { + return false; + } +} +static void clientAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { // impl later + static const int CAPACITY = 512; + + SCliConn* conn = handle->data; + SConnBuffer* pBuf = &conn->readBuf; + if (pBuf->cap == 0) { + pBuf->buf = (char*)calloc(CAPACITY, sizeof(char)); + pBuf->len = 0; + pBuf->cap = CAPACITY; + pBuf->left = -1; + buf->base = pBuf->buf; + buf->len = CAPACITY; + } else { + if (pBuf->len >= pBuf->cap) { + if (pBuf->left == -1) { + pBuf->cap *= 2; + pBuf->buf = realloc(pBuf->buf, pBuf->cap); + } else if (pBuf->len + pBuf->left > pBuf->cap) { + pBuf->cap = pBuf->len + pBuf->left; + pBuf->buf = realloc(pBuf->buf, pBuf->len + pBuf->left); + } + } + buf->base = pBuf->buf + pBuf->len; + buf->len = pBuf->cap - pBuf->len; + } } static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { // impl later - SCliConn* conn = handle->data; + SCliConn* conn = handle->data; + SConnBuffer* pBuf = &conn->readBuf; if (nread > 0) { + pBuf->len += nread; + if (clientReadComplete(pBuf)) { + tDebug("alread read complete pack"); + clientProcessData(conn); + } else { + tDebug("read halp packet, continue to read"); + } return; } + + if (nread != UV_EOF) { + tDebug("Read error %s\n", uv_err_name(nread)); + } // uv_close((uv_handle_t*)handle, clientDestroy); } @@ -96,15 +164,17 @@ static void clientWriteCb(uv_write_t* req, int status) { return; } - uv_read_start((uv_stream_t*)pConn->stream, clientAllocrReadBufferCb, clientReadCb); + uv_read_start((uv_stream_t*)pConn->stream, clientAllocReadBufferCb, clientReadCb); // impl later } static void clientWrite(SCliConn* pConn) { - SCliMsg* pMsg = pConn->data; - SRpcHead* pHead = rpcHeadFromCont(pMsg->context->pCont); - int msgLen = rpcMsgLenFromCont(pMsg->context->contLen); - char* msg = (char*)(pHead); + SCliMsg* pCliMsg = pConn->data; + SRpcMsg* pMsg = (SRpcMsg*)(&pCliMsg->msg); + STransMsgHead* pHead = transHeadFromCont(pMsg->pCont); + + int msgLen = transMsgLenFromCont(pMsg->contLen); + char* msg = (char*)(pHead); uv_buf_t wb = uv_buf_init(msg, msgLen); uv_write(pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, clientWriteCb); @@ -118,23 +188,22 @@ static void clientConnCb(uv_connect_t* req, int status) { return; } - SCliMsg* pMsg = pConn->data; - SEpSet* pEpSet = &pMsg->context->epSet; - SRpcMsg rpcMsg; - // rpcMsg.ahandle = pMsg->context->ahandle; - // rpcMsg.pCont = NULL; + SCliMsg* pMsg = pConn->data; + STransConnCtx* pCtx = ((SCliMsg*)(pConn->data))->ctx; + + SRpcMsg rpcMsg; + rpcMsg.ahandle = pCtx->ahandle; - char* fqdn = pEpSet->fqdn[pEpSet->inUse]; - uint32_t port = pEpSet->port[pEpSet->inUse]; if (status != 0) { // call user fp later - tError("failed to connect server(%s, %d), errmsg: %s", fqdn, port, uv_strerror(status)); - SRpcInfo* pRpc = pMsg->context->pRpc; - (pRpc->cfp)(NULL, &rpcMsg, pEpSet); + tError("failed to connect server(%s, %d), errmsg: %s", pCtx->ip, pCtx->port, uv_strerror(status)); + SRpcInfo* pRpc = pMsg->ctx->pRpc; + (pRpc->cfp)(NULL, &rpcMsg, NULL); uv_close((uv_handle_t*)req->handle, clientDestroy); return; } assert(pConn->stream == req->handle); + clientWrite(pConn); } static SCliConn* getConnFromCache(void* cache, char* ip, uint32_t port) { @@ -147,24 +216,20 @@ static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn) } static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { - SEpSet* pEpSet = &pMsg->context->epSet; - - char* fqdn = pEpSet->fqdn[pEpSet->inUse]; - uint32_t port = pEpSet->port[pEpSet->inUse]; - - uint64_t el = taosGetTimestampUs() - pMsg->st; + uint64_t et = taosGetTimestampUs(); + uint64_t el = et - pMsg->st; tDebug("msg tran time cost: %" PRIu64 "", el); + et = taosGetTimestampUs(); - SCliConn* conn = getConnFromCache(pThrd->cache, fqdn, port); + STransConnCtx* pCtx = pMsg->ctx; + SCliConn* conn = getConnFromCache(pThrd->cache, pCtx->ip, pCtx->port); if (conn != NULL) { // impl later conn->data = pMsg; conn->writeReq->data = conn; clientWrite(conn); - // uv_buf_t wb; - // uv_write(conn->writeReq, (uv_stream_t*)conn->stream, &wb, 1, clientWriteCb); } else { - SCliConn* conn = malloc(sizeof(SCliConn)); + SCliConn* conn = calloc(1, sizeof(SCliConn)); conn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t)); uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream)); @@ -172,23 +237,11 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { conn->connReq.data = conn; conn->data = pMsg; + struct sockaddr_in addr; - uv_ip4_addr(fqdn, port, &addr); + uv_ip4_addr(pMsg->ctx->ip, pMsg->ctx->port, &addr); // handle error in callback if fail to connect uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, clientConnCb); - - // SRpcMsg rpcMsg; - // SEpSet* pEpSet = &pMsg->context->epSet; - // SRpcInfo* pRpc = pMsg->context->pRpc; - //// rpcMsg.ahandle = pMsg->context->ahandle; - // rpcMsg.pCont = NULL; - // rpcMsg.ahandle = pMsg->context->ahandle; - // uint64_t el1 = taosGetTimestampUs() - et; - // tError("msg tran back first: time cost: %" PRIu64 "", el1); - // et = taosGetTimestampUs(); - //(pRpc->cfp)(NULL, &rpcMsg, pEpSet); - // uint64_t el2 = taosGetTimestampUs() - et; - // tError("msg tran back second: time cost: %" PRIu64 "", el2); } } static void clientAsyncCb(uv_async_t* handle) { @@ -205,7 +258,8 @@ static void clientAsyncCb(uv_async_t* handle) { while (!QUEUE_IS_EMPTY(&wq)) { queue* h = QUEUE_HEAD(&wq); QUEUE_REMOVE(h); - pMsg = QUEUE_DATA(h, SCliMsg, q); + + SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q); clientHandleReq(pMsg, pThrd); count++; if (count >= 2) { @@ -221,6 +275,7 @@ static void* clientThread(void* arg) { void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) { SClientObj* cli = calloc(1, sizeof(SClientObj)); + memcpy(cli->label, label, strlen(label)); cli->numOfThreads = numOfThreads; cli->pThreadObj = (SCliThrdObj**)calloc(cli->numOfThreads, sizeof(SCliThrdObj*)); @@ -245,22 +300,44 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, } return cli; } +static void clientMsgDestroy(SCliMsg* pMsg) { + // impl later + free(pMsg); +} +void taosCloseClient(void* arg) { + // impl later + SClientObj* cli = arg; + for (int i = 0; i < cli->numOfThreads; i++) { + SCliThrdObj* pThrd = 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); +} void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* pRid) { // impl later + char* ip = (char*)(pEpSet->fqdn[pEpSet->inUse]); + uint32_t port = pEpSet->port[pEpSet->inUse]; + SRpcInfo* pRpc = (SRpcInfo*)shandle; - int len = rpcCompressRpcMsg(pMsg->pCont, pMsg->contLen); + int32_t flen = 0; + if (transCompressMsg(pMsg->pCont, pMsg->contLen, &flen)) { + // imp later + } + + STransConnCtx* pCtx = calloc(1, sizeof(STransConnCtx)); - SRpcReqContext* pContext; - pContext = (SRpcReqContext*)((char*)pMsg->pCont - sizeof(SRpcHead) - sizeof(SRpcReqContext)); - pContext->ahandle = pMsg->ahandle; - pContext->pRpc = (SRpcInfo*)shandle; - pContext->epSet = *pEpSet; - pContext->contLen = len; - pContext->pCont = pMsg->pCont; - pContext->msgType = pMsg->msgType; - pContext->oldInUse = pEpSet->inUse; + pCtx->pRpc = (SRpcInfo*)shandle; + pCtx->ahandle = pMsg->ahandle; + pCtx->msgType = pMsg->msgType; + pCtx->ip = strdup(ip); + pCtx->port = port; assert(pRpc->connType == TAOS_CONN_CLIENT); // atomic or not @@ -268,14 +345,15 @@ void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* if (pRpc->index++ >= pRpc->numOfThreads) { pRpc->index = 0; } - SCliMsg* msg = malloc(sizeof(SCliMsg)); - msg->context = pContext; - msg->st = taosGetTimestampUs(); + SCliMsg* cliMsg = malloc(sizeof(SCliMsg)); + cliMsg->ctx = pCtx; + cliMsg->msg = *pMsg; + cliMsg->st = taosGetTimestampUs(); SCliThrdObj* thrd = ((SClientObj*)pRpc->tcphandle)->pThreadObj[index % pRpc->numOfThreads]; pthread_mutex_lock(&thrd->msgMtx); - QUEUE_PUSH(&thrd->msg, &msg->q); + QUEUE_PUSH(&thrd->msg, &cliMsg->q); pthread_mutex_unlock(&thrd->msgMtx); uv_async_send(thrd->cliAsync); diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index f23cfb6e2d45e9bedaef58cb562834ce49c44a45..617abeea3953b120a876763020e1e7ffcb035307 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -30,6 +30,20 @@ int rpcAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) { return ret; } +int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) { + T_MD5_CTX context; + int ret = -1; + + tMD5Init(&context); + tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN); + tMD5Update(&context, (uint8_t*)pMsg, msgLen); + tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN); + tMD5Final(&context); + + if (memcmp(context.digest, pAuth, sizeof(context.digest)) == 0) ret = 0; + + return ret; +} void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey) { T_MD5_CTX context; @@ -41,6 +55,17 @@ void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey) { memcpy(pAuth, context.digest, sizeof(context.digest)); } +void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey) { + T_MD5_CTX context; + + tMD5Init(&context); + tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN); + tMD5Update(&context, (uint8_t*)pMsg, msgLen); + tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN); + tMD5Final(&context); + + memcpy(pAuth, context.digest, sizeof(context.digest)); +} int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) { SRpcHead* pHead = rpcHeadFromCont(pCont); @@ -81,6 +106,54 @@ int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) { return finalLen; } +bool transCompressMsg(char* msg, int32_t len, int32_t* flen) { + return false; + // SRpcHead* pHead = rpcHeadFromCont(pCont); + bool succ = false; + int overhead = sizeof(STransCompMsg); + if (!NEEDTO_COMPRESSS_MSG(len)) { + return succ; + } + + char* buf = malloc(len + overhead + 8); // 8 extra bytes + if (buf == NULL) { + tError("failed to allocate memory for rpc msg compression, contLen:%d", len); + *flen = len; + return succ; + } + + int32_t clen = LZ4_compress_default(msg, buf, len, len + overhead); + tDebug("compress rpc msg, before:%d, after:%d, overhead:%d", len, clen, overhead); + /* + * only the compressed size is less than the value of contLen - overhead, the compression is applied + * The first four bytes is set to 0, the second four bytes are utilized to keep the original length of message + */ + if (clen > 0 && clen < len - overhead) { + STransCompMsg* pComp = (STransCompMsg*)msg; + pComp->reserved = 0; + pComp->contLen = htonl(len); + memcpy(msg + overhead, buf, clen); + + tDebug("compress rpc msg, before:%d, after:%d", len, clen); + *flen = clen + overhead; + succ = true; + } else { + *flen = len; + succ = false; + } + free(buf); + return succ; +} +bool transDecompressMsg(char* msg, int32_t len, int32_t* flen) { + // impl later + return false; + STransCompMsg* pComp = (STransCompMsg*)msg; + + int overhead = sizeof(STransCompMsg); + int clen = 0; + return false; +} + SRpcHead* rpcDecompressRpcMsg(SRpcHead* pHead) { int overhead = sizeof(SRpcComp); SRpcHead* pNewHead = NULL; @@ -114,4 +187,8 @@ SRpcHead* rpcDecompressRpcMsg(SRpcHead* pHead) { return pHead; } +void transConnCtxDestroy(STransConnCtx* ctx) { + free(ctx->ip); + free(ctx); +} #endif diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c index bc4cc695b008328614eab4385482d89094d3b901..d096ab78139efaf9e613ddaf7be1bc63b4689192 100644 --- a/source/libs/transport/src/transSrv.c +++ b/source/libs/transport/src/transSrv.c @@ -16,13 +16,6 @@ #ifdef USE_UV #include "transComm.h" -typedef struct SConnBuffer { - char* buf; - int len; - int cap; - int left; -} SConnBuffer; - typedef struct SConn { uv_tcp_t* pTcp; uv_write_t* pWriter; @@ -100,31 +93,32 @@ static void* acceptThread(void* arg); void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { /* * formate of data buffer: - * |<-------SRpcReqContext------->|<------------data read from socket----------->| + * |<--------------------------data from socket------------------------------->| + * |<------STransMsgHead------->|<-------------------other data--------------->| */ static const int CAPACITY = 1024; SConn* conn = handle->data; SConnBuffer* pBuf = &conn->connBuf; if (pBuf->cap == 0) { - pBuf->buf = (char*)calloc(CAPACITY + RPC_RESERVE_SIZE, sizeof(char)); + pBuf->buf = (char*)calloc(CAPACITY, sizeof(char)); pBuf->len = 0; pBuf->cap = CAPACITY; pBuf->left = -1; - buf->base = pBuf->buf + RPC_RESERVE_SIZE; + buf->base = pBuf->buf; buf->len = CAPACITY; } else { if (pBuf->len >= pBuf->cap) { if (pBuf->left == -1) { pBuf->cap *= 2; - pBuf->buf = realloc(pBuf->buf, pBuf->cap + RPC_RESERVE_SIZE); + pBuf->buf = realloc(pBuf->buf, pBuf->cap); } else if (pBuf->len + pBuf->left > pBuf->cap) { pBuf->cap = pBuf->len + pBuf->left; - pBuf->buf = realloc(pBuf->buf, pBuf->len + pBuf->left + RPC_RESERVE_SIZE); + pBuf->buf = realloc(pBuf->buf, pBuf->len + pBuf->left); } } - buf->base = pBuf->buf + pBuf->len + RPC_RESERVE_SIZE; + buf->base = pBuf->buf + pBuf->len; buf->len = pBuf->cap - pBuf->len; } } @@ -133,11 +127,11 @@ void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b // static bool readComplete(SConnBuffer* data) { // TODO(yihao): handle pipeline later - SRpcHead rpcHead; - int32_t headLen = sizeof(rpcHead); + STransMsgHead head; + int32_t headLen = sizeof(head); if (data->len >= headLen) { - memcpy((char*)&rpcHead, data->buf + RPC_RESERVE_SIZE, headLen); - int32_t msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen); + memcpy((char*)&head, data->buf, headLen); + int32_t msgLen = (int32_t)htonl((uint32_t)head.msgLen); if (msgLen > data->len) { data->left = msgLen - data->len; return false; @@ -150,21 +144,21 @@ static bool readComplete(SConnBuffer* data) { } static void uvDoProcess(SRecvInfo* pRecv) { - SRpcHead* pHead = (SRpcHead*)pRecv->msg; - SRpcInfo* pRpc = (SRpcInfo*)pRecv->shandle; - SConn* pConn = pRecv->thandle; - + // impl later + STransMsgHead* pHead = (STransMsgHead*)pRecv->msg; + SRpcInfo* pRpc = (SRpcInfo*)pRecv->shandle; + SConn* pConn = pRecv->thandle; tDump(pRecv->msg, pRecv->msgLen); - terrno = 0; - SRpcReqContext* pContest; + // SRpcReqContext* pContest; // do auth and check } static int uvAuthMsg(SConn* pConn, char* msg, int len) { - SRpcHead* pHead = (SRpcHead*)msg; - int code = 0; + STransMsgHead* pHead = (STransMsgHead*)msg; + + int code = 0; if ((pConn->secured && pHead->spi == 0) || (pHead->spi == 0 && pConn->spi == 0)) { // secured link, or no authentication @@ -224,7 +218,7 @@ static void uvProcessData(SConn* pConn) { SRecvInfo info; SRecvInfo* p = &info; SConnBuffer* pBuf = &pConn->connBuf; - p->msg = pBuf->buf + RPC_RESERVE_SIZE; + p->msg = pBuf->buf; p->msgLen = pBuf->len; p->ip = 0; p->port = 0; @@ -233,11 +227,10 @@ static void uvProcessData(SConn* pConn) { p->chandle = NULL; // - SRpcHead* pHead = (SRpcHead*)p->msg; - assert(rpcIsReq(pHead->msgType)); + STransMsgHead* pHead = (STransMsgHead*)p->msg; + assert(transIsReq(pHead->msgType)); SRpcInfo* pRpc = (SRpcInfo*)p->shandle; - pConn->ahandle = (void*)pHead->ahandle; // auth here int8_t code = uvAuthMsg(pConn, (char*)pHead, p->msgLen); @@ -247,14 +240,19 @@ static void uvProcessData(SConn* pConn) { } pHead->code = htonl(pHead->code); + int32_t dlen = 0; SRpcMsg rpcMsg; - - pHead = rpcDecompressRpcMsg(pHead); + if (transDecompressMsg(NULL, 0, NULL)) { + // add compress later + // pHead = rpcDecompressRpcMsg(pHead); + } else { + // impl later + } rpcMsg.contLen = rpcContLenFromMsg(pHead->msgLen); rpcMsg.pCont = pHead->content; rpcMsg.msgType = pHead->msgType; rpcMsg.code = pHead->code; - rpcMsg.ahandle = pConn->ahandle; + rpcMsg.ahandle = NULL; rpcMsg.handle = pConn; (*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL); @@ -265,13 +263,13 @@ static void uvProcessData(SConn* pConn) { void uvOnReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) { // opt - SConn* ctx = cli->data; - SConnBuffer* pBuf = &ctx->connBuf; + SConn* conn = cli->data; + SConnBuffer* pBuf = &conn->connBuf; if (nread > 0) { pBuf->len += nread; if (readComplete(pBuf)) { tDebug("alread read complete packet"); - uvProcessData(ctx); + uvProcessData(conn); } else { tDebug("read half packet, continue to read"); } @@ -423,7 +421,7 @@ void* workerThread(void* arg) { uv_loop_init(pThrd->loop); // SRpcInfo* pRpc = pThrd->shandle; - uv_pipe_init(pThrd->loop, pThrd->pipe, 1); + uv_pipe_init(pThrd->loop, pThrd->pipe, 0); uv_pipe_open(pThrd->pipe, pThrd->fd); pThrd->pipe->data = pThrd; @@ -491,6 +489,7 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, for (int i = 0; i < srv->numOfThreads; i++) { SWorkThrdObj* thrd = (SWorkThrdObj*)calloc(1, sizeof(SWorkThrdObj)); + srv->pipe[i] = (uv_pipe_t*)calloc(2, sizeof(uv_pipe_t)); int fds[2]; if (uv_socketpair(AF_UNIX, SOCK_STREAM, fds, UV_NONBLOCK_PIPE, UV_NONBLOCK_PIPE) != 0) { @@ -522,6 +521,22 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, return srv; } +void taosCloseServer(void* arg) { + // impl later + SServerObj* srv = arg; + for (int i = 0; i < srv->numOfThreads; i++) { + SWorkThrdObj* pThrd = srv->pThreadObj[i]; + pthread_join(pThrd->thread, NULL); + free(srv->pipe[i]); + free(pThrd->loop); + free(pThrd); + } + free(srv->loop); + free(srv->pipe); + free(srv->pThreadObj); + pthread_join(srv->thread, NULL); + free(srv); +} void rpcSendResponse(const SRpcMsg* pMsg) { SConn* pConn = pMsg->handle; diff --git a/source/libs/transport/test/rsclient.c b/source/libs/transport/test/rsclient.c new file mode 100644 index 0000000000000000000000000000000000000000..65170d4abb2745a469dfda3e4146c2ea85405b33 --- /dev/null +++ b/source/libs/transport/test/rsclient.c @@ -0,0 +1,196 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * 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 . + */ + + +#include "os.h" +#include "tutil.h" +#include "tglobal.h" +#include "rpcLog.h" +#include "trpc.h" +#include "taoserror.h" + +typedef struct { + int index; + SRpcEpSet epSet; + int num; + int numOfReqs; + int msgSize; + tsem_t rspSem; + tsem_t *pOverSem; + pthread_t thread; + void *pRpc; +} SInfo; + + +static int tcount = 0; +static int terror = 0; + +static void *sendRequest(void *param) { + SInfo *pInfo = (SInfo *)param; + SRpcMsg rpcMsg, rspMsg; + + tDebug("thread:%d, start to send request", pInfo->index); + + while ( pInfo->numOfReqs == 0 || pInfo->num < pInfo->numOfReqs) { + pInfo->num++; + rpcMsg.pCont = rpcMallocCont(pInfo->msgSize); + rpcMsg.contLen = pInfo->msgSize; + rpcMsg.handle = pInfo; + rpcMsg.msgType = 1; + tDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num); + + rpcSendRecv(pInfo->pRpc, &pInfo->epSet, &rpcMsg, &rspMsg); + + // handle response + if (rspMsg.code != 0) terror++; + + tDebug("thread:%d, rspLen:%d code:%d", pInfo->index, rspMsg.contLen, rspMsg.code); + + rpcFreeCont(rspMsg.pCont); + + if ( pInfo->num % 20000 == 0 ) + tInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num); + } + + tDebug("thread:%d, it is over", pInfo->index); + tcount++; + + return NULL; +} + +int main(int argc, char *argv[]) { + SRpcInit rpcInit; + SRpcEpSet epSet; + int msgSize = 128; + int numOfReqs = 0; + int appThreads = 1; + char serverIp[40] = "127.0.0.1"; + char secret[TSDB_KEY_LEN] = "mypassword"; + struct timeval systemTime; + int64_t startTime, endTime; + pthread_attr_t thattr; + + // server info + epSet.numOfEps = 1; + epSet.inUse = 0; + epSet.port[0] = 7000; + epSet.port[1] = 7000; + strcpy(epSet.fqdn[0], serverIp); + strcpy(epSet.fqdn[1], "192.168.0.1"); + + // client info + memset(&rpcInit, 0, sizeof(rpcInit)); + //rpcInit.localIp = "0.0.0.0"; + rpcInit.localPort = 0; + rpcInit.label = "APP"; + rpcInit.numOfThreads = 1; + rpcInit.sessions = 100; + rpcInit.idleTime = tsShellActivityTimer*1000; + rpcInit.user = "michael"; + rpcInit.secret = secret; + rpcInit.ckey = "key"; + rpcInit.spi = 1; + rpcInit.connType = TAOS_CONN_CLIENT; + + for (int i=1; iindex = i; + pInfo->epSet = epSet; + pInfo->numOfReqs = numOfReqs; + pInfo->msgSize = msgSize; + tsem_init(&pInfo->rspSem, 0, 0); + pInfo->pRpc = pRpc; + pthread_create(&pInfo->thread, &thattr, sendRequest, pInfo); + pInfo++; + } + + do { + usleep(1); + } while ( tcount < appThreads); + + gettimeofday(&systemTime, NULL); + endTime = systemTime.tv_sec*1000000 + systemTime.tv_usec; + float usedTime = (endTime - startTime)/1000.0; // mseconds + + tInfo("it takes %.3f mseconds to send %d requests to server, error num:%d", usedTime, numOfReqs*appThreads, terror); + tInfo("Performance: %.3f requests per second, msgSize:%d bytes", 1000.0*numOfReqs*appThreads/usedTime, msgSize); + + taosCloseLog(); + + return 0; +} + + diff --git a/source/libs/transport/test/uv.c b/source/libs/transport/test/uv.c new file mode 100644 index 0000000000000000000000000000000000000000..4c7d30900b220c5b1ea87fb55ffb16a415541986 --- /dev/null +++ b/source/libs/transport/test/uv.c @@ -0,0 +1,207 @@ +#include +#include +#include +#include +#include + +#include "task.h" +#include + +#define NUM_OF_THREAD 1 +#define TIMEOUT 10000 + +typedef struct SThreadObj { + pthread_t thread; + uv_pipe_t *pipe; + uv_loop_t *loop; + uv_async_t *workerAsync; // + int fd; +} SThreadObj; + +typedef struct SServerObj { + uv_tcp_t server; + uv_loop_t *loop; + int workerIdx; + int numOfThread; + SThreadObj **pThreadObj; + uv_pipe_t **pipe; +} SServerObj; + +typedef struct SConnCtx { + uv_tcp_t *pClient; + uv_timer_t *pTimer; + uv_async_t *pWorkerAsync; + int ref; +} SConnCtx; + +void echo_write(uv_write_t *req, int status) { + if (status < 0) { + fprintf(stderr, "Write error %s\n", uv_err_name(status)); + } + printf("write data to client\n"); + free(req); +} + +void echo_read(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) { + + SConnCtx *pConn = container_of(client, SConnCtx, pClient); + pConn->ref += 1; + printf("read data %d\n", nread, buf->base, buf->len); + if (nread > 0) { + uv_write_t *req = (uv_write_t *)malloc(sizeof(uv_write_t)); + // dispatch request to database other process thread + // just write out + uv_buf_t write_out; + write_out.base = buf->base; + write_out.len = nread; + uv_write((uv_write_t *)req, client, &write_out, 1, echo_write); + free(buf->base); + return; + } + + if (nread < 0) { + if (nread != UV_EOF) + fprintf(stderr, "Read error %s\n", uv_err_name(nread)); + uv_close((uv_handle_t *)client, NULL); + } + free(buf->base); +} + +void alloc_buffer(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) { + buf->base = malloc(suggested_size); + buf->len = suggested_size; +} + +void on_new_connection(uv_stream_t *s, int status) { + if (status == -1) { + // error! + return; + } + SServerObj *pObj = container_of(s, SServerObj, server); + printf("new_connection from client\n"); + + uv_tcp_t *client = (uv_tcp_t *)malloc(sizeof(uv_tcp_t)); + uv_tcp_init(pObj->loop, client); + if (uv_accept(s, (uv_stream_t *)client) == 0) { + uv_write_t *write_req = (uv_write_t *)malloc(sizeof(uv_write_t)); + uv_buf_t dummy_buf = uv_buf_init("a", 1); + // despatch to worker thread + pObj->workerIdx = (pObj->workerIdx + 1) % pObj->numOfThread; + uv_write2(write_req, (uv_stream_t *)&(pObj->pipe[pObj->workerIdx][0]), + &dummy_buf, 1, (uv_stream_t *)client, echo_write); + } else { + uv_close((uv_handle_t *)client, NULL); + } +} +void child_on_new_connection(uv_stream_t *q, ssize_t nread, + const uv_buf_t *buf) { + printf("x child_on_new_connection \n"); + if (nread < 0) { + if (nread != UV_EOF) + fprintf(stderr, "Read error %s\n", uv_err_name(nread)); + uv_close((uv_handle_t *)q, NULL); + return; + } + SThreadObj *pObj = (SThreadObj *)container_of(q, struct SThreadObj, pipe); + + uv_pipe_t *pipe = (uv_pipe_t *)q; + if (!uv_pipe_pending_count(pipe)) { + fprintf(stderr, "No pending count\n"); + return; + } + + uv_handle_type pending = uv_pipe_pending_type(pipe); + assert(pending == UV_TCP); + + SConnCtx *pConn = malloc(sizeof(SConnCtx)); + + /* init conn timer*/ + pConn->pTimer = malloc(sizeof(uv_timer_t)); + uv_timer_init(pObj->loop, pConn->pTimer); + + pConn->pClient = (uv_tcp_t *)malloc(sizeof(uv_tcp_t)); + pConn->pWorkerAsync = pObj->workerAsync; // thread safty + uv_tcp_init(pObj->loop, pConn->pClient); + + if (uv_accept(q, (uv_stream_t *)(pConn->pClient)) == 0) { + uv_os_fd_t fd; + uv_fileno((const uv_handle_t *)pConn->pClient, &fd); + fprintf(stderr, "Worker Accepted fd %d\n", fd); + uv_timer_start(pConn->pTimer, timeOutCallBack, TIMEOUT, 0); + uv_read_start((uv_stream_t *)(pConn->pClient), alloc_buffer, echo_read); + } else { + uv_timer_stop(pConn->pTimer); + free(pConn->pTimer); + uv_close((uv_handle_t *)pConn->pClient, NULL); + free(pConn->pClient); + free(pConn); + } +} + +static void workerAsyncCallback(uv_async_t *handle) { + SThreadObj *pObj = container_of(handle, SThreadObj, workerAsync); + // do nothing +} +void *worker_thread(void *arg) { + SThreadObj *pObj = (SThreadObj *)arg; + int fd = pObj->fd; + pObj->loop = (uv_loop_t *)malloc(sizeof(uv_loop_t)); + uv_loop_init(pObj->loop); + + uv_pipe_init(pObj->loop, pObj->pipe, 1); + uv_pipe_open(pObj->pipe, fd); + + pObj->workerAsync = malloc(sizeof(uv_async_t)); + uv_async_init(pObj->loop, pObj->workerAsync, workerAsyncCallback); + uv_read_start((uv_stream_t *)pObj->pipe, alloc_buffer, + child_on_new_connection); + + uv_run(pObj->loop, UV_RUN_DEFAULT); +} +int main() { + + SServerObj *server = calloc(1, sizeof(SServerObj)); + server->loop = (uv_loop_t *)malloc(sizeof(uv_loop_t)); + server->numOfThread = NUM_OF_THREAD; + server->workerIdx = 0; + server->pThreadObj = + (SThreadObj **)calloc(server->numOfThread, sizeof(SThreadObj *)); + server->pipe = (uv_pipe_t **)calloc(server->numOfThread, sizeof(uv_pipe_t *)); + + uv_loop_init(server->loop); + + for (int i = 0; i < server->numOfThread; i++) { + server->pThreadObj[i] = (SThreadObj *)calloc(1, sizeof(SThreadObj)); + server->pipe[i] = (uv_pipe_t *)calloc(2, sizeof(uv_pipe_t)); + int fds[2]; + if (uv_socketpair(AF_UNIX, SOCK_STREAM, fds, UV_NONBLOCK_PIPE, + UV_NONBLOCK_PIPE) != 0) { + return -1; + } + uv_pipe_init(server->loop, &(server->pipe[i][0]), 1); + uv_pipe_open(&(server->pipe[i][0]), fds[1]); // init write + + server->pThreadObj[i]->fd = fds[0]; + server->pThreadObj[i]->pipe = &(server->pipe[i][1]); // init read + int err = pthread_create(&(server->pThreadObj[i]->thread), NULL, + worker_thread, (void *)(server->pThreadObj[i])); + if (err == 0) { + printf("thread %d create\n", i); + } else { + printf("thread %d create failed", i); + } + + uv_tcp_init(server->loop, &server->server); + struct sockaddr_in bind_addr; + uv_ip4_addr("0.0.0.0", 7000, &bind_addr); + uv_tcp_bind(&server->server, (const struct sockaddr *)&bind_addr, 0); + int err = 0; + if ((err = uv_listen((uv_stream_t *)&server->server, 128, + on_new_connection)) != 0) { + fprintf(stderr, "Listen error %s\n", uv_err_name(err)); + return 2; + } + uv_run(server->loop, UV_RUN_DEFAULT); + return 0; + } +} diff --git a/tests/test/c/create_table.c b/tests/test/c/create_table.c index 2dd5f04a58a7e202d965341c48c0fe1625f2a6a6..080f1551c2a4aba707eb071d3d6d2bbf2ddcd259 100644 --- a/tests/test/c/create_table.c +++ b/tests/test/c/create_table.c @@ -26,6 +26,7 @@ char dbName[32] = "db"; char stbName[64] = "st"; int32_t numOfThreads = 1; int64_t numOfTables = 200000; +int64_t startOffset = 0; int32_t createTable = 1; int32_t insertData = 0; int32_t batchNumOfTbl = 100; @@ -84,7 +85,7 @@ void createDbAndStb() { } taos_free_result(pRes); - sprintf(qstr, "create table %s (ts timestamp, i int) tags (j bigint)", stbName); + sprintf(qstr, "create table if not exists %s (ts timestamp, i int) tags (j int)", stbName); pRes = taos_query(con, qstr); code = taos_errno(pRes); if (code != 0) { @@ -181,8 +182,19 @@ void *threadFunc(void *param) { exit(1); } - // printf("thread:%d, table range: %"PRId64 " - %"PRId64 "\n", pInfo->threadIndex, pInfo->tableBeginIndex, - // pInfo->tableEndIndex); + pError("====before thread:%d, table range: %"PRId64 " - %"PRId64 "\n", + pInfo->threadIndex, + pInfo->tableBeginIndex, + pInfo->tableEndIndex); + + pInfo->tableBeginIndex += startOffset; + pInfo->tableEndIndex += startOffset; + + pError("====after thread:%d, table range: %"PRId64 " - %"PRId64 "\n", + pInfo->threadIndex, + pInfo->tableBeginIndex, + pInfo->tableEndIndex); + sprintf(qstr, "use %s", pInfo->dbName); TAOS_RES *pRes = taos_query(con, qstr); taos_free_result(pRes); @@ -210,7 +222,7 @@ void *threadFunc(void *param) { TAOS_RES *pRes = taos_query(con, qstr); code = taos_errno(pRes); if ((code != 0) && (code != TSDB_CODE_RPC_AUTH_REQUIRED)) { - pError("failed to create table t%" PRId64 ", reason:%s", t, tstrerror(code)); + pError("failed to create table reason:%s, sql: %s", tstrerror(code), qstr); } taos_free_result(pRes); int64_t endTs = taosGetTimestampUs(); @@ -296,6 +308,8 @@ void printHelp() { printf("%s%s%s%d\n", indent, indent, "numOfThreads, default is ", numOfThreads); printf("%s%s\n", indent, "-n"); printf("%s%s%s%" PRId64 "\n", indent, indent, "numOfTables, default is ", numOfTables); + printf("%s%s\n", indent, "-g"); + printf("%s%s%s%" PRId64 "\n", indent, indent, "startOffset, default is ", startOffset); printf("%s%s\n", indent, "-v"); printf("%s%s%s%d\n", indent, indent, "numOfVgroups, default is ", numOfVgroups); printf("%s%s\n", indent, "-a"); @@ -329,6 +343,8 @@ void parseArgument(int32_t argc, char *argv[]) { numOfThreads = atoi(argv[++i]); } else if (strcmp(argv[i], "-n") == 0) { numOfTables = atoll(argv[++i]); + } else if (strcmp(argv[i], "-g") == 0) { + startOffset = atoll(argv[++i]); } else if (strcmp(argv[i], "-v") == 0) { numOfVgroups = atoi(argv[++i]); } else if (strcmp(argv[i], "-a") == 0) { @@ -352,6 +368,7 @@ void parseArgument(int32_t argc, char *argv[]) { pPrint("%s stbName:%s %s", GREEN, stbName, NC); pPrint("%s configDir:%s %s", GREEN, configDir, NC); pPrint("%s numOfTables:%" PRId64 " %s", GREEN, numOfTables, NC); + pPrint("%s startOffset:%" PRId64 " %s", GREEN, startOffset, NC); pPrint("%s numOfThreads:%d %s", GREEN, numOfThreads, NC); pPrint("%s numOfVgroups:%d %s", GREEN, numOfVgroups, NC); pPrint("%s createTable:%d %s", GREEN, createTable, NC); @@ -381,7 +398,7 @@ int32_t main(int32_t argc, char *argv[]) { createDbAndStb(); } - pPrint("%d threads are spawned to create %" PRId64 " tables", numOfThreads, numOfTables); + pPrint("%d threads are spawned to create %" PRId64 " tables, offset is %" PRId64 " ", numOfThreads, numOfTables, startOffset); pthread_attr_t thattr; pthread_attr_init(&thattr); @@ -407,7 +424,7 @@ int32_t main(int32_t argc, char *argv[]) { int64_t tableFrom = 0; for (int32_t i = 0; i < numOfThreads; ++i) { pInfo[i].tableBeginIndex = tableFrom; - pInfo[i].tableEndIndex = i < b ? tableFrom + a : tableFrom + a - 1; + pInfo[i].tableEndIndex = (i < b ? tableFrom + a : tableFrom + a - 1); tableFrom = pInfo[i].tableEndIndex + 1; pInfo[i].threadIndex = i; pInfo[i].minDelay = INT64_MAX;