Skip to content

Commit

Permalink
Merge pull request taosdata#15521 from shiyi23/hsy_dev_newest
Browse files Browse the repository at this point in the history
fix: Cover Scan
  • Loading branch information
guanshengliang authored Oct 24, 2022
2 parents 77ca715 + e2cbf7c commit 3ef0c27
Show file tree
Hide file tree
Showing 2 changed files with 43 additions and 14 deletions.
2 changes: 1 addition & 1 deletion src/inc/tsync.h
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ typedef enum {
typedef struct {
uint32_t nodeId; // node ID assigned by TDengine
uint16_t nodePort; // node sync Port
char nodeFqdn[TSDB_FQDN_LEN]; // node FQDN
char nodeFqdn[TSDB_FQDN_LEN * 2]; // node FQDN
} SNodeInfo;

typedef struct {
Expand Down
55 changes: 42 additions & 13 deletions src/sync/src/syncMain.c
Original file line number Diff line number Diff line change
Expand Up @@ -418,8 +418,12 @@ void syncRecover(int64_t rid) {
nodeRole = TAOS_SYNC_ROLE_UNSYNCED;
(*pNode->notifyRoleFp)(pNode->vgId, nodeRole);

pthread_mutex_lock(&pNode->mutex);

// pthread_mutex_lock(&pNode->mutex);
int err = pthread_mutex_trylock(&pNode->mutex);
if(EBUSY == err)
{
sDebug("pthread_mutex_trylock failed!");
}
nodeVersion = 0;

for (int32_t i = 0; i < pNode->replica; ++i) {
Expand All @@ -431,7 +435,7 @@ void syncRecover(int64_t rid) {
}
}

pthread_mutex_unlock(&pNode->mutex);
if(0 == err) pthread_mutex_unlock(&pNode->mutex);

syncReleaseNode(pNode);
}
Expand Down Expand Up @@ -987,12 +991,17 @@ static void syncNotStarted(void *param, void *tmrId) {

SSyncNode *pNode = pPeer->pSyncNode;

pthread_mutex_lock(&pNode->mutex);
// pthread_mutex_lock(&pNode->mutex);
int err = pthread_mutex_trylock(&pNode->mutex);
if(EBUSY == err)
{
sDebug("pthread_mutex_trylock failed!");
}
pPeer->timer = NULL;
pPeer->sstatus = TAOS_SYNC_STATUS_INIT;
sInfo("%s, sync conn is still not up, restart and set sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
syncRestartConnection(pPeer);
pthread_mutex_unlock(&pNode->mutex);
if(0 == err) pthread_mutex_unlock(&pNode->mutex);

syncReleasePeer(pPeer);
}
Expand Down Expand Up @@ -1136,7 +1145,12 @@ static int32_t syncProcessPeerMsg(int64_t rid, void *buffer) {
SSyncHead *pHead = buffer;
SSyncNode *pNode = pPeer->pSyncNode;

pthread_mutex_lock(&pNode->mutex);
// pthread_mutex_lock(&pNode->mutex);
int err = pthread_mutex_trylock(&pNode->mutex);
if(EBUSY == err)
{
sDebug("pthread_mutex_trylock failed!");
}

int32_t code = syncReadPeerMsg(pPeer, pHead);

Expand All @@ -1152,7 +1166,7 @@ static int32_t syncProcessPeerMsg(int64_t rid, void *buffer) {
}
}

pthread_mutex_unlock(&pNode->mutex);
if(0 == err) pthread_mutex_unlock(&pNode->mutex);
syncReleasePeer(pPeer);

return code;
Expand Down Expand Up @@ -1245,12 +1259,17 @@ static void syncCheckPeerConnection(void *param, void *tmrId) {

SSyncNode *pNode = pPeer->pSyncNode;

pthread_mutex_lock(&pNode->mutex);
// pthread_mutex_lock(&pNode->mutex);
int err = pthread_mutex_trylock(&pNode->mutex);
if(EBUSY == err)
{
sDebug("pthread_mutex_trylock failed!");
}

sDebug("%s, check peer connection", pPeer->id);
syncSetupPeerConnection(pPeer);

pthread_mutex_unlock(&pNode->mutex);
if(0 == err) pthread_mutex_unlock(&pNode->mutex);

syncReleasePeer(pPeer);
}
Expand Down Expand Up @@ -1331,7 +1350,12 @@ static void syncProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp) {
sDebug("vgId:%d, sync connection is incoming, tranId:%u", vgId, msg.tranId);

SSyncNode *pNode = *ppNode;
pthread_mutex_lock(&pNode->mutex);
// pthread_mutex_lock(&pNode->mutex);
int err = pthread_mutex_trylock(&pNode->mutex);
if(EBUSY == err)
{
sDebug("pthread_mutex_trylock failed!");
}

SSyncPeer *pPeer;
for (i = 0; i < pNode->replica; ++i) {
Expand Down Expand Up @@ -1362,7 +1386,7 @@ static void syncProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp) {
}
}

pthread_mutex_unlock(&pNode->mutex);
if(0 == err) pthread_mutex_unlock(&pNode->mutex);
}

static void syncProcessBrokenLink(int64_t rid, int32_t closedByApp) {
Expand All @@ -1371,7 +1395,12 @@ static void syncProcessBrokenLink(int64_t rid, int32_t closedByApp) {

SSyncNode *pNode = pPeer->pSyncNode;

pthread_mutex_lock(&pNode->mutex);
// pthread_mutex_lock(&pNode->mutex);
int err = pthread_mutex_trylock(&pNode->mutex);
if(EBUSY == err)
{
sDebug("pthread_mutex_trylock failed!");
}

sDebug("%s, TCP link is broken since %s, pfd:%d sfd:%d closedByApp:%d",
pPeer->id, strerror(errno), pPeer->peerFd, pPeer->syncFd, closedByApp);
Expand All @@ -1381,7 +1410,7 @@ static void syncProcessBrokenLink(int64_t rid, int32_t closedByApp) {
}

syncRestartConnection(pPeer);
pthread_mutex_unlock(&pNode->mutex);
if (0 == err) pthread_mutex_unlock(&pNode->mutex);

syncReleasePeer(pPeer);
}
Expand Down

0 comments on commit 3ef0c27

Please sign in to comment.