Skip to content

Commit

Permalink
Merge pull request #22287 from taosdata/fix/3_liaohj
Browse files Browse the repository at this point in the history 
other: merge 3.0
  • Loading branch information
@gccgdb1234
gccgdb1234 committed Aug 2, 2023
2 parents cbca55b + 0feffc6 commit 12ac393
Show file tree
Hide file tree
Showing 3 changed files with 93 additions and 32 deletions.
83 changes: 57 additions & 26 deletions source/dnode/vnode/src/tsdb/tsdbMergeTree.c
View file
Expand Up @@ -15,6 +15,7 @@

#include "tsdb.h"
#include "tsdbFSet2.h"
#include "tsdbMerge.h"
#include "tsdbReadUtil.h"
#include "tsdbSttFileRW.h"

Expand Down Expand Up @@ -352,10 +353,14 @@ static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray
return TSDB_CODE_SUCCESS;
}

static int32_t uidComparFn(const void *p1, const void *p2) {
const uint64_t *uid1 = p1;
static int32_t suidComparFn(const void *target, const void *p2) {
const uint64_t *targetUid = target;
const uint64_t *uid2 = p2;
return (*uid1) - (*uid2);
if (*uid2 == (*targetUid)) {
return 0;
} else {
return (*targetUid) < (*uid2) ? -1:1;
}
}

static bool existsFromSttBlkStatis(const TStatisBlkArray *pStatisBlkArray, uint64_t suid, uint64_t uid,
Expand All @@ -372,29 +377,55 @@ static bool existsFromSttBlkStatis(const TStatisBlkArray *pStatisBlkArray, uint6
}
}

// for (; i < TARRAY2_SIZE(pStatisBlkArray); ++i) {
// SStatisBlk *p = &pStatisBlkArray->data[i];
// if (p->minTbid.uid <= uid && p->maxTbid.uid >= uid) {
// break;
// }
//
// if (p->maxTbid.uid < uid) {
// break;
// }
// }

if (i >= TARRAY2_SIZE(pStatisBlkArray)) {
return false;
}

SStatisBlk *p = &pStatisBlkArray->data[i];
STbStatisBlock block = {0};
tsdbSttFileReadStatisBlock(pReader, p, &block);
while(i < TARRAY2_SIZE(pStatisBlkArray)) {
SStatisBlk *p = &pStatisBlkArray->data[i];
if (p->minTbid.suid > suid) {
return false;
}

STbStatisBlock block = {0};
tsdbSttFileReadStatisBlock(pReader, p, &block);

int32_t index = tarray2SearchIdx(block.suid, &suid, sizeof(int64_t), suidComparFn, TD_EQ);
if (index == -1) {
tStatisBlockDestroy(&block);
return false;
}

int32_t j = index;
if (block.uid->data[j] == uid) {
tStatisBlockDestroy(&block);
return true;
} else if (block.uid->data[j] > uid) {
while (j >= 0 && block.suid->data[j] == suid) {
if (block.uid->data[j] == uid) {
tStatisBlockDestroy(&block);
return true;
} else {
j -= 1;
}
}
} else {
j = index + 1;
while (j < block.suid->size && block.suid->data[j] == suid) {
if (block.uid->data[j] == uid) {
tStatisBlockDestroy(&block);
return true;
} else {
j += 1;
}
}
}

int32_t index = tarray2SearchIdx(block.uid, &uid, sizeof(int64_t), uidComparFn, TD_EQ);
tStatisBlockDestroy(&block);
tStatisBlockDestroy(&block);
i += 1;
}

return (index != -1);
return false;
}

