#3660

Committed 15 Mar 2025 09:06AM UTC coverage: 62.039% (-1.3%) from 63.314%

Build # #3660

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(stream): support stream processing for virtual tables (#30144)

* enh: add client processing

* enh: add mnode vtables processing

* enh: add mnode vtable processing

* enh: add normal child vtable support

* fix: compile issues

* fix: compile issues

* fix: create stream issues

* fix: multi stream scan issue

* fix: remove debug info

* fix: agg task and task level issues

* fix: correct task output type

* fix: split vtablescan from agg

* fix: memory leak issues

* fix: add limitations

* Update 09-error-code.md

* Update 09-error-code.md

* fix: remove usless case

* feat(stream): extract original table data in source scan task

Implemented functionality in the source task to extract data
corresponding to the virtual table from the original table using WAL.
The extracted data is then sent to the downstream merge task for further
processing.

* feat(stream): multi-way merge using loser tree in virtual merge task

Implemented multi-way merge in the merge task using a loser tree to
combine data from multiple original table into a single virtual table.
The merged virtual table data is then pushed downstream for further
processing.  Introduced memory limit handling during the merge process
with configurable behavior when the memory limit is reached.

* fix(test): remove useless cases

---------

Co-authored-by: dapan1121 <wpan@taosdata.com>
Co-authored-by: Pan Wei <72057773+dapan1121@users.noreply.github.com>

Run Details

154078 of 317582 branches covered (48.52%)

Branch coverage included in aggregate %.

313 of 2391 new or added lines in 34 files covered. (13.09%)

26134 existing lines in 205 files now uncovered.

240261 of 318051 relevant lines covered (75.54%)

16655189.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.77

/source/os/src/osThread.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#define ALLOW_FORBID_FUNC
#include <pthread.h>
#include "os.h"

int32_t taosThreadCreate(TdThread *tid, const TdThreadAttr *attr, void *(*start)(void *), void *arg) {
  OS_PARAM_CHECK(tid);
  OS_PARAM_CHECK(start);
#ifdef TD_ASTRA
  int32_t code = 0;
  if (!attr) {
    pthread_attr_t threadAttr;
    pthread_attr_init(&threadAttr);
    pthread_attr_setstacksize(&threadAttr, STACK_SIZE_DEFAULT);
    code = pthread_create(tid, &threadAttr, start, arg);
    pthread_attr_destroy(&threadAttr);
  } else {
    int32_t stackSize = 0;
    pthread_attr_getstacksize(attr, &stackSize);
    if (stackSize == 0) {
      pthread_attr_setstacksize(attr, STACK_SIZE_DEFAULT);
    }
    code = pthread_create(tid, attr, start, arg);
  }
#else
  int32_t code = pthread_create(tid, attr, start, arg);
#endif
  if (code) {
    taosThreadClear(tid);
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrDestroy(TdThreadAttr *attr) {
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_attr_destroy(attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrGetDetachState(const TdThreadAttr *attr, int32_t *detachstate) {
  OS_PARAM_CHECK(attr);
  OS_PARAM_CHECK(detachstate);
  int32_t code = pthread_attr_getdetachstate(attr, detachstate);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrGetInheritSched(const TdThreadAttr *attr, int32_t *inheritsched) {
  OS_PARAM_CHECK(attr);
  OS_PARAM_CHECK(inheritsched);
  int32_t code = pthread_attr_getinheritsched(attr, inheritsched);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrGetSchedParam(const TdThreadAttr *attr, struct sched_param *param) {
  OS_PARAM_CHECK(attr);
  OS_PARAM_CHECK(param);
  int32_t code = pthread_attr_getschedparam(attr, param);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrGetSchedPolicy(const TdThreadAttr *attr, int32_t *policy) {
  OS_PARAM_CHECK(attr);
  OS_PARAM_CHECK(policy);
  int32_t code = pthread_attr_getschedpolicy(attr, policy);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrGetScope(const TdThreadAttr *attr, int32_t *contentionscope) {
  OS_PARAM_CHECK(attr);
  OS_PARAM_CHECK(contentionscope);
  int32_t code = pthread_attr_getscope(attr, contentionscope);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrGetStackSize(const TdThreadAttr *attr, size_t *stacksize) {
  OS_PARAM_CHECK(attr);
  OS_PARAM_CHECK(stacksize);
  int32_t code = pthread_attr_getstacksize(attr, stacksize);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrInit(TdThreadAttr *attr) {
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_attr_init(attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrSetDetachState(TdThreadAttr *attr, int32_t detachstate) {
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_attr_setdetachstate(attr, detachstate);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrSetInheritSched(TdThreadAttr *attr, int32_t inheritsched) {
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_attr_setinheritsched(attr, inheritsched);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrSetSchedParam(TdThreadAttr *attr, const struct sched_param *param) {
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_attr_setschedparam(attr, param);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrSetSchedPolicy(TdThreadAttr *attr, int32_t policy) {
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_attr_setschedpolicy(attr, policy);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrSetScope(TdThreadAttr *attr, int32_t contentionscope) {
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_attr_setscope(attr, contentionscope);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadAttrSetStackSize(TdThreadAttr *attr, size_t stacksize) {
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_attr_setstacksize(attr, stacksize);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadCancel(TdThread thread) {
  int32_t code = pthread_cancel(thread);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadCondDestroy(TdThreadCond *cond) {
  OS_PARAM_CHECK(cond);
#ifdef __USE_WIN_THREAD
  return 0;
#else
  int32_t code = pthread_cond_destroy(cond);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondInit(TdThreadCond *cond, const TdThreadCondAttr *attr) {
  OS_PARAM_CHECK(cond);
#ifdef __USE_WIN_THREAD
  InitializeConditionVariable(cond);
  return 0;
#else
  int32_t code = pthread_cond_init(cond, attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondSignal(TdThreadCond *cond) {
  OS_PARAM_CHECK(cond);
#ifdef __USE_WIN_THREAD
  WakeConditionVariable(cond);
  return 0;
#else
  int32_t code = pthread_cond_signal(cond);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondBroadcast(TdThreadCond *cond) {
  OS_PARAM_CHECK(cond);
#ifdef __USE_WIN_THREAD
  WakeAllConditionVariable(cond);
  return 0;
#else
  int32_t code = pthread_cond_broadcast(cond);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondWait(TdThreadCond *cond, TdThreadMutex *mutex) {
  OS_PARAM_CHECK(cond);
  OS_PARAM_CHECK(mutex);
#ifdef __USE_WIN_THREAD
  if (!SleepConditionVariableCS(cond, mutex, INFINITE)) {
    return EINVAL;
  }
  return 0;
#else
  int32_t code = pthread_cond_wait(cond, mutex);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondTimedWait(TdThreadCond *cond, TdThreadMutex *mutex, const struct timespec *abstime) {
  if (!abstime) return 0;
  OS_PARAM_CHECK(cond);
  OS_PARAM_CHECK(mutex);
#ifdef __USE_WIN_THREAD
  if (SleepConditionVariableCS(cond, mutex, (DWORD)(abstime->tv_sec * 1e3 + abstime->tv_nsec / 1e6))) return 0;
  DWORD error = GetLastError();
  if (error == ERROR_TIMEOUT) {
    return TSDB_CODE_TIMEOUT_ERROR;
  }
  return TAOS_SYSTEM_WINAPI_ERROR(error);
#else
  int32_t code = pthread_cond_timedwait(cond, mutex, abstime);
  if (code == ETIMEDOUT) {
    return TSDB_CODE_TIMEOUT_ERROR;
  } else if (code) {
    return TAOS_SYSTEM_ERROR(code);
  } else {
    return 0;
  }
#endif
}

int32_t taosThreadCondAttrDestroy(TdThreadCondAttr *attr) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_condattr_destroy(attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondAttrGetPshared(const TdThreadCondAttr *attr, int32_t *pshared) {
  OS_PARAM_CHECK(attr);
  OS_PARAM_CHECK(pshared);
#ifdef __USE_WIN_THREAD
  if (pshared) *pshared = PTHREAD_PROCESS_PRIVATE;
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_condattr_getpshared(attr, pshared);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondAttrInit(TdThreadCondAttr *attr) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_condattr_init(attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondAttrSetclock(TdThreadCondAttr *attr, int clockId) {
#ifdef __USE_WIN_THREAD
  return 0;
#elif defined(__APPLE__)
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_condattr_setclock(attr, clockId);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadCondAttrSetPshared(TdThreadCondAttr *attr, int32_t pshared) {
  OS_PARAM_CHECK(attr);
#ifdef __USE_WIN_THREAD
  return 0;
#else
  int32_t code = pthread_condattr_setpshared(attr, pshared);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadDetach(TdThread thread) {
  int32_t code = pthread_detach(thread);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadEqual(TdThread t1, TdThread t2) { return pthread_equal(t1, t2); }

void taosThreadExit(void *valuePtr) {
  if (valuePtr) return pthread_exit(valuePtr);
}

int32_t taosThreadGetSchedParam(TdThread thread, int32_t *policy, struct sched_param *param) {
  OS_PARAM_CHECK(policy);
  OS_PARAM_CHECK(param);
  int32_t code = pthread_getschedparam(thread, policy, param);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

void *taosThreadGetSpecific(TdThreadKey key) { return pthread_getspecific(key); }

int32_t taosThreadJoin(TdThread thread, void **valuePtr) {
  int32_t code = pthread_join(thread, valuePtr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadKeyCreate(TdThreadKey *key, void (*destructor)(void *)) {
  OS_PARAM_CHECK(key);
  int32_t code = pthread_key_create(key, destructor);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadKeyDelete(TdThreadKey key) {
  int32_t code = pthread_key_delete(key);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadKill(TdThread thread, int32_t sig) {
  int32_t code = pthread_kill(thread, sig);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

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

int32_t taosThreadMutexDestroy(TdThreadMutex *mutex) {
  OS_PARAM_CHECK(mutex);
#ifdef __USE_WIN_THREAD
  DeleteCriticalSection(mutex);
  return 0;
#else
  int32_t code = pthread_mutex_destroy(mutex);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadMutexInit(TdThreadMutex *mutex, const TdThreadMutexAttr *attr) {
  OS_PARAM_CHECK(mutex);
#ifdef __USE_WIN_THREAD
  /**
   * Windows Server 2003 and Windows XP:  In low memory situations, InitializeCriticalSection can raise a
   * STATUS_NO_MEMORY exception. Starting with Windows Vista, this exception was eliminated and
   * InitializeCriticalSection always succeeds, even in low memory situations.
   */
  InitializeCriticalSection(mutex);
  return 0;
#else
  int32_t code = pthread_mutex_init(mutex, attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadMutexLock(TdThreadMutex *mutex) {
  OS_PARAM_CHECK(mutex);
#ifdef __USE_WIN_THREAD
  EnterCriticalSection(mutex);
  return 0;
#else
  int32_t code = pthread_mutex_lock(mutex);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

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

int32_t taosThreadMutexTryLock(TdThreadMutex *mutex) {
  OS_PARAM_CHECK(mutex);
#ifdef __USE_WIN_THREAD
  if (TryEnterCriticalSection(mutex)) return 0;
  return EBUSY;
#else
  int32_t code = pthread_mutex_trylock(mutex);
  if (code && code != EBUSY) {
    code = TAOS_SYSTEM_ERROR(code);
  }
  return code;
#endif
}

int32_t taosThreadMutexUnlock(TdThreadMutex *mutex) {
  OS_PARAM_CHECK(mutex);
#ifdef __USE_WIN_THREAD
  LeaveCriticalSection(mutex);
  return 0;
#else
  int32_t code = pthread_mutex_unlock(mutex);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadMutexAttrDestroy(TdThreadMutexAttr *attr) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_mutexattr_destroy(attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadMutexAttrGetPshared(const TdThreadMutexAttr *attr, int32_t *pshared) {
  OS_PARAM_CHECK(pshared);
#ifdef __USE_WIN_THREAD
  if (pshared) *pshared = PTHREAD_PROCESS_PRIVATE;
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_mutexattr_getpshared(attr, pshared);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

// int32_t taosThreadMutexAttrGetRobust(const TdThreadMutexAttr * attr, int32_t * robust) {
//   return pthread_mutexattr_getrobust(attr, robust);
// }

int32_t taosThreadMutexAttrGetType(const TdThreadMutexAttr *attr, int32_t *kind) {
  OS_PARAM_CHECK(kind);
#ifdef __USE_WIN_THREAD
  if (kind) *kind = PTHREAD_MUTEX_NORMAL;
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_mutexattr_gettype(attr, kind);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadMutexAttrInit(TdThreadMutexAttr *attr) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_mutexattr_init(attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadMutexAttrSetPshared(TdThreadMutexAttr *attr, int32_t pshared) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_mutexattr_setpshared(attr, pshared);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

// int32_t taosThreadMutexAttrSetRobust(TdThreadMutexAttr * attr, int32_t robust) {
//   return pthread_mutexattr_setrobust(attr, robust);
// }

int32_t taosThreadMutexAttrSetType(TdThreadMutexAttr *attr, int32_t kind) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_mutexattr_settype(attr, kind);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadOnce(TdThreadOnce *onceControl, void (*initRoutine)(void)) {
  int32_t code = pthread_once(onceControl, initRoutine);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadRwlockDestroy(TdThreadRwlock *rwlock) {
#ifdef __USE_WIN_THREAD
  /* SRWLock does not need explicit destruction so long as there are no waiting threads
   * See: https://docs.microsoft.com/windows/win32/api/synchapi/nf-synchapi-initializesrwlock#remarks
   */
  return 0;
#else
  OS_PARAM_CHECK(rwlock);
  int32_t code = pthread_rwlock_destroy(rwlock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockInit(TdThreadRwlock *rwlock, const TdThreadRwlockAttr *attr) {
  OS_PARAM_CHECK(rwlock);
#ifdef __USE_WIN_THREAD
  memset(rwlock, 0, sizeof(*rwlock));
  InitializeSRWLock(&rwlock->lock);
  return 0;
#else
  int32_t code = pthread_rwlock_init(rwlock, attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockRdlock(TdThreadRwlock *rwlock) {
  OS_PARAM_CHECK(rwlock);
#ifdef __USE_WIN_THREAD
  AcquireSRWLockShared(&rwlock->lock);
  return 0;
#else
  int32_t code = pthread_rwlock_rdlock(rwlock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

// int32_t taosThreadRwlockTimedRdlock(TdThreadRwlock * rwlock, const struct timespec *abstime) {
//   return pthread_rwlock_timedrdlock(rwlock, abstime);
// }

// int32_t taosThreadRwlockTimedWrlock(TdThreadRwlock * rwlock, const struct timespec *abstime) {
//   return pthread_rwlock_timedwrlock(rwlock, abstime);
// }

int32_t taosThreadRwlockTryRdlock(TdThreadRwlock *rwlock) {
  OS_PARAM_CHECK(rwlock);
#ifdef __USE_WIN_THREAD
  if (!TryAcquireSRWLockShared(&rwlock->lock)) return EBUSY;
  return 0;
#else
  int32_t code = pthread_rwlock_tryrdlock(rwlock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockTryWrlock(TdThreadRwlock *rwlock) {
  OS_PARAM_CHECK(rwlock);
#ifdef __USE_WIN_THREAD
  if (!TryAcquireSRWLockExclusive(&rwlock->lock)) return EBUSY;
  atomic_store_8(&rwlock->excl, 1);
  return 0;
#else
  int32_t code = pthread_rwlock_trywrlock(rwlock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockUnlock(TdThreadRwlock *rwlock) {
  OS_PARAM_CHECK(rwlock);
#ifdef __USE_WIN_THREAD
  if (1 == atomic_val_compare_exchange_8(&rwlock->excl, 1, 0)) {
    ReleaseSRWLockExclusive(&rwlock->lock);
  } else {
    ReleaseSRWLockShared(&rwlock->lock);
  }
  return 0;
#else
  int32_t code = pthread_rwlock_unlock(rwlock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockWrlock(TdThreadRwlock *rwlock) {
  OS_PARAM_CHECK(rwlock);
#ifdef __USE_WIN_THREAD
  AcquireSRWLockExclusive(&rwlock->lock);
  atomic_store_8(&rwlock->excl, 1);
  return 0;
#else
  int32_t code = pthread_rwlock_wrlock(rwlock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockAttrDestroy(TdThreadRwlockAttr *attr) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_rwlockattr_destroy(attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockAttrGetPshared(const TdThreadRwlockAttr *attr, int32_t *pshared) {
  OS_PARAM_CHECK(attr);
  OS_PARAM_CHECK(pshared);
#ifdef __USE_WIN_THREAD
  if (pshared) *pshared = PTHREAD_PROCESS_PRIVATE;
  return 0;
#else
  int32_t code = pthread_rwlockattr_getpshared(attr, pshared);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockAttrInit(TdThreadRwlockAttr *attr) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_rwlockattr_init(attr);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadRwlockAttrSetPshared(TdThreadRwlockAttr *attr, int32_t pshared) {
#ifdef __USE_WIN_THREAD
  return 0;
#else
  OS_PARAM_CHECK(attr);
  int32_t code = pthread_rwlockattr_setpshared(attr, pshared);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

TdThread taosThreadSelf(void) { return pthread_self(); }

int32_t taosThreadSetCancelState(int32_t state, int32_t *oldstate) {
  int32_t code = pthread_setcancelstate(state, oldstate);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadSetCancelType(int32_t type, int32_t *oldtype) {
  int32_t code = pthread_setcanceltype(type, oldtype);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadSetSchedParam(TdThread thread, int32_t policy, const struct sched_param *param) {
  OS_PARAM_CHECK(param);
  int32_t code = pthread_setschedparam(thread, policy, param);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadSetSpecific(TdThreadKey key, const void *value) {
  OS_PARAM_CHECK(value);
  int32_t code = pthread_setspecific(key, value);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
}

int32_t taosThreadSpinDestroy(TdThreadSpinlock *lock) {
  OS_PARAM_CHECK(lock);
#ifdef TD_USE_SPINLOCK_AS_MUTEX
  return pthread_mutex_destroy((pthread_mutex_t *)lock);
#else
  int32_t code = pthread_spin_destroy((pthread_spinlock_t *)lock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadSpinInit(TdThreadSpinlock *lock, int32_t pshared) {
  OS_PARAM_CHECK(lock);
#ifdef TD_USE_SPINLOCK_AS_MUTEX
  if (pshared != 0) return TSDB_CODE_INVALID_PARA;
  return pthread_mutex_init((pthread_mutex_t *)lock, NULL);
#else
  int32_t code = pthread_spin_init((pthread_spinlock_t *)lock, pshared);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadSpinLock(TdThreadSpinlock *lock) {
  OS_PARAM_CHECK(lock);
#ifdef TD_USE_SPINLOCK_AS_MUTEX
  return pthread_mutex_lock((pthread_mutex_t *)lock);
#else
  int32_t code = pthread_spin_lock((pthread_spinlock_t *)lock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

int32_t taosThreadSpinTrylock(TdThreadSpinlock *lock) {
  OS_PARAM_CHECK(lock);
#ifdef TD_USE_SPINLOCK_AS_MUTEX
  return pthread_mutex_trylock((pthread_mutex_t *)lock);
#else
  int32_t code = pthread_spin_trylock((pthread_spinlock_t *)lock);
  if (code && code != EBUSY) {
    code = TAOS_SYSTEM_ERROR(code);
  }
  return code;
#endif
}

int32_t taosThreadSpinUnlock(TdThreadSpinlock *lock) {
  OS_PARAM_CHECK(lock);
#ifdef TD_USE_SPINLOCK_AS_MUTEX
  return pthread_mutex_unlock((pthread_mutex_t *)lock);
#else
  int32_t code = pthread_spin_unlock((pthread_spinlock_t *)lock);
  if (code) {
    return (terrno = TAOS_SYSTEM_ERROR(code));
  }
  return code;
#endif
}

void taosThreadTestCancel(void) { return pthread_testcancel(); }

void taosThreadClear(TdThread *thread) {
  if (!thread) return;
  (void)memset(thread, 0, sizeof(TdThread));
}

#ifdef WINDOWS
bool taosThreadIsMain() {
  DWORD curProcessId = GetCurrentProcessId();
  DWORD curThreadId = GetCurrentThreadId();
  DWORD dwThreadId = -1;

  HANDLE hThreadSnapshot = CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0);
  if (hThreadSnapshot == INVALID_HANDLE_VALUE) {
    return false;
  }

  THREADENTRY32 te32;
  te32.dwSize = sizeof(THREADENTRY32);

  if (!Thread32First(hThreadSnapshot, &te32)) {
    CloseHandle(hThreadSnapshot);
    return false;
  }

  do {
    if (te32.th32OwnerProcessID == curProcessId) {
      dwThreadId = te32.th32ThreadID;
      break;
    }
  } while (Thread32Next(hThreadSnapshot, &te32));

  CloseHandle(hThreadSnapshot);

  return curThreadId == dwThreadId;
}
#endif

taosdata / TDengine / #3660

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous