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

fix/TD-32838-add-writer-lock

This commit is contained in:
dmchen 2024-11-11 12:00:09 +08:00
parent 82cf87369a
commit aedeca9ecb
2 changed files with 24 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/sync/inc/syncSnapshot.h
View File

@ -91,6 +91,7 @@ typedef struct SSyncSnapshotReceiver {
// update when begin // update when begin
void *pWriter; void *pWriter;
TdThreadMutex writerMutex;
SSnapshotParam snapshotParam; SSnapshotParam snapshotParam;
SSnapshot snapshot; SSnapshot snapshot;

32
source/libs/sync/src/syncSnapshot.c
View File

@ -429,6 +429,7 @@ int32_t snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId, SSyncSnapsh
pReceiver->startTime = 0; pReceiver->startTime = 0;
pReceiver->ack = SYNC_SNAPSHOT_SEQ_BEGIN; pReceiver->ack = SYNC_SNAPSHOT_SEQ_BEGIN;
pReceiver->pWriter = NULL; pReceiver->pWriter = NULL;
(void)taosThreadMutexInit(&pReceiver->writerMutex, NULL);
pReceiver->pSyncNode = pSyncNode; pReceiver->pSyncNode = pSyncNode;
pReceiver->fromId = fromId; pReceiver->fromId = fromId;
pReceiver->term = raftStoreGetTerm(pSyncNode); pReceiver->term = raftStoreGetTerm(pSyncNode);
@ -468,6 +469,7 @@ static int32_t snapshotReceiverClearInfoData(SSyncSnapshotReceiver *pReceiver) {
void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) { void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) {
if (pReceiver == NULL) return; if (pReceiver == NULL) return;
(void)taosThreadMutexLock(&pReceiver->writerMutex);
// close writer // close writer
if (pReceiver->pWriter != NULL) { if (pReceiver->pWriter != NULL) {
int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter,
@ -478,6 +480,9 @@ void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) {
} }
pReceiver->pWriter = NULL; pReceiver->pWriter = NULL;
} }
(void)taosThreadMutexUnlock(&pReceiver->writerMutex);
(void)taosThreadMutexDestroy(&pReceiver->writerMutex);
// free snap buf // free snap buf
if (pReceiver->pRcvBuf) { if (pReceiver->pRcvBuf) {
@ -556,7 +561,8 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
int8_t stopped = !atomic_val_compare_exchange_8(&pReceiver->start, true, false); int8_t stopped = !atomic_val_compare_exchange_8(&pReceiver->start, true, false);
if (stopped) return; if (stopped) return;
(void)taosThreadMutexLock(&pReceiver->pRcvBuf->mutex);
(void)taosThreadMutexLock(&pReceiver->writerMutex);
{ {
if (pReceiver->pWriter != NULL) { if (pReceiver->pWriter != NULL) {
int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter,
@ -568,7 +574,11 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
} else { } else {
sRInfo(pReceiver, "snapshot receiver stop, writer is null"); sRInfo(pReceiver, "snapshot receiver stop, writer is null");
} }
}
(void)taosThreadMutexUnlock(&pReceiver->writerMutex);
(void)taosThreadMutexLock(&pReceiver->pRcvBuf->mutex);
{
syncSnapBufferReset(pReceiver->pRcvBuf); syncSnapBufferReset(pReceiver->pRcvBuf);
(void)snapshotReceiverClearInfoData(pReceiver); (void)snapshotReceiverClearInfoData(pReceiver);
@ -600,15 +610,19 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
raftStoreSetTerm(pReceiver->pSyncNode, pReceiver->snapshot.lastApplyTerm); raftStoreSetTerm(pReceiver->pSyncNode, pReceiver->snapshot.lastApplyTerm);
} }
// stop writer, apply data (void)taosThreadMutexLock(&pReceiver->writerMutex);
code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true, if (pReceiver->pWriter != NULL) {
&pReceiver->snapshot); // stop writer, apply data
if (code != 0) { code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true,
sRError(pReceiver, "snapshot receiver apply failed since %s", tstrerror(code)); &pReceiver->snapshot);
TAOS_RETURN(code); if (code != 0) {
sRError(pReceiver, "snapshot receiver apply failed since %s", tstrerror(code));
TAOS_RETURN(code);
}
pReceiver->pWriter = NULL;
sRInfo(pReceiver, "snapshot receiver write stopped");
} }
pReceiver->pWriter = NULL; (void)taosThreadMutexUnlock(&pReceiver->writerMutex);
sRInfo(pReceiver, "snapshot receiver write stopped");
// update progress // update progress
pReceiver->ack = SYNC_SNAPSHOT_SEQ_END; pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;