int32_t tLDataIterOpen2(struct SLDataIter *pIter, SSttFileReader *pSttFileReader, int32_t iStt, int8_t backward,
Expand Down Expand Up @@ -445,12 +476,12 @@ int32_t tLDataIterOpen2(struct SLDataIter *pIter, SSttFileReader *pSttFileReader
tsdbDebug("load the stt file info completed, elapsed time:%.2fms, %s", el, idStr);
}

// bool exists = existsFromSttBlkStatis(pBlockLoadInfo->pSttStatisBlkArray, suid, uid, pIter->pReader);
// if (!exists) {
// pIter->iSttBlk = -1;
// pIter->pSttBlk = NULL;
// return TSDB_CODE_SUCCESS;
// }
bool exists = existsFromSttBlkStatis(pBlockLoadInfo->pSttStatisBlkArray, suid, uid, pIter->pReader);
if (!exists) {
pIter->iSttBlk = -1;
pIter->pSttBlk = NULL;
return TSDB_CODE_SUCCESS;
}

// find the start block, actually we could load the position to avoid repeatly searching for the start position when
// the skey is updated.
Expand Down
2 changes: 1 addition & 1 deletion source/dnode/vnode/src/tsdb/tsdbRetention.c
View file
Expand Up @@ -100,7 +100,7 @@ static int32_t tsdbDoMigrateFileObj(SRTNer *rtner, const STFileObj *fobj, const
.type = fobj->f->type,
.did = did[0],
.fid = fobj->f->fid,
.cid = rtner->cid,
.cid = fobj->f->cid,
.size = fobj->f->size,
.stt[0] =
{
Expand Down
40 changes: 35 additions & 5 deletions source/os/src/osThread.c
View file
Expand Up @@ -17,6 +17,15 @@
#include <pthread.h>
#include "os.h"

#ifdef WINDOWS
#define THREAD_PTR_CHECK(p) \
do { \
if (!(p) || !(*(p))) return 0; \
} while (0);
#else
#define THREAD_PTR_CHECK(p)
#endif

int32_t taosThreadCreate(TdThread *tid, const TdThreadAttr *attr, void *(*start)(void *), void *arg) {
return pthread_create(tid, attr, start, arg);
}
Expand Down Expand Up @@ -83,9 +92,13 @@ int32_t taosThreadCondSignal(TdThreadCond *cond) { return pthread_cond_signal(co

int32_t taosThreadCondBroadcast(TdThreadCond *cond) { return pthread_cond_broadcast(cond); }

int32_t taosThreadCondWait(TdThreadCond *cond, TdThreadMutex *mutex) { return pthread_cond_wait(cond, mutex); }
int32_t taosThreadCondWait(TdThreadCond *cond, TdThreadMutex *mutex) {
THREAD_PTR_CHECK(mutex)
return pthread_cond_wait(cond, mutex);
}

int32_t taosThreadCondTimedWait(TdThreadCond *cond, TdThreadMutex *mutex, const struct timespec *abstime) {
THREAD_PTR_CHECK(mutex)
return pthread_cond_timedwait(cond, mutex, abstime);
}

Expand Down Expand Up @@ -124,24 +137,37 @@ int32_t taosThreadKeyDelete(TdThreadKey key) { return pthread_key_delete(key); }
int32_t taosThreadKill(TdThread thread, int32_t sig) { return pthread_kill(thread, sig); }

// int32_t taosThreadMutexConsistent(TdThreadMutex* mutex) {
// THREAD_PTR_CHECK(mutex)
// return pthread_mutex_consistent(mutex);
// }

int32_t taosThreadMutexDestroy(TdThreadMutex *mutex) { return pthread_mutex_destroy(mutex); }
int32_t taosThreadMutexDestroy(TdThreadMutex *mutex) {
THREAD_PTR_CHECK(mutex)
return pthread_mutex_destroy(mutex);
}

int32_t taosThreadMutexInit(TdThreadMutex *mutex, const TdThreadMutexAttr *attr) {
return pthread_mutex_init(mutex, attr);
}

int32_t taosThreadMutexLock(TdThreadMutex *mutex) { return pthread_mutex_lock(mutex); }
int32_t taosThreadMutexLock(TdThreadMutex *mutex) {
THREAD_PTR_CHECK(mutex)
return pthread_mutex_lock(mutex);
}

// int32_t taosThreadMutexTimedLock(TdThreadMutex * mutex, const struct timespec *abstime) {
// return pthread_mutex_timedlock(mutex, abstime);
// }

int32_t taosThreadMutexTryLock(TdThreadMutex *mutex) { return pthread_mutex_trylock(mutex); }
int32_t taosThreadMutexTryLock(TdThreadMutex *mutex) {
THREAD_PTR_CHECK(mutex)
return pthread_mutex_trylock(mutex);
}

int32_t taosThreadMutexUnlock(TdThreadMutex *mutex) { return pthread_mutex_unlock(mutex); }
int32_t taosThreadMutexUnlock(TdThreadMutex *mutex) {
THREAD_PTR_CHECK(mutex)
return pthread_mutex_unlock(mutex);
}

int32_t taosThreadMutexAttrDestroy(TdThreadMutexAttr *attr) { return pthread_mutexattr_destroy(attr); }

Expand Down Expand Up @@ -224,6 +250,7 @@ int32_t taosThreadSetSchedParam(TdThread thread, int32_t policy, const struct sc
int32_t taosThreadSetSpecific(TdThreadKey key, const void *value) { return pthread_setspecific(key, value); }

int32_t taosThreadSpinDestroy(TdThreadSpinlock *lock) {
THREAD_PTR_CHECK(lock)
#ifdef TD_USE_SPINLOCK_AS_MUTEX
return pthread_mutex_destroy((pthread_mutex_t *)lock);
#else
Expand All @@ -242,6 +269,7 @@ int32_t taosThreadSpinInit(TdThreadSpinlock *lock, int32_t pshared) {
}

int32_t taosThreadSpinLock(TdThreadSpinlock *lock) {
THREAD_PTR_CHECK(lock)
#ifdef TD_USE_SPINLOCK_AS_MUTEX
return pthread_mutex_lock((pthread_mutex_t *)lock);
#else
Expand All @@ -250,6 +278,7 @@ int32_t taosThreadSpinLock(TdThreadSpinlock *lock) {
}

int32_t taosThreadSpinTrylock(TdThreadSpinlock *lock) {
THREAD_PTR_CHECK(lock)
#ifdef TD_USE_SPINLOCK_AS_MUTEX
return pthread_mutex_trylock((pthread_mutex_t *)lock);
#else
Expand All @@ -258,6 +287,7 @@ int32_t taosThreadSpinTrylock(TdThreadSpinlock *lock) {
}

int32_t taosThreadSpinUnlock(TdThreadSpinlock *lock) {
THREAD_PTR_CHECK(lock)
#ifdef TD_USE_SPINLOCK_AS_MUTEX
return pthread_mutex_unlock((pthread_mutex_t *)lock);
#else
Expand Down

0 comments on commit 12ac393

Please sign in to comment.