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

fix exchange operator memory access problem

This commit is contained in:
wangjiaming0909 2024-05-31 10:08:15 +00:00 committed by wangjiaming0909
parent 812777a664
commit 6fcd071144
2 changed files with 8 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/os/osAtomic.h
View File

@ -114,6 +114,12 @@ int32_t atomic_fetch_xor_32(int32_t volatile *ptr, int32_t val);
int64_t atomic_fetch_xor_64(int64_t volatile *ptr, int64_t val); int64_t atomic_fetch_xor_64(int64_t volatile *ptr, int64_t val);
void *atomic_fetch_xor_ptr(void *ptr, void *val); void *atomic_fetch_xor_ptr(void *ptr, void *val);
#ifdef _MSC_VER
#define tmemory_barrier(order) MemoryBarrier()
#else
#define tmemory_barrier(order) __sync_synchronize()
#endif
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

2
source/libs/executor/src/exchangeoperator.c
View File

@ -93,6 +93,7 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
goto _error; goto _error;
} }
tmemory_barrier();
SRetrieveTableRsp* pRsp = pDataInfo->pRsp; SRetrieveTableRsp* pRsp = pDataInfo->pRsp;
SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pDataInfo->index); SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pDataInfo->index);
@ -428,6 +429,7 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
} }
} }
tmemory_barrier();
pSourceDataInfo->status = EX_SOURCE_DATA_READY; pSourceDataInfo->status = EX_SOURCE_DATA_READY;
code = tsem_post(&pExchangeInfo->ready); code = tsem_post(&pExchangeInfo->ready);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {