diff --git a/source/libs/transport/inc/rpcLog.h b/source/libs/transport/inc/rpcLog.h index 904680bbe6..621504091c 100644 --- a/source/libs/transport/inc/rpcLog.h +++ b/source/libs/transport/inc/rpcLog.h @@ -24,13 +24,49 @@ extern "C" { extern int32_t rpcDebugFlag; -#define tFatal(...) { if (rpcDebugFlag & DEBUG_FATAL) { taosPrintLog("RPC FATAL ", rpcDebugFlag, __VA_ARGS__); }} -#define tError(...) { if (rpcDebugFlag & DEBUG_ERROR) { taosPrintLog("RPC ERROR ", rpcDebugFlag, __VA_ARGS__); }} -#define tWarn(...) { if (rpcDebugFlag & DEBUG_WARN) { taosPrintLog("RPC WARN ", rpcDebugFlag, __VA_ARGS__); }} -#define tInfo(...) { if (rpcDebugFlag & DEBUG_INFO) { taosPrintLog("RPC ", tscEmbedded ? 255 : rpcDebugFlag, __VA_ARGS__); }} -#define tDebug(...) { if (rpcDebugFlag & DEBUG_DEBUG) { taosPrintLog("RPC ", rpcDebugFlag, __VA_ARGS__); }} -#define tTrace(...) { if (rpcDebugFlag & DEBUG_TRACE) { taosPrintLog("RPC ", rpcDebugFlag, __VA_ARGS__); }} -#define tDump(x, y) { if (rpcDebugFlag & DEBUG_DUMP) { taosDumpData((unsigned char *)x, y); }} +// rpcDebugFlag = 143 +#define tFatal(...) \ + { \ + if (rpcDebugFlag & DEBUG_FATAL) { \ + taosPrintLog("RPC FATAL ", rpcDebugFlag, __VA_ARGS__); \ + } \ + } +#define tError(...) \ + { \ + if (rpcDebugFlag & DEBUG_ERROR) { \ + taosPrintLog("RPC ERROR ", rpcDebugFlag, __VA_ARGS__); \ + } \ + } +#define tWarn(...) \ + { \ + if (rpcDebugFlag & DEBUG_WARN) { \ + taosPrintLog("RPC WARN ", rpcDebugFlag, __VA_ARGS__); \ + } \ + } +#define tInfo(...) \ + { \ + if (rpcDebugFlag & DEBUG_INFO) { \ + taosPrintLog("RPC ", rpcDebugFlag, __VA_ARGS__); \ + } \ + } +#define tDebug(...) \ + { \ + if (rpcDebugFlag & DEBUG_DEBUG) { \ + taosPrintLog("RPC ", rpcDebugFlag, __VA_ARGS__); \ + } \ + } +#define tTrace(...) \ + { \ + if (rpcDebugFlag & DEBUG_TRACE) { \ + taosPrintLog("RPC ", rpcDebugFlag, __VA_ARGS__); \ + } \ + } +#define tDump(x, y) \ + { \ + if (rpcDebugFlag & DEBUG_DUMP) { \ + taosDumpData((unsigned char *)x, y); \ + } \ + } #ifdef __cplusplus } diff --git a/source/libs/transport/src/transport.c b/source/libs/transport/src/transport.c index 93bbaf2820..6cc2ca8c49 100644 --- a/source/libs/transport/src/transport.c +++ b/source/libs/transport/src/transport.c @@ -101,6 +101,13 @@ typedef struct SThreadObj { void* shandle; } SThreadObj; +typedef struct SClientObj { + char label[TSDB_LABEL_LEN]; + int32_t index; + int numOfThreads; + SThreadObj** pThreadObj; +} SClientObj; + #define RPC_MSG_OVERHEAD (sizeof(SRpcReqContext) + sizeof(SRpcHead) + sizeof(SRpcDigest)) #define rpcHeadFromCont(cont) ((SRpcHead*)((char*)cont - sizeof(SRpcHead))) #define rpcContFromHead(msg) (msg + sizeof(SRpcHead)) @@ -113,7 +120,7 @@ typedef struct SServerObj { uv_tcp_t server; uv_loop_t* loop; int workerIdx; - int numOfThread; + int numOfThreads; SThreadObj** pThreadObj; uv_pipe_t** pipe; uint32_t ip; @@ -179,18 +186,37 @@ static void* acceptThread(void* arg); 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* (*taosHandle[])(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) = {taosInitServer, taosInitClient}; + +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 = (SThreadObj**)calloc(cli->numOfThreads, sizeof(SThreadObj*)); + + for (int i = 0; i < cli->numOfThreads; i++) { + SThreadObj* thrd = (SThreadObj*)calloc(1, sizeof(SThreadObj)); + + int err = pthread_create(&thrd->thread, NULL, workerThread, (void*)(thrd)); + if (err == 0) { + tDebug("sucess to create tranport-client thread %d", i); + } + } + return cli; +} + void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) { SServerObj* srv = calloc(1, sizeof(SServerObj)); srv->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t)); - srv->numOfThread = numOfThreads; + srv->numOfThreads = numOfThreads; srv->workerIdx = 0; - srv->pThreadObj = (SThreadObj**)calloc(srv->numOfThread, sizeof(SThreadObj*)); - srv->pipe = (uv_pipe_t**)calloc(srv->numOfThread, sizeof(uv_pipe_t*)); + srv->pThreadObj = (SThreadObj**)calloc(srv->numOfThreads, sizeof(SThreadObj*)); + srv->pipe = (uv_pipe_t**)calloc(srv->numOfThreads, sizeof(uv_pipe_t*)); srv->ip = ip; srv->port = port; uv_loop_init(srv->loop); - for (int i = 0; i < srv->numOfThread; i++) { + for (int i = 0; i < srv->numOfThreads; i++) { SThreadObj* thrd = (SThreadObj*)calloc(1, sizeof(SThreadObj)); srv->pipe[i] = (uv_pipe_t*)calloc(2, sizeof(uv_pipe_t)); int fds[2]; @@ -299,8 +325,7 @@ static int uvAuthMsg(SRpcConn* pConn, char* msg, int len) { if (!rpcIsReq(pHead->msgType)) { // for response, if code is auth failure, it shall bypass the auth process code = htonl(pHead->code); - if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE || - code == TSDB_CODE_RPC_INVALID_VERSION || code == TSDB_CODE_RPC_AUTH_REQUIRED || + if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE || code == TSDB_CODE_RPC_INVALID_VERSION || code == TSDB_CODE_RPC_AUTH_REQUIRED || code == TSDB_CODE_MND_USER_NOT_EXIST || code == TSDB_CODE_RPC_NOT_READY) { pHead->msgLen = (int32_t)htonl((uint32_t)pHead->msgLen); // tTrace("%s, dont check authentication since code is:0x%x", pConn->info, code); @@ -460,7 +485,7 @@ void uvOnAcceptCb(uv_stream_t* stream, int status) { uv_buf_t buf = uv_buf_init((char*)notify, strlen(notify)); - pObj->workerIdx = (pObj->workerIdx + 1) % pObj->numOfThread; + pObj->workerIdx = (pObj->workerIdx + 1) % pObj->numOfThreads; tDebug("new conntion accepted by main server, dispatch to %dth worker-thread", pObj->workerIdx); uv_write2(wr, (uv_stream_t*)&(pObj->pipe[pObj->workerIdx][0]), &buf, 1, (uv_stream_t*)cli, uvOnWriteCb); } else { @@ -587,7 +612,9 @@ void* rpcOpen(const SRpcInit* pInit) { tstrncpy(pRpc->label, pInit->label, strlen(pInit->label)); } pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads; - pRpc->tcphandle = taosInitServer(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc); + pRpc->connType = pInit->connType; + pRpc->tcphandle = (*taosHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc); + // pRpc->taosInitServer(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc); return pRpc; } void rpcClose(void* arg) { return; } diff --git a/source/libs/transport/test/CMakeLists.txt b/source/libs/transport/test/CMakeLists.txt index 9e58bf08cd..c61f688060 100644 --- a/source/libs/transport/test/CMakeLists.txt +++ b/source/libs/transport/test/CMakeLists.txt @@ -1,8 +1,19 @@ add_executable(transportTest "") +add_executable(client "") +add_executable(server "") + target_sources(transportTest PRIVATE "transportTests.cc" ) +target_sources (client + PRIVATE + "rclient.c" +) +target_sources (server + PRIVATE + "rserver.c" +) target_include_directories(transportTest PUBLIC @@ -18,4 +29,30 @@ target_link_libraries (transportTest transport ) +target_include_directories(client + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/transport" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) + +target_link_libraries (client + os + util + common + gtest_main + transport +) +target_include_directories(server + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/transport" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) + +target_link_libraries (server + os + util + common + gtest_main + transport +) diff --git a/source/libs/transport/test/rclient.c b/source/libs/transport/test/rclient.c new file mode 100644 index 0000000000..045fb8520e --- /dev/null +++ b/source/libs/transport/test/rclient.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 "rpcLog.h" +#include "taoserror.h" +#include "tglobal.h" +#include "trpc.h" +#include "tutil.h" + +typedef struct { + int index; + SEpSet epSet; + int num; + int numOfReqs; + int msgSize; + tsem_t rspSem; + tsem_t * pOverSem; + pthread_t thread; + void * pRpc; +} SInfo; + +static void processResponse(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) { + SInfo *pInfo = (SInfo *)pMsg->ahandle; + tDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen, pMsg->code); + + if (pEpSet) pInfo->epSet = *pEpSet; + + rpcFreeCont(pMsg->pCont); + tsem_post(&pInfo->rspSem); +} + +static int tcount = 0; + +static void *sendRequest(void *param) { + SInfo * pInfo = (SInfo *)param; + SRpcMsg rpcMsg = {0}; + + 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.ahandle = pInfo; + rpcMsg.msgType = 1; + tDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num); + rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg, NULL); + if (pInfo->num % 20000 == 0) tInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num); + tsem_wait(&pInfo->rspSem); + } + + tDebug("thread:%d, it is over", pInfo->index); + tcount++; + + return NULL; +} + +int main(int argc, char *argv[]) { + SRpcInit rpcInit; + SEpSet epSet; + int msgSize = 128; + int numOfReqs = 0; + int appThreads = 1; + char serverIp[40] = "127.0.0.1"; + char secret[20] = "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.localPort = 0; + rpcInit.label = "APP"; + rpcInit.numOfThreads = 1; + rpcInit.cfp = processResponse; + 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; i < argc; ++i) { + if (strcmp(argv[i], "-p") == 0 && i < argc - 1) { + epSet.port[0] = atoi(argv[++i]); + } else if (strcmp(argv[i], "-i") == 0 && i < argc - 1) { + tstrncpy(epSet.fqdn[0], argv[++i], sizeof(epSet.fqdn[0])); + } else if (strcmp(argv[i], "-t") == 0 && i < argc - 1) { + rpcInit.numOfThreads = atoi(argv[++i]); + } else if (strcmp(argv[i], "-m") == 0 && i < argc - 1) { + msgSize = atoi(argv[++i]); + } else if (strcmp(argv[i], "-s") == 0 && i < argc - 1) { + rpcInit.sessions = atoi(argv[++i]); + } else if (strcmp(argv[i], "-n") == 0 && i < argc - 1) { + numOfReqs = atoi(argv[++i]); + } else if (strcmp(argv[i], "-a") == 0 && i < argc - 1) { + appThreads = atoi(argv[++i]); + } else if (strcmp(argv[i], "-o") == 0 && i < argc - 1) { + tsCompressMsgSize = atoi(argv[++i]); + } else if (strcmp(argv[i], "-u") == 0 && i < argc - 1) { + rpcInit.user = argv[++i]; + } else if (strcmp(argv[i], "-k") == 0 && i < argc - 1) { + rpcInit.secret = argv[++i]; + } else if (strcmp(argv[i], "-spi") == 0 && i < argc - 1) { + rpcInit.spi = atoi(argv[++i]); + } else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) { + rpcDebugFlag = atoi(argv[++i]); + } else { + printf("\nusage: %s [options] \n", argv[0]); + printf(" [-i ip]: first server IP address, default is:%s\n", serverIp); + printf(" [-p port]: server port number, default is:%d\n", epSet.port[0]); + printf(" [-t threads]: number of rpc threads, default is:%d\n", rpcInit.numOfThreads); + printf(" [-s sessions]: number of rpc sessions, default is:%d\n", rpcInit.sessions); + printf(" [-m msgSize]: message body size, default is:%d\n", msgSize); + printf(" [-a threads]: number of app threads, default is:%d\n", appThreads); + printf(" [-n requests]: number of requests per thread, default is:%d\n", numOfReqs); + printf(" [-o compSize]: compression message size, default is:%d\n", tsCompressMsgSize); + printf(" [-u user]: user name for the connection, default is:%s\n", rpcInit.user); + printf(" [-k secret]: password for the connection, default is:%s\n", rpcInit.secret); + printf(" [-spi SPI]: security parameter index, default is:%d\n", rpcInit.spi); + printf(" [-d debugFlag]: debug flag, default:%d\n", rpcDebugFlag); + printf(" [-h help]: print out this help\n\n"); + exit(0); + } + } + + taosInitLog("client.log", 100000, 10); + + void *pRpc = rpcOpen(&rpcInit); + if (pRpc == NULL) { + tError("failed to initialize RPC"); + return -1; + } + + tInfo("client is initialized"); + tInfo("threads:%d msgSize:%d requests:%d", appThreads, msgSize, numOfReqs); + + // gettimeofday(&systemTime, NULL); + // startTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec; + + SInfo *pInfo = (SInfo *)calloc(1, sizeof(SInfo) * appThreads); + + pthread_attr_init(&thattr); + pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE); + + for (int i = 0; i < appThreads; ++i) { + pInfo->index = 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.0f; // mseconds + + // tInfo("it takes %.3f mseconds to send %d requests to server", usedTime, numOfReqs * appThreads); + // tInfo("Performance: %.3f requests per second, msgSize:%d bytes", 1000.0 * numOfReqs * appThreads / usedTime, msgSize); + + int ch = getchar(); + UNUSED(ch); + + taosCloseLog(); + + return 0; +} diff --git a/source/libs/transport/test/rserver.c b/source/libs/transport/test/rserver.c new file mode 100644 index 0000000000..2e32aa57ca --- /dev/null +++ b/source/libs/transport/test/rserver.c @@ -0,0 +1,188 @@ +/* + * 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 . + */ + +//#define _DEFAULT_SOURCE +#include "os.h" +#include "rpcLog.h" +#include "tglobal.h" +#include "tqueue.h" +#include "trpc.h" + +int msgSize = 128; +int commit = 0; +int dataFd = -1; +STaosQueue *qhandle = NULL; +STaosQset * qset = NULL; + +void processShellMsg() { + static int num = 0; + STaosQall *qall; + SRpcMsg * pRpcMsg, rpcMsg; + int type; + void * pvnode; + + qall = taosAllocateQall(); + + while (1) { + int numOfMsgs = taosReadAllQitemsFromQset(qset, qall, &pvnode, NULL); + tDebug("%d shell msgs are received", numOfMsgs); + if (numOfMsgs <= 0) break; + + for (int i = 0; i < numOfMsgs; ++i) { + taosGetQitem(qall, (void **)&pRpcMsg); + + if (dataFd >= 0) { + if (write(dataFd, pRpcMsg->pCont, pRpcMsg->contLen) < 0) { + tInfo("failed to write data file, reason:%s", strerror(errno)); + } + } + } + + if (commit >= 2) { + num += numOfMsgs; + // if (taosFsync(dataFd) < 0) { + // tInfo("failed to flush data to file, reason:%s", strerror(errno)); + //} + + if (num % 10000 == 0) { + tInfo("%d request have been written into disk", num); + } + } + + taosResetQitems(qall); + for (int i = 0; i < numOfMsgs; ++i) { + taosGetQitem(qall, (void **)&pRpcMsg); + rpcFreeCont(pRpcMsg->pCont); + + memset(&rpcMsg, 0, sizeof(rpcMsg)); + rpcMsg.pCont = rpcMallocCont(msgSize); + rpcMsg.contLen = msgSize; + rpcMsg.handle = pRpcMsg->handle; + rpcMsg.code = 0; + rpcSendResponse(&rpcMsg); + + taosFreeQitem(pRpcMsg); + } + } + + taosFreeQall(qall); +} + +int retrieveAuthInfo(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey) { + // app shall retrieve the auth info based on meterID from DB or a data file + // demo code here only for simple demo + int ret = 0; + + if (strcmp(meterId, "michael") == 0) { + *spi = 1; + *encrypt = 0; + strcpy(secret, "mypassword"); + strcpy(ckey, "key"); + } else if (strcmp(meterId, "jeff") == 0) { + *spi = 0; + *encrypt = 0; + } else { + ret = -1; // user not there + } + + return ret; +} + +void processRequestMsg(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) { + SRpcMsg *pTemp; + + pTemp = taosAllocateQitem(sizeof(SRpcMsg)); + memcpy(pTemp, pMsg, sizeof(SRpcMsg)); + + tDebug("request is received, type:%d, contLen:%d, item:%p", pMsg->msgType, pMsg->contLen, pTemp); + taosWriteQitem(qhandle, pTemp); +} + +int main(int argc, char *argv[]) { + SRpcInit rpcInit; + char dataName[20] = "server.data"; + + taosBlockSIGPIPE(); + + memset(&rpcInit, 0, sizeof(rpcInit)); + rpcInit.localPort = 7000; + rpcInit.label = "SER"; + rpcInit.numOfThreads = 1; + rpcInit.cfp = processRequestMsg; + rpcInit.sessions = 1000; + rpcInit.idleTime = tsShellActivityTimer * 1500; + rpcInit.afp = retrieveAuthInfo; + + for (int i = 1; i < argc; ++i) { + if (strcmp(argv[i], "-p") == 0 && i < argc - 1) { + rpcInit.localPort = atoi(argv[++i]); + } else if (strcmp(argv[i], "-t") == 0 && i < argc - 1) { + rpcInit.numOfThreads = atoi(argv[++i]); + } else if (strcmp(argv[i], "-m") == 0 && i < argc - 1) { + msgSize = atoi(argv[++i]); + } else if (strcmp(argv[i], "-s") == 0 && i < argc - 1) { + rpcInit.sessions = atoi(argv[++i]); + } else if (strcmp(argv[i], "-o") == 0 && i < argc - 1) { + tsCompressMsgSize = atoi(argv[++i]); + } else if (strcmp(argv[i], "-w") == 0 && i < argc - 1) { + commit = atoi(argv[++i]); + } else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) { + rpcDebugFlag = atoi(argv[++i]); + dDebugFlag = rpcDebugFlag; + uDebugFlag = rpcDebugFlag; + } else { + printf("\nusage: %s [options] \n", argv[0]); + printf(" [-p port]: server port number, default is:%d\n", rpcInit.localPort); + printf(" [-t threads]: number of rpc threads, default is:%d\n", rpcInit.numOfThreads); + printf(" [-s sessions]: number of sessions, default is:%d\n", rpcInit.sessions); + printf(" [-m msgSize]: message body size, default is:%d\n", msgSize); + printf(" [-o compSize]: compression message size, default is:%d\n", tsCompressMsgSize); + printf(" [-w write]: write received data to file(0, 1, 2), default is:%d\n", commit); + printf(" [-d debugFlag]: debug flag, default:%d\n", rpcDebugFlag); + printf(" [-h help]: print out this help\n\n"); + exit(0); + } + } + + tsAsyncLog = 0; + rpcInit.connType = TAOS_CONN_SERVER; + taosInitLog("server.log", 100000, 10); + + void *pRpc = rpcOpen(&rpcInit); + if (pRpc == NULL) { + tError("failed to start RPC server"); + return -1; + } + + tInfo("RPC server is running, ctrl-c to exit"); + + if (commit) { + dataFd = open(dataName, O_APPEND | O_CREAT | O_WRONLY, S_IRWXU | S_IRWXG | S_IRWXO); + if (dataFd < 0) tInfo("failed to open data file, reason:%s", strerror(errno)); + } + + qhandle = taosOpenQueue(); + qset = taosOpenQset(); + taosAddIntoQset(qset, qhandle, NULL); + + processShellMsg(); + + if (dataFd >= 0) { + close(dataFd); + remove(dataName); + } + + return 0; +}