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

TD-2616

This commit is contained in:
Shengliang Guan 2020-12-30 14:43:50 +08:00
parent e7d5499372
commit 44b170427d
15 changed files with 84 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/client/src/tscUtil.c
View File

@ -2503,7 +2503,7 @@ bool tscSetSqlOwner(SSqlObj* pSql) {
SSqlRes* pRes = &pSql->res;
// set the sql object owner
uint64_t threadId = taosGetPthreadId();
uint64_t threadId = taosGetSelfPthreadId();
if (atomic_val_compare_exchange_64(&pSql->owner, 0, threadId) != 0) {
pRes->code = TSDB_CODE_QRY_IN_EXEC;
return false;

7
src/os/inc/osSemphone.h
View File

@ -30,11 +30,12 @@ extern "C" {
// TAOS_OS_FUNC_SEMPHONE_PTHREAD
bool taosCheckPthreadValid(pthread_t thread);
int64_t taosGetPthreadId();
void taosResetPthread(pthread_t *thread);
int64_t taosGetSelfPthreadId();
int64_t taosGetPthreadId(pthread_t thread);
void taosResetPthread(pthread_t* thread);
bool taosComparePthread(pthread_t first, pthread_t second);
int32_t taosGetPId();
int32_t taosGetCurrentAPPName(char *name, int32_t* len);
int32_t taosGetCurrentAPPName(char* name, int32_t* len);
#ifdef __cplusplus
}

3
src/os/src/detail/osSemphone.c
View File

@ -31,7 +31,8 @@ int tsem_wait(tsem_t* sem) {
#ifndef TAOS_OS_FUNC_SEMPHONE_PTHREAD
bool taosCheckPthreadValid(pthread_t thread) { return thread != 0; }
int64_t taosGetPthreadId() { return (int64_t)pthread_self(); }
int64_t taosGetSelfPthreadId() { return (int64_t)pthread_self(); }
int64_t taosGetPthreadId(pthread_t thread) { return (int64_t)thread; }
void taosResetPthread(pthread_t *thread) { *thread = 0; }
bool taosComparePthread(pthread_t first, pthread_t second) { return first == second; }
int32_t taosGetPId() { return getpid(); }

8
src/os/src/windows/wSemphone.c
View File

@ -25,14 +25,16 @@ bool taosCheckPthreadValid(pthread_t thread) { return thread.p != NULL; }
void taosResetPthread(pthread_t *thread) { thread->p = 0; }
int64_t taosGetPthreadId() {
int64_t taosGetPthreadId(pthread_t thread) {
#ifdef PTW32_VERSION
return pthread_getw32threadid_np(pthread_self());
return pthread_getw32threadid_np(thread);
#else
return (int64_t)pthread_self();
return (int64_t)thread;
#endif
}
int64_t taosGetSelfPthreadId() { return taosGetPthreadId(pthread_self()); }
bool taosComparePthread(pthread_t first, pthread_t second) {
return first.p == second.p;
}

4
src/query/src/qExecutor.c
View File

@ -7250,7 +7250,7 @@ static bool doBuildResCheck(SQInfo* pQInfo) {
// clear qhandle owner, it must be in the secure area. other thread may run ahead before current, after it is
// put into task to be executed.
assert(pQInfo->owner == taosGetPthreadId());
assert(pQInfo->owner == taosGetSelfPthreadId());
pQInfo->owner = 0;
pthread_mutex_unlock(&pQInfo->lock);
@ -7263,7 +7263,7 @@ static bool doBuildResCheck(SQInfo* pQInfo) {
bool qTableQuery(qinfo_t qinfo) {
SQInfo *pQInfo = (SQInfo *)qinfo;
assert(pQInfo && pQInfo->signature == pQInfo);
int64_t threadId = taosGetPthreadId();
int64_t threadId = taosGetSelfPthreadId();
int64_t curOwner = 0;
if ((curOwner = atomic_val_compare_exchange_64(&pQInfo->owner, 0, threadId)) != 0) {

4
src/rpc/src/rpcCache.c
View File

@ -272,7 +272,7 @@ static int rpcHashConn(void *handle, char *fqdn, uint16_t port, int8_t connType)
}
static void rpcLockCache(int64_t *lockedBy) {
int64_t tid = taosGetPthreadId();
int64_t tid = taosGetSelfPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(lockedBy, 0, tid) != 0) {
if (++i % 100 == 0) {
@ -282,7 +282,7 @@ static void rpcLockCache(int64_t *lockedBy) {
}
static void rpcUnlockCache(int64_t *lockedBy) {
int64_t tid = taosGetPthreadId();
int64_t tid = taosGetSelfPthreadId();
if (atomic_val_compare_exchange_64(lockedBy, tid, 0) != tid) {
assert(false);
}

4
src/rpc/src/rpcMain.c
View File

@ -1604,7 +1604,7 @@ static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen) {
}
static void rpcLockConn(SRpcConn *pConn) {
int64_t tid = taosGetPthreadId();
int64_t tid = taosGetSelfPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(&(pConn->lockedBy), 0, tid) != 0) {
if (++i % 1000 == 0) {
@ -1614,7 +1614,7 @@ static void rpcLockConn(SRpcConn *pConn) {
}
static void rpcUnlockConn(SRpcConn *pConn) {
int64_t tid = taosGetPthreadId();
int64_t tid = taosGetSelfPthreadId();
if (atomic_val_compare_exchange_64(&(pConn->lockedBy), tid, 0) != tid) {
assert(false);
}

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

@ -478,9 +478,7 @@ static void syncAddArbitrator(SSyncNode *pNode) {
static void syncFreeNode(void *param) {
SSyncNode *pNode = param;
int32_t refCount = atomic_sub_fetch_32(&pNode->refCount, 1);
sDebug("vgId:%d, syncnode is freed, refCount:%d", pNode->vgId, refCount);
sDebug("vgId:%d, node is freed, refCount:%d", pNode->vgId, pNode->refCount);
pthread_mutex_destroy(&pNode->mutex);
tfree(pNode->pRecv);
@ -491,10 +489,10 @@ static void syncFreeNode(void *param) {
SSyncNode *syncAcquireNode(int64_t rid) {
SSyncNode *pNode = taosAcquireRef(tsNodeRefId, rid);
if (pNode == NULL) {
sDebug("failed to acquire syncnode from refId:%" PRId64, rid);
sDebug("failed to acquire node from refId:%" PRId64, rid);
} else {
int32_t refCount = atomic_add_fetch_32(&pNode->refCount, 1);
sTrace("vgId:%d, acquire syncnode refId:%" PRId64 ", refCount:%d", pNode->vgId, rid, refCount);
sTrace("vgId:%d, acquire node refId:%" PRId64 ", refCount:%d", pNode->vgId, rid, refCount);
}
return pNode;
@ -502,16 +500,14 @@ SSyncNode *syncAcquireNode(int64_t rid) {
void syncReleaseNode(SSyncNode *pNode) {
int32_t refCount = atomic_sub_fetch_32(&pNode->refCount, 1);
sTrace("vgId:%d, dec syncnode refId:%" PRId64 " refCount:%d", pNode->vgId, pNode->rid, refCount);
sTrace("vgId:%d, release node refId:%" PRId64 ", refCount:%d", pNode->vgId, pNode->rid, refCount);
taosReleaseRef(tsNodeRefId, pNode->rid);
}
static void syncFreePeer(void *param) {
SSyncPeer *pPeer = param;
int32_t refCount = atomic_sub_fetch_32(&pPeer->refCount, 1);
sDebug("%s, peer is freed, refCount:%d", pPeer->id, refCount);
sDebug("%s, peer is freed, refCount:%d", pPeer->id, pPeer->refCount);
syncReleaseNode(pPeer->pSyncNode);
tfree(pPeer);
@ -531,7 +527,7 @@ SSyncPeer *syncAcquirePeer(int64_t rid) {
void syncReleasePeer(SSyncPeer *pPeer) {
int32_t refCount = atomic_sub_fetch_32(&pPeer->refCount, 1);
sTrace("%s, dec peer refId:%" PRId64 ", refCount:%d", pPeer->id, pPeer->rid, refCount);
sTrace("%s, release peer refId:%" PRId64 ", refCount:%d", pPeer->id, pPeer->rid, refCount);
taosReleaseRef(tsPeerRefId, pPeer->rid);
}
@ -879,14 +875,14 @@ static void syncProcessSyncRequest(char *msg, SSyncPeer *pPeer) {
int32_t ret = pthread_create(&thread, &thattr, syncRetrieveData, (void *)pPeer->rid);
pthread_attr_destroy(&thattr);
if (ret != 0) {
sError("%s, failed to create sync thread since %s", pPeer->id, strerror(errno));
if (ret < 0) {
sError("%s, failed to create sync retrieve thread since %s", pPeer->id, strerror(errno));
syncReleasePeer(pPeer);
} else {
pPeer->sstatus = TAOS_SYNC_STATUS_START;
sDebug("%s, thread is created to retrieve data, set sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
sDebug("%s, sync retrieve thread:0x%08" PRIx64 " create successfully, rid:%" PRId64 ", set sstatus:%s", pPeer->id,
taosGetPthreadId(thread), pPeer->rid, syncStatus[pPeer->sstatus]);
}
syncReleasePeer(pPeer);
}
static void syncNotStarted(void *param, void *tmrId) {
@ -1154,19 +1150,19 @@ static void syncCreateRestoreDataThread(SSyncPeer *pPeer) {
(void)syncAcquirePeer(pPeer->rid);
int32_t ret = pthread_create(&(thread), &thattr, (void *)syncRestoreData, (void *)pPeer->rid);
int32_t ret = pthread_create(&thread, &thattr, (void *)syncRestoreData, (void *)pPeer->rid);
pthread_attr_destroy(&thattr);
if (ret < 0) {
SSyncNode *pNode = pPeer->pSyncNode;
nodeSStatus = TAOS_SYNC_STATUS_INIT;
sError("%s, failed to create sync thread, set sstatus:%s", pPeer->id, syncStatus[nodeSStatus]);
sError("%s, failed to create sync restore thread, set sstatus:%s", pPeer->id, syncStatus[nodeSStatus]);
taosClose(pPeer->syncFd);
} else {
sInfo("%s, sync connection is up", pPeer->id);
}
syncReleasePeer(pPeer);
} else {
sInfo("%s, sync restore thread:0x%08" PRIx64 " create successfully, rid:%" PRId64, pPeer->id,
taosGetPthreadId(thread), pPeer->rid);
}
}
static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {

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

@ -353,12 +353,16 @@ static int32_t syncRestoreDataStepByStep(SSyncPeer *pPeer) {
void *syncRestoreData(void *param) {
int64_t rid = (int64_t)param;
SSyncPeer *pPeer = syncAcquirePeer(rid);
if (pPeer == NULL) return NULL;
if (pPeer == NULL) {
sError("failed to restore data, invalid peer rid:%" PRId64, rid);
return NULL;
}
SSyncNode *pNode = pPeer->pSyncNode;
taosBlockSIGPIPE();
__sync_fetch_and_add(&tsSyncNum, 1);
sInfo("%s, start to restore data, sstatus:%s", pPeer->id, syncStatus[nodeSStatus]);
(*pNode->notifyRole)(pNode->vgId, TAOS_SYNC_ROLE_SYNCING);
@ -380,11 +384,14 @@ void *syncRestoreData(void *param) {
(*pNode->notifyRole)(pNode->vgId, nodeRole);
nodeSStatus = TAOS_SYNC_STATUS_INIT;
sInfo("%s, sync over, set sstatus:%s", pPeer->id, syncStatus[nodeSStatus]);
sInfo("%s, restore data over, set sstatus:%s", pPeer->id, syncStatus[nodeSStatus]);
taosClose(pPeer->syncFd);
syncCloseRecvBuffer(pNode);
__sync_fetch_and_sub(&tsSyncNum, 1);
// The ref is obtained in both the create thread and the current thread, so it is released twice
syncReleasePeer(pPeer);
syncReleasePeer(pPeer);
return NULL;

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

@ -194,7 +194,7 @@ static int32_t syncReadOneWalRecord(int32_t sfd, SWalHead *pHead) {
}
if (ret == 0) {
sTrace("sfd:%d, read to the end of file, ret:%d", sfd, ret);
sDebug("sfd:%d, read to the end of file, ret:%d", sfd, ret);
return 0;
}
@ -253,7 +253,7 @@ static int32_t syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversi
break;
}
sTrace("%s, last wal is forwarded, hver:%" PRIu64, pPeer->id, pHead->version);
sDebug("%s, last wal is forwarded, hver:%" PRIu64, pPeer->id, pHead->version);
int32_t wsize = code;
int32_t ret = taosWriteMsg(pPeer->syncFd, pHead, wsize);
@ -466,10 +466,15 @@ static int32_t syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
void *syncRetrieveData(void *param) {
int64_t rid = (int64_t)param;
SSyncPeer *pPeer = syncAcquirePeer(rid);
if (pPeer == NULL) return NULL;
if (pPeer == NULL) {
sError("failed to retrieve data, invalid peer rid:%" PRId64, rid);
return NULL;
}
SSyncNode *pNode = pPeer->pSyncNode;
taosBlockSIGPIPE();
sInfo("%s, start to retrieve data, sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->vgId, pPeer->numOfRetrieves);
@ -496,7 +501,11 @@ void *syncRetrieveData(void *param) {
pPeer->fileChanged = 0;
taosClose(pPeer->syncFd);
// The ref is obtained in both the create thread and the current thread, so it is released twice
syncReleasePeer(pPeer);
syncReleasePeer(pPeer);
sInfo("%s, sync retrieve data over, sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
return NULL;
}

4
src/util/src/tlog.c
View File

@ -364,7 +364,7 @@ void taosPrintLog(const char *flags, int32_t dflag, const char *format, ...) {
ptm = localtime_r(&curTime, &Tm);
len = sprintf(buffer, "%02d/%02d %02d:%02d:%02d.%06d 0x%08" PRIx64 " ", ptm->tm_mon + 1, ptm->tm_mday, ptm->tm_hour,
ptm->tm_min, ptm->tm_sec, (int32_t)timeSecs.tv_usec, taosGetPthreadId());
ptm->tm_min, ptm->tm_sec, (int32_t)timeSecs.tv_usec, taosGetSelfPthreadId());
len += sprintf(buffer + len, "%s", flags);
va_start(argpointer, format);
@ -450,7 +450,7 @@ void taosPrintLongString(const char *flags, int32_t dflag, const char *format, .
ptm = localtime_r(&curTime, &Tm);
len = sprintf(buffer, "%02d/%02d %02d:%02d:%02d.%06d 0x%08" PRIx64 " ", ptm->tm_mon + 1, ptm->tm_mday, ptm->tm_hour,
ptm->tm_min, ptm->tm_sec, (int32_t)timeSecs.tv_usec, taosGetPthreadId());
ptm->tm_min, ptm->tm_sec, (int32_t)timeSecs.tv_usec, taosGetSelfPthreadId());
len += sprintf(buffer + len, "%s", flags);
va_start(argpointer, format);

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

@ -249,7 +249,7 @@ void taosNotePrint(SNoteObj *pNote, const char *const format, ...) {
curTime = timeSecs.tv_sec;
ptm = localtime_r(&curTime, &Tm);
len = sprintf(buffer, "%02d/%02d %02d:%02d:%02d.%06d 0x%08" PRIx64 " ", ptm->tm_mon + 1, ptm->tm_mday, ptm->tm_hour,
ptm->tm_min, ptm->tm_sec, (int32_t)timeSecs.tv_usec, taosGetPthreadId());
ptm->tm_min, ptm->tm_sec, (int32_t)timeSecs.tv_usec, taosGetSelfPthreadId());
va_start(argpointer, format);
len += vsnprintf(buffer + len, MAX_NOTE_LINE_SIZE - len, format, argpointer);
va_end(argpointer);

4
src/util/src/tref.c
View File

@ -447,7 +447,7 @@ static int taosDecRefCount(int rsetId, int64_t rid, int remove) {
}
static void taosLockList(int64_t *lockedBy) {
int64_t tid = taosGetPthreadId();
int64_t tid = taosGetSelfPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(lockedBy, 0, tid) != 0) {
if (++i % 100 == 0) {
@ -457,7 +457,7 @@ static void taosLockList(int64_t *lockedBy) {
}
static void taosUnlockList(int64_t *lockedBy) {
int64_t tid = taosGetPthreadId();
int64_t tid = taosGetSelfPthreadId();
if (atomic_val_compare_exchange_64(lockedBy, tid, 0) != tid) {
assert(false);
}

8
src/util/src/ttimer.c
View File

@ -119,7 +119,7 @@ static void timerDecRef(tmr_obj_t* timer) {
}
static void lockTimerList(timer_list_t* list) {
int64_t tid = taosGetPthreadId();
int64_t tid = taosGetSelfPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(&(list->lockedBy), 0, tid) != 0) {
if (++i % 1000 == 0) {
@ -129,7 +129,7 @@ static void lockTimerList(timer_list_t* list) {
}
static void unlockTimerList(timer_list_t* list) {
int64_t tid = taosGetPthreadId();
int64_t tid = taosGetSelfPthreadId();
if (atomic_val_compare_exchange_64(&(list->lockedBy), tid, 0) != tid) {
assert(false);
tmrError("%" PRId64 " trying to unlock a timer list not locked by current thread.", tid);
@ -257,7 +257,7 @@ static bool removeFromWheel(tmr_obj_t* timer) {
static void processExpiredTimer(void* handle, void* arg) {
tmr_obj_t* timer = (tmr_obj_t*)handle;
timer->executedBy = taosGetPthreadId();
timer->executedBy = taosGetSelfPthreadId();
uint8_t state = atomic_val_compare_exchange_8(&timer->state, TIMER_STATE_WAITING, TIMER_STATE_EXPIRED);
if (state == TIMER_STATE_WAITING) {
const char* fmt = "%s timer[id=%" PRIuPTR ", fp=%p, param=%p] execution start.";
@ -406,7 +406,7 @@ static bool doStopTimer(tmr_obj_t* timer, uint8_t state) {
return false;
}
if (timer->executedBy == taosGetPthreadId()) {
if (timer->executedBy == taosGetSelfPthreadId()) {
// taosTmrReset is called in the timer callback, should do nothing in this
// case to avoid dead lock. note taosTmrReset must be the last statement
// of the callback funtion, will be a bug otherwise.

18
tests/script/unique/mnode/mgmt20.sim
View File

@ -50,6 +50,24 @@ $d1_first = $rows
sql select * from log.dn2
$d2_first = $rows
$x = 0
show4:
$x = $x + 1
sleep 1000
if $x == 20 then
return -1
endi
sql show mnodes
print dnode1 ==> $data2_1
print dnode2 ==> $data2_2
if $data2_1 != master then
goto show4
endi
if $data2_2 != slave then
goto show4
endi
sleep 3000
sql select * from log.dn1
$d1_second = $rows