Skip to content

TDengine
TDengine

taosdata / TDengine
接近 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
TDengine
TDengine
提交 fcbaa608 编写于 作者: H Hongze Cheng
浏览文件
操作

more code refact

上级 5a45dab0
展开全部 隐藏空白更改
内联 并排
Showing with 2759 addition and 3186 deletion
include/os/osEnv.h
浏览文件 @ fcbaa608
......@@ -17,6 +17,7 @@
#define _TD_OS_ENV_H_
#include "osSysinfo.h"
#include "osTimezone.h"
#ifdef __cplusplus
extern "C" {
......
include/util/tdigest.h
浏览文件 @ fcbaa608
......@@ -22,6 +22,8 @@
#ifndef TDIGEST_H
#define TDIGEST_H
#include "os.h"
#ifndef M_PI
#define M_PI 3.14159265358979323846264338327950288 /* pi */
#endif
......
source/os/src/osAtomic.c
浏览文件 @ fcbaa608
......@@ -193,11 +193,11 @@ void* interlocked_sub_fetch_ptr(void* volatile* ptr, void* val) {
}
int32_t interlocked_fetch_sub_32(int32_t volatile* ptr, int32_t val) { return _InterlockedExchangeAdd(ptr, -val); }
int64_t interlocked_fetch_sub_64(int64_t volatile* ptr, int64_t val) {
int64_t interlocked_fetch_sub_64(int64_t volatile* ptr, int64_t val) {
#ifdef _TD_WINDOWS_32
return _InterlockedExchangeAdd((int32_t volatile*)ptr, -(int32_t)val);
return _InterlockedExchangeAdd((int32_t volatile*)ptr, -(int32_t)val);
#else
return _InterlockedExchangeAdd64(ptr, -val);
return _InterlockedExchangeAdd64(ptr, -val);
#endif
}
......
source/os/src/osDir.c
浏览文件 @ fcbaa608
......@@ -31,14 +31,13 @@ typedef struct TdDir {
HANDLE hFind;
} TdDir;
enum
{
WRDE_NOSPACE = 1, /* Ran out of memory. */
WRDE_BADCHAR, /* A metachar appears in the wrong place. */
WRDE_BADVAL, /* Undefined var reference with WRDE_UNDEF. */
WRDE_CMDSUB, /* Command substitution with WRDE_NOCMD. */
WRDE_SYNTAX /* Shell syntax error. */
};
enum {
WRDE_NOSPACE = 1, /* Ran out of memory. */
WRDE_BADCHAR, /* A metachar appears in the wrong place. */
WRDE_BADVAL, /* Undefined var reference with WRDE_UNDEF. */
WRDE_CMDSUB, /* Command substitution with WRDE_NOCMD. */
WRDE_SYNTAX /* Shell syntax error. */
};
int wordexp(char *words, wordexp_t *pwordexp, int flags) {
pwordexp->we_offs = 0;
......@@ -175,7 +174,7 @@ int32_t taosMulMkDir(const char *dirname) {
#ifdef WINDOWS
code = _mkdir(temp, 0755);
#elif defined(DARWIN)
code = mkdir(dirname, 0777);
code = mkdir(dirname, 0777);
#else
code = mkdir(temp, 0755);
#endif
......@@ -237,7 +236,7 @@ int32_t taosMulModeMkDir(const char *dirname, int mode) {
#ifdef WINDOWS
code = _mkdir(temp, mode);
#elif defined(DARWIN)
code = mkdir(dirname, 0777);
code = mkdir(dirname, 0777);
#else
code = mkdir(temp, mode);
#endif
......@@ -301,22 +300,24 @@ void taosRemoveOldFiles(const char *dirname, int32_t keepDays) {
int32_t taosExpandDir(const char *dirname, char *outname, int32_t maxlen) {
wordexp_t full_path;
switch (wordexp (dirname, &full_path, 0)) {
case 0:
break;
case WRDE_NOSPACE:
wordfree (&full_path);
// printf("failed to expand path:%s since Out of memory\n", dirname);
return -1;
case WRDE_BADCHAR:
// printf("failed to expand path:%s since illegal occurrence of newline or one of |, &, ;, <, >, (, ), {, }\n", dirname);
return -1;
case WRDE_SYNTAX:
// printf("failed to expand path:%s since Shell syntax error, such as unbalanced parentheses or unmatched quotes\n", dirname);
return -1;
default:
// printf("failed to expand path:%s since %s\n", dirname, strerror(errno));
return -1;
switch (wordexp(dirname, &full_path, 0)) {
case 0:
break;
case WRDE_NOSPACE:
wordfree(&full_path);
// printf("failed to expand path:%s since Out of memory\n", dirname);
return -1;
case WRDE_BADCHAR:
// printf("failed to expand path:%s since illegal occurrence of newline or one of |, &, ;, <, >, (, ), {, }\n",
// dirname);
return -1;
case WRDE_SYNTAX:
// printf("failed to expand path:%s since Shell syntax error, such as unbalanced parentheses or unmatched
// quotes\n", dirname);
return -1;
default:
// printf("failed to expand path:%s since %s\n", dirname, strerror(errno));
return -1;
}
if (full_path.we_wordv != NULL && full_path.we_wordv[0] != NULL) {
......@@ -417,7 +418,7 @@ TdDirPtr taosOpenDir(const char *dirname) {
DIR *pDir = opendir(dirname);
if (pDir == NULL) return NULL;
TdDirPtr dirPtr = (TdDirPtr)taosMemoryMalloc(sizeof(TdDir));
dirPtr->dirEntryPtr = (TdDirEntryPtr)&(dirPtr->dirEntry1);
dirPtr->dirEntryPtr = (TdDirEntryPtr) & (dirPtr->dirEntry1);
dirPtr->pDir = pDir;
return dirPtr;
#else
......@@ -435,7 +436,7 @@ TdDirEntryPtr taosReadDir(TdDirPtr pDir) {
}
return (TdDirEntryPtr) & (pDir->dirEntry.findFileData);
#elif defined(DARWIN)
if (readdir_r(pDir->pDir, (dirent*)&(pDir->dirEntry), (dirent**)&(pDir->dirEntryPtr)) == 0) {
if (readdir_r(pDir->pDir, (dirent *)&(pDir->dirEntry), (dirent **)&(pDir->dirEntryPtr)) == 0) {
return pDir->dirEntryPtr;
} else {
return NULL;
......
source/os/src/osEnv.c
浏览文件 @ fcbaa608
......@@ -35,7 +35,7 @@ int64_t tsOpenMax = 0;
int64_t tsStreamMax = 0;
float tsNumOfCores = 0;
int64_t tsTotalMemoryKB = 0;
char* tsProcPath = NULL;
char *tsProcPath = NULL;
void osDefaultInit() {
taosSeedRand(taosSafeRand());
......
source/os/src/osFile.c
浏览文件 @ fcbaa608
......@@ -58,14 +58,14 @@ typedef struct TdFile {
#define FILE_WITH_LOCK 1
typedef struct AutoDelFile * AutoDelFilePtr;
typedef struct AutoDelFile *AutoDelFilePtr;
typedef struct AutoDelFile {
char *name;
AutoDelFilePtr lastAutoDelFilePtr;
char *name;
AutoDelFilePtr lastAutoDelFilePtr;
} AutoDelFile;
static TdThreadMutex autoDelFileLock;
static TdThreadMutex autoDelFileLock;
static AutoDelFilePtr nowAutoDelFilePtr = NULL;
static TdThreadOnce autoDelFileInit = PTHREAD_ONCE_INIT;
static TdThreadOnce autoDelFileInit = PTHREAD_ONCE_INIT;
void taosGetTmpfilePath(const char *inputTmpDir, const char *fileNamePrefix, char *dstPath) {
#ifdef WINDOWS
......@@ -205,10 +205,10 @@ int32_t taosRenameFile(const char *oldName, const char *newName) {
int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime) {
#ifdef WINDOWS
struct _stati64 fileStat;
int32_t code = _stati64(path, &fileStat);
int32_t code = _stati64(path, &fileStat);
#else
struct stat fileStat;
int32_t code = stat(path, &fileStat);
int32_t code = stat(path, &fileStat);
#endif
if (code < 0) {
return code;
......@@ -706,11 +706,11 @@ int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, in
int64_t sentbytes;
while (leftbytes > 0) {
#ifdef _TD_ARM_32
sentbytes = sendfile(pFileOut->fd, pFileIn->fd, (long int*)offset, leftbytes);
#else
#ifdef _TD_ARM_32
sentbytes = sendfile(pFileOut->fd, pFileIn->fd, (long int *)offset, leftbytes);
#else
sentbytes = sendfile(pFileOut->fd, pFileIn->fd, offset, leftbytes);
#endif
#endif
if (sentbytes == -1) {
if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
continue;
......
source/os/src/osLocale.c
浏览文件 @ fcbaa608
......@@ -77,7 +77,7 @@ void taosSetSystemLocale(const char *inLocale, const char *inCharSet) {
// default locale or user specified locale is not valid, abort launch
if (inLocale == NULL || strlen(inLocale) == 0) {
//printf("Invalid locale:%s, please set the valid locale in config file\n", inLocale);
// printf("Invalid locale:%s, please set the valid locale in config file\n", inLocale);
}
if (!taosValidateEncodec(inCharSet)) {
......
source/os/src/osMemory.c
浏览文件 @ fcbaa608
......@@ -52,36 +52,38 @@ int32_t taosBackTrace(void **buffer, int32_t size) {
#pragma comment(lib, "dbghelp.lib")
void taosPrintBackTrace() {
#define MAX_STACK_FRAMES 20
void *pStack[MAX_STACK_FRAMES];
HANDLE process = GetCurrentProcess();
SymInitialize(process, NULL, TRUE);
WORD frames = CaptureStackBackTrace(1, MAX_STACK_FRAMES, pStack, NULL);
#define MAX_STACK_FRAMES 20
void *pStack[MAX_STACK_FRAMES];
HANDLE process = GetCurrentProcess();
SymInitialize(process, NULL, TRUE);
WORD frames = CaptureStackBackTrace(1, MAX_STACK_FRAMES, pStack, NULL);
char buf_tmp[1024];
for (WORD i = 0; i < frames; ++i) {
DWORD64 address = (DWORD64)(pStack[i]);
DWORD64 displacementSym = 0;
char buffer[sizeof(SYMBOL_INFO) + MAX_SYM_NAME * sizeof(TCHAR)];
PSYMBOL_INFO pSymbol = (PSYMBOL_INFO)buffer;
pSymbol->SizeOfStruct = sizeof(SYMBOL_INFO);
pSymbol->MaxNameLen = MAX_SYM_NAME;
DWORD displacementLine = 0;
IMAGEHLP_LINE64 line;
//SymSetOptions(SYMOPT_LOAD_LINES);
line.SizeOfStruct = sizeof(IMAGEHLP_LINE64);
if (SymFromAddr(process, address, &displacementSym, pSymbol) && SymGetLineFromAddr64(process, address, &displacementLine, &line)) {
snprintf(buf_tmp,sizeof(buf_tmp),"BackTrace %08" PRId64 " %s:%d %s\n", taosGetSelfPthreadId(), line.FileName, line.LineNumber, pSymbol->Name);
} else {
snprintf(buf_tmp,sizeof(buf_tmp),"BackTrace error: %d\n",GetLastError());
}
write(1,buf_tmp,strlen(buf_tmp));
}
for (WORD i = 0; i < frames; ++i) {
DWORD64 address = (DWORD64)(pStack[i]);
DWORD64 displacementSym = 0;
char buffer[sizeof(SYMBOL_INFO) + MAX_SYM_NAME * sizeof(TCHAR)];
PSYMBOL_INFO pSymbol = (PSYMBOL_INFO)buffer;
pSymbol->SizeOfStruct = sizeof(SYMBOL_INFO);
pSymbol->MaxNameLen = MAX_SYM_NAME;
DWORD displacementLine = 0;
IMAGEHLP_LINE64 line;
// SymSetOptions(SYMOPT_LOAD_LINES);
line.SizeOfStruct = sizeof(IMAGEHLP_LINE64);
if (SymFromAddr(process, address, &displacementSym, pSymbol) &&
SymGetLineFromAddr64(process, address, &displacementLine, &line)) {
snprintf(buf_tmp, sizeof(buf_tmp), "BackTrace %08" PRId64 " %s:%d %s\n", taosGetSelfPthreadId(), line.FileName,
line.LineNumber, pSymbol->Name);
} else {
snprintf(buf_tmp, sizeof(buf_tmp), "BackTrace error: %d\n", GetLastError());
}
write(1, buf_tmp, strlen(buf_tmp));
}
}
#endif
#else
......@@ -126,27 +128,26 @@ int32_t taosBackTrace(void **buffer, int32_t size) {
#ifdef USE_ADDR2LINE
#include "osThread.h"
#include "libdwarf.h"
#include "dwarf.h"
#include "libdwarf.h"
#include "osThread.h"
#define DW_PR_DUu "llu"
typedef struct lookup_table
{
Dwarf_Line *table;
Dwarf_Line_Context *ctxts;
int cnt;
Dwarf_Addr low;
Dwarf_Addr high;
typedef struct lookup_table {
Dwarf_Line *table;
Dwarf_Line_Context *ctxts;
int cnt;
Dwarf_Addr low;
Dwarf_Addr high;
} lookup_tableT;
extern int create_lookup_table(Dwarf_Debug dbg, lookup_tableT *lookup_table);
extern int create_lookup_table(Dwarf_Debug dbg, lookup_tableT *lookup_table);
extern void delete_lookup_table(lookup_tableT *lookup_table);
size_t addr = 0;
lookup_tableT lookup_table;
Dwarf_Debug tDbg;
size_t addr = 0;
lookup_tableT lookup_table;
Dwarf_Debug tDbg;
static TdThreadOnce traceThreadInit = PTHREAD_ONCE_INIT;
void endTrace() {
......@@ -157,7 +158,7 @@ void endTrace() {
}
}
void startTrace() {
int ret;
int ret;
Dwarf_Ptr errarg = 0;
FILE *fp = fopen("/proc/self/maps", "r");
......@@ -178,7 +179,7 @@ void startTrace() {
atexit(endTrace);
}
static void print_line(Dwarf_Debug dbg, Dwarf_Line line, Dwarf_Addr pc) {
char *linesrc = "??";
char *linesrc = "??";
Dwarf_Unsigned lineno = 0;
if (line) {
......@@ -189,18 +190,18 @@ static void print_line(Dwarf_Debug dbg, Dwarf_Line line, Dwarf_Addr pc) {
if (line) dwarf_dealloc(dbg, linesrc, DW_DLA_STRING);
}
void taosPrintBackTrace() {
int size = 20;
void **buffer[20];
int size = 20;
void **buffer[20];
Dwarf_Addr pc;
int32_t frame = 0;
void **ebp;
void **ret = NULL;
size_t func_frame_distance = 0;
int32_t frame = 0;
void **ebp;
void **ret = NULL;
size_t func_frame_distance = 0;
taosThreadOnce(&traceThreadInit, startTrace);
if (buffer != NULL && size > 0) {
ebp = taosGetEbp();
ebp = taosGetEbp();
func_frame_distance = (size_t)*ebp - (size_t)ebp;
while (ebp && frame < size && (func_frame_distance < (1ULL << 24)) && (func_frame_distance > 0)) {
ret = ebp + 1;
......
source/os/src/osProc.c
浏览文件 @ fcbaa608
......@@ -44,11 +44,11 @@ void taosWaitProc(int32_t pid) {
#endif
}
void taosKillProc(int32_t pid) {
void taosKillProc(int32_t pid) {
#ifdef WINDOWS
assert(0);
#else
kill(pid, SIGINT);
kill(pid, SIGINT);
#endif
}
......
source/os/src/osRand.c
浏览文件 @ fcbaa608
......@@ -16,8 +16,8 @@
#define _DEFAULT_SOURCE
#include "os.h"
#ifdef WINDOWS
#include "windows.h"
#include "wincrypt.h"
#include "windows.h"
#else
#include <sys/file.h>
#include <unistd.h>
......@@ -27,17 +27,17 @@ void taosSeedRand(uint32_t seed) { return srand(seed); }
uint32_t taosRand(void) { return rand(); }
uint32_t taosRandR(uint32_t *pSeed) {
uint32_t taosRandR(uint32_t* pSeed) {
#ifdef WINDOWS
return rand_s(pSeed);
return rand_s(pSeed);
#else
return rand_r(pSeed);
return rand_r(pSeed);
#endif
}
uint32_t taosSafeRand(void) {
#ifdef WINDOWS
uint32_t seed = taosRand();
uint32_t seed = taosRand();
HCRYPTPROV hCryptProv;
if (!CryptAcquireContext(&hCryptProv, NULL, NULL, PROV_RSA_FULL, 0)) {
if (!CryptAcquireContext(&hCryptProv, NULL, NULL, PROV_RSA_FULL, CRYPT_NEWKEYSET)) {
......@@ -51,7 +51,7 @@ uint32_t taosSafeRand(void) {
return seed;
#else
TdFilePtr pFile;
int seed;
int seed;
pFile = taosOpenFile("/dev/urandom", TD_FILE_READ);
if (pFile == NULL) {
......
source/os/src/osSignal.c
浏览文件 @ fcbaa608
......@@ -47,7 +47,7 @@ void taosDflSignal(int32_t signum) {
signal(signum, SIG_DFL);
}
void taosKillChildOnParentStopped() { }
void taosKillChildOnParentStopped() {}
#else
......@@ -73,9 +73,9 @@ void taosIgnSignal(int32_t signum) { signal(signum, SIG_IGN); }
void taosDflSignal(int32_t signum) { signal(signum, SIG_DFL); }
void taosKillChildOnParentStopped() {
void taosKillChildOnParentStopped() {
#ifndef _TD_DARWIN_64
prctl(PR_SET_PDEATHSIG, SIGKILL);
prctl(PR_SET_PDEATHSIG, SIGKILL);
#endif
}
......
source/os/src/osSleep.c
浏览文件 @ fcbaa608
......@@ -17,14 +17,13 @@
#define _DEFAULT_SOURCE
#include "os.h"
#if !(defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32))
#include <unistd.h>
#endif
void taosSsleep(int32_t s) {
#ifdef WINDOWS
Sleep(1000 * s);
Sleep(1000 * s);
#else
sleep(s);
#endif
......@@ -32,7 +31,7 @@ void taosSsleep(int32_t s) {
void taosMsleep(int32_t ms) {
#ifdef WINDOWS
Sleep(ms);
Sleep(ms);
#else
usleep(ms * 1000);
#endif
......@@ -40,14 +39,14 @@ void taosMsleep(int32_t ms) {
void taosUsleep(int32_t us) {
#ifdef WINDOWS
HANDLE timer;
LARGE_INTEGER interval;
interval.QuadPart = (10 * us);
HANDLE timer;
LARGE_INTEGER interval;
interval.QuadPart = (10 * us);
timer = CreateWaitableTimer(NULL, TRUE, NULL);
SetWaitableTimer(timer, &interval, 0, NULL, NULL, 0);
WaitForSingleObject(timer, INFINITE);
CloseHandle(timer);
timer = CreateWaitableTimer(NULL, TRUE, NULL);
SetWaitableTimer(timer, &interval, 0, NULL, NULL, 0);
WaitForSingleObject(timer, INFINITE);
CloseHandle(timer);
#else
usleep(us);
#endif
......
source/os/src/osString.c
浏览文件 @ fcbaa608
......@@ -26,10 +26,10 @@ extern int wcswidth(const wchar_t *s, size_t n);
#ifdef WINDOWS
char *strsep(char **stringp, const char *delim) {
char * s;
char *s;
const char *spanp;
int32_t c, sc;
char * tok;
char *tok;
if ((s = *stringp) == NULL) return (NULL);
for (tok = s;;) {
c = *s++;
......@@ -50,7 +50,7 @@ char *strsep(char **stringp, const char *delim) {
/* Duplicate a string, up to at most size characters */
char *strndup(const char *s, int size) {
size_t l;
char * s2;
char *s2;
l = strlen(s);
if (l > size) l = size;
s2 = malloc(l + 1);
......@@ -139,7 +139,7 @@ typedef struct {
int8_t inUse;
} SConv;
SConv *gConv = NULL;
SConv *gConv = NULL;
int32_t convUsed = 0;
int32_t gConvMaxNum = 0;
......@@ -167,7 +167,7 @@ iconv_t taosAcquireConv(int32_t *idx) {
*idx = -1;
return iconv_open(DEFAULT_UNICODE_ENCODEC, tsCharset);
}
while (true) {
int32_t used = atomic_add_fetch_32(&convUsed, 1);
if (used > gConvMaxNum) {
......@@ -175,7 +175,7 @@ iconv_t taosAcquireConv(int32_t *idx) {
sched_yield();
continue;
}
break;
}
......
source/os/src/osSysinfo.c
浏览文件 @ fcbaa608
......@@ -136,7 +136,7 @@ static int32_t taosGetSysCpuInfo(SysCpuInfo *cpuInfo) {
FILETIME idleTime;
FILETIME kernelTime;
FILETIME userTime;
bool res = GetSystemTimes(&idleTime, &kernelTime, &userTime);
bool res = GetSystemTimes(&idleTime, &kernelTime, &userTime);
if (res) {
cpuInfo->idle = CompareFileTime(&pre_idleTime, &idleTime);
cpuInfo->system = CompareFileTime(&pre_kernelTime, &kernelTime);
......@@ -174,14 +174,14 @@ static int32_t taosGetProcCpuInfo(ProcCpuInfo *cpuInfo) {
#ifdef WINDOWS
FILETIME pre_krnlTm = {0};
FILETIME pre_usrTm = {0};
FILETIME creatTm, exitTm, krnlTm, usrTm;
FILETIME creatTm, exitTm, krnlTm, usrTm;
if (GetThreadTimes(GetCurrentThread(), &creatTm, &exitTm, &krnlTm, &usrTm)) {
if (GetThreadTimes(GetCurrentThread(), &creatTm, &exitTm, &krnlTm, &usrTm)) {
cpuInfo->stime = CompareFileTime(&pre_krnlTm, &krnlTm);
cpuInfo->utime = CompareFileTime(&pre_usrTm, &usrTm);
cpuInfo->cutime = 0;
cpuInfo->cstime = 0;
}
}
#elif defined(DARWIN)
cpuInfo->stime = 0;
cpuInfo->utime = 0;
......@@ -214,7 +214,6 @@ static int32_t taosGetProcCpuInfo(ProcCpuInfo *cpuInfo) {
return 0;
}
bool taosCheckSystemIsLittleEnd() {
union check {
int16_t i;
......@@ -336,12 +335,13 @@ int32_t taosGetOsReleaseName(char *releaseName, int32_t maxLen) {
int32_t taosGetCpuInfo(char *cpuModel, int32_t maxLen, float *numOfCores) {
#ifdef WINDOWS
char value[100];
char value[100];
DWORD bufferSize = sizeof(value);
RegGetValue(HKEY_LOCAL_MACHINE, "HARDWARE\\DESCRIPTION\\System\\CentralProcessor\\0", "ProcessorNameString", RRF_RT_ANY, NULL, (PVOID)&value, &bufferSize);
RegGetValue(HKEY_LOCAL_MACHINE, "HARDWARE\\DESCRIPTION\\System\\CentralProcessor\\0", "ProcessorNameString",
RRF_RT_ANY, NULL, (PVOID)&value, &bufferSize);
tstrncpy(cpuModel, value, maxLen);
SYSTEM_INFO si;
memset(&si,0,sizeof(SYSTEM_INFO));
memset(&si, 0, sizeof(SYSTEM_INFO));
GetSystemInfo(&si);
*numOfCores = si.dwNumberOfProcessors;
return 0;
......@@ -416,10 +416,10 @@ int32_t taosGetCpuInfo(char *cpuModel, int32_t maxLen, float *numOfCores) {
}
if ((done & 2) == 0) {
*numOfCores = coreCount;
done |= 2;
*numOfCores = coreCount;
done |= 2;
}
return code;
#endif
}
......@@ -702,7 +702,7 @@ int32_t taosGetCardInfo(int64_t *receive_bytes, int64_t *transmit_bytes) {
if (pFile == NULL) return -1;
ssize_t _bytes = 0;
char line[1024];
char line[1024];
while (!taosEOFFile(pFile)) {
int64_t o_rbytes = 0;
......@@ -776,13 +776,14 @@ int32_t taosGetSystemUUID(char *uid, int32_t uidlen) {
#ifdef WINDOWS
GUID guid;
CoCreateGuid(&guid);
snprintf(uid, uidlen, "%08X-%04X-%04X-%02X%02X-%02X%02X%02X%02X%02X%02X", guid.Data1, guid.Data2, guid.Data3, guid.Data4[0],
guid.Data4[1], guid.Data4[2], guid.Data4[3], guid.Data4[4], guid.Data4[5], guid.Data4[6], guid.Data4[7]);
snprintf(uid, uidlen, "%08X-%04X-%04X-%02X%02X-%02X%02X%02X%02X%02X%02X", guid.Data1, guid.Data2, guid.Data3,
guid.Data4[0], guid.Data4[1], guid.Data4[2], guid.Data4[3], guid.Data4[4], guid.Data4[5], guid.Data4[6],
guid.Data4[7]);
return 0;
#elif defined(_TD_DARWIN_64)
uuid_t uuid = {0};
char buf[37] = {0};
char buf[37] = {0};
uuid_generate(uuid);
// it's caller's responsibility to make enough space for `uid`, that's 36-char + 1-null
uuid_unparse_lower(uuid, buf);
......@@ -929,7 +930,7 @@ void taosSetCoreDump(bool enable) {
SysNameInfo taosGetSysNameInfo() {
#ifdef WINDOWS
SysNameInfo info = {0};
DWORD dwVersion = GetVersion();
DWORD dwVersion = GetVersion();
char *tmp = NULL;
tmp = getenv("OS");
......@@ -971,14 +972,17 @@ SysNameInfo taosGetSysNameInfo() {
#endif
}
bool taosCheckCurrentInDll() {
#ifdef WINDOWS
MEMORY_BASIC_INFORMATION mbi;
char path[PATH_MAX] = {0};
GetModuleFileName(((VirtualQuery(taosCheckCurrentInDll,&mbi,sizeof(mbi)) != 0) ? (HMODULE)mbi.AllocationBase : NULL), path, PATH_MAX);
char path[PATH_MAX] = {0};
GetModuleFileName(
((VirtualQuery(taosCheckCurrentInDll, &mbi, sizeof(mbi)) != 0) ? (HMODULE)mbi.AllocationBase : NULL), path,
PATH_MAX);
int strLastIndex = strlen(path);
if ((path[strLastIndex-3] == 'd' || path[strLastIndex-3] == 'D') && (path[strLastIndex-2] == 'l' || path[strLastIndex-2] == 'L') && (path[strLastIndex-1] == 'l' || path[strLastIndex-1] == 'L')) {
if ((path[strLastIndex - 3] == 'd' || path[strLastIndex - 3] == 'D') &&
(path[strLastIndex - 2] == 'l' || path[strLastIndex - 2] == 'L') &&
(path[strLastIndex - 1] == 'l' || path[strLastIndex - 1] == 'L')) {
return true;
}
return false;
......
source/os/src/osSystem.c
浏览文件 @ fcbaa608
......@@ -18,52 +18,52 @@
#include "os.h"
#if defined(WINDOWS)
typedef void (*MainWindows)(int argc,char** argv);
typedef void (*MainWindows)(int argc, char** argv);
MainWindows mainWindowsFunc = NULL;
SERVICE_STATUS ServiceStatus;
SERVICE_STATUS ServiceStatus;
SERVICE_STATUS_HANDLE hServiceStatusHandle;
void WINAPI windowsServiceCtrlHandle(DWORD request) {
switch (request) {
case SERVICE_CONTROL_STOP:
case SERVICE_CONTROL_SHUTDOWN:
raise(SIGINT);
ServiceStatus.dwCurrentState = SERVICE_STOP_PENDING;
if (!SetServiceStatus(hServiceStatusHandle, &ServiceStatus)) {
DWORD nError = GetLastError();
printf("failed to send stopped status to windows service: %d",nError);
}
break;
default:
return;
}
void WINAPI windowsServiceCtrlHandle(DWORD request) {
switch (request) {
case SERVICE_CONTROL_STOP:
case SERVICE_CONTROL_SHUTDOWN:
raise(SIGINT);
ServiceStatus.dwCurrentState = SERVICE_STOP_PENDING;
if (!SetServiceStatus(hServiceStatusHandle, &ServiceStatus)) {
DWORD nError = GetLastError();
printf("failed to send stopped status to windows service: %d", nError);
}
break;
default:
return;
}
}
void WINAPI mainWindowsService(int argc,char** argv) {
int ret = 0;
ServiceStatus.dwServiceType = SERVICE_WIN32;
ServiceStatus.dwControlsAccepted = SERVICE_ACCEPT_PAUSE_CONTINUE | SERVICE_ACCEPT_STOP | SERVICE_ACCEPT_SHUTDOWN;
ServiceStatus.dwCurrentState = SERVICE_START_PENDING;
ServiceStatus.dwWin32ExitCode = 0;
ServiceStatus.dwCheckPoint = 0;
ServiceStatus.dwWaitHint = 0;
ServiceStatus.dwServiceSpecificExitCode = 0;
hServiceStatusHandle = RegisterServiceCtrlHandler("taosd", &windowsServiceCtrlHandle);
if (hServiceStatusHandle == 0) {
DWORD nError = GetLastError();
printf("failed to register windows service ctrl handler: %d",nError);
}
ServiceStatus.dwCurrentState = SERVICE_RUNNING;
if (SetServiceStatus(hServiceStatusHandle, &ServiceStatus)) {
DWORD nError = GetLastError();
printf("failed to send running status to windows service: %d",nError);
}
void WINAPI mainWindowsService(int argc, char** argv) {
int ret = 0;
ServiceStatus.dwServiceType = SERVICE_WIN32;
ServiceStatus.dwControlsAccepted = SERVICE_ACCEPT_PAUSE_CONTINUE | SERVICE_ACCEPT_STOP | SERVICE_ACCEPT_SHUTDOWN;
ServiceStatus.dwCurrentState = SERVICE_START_PENDING;
ServiceStatus.dwWin32ExitCode = 0;
ServiceStatus.dwCheckPoint = 0;
ServiceStatus.dwWaitHint = 0;
ServiceStatus.dwServiceSpecificExitCode = 0;
hServiceStatusHandle = RegisterServiceCtrlHandler("taosd", &windowsServiceCtrlHandle);
if (hServiceStatusHandle == 0) {
DWORD nError = GetLastError();
printf("failed to register windows service ctrl handler: %d", nError);
}
ServiceStatus.dwCurrentState = SERVICE_RUNNING;
if (SetServiceStatus(hServiceStatusHandle, &ServiceStatus)) {
DWORD nError = GetLastError();
printf("failed to send running status to windows service: %d", nError);
}
if (mainWindowsFunc != NULL) mainWindowsFunc(argc, argv);
ServiceStatus.dwCurrentState = SERVICE_STOPPED;
if (!SetServiceStatus(hServiceStatusHandle, &ServiceStatus)) {
DWORD nError = GetLastError();
printf("failed to send stopped status to windows service: %d",nError);
}
if (!SetServiceStatus(hServiceStatusHandle, &ServiceStatus)) {
DWORD nError = GetLastError();
printf("failed to send stopped status to windows service: %d", nError);
}
}
void stratWindowsService(MainWindows mainWindows) {
mainWindowsFunc = mainWindows;
......@@ -248,7 +248,7 @@ TdCmdPtr taosOpenCmd(const char* cmd) {
#endif
}
int64_t taosGetsCmd(TdCmdPtr pCmd, int32_t maxSize, char *__restrict buf) {
int64_t taosGetsCmd(TdCmdPtr pCmd, int32_t maxSize, char* __restrict buf) {
if (pCmd == NULL || buf == NULL) {
return -1;
}
......
source/os/src/osThread.c
浏览文件 @ fcbaa608
......@@ -17,179 +17,135 @@
#include <pthread.h>
#include "os.h"
int32_t taosThreadCreate(TdThread * tid, const TdThreadAttr * attr, void *(*start)(void *), void *arg) {
int32_t taosThreadCreate(TdThread *tid, const TdThreadAttr *attr, void *(*start)(void *), void *arg) {
return pthread_create(tid, attr, start, arg);
}
int32_t taosThreadAttrDestroy(TdThreadAttr * attr) {
return pthread_attr_destroy(attr);
}
int32_t taosThreadAttrDestroy(TdThreadAttr *attr) { return pthread_attr_destroy(attr); }
int32_t taosThreadAttrGetDetachState(const TdThreadAttr * attr, int32_t *detachstate) {
int32_t taosThreadAttrGetDetachState(const TdThreadAttr *attr, int32_t *detachstate) {
return pthread_attr_getdetachstate(attr, detachstate);
}
int32_t taosThreadAttrGetInheritSched(const TdThreadAttr * attr, int32_t *inheritsched) {
int32_t taosThreadAttrGetInheritSched(const TdThreadAttr *attr, int32_t *inheritsched) {
return pthread_attr_getinheritsched(attr, inheritsched);
}
int32_t taosThreadAttrGetSchedParam(const TdThreadAttr * attr, struct sched_param *param) {
int32_t taosThreadAttrGetSchedParam(const TdThreadAttr *attr, struct sched_param *param) {
return pthread_attr_getschedparam(attr, param);
}
int32_t taosThreadAttrGetSchedPolicy(const TdThreadAttr * attr, int32_t *policy) {
int32_t taosThreadAttrGetSchedPolicy(const TdThreadAttr *attr, int32_t *policy) {
return pthread_attr_getschedpolicy(attr, policy);
}
int32_t taosThreadAttrGetScope(const TdThreadAttr * attr, int32_t *contentionscope) {
int32_t taosThreadAttrGetScope(const TdThreadAttr *attr, int32_t *contentionscope) {
return pthread_attr_getscope(attr, contentionscope);
}
int32_t taosThreadAttrGetStackSize(const TdThreadAttr * attr, size_t * stacksize) {
int32_t taosThreadAttrGetStackSize(const TdThreadAttr *attr, size_t *stacksize) {
return pthread_attr_getstacksize(attr, stacksize);
}
int32_t taosThreadAttrInit(TdThreadAttr * attr) {
return pthread_attr_init(attr);
}
int32_t taosThreadAttrInit(TdThreadAttr *attr) { return pthread_attr_init(attr); }
int32_t taosThreadAttrSetDetachState(TdThreadAttr * attr, int32_t detachstate) {
int32_t taosThreadAttrSetDetachState(TdThreadAttr *attr, int32_t detachstate) {
return pthread_attr_setdetachstate(attr, detachstate);
}
int32_t taosThreadAttrSetInheritSched(TdThreadAttr * attr, int32_t inheritsched) {
int32_t taosThreadAttrSetInheritSched(TdThreadAttr *attr, int32_t inheritsched) {
return pthread_attr_setinheritsched(attr, inheritsched);
}
int32_t taosThreadAttrSetSchedParam(TdThreadAttr * attr, const struct sched_param *param) {
int32_t taosThreadAttrSetSchedParam(TdThreadAttr *attr, const struct sched_param *param) {
return pthread_attr_setschedparam(attr, param);
}
int32_t taosThreadAttrSetSchedPolicy(TdThreadAttr * attr, int32_t policy) {
int32_t taosThreadAttrSetSchedPolicy(TdThreadAttr *attr, int32_t policy) {
return pthread_attr_setschedpolicy(attr, policy);
}
int32_t taosThreadAttrSetScope(TdThreadAttr * attr, int32_t contentionscope) {
int32_t taosThreadAttrSetScope(TdThreadAttr *attr, int32_t contentionscope) {
return pthread_attr_setscope(attr, contentionscope);
}
int32_t taosThreadAttrSetStackSize(TdThreadAttr * attr, size_t stacksize) {
int32_t taosThreadAttrSetStackSize(TdThreadAttr *attr, size_t stacksize) {
return pthread_attr_setstacksize(attr, stacksize);
}
int32_t taosThreadCancel(TdThread thread) {
return pthread_cancel(thread);
}
int32_t taosThreadCancel(TdThread thread) { return pthread_cancel(thread); }
int32_t taosThreadCondDestroy(TdThreadCond * cond) {
return pthread_cond_destroy(cond);
}
int32_t taosThreadCondDestroy(TdThreadCond *cond) { return pthread_cond_destroy(cond); }
int32_t taosThreadCondInit(TdThreadCond * cond, const TdThreadCondAttr * attr) {
return pthread_cond_init(cond, attr);
}
int32_t taosThreadCondInit(TdThreadCond *cond, const TdThreadCondAttr *attr) { return pthread_cond_init(cond, attr); }
int32_t taosThreadCondSignal(TdThreadCond * cond) {
return pthread_cond_signal(cond);
}
int32_t taosThreadCondSignal(TdThreadCond *cond) { return pthread_cond_signal(cond); }
int32_t taosThreadCondBroadcast(TdThreadCond * cond) {
return pthread_cond_broadcast(cond);
}
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) { return pthread_cond_wait(cond, mutex); }
int32_t taosThreadCondTimedWait(TdThreadCond * cond, TdThreadMutex * mutex, const struct timespec *abstime) {
int32_t taosThreadCondTimedWait(TdThreadCond *cond, TdThreadMutex *mutex, const struct timespec *abstime) {
return pthread_cond_timedwait(cond, mutex, abstime);
}
int32_t taosThreadCondAttrDestroy(TdThreadCondAttr * attr) {
return pthread_condattr_destroy(attr);
}
int32_t taosThreadCondAttrDestroy(TdThreadCondAttr *attr) { return pthread_condattr_destroy(attr); }
int32_t taosThreadCondAttrGetPshared(const TdThreadCondAttr * attr, int32_t *pshared) {
int32_t taosThreadCondAttrGetPshared(const TdThreadCondAttr *attr, int32_t *pshared) {
return pthread_condattr_getpshared(attr, pshared);
}
int32_t taosThreadCondAttrInit(TdThreadCondAttr * attr) {
return pthread_condattr_init(attr);
}
int32_t taosThreadCondAttrInit(TdThreadCondAttr *attr) { return pthread_condattr_init(attr); }
int32_t taosThreadCondAttrSetPshared(TdThreadCondAttr * attr, int32_t pshared) {
int32_t taosThreadCondAttrSetPshared(TdThreadCondAttr *attr, int32_t pshared) {
return pthread_condattr_setpshared(attr, pshared);
}
int32_t taosThreadDetach(TdThread thread) {
return pthread_detach(thread);
}
int32_t taosThreadDetach(TdThread thread) { return pthread_detach(thread); }
int32_t taosThreadEqual(TdThread t1, TdThread t2) {
return pthread_equal(t1, t2);
}
int32_t taosThreadEqual(TdThread t1, TdThread t2) { return pthread_equal(t1, t2); }
void taosThreadExit(void *valuePtr) {
return pthread_exit(valuePtr);
}
void taosThreadExit(void *valuePtr) { return pthread_exit(valuePtr); }
int32_t taosThreadGetSchedParam(TdThread thread, int32_t *policy, struct sched_param *param) {
return pthread_getschedparam(thread, policy, param);
}
void *taosThreadGetSpecific(TdThreadKey key) {
return pthread_getspecific(key);
}
void *taosThreadGetSpecific(TdThreadKey key) { return pthread_getspecific(key); }
int32_t taosThreadJoin(TdThread thread, void **valuePtr) {
return pthread_join(thread, valuePtr);
}
int32_t taosThreadJoin(TdThread thread, void **valuePtr) { return pthread_join(thread, valuePtr); }
int32_t taosThreadKeyCreate(TdThreadKey * key, void(*destructor)(void *)) {
int32_t taosThreadKeyCreate(TdThreadKey *key, void (*destructor)(void *)) {
return pthread_key_create(key, destructor);
}
int32_t taosThreadKeyDelete(TdThreadKey key) {
return pthread_key_delete(key);
}
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 taosThreadKill(TdThread thread, int32_t sig) { return pthread_kill(thread, sig); }
// int32_t taosThreadMutexConsistent(TdThreadMutex* mutex) {
// return pthread_mutex_consistent(mutex);
// }
int32_t taosThreadMutexDestroy(TdThreadMutex * mutex) {
return pthread_mutex_destroy(mutex);
}
int32_t taosThreadMutexDestroy(TdThreadMutex *mutex) { return pthread_mutex_destroy(mutex); }
int32_t taosThreadMutexInit(TdThreadMutex * mutex, const TdThreadMutexAttr * attr) {
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) { 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) { return pthread_mutex_trylock(mutex); }
int32_t taosThreadMutexUnlock(TdThreadMutex * mutex) {
return pthread_mutex_unlock(mutex);
}
int32_t taosThreadMutexUnlock(TdThreadMutex *mutex) { return pthread_mutex_unlock(mutex); }
int32_t taosThreadMutexAttrDestroy(TdThreadMutexAttr * attr) {
return pthread_mutexattr_destroy(attr);
}
int32_t taosThreadMutexAttrDestroy(TdThreadMutexAttr *attr) { return pthread_mutexattr_destroy(attr); }
int32_t taosThreadMutexAttrGetPshared(const TdThreadMutexAttr * attr, int32_t *pshared) {
int32_t taosThreadMutexAttrGetPshared(const TdThreadMutexAttr *attr, int32_t *pshared) {
return pthread_mutexattr_getpshared(attr, pshared);
}
......@@ -197,15 +153,13 @@ int32_t taosThreadMutexAttrGetPshared(const TdThreadMutexAttr * attr, int32_t *p
// return pthread_mutexattr_getrobust(attr, robust);
// }
int32_t taosThreadMutexAttrGetType(const TdThreadMutexAttr * attr, int32_t *kind) {
int32_t taosThreadMutexAttrGetType(const TdThreadMutexAttr *attr, int32_t *kind) {
return pthread_mutexattr_gettype(attr, kind);
}
int32_t taosThreadMutexAttrInit(TdThreadMutexAttr * attr) {
return pthread_mutexattr_init(attr);
}
int32_t taosThreadMutexAttrInit(TdThreadMutexAttr *attr) { return pthread_mutexattr_init(attr); }
int32_t taosThreadMutexAttrSetPshared(TdThreadMutexAttr * attr, int32_t pshared) {
int32_t taosThreadMutexAttrSetPshared(TdThreadMutexAttr *attr, int32_t pshared) {
return pthread_mutexattr_setpshared(attr, pshared);
}
......@@ -213,25 +167,21 @@ int32_t taosThreadMutexAttrSetPshared(TdThreadMutexAttr * attr, int32_t pshared)
// return pthread_mutexattr_setrobust(attr, robust);
// }
int32_t taosThreadMutexAttrSetType(TdThreadMutexAttr * attr, int32_t kind) {
int32_t taosThreadMutexAttrSetType(TdThreadMutexAttr *attr, int32_t kind) {
return pthread_mutexattr_settype(attr, kind);
}
int32_t taosThreadOnce(TdThreadOnce * onceControl, void(*initRoutine)(void)) {
int32_t taosThreadOnce(TdThreadOnce *onceControl, void (*initRoutine)(void)) {
return pthread_once(onceControl, initRoutine);
}
int32_t taosThreadRwlockDestroy(TdThreadRwlock * rwlock) {
return pthread_rwlock_destroy(rwlock);
}
int32_t taosThreadRwlockDestroy(TdThreadRwlock *rwlock) { return pthread_rwlock_destroy(rwlock); }
int32_t taosThreadRwlockInit(TdThreadRwlock * rwlock, const TdThreadRwlockAttr * attr) {
int32_t taosThreadRwlockInit(TdThreadRwlock *rwlock, const TdThreadRwlockAttr *attr) {
return pthread_rwlock_init(rwlock, attr);
}
int32_t taosThreadRwlockRdlock(TdThreadRwlock * rwlock) {
return pthread_rwlock_rdlock(rwlock);
}
int32_t taosThreadRwlockRdlock(TdThreadRwlock *rwlock) { return pthread_rwlock_rdlock(rwlock); }
// int32_t taosThreadRwlockTimedRdlock(TdThreadRwlock * rwlock, const struct timespec *abstime) {
// return pthread_rwlock_timedrdlock(rwlock, abstime);
......@@ -241,103 +191,79 @@ int32_t taosThreadRwlockRdlock(TdThreadRwlock * rwlock) {
// return pthread_rwlock_timedwrlock(rwlock, abstime);
// }
int32_t taosThreadRwlockTryRdlock(TdThreadRwlock * rwlock) {
return pthread_rwlock_tryrdlock(rwlock);
}
int32_t taosThreadRwlockTryRdlock(TdThreadRwlock *rwlock) { return pthread_rwlock_tryrdlock(rwlock); }
int32_t taosThreadRwlockTryWrlock(TdThreadRwlock * rwlock) {
return pthread_rwlock_trywrlock(rwlock);
}
int32_t taosThreadRwlockTryWrlock(TdThreadRwlock *rwlock) { return pthread_rwlock_trywrlock(rwlock); }
int32_t taosThreadRwlockUnlock(TdThreadRwlock * rwlock) {
return pthread_rwlock_unlock(rwlock);
}
int32_t taosThreadRwlockUnlock(TdThreadRwlock *rwlock) { return pthread_rwlock_unlock(rwlock); }
int32_t taosThreadRwlockWrlock(TdThreadRwlock * rwlock) {
return pthread_rwlock_wrlock(rwlock);
}
int32_t taosThreadRwlockWrlock(TdThreadRwlock *rwlock) { return pthread_rwlock_wrlock(rwlock); }
int32_t taosThreadRwlockAttrDestroy(TdThreadRwlockAttr * attr) {
return pthread_rwlockattr_destroy(attr);
}
int32_t taosThreadRwlockAttrDestroy(TdThreadRwlockAttr *attr) { return pthread_rwlockattr_destroy(attr); }
int32_t taosThreadRwlockAttrGetPshared(const TdThreadRwlockAttr * attr, int32_t *pshared) {
int32_t taosThreadRwlockAttrGetPshared(const TdThreadRwlockAttr *attr, int32_t *pshared) {
return pthread_rwlockattr_getpshared(attr, pshared);
}
int32_t taosThreadRwlockAttrInit(TdThreadRwlockAttr * attr) {
return pthread_rwlockattr_init(attr);
}
int32_t taosThreadRwlockAttrInit(TdThreadRwlockAttr *attr) { return pthread_rwlockattr_init(attr); }
int32_t taosThreadRwlockAttrSetPshared(TdThreadRwlockAttr * attr, int32_t pshared) {
int32_t taosThreadRwlockAttrSetPshared(TdThreadRwlockAttr *attr, int32_t pshared) {
return pthread_rwlockattr_setpshared(attr, pshared);
}
TdThread taosThreadSelf(void) {
return pthread_self();
}
TdThread taosThreadSelf(void) { return pthread_self(); }
int32_t taosThreadSetCancelState(int32_t state, int32_t *oldstate) {
return pthread_setcancelstate(state, oldstate);
}
int32_t taosThreadSetCancelState(int32_t state, int32_t *oldstate) { return pthread_setcancelstate(state, oldstate); }
int32_t taosThreadSetCancelType(int32_t type, int32_t *oldtype) {
return pthread_setcanceltype(type, oldtype);
}
int32_t taosThreadSetCancelType(int32_t type, int32_t *oldtype) { return pthread_setcanceltype(type, oldtype); }
int32_t taosThreadSetSchedParam(TdThread thread, int32_t policy, const struct sched_param *param) {
return pthread_setschedparam(thread, policy, param);
}
int32_t taosThreadSetSpecific(TdThreadKey key, const void *value) {
return pthread_setspecific(key, value);
}
int32_t taosThreadSetSpecific(TdThreadKey key, const void *value) { return pthread_setspecific(key, value); }
int32_t taosThreadSpinDestroy(TdThreadSpinlock * lock) {
int32_t taosThreadSpinDestroy(TdThreadSpinlock *lock) {
#ifdef TD_USE_SPINLOCK_AS_MUTEX
return pthread_mutex_destroy((pthread_mutex_t*)lock);
return pthread_mutex_destroy((pthread_mutex_t *)lock);
#else
return pthread_spin_destroy((pthread_spinlock_t*)lock);
return pthread_spin_destroy((pthread_spinlock_t *)lock);
#endif
}
int32_t taosThreadSpinInit(TdThreadSpinlock * lock, int32_t pshared) {
int32_t taosThreadSpinInit(TdThreadSpinlock *lock, int32_t pshared) {
#ifdef TD_USE_SPINLOCK_AS_MUTEX
assert(pshared == 0);
return pthread_mutex_init((pthread_mutex_t*)lock, NULL);
return pthread_mutex_init((pthread_mutex_t *)lock, NULL);
#else
return pthread_spin_init((pthread_spinlock_t*)lock, pshared);
return pthread_spin_init((pthread_spinlock_t *)lock, pshared);
#endif
}
int32_t taosThreadSpinLock(TdThreadSpinlock * lock) {
int32_t taosThreadSpinLock(TdThreadSpinlock *lock) {
#ifdef TD_USE_SPINLOCK_AS_MUTEX
return pthread_mutex_lock((pthread_mutex_t*)lock);
return pthread_mutex_lock((pthread_mutex_t *)lock);
#else
return pthread_spin_lock((pthread_spinlock_t*)lock);
return pthread_spin_lock((pthread_spinlock_t *)lock);
#endif
}
int32_t taosThreadSpinTrylock(TdThreadSpinlock * lock) {
int32_t taosThreadSpinTrylock(TdThreadSpinlock *lock) {
#ifdef TD_USE_SPINLOCK_AS_MUTEX
return pthread_mutex_trylock((pthread_mutex_t*)lock);
return pthread_mutex_trylock((pthread_mutex_t *)lock);
#else
return pthread_spin_trylock((pthread_spinlock_t*)lock);
return pthread_spin_trylock((pthread_spinlock_t *)lock);
#endif
}
int32_t taosThreadSpinUnlock(TdThreadSpinlock * lock) {
int32_t taosThreadSpinUnlock(TdThreadSpinlock *lock) {
#ifdef TD_USE_SPINLOCK_AS_MUTEX
return pthread_mutex_unlock((pthread_mutex_t*)lock);
return pthread_mutex_unlock((pthread_mutex_t *)lock);
#else
return pthread_spin_unlock((pthread_spinlock_t*)lock);
return pthread_spin_unlock((pthread_spinlock_t *)lock);
#endif
}
void taosThreadTestCancel(void) {
return pthread_testcancel();
}
void taosThreadTestCancel(void) { return pthread_testcancel(); }
void taosThreadClear(TdThread *thread) {
memset(thread, 0, sizeof(TdThread));
}
\ No newline at end of file
void taosThreadClear(TdThread *thread) { memset(thread, 0, sizeof(TdThread)); }
\ No newline at end of file
source/os/src/osTimer.c
浏览文件 @ fcbaa608
......@@ -44,7 +44,7 @@ static MMRESULT timerId;
static void (*timer_callback)(int);
static int timer_ms = 0;
static TdThread timer_thread;
static TdThread timer_thread;
static int timer_kq = -1;
static volatile int timer_stop = 0;
......@@ -83,11 +83,11 @@ static void taosDeleteTimer(void *tharg) {
timer_delete(*pTimer);
}
static TdThread timerThread;
static TdThread timerThread;
static timer_t timerId;
static volatile bool stopTimer = false;
static void * taosProcessAlarmSignal(void *tharg) {
// Block the signal
static void *taosProcessAlarmSignal(void *tharg) {
// Block the signal
sigset_t sigset;
sigemptyset(&sigset);
sigaddset(&sigset, SIGALRM);
......@@ -109,7 +109,7 @@ static void * taosProcessAlarmSignal(void *tharg) {
sevent.sigev_signo = SIGALRM;
if (timer_create(CLOCK_REALTIME, &sevent, &timerId) == -1) {
// printf("Failed to create timer");
// printf("Failed to create timer");
}
taosThreadCleanupPush(taosDeleteTimer, &timerId);
......@@ -121,19 +121,19 @@ static void * taosProcessAlarmSignal(void *tharg) {
ts.it_interval.tv_nsec = 1000000 * MSECONDS_PER_TICK;
if (timer_settime(timerId, 0, &ts, NULL)) {
// printf("Failed to init timer");
// printf("Failed to init timer");
return NULL;
}
int signo;
while (!stopTimer) {
if (sigwait(&sigset, &signo)) {
// printf("Failed to wait signal: number %d", signo);
if (sigwait(&sigset, &signo)) {
// printf("Failed to wait signal: number %d", signo);
continue;
}
/* //printf("Signal handling: number %d ......\n", signo); */
/* //printf("Signal handling: number %d ......\n", signo); */
callback(0);
callback(0);
}
taosThreadCleanupPop(1);
......@@ -167,7 +167,8 @@ int taosInitTimer(void (*callback)(int), int ms) {
r = taosThreadCreate(&timer_thread, NULL, timer_routine, NULL);
if (r) {
fprintf(stderr, "==%s[%d]%s()==failed to create timer thread\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__);
fprintf(stderr, "==%s[%d]%s()==failed to create timer thread\n", taosDirEntryBaseName(__FILE__), __LINE__,
__func__);
// since no caller of this func checks the return value for the moment
abort();
}
......
source/os/src/osTimezone.c
浏览文件 @ fcbaa608
此差异已折叠。
source/util/src/talgo.c
浏览文件 @ fcbaa608
......@@ -175,11 +175,11 @@ void *taosbsearch(const void *key, const void *base, int32_t nmemb, int32_t size
c = compar(key, p);
if (c == 0) {
if (flags == TD_GT){
if (flags == TD_GT) {
lidx = midx + 1;
} else if(flags == TD_LT){
} else if (flags == TD_LT) {
ridx = midx - 1;
}else{
} else {
break;
}
} else if (c < 0) {
......
source/util/src/tbloomfilter.c
浏览文件 @ fcbaa608
......@@ -19,7 +19,7 @@
#include "taos.h"
#include "taoserror.h"
#define UNIT_NUM_BITS 64
#define UNIT_NUM_BITS 64
#define UNIT_ADDR_NUM_BITS 6
static FORCE_INLINE bool setBit(uint64_t *buf, uint64_t index) {
......@@ -51,12 +51,12 @@ SBloomFilter *tBloomFilterInit(uint64_t expectedEntries, double errorRate) {
// ln(2)^2 = 0.480453013918201
// m = - n * ln(P) / ( ln(2) )^2
// m is the size of bloom filter, n is expected entries, P is false positive probability
pBF->numUnits = (uint64_t) ceil(expectedEntries * lnRate / 0.480453013918201 / UNIT_NUM_BITS);
pBF->numUnits = (uint64_t)ceil(expectedEntries * lnRate / 0.480453013918201 / UNIT_NUM_BITS);
pBF->numBits = pBF->numUnits * 64;
pBF->size = 0;
// ln(2) = 0.693147180559945
pBF->hashFunctions = (uint32_t) ceil(lnRate / 0.693147180559945);
pBF->hashFunctions = (uint32_t)ceil(lnRate / 0.693147180559945);
pBF->hashFn1 = taosGetDefaultHashFunction(TSDB_DATA_TYPE_TIMESTAMP);
pBF->hashFn2 = taosGetDefaultHashFunction(TSDB_DATA_TYPE_NCHAR);
pBF->buffer = taosMemoryCalloc(pBF->numUnits, sizeof(uint64_t));
......@@ -69,11 +69,11 @@ SBloomFilter *tBloomFilterInit(uint64_t expectedEntries, double errorRate) {
int32_t tBloomFilterPut(SBloomFilter *pBF, const void *keyBuf, uint32_t len) {
ASSERT(!tBloomFilterIsFull(pBF));
uint64_t h1 = (uint64_t)pBF->hashFn1(keyBuf, len);
uint64_t h2 = (uint64_t)pBF->hashFn2(keyBuf, len);
bool hasChange = false;
uint64_t h1 = (uint64_t)pBF->hashFn1(keyBuf, len);
uint64_t h2 = (uint64_t)pBF->hashFn2(keyBuf, len);
bool hasChange = false;
const register uint64_t size = pBF->numBits;
uint64_t cbHash = h1;
uint64_t cbHash = h1;
for (uint64_t i = 0; i < pBF->hashFunctions; ++i) {
hasChange |= setBit(pBF->buffer, cbHash % size);
cbHash += h2;
......@@ -85,12 +85,11 @@ int32_t tBloomFilterPut(SBloomFilter *pBF, const void *keyBuf, uint32_t len) {
return TSDB_CODE_FAILED;
}
int32_t tBloomFilterNoContain(const SBloomFilter *pBF, const void *keyBuf,
uint32_t len) {
uint64_t h1 = (uint64_t)pBF->hashFn1(keyBuf, len);
uint64_t h2 = (uint64_t)pBF->hashFn2(keyBuf, len);
int32_t tBloomFilterNoContain(const SBloomFilter *pBF, const void *keyBuf, uint32_t len) {
uint64_t h1 = (uint64_t)pBF->hashFn1(keyBuf, len);
uint64_t h2 = (uint64_t)pBF->hashFn2(keyBuf, len);
const register uint64_t size = pBF->numBits;
uint64_t cbHash = h1;
uint64_t cbHash = h1;
for (uint64_t i = 0; i < pBF->hashFunctions; ++i) {
if (!getBit(pBF->buffer, cbHash % size)) {
return TSDB_CODE_SUCCESS;
......@@ -108,21 +107,21 @@ void tBloomFilterDestroy(SBloomFilter *pBF) {
taosMemoryFree(pBF);
}
int32_t tBloomFilterEncode(const SBloomFilter *pBF, SEncoder* pEncoder) {
int32_t tBloomFilterEncode(const SBloomFilter *pBF, SEncoder *pEncoder) {
if (tEncodeU32(pEncoder, pBF->hashFunctions) < 0) return -1;
if (tEncodeU64(pEncoder, pBF->expectedEntries) < 0) return -1;
if (tEncodeU64(pEncoder, pBF->numUnits) < 0) return -1;
if (tEncodeU64(pEncoder, pBF->numBits) < 0) return -1;
if (tEncodeU64(pEncoder, pBF->size) < 0) return -1;
for (uint64_t i = 0; i < pBF->numUnits; i++) {
uint64_t* pUnits = (uint64_t*)pBF->buffer;
uint64_t *pUnits = (uint64_t *)pBF->buffer;
if (tEncodeU64(pEncoder, pUnits[i]) < 0) return -1;
}
if (tEncodeDouble(pEncoder, pBF->errorRate) < 0) return -1;
return 0;
}
SBloomFilter* tBloomFilterDecode(SDecoder* pDecoder) {
SBloomFilter *tBloomFilterDecode(SDecoder *pDecoder) {
SBloomFilter *pBF = taosMemoryCalloc(1, sizeof(SBloomFilter));
pBF->buffer = NULL;
if (tDecodeU32(pDecoder, &pBF->hashFunctions) < 0) goto _error;
......@@ -132,7 +131,7 @@ SBloomFilter* tBloomFilterDecode(SDecoder* pDecoder) {
if (tDecodeU64(pDecoder, &pBF->size) < 0) goto _error;
pBF->buffer = taosMemoryCalloc(pBF->numUnits, sizeof(uint64_t));
for (int32_t i = 0; i < pBF->numUnits; i++) {
uint64_t* pUnits = (uint64_t*)pBF->buffer;
uint64_t *pUnits = (uint64_t *)pBF->buffer;
if (tDecodeU64(pDecoder, pUnits + i) < 0) goto _error;
}
if (tDecodeDouble(pDecoder, &pBF->errorRate) < 0) goto _error;
......@@ -145,6 +144,4 @@ _error:
return NULL;
}
bool tBloomFilterIsFull(const SBloomFilter *pBF) {
return pBF->size >= pBF->expectedEntries;
}
\ No newline at end of file
bool tBloomFilterIsFull(const SBloomFilter *pBF) { return pBF->size >= pBF->expectedEntries; }
\ No newline at end of file
source/util/src/tcache.c
浏览文件 @ fcbaa608
......@@ -15,8 +15,8 @@
#define _DEFAULT_SOURCE
#include "tcache.h"
#include "taoserror.h"
#include "osThread.h"
#include "taoserror.h"
#include "tlog.h"
#include "tutil.h"
......@@ -35,7 +35,7 @@ typedef struct SCacheNode {
uint64_t addedTime; // the added time when this element is added or updated into cache
uint64_t lifespan; // life duration when this element should be remove from cache
int64_t expireTime; // expire time
void* signature;
void *signature;
struct STrashElem *pTNodeHeader; // point to trash node head
uint16_t keyLen : 15; // max key size: 32kb
bool inTrashcan : 1; // denote if it is in trash or not
......@@ -279,7 +279,7 @@ static void removeNodeInEntryList(SCacheEntry *pe, SCacheNode *prev, SCacheNode
pNode->pNext = NULL;
pe->num -= 1;
ASSERT((pe->next && pe->num > 0) || (NULL == pe->next && pe->num == 0));
ASSERT((pe->next && pe->num > 0) || (NULL == pe->next && pe->num == 0));
}
static FORCE_INLINE SCacheEntry *doFindEntry(SCacheObj *pCacheObj, const void *key, size_t keyLen) {
......@@ -660,7 +660,7 @@ void doTraverseElems(SCacheObj *pCacheObj, bool (*fp)(void *param, SCacheNode *p
taosWLockLatch(&pEntry->latch);
SCacheNode **pPre = &pEntry->next;
SCacheNode *pNode = pEntry->next;
SCacheNode *pNode = pEntry->next;
while (pNode != NULL) {
SCacheNode *next = pNode->pNext;
......@@ -945,7 +945,7 @@ bool taosCacheIterNext(SCacheIter *pIter) {
char *p = pIter->pCurrent[i]->data;
taosCacheRelease(pCacheObj, (void **)&p, false);
pIter->pCurrent[i] = NULL;
}
}
if (pIter->entryIndex + 1 >= pCacheObj->capacity) {
return false;
......
source/util/src/tcompare.c
浏览文件 @ fcbaa608
此差异已折叠。
source/util/src/tconfig.c
浏览文件 @ fcbaa608
......@@ -15,13 +15,13 @@
#define _DEFAULT_SOURCE
#include "tconfig.h"
#include "cJSON.h"
#include "taoserror.h"
#include "tlog.h"
#include "tutil.h"
#include "tenv.h"
#include "cJSON.h"
#include "tjson.h"
#include "tgrant.h"
#include "tjson.h"
#include "tlog.h"
#include "tutil.h"
#define CFG_NAME_PRINT_LEN 24
#define CFG_SRC_PRINT_LEN 12
......@@ -508,7 +508,7 @@ const char *cfgDtypeStr(ECfgDataType type) {
}
}
void cfgDumpItemValue(SConfigItem *pItem, char* buf, int32_t bufSize, int32_t* pLen) {
void cfgDumpItemValue(SConfigItem *pItem, char *buf, int32_t bufSize, int32_t *pLen) {
int32_t len = 0;
switch (pItem->dtype) {
case CFG_DTYPE_BOOL:
......@@ -629,16 +629,16 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) {
}
int32_t cfgLoadFromEnvVar(SConfig *pConfig) {
char line[1024], *name, *value, *value2, *value3;
char line[1024], *name, *value, *value2, *value3;
int32_t olen, vlen, vlen2, vlen3;
int32_t code = 0;
char **pEnv = environ;
char **pEnv = environ;
line[1023] = 0;
while(*pEnv != NULL) {
while (*pEnv != NULL) {
name = value = value2 = value3 = NULL;
olen = vlen = vlen2 = vlen3 = 0;
strncpy(line, *pEnv, sizeof(line)-1);
strncpy(line, *pEnv, sizeof(line) - 1);
pEnv++;
taosEnvToCfg(line, line);
......@@ -676,12 +676,12 @@ int32_t cfgLoadFromEnvCmd(SConfig *pConfig, const char **envCmd) {
int32_t code = 0;
int32_t index = 0;
if (envCmd == NULL) return 0;
while (envCmd[index]!=NULL) {
strncpy(buf, envCmd[index], sizeof(buf)-1);
buf[sizeof(buf)-1] = 0;
while (envCmd[index] != NULL) {
strncpy(buf, envCmd[index], sizeof(buf) - 1);
buf[sizeof(buf) - 1] = 0;
taosEnvToCfg(buf, buf);
index++;
name = value = value2 = value3 = NULL;
olen = vlen = vlen2 = vlen3 = 0;
......@@ -720,13 +720,13 @@ int32_t cfgLoadFromEnvFile(SConfig *pConfig, const char *envFile) {
ssize_t _bytes = 0;
const char *filepath = ".env";
if (envFile != NULL && strlen(envFile)>0) {
if (envFile != NULL && strlen(envFile) > 0) {
if (!taosCheckExistFile(envFile)) {
uError("failed to load env file: %s", envFile);
return -1;
}
filepath = envFile;
}else {
} else {
if (!taosCheckExistFile(filepath)) {
uInfo("failed to load env file: %s", filepath);
return 0;
......@@ -747,7 +747,7 @@ int32_t cfgLoadFromEnvFile(SConfig *pConfig, const char *envFile) {
if (_bytes <= 0) {
break;
}
if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
if (line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
taosEnvToCfg(line, line);
paGetToken(line, &name, &olen);
......@@ -808,7 +808,7 @@ int32_t cfgLoadFromCfgFile(SConfig *pConfig, const char *filepath) {
break;
}
if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
if (line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
paGetToken(line, &name, &olen);
if (olen == 0) continue;
......@@ -918,7 +918,7 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
uInfo("fail to load apoll url");
return 0;
}
char *p = strchr(url, ':');
if (p == NULL) {
uError("fail to load apoll url: %s, unknown format", url);
......@@ -939,15 +939,15 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
return -1;
}
size_t fileSize = taosLSeekFile(pFile, 0, SEEK_END);
char *buf = taosMemoryMalloc(fileSize);
char *buf = taosMemoryMalloc(fileSize);
taosLSeekFile(pFile, 0, SEEK_SET);
if(taosReadFile(pFile, buf, fileSize) <= 0) {
if (taosReadFile(pFile, buf, fileSize) <= 0) {
taosCloseFile(&pFile);
uError("load json file error: %s", filepath);
return -1;
}
taosCloseFile(&pFile);
SJson* pJson = tjsonParse(buf);
SJson *pJson = tjsonParse(buf);
if (NULL == pJson) {
const char *jsonParseError = tjsonGetError();
if (jsonParseError != NULL) {
......@@ -958,8 +958,8 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
taosMemoryFreeClear(buf);
int32_t jsonArraySize = tjsonGetArraySize(pJson);
for(int32_t i = 0; i < jsonArraySize; i++) {
cJSON* item = tjsonGetArrayItem(pJson, i);
for (int32_t i = 0; i < jsonArraySize; i++) {
cJSON *item = tjsonGetArrayItem(pJson, i);
if (item == NULL) break;
char *itemName = NULL, *itemValueString = NULL;
tjsonGetObjectName(item, &itemName);
......@@ -971,7 +971,7 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
cfgLineBuf = taosMemoryMalloc(itemNameLen + itemValueStringLen + 2);
memcpy(cfgLineBuf, itemName, itemNameLen);
cfgLineBuf[itemNameLen] = ' ';
memcpy(&cfgLineBuf[itemNameLen+1], itemValueString, itemValueStringLen);
memcpy(&cfgLineBuf[itemNameLen + 1], itemValueString, itemValueStringLen);
cfgLineBuf[itemNameLen + itemValueStringLen + 1] = '\0';
paGetToken(cfgLineBuf, &name, &olen);
......@@ -999,8 +999,8 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
}
tjsonDelete(pJson);
// } else if (strncmp(url, "jsonUrl", 7) == 0) {
// } else if (strncmp(url, "etcdUrl", 7) == 0) {
// } else if (strncmp(url, "jsonUrl", 7) == 0) {
// } else if (strncmp(url, "etcdUrl", 7) == 0) {
} else {
uError("Unsupported url: %s", url);
return -1;
......@@ -1010,19 +1010,19 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
return 0;
}
int32_t cfgGetApollUrl(const char **envCmd, const char *envFile, char* apolloUrl) {
int32_t cfgGetApollUrl(const char **envCmd, const char *envFile, char *apolloUrl) {
int32_t index = 0;
if (envCmd == NULL) return 0;
while (envCmd[index]!=NULL) {
while (envCmd[index] != NULL) {
if (strncmp(envCmd[index], "TAOS_APOLLO_URL", 14) == 0) {
char *p = strchr(envCmd[index], '=');
if (p != NULL) {
p++;
if (*p == '\'') {
p++;
p[strlen(p)-1] = '\0';
p[strlen(p) - 1] = '\0';
}
memcpy(apolloUrl, p, TMIN(strlen(p)+1,PATH_MAX));
memcpy(apolloUrl, p, TMIN(strlen(p) + 1, PATH_MAX));
uInfo("get apollo url from env cmd success");
return 0;
}
......@@ -1033,8 +1033,8 @@ int32_t cfgGetApollUrl(const char **envCmd, const char *envFile, char* apolloUrl
char line[1024];
char **pEnv = environ;
line[1023] = 0;
while(*pEnv != NULL) {
strncpy(line, *pEnv, sizeof(line)-1);
while (*pEnv != NULL) {
strncpy(line, *pEnv, sizeof(line) - 1);
pEnv++;
if (strncmp(line, "TAOS_APOLLO_URL", 14) == 0) {
char *p = strchr(line, '=');
......@@ -1042,29 +1042,29 @@ int32_t cfgGetApollUrl(const char **envCmd, const char *envFile, char* apolloUrl
p++;
if (*p == '\'') {
p++;
p[strlen(p)-1] = '\0';
p[strlen(p) - 1] = '\0';
}
memcpy(apolloUrl, p, TMIN(strlen(p)+1,PATH_MAX));
uInfo("get apollo url from env variables success, apolloUrl=%s",apolloUrl);
memcpy(apolloUrl, p, TMIN(strlen(p) + 1, PATH_MAX));
uInfo("get apollo url from env variables success, apolloUrl=%s", apolloUrl);
return 0;
}
}
}
const char *filepath = ".env";
if (envFile != NULL && strlen(envFile)>0) {
if (envFile != NULL && strlen(envFile) > 0) {
if (!taosCheckExistFile(envFile)) {
uError("failed to load env file: %s", envFile);
return -1;
}
filepath = envFile;
}else {
} else {
if (!taosCheckExistFile(filepath)) {
uInfo("failed to load env file: %s", filepath);
return 0;
}
}
int64_t _bytes;
int64_t _bytes;
TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_READ | TD_FILE_STREAM);
if (pFile != NULL) {
while (!taosEOFFile(pFile)) {
......@@ -1072,16 +1072,16 @@ int32_t cfgGetApollUrl(const char **envCmd, const char *envFile, char* apolloUrl
if (_bytes <= 0) {
break;
}
if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
if (line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
if (strncmp(line, "TAOS_APOLLO_URL", 14) == 0) {
char *p = strchr(line, '=');
if (p != NULL) {
p++;
if (*p == '\'') {
p++;
p[strlen(p)-1] = '\0';
p[strlen(p) - 1] = '\0';
}
memcpy(apolloUrl, p, TMIN(strlen(p)+1,PATH_MAX));
memcpy(apolloUrl, p, TMIN(strlen(p) + 1, PATH_MAX));
taosCloseFile(&pFile);
uInfo("get apollo url from env file success");
return 0;
......
source/util/src/tcrc32c.c
浏览文件 @ fcbaa608
此差异已折叠。
source/util/src/tdigest.c
浏览文件 @ fcbaa608
......@@ -24,296 +24,273 @@
* Copyright (c) 2016, Usman Masood <usmanm at fastmail dot fm>
*/
#include "tdigest.h"
#include "os.h"
#include "osMath.h"
#include "tdigest.h"
#define INTERPOLATE(x, x0, x1) (((x) - (x0)) / ((x1) - (x0)))
//#define INTEGRATED_LOCATION(compression, q) ((compression) * (asin(2 * (q) - 1) + M_PI / 2) / M_PI)
#define INTEGRATED_LOCATION(compression, q) ((compression) * (asin(2 * (double)(q) - 1)/M_PI + (double)1/2))
#define FLOAT_EQ(f1, f2) (fabs((f1) - (f2)) <= FLT_EPSILON)
#define INTEGRATED_LOCATION(compression, q) ((compression) * (asin(2 * (double)(q)-1) / M_PI + (double)1 / 2))
#define FLOAT_EQ(f1, f2) (fabs((f1) - (f2)) <= FLT_EPSILON)
typedef struct SMergeArgs {
TDigest *t;
SCentroid *centroids;
int32_t idx;
double weight_so_far;
double k1;
double min;
double max;
}SMergeArgs;
void tdigestAutoFill(TDigest* t, int32_t compression) {
t->centroids = (SCentroid*)((char*)t + sizeof(TDigest));
t->buffered_pts = (SPt*) ((char*)t + sizeof(TDigest) + sizeof(SCentroid) * (int32_t)GET_CENTROID(compression));
TDigest *t;
SCentroid *centroids;
int32_t idx;
double weight_so_far;
double k1;
double min;
double max;
} SMergeArgs;
void tdigestAutoFill(TDigest *t, int32_t compression) {
t->centroids = (SCentroid *)((char *)t + sizeof(TDigest));
t->buffered_pts = (SPt *)((char *)t + sizeof(TDigest) + sizeof(SCentroid) * (int32_t)GET_CENTROID(compression));
}
TDigest *tdigestNewFrom(void* pBuf, int32_t compression) {
memset(pBuf, 0, (size_t)TDIGEST_SIZE(compression));
TDigest* t = (TDigest*)pBuf;
tdigestAutoFill(t, compression);
TDigest *tdigestNewFrom(void *pBuf, int32_t compression) {
memset(pBuf, 0, (size_t)TDIGEST_SIZE(compression));
TDigest *t = (TDigest *)pBuf;
tdigestAutoFill(t, compression);
t->compression = compression;
t->size = (int64_t)GET_CENTROID(compression);
t->threshold = (int32_t)GET_THRESHOLD(compression);
t->min = DOUBLE_MAX;
t->max = -DOUBLE_MAX;
t->compression = compression;
t->size = (int64_t)GET_CENTROID(compression);
t->threshold = (int32_t)GET_THRESHOLD(compression);
t->min = DOUBLE_MAX;
t->max = -DOUBLE_MAX;
return t;
return t;
}
static int32_t cmpCentroid(const void *a, const void *b) {
SCentroid *c1 = (SCentroid *) a;
SCentroid *c2 = (SCentroid *) b;
if (c1->mean < c2->mean)
return -1;
if (c1->mean > c2->mean)
return 1;
return 0;
SCentroid *c1 = (SCentroid *)a;
SCentroid *c2 = (SCentroid *)b;
if (c1->mean < c2->mean) return -1;
if (c1->mean > c2->mean) return 1;
return 0;
}
static void mergeCentroid(SMergeArgs *args, SCentroid *merge) {
double k2;
SCentroid *c = &args->centroids[args->idx];
args->weight_so_far += merge->weight;
k2 = INTEGRATED_LOCATION(args->t->size,
args->weight_so_far / args->t->total_weight);
//idx++
if(k2 - args->k1 > 1 && c->weight > 0) {
if(args->idx + 1 < args->t->size
&& merge->mean != args->centroids[args->idx].mean) {
args->idx++;
}
args->k1 = k2;
double k2;
SCentroid *c = &args->centroids[args->idx];
args->weight_so_far += merge->weight;
k2 = INTEGRATED_LOCATION(args->t->size, args->weight_so_far / args->t->total_weight);
// idx++
if (k2 - args->k1 > 1 && c->weight > 0) {
if (args->idx + 1 < args->t->size && merge->mean != args->centroids[args->idx].mean) {
args->idx++;
}
c = &args->centroids[args->idx];
if(c->mean == merge->mean) {
c->weight += merge->weight;
} else {
c->weight += merge->weight;
c->mean += (merge->mean - c->mean) * merge->weight / c->weight;
if (merge->weight > 0) {
args->min = TMIN(merge->mean, args->min);
args->max = TMAX(merge->mean, args->max);
}
args->k1 = k2;
}
c = &args->centroids[args->idx];
if (c->mean == merge->mean) {
c->weight += merge->weight;
} else {
c->weight += merge->weight;
c->mean += (merge->mean - c->mean) * merge->weight / c->weight;
if (merge->weight > 0) {
args->min = TMIN(merge->mean, args->min);
args->max = TMAX(merge->mean, args->max);
}
}
}
void tdigestCompress(TDigest *t) {
SCentroid *unmerged_centroids;
int64_t unmerged_weight = 0;
int32_t num_unmerged = t->num_buffered_pts;
int32_t i, j;
SMergeArgs args;
if (t->num_buffered_pts <= 0)
return;
unmerged_centroids = (SCentroid*)taosMemoryMalloc(sizeof(SCentroid) * t->num_buffered_pts);
for (i = 0; i < num_unmerged; i++) {
SPt *p = t->buffered_pts + i;
SCentroid *c = &unmerged_centroids[i];
c->mean = p->value;
c->weight = p->weight;
unmerged_weight += c->weight;
SCentroid *unmerged_centroids;
int64_t unmerged_weight = 0;
int32_t num_unmerged = t->num_buffered_pts;
int32_t i, j;
SMergeArgs args;
if (t->num_buffered_pts <= 0) return;
unmerged_centroids = (SCentroid *)taosMemoryMalloc(sizeof(SCentroid) * t->num_buffered_pts);
for (i = 0; i < num_unmerged; i++) {
SPt *p = t->buffered_pts + i;
SCentroid *c = &unmerged_centroids[i];
c->mean = p->value;
c->weight = p->weight;
unmerged_weight += c->weight;
}
t->num_buffered_pts = 0;
t->total_weight += unmerged_weight;
taosSort(unmerged_centroids, num_unmerged, sizeof(SCentroid), cmpCentroid);
memset(&args, 0, sizeof(SMergeArgs));
args.centroids = (SCentroid *)taosMemoryMalloc((size_t)(sizeof(SCentroid) * t->size));
memset(args.centroids, 0, (size_t)(sizeof(SCentroid) * t->size));
args.t = t;
args.min = DOUBLE_MAX;
args.max = -DOUBLE_MAX;
i = 0;
j = 0;
while (i < num_unmerged && j < t->num_centroids) {
SCentroid *a = &unmerged_centroids[i];
SCentroid *b = &t->centroids[j];
if (a->mean <= b->mean) {
mergeCentroid(&args, a);
assert(args.idx < t->size);
i++;
} else {
mergeCentroid(&args, b);
assert(args.idx < t->size);
j++;
}
t->num_buffered_pts = 0;
t->total_weight += unmerged_weight;
taosSort(unmerged_centroids, num_unmerged, sizeof(SCentroid), cmpCentroid);
memset(&args, 0, sizeof(SMergeArgs));
args.centroids = (SCentroid*)taosMemoryMalloc((size_t)(sizeof(SCentroid) * t->size));
memset(args.centroids, 0, (size_t)(sizeof(SCentroid) * t->size));
args.t = t;
args.min = DOUBLE_MAX;
args.max = -DOUBLE_MAX;
i = 0;
j = 0;
while (i < num_unmerged && j < t->num_centroids) {
SCentroid *a = &unmerged_centroids[i];
SCentroid *b = &t->centroids[j];
if (a->mean <= b->mean) {
mergeCentroid(&args, a);
assert(args.idx < t->size);
i++;
} else {
mergeCentroid(&args, b);
assert(args.idx < t->size);
j++;
}
}
while (i < num_unmerged) {
mergeCentroid(&args, &unmerged_centroids[i++]);
assert(args.idx < t->size);
}
taosMemoryFree((void *)unmerged_centroids);
while (j < t->num_centroids) {
mergeCentroid(&args, &t->centroids[j++]);
assert(args.idx < t->size);
}
if (t->total_weight > 0) {
t->min = TMIN(t->min, args.min);
if (args.centroids[args.idx].weight <= 0) {
args.idx--;
}
t->num_centroids = args.idx + 1;
t->max = TMAX(t->max, args.max);
}
while (i < num_unmerged) {
mergeCentroid(&args, &unmerged_centroids[i++]);
assert(args.idx < t->size);
}
taosMemoryFree((void*)unmerged_centroids);
memcpy(t->centroids, args.centroids, sizeof(SCentroid) * t->num_centroids);
taosMemoryFree((void *)args.centroids);
}
while (j < t->num_centroids) {
mergeCentroid(&args, &t->centroids[j++]);
assert(args.idx < t->size);
}
void tdigestAdd(TDigest *t, double x, int64_t w) {
if (w == 0) return;
if (t->total_weight > 0) {
t->min = TMIN(t->min, args.min);
if (args.centroids[args.idx].weight <= 0) {
args.idx--;
}
t->num_centroids = args.idx + 1;
t->max = TMAX(t->max, args.max);
}
int32_t i = t->num_buffered_pts;
if (i > 0 && t->buffered_pts[i - 1].value == x) {
t->buffered_pts[i].weight = w;
} else {
t->buffered_pts[i].value = x;
t->buffered_pts[i].weight = w;
t->num_buffered_pts++;
}
memcpy(t->centroids, args.centroids, sizeof(SCentroid) * t->num_centroids);
taosMemoryFree((void*)args.centroids);
if (t->num_buffered_pts >= t->threshold) tdigestCompress(t);
}
void tdigestAdd(TDigest* t, double x, int64_t w) {
if (w == 0)
return;
double tdigestCDF(TDigest *t, double x) {
if (t == NULL) return 0;
int32_t i = t->num_buffered_pts;
if(i > 0 && t->buffered_pts[i-1].value == x ) {
t->buffered_pts[i].weight = w;
} else {
t->buffered_pts[i].value = x;
t->buffered_pts[i].weight = w;
t->num_buffered_pts++;
}
int32_t i;
double left, right;
int64_t weight_so_far;
SCentroid *a, *b, tmp;
tdigestCompress(t);
if (t->num_centroids == 0) return NAN;
if (x < t->min) return 0;
if (x > t->max) return 1;
if (t->num_centroids == 1) {
if (FLOAT_EQ(t->max, t->min)) return 0.5;
if (t->num_buffered_pts >= t->threshold)
tdigestCompress(t);
}
return INTERPOLATE(x, t->min, t->max);
}
double tdigestCDF(TDigest *t, double x) {
if (t == NULL)
return 0;
int32_t i;
double left, right;
int64_t weight_so_far;
SCentroid *a, *b, tmp;
tdigestCompress(t);
if (t->num_centroids == 0)
return NAN;
if (x < t->min)
return 0;
if (x > t->max)
return 1;
if (t->num_centroids == 1) {
if (FLOAT_EQ(t->max, t->min))
return 0.5;
return INTERPOLATE(x, t->min, t->max);
}
weight_so_far = 0;
a = b = &tmp;
b->mean = t->min;
b->weight = 0;
right = 0;
weight_so_far = 0;
a = b = &tmp;
b->mean = t->min;
b->weight = 0;
right = 0;
for (i = 0; i < t->num_centroids; i++) {
SCentroid *c = &t->centroids[i];
left = b->mean - (a->mean + right);
a = b;
b = c;
right = (b->mean - a->mean) * a->weight / (a->weight + b->weight);
if (x < a->mean + right) {
double cdf = (weight_so_far
+ a->weight
* INTERPOLATE(x, a->mean - left, a->mean + right))
/ t->total_weight;
return TMAX(cdf, 0.0);
}
weight_so_far += a->weight;
}
for (i = 0; i < t->num_centroids; i++) {
SCentroid *c = &t->centroids[i];
left = b->mean - (a->mean + right);
a = b;
right = t->max - a->mean;
b = c;
right = (b->mean - a->mean) * a->weight / (a->weight + b->weight);
if (x < a->mean + right) {
return (weight_so_far + a->weight * INTERPOLATE(x, a->mean - left, a->mean + right))
/ t->total_weight;
double cdf = (weight_so_far + a->weight * INTERPOLATE(x, a->mean - left, a->mean + right)) / t->total_weight;
return TMAX(cdf, 0.0);
}
return 1;
weight_so_far += a->weight;
}
left = b->mean - (a->mean + right);
a = b;
right = t->max - a->mean;
if (x < a->mean + right) {
return (weight_so_far + a->weight * INTERPOLATE(x, a->mean - left, a->mean + right)) / t->total_weight;
}
return 1;
}
double tdigestQuantile(TDigest *t, double q) {
if (t == NULL)
return 0;
int32_t i;
double left, right, idx;
int64_t weight_so_far;
SCentroid *a, *b, tmp;
tdigestCompress(t);
if (t->num_centroids == 0)
return NAN;
if (t->num_centroids == 1)
return t->centroids[0].mean;
if (FLOAT_EQ(q, 0.0))
return t->min;
if (FLOAT_EQ(q, 1.0))
return t->max;
idx = q * t->total_weight;
weight_so_far = 0;
b = &tmp;
b->mean = t->min;
b->weight = 0;
right = t->min;
for (i = 0; i < t->num_centroids; i++) {
SCentroid *c = &t->centroids[i];
a = b;
left = right;
b = c;
right = (b->weight * a->mean + a->weight * b->mean)/ (a->weight + b->weight);
if (idx < weight_so_far + a->weight) {
double p = (idx - weight_so_far) / a->weight;
return left * (1 - p) + right * p;
}
weight_so_far += a->weight;
}
left = right;
if (t == NULL) return 0;
int32_t i;
double left, right, idx;
int64_t weight_so_far;
SCentroid *a, *b, tmp;
tdigestCompress(t);
if (t->num_centroids == 0) return NAN;
if (t->num_centroids == 1) return t->centroids[0].mean;
if (FLOAT_EQ(q, 0.0)) return t->min;
if (FLOAT_EQ(q, 1.0)) return t->max;
idx = q * t->total_weight;
weight_so_far = 0;
b = &tmp;
b->mean = t->min;
b->weight = 0;
right = t->min;
for (i = 0; i < t->num_centroids; i++) {
SCentroid *c = &t->centroids[i];
a = b;
right = t->max;
left = right;
if (idx < weight_so_far + a->weight && a->weight != 0) {
double p = (idx - weight_so_far) / a->weight;
return left * (1 - p) + right * p;
b = c;
right = (b->weight * a->mean + a->weight * b->mean) / (a->weight + b->weight);
if (idx < weight_so_far + a->weight) {
double p = (idx - weight_so_far) / a->weight;
return left * (1 - p) + right * p;
}
weight_so_far += a->weight;
}
left = right;
a = b;
right = t->max;
return t->max;
if (idx < weight_so_far + a->weight && a->weight != 0) {
double p = (idx - weight_so_far) / a->weight;
return left * (1 - p) + right * p;
}
return t->max;
}
void tdigestMerge(TDigest *t1, TDigest *t2) {
// SPoints
int32_t num_pts = t2->num_buffered_pts;
for(int32_t i = num_pts - 1; i >= 0; i--) {
SPt* p = t2->buffered_pts + i;
tdigestAdd(t1, p->value, p->weight);
t2->num_buffered_pts --;
}
// centroids
for (int32_t i = 0; i < t2->num_centroids; i++) {
tdigestAdd(t1, t2->centroids[i].mean, t2->centroids[i].weight);
}
// SPoints
int32_t num_pts = t2->num_buffered_pts;
for (int32_t i = num_pts - 1; i >= 0; i--) {
SPt *p = t2->buffered_pts + i;
tdigestAdd(t1, p->value, p->weight);
t2->num_buffered_pts--;
}
// centroids
for (int32_t i = 0; i < t2->num_centroids; i++) {
tdigestAdd(t1, t2->centroids[i].mean, t2->centroids[i].weight);
}
}
source/util/src/tenv.c
浏览文件 @ fcbaa608
......@@ -59,12 +59,12 @@ int32_t taosEnvToCfg(const char *envStr, char *cfgStr) {
}
if (cfgStr != envStr) strcpy(cfgStr, envStr);
char *p = strchr(cfgStr, '=');
if (p != NULL) {
char buf[CFG_NAME_MAX_LEN];
if (*(p+1) == '\'') {
*(p+1)= ' ';
char *pEnd = &cfgStr[strlen(cfgStr)-1];
if (*(p + 1) == '\'') {
*(p + 1) = ' ';
char *pEnd = &cfgStr[strlen(cfgStr) - 1];
if (*pEnd == '\'') *pEnd = '\0';
}
*p = '\0';
......
source/util/src/thash.c
浏览文件 @ fcbaa608
......@@ -333,7 +333,7 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, const vo
// disable resize
taosHashRLock(pHashObj);
uint32_t slot = HASH_INDEX(hashVal, pHashObj->capacity);
uint32_t slot = HASH_INDEX(hashVal, pHashObj->capacity);
SHashEntry *pe = pHashObj->hashList[slot];
taosHashEntryWLock(pHashObj, pe);
......@@ -890,5 +890,3 @@ void *taosHashAcquire(SHashObj *pHashObj, const void *key, size_t keyLen) {
void taosHashRelease(SHashObj *pHashObj, void *p) { taosHashCancelIterate(pHashObj, p); }
int64_t taosHashGetCompTimes(SHashObj *pHashObj) { return atomic_load_64(&pHashObj->compTimes); }
source/util/src/thashutil.c
浏览文件 @ fcbaa608
......@@ -14,13 +14,13 @@
*/
#define _DEFAULT_SOURCE
#include "thash.h"
#include "tcompare.h"
#include "thash.h"
#include "types.h"
#define ROTL32(x, r) ((x) << (r) | (x) >> (32u - (r)))
#define DLT (FLT_COMPAR_TOL_FACTOR * FLT_EPSILON)
#define DLT (FLT_COMPAR_TOL_FACTOR * FLT_EPSILON)
#define BASE 1000
#define FMIX32(h) \
......@@ -29,11 +29,12 @@
(h) *= 0x85ebca6b; \
(h) ^= (h) >> 13; \
(h) *= 0xc2b2ae35; \
(h) ^= (h) >> 16; } while (0)
(h) ^= (h) >> 16; \
} while (0)
uint32_t MurmurHash3_32(const char *key, uint32_t len) {
const uint8_t *data = (const uint8_t *)key;
const int32_t nblocks = len >> 2u;
const int32_t nblocks = len >> 2u;
uint32_t h1 = 0x12345678;
......@@ -80,14 +81,14 @@ uint32_t MurmurHash3_32(const char *key, uint32_t len) {
uint64_t MurmurHash3_64(const char *key, uint32_t len) {
const uint64_t m = 0x87c37b91114253d5;
const int r = 47;
uint32_t seed = 0x12345678;
uint64_t h = seed ^ (len * m);
const int r = 47;
uint32_t seed = 0x12345678;
uint64_t h = seed ^ (len * m);
const uint8_t *data = (const uint8_t *)key;
const uint8_t *end = data + (len-(len&7));
const uint8_t *end = data + (len - (len & 7));
while(data != end) {
uint64_t k = *((uint64_t*)data);
while (data != end) {
uint64_t k = *((uint64_t *)data);
k *= m;
k ^= k >> r;
......@@ -97,14 +98,21 @@ uint64_t MurmurHash3_64(const char *key, uint32_t len) {
data += 8;
}
switch(len & 7) {
case 7: h ^= (uint64_t)data[6] << 48; /* fall-thru */
case 6: h ^= (uint64_t)data[5] << 40; /* fall-thru */
case 5: h ^= (uint64_t)data[4] << 32; /* fall-thru */
case 4: h ^= (uint64_t)data[3] << 24; /* fall-thru */
case 3: h ^= (uint64_t)data[2] << 16; /* fall-thru */
case 2: h ^= (uint64_t)data[1] << 8; /* fall-thru */
case 1: h ^= (uint64_t)data[0];
switch (len & 7) {
case 7:
h ^= (uint64_t)data[6] << 48; /* fall-thru */
case 6:
h ^= (uint64_t)data[5] << 40; /* fall-thru */
case 5:
h ^= (uint64_t)data[4] << 32; /* fall-thru */
case 4:
h ^= (uint64_t)data[3] << 24; /* fall-thru */
case 3:
h ^= (uint64_t)data[2] << 16; /* fall-thru */
case 2:
h ^= (uint64_t)data[1] << 8; /* fall-thru */
case 1:
h ^= (uint64_t)data[0];
h *= m; /* fall-thru */
};
......@@ -126,11 +134,11 @@ uint32_t taosFloatHash(const char *key, uint32_t UNUSED_PARAM(len)) {
if (FLT_EQUAL(f, 0.0)) {
return 0;
}
if (fabs(f) < FLT_MAX/BASE - DLT) {
int32_t t = (int32_t)(round(BASE * (f + DLT)));
return (uint32_t)t;
if (fabs(f) < FLT_MAX / BASE - DLT) {
int32_t t = (int32_t)(round(BASE * (f + DLT)));
return (uint32_t)t;
} else {
return 0x7fc00000;
return 0x7fc00000;
}
}
uint32_t taosDoubleHash(const char *key, uint32_t UNUSED_PARAM(len)) {
......@@ -142,11 +150,11 @@ uint32_t taosDoubleHash(const char *key, uint32_t UNUSED_PARAM(len)) {
if (FLT_EQUAL(f, 0.0)) {
return 0;
}
if (fabs(f) < DBL_MAX/BASE - DLT) {
int32_t t = (int32_t)(round(BASE * (f + DLT)));
return (uint32_t)t;
if (fabs(f) < DBL_MAX / BASE - DLT) {
int32_t t = (int32_t)(round(BASE * (f + DLT)));
return (uint32_t)t;
} else {
return 0x7fc00000;
return 0x7fc00000;
}
}
uint32_t taosIntHash_64(const char *key, uint32_t UNUSED_PARAM(len)) {
......@@ -160,7 +168,7 @@ uint32_t taosIntHash_64(const char *key, uint32_t UNUSED_PARAM(len)) {
_hash_fn_t taosGetDefaultHashFunction(int32_t type) {
_hash_fn_t fn = NULL;
switch(type) {
switch (type) {
case TSDB_DATA_TYPE_TIMESTAMP:
case TSDB_DATA_TYPE_UBIGINT:
case TSDB_DATA_TYPE_BIGINT:
......@@ -210,9 +218,15 @@ int32_t taosDoubleEqual(const void *a, const void *b, size_t UNUSED_PARAM(sz)) {
_equal_fn_t taosGetDefaultEqualFunction(int32_t type) {
_equal_fn_t fn = NULL;
switch (type) {
case TSDB_DATA_TYPE_FLOAT: fn = taosFloatEqual; break;
case TSDB_DATA_TYPE_DOUBLE: fn = taosDoubleEqual; break;
default: fn = memcmp; break;
case TSDB_DATA_TYPE_FLOAT:
fn = taosFloatEqual;
break;
case TSDB_DATA_TYPE_DOUBLE:
fn = taosDoubleEqual;
break;
default:
fn = memcmp;
break;
}
return fn;
}
source/util/src/tmempool.c
浏览文件 @ fcbaa608
......@@ -19,12 +19,12 @@
#include "tutil.h"
typedef struct {
int32_t numOfFree; /* number of free slots */
int32_t first; /* the first free slot */
int32_t numOfBlock; /* the number of blocks */
int32_t blockSize; /* block size in bytes */
int32_t *freeList; /* the index list */
char *pool; /* the actual mem block */
int32_t numOfFree; /* number of free slots */
int32_t first; /* the first free slot */
int32_t numOfBlock; /* the number of blocks */
int32_t blockSize; /* block size in bytes */
int32_t *freeList; /* the index list */
char *pool; /* the actual mem block */
TdThreadMutex mutex;
} pool_t;
......
source/util/src/tpagedbuf.c
浏览文件 @ fcbaa608
......@@ -42,13 +42,14 @@ struct SDiskbasedBuf {
bool comp; // compressed before flushed to disk
uint64_t nextPos; // next page flush position
char* id; // for debug purpose
bool printStatis; // Print statistics info when closing this buffer.
char* id; // for debug purpose
bool printStatis; // Print statistics info when closing this buffer.
SDiskbasedBufStatis statis;
};
static int32_t createDiskFile(SDiskbasedBuf* pBuf) {
pBuf->pFile = taosOpenFile(pBuf->path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
pBuf->pFile =
taosOpenFile(pBuf->path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
if (pBuf->pFile == NULL) {
return TAOS_SYSTEM_ERROR(errno);
}
......@@ -247,12 +248,12 @@ static SPageInfo* registerPage(SDiskbasedBuf* pBuf, int32_t pageId) {
SPageInfo* ppi = taosMemoryMalloc(sizeof(SPageInfo));
ppi->pageId = pageId;
ppi->pData = NULL;
ppi->pData = NULL;
ppi->offset = -1;
ppi->length = -1;
ppi->used = true;
ppi->pn = NULL;
ppi->dirty = false;
ppi->used = true;
ppi->pn = NULL;
ppi->dirty = false;
return *(SPageInfo**)taosArrayPush(pBuf->pIdList, &ppi);
}
......@@ -318,7 +319,7 @@ static void lruListMoveToFront(SList* pList, SPageInfo* pi) {
}
static SPageInfo* getPageInfoFromPayload(void* page) {
char* p = (char *)page - POINTER_BYTES;
char* p = (char*)page - POINTER_BYTES;
SPageInfo* ppi = ((SPageInfo**)p)[0];
return ppi;
......@@ -334,12 +335,12 @@ int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMem
}
pPBuf->pageSize = pagesize;
pPBuf->numOfPages = 0; // all pages are in buffer in the first place
pPBuf->numOfPages = 0; // all pages are in buffer in the first place
pPBuf->totalBufSize = 0;
pPBuf->inMemPages = inMemBufSize / pagesize; // maximum allowed pages, it is a soft limit.
pPBuf->allocateId = -1;
pPBuf->pFile = NULL;
pPBuf->id = strdup(id);
pPBuf->pFile = NULL;
pPBuf->id = strdup(id);
pPBuf->fileSize = 0;
pPBuf->pFree = taosArrayInit(4, sizeof(SFreeListItem));
pPBuf->freePgList = tdListNew(POINTER_BYTES);
......@@ -402,14 +403,15 @@ void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
// allocate buf
if (availablePage == NULL) {
pi->pData = taosMemoryCalloc(1, getAllocPageSize(pBuf->pageSize)); // add extract bytes in case of zipped buffer increased.
pi->pData =
taosMemoryCalloc(1, getAllocPageSize(pBuf->pageSize)); // add extract bytes in case of zipped buffer increased.
} else {
pi->pData = availablePage;
}
((void**)pi->pData)[0] = pi;
#ifdef BUF_PAGE_DEBUG
uDebug("page_getNewBufPage , pi->pData:%p, pageId:%d, offset:%"PRId64, pi->pData, pi->pageId, pi->offset);
uDebug("page_getNewBufPage , pi->pData:%p, pageId:%d, offset:%" PRId64, pi->pData, pi->pageId, pi->offset);
#endif
return (void*)(GET_DATA_PAYLOAD(pi));
}
......@@ -434,11 +436,12 @@ void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
lruListMoveToFront(pBuf->lruList, (*pi));
(*pi)->used = true;
#ifdef BUF_PAGE_DEBUG
uDebug("page_getBufPage1 pageId:%d, offset:%"PRId64, (*pi)->pageId, (*pi)->offset);
uDebug("page_getBufPage1 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
#endif
return (void*)(GET_DATA_PAYLOAD(*pi));
} else { // not in memory
assert((*pi)->pData == NULL && (*pi)->pn == NULL && (((*pi)->length >= 0 && (*pi)->offset >= 0) || ((*pi)->length == -1 && (*pi)->offset == -1)));
assert((*pi)->pData == NULL && (*pi)->pn == NULL &&
(((*pi)->length >= 0 && (*pi)->offset >= 0) || ((*pi)->length == -1 && (*pi)->offset == -1)));
char* availablePage = NULL;
if (NO_IN_MEM_AVAILABLE_PAGES(pBuf)) {
......@@ -468,7 +471,7 @@ void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
}
}
#ifdef BUF_PAGE_DEBUG
uDebug("page_getBufPage2 pageId:%d, offset:%"PRId64, (*pi)->pageId, (*pi)->offset);
uDebug("page_getBufPage2 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
#endif
return (void*)(GET_DATA_PAYLOAD(*pi));
}
......@@ -482,7 +485,7 @@ void releaseBufPage(SDiskbasedBuf* pBuf, void* page) {
void releaseBufPageInfo(SDiskbasedBuf* pBuf, SPageInfo* pi) {
#ifdef BUF_PAGE_DEBUG
uDebug("page_releaseBufPageInfo pageId:%d, used:%d, offset:%"PRId64, pi->pageId, pi->used, pi->offset);
uDebug("page_releaseBufPageInfo pageId:%d, used:%d, offset:%" PRId64, pi->pageId, pi->used, pi->offset);
#endif
// assert(pi->pData != NULL && pi->used == true);
assert(pi->pData != NULL);
......@@ -520,12 +523,12 @@ void destroyDiskbasedBuf(SDiskbasedBuf* pBuf) {
{
SDiskbasedBufStatis* ps = &pBuf->statis;
if (ps->loadPages == 0) {
uDebug(
"Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages)",
ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f, ps->loadPages);
uDebug("Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages)", ps->getPages,
ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f, ps->loadPages);
} else {
uDebug(
"Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f Kb",
"Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f "
"Kb",
ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
}
......@@ -614,7 +617,8 @@ void dBufPrintStatis(const SDiskbasedBuf* pBuf) {
if (ps->loadPages > 0) {
printf(
"Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f Kb\n",
"Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f "
"Kb\n",
ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
} else {
......@@ -640,7 +644,7 @@ void clearDiskbasedBuf(SDiskbasedBuf* pBuf) {
taosHashClear(pBuf->all);
pBuf->numOfPages = 0; // all pages are in buffer in the first place
pBuf->numOfPages = 0; // all pages are in buffer in the first place
pBuf->totalBufSize = 0;
pBuf->allocateId = -1;
pBuf->fileSize = 0;
......
source/util/src/tscalablebf.c
浏览文件 @ fcbaa608
......@@ -18,11 +18,10 @@
#include "tscalablebf.h"
#include "taoserror.h"
#define DEFAULT_GROWTH 2
#define DEFAULT_GROWTH 2
#define DEFAULT_TIGHTENING_RATIO 0.5
static SBloomFilter *tScalableBfAddFilter(SScalableBf *pSBf, uint64_t expectedEntries,
double errorRate);
static SBloomFilter *tScalableBfAddFilter(SScalableBf *pSBf, uint64_t expectedEntries, double errorRate);
SScalableBf *tScalableBfInit(uint64_t expectedEntries, double errorRate) {
const uint32_t defaultSize = 8;
......@@ -35,7 +34,7 @@ SScalableBf *tScalableBfInit(uint64_t expectedEntries, double errorRate) {
}
pSBf->numBits = 0;
pSBf->bfArray = taosArrayInit(defaultSize, sizeof(void *));
if (tScalableBfAddFilter(pSBf, expectedEntries, errorRate * DEFAULT_TIGHTENING_RATIO) == NULL ) {
if (tScalableBfAddFilter(pSBf, expectedEntries, errorRate * DEFAULT_TIGHTENING_RATIO) == NULL) {
tScalableBfDestroy(pSBf);
return NULL;
}
......@@ -46,8 +45,7 @@ SScalableBf *tScalableBfInit(uint64_t expectedEntries, double errorRate) {
int32_t tScalableBfPut(SScalableBf *pSBf, const void *keyBuf, uint32_t len) {
int32_t size = taosArrayGetSize(pSBf->bfArray);
for (int32_t i = size - 2; i >= 0; --i) {
if (tBloomFilterNoContain(taosArrayGetP(pSBf->bfArray, i),
keyBuf, len) != TSDB_CODE_SUCCESS) {
if (tBloomFilterNoContain(taosArrayGetP(pSBf->bfArray, i), keyBuf, len) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_FAILED;
}
}
......@@ -55,8 +53,7 @@ int32_t tScalableBfPut(SScalableBf *pSBf, const void *keyBuf, uint32_t len) {
SBloomFilter *pNormalBf = taosArrayGetP(pSBf->bfArray, size - 1);
ASSERT(pNormalBf);
if (tBloomFilterIsFull(pNormalBf)) {
pNormalBf = tScalableBfAddFilter(pSBf,
pNormalBf->expectedEntries * pSBf->growth,
pNormalBf = tScalableBfAddFilter(pSBf, pNormalBf->expectedEntries * pSBf->growth,
pNormalBf->errorRate * DEFAULT_TIGHTENING_RATIO);
if (pNormalBf == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
......@@ -65,25 +62,22 @@ int32_t tScalableBfPut(SScalableBf *pSBf, const void *keyBuf, uint32_t len) {
return tBloomFilterPut(pNormalBf, keyBuf, len);
}
int32_t tScalableBfNoContain(const SScalableBf *pSBf, const void *keyBuf,
uint32_t len) {
int32_t tScalableBfNoContain(const SScalableBf *pSBf, const void *keyBuf, uint32_t len) {
int32_t size = taosArrayGetSize(pSBf->bfArray);
for (int32_t i = size - 1; i >= 0; --i) {
if (tBloomFilterNoContain(taosArrayGetP(pSBf->bfArray, i),
keyBuf, len) != TSDB_CODE_SUCCESS) {
if (tBloomFilterNoContain(taosArrayGetP(pSBf->bfArray, i), keyBuf, len) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_FAILED;
}
}
return TSDB_CODE_SUCCESS;
}
static SBloomFilter *tScalableBfAddFilter(SScalableBf *pSBf, uint64_t expectedEntries,
double errorRate) {
static SBloomFilter *tScalableBfAddFilter(SScalableBf *pSBf, uint64_t expectedEntries, double errorRate) {
SBloomFilter *pNormalBf = tBloomFilterInit(expectedEntries, errorRate);
if (pNormalBf == NULL) {
return NULL;
}
if(taosArrayPush(pSBf->bfArray, &pNormalBf) == NULL) {
if (taosArrayPush(pSBf->bfArray, &pNormalBf) == NULL) {
tBloomFilterDestroy(pNormalBf);
return NULL;
}
......@@ -101,7 +95,7 @@ void tScalableBfDestroy(SScalableBf *pSBf) {
taosMemoryFree(pSBf);
}
int32_t tScalableBfEncode(const SScalableBf *pSBf, SEncoder* pEncoder) {
int32_t tScalableBfEncode(const SScalableBf *pSBf, SEncoder *pEncoder) {
if (!pSBf) {
if (tEncodeI32(pEncoder, 0) < 0) return -1;
return 0;
......@@ -109,7 +103,7 @@ int32_t tScalableBfEncode(const SScalableBf *pSBf, SEncoder* pEncoder) {
int32_t size = taosArrayGetSize(pSBf->bfArray);
if (tEncodeI32(pEncoder, size) < 0) return -1;
for (int32_t i = 0; i < size; i++) {
SBloomFilter* pBF = taosArrayGetP(pSBf->bfArray, i);
SBloomFilter *pBF = taosArrayGetP(pSBf->bfArray, i);
if (tBloomFilterEncode(pBF, pEncoder) < 0) return -1;
}
if (tEncodeU32(pEncoder, pSBf->growth) < 0) return -1;
......@@ -117,7 +111,7 @@ int32_t tScalableBfEncode(const SScalableBf *pSBf, SEncoder* pEncoder) {
return 0;
}
SScalableBf* tScalableBfDecode(SDecoder* pDecoder) {
SScalableBf *tScalableBfDecode(SDecoder *pDecoder) {
SScalableBf *pSBf = taosMemoryCalloc(1, sizeof(SScalableBf));
pSBf->bfArray = NULL;
int32_t size = 0;
......@@ -128,7 +122,7 @@ SScalableBf* tScalableBfDecode(SDecoder* pDecoder) {
}
pSBf->bfArray = taosArrayInit(size * 2, sizeof(void *));
for (int32_t i = 0; i < size; i++) {
SBloomFilter* pBF = tBloomFilterDecode(pDecoder);
SBloomFilter *pBF = tBloomFilterDecode(pDecoder);
if (!pBF) goto _error;
taosArrayPush(pSBf->bfArray, &pBF);
}
......
source/util/src/tthread.c
浏览文件 @ fcbaa608
......@@ -17,7 +17,7 @@
#include "tthread.h"
TdThread* taosCreateThread(void* (*__start_routine)(void*), void* param) {
TdThread* pthread = (TdThread*)taosMemoryMalloc(sizeof(TdThread));
TdThread* pthread = (TdThread*)taosMemoryMalloc(sizeof(TdThread));
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
taosThreadAttrSetDetachState(&thattr, PTHREAD_CREATE_JOINABLE);
......
source/util/src/ttimer.c
浏览文件 @ fcbaa608
......@@ -16,9 +16,9 @@
#define _DEFAULT_SOURCE
#include "ttimer.h"
#include "taoserror.h"
#include "tdef.h"
#include "tlog.h"
#include "tsched.h"
#include "tdef.h"
#define tmrFatal(...) \
{ \
......@@ -104,21 +104,21 @@ typedef struct timer_map_t {
typedef struct time_wheel_t {
TdThreadMutex mutex;
int64_t nextScanAt;
uint32_t resolution;
uint16_t size;
uint16_t index;
tmr_obj_t** slots;
int64_t nextScanAt;
uint32_t resolution;
uint16_t size;
uint16_t index;
tmr_obj_t** slots;
} time_wheel_t;
static int32_t tsMaxTmrCtrl = TSDB_MAX_VNODES_PER_DB + 100;
static TdThreadOnce tmrModuleInit = PTHREAD_ONCE_INIT;
static TdThreadMutex tmrCtrlMutex;
static tmr_ctrl_t* tmrCtrls;
static tmr_ctrl_t* unusedTmrCtrl = NULL;
static void* tmrQhandle;
static int32_t numOfTmrCtrl = 0;
static tmr_ctrl_t* tmrCtrls;
static tmr_ctrl_t* unusedTmrCtrl = NULL;
static void* tmrQhandle;
static int32_t numOfTmrCtrl = 0;
int32_t taosTmrThreads = 1;
static uintptr_t nextTimerId = 0;
......@@ -133,7 +133,7 @@ static timer_map_t timerMap;
static uintptr_t getNextTimerId() {
uintptr_t id;
do {
id = (uintptr_t)atomic_add_fetch_ptr((void **)&nextTimerId, 1);
id = (uintptr_t)atomic_add_fetch_ptr((void**)&nextTimerId, 1);
} while (id == 0);
return id;
}
......
source/util/src/tworker.c
浏览文件 @ fcbaa608
......@@ -79,7 +79,7 @@ static void *tQWorkerThreadFp(SQWorker *worker) {
uDebug("worker:%s:%d is running", pool->name, worker->id);
while (1) {
if (taosReadQitemFromQset(pool->qset, (void **)&msg, &qinfo) == 0) {
if (taosReadQitemFromQset(pool->qset, (void **)&msg, &qinfo) == 0) {
uDebug("worker:%s:%d qset:%p, got no message and exiting", pool->name, worker->id, pool->qset);
break;
}
......
source/util/test/arrayTest.cpp
浏览文件 @ fcbaa608
......@@ -9,21 +9,21 @@
namespace {
static void remove_batch_test() {
SArray *pa = (SArray*) taosArrayInit(4, sizeof(int32_t));
SArray* pa = (SArray*)taosArrayInit(4, sizeof(int32_t));
for(int32_t i = 0; i < 20; ++i) {
for (int32_t i = 0; i < 20; ++i) {
int32_t a = i;
taosArrayPush(pa, &a);
}
SArray* delList = (SArray*)taosArrayInit(4, sizeof(int32_t));
taosArrayRemoveBatch(pa, (const int32_t*) TARRAY_GET_START(delList), taosArrayGetSize(delList));
taosArrayRemoveBatch(pa, (const int32_t*)TARRAY_GET_START(delList), taosArrayGetSize(delList));
EXPECT_EQ(taosArrayGetSize(pa), 20);
int32_t a = 5;
taosArrayPush(delList, &a);
taosArrayRemoveBatch(pa, (const int32_t*) TARRAY_GET_START(delList), taosArrayGetSize(delList));
taosArrayRemoveBatch(pa, (const int32_t*)TARRAY_GET_START(delList), taosArrayGetSize(delList));
EXPECT_EQ(taosArrayGetSize(pa), 19);
EXPECT_EQ(*(int*)taosArrayGet(pa, 5), 6);
......@@ -41,7 +41,7 @@ static void remove_batch_test() {
a = 14;
taosArrayPush(delList, &a);
taosArrayRemoveBatch(pa, (const int32_t*) TARRAY_GET_START(delList), taosArrayGetSize(delList));
taosArrayRemoveBatch(pa, (const int32_t*)TARRAY_GET_START(delList), taosArrayGetSize(delList));
EXPECT_EQ(taosArrayGetSize(pa), 17);
taosArrayDestroy(pa);
......@@ -49,24 +49,22 @@ static void remove_batch_test() {
}
} // namespace
TEST(arrayTest, array_list_test) {
remove_batch_test();
}
TEST(arrayTest, array_list_test) { remove_batch_test(); }
TEST(arrayTest, array_search_test) {
SArray *pa = (SArray*) taosArrayInit(4, sizeof(int32_t));
SArray* pa = (SArray*)taosArrayInit(4, sizeof(int32_t));
for(int32_t i = 10; i < 20; ++i) {
for (int32_t i = 10; i < 20; ++i) {
int32_t a = i;
taosArrayPush(pa, &a);
}
for(int i = 0; i < 30; i++) {
int32_t k = i;
for (int i = 0; i < 30; i++) {
int32_t k = i;
int32_t* pRet = (int32_t*)taosArraySearch(pa, &k, compareInt32Val, TD_GE);
int32_t idx = taosArraySearchIdx(pa, &k, compareInt32Val, TD_GE);
int32_t idx = taosArraySearchIdx(pa, &k, compareInt32Val, TD_GE);
if(pRet == NULL) {
if (pRet == NULL) {
ASSERT_EQ(idx, -1);
} else {
ASSERT_EQ(taosArrayGet(pa, idx), pRet);
......@@ -75,12 +73,11 @@ TEST(arrayTest, array_search_test) {
pRet = (int32_t*)taosArraySearch(pa, &k, compareInt32Val, TD_LE);
idx = taosArraySearchIdx(pa, &k, compareInt32Val, TD_LE);
if(pRet == NULL) {
if (pRet == NULL) {
ASSERT_EQ(idx, -1);
} else {
ASSERT_EQ(taosArrayGet(pa, idx), pRet);
}
}
taosArrayDestroy(pa);
......
source/util/test/hashTest.cpp
浏览文件 @ fcbaa608
此差异已折叠。
source/util/test/pageBufferTest.cpp
浏览文件 @ fcbaa608
此差异已折叠。
source/util/test/stringTest.cpp
浏览文件 @ fcbaa608
......@@ -19,7 +19,7 @@ TEST(testCase, string_dequote_test) {
EXPECT_EQ(3, len);
EXPECT_STRCASEEQ(t1, "abc");
char t21[] = " abc ";
char t21[] = " abc ";
int32_t lx = strtrim(t21);
EXPECT_STREQ("abc", t21);
......
source/util/test/trefTest.c
浏览文件 @ fcbaa608
此差异已折叠。
tools/scripts/codeFormat.sh
浏览文件 @ fcbaa608
......@@ -9,7 +9,8 @@ cd ${PRJ_ROOT_DIR}
FORMAT_DIR_LIST=(
"include"
# "source"
"source/os"
"source/util"
)
for d in ${FORMAT_DIR_LIST[@]}; do
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或