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

Merge branch 'feature/update' into feature/update_test

This commit is contained in:
Ping Xiao 2020-11-04 16:49:53 +08:00
parent ce8e1177a6 d9d28841ef
commit ba4b483bc1
19 changed files with 138 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/dnode/inc/dnodeVWrite.h
View File

@ -20,9 +20,12 @@
extern "C" { extern "C" {
#endif #endif
int32_t dnodeInitVnodeWrite(); int32_t dnodeInitVWrite();
void dnodeCleanupVnodeWrite(); void dnodeCleanupVWrite();
void dnodeDispatchToVnodeWriteQueue(SRpcMsg *pMsg); void dnodeDispatchToVWriteQueue(SRpcMsg *pMsg);
void * dnodeAllocVWriteQueue(void *pVnode);
void dnodeFreeVWriteQueue(void *wqueue);
void dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
src/dnode/src/dnodeMain.c
View File

@ -62,7 +62,7 @@ static const SDnodeComponent tsDnodeComponents[] = {
{"wal", walInit, walCleanUp}, {"wal", walInit, walCleanUp},
{"check", dnodeInitCheck, dnodeCleanupCheck}, // NOTES: dnodeInitCheck must be behind the dnodeinitStorage component !!! {"check", dnodeInitCheck, dnodeCleanupCheck}, // NOTES: dnodeInitCheck must be behind the dnodeinitStorage component !!!
{"vread", dnodeInitVnodeRead, dnodeCleanupVnodeRead}, {"vread", dnodeInitVnodeRead, dnodeCleanupVnodeRead},
{"vwrite", dnodeInitVnodeWrite, dnodeCleanupVnodeWrite}, {"vwrite", dnodeInitVWrite, dnodeCleanupVWrite},
{"mread", dnodeInitMnodeRead, dnodeCleanupMnodeRead}, {"mread", dnodeInitMnodeRead, dnodeCleanupMnodeRead},
{"mwrite", dnodeInitMnodeWrite, dnodeCleanupMnodeWrite}, {"mwrite", dnodeInitMnodeWrite, dnodeCleanupMnodeWrite},
{"mpeer", dnodeInitMnodePeer, dnodeCleanupMnodePeer}, {"mpeer", dnodeInitMnodePeer, dnodeCleanupMnodePeer},

8
src/dnode/src/dnodePeer.c
View File

@ -38,10 +38,10 @@ static void *tsDnodeServerRpc = NULL;
static void *tsDnodeClientRpc = NULL; static void *tsDnodeClientRpc = NULL;
int32_t dnodeInitServer() { int32_t dnodeInitServer() {
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeDispatchToVnodeWriteQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeDispatchToVWriteQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE] = dnodeDispatchToVnodeWriteQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE] = dnodeDispatchToVWriteQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_TABLE] = dnodeDispatchToVnodeWriteQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_TABLE] = dnodeDispatchToVWriteQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_STABLE] = dnodeDispatchToVnodeWriteQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_STABLE] = dnodeDispatchToVWriteQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeDispatchToMgmtQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE] = dnodeDispatchToMgmtQueue;

4
src/dnode/src/dnodeShell.c
View File

@ -38,10 +38,10 @@ static int32_t tsDnodeQueryReqNum = 0;
static int32_t tsDnodeSubmitReqNum = 0; static int32_t tsDnodeSubmitReqNum = 0;
int32_t dnodeInitShell() { int32_t dnodeInitShell() {
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeDispatchToVnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeDispatchToVWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_QUERY] = dnodeDispatchToVnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_QUERY] = dnodeDispatchToVnodeReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_FETCH] = dnodeDispatchToVnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_FETCH] = dnodeDispatchToVnodeReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_UPDATE_TAG_VAL] = dnodeDispatchToVnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_UPDATE_TAG_VAL] = dnodeDispatchToVWriteQueue;
// the following message shall be treated as mnode write // the following message shall be treated as mnode write
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_ACCT] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_ACCT] = dnodeDispatchToMnodeWriteQueue;

4
src/dnode/src/dnodeVRead.c
View File

@ -132,7 +132,7 @@ void dnodeDispatchToVnodeReadQueue(SRpcMsg *pMsg) {
} }
} }
void *dnodeAllocateVnodeRqueue(void *pVnode) { void *dnodeAllocVReadQueue(void *pVnode) {
pthread_mutex_lock(&readPool.mutex); pthread_mutex_lock(&readPool.mutex);
taos_queue queue = taosOpenQueue(); taos_queue queue = taosOpenQueue();
if (queue == NULL) { if (queue == NULL) {
@ -167,7 +167,7 @@ void *dnodeAllocateVnodeRqueue(void *pVnode) {
return queue; return queue;
} }
void dnodeFreeVnodeRqueue(void *rqueue) { void dnodeFreeVReadQueue(void *rqueue) {
taosCloseQueue(rqueue); taosCloseQueue(rqueue);
// dynamically adjust the number of threads // dynamically adjust the number of threads

154
src/dnode/src/dnodeVWrite.c
View File

@ -15,74 +15,65 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "taosmsg.h" #include "tglobal.h"
#include "taoserror.h"
#include "tutil.h"
#include "tqueue.h" #include "tqueue.h"
#include "trpc.h"
#include "tsdb.h" #include "tsdb.h"
#include "twal.h" #include "twal.h"
#include "tdataformat.h"
#include "tglobal.h"
#include "tsync.h" #include "tsync.h"
#include "vnode.h" #include "vnode.h"
#include "dnodeInt.h"
#include "syncInt.h" #include "syncInt.h"
#include "dnodeVWrite.h" #include "dnodeInt.h"
#include "dnodeMgmt.h"
typedef struct { typedef struct {
taos_qall qall; taos_qall qall;
taos_qset qset; // queue set taos_qset qset; // queue set
pthread_t thread; // thread
int32_t workerId; // worker ID int32_t workerId; // worker ID
pthread_t thread; // thread
} SWriteWorker; } SWriteWorker;
typedef struct { typedef struct {
SRspRet rspRet; SRspRet rspRet;
SRpcMsg rpcMsg;
int32_t processedCount; int32_t processedCount;
int32_t code; int32_t code;
void *pCont;
int32_t contLen; int32_t contLen;
SRpcMsg rpcMsg; void * pCont;
} SWriteMsg; } SWriteMsg;
typedef struct { typedef struct {
int32_t max; // max number of workers int32_t max; // max number of workers
int32_t nextId; // from 0 to max-1, cyclic int32_t nextId; // from 0 to max-1, cyclic
SWriteWorker *writeWorker; SWriteWorker *worker;
pthread_mutex_t mutex; pthread_mutex_t mutex;
} SWriteWorkerPool; } SWriteWorkerPool;
static SWriteWorkerPool tsVWriteWP;
static void *dnodeProcessWriteQueue(void *param); static void *dnodeProcessWriteQueue(void *param);
static void dnodeHandleIdleWorker(SWriteWorker *pWorker);
SWriteWorkerPool wWorkerPool; int32_t dnodeInitVWrite() {
tsVWriteWP.max = tsNumOfCores;
tsVWriteWP.worker = (SWriteWorker *)tcalloc(sizeof(SWriteWorker), tsVWriteWP.max);
if (tsVWriteWP.worker == NULL) return -1;
pthread_mutex_init(&tsVWriteWP.mutex, NULL);
int32_t dnodeInitVnodeWrite() { for (int32_t i = 0; i < tsVWriteWP.max; ++i) {
wWorkerPool.max = tsNumOfCores; tsVWriteWP.worker[i].workerId = i;
wWorkerPool.writeWorker = (SWriteWorker *)calloc(sizeof(SWriteWorker), wWorkerPool.max);
if (wWorkerPool.writeWorker == NULL) return -1;
pthread_mutex_init(&wWorkerPool.mutex, NULL);
for (int32_t i = 0; i < wWorkerPool.max; ++i) {
wWorkerPool.writeWorker[i].workerId = i;
} }
dInfo("dnode write is initialized, max worker %d", wWorkerPool.max); dInfo("dnode vwrite is initialized, max worker %d", tsVWriteWP.max);
return 0; return 0;
} }
void dnodeCleanupVnodeWrite() { void dnodeCleanupVWrite() {
for (int32_t i = 0; i < wWorkerPool.max; ++i) { for (int32_t i = 0; i < tsVWriteWP.max; ++i) {
SWriteWorker *pWorker = wWorkerPool.writeWorker + i; SWriteWorker *pWorker = tsVWriteWP.worker + i;
if (pWorker->thread) { if (pWorker->thread) {
taosQsetThreadResume(pWorker->qset); taosQsetThreadResume(pWorker->qset);
} }
} }
for (int32_t i = 0; i < wWorkerPool.max; ++i) { for (int32_t i = 0; i < tsVWriteWP.max; ++i) {
SWriteWorker *pWorker = wWorkerPool.writeWorker + i; SWriteWorker *pWorker = tsVWriteWP.worker + i;
if (pWorker->thread) { if (pWorker->thread) {
pthread_join(pWorker->thread, NULL); pthread_join(pWorker->thread, NULL);
taosFreeQall(pWorker->qall); taosFreeQall(pWorker->qall);
@ -90,13 +81,13 @@ void dnodeCleanupVnodeWrite() {
} }
} }
pthread_mutex_destroy(&wWorkerPool.mutex); pthread_mutex_destroy(&tsVWriteWP.mutex);
free(wWorkerPool.writeWorker); tfree(tsVWriteWP.worker);
dInfo("dnode write is closed"); dInfo("dnode vwrite is closed");
} }
void dnodeDispatchToVnodeWriteQueue(SRpcMsg *pMsg) { void dnodeDispatchToVWriteQueue(SRpcMsg *pMsg) {
char *pCont = (char *)pMsg->pCont; char *pCont = pMsg->pCont;
if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) { if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) {
SMsgDesc *pDesc = (SMsgDesc *)pCont; SMsgDesc *pDesc = (SMsgDesc *)pCont;
@ -111,7 +102,7 @@ void dnodeDispatchToVnodeWriteQueue(SRpcMsg *pMsg) {
taos_queue queue = vnodeAcquireWqueue(pHead->vgId); taos_queue queue = vnodeAcquireWqueue(pHead->vgId);
if (queue) { if (queue) {
// put message into queue // put message into queue
SWriteMsg *pWrite = (SWriteMsg *)taosAllocateQitem(sizeof(SWriteMsg)); SWriteMsg *pWrite = taosAllocateQitem(sizeof(SWriteMsg));
pWrite->rpcMsg = *pMsg; pWrite->rpcMsg = *pMsg;
pWrite->pCont = pCont; pWrite->pCont = pCont;
pWrite->contLen = pHead->contLen; pWrite->contLen = pHead->contLen;
@ -130,12 +121,12 @@ void dnodeDispatchToVnodeWriteQueue(SRpcMsg *pMsg) {
} }
} }
void *dnodeAllocateVnodeWqueue(void *pVnode) { void *dnodeAllocVWriteQueue(void *pVnode) {
pthread_mutex_lock(&wWorkerPool.mutex); pthread_mutex_lock(&tsVWriteWP.mutex);
SWriteWorker *pWorker = wWorkerPool.writeWorker + wWorkerPool.nextId; SWriteWorker *pWorker = tsVWriteWP.worker + tsVWriteWP.nextId;
void *queue = taosOpenQueue(); void *queue = taosOpenQueue();
if (queue == NULL) { if (queue == NULL) {
pthread_mutex_unlock(&wWorkerPool.mutex); pthread_mutex_unlock(&tsVWriteWP.mutex);
return NULL; return NULL;
} }
@ -143,7 +134,7 @@ void *dnodeAllocateVnodeWqueue(void *pVnode) {
pWorker->qset = taosOpenQset(); pWorker->qset = taosOpenQset();
if (pWorker->qset == NULL) { if (pWorker->qset == NULL) {
taosCloseQueue(queue); taosCloseQueue(queue);
pthread_mutex_unlock(&wWorkerPool.mutex); pthread_mutex_unlock(&tsVWriteWP.mutex);
return NULL; return NULL;
} }
@ -152,7 +143,7 @@ void *dnodeAllocateVnodeWqueue(void *pVnode) {
if (pWorker->qall == NULL) { if (pWorker->qall == NULL) {
taosCloseQset(pWorker->qset); taosCloseQset(pWorker->qset);
taosCloseQueue(queue); taosCloseQueue(queue);
pthread_mutex_unlock(&wWorkerPool.mutex); pthread_mutex_unlock(&tsVWriteWP.mutex);
return NULL; return NULL;
} }
pthread_attr_t thAttr; pthread_attr_t thAttr;
@ -160,37 +151,35 @@ void *dnodeAllocateVnodeWqueue(void *pVnode) {
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessWriteQueue, pWorker) != 0) { if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessWriteQueue, pWorker) != 0) {
dError("failed to create thread to process read queue, reason:%s", strerror(errno)); dError("failed to create thread to process vwrite queue since %s", strerror(errno));
taosFreeQall(pWorker->qall); taosFreeQall(pWorker->qall);
taosCloseQset(pWorker->qset); taosCloseQset(pWorker->qset);
taosCloseQueue(queue); taosCloseQueue(queue);
queue = NULL; queue = NULL;
} else { } else {
dDebug("write worker:%d is launched", pWorker->workerId); dDebug("dnode vwrite worker:%d is launched", pWorker->workerId);
wWorkerPool.nextId = (wWorkerPool.nextId + 1) % wWorkerPool.max; tsVWriteWP.nextId = (tsVWriteWP.nextId + 1) % tsVWriteWP.max;
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
} else { } else {
taosAddIntoQset(pWorker->qset, queue, pVnode); taosAddIntoQset(pWorker->qset, queue, pVnode);
wWorkerPool.nextId = (wWorkerPool.nextId + 1) % wWorkerPool.max; tsVWriteWP.nextId = (tsVWriteWP.nextId + 1) % tsVWriteWP.max;
} }
pthread_mutex_unlock(&wWorkerPool.mutex); pthread_mutex_unlock(&tsVWriteWP.mutex);
dDebug("pVnode:%p, write queue:%p is allocated", pVnode, queue); dDebug("pVnode:%p, dnode vwrite queue:%p is allocated", pVnode, queue);
return queue; return queue;
} }
void dnodeFreeVnodeWqueue(void *wqueue) { void dnodeFreeVWriteQueue(void *wqueue) {
taosCloseQueue(wqueue); taosCloseQueue(wqueue);
// dynamically adjust the number of threads
} }
void dnodeSendRpcVnodeWriteRsp(void *pVnode, void *param, int32_t code) { void dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code) {
SWriteMsg *pWrite = (SWriteMsg *)param; if (param == NULL) return;
if (pWrite == NULL) return; SWriteMsg *pWrite = param;
if (code < 0) pWrite->code = code; if (code < 0) pWrite->code = code;
int32_t count = atomic_add_fetch_32(&pWrite->processedCount, 1); int32_t count = atomic_add_fetch_32(&pWrite->processedCount, 1);
@ -215,44 +204,45 @@ static void *dnodeProcessWriteQueue(void *param) {
SWriteWorker *pWorker = (SWriteWorker *)param; SWriteWorker *pWorker = (SWriteWorker *)param;
SWriteMsg * pWrite; SWriteMsg * pWrite;
SWalHead * pHead; SWalHead * pHead;
int32_t numOfMsgs;
int type;
void * pVnode, *item;
SRspRet * pRspRet; SRspRet * pRspRet;
void * pVnode;
void * pItem;
int32_t numOfMsgs;
int32_t qtype;
dDebug("write worker:%d is running", pWorker->workerId); dDebug("dnode vwrite worker:%d is running", pWorker->workerId);
while (1) { while (1) {
numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, pWorker->qall, &pVnode); numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, pWorker->qall, &pVnode);
if (numOfMsgs == 0) { if (numOfMsgs == 0) {
dDebug("qset:%p, dnode write got no message from qset, exiting", pWorker->qset); dDebug("qset:%p, dnode vwrite got no message from qset, exiting", pWorker->qset);
break; break;
} }
for (int32_t i = 0; i < numOfMsgs; ++i) { for (int32_t i = 0; i < numOfMsgs; ++i) {
pWrite = NULL; pWrite = NULL;
pRspRet = NULL; pRspRet = NULL;
taosGetQitem(pWorker->qall, &type, &item); taosGetQitem(pWorker->qall, &qtype, &pItem);
if (type == TAOS_QTYPE_RPC) { if (qtype == TAOS_QTYPE_RPC) {
pWrite = (SWriteMsg *)item; pWrite = pItem;
pRspRet = &pWrite->rspRet; pRspRet = &pWrite->rspRet;
pHead = (SWalHead *)(pWrite->pCont - sizeof(SWalHead)); pHead = (SWalHead *)((char *)pWrite->pCont - sizeof(SWalHead));
pHead->msgType = pWrite->rpcMsg.msgType; pHead->msgType = pWrite->rpcMsg.msgType;
pHead->version = 0; pHead->version = 0;
pHead->len = pWrite->contLen; pHead->len = pWrite->contLen;
dDebug("%p, rpc msg:%s will be processed in vwrite queue", pWrite->rpcMsg.ahandle, dDebug("%p, rpc msg:%s will be processed in vwrite queue", pWrite->rpcMsg.ahandle,
taosMsg[pWrite->rpcMsg.msgType]); taosMsg[pWrite->rpcMsg.msgType]);
} else if (type == TAOS_QTYPE_CQ) { } else if (qtype == TAOS_QTYPE_CQ) {
pHead = (SWalHead *)((char*)item + sizeof(SSyncHead)); pHead = (SWalHead *)((char *)pItem + sizeof(SSyncHead));
dTrace("%p, CQ wal msg:%s will be processed in vwrite queue, version:%" PRIu64, pHead, taosMsg[pHead->msgType], dTrace("%p, CQ wal msg:%s will be processed in vwrite queue, version:%" PRIu64, pHead, taosMsg[pHead->msgType],
pHead->version); pHead->version);
} else { } else {
pHead = (SWalHead *)item; pHead = pItem;
dTrace("%p, wal msg:%s will be processed in vwrite queue, version:%" PRIu64, pHead, taosMsg[pHead->msgType], dTrace("%p, wal msg:%s will be processed in vwrite queue, version:%" PRIu64, pHead, taosMsg[pHead->msgType],
pHead->version); pHead->version);
} }
int32_t code = vnodeProcessWrite(pVnode, type, pHead, pRspRet); int32_t code = vnodeProcessWrite(pVnode, qtype, pHead, pRspRet);
dTrace("%p, msg:%s is processed in vwrite queue, version:%" PRIu64 ", result:%s", pHead, taosMsg[pHead->msgType], dTrace("%p, msg:%s is processed in vwrite queue, version:%" PRIu64 ", result:%s", pHead, taosMsg[pHead->msgType],
pHead->version, tstrerror(code)); pHead->version, tstrerror(code));
@ -267,17 +257,17 @@ static void *dnodeProcessWriteQueue(void *param) {
// browse all items, and process them one by one // browse all items, and process them one by one
taosResetQitems(pWorker->qall); taosResetQitems(pWorker->qall);
for (int32_t i = 0; i < numOfMsgs; ++i) { for (int32_t i = 0; i < numOfMsgs; ++i) {
taosGetQitem(pWorker->qall, &type, &item); taosGetQitem(pWorker->qall, &qtype, &pItem);
if (type == TAOS_QTYPE_RPC) { if (qtype == TAOS_QTYPE_RPC) {
pWrite = (SWriteMsg *)item; pWrite = pItem;
dnodeSendRpcVnodeWriteRsp(pVnode, item, pWrite->rpcMsg.code); dnodeSendRpcVWriteRsp(pVnode, pItem, pWrite->rpcMsg.code);
} else if (type == TAOS_QTYPE_FWD) { } else if (qtype == TAOS_QTYPE_FWD) {
pHead = (SWalHead *)item; pHead = pItem;
vnodeConfirmForward(pVnode, pHead->version, 0); vnodeConfirmForward(pVnode, pHead->version, 0);
taosFreeQitem(item); taosFreeQitem(pItem);
vnodeRelease(pVnode); vnodeRelease(pVnode);
} else { } else {
taosFreeQitem(item); taosFreeQitem(pItem);
vnodeRelease(pVnode); vnodeRelease(pVnode);
} }
} }
@ -285,19 +275,3 @@ static void *dnodeProcessWriteQueue(void *param) {
return NULL; return NULL;
} }
UNUSED_FUNC
static void dnodeHandleIdleWorker(SWriteWorker *pWorker) {
int32_t num = taosGetQueueNumber(pWorker->qset);
if (num > 0) {
usleep(30000);
sched_yield();
} else {
taosFreeQall(pWorker->qall);
taosCloseQset(pWorker->qset);
pWorker->qset = NULL;
dDebug("write worker:%d is released", pWorker->workerId);
pthread_exit(NULL);
}
}

10
src/inc/dnode.h
View File

@ -53,11 +53,11 @@ void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp);
void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet); void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet);
void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid); void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid);
void *dnodeAllocateVnodeWqueue(void *pVnode); void *dnodeAllocVWriteQueue(void *pVnode);
void dnodeFreeVnodeWqueue(void *queue); void dnodeFreeVWriteQueue(void *wqueue);
void *dnodeAllocateVnodeRqueue(void *pVnode); void *dnodeAllocVReadQueue(void *pVnode);
void dnodeFreeVnodeRqueue(void *rqueue); void dnodeFreeVReadQueue(void *rqueue);
void dnodeSendRpcVnodeWriteRsp(void *pVnode, void *param, int32_t code); void dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code);
int32_t dnodeAllocateMnodePqueue(); int32_t dnodeAllocateMnodePqueue();
void dnodeFreeMnodePqueue(); void dnodeFreeMnodePqueue();

4
src/os/inc/osAlloc.h
View File

@ -22,14 +22,14 @@ extern "C" {
#ifndef TAOS_OS_FUNC_ALLOC #ifndef TAOS_OS_FUNC_ALLOC
#define tmalloc(size) malloc(size) #define tmalloc(size) malloc(size)
#define tcalloc(size) calloc(1, size) #define tcalloc(nmemb, size) calloc(nmemb, size)
#define trealloc(p, size) realloc(p, size) #define trealloc(p, size) realloc(p, size)
#define tmemalign(alignment, size) malloc(size) #define tmemalign(alignment, size) malloc(size)
#define tfree(p) free(p) #define tfree(p) free(p)
#define tmemzero(p, size) memset(p, 0, size) #define tmemzero(p, size) memset(p, 0, size)
#else #else
void *tmalloc(int32_t size); void *tmalloc(int32_t size);
void *tcalloc(int32_t size); void *tcalloc(int32_t nmemb, int32_t size);
void *trealloc(void *p, int32_t size); void *trealloc(void *p, int32_t size);
void *tmemalign(int32_t alignment, int32_t size); void *tmemalign(int32_t alignment, int32_t size);
void tfree(void *p); void tfree(void *p);

18
src/os/inc/osFile.h
View File

@ -31,10 +31,14 @@ extern "C" {
} \ } \
} }
int64_t taosRead(int32_t fd, void *buf, int64_t count); int64_t taosReadImp(int32_t fd, void *buf, int64_t count);
int64_t taosWrite(int32_t fd, void *buf, int64_t count); int64_t taosWriteImp(int32_t fd, void *buf, int64_t count);
int64_t taosLSeek(int32_t fd, int64_t offset, int32_t whence); int64_t taosLSeekImp(int32_t fd, int64_t offset, int32_t whence);
int32_t taosRenameFile(char *fullPath, char *suffix, char delimiter, char **dstPath); int32_t taosRenameFile(char *fullPath, char *suffix, char delimiter, char **dstPath);
#define taosRead(fd, buf, count) taosReadImp(fd, buf, count)
#define taosWrite(fd, buf, count) taosWriteImp(fd, buf, count)
#define taosLSeek(fd, offset, whence) taosLSeekImp(fd, offset, whence)
#define taosClose(x) tclose(x) #define taosClose(x) tclose(x)
// TAOS_OS_FUNC_FILE_SENDIFLE // TAOS_OS_FUNC_FILE_SENDIFLE
@ -42,12 +46,12 @@ int64_t taosSendFile(int32_t dfd, int32_t sfd, int64_t *offset, int64_t size);
int64_t taosFSendFile(FILE *outfile, FILE *infile, int64_t *offset, int64_t size); int64_t taosFSendFile(FILE *outfile, FILE *infile, int64_t *offset, int64_t size);
#ifdef TAOS_RANDOM_FILE_FAIL #ifdef TAOS_RANDOM_FILE_FAIL
void taosSetRandomFileFailFactor(int factor); void taosSetRandomFileFailFactor(int32_t factor);
void taosSetRandomFileFailOutput(const char *path); void taosSetRandomFileFailOutput(const char *path);
#ifdef TAOS_RANDOM_FILE_FAIL_TEST #ifdef TAOS_RANDOM_FILE_FAIL_TEST
ssize_t taosReadFileRandomFail(int fd, void *buf, size_t count, const char *file, uint32_t line); int64_t taosReadFileRandomFail(int32_t fd, void *buf, int32_t count, const char *file, uint32_t line);
ssize_t taosWriteFileRandomFail(int fd, void *buf, size_t count, const char *file, uint32_t line); int64_t taosWriteFileRandomFail(int32_t fd, void *buf, int32_t count, const char *file, uint32_t line);
off_t taosLSeekRandomFail(int fd, off_t offset, int whence, const char *file, uint32_t line); int64_t taosLSeekRandomFail(int32_t fd, int64_t offset, int32_t whence, const char *file, uint32_t line);
#undef taosRead #undef taosRead
#undef taosWrite #undef taosWrite
#undef taosLSeek #undef taosLSeek

8
src/os/inc/osSocket.h
View File

@ -42,10 +42,10 @@ extern "C" {
#ifdef TAOS_RANDOM_NETWORK_FAIL #ifdef TAOS_RANDOM_NETWORK_FAIL
#ifdef TAOS_RANDOM_NETWORK_FAIL_TEST #ifdef TAOS_RANDOM_NETWORK_FAIL_TEST
ssize_t taosSendRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t flags); int64_t taosSendRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t flags);
ssize_t taosSendToRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t flags, const struct sockaddr *dest_addr, socklen_t addrlen); int64_t taosSendToRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t flags, const struct sockaddr *dest_addr, socklen_t addrlen);
ssize_t taosReadSocketRandomFail(int32_t fd, void *buf, size_t count); int64_t taosReadSocketRandomFail(int32_t fd, void *buf, size_t count);
ssize_t taosWriteSocketRandomFail(int32_t fd, const void *buf, size_t count); int64_t taosWriteSocketRandomFail(int32_t fd, const void *buf, size_t count);
#undef taosSend #undef taosSend
#undef taosSendto #undef taosSendto
#undef taosReadSocket #undef taosReadSocket

6
src/os/src/detail/osAlloc.c
View File

@ -32,11 +32,11 @@ void *tmalloc(int32_t size) {
return p; return p;
} }
void *tcalloc(int32_t size) { void *tcalloc(int32_t nmemb, int32_t size) {
void *p = calloc(1, size); void *p = calloc(nmemb, size);
if (p == NULL) { if (p == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to calloc memory, size:%d reason:%s", size, strerror(errno)); uError("failed to calloc memory, nmemb:%d size:%d reason:%s", nmemb, size, strerror(errno));
} }
return p; return p;

22
src/os/src/detail/osFail.c
View File

@ -20,7 +20,7 @@
#ifdef TAOS_RANDOM_NETWORK_FAIL #ifdef TAOS_RANDOM_NETWORK_FAIL
ssize_t taosSendRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t flags) { int64_t taosSendRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t flags) {
if (rand() % RANDOM_NETWORK_FAIL_FACTOR == 0) { if (rand() % RANDOM_NETWORK_FAIL_FACTOR == 0) {
errno = ECONNRESET; errno = ECONNRESET;
return -1; return -1;
@ -29,8 +29,8 @@ ssize_t taosSendRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t
return send(sockfd, buf, len, flags); return send(sockfd, buf, len, flags);
} }
ssize_t taosSendToRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t flags, const struct sockaddr *dest_addr, int64_t taosSendToRandomFail(int32_t sockfd, const void *buf, size_t len, int32_t flags,
socklen_t addrlen) { const struct sockaddr *dest_addr, socklen_t addrlen) {
if (rand() % RANDOM_NETWORK_FAIL_FACTOR == 0) { if (rand() % RANDOM_NETWORK_FAIL_FACTOR == 0) {
errno = ECONNRESET; errno = ECONNRESET;
return -1; return -1;
@ -39,7 +39,7 @@ ssize_t taosSendToRandomFail(int32_t sockfd, const void *buf, size_t len, int32_
return sendto(sockfd, buf, len, flags, dest_addr, addrlen); return sendto(sockfd, buf, len, flags, dest_addr, addrlen);
} }
ssize_t taosReadSocketRandomFail(int32_t fd, void *buf, size_t count) { int64_t taosReadSocketRandomFail(int32_t fd, void *buf, size_t count) {
if (rand() % RANDOM_NETWORK_FAIL_FACTOR == 0) { if (rand() % RANDOM_NETWORK_FAIL_FACTOR == 0) {
errno = ECONNRESET; errno = ECONNRESET;
return -1; return -1;
@ -48,7 +48,7 @@ ssize_t taosReadSocketRandomFail(int32_t fd, void *buf, size_t count) {
return read(fd, buf, count); return read(fd, buf, count);
} }
ssize_t taosWriteSocketRandomFail(int32_t fd, const void *buf, size_t count) { int64_t taosWriteSocketRandomFail(int32_t fd, const void *buf, size_t count) {
if (rand() % RANDOM_NETWORK_FAIL_FACTOR == 0) { if (rand() % RANDOM_NETWORK_FAIL_FACTOR == 0) {
errno = EINTR; errno = EINTR;
return -1; return -1;
@ -105,7 +105,7 @@ void taosSetRandomFileFailOutput(const char *path) {
sigaction(SIGILL, &act, NULL); sigaction(SIGILL, &act, NULL);
} }
ssize_t taosReadFileRandomFail(int32_t fd, void *buf, size_t count, const char *file, uint32_t line) { int64_t taosReadFileRandomFail(int32_t fd, void *buf, int32_t count, const char *file, uint32_t line) {
if (random_file_fail_factor > 0) { if (random_file_fail_factor > 0) {
if (rand() % random_file_fail_factor == 0) { if (rand() % random_file_fail_factor == 0) {
errno = EIO; errno = EIO;
@ -113,10 +113,10 @@ ssize_t taosReadFileRandomFail(int32_t fd, void *buf, size_t count, const char *
} }
} }
return taosRead(fd, buf, count); return taosReadImp(fd, buf, count);
} }
ssize_t taosWriteFileRandomFail(int32_t fd, void *buf, size_t count, const char *file, uint32_t line) { int64_t taosWriteFileRandomFail(int32_t fd, void *buf, int32_t count, const char *file, uint32_t line) {
if (random_file_fail_factor > 0) { if (random_file_fail_factor > 0) {
if (rand() % random_file_fail_factor == 0) { if (rand() % random_file_fail_factor == 0) {
errno = EIO; errno = EIO;
@ -124,10 +124,10 @@ ssize_t taosWriteFileRandomFail(int32_t fd, void *buf, size_t count, const char
} }
} }
return taosWrite(fd, buf, count); return taosWriteImp(fd, buf, count);
} }
off_t taosLSeekRandomFail(int32_t fd, off_t offset, int32_t whence, const char *file, uint32_t line) { int64_t taosLSeekRandomFail(int32_t fd, int64_t offset, int32_t whence, const char *file, uint32_t line) {
if (random_file_fail_factor > 0) { if (random_file_fail_factor > 0) {
if (rand() % random_file_fail_factor == 0) { if (rand() % random_file_fail_factor == 0) {
errno = EIO; errno = EIO;
@ -135,7 +135,7 @@ off_t taosLSeekRandomFail(int32_t fd, off_t offset, int32_t whence, const char *
} }
} }
return taosLSeek(fd, offset, whence); return taosLSeekImp(fd, offset, whence);
} }
#endif //TAOS_RANDOM_FILE_FAIL #endif //TAOS_RANDOM_FILE_FAIL

8
src/os/src/detail/osFile.c
View File

@ -71,7 +71,7 @@ int32_t taosRenameFile(char *fullPath, char *suffix, char delimiter, char **dstP
return rename(fullPath, *dstPath); return rename(fullPath, *dstPath);
} }
int64_t taosRead(int32_t fd, void *buf, int64_t count) { int64_t taosReadImp(int32_t fd, void *buf, int64_t count) {
int64_t leftbytes = count; int64_t leftbytes = count;
int64_t readbytes; int64_t readbytes;
char * tbuf = (char *)buf; char * tbuf = (char *)buf;
@ -95,7 +95,7 @@ int64_t taosRead(int32_t fd, void *buf, int64_t count) {
return count; return count;
} }
int64_t taosWrite(int32_t fd, void *buf, int64_t n) { int64_t taosWriteImp(int32_t fd, void *buf, int64_t n) {
int64_t nleft = n; int64_t nleft = n;
int64_t nwritten = 0; int64_t nwritten = 0;
char * tbuf = (char *)buf; char * tbuf = (char *)buf;
@ -115,6 +115,10 @@ int64_t taosWrite(int32_t fd, void *buf, int64_t n) {
return n; return n;
} }
int64_t taosLSeekImp(int32_t fd, int64_t offset, int32_t whence) {
return (int64_t)tlseek(fd, (long)offset, whence);
}
#ifndef TAOS_OS_FUNC_FILE_SENDIFLE #ifndef TAOS_OS_FUNC_FILE_SENDIFLE
int64_t taosSendFile(int32_t dfd, int32_t sfd, int64_t *offset, int64_t size) { int64_t taosSendFile(int32_t dfd, int32_t sfd, int64_t *offset, int64_t size) {

4
src/tsdb/src/tsdbFile.c
View File

@ -13,10 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include <regex.h>
#define TAOS_RANDOM_FILE_FAIL_TEST #define TAOS_RANDOM_FILE_FAIL_TEST
#include <regex.h>
#include "os.h" #include "os.h"
#include "talgo.h" #include "talgo.h"
#include "tchecksum.h" #include "tchecksum.h"

2
src/tsdb/src/tsdbRWHelper.c
View File

@ -14,9 +14,7 @@
*/ */
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#define TAOS_RANDOM_FILE_FAIL_TEST #define TAOS_RANDOM_FILE_FAIL_TEST
#include "os.h" #include "os.h"
#include "talgo.h" #include "talgo.h"
#include "tchecksum.h" #include "tchecksum.h"

2
src/util/src/tkvstore.c
View File

@ -14,9 +14,7 @@
*/ */
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#define TAOS_RANDOM_FILE_FAIL_TEST #define TAOS_RANDOM_FILE_FAIL_TEST
#include "os.h" #include "os.h"
#include "hash.h" #include "hash.h"
#include "taoserror.h" #include "taoserror.h"

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

@ -255,8 +255,8 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->fversion = pVnode->version; pVnode->fversion = pVnode->version;
pVnode->wqueue = dnodeAllocateVnodeWqueue(pVnode); pVnode->wqueue = dnodeAllocVWriteQueue(pVnode);
pVnode->rqueue = dnodeAllocateVnodeRqueue(pVnode); pVnode->rqueue = dnodeAllocVReadQueue(pVnode);
if (pVnode->wqueue == NULL || pVnode->rqueue == NULL) { if (pVnode->wqueue == NULL || pVnode->rqueue == NULL) {
vnodeCleanUp(pVnode); vnodeCleanUp(pVnode);
return terrno; return terrno;
@ -322,7 +322,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
syncInfo.getWalInfo = vnodeGetWalInfo; syncInfo.getWalInfo = vnodeGetWalInfo;
syncInfo.getFileInfo = vnodeGetFileInfo; syncInfo.getFileInfo = vnodeGetFileInfo;
syncInfo.writeToCache = vnodeWriteToQueue; syncInfo.writeToCache = vnodeWriteToQueue;
syncInfo.confirmForward = dnodeSendRpcVnodeWriteRsp; syncInfo.confirmForward = dnodeSendRpcVWriteRsp;
syncInfo.notifyRole = vnodeNotifyRole; syncInfo.notifyRole = vnodeNotifyRole;
syncInfo.notifyFlowCtrl = vnodeCtrlFlow; syncInfo.notifyFlowCtrl = vnodeCtrlFlow;
syncInfo.notifyFileSynced = vnodeNotifyFileSynced; syncInfo.notifyFileSynced = vnodeNotifyFileSynced;
@ -409,12 +409,12 @@ void vnodeRelease(void *pVnodeRaw) {
} }
if (pVnode->wqueue) { if (pVnode->wqueue) {
dnodeFreeVnodeWqueue(pVnode->wqueue); dnodeFreeVWriteQueue(pVnode->wqueue);
pVnode->wqueue = NULL; pVnode->wqueue = NULL;
} }
if (pVnode->rqueue) { if (pVnode->rqueue) {
dnodeFreeVnodeRqueue(pVnode->rqueue); dnodeFreeVReadQueue(pVnode->rqueue);
pVnode->rqueue = NULL; pVnode->rqueue = NULL;
} }

2
src/wal/src/walMgmt.c
View File

@ -55,7 +55,7 @@ void walCleanUp() {
} }
void *walOpen(char *path, SWalCfg *pCfg) { void *walOpen(char *path, SWalCfg *pCfg) {
SWal *pWal = tcalloc(sizeof(SWal)); SWal *pWal = tcalloc(1, sizeof(SWal));
if (pWal == NULL) { if (pWal == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
return NULL; return NULL;

1
src/wal/src/walWrite.c
View File

@ -14,6 +14,7 @@
*/ */
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#define TAOS_RANDOM_FILE_FAIL_TEST
#include "os.h" #include "os.h"
#include "taoserror.h" #include "taoserror.h"
#include "tchecksum.h" #include "tchecksum.h"