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

TD-1617

This commit is contained in:
Shengliang Guan 2020-09-27 15:34:44 +00:00
parent 6d04062576
commit 2caeccf623
8 changed files with 336 additions and 348 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/sync/inc/syncInt.h
View File

@ -171,7 +171,6 @@ void syncBroadcastStatus(SSyncNode *pNode);
void syncAddPeerRef(SSyncPeer *pPeer);
int syncDecPeerRef(SSyncPeer *pPeer);
#ifdef __cplusplus
}
#endif

1
src/sync/inc/taosTcpPool.h
View File

@ -38,7 +38,6 @@ void taosCloseTcpThreadPool(ttpool_h);
void *taosAllocateTcpConn(void *, void *ahandle, int connFd);
void taosFreeTcpConn(void *);
#ifdef __cplusplus
}
#endif

14
src/sync/src/syncMain.c
View File

@ -178,7 +178,8 @@ void *syncStart(const SSyncInfo *pInfo) {
nodeVersion = pInfo->version; // set the initial version
nodeRole = (pNode->replica > 1) ? TAOS_SYNC_ROLE_UNSYNCED : TAOS_SYNC_ROLE_MASTER;
sInfo("vgId:%d, %d replicas are configured, quorum:%d role:%s", pNode->vgId, pNode->replica, pNode->quorum, syncRole[nodeRole]);
sInfo("vgId:%d, %d replicas are configured, quorum:%d role:%s", pNode->vgId, pNode->replica, pNode->quorum,
syncRole[nodeRole]);
pNode->pSyncFwds = calloc(sizeof(SSyncFwds) + tsMaxFwdInfo * sizeof(SFwdInfo), 1);
if (pNode->pSyncFwds == NULL) {
@ -443,9 +444,7 @@ static void syncAddArbitrator(SSyncNode *pNode) {
pNode->peerInfo[TAOS_SYNC_MAX_REPLICA] = syncAddPeer(pNode, &nodeInfo);
}
static void syncAddNodeRef(SSyncNode *pNode) {
atomic_add_fetch_8(&pNode->refCount, 1);
}
static void syncAddNodeRef(SSyncNode *pNode) { atomic_add_fetch_8(&pNode->refCount, 1); }
static void syncDecNodeRef(SSyncNode *pNode) {
if (atomic_sub_fetch_8(&pNode->refCount, 1) == 0) {
@ -456,9 +455,7 @@ static void syncDecNodeRef(SSyncNode *pNode) {
}
}
void syncAddPeerRef(SSyncPeer *pPeer) {
atomic_add_fetch_8(&pPeer->refCount, 1);
}
void syncAddPeerRef(SSyncPeer *pPeer) { atomic_add_fetch_8(&pPeer->refCount, 1); }
int syncDecPeerRef(SSyncPeer *pPeer) {
if (atomic_sub_fetch_8(&pPeer->refCount, 1) == 0) {
@ -501,6 +498,7 @@ static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo) {
tstrncpy(pPeer->fqdn, pInfo->nodeFqdn, sizeof(pPeer->fqdn));
pPeer->ip = ip;
pPeer->port = pInfo->nodePort;
pPeer->fqdn[sizeof(pPeer->fqdn) - 1] = 0;
snprintf(pPeer->id, sizeof(pPeer->id), "vgId:%d peer:%s:%d", pNode->vgId, pPeer->fqdn, pPeer->port);
pPeer->peerFd = -1;
@ -877,8 +875,6 @@ static void syncProcessForwardFromPeer(char *cont, SSyncPeer *pPeer) {
sError("%s, forward discarded, ver:%" PRIu64, pPeer->id, pHead->version);
}
}
return;
}
static void syncProcessPeersStatusMsg(char *cont, SSyncPeer *pPeer) {

19
src/sync/src/syncRestore.c
View File

@ -23,7 +23,7 @@
#include "tsync.h"
#include "syncInt.h"
static void syncRemoveExtraFile(SSyncPeer *pPeer, uint32_t sindex, uint32_t eindex) {
static void syncRemoveExtraFile(SSyncPeer *pPeer, int32_t sindex, int32_t eindex) {
char name[TSDB_FILENAME_LEN * 2] = {0};
char fname[TSDB_FILENAME_LEN * 3] = {0};
uint32_t magic;
@ -40,7 +40,7 @@ static void syncRemoveExtraFile(SSyncPeer *pPeer, uint32_t sindex, uint32_t eind
if (magic == 0) break;
snprintf(fname, sizeof(fname), "%s/%s", pNode->path, name);
remove(fname);
(void)remove(fname);
sDebug("%s, %s is removed", pPeer->id, fname);
index++;
@ -81,7 +81,8 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
// check the file info
sinfo = minfo;
sDebug("%s, get file info:%s", pPeer->id, minfo.name);
sinfo.magic = (*pNode->getFileInfo)(pNode->ahandle, sinfo.name, &sinfo.index, TAOS_SYNC_MAX_INDEX, &sinfo.size, &sinfo.fversion);
sinfo.magic = (*pNode->getFileInfo)(pNode->ahandle, sinfo.name, &sinfo.index, TAOS_SYNC_MAX_INDEX, &sinfo.size,
&sinfo.fversion);
// if file not there or magic is not the same, file shall be synced
memset(&fileAck, 0, sizeof(fileAck));
@ -99,6 +100,7 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
// if sync is required, open file, receive from master, and write to file
// get the full path to file
minfo.name[sizeof(minfo.name) - 1] = 0;
snprintf(name, sizeof(name), "%s/%s", pNode->path, minfo.name);
int dfd = open(name, O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU | S_IRWXG | S_IRWXO);
@ -113,7 +115,6 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
if (ret < 0) break;
sDebug("%s, %s is received, size:%" PRId64, pPeer->id, minfo.name, minfo.size);
}
if (code == 0 && (minfo.fversion != sinfo.fversion)) {
@ -142,7 +143,10 @@ static int syncRestoreWal(SSyncPeer *pPeer) {
ret = taosReadMsg(pPeer->syncFd, pHead, sizeof(SWalHead));
if (ret < 0) break;
if (pHead->len == 0) {code = 0; break;} // wal sync over
if (pHead->len == 0) {
code = 0;
break;
} // wal sync over
ret = taosReadMsg(pPeer->syncFd, pHead->cont, pHead->len);
if (ret < 0) break;
@ -159,8 +163,7 @@ static int syncRestoreWal(SSyncPeer *pPeer) {
return code;
}
static char *syncProcessOneBufferedFwd(SSyncPeer *pPeer, char *offset)
{
static char *syncProcessOneBufferedFwd(SSyncPeer *pPeer, char *offset) {
SSyncNode *pNode = pPeer->pSyncNode;
SWalHead * pHead = (SWalHead *)offset;
@ -311,7 +314,7 @@ void *syncRestoreData(void *param) {
(*pNode->notifyRole)(pNode->ahandle, nodeRole);
nodeSStatus = TAOS_SYNC_STATUS_INIT;
taosClose(pPeer->syncFd)
taosClose(pPeer->syncFd);
syncCloseRecvBuffer(pNode);
__sync_fetch_and_sub(&tsSyncNum, 1);
syncDecPeerRef(pPeer);

29
src/sync/src/syncRetrieve.c
View File

@ -110,7 +110,8 @@ static int syncRetrieveFile(SSyncPeer *pPeer) {
while (1) {
// retrieve file info
fileInfo.name[0] = 0;
fileInfo.magic = (*pNode->getFileInfo)(pNode->ahandle, fileInfo.name, &fileInfo.index, TAOS_SYNC_MAX_INDEX, &fileInfo.size, &fileInfo.fversion);
fileInfo.magic = (*pNode->getFileInfo)(pNode->ahandle, fileInfo.name, &fileInfo.index, TAOS_SYNC_MAX_INDEX,
&fileInfo.size, &fileInfo.fversion);
// fileInfo.size = htonl(size);
// send the file info
@ -120,7 +121,8 @@ static int syncRetrieveFile(SSyncPeer *pPeer) {
// if no file anymore, break
if (fileInfo.magic == 0 || fileInfo.name[0] == 0) {
sDebug("%s, no more files to sync", pPeer->id);
code = 0; break;
code = 0;
break;
}
// wait for the ack from peer
@ -219,7 +221,7 @@ static int syncMonitorLastWal(SSyncPeer *pPeer, char *name) {
return 0;
}
static uint32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
static int32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
char buf[2048];
int len = read(pPeer->notifyFd, buf, sizeof(buf));
if (len < 0 && errno != EAGAIN) {
@ -236,21 +238,24 @@ static uint32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
if (event->mask & IN_CLOSE_WRITE) *pEvent = *pEvent | IN_CLOSE_WRITE;
}
if (pEvent != 0)
sDebug("%s, last wal event:0x%x", pPeer->id, *pEvent);
if (pEvent != 0) sDebug("%s, last wal event:0x%x", pPeer->id, *pEvent);
return 0;
}
static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion, int64_t offset, uint32_t *pEvent) {
SWalHead *pHead = (SWalHead *) malloc(640000);
SWalHead *pHead = malloc(640000);
int code = -1;
int32_t bytes = 0;
int sfd;
sfd = open(name, O_RDONLY);
if (sfd < 0) return -1;
lseek(sfd, offset, SEEK_SET);
if (sfd < 0) {
free(pHead);
return -1;
}
(void)lseek(sfd, offset, SEEK_SET);
sDebug("%s, retrieve last wal, offset:%" PRId64 " fversion:%" PRIu64, pPeer->id, offset, fversion);
while (1) {
@ -276,7 +281,7 @@ static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion,
}
free(pHead);
taosClose(sfd);
close(sfd);
if (code == 0) return bytes;
return -1;
@ -447,8 +452,7 @@ static int syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
}
// if no files are synced, there must be wal to sync, sversion must be larger than one
if (pPeer->sversion == 0)
pPeer->sversion = 1;
if (pPeer->sversion == 0) pPeer->sversion = 1;
sDebug("%s, start to retrieve wal", pPeer->id);
if (syncRetrieveWal(pPeer) < 0) {
@ -486,8 +490,7 @@ void *syncRetrieveData(void *param) {
(*pNode->notifyFlowCtrl)(pNode->ahandle, 4 << (pPeer->numOfRetrieves - 2));
} else {
pPeer->numOfRetrieves = 0;
if (pNode->notifyFlowCtrl)
(*pNode->notifyFlowCtrl)(pNode->ahandle, 0);
if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->ahandle, 0);
}
pPeer->fileChanged = 0;

15
src/sync/src/taosTcpPool.c
View File

@ -45,8 +45,8 @@ typedef struct {
static void *taosAcceptPeerTcpConnection(void *argv);
static void *taosProcessTcpData(void *param);
static SThreadObj *taosGetTcpThread(SPoolObj *pPool);
static void taosStopPoolThread(SThreadObj *pThread);
static SThreadObj *taosGetTcpThread(SPoolObj *pPool);
void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
pthread_attr_t thattr;
@ -68,7 +68,8 @@ void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
pPool->acceptFd = taosOpenTcpServerSocket(pInfo->serverIp, pInfo->port);
if (pPool->acceptFd < 0) {
free(pPool->pThread); free(pPool);
free(pPool->pThread);
free(pPool);
uError("failed to create TCP server socket, port:%d (%s)", pInfo->port, strerror(errno));
return NULL;
}
@ -78,7 +79,8 @@ void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
if (pthread_create(&(pPool->thread), &thattr, (void *)taosAcceptPeerTcpConnection, pPool) != 0) {
uError("TCP server, failed to create accept thread, reason:%s", strerror(errno));
close(pPool->acceptFd);
free(pPool->pThread); free(pPool);
free(pPool->pThread);
free(pPool);
return NULL;
}
@ -100,9 +102,10 @@ void taosCloseTcpThreadPool(void *param) {
if (pThread) taosStopPoolThread(pThread);
}
uDebug("%p TCP pool is closed", pPool);
taosTFree(pPool->pThread);
free(pPool);
uDebug("%p TCP pool is closed", pPool);
}
void *taosAllocateTcpConn(void *param, void *pPeer, int connFd) {
@ -219,10 +222,11 @@ static void *taosProcessTcpData(void *param) {
}
}
uDebug("%p TCP epoll thread exits", pThread);
close(pThread->pollFd);
free(pThread);
free(buffer);
uDebug("%p TCP epoll thread exits", pThread);
return NULL;
}
@ -319,4 +323,3 @@ static void taosStopPoolThread(SThreadObj* pThread) {
pthread_join(thread, NULL);
taosClose(fd);
}

10
src/sync/test/syncClient.c
View File

@ -33,7 +33,8 @@ typedef struct {
void processResponse(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
SInfo *pInfo = (SInfo *)pMsg->ahandle;
uDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen, pMsg->code);
uDebug("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);
@ -57,8 +58,9 @@ void *sendRequest(void *param) {
rpcMsg.msgType = 1;
uDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg);
if ( pInfo->num % 20000 == 0 )
if (pInfo->num % 20000 == 0) {
uInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num);
}
tsem_wait(&pInfo->rspSem);
}
@ -106,7 +108,7 @@ int main(int argc, char *argv[]) {
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) {
strcpy(epSet.fqdn[0], argv[++i]);
tstrncpy(epSet.fqdn[0], argv[++i], TSDB_FQDN_LEN);
} 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) {
@ -190,5 +192,3 @@ int main(int argc, char *argv[]) {
return 0;
}

59
src/sync/test/syncServer.c
View File

@ -38,12 +38,11 @@ int numOfWrites ;
SSyncInfo syncInfo;
SSyncCfg *pCfg;
int writeIntoWal(SWalHead *pHead)
{
int writeIntoWal(SWalHead *pHead) {
if (dataFd < 0) {
char walName[280];
snprintf(walName, sizeof(walName), "%s/wal/wal.%d", path, walNum);
remove(walName);
(void)remove(walName);
dataFd = open(walName, O_CREAT | O_WRONLY, S_IRWXU | S_IRWXG | S_IRWXO);
if (dataFd < 0) {
uInfo("failed to open wal file:%s(%s)", walName, strerror(errno));
@ -71,8 +70,7 @@ int writeIntoWal(SWalHead *pHead)
return 0;
}
void confirmForward(void *ahandle, void *mhandle, int32_t code)
{
void confirmForward(void *ahandle, void *mhandle, int32_t code) {
SRpcMsg * pMsg = (SRpcMsg *)mhandle;
SWalHead *pHead = (SWalHead *)(((char *)pMsg->pCont) - sizeof(SWalHead));
@ -80,7 +78,7 @@ void confirmForward(void *ahandle, void *mhandle, int32_t code)
rpcFreeCont(pMsg->pCont);
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = rpcMallocCont(msgSize);
rpcMsg.contLen = msgSize;
rpcMsg.handle = pMsg->handle;
@ -98,7 +96,6 @@ int processRpcMsg(void *item) {
if (role != TAOS_SYNC_ROLE_MASTER) {
uError("not master, write failed, role:%s", syncRole[role]);
} else {
pHead->version = ++tversion;
pHead->msgType = pMsg->msgType;
pHead->len = pMsg->contLen;
@ -111,10 +108,10 @@ int processRpcMsg(void *item) {
}
if (pCfg->quorum <= 1) {
taosFreeQitem(item);
rpcFreeCont(pMsg->pCont);
taosFreeQitem(item);
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = rpcMallocCont(msgSize);
rpcMsg.contLen = msgSize;
rpcMsg.handle = pMsg->handle;
@ -126,7 +123,6 @@ int processRpcMsg(void *item) {
}
int processFwdMsg(void *item) {
SWalHead *pHead = (SWalHead *)item;
if (pHead->version <= tversion) {
@ -154,7 +150,6 @@ int processFwdMsg(void *item) {
}
int processWalMsg(void *item) {
SWalHead *pHead = (SWalHead *)item;
if (pHead->version <= tversion) {
@ -197,7 +192,6 @@ void *processWriteQueue(void *param) {
} else if (type == TAOS_QTYPE_FWD) {
processFwdMsg(item);
}
}
return NULL;
@ -224,7 +218,6 @@ int retrieveAuthInfo(char *meterId, char *spi, char *encrypt, char *secret, char
}
void processRequestMsg(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
SRpcMsg *pTemp;
pTemp = taosAllocateQitem(sizeof(SRpcMsg));
@ -234,8 +227,7 @@ void processRequestMsg(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
taosWriteQitem(qhandle, TAOS_QTYPE_RPC, pTemp);
}
uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int64_t *size, uint64_t *fversion)
{
uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int64_t *size, uint64_t *fversion) {
uint32_t magic;
struct stat fstat;
char aname[280];
@ -263,7 +255,6 @@ uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex
}
int getWalInfo(void *ahandle, char *name, uint32_t *index) {
struct stat fstat;
char aname[280];
@ -279,7 +270,6 @@ int getWalInfo(void *ahandle, char *name, uint32_t *index) {
if (*index >= walNum - 1) return 0; // no more
return 1;
}
int writeToCache(void *ahandle, void *data, int type) {
@ -295,19 +285,14 @@ int writeToCache(void *ahandle, void *data, int type) {
return 0;
}
void confirmFwd(void *ahandle, int64_t version) {
return;
}
void confirmFwd(void *ahandle, int64_t version) { return; }
void notifyRole(void *ahandle, int8_t r) {
role = r;
printf("current role:%s\n", syncRole[role]);
}
void initSync() {
pCfg->replica = 1;
pCfg->quorum = 1;
syncInfo.vgId = 1;
@ -339,15 +324,13 @@ void initSync() {
taosGetFqdn(pCfg->nodeInfo[4].nodeFqdn);
}
void doSync()
{
void doSync() {
for (int i = 0; i < 5; ++i) {
if (tsSyncPort == pCfg->nodeInfo[i].nodePort)
nodeId = pCfg->nodeInfo[i].nodeId;
if (tsSyncPort == pCfg->nodeInfo[i].nodePort) nodeId = pCfg->nodeInfo[i].nodeId;
}
snprintf(path, sizeof(path), "/root/test/d%d", nodeId);
strcpy(syncInfo.path, path);
tstrncpy(syncInfo.path, path, sizeof(syncInfo.path));
if (syncHandle == NULL) {
syncHandle = syncStart(&syncInfo);
@ -443,23 +426,28 @@ int main(int argc, char *argv[]) {
SNodesRole nroles;
while (1) {
char c = getchar();
int c = getchar();
switch (c) {
case '1':
pCfg->replica = 1; doSync();
pCfg->replica = 1;
doSync();
break;
case '2':
pCfg->replica = 2; doSync();
pCfg->replica = 2;
doSync();
break;
case '3':
pCfg->replica = 3; doSync();
pCfg->replica = 3;
doSync();
break;
case '4':
pCfg->replica = 4; doSync();
pCfg->replica = 4;
doSync();
break;
case '5':
pCfg->replica = 5; doSync();
pCfg->replica = 5;
doSync();
break;
case 's':
syncGetNodesRole(syncHandle, &nroles);
@ -471,7 +459,6 @@ int main(int argc, char *argv[]) {
}
if (c == 'q') break;
}
syncStop(syncHandle);
@ -483,5 +470,3 @@ int main(int argc, char *argv[]) {
return 0;
}