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

Merge pull request #17241 from taosdata/enh/optLogTime 

enh: opt log time
This commit is contained in:
Shengliang Guan 2022-10-10 09:59:37 +08:00 committed by GitHub
parent 9273ec149e 86c98f1f21
commit 60be9f5771
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 170 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/os/osTime.h
View File

@ -82,8 +82,9 @@ static FORCE_INLINE int64_t taosGetTimestampNs() {
return (int64_t)systemTime.tv_sec * 1000000000LL + (int64_t)systemTime.tv_nsec;
}
char * taosStrpTime(const char *buf, const char *fmt, struct tm *tm);
char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm);
struct tm *taosLocalTime(const time_t *timep, struct tm *result);
struct tm *taosLocalTimeNolock(struct tm *result, const time_t *timep, int dst);
time_t taosTime(time_t *t);
time_t taosMktime(struct tm *timep);

4
source/libs/index/src/indexFst.c
View File

@ -24,7 +24,7 @@ static FORCE_INLINE void fstPackDeltaIn(IdxFstFile* wrt, CompiledAddr nodeAddr,
CompiledAddr deltaAddr = (transAddr == EMPTY_ADDRESS) ? EMPTY_ADDRESS : nodeAddr - transAddr;
idxFilePackUintIn(wrt, deltaAddr, nBytes);
}
static FORCE_INLINE uint8_t fstPackDetla(IdxFstFile* wrt, CompiledAddr nodeAddr, CompiledAddr transAddr) {
static FORCE_INLINE uint8_t fstPackDelta(IdxFstFile* wrt, CompiledAddr nodeAddr, CompiledAddr transAddr) {
uint8_t nBytes = packDeltaSize(nodeAddr, transAddr);
fstPackDeltaIn(wrt, nodeAddr, transAddr, nBytes);
return nBytes;
@ -226,7 +226,7 @@ void fstStateCompileForOneTransNext(IdxFstFile* w, CompiledAddr addr, uint8_t in
void fstStateCompileForOneTrans(IdxFstFile* w, CompiledAddr addr, FstTransition* trn) {
Output out = trn->out;
uint8_t outPackSize = (out == 0 ? 0 : idxFilePackUint(w, out));
uint8_t transPackSize = fstPackDetla(w, addr, trn->addr);
uint8_t transPackSize = fstPackDelta(w, addr, trn->addr);
PackSizes packSizes = 0;
FST_SET_OUTPUT_PACK_SIZE(packSizes, outPackSize);

42
source/libs/transport/src/transCli.c
View File

@ -79,7 +79,8 @@ typedef struct SCliThrd {
uint64_t nextTimeout; // next timeout
void* pTransInst; //
SCvtAddr cvtAddr;
SHashObj* fqdn2ipCache;
SCvtAddr cvtAddr;
SCliMsg* stopMsg;
@ -135,6 +136,9 @@ static FORCE_INLINE void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr);
static FORCE_INLINE int32_t cliBuildExceptResp(SCliMsg* pMsg, STransMsg* resp);
static FORCE_INLINE uint32_t cliGetIpFromFqdnCache(SHashObj* cache, char* fqdn);
static FORCE_INLINE void cliUpdateFqdnCache(SHashObj* cache, char* fqdn);
// process data read from server, add decompress etc later
static void cliHandleResp(SCliConn* conn);
// handle except about conn
@ -154,7 +158,7 @@ static FORCE_INLINE int cliRBChoseIdx(STrans* pTransInst);
static FORCE_INLINE void transDestroyConnCtx(STransConnCtx* ctx);
// thread obj
static SCliThrd* createThrdObj();
static SCliThrd* createThrdObj(void* trans);
static void destroyThrdObj(SCliThrd* pThrd);
static void cliWalkCb(uv_handle_t* handle, void* arg);
@ -930,6 +934,21 @@ FORCE_INLINE int32_t cliBuildExceptResp(SCliMsg* pMsg, STransMsg* pResp) {
return 0;
}
static FORCE_INLINE uint32_t cliGetIpFromFqdnCache(SHashObj* cache, char* fqdn) {
uint32_t addr = 0;
uint32_t* v = taosHashGet(cache, fqdn, strlen(fqdn));
if (v == NULL) {
addr = taosGetIpv4FromFqdn(fqdn);
taosHashPut(cache, fqdn, strlen(fqdn), &addr, sizeof(addr));
} else {
addr = *v;
}
return addr;
}
static FORCE_INLINE void cliUpdateFqdnCache(SHashObj* cache, char* fqdn) {
// impl later
return;
}
void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
STrans* pTransInst = pThrd->pTransInst;
@ -985,7 +1004,8 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
struct sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = taosGetIpv4FromFqdn(conn->ip);
addr.sin_addr.s_addr = cliGetIpFromFqdnCache(pThrd->fqdn2ipCache, conn->ip);
addr.sin_port = (uint16_t)htons((uint16_t)conn->port);
tTrace("%s conn %p try to connect to %s:%d", pTransInst->label, conn, conn->ip, conn->port);
ret = uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, cliConnCb);
@ -1132,11 +1152,8 @@ void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
cli->pThreadObj = (SCliThrd**)taosMemoryCalloc(cli->numOfThreads, sizeof(SCliThrd*));
for (int i = 0; i < cli->numOfThreads; i++) {
SCliThrd* pThrd = createThrdObj();
pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
pThrd->pTransInst = shandle;
int err = taosThreadCreate(&pThrd->thread, NULL, cliWorkThread, (void*)(pThrd));
SCliThrd* pThrd = createThrdObj(shandle);
int err = taosThreadCreate(&pThrd->thread, NULL, cliWorkThread, (void*)(pThrd));
if (err == 0) {
tDebug("success to create tranport-cli thread:%d", i);
}
@ -1164,7 +1181,9 @@ static FORCE_INLINE void destroyCmsg(void* arg) {
taosMemoryFree(pMsg);
}
static SCliThrd* createThrdObj() {
static SCliThrd* createThrdObj(void* trans) {
STrans* pTransInst = trans;
SCliThrd* pThrd = (SCliThrd*)taosMemoryCalloc(1, sizeof(SCliThrd));
QUEUE_INIT(&pThrd->msg);
@ -1193,6 +1212,10 @@ static SCliThrd* createThrdObj() {
transDQCreate(pThrd->loop, &pThrd->timeoutQueue);
pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
pThrd->pTransInst = trans;
pThrd->fqdn2ipCache = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
pThrd->quit = false;
return pThrd;
}
@ -1217,6 +1240,7 @@ static void destroyThrdObj(SCliThrd* pThrd) {
taosArrayDestroy(pThrd->timerList);
taosMemoryFree(pThrd->prepare);
taosMemoryFree(pThrd->loop);
taosHashCleanup(pThrd->fqdn2ipCache);
taosMemoryFree(pThrd);
}

129
source/os/src/osTime.c
View File

@ -359,15 +359,15 @@ time_t taosTime(time_t *t) { return time(t); }
time_t taosMktime(struct tm *timep) {
#ifdef WINDOWS
struct tm tm1 = {0};
LARGE_INTEGER t;
FILETIME f;
SYSTEMTIME s;
FILETIME ff;
SYSTEMTIME ss;
LARGE_INTEGER offset;
struct tm tm1 = {0};
LARGE_INTEGER t;
FILETIME f;
SYSTEMTIME s;
FILETIME ff;
SYSTEMTIME ss;
LARGE_INTEGER offset;
time_t tt = 0;
time_t tt = 0;
localtime_s(&tm1, &tt);
ss.wYear = tm1.tm_year + 1900;
ss.wMonth = tm1.tm_mon + 1;
@ -394,11 +394,11 @@ time_t taosMktime(struct tm *timep) {
t.QuadPart |= f.dwLowDateTime;
t.QuadPart -= offset.QuadPart;
return (time_t)(t.QuadPart / 10000000);
return (time_t)(t.QuadPart / 10000000);
#else
return mktime(timep);
#endif
}
}
struct tm *taosLocalTime(const time_t *timep, struct tm *result) {
if (result == NULL) {
@ -406,8 +406,8 @@ struct tm *taosLocalTime(const time_t *timep, struct tm *result) {
}
#ifdef WINDOWS
if (*timep < 0) {
SYSTEMTIME ss,s;
FILETIME ff,f;
SYSTEMTIME ss, s;
FILETIME ff, f;
LARGE_INTEGER offset;
struct tm tm1;
time_t tt = 0;
@ -431,8 +431,8 @@ struct tm *taosLocalTime(const time_t *timep, struct tm *result) {
result->tm_min = s.wMinute;
result->tm_hour = s.wHour;
result->tm_mday = s.wDay;
result->tm_mon = s.wMonth-1;
result->tm_year = s.wYear-1900;
result->tm_mon = s.wMonth - 1;
result->tm_year = s.wYear - 1900;
result->tm_wday = s.wDayOfWeek;
result->tm_yday = 0;
result->tm_isdst = 0;
@ -445,6 +445,103 @@ struct tm *taosLocalTime(const time_t *timep, struct tm *result) {
return result;
}
static int isLeapYear(time_t year) {
if (year % 4)
return 0;
else if (year % 100)
return 1;
else if (year % 400)
return 0;
else
return 1;
}
struct tm *taosLocalTimeNolock(struct tm *result, const time_t *timep, int dst) {
if (result == NULL) {
return localtime(timep);
}
#ifdef WINDOWS
if (*timep < 0) {
SYSTEMTIME ss, s;
FILETIME ff, f;
LARGE_INTEGER offset;
struct tm tm1;
time_t tt = 0;
localtime_s(&tm1, &tt);
ss.wYear = tm1.tm_year + 1900;
ss.wMonth = tm1.tm_mon + 1;
ss.wDay = tm1.tm_mday;
ss.wHour = tm1.tm_hour;
ss.wMinute = tm1.tm_min;
ss.wSecond = tm1.tm_sec;
ss.wMilliseconds = 0;
SystemTimeToFileTime(&ss, &ff);
offset.QuadPart = ff.dwHighDateTime;
offset.QuadPart <<= 32;
offset.QuadPart |= ff.dwLowDateTime;
offset.QuadPart += *timep * 10000000;
f.dwLowDateTime = offset.QuadPart & 0xffffffff;
f.dwHighDateTime = (offset.QuadPart >> 32) & 0xffffffff;
FileTimeToSystemTime(&f, &s);
result->tm_sec = s.wSecond;
result->tm_min = s.wMinute;
result->tm_hour = s.wHour;
result->tm_mday = s.wDay;
result->tm_mon = s.wMonth - 1;
result->tm_year = s.wYear - 1900;
result->tm_wday = s.wDayOfWeek;
result->tm_yday = 0;
result->tm_isdst = 0;
} else {
localtime_s(result, timep);
}
#elif defined(LINUX)
time_t secsMin = 60, secsHour = 3600, secsDay = 3600 * 24;
long tz = timezone;
time_t t = *timep;
t -= tz; /* Adjust for timezone. */
t += 3600 * dst; /* Adjust for daylight time. */
time_t days = t / secsDay; /* Days passed since epoch. */
time_t seconds = t % secsDay; /* Remaining seconds. */
result->tm_isdst = dst;
result->tm_hour = seconds / secsHour;
result->tm_min = (seconds % secsHour) / secsMin;
result->tm_sec = (seconds % secsHour) % secsMin;
/* 1/1/1970 was a Thursday, that is, day 4 from the POV of the tm structure
* where sunday = 0, so to calculate the day of the week we have to add 4
* and take the modulo by 7. */
result->tm_wday = (days + 4) % 7;
/* Calculate the current year. */
result->tm_year = 1970;
while (1) {
/* Leap years have one day more. */
time_t daysOfYear = 365 + isLeapYear(result->tm_year);
if (daysOfYear > days) break;
days -= daysOfYear;
result->tm_year++;
}
result->tm_yday = days; /* Number of day of the current year. */
/* We need to calculate in which month and day of the month we are. To do
* so we need to skip days according to how many days there are in each
* month, and adjust for the leap year that has one more day in February. */
int mdays[12] = {31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};
mdays[1] += isLeapYear(result->tm_year);
result->tm_mon = 0;
while (days >= mdays[result->tm_mon]) {
days -= mdays[result->tm_mon];
result->tm_mon++;
}
result->tm_mday = days + 1; /* Add 1 since our 'days' is zero-based. */
result->tm_year -= 1900; /* Surprisingly tm_year is year-1900. */
#else
localtime_r(timep, result);
#endif
return result;
}
int32_t taosGetTimestampSec() { return (int32_t)time(NULL); }
int32_t taosClockGetTime(int clock_id, struct timespec *pTS) {
#ifdef WINDOWS
@ -473,9 +570,9 @@ int32_t taosClockGetTime(int clock_id, struct timespec *pTS) {
t.QuadPart -= offset.QuadPart;
pTS->tv_sec = t.QuadPart / 10000000;
pTS->tv_nsec = (t.QuadPart % 10000000)*100;
pTS->tv_nsec = (t.QuadPart % 10000000) * 100;
return (0);
#else
return clock_gettime(clock_id, pTS);
#endif
}
}

21
source/util/src/tlog.c
View File

@ -68,6 +68,7 @@ static int8_t tsLogInited = 0;
static SLogObj tsLogObj = {.fileNum = 1};
static int64_t tsAsyncLogLostLines = 0;
static int32_t tsWriteInterval = LOG_DEFAULT_INTERVAL;
static int32_t tsDaylightActive; /* Currently in daylight saving time. */
bool tsLogEmbedded = 0;
bool tsAsyncLog = true;
@ -113,6 +114,16 @@ static void taosCloseLogByFd(TdFilePtr pFile);
static int32_t taosOpenLogFile(char *fn, int32_t maxLines, int32_t maxFileNum);
static int32_t taosCompressFile(char *srcFileName, char *destFileName);
static FORCE_INLINE void taosUpdateDaylight() {
struct tm Tm, *ptm;
struct timeval timeSecs;
taosGetTimeOfDay(&timeSecs);
time_t curTime = timeSecs.tv_sec;
ptm = taosLocalTime(&curTime, &Tm);
tsDaylightActive = ptm->tm_isdst;
}
static FORCE_INLINE int32_t taosGetDaylight() { return tsDaylightActive; }
static int32_t taosStartLog() {
TdThreadAttr threadAttr;
taosThreadAttrInit(&threadAttr);
@ -133,6 +144,7 @@ int32_t taosInitLog(const char *logName, int32_t maxFiles) {
} else {
snprintf(fullName, PATH_MAX, "%s", logName);
}
taosUpdateDaylight();
tsLogObj.logHandle = taosLogBuffNew(LOG_DEFAULT_BUF_SIZE);
if (tsLogObj.logHandle == NULL) return -1;
@ -422,7 +434,7 @@ static inline int32_t taosBuildLogHead(char *buffer, const char *flags) {
taosGetTimeOfDay(&timeSecs);
time_t curTime = timeSecs.tv_sec;
ptm = taosLocalTime(&curTime, &Tm);
ptm = taosLocalTimeNolock(&Tm, &curTime, taosGetDaylight());
return sprintf(buffer, "%02d/%02d %02d:%02d:%02d.%06d %08" PRId64 " %s", ptm->tm_mon + 1, ptm->tm_mday, ptm->tm_hour,
ptm->tm_min, ptm->tm_sec, (int32_t)timeSecs.tv_usec, taosGetSelfPthreadId(), flags);
@ -694,8 +706,10 @@ static void *taosAsyncOutputLog(void *param) {
SLogBuff *pLogBuf = (SLogBuff *)param;
setThreadName("log");
int32_t count = 0;
int32_t updateCron = 0;
while (1) {
count += tsWriteInterval;
updateCron++;
taosMsleep(tsWriteInterval);
if (count > 1000) {
osUpdate();
@ -705,6 +719,11 @@ static void *taosAsyncOutputLog(void *param) {
// Polling the buffer
taosWriteLog(pLogBuf);
if (updateCron >= 3600 * 24 * 40 / 2) {
taosUpdateDaylight();
updateCron = 0;
}
if (pLogBuf->stop) break;
}