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

Merge branch '3.0' into feature/TD-11463-3.0

This commit is contained in:
Cary Xu 2022-03-11 19:39:23 +08:00
parent 0c307568e4 3ef2facd8d
commit 00f6807192
35 changed files with 921 additions and 563 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/os/osDef.h
View File

@ -65,8 +65,6 @@ extern "C" {
#define in_addr_t unsigned long
#define socklen_t int
struct tm *localtime_r(const time_t *timep, struct tm *result);
char * strptime(const char *buf, const char *fmt, struct tm *tm);
char * strsep(char **stringp, const char *delim);
char * getpass(const char *prefix);
char * strndup(const char *s, size_t n);

12
include/os/osTime.h
View File

@ -20,6 +20,15 @@
extern "C" {
#endif
// If the error is in a third-party library, place this header file under the third-party library header file.
#ifndef ALLOW_FORBID_FUNC
#define strptime STRPTIME_FUNC_TAOS_FORBID
#define gettimeofday GETTIMEOFDAY_FUNC_TAOS_FORBID
#define localtime_s LOCALTIMES_FUNC_TAOS_FORBID
#define localtime_r LOCALTIMER_FUNC_TAOS_FORBID
#define time TIME_FUNC_TAOS_FORBID
#endif
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
#ifdef _TD_GO_DLL_
#define MILLISECOND_PER_SECOND (1000LL)
@ -61,6 +70,9 @@ static FORCE_INLINE int64_t taosGetTimestampNs() {
return (int64_t)systemTime.tv_sec * 1000000000L + (int64_t)systemTime.tv_nsec;
}
char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm);
struct tm *taosLocalTime(const time_t *timep, struct tm *result);
#ifdef __cplusplus
}
#endif

9
include/os/osTimer.h
View File

@ -20,6 +20,15 @@
extern "C" {
#endif
// If the error is in a third-party library, place this header file under the third-party library header file.
#ifndef ALLOW_FORBID_FUNC
#define timer_create TIMER_CREATE_FUNC_TAOS_FORBID
#define timer_settime TIMER_SETTIME_FUNC_TAOS_FORBID
#define timer_delete TIMER_DELETE_FUNC_TAOS_FORBID
#define timeSetEvent TIMESETEVENT_SETTIME_FUNC_TAOS_FORBID
#define timeKillEvent TIMEKILLEVENT_SETTIME_FUNC_TAOS_FORBID
#endif
#define MSECONDS_PER_TICK 5
int32_t taosInitTimer(void (*callback)(int32_t), int32_t ms);

2
source/common/src/tname.c
View File

@ -61,7 +61,7 @@ int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, in
}
struct tm tm;
time_t t = (time_t)start;
localtime_r(&t, &tm);
taosLocalTime(&t, &tm);
tm.tm_sec = 0;
tm.tm_min = 0;
tm.tm_hour = 0;

24
source/common/src/ttime.c
View File

@ -69,7 +69,7 @@ static int64_t m_deltaUtc = 0;
void deltaToUtcInitOnce() {
struct tm tm = {0};
(void)strptime("1970-01-01 00:00:00", (const char*)("%Y-%m-%d %H:%M:%S"), &tm);
(void)taosStrpTime("1970-01-01 00:00:00", (const char*)("%Y-%m-%d %H:%M:%S"), &tm);
m_deltaUtc = (int64_t)mktime(&tm);
// printf("====delta:%lld\n\n", seconds);
}
@ -236,9 +236,9 @@ int32_t parseTimeWithTz(const char* timestr, int64_t* time, int32_t timePrec, ch
char* str;
if (delim == 'T') {
str = strptime(timestr, "%Y-%m-%dT%H:%M:%S", &tm);
str = taosStrpTime(timestr, "%Y-%m-%dT%H:%M:%S", &tm);
} else if (delim == 0) {
str = strptime(timestr, "%Y-%m-%d %H:%M:%S", &tm);
str = taosStrpTime(timestr, "%Y-%m-%d %H:%M:%S", &tm);
} else {
str = NULL;
}
@ -303,7 +303,7 @@ int32_t parseLocaltime(char* timestr, int64_t* time, int32_t timePrec) {
*time = 0;
struct tm tm = {0};
char* str = strptime(timestr, "%Y-%m-%d %H:%M:%S", &tm);
char* str = taosStrpTime(timestr, "%Y-%m-%d %H:%M:%S", &tm);
if (str == NULL) {
return -1;
}
@ -338,7 +338,7 @@ int32_t parseLocaltimeDst(char* timestr, int64_t* time, int32_t timePrec) {
struct tm tm = {0};
tm.tm_isdst = -1;
char* str = strptime(timestr, "%Y-%m-%d %H:%M:%S", &tm);
char* str = taosStrpTime(timestr, "%Y-%m-%d %H:%M:%S", &tm);
if (str == NULL) {
return -1;
}
@ -466,7 +466,7 @@ int64_t taosTimeAdd(int64_t t, int64_t duration, char unit, int32_t precision) {
struct tm tm;
time_t tt = (time_t)(t / TSDB_TICK_PER_SECOND(precision));
localtime_r(&tt, &tm);
taosLocalTime(&tt, &tm);
int32_t mon = tm.tm_year * 12 + tm.tm_mon + (int32_t)duration;
tm.tm_year = mon / 12;
tm.tm_mon = mon % 12;
@ -489,11 +489,11 @@ int32_t taosTimeCountInterval(int64_t skey, int64_t ekey, int64_t interval, char
struct tm tm;
time_t t = (time_t)skey;
localtime_r(&t, &tm);
taosLocalTime(&t, &tm);
int32_t smon = tm.tm_year * 12 + tm.tm_mon;
t = (time_t)ekey;
localtime_r(&t, &tm);
taosLocalTime(&t, &tm);
int32_t emon = tm.tm_year * 12 + tm.tm_mon;
if (unit == 'y') {
@ -514,7 +514,7 @@ int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precisio
start /= (int64_t)(TSDB_TICK_PER_SECOND(precision));
struct tm tm;
time_t tt = (time_t)start;
localtime_r(&tt, &tm);
taosLocalTime(&tt, &tm);
tm.tm_sec = 0;
tm.tm_min = 0;
tm.tm_hour = 0;
@ -597,13 +597,13 @@ const char* fmtts(int64_t ts) {
if (ts > -62135625943 && ts < 32503651200) {
time_t t = (time_t)ts;
localtime_r(&t, &tm);
taosLocalTime(&t, &tm);
pos += strftime(buf + pos, sizeof(buf), "s=%Y-%m-%d %H:%M:%S", &tm);
}
if (ts > -62135625943000 && ts < 32503651200000) {
time_t t = (time_t)(ts / 1000);
localtime_r(&t, &tm);
taosLocalTime(&t, &tm);
if (pos > 0) {
buf[pos++] = ' ';
buf[pos++] = '|';
@ -615,7 +615,7 @@ const char* fmtts(int64_t ts) {
{
time_t t = (time_t)(ts / 1000000);
localtime_r(&t, &tm);
taosLocalTime(&t, &tm);
if (pos > 0) {
buf[pos++] = ' ';
buf[pos++] = '|';

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

@ -163,7 +163,7 @@ static void getNextTimeWindow(SInterval* pInterval, int32_t precision, int32_t o
struct tm tm;
time_t t = (time_t)key;
localtime_r(&t, &tm);
taosLocalTime(&t, &tm);
int mon = (int)(tm.tm_year * 12 + tm.tm_mon + interval * factor);
tm.tm_year = mon / 12;

6
source/libs/executor/test/executorTests.cpp
View File

@ -1019,7 +1019,7 @@ TEST(testCase, external_sort_Test) {
return;
#endif
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
SArray* pOrderVal = taosArrayInit(4, sizeof(SOrder));
SOrder o = {0};
@ -1080,7 +1080,7 @@ TEST(testCase, external_sort_Test) {
}
TEST(testCase, sorted_merge_Test) {
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
SArray* pOrderVal = taosArrayInit(4, sizeof(SOrder));
SOrder o = {0};
@ -1152,7 +1152,7 @@ TEST(testCase, sorted_merge_Test) {
}
TEST(testCase, time_interval_Operator_Test) {
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
SArray* pOrderVal = taosArrayInit(4, sizeof(SOrder));
SOrder o = {0};

2
source/libs/executor/test/lhashTests.cpp
View File

@ -25,7 +25,7 @@
#pragma GCC diagnostic ignored "-Wsign-compare"
TEST(testCase, linear_hash_Tests) {
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
_hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);
#if 0

14
source/libs/qworker/test/qworkerTests.cpp
View File

@ -963,7 +963,7 @@ TEST(seqTest, randCase) {
stubSetRpcSendResponse();
stubSetCreateExecTask();
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
ASSERT_EQ(code, 0);
@ -1025,7 +1025,7 @@ TEST(seqTest, multithreadRand) {
stubSetStringToPlan();
stubSetRpcSendResponse();
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
ASSERT_EQ(code, 0);
@ -1076,7 +1076,7 @@ TEST(rcTest, shortExecshortDelay) {
stubSetPutDataBlock();
stubSetGetDataBlock();
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
qwtTestStop = false;
qwtTestQuitThreadNum = 0;
@ -1157,7 +1157,7 @@ TEST(rcTest, longExecshortDelay) {
stubSetPutDataBlock();
stubSetGetDataBlock();
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
qwtTestStop = false;
qwtTestQuitThreadNum = 0;
@ -1240,7 +1240,7 @@ TEST(rcTest, shortExeclongDelay) {
stubSetPutDataBlock();
stubSetGetDataBlock();
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
qwtTestStop = false;
qwtTestQuitThreadNum = 0;
@ -1324,7 +1324,7 @@ TEST(rcTest, dropTest) {
stubSetPutDataBlock();
stubSetGetDataBlock();
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
ASSERT_EQ(code, 0);
@ -1358,7 +1358,7 @@ TEST(rcTest, dropTest) {
int main(int argc, char** argv) {
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

2
source/libs/scalar/test/filter/filterTests.cpp
View File

@ -1286,7 +1286,7 @@ TEST(scalarModelogicTest, diff_columns_or_and_or) {
int main(int argc, char** argv) {
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

2
source/libs/scalar/test/scalar/scalarTests.cpp
View File

@ -1435,7 +1435,7 @@ TEST(columnTest, greater_and_lower) {
int main(int argc, char** argv) {
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

4
source/libs/scheduler/test/schedulerTests.cpp
View File

@ -724,7 +724,7 @@ TEST(queryTest, flowCtrlCase) {
schtInitLogFile();
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
SArray *qnodeList = taosArrayInit(1, sizeof(SEp));
@ -873,7 +873,7 @@ TEST(multiThread, forceFree) {
}
int main(int argc, char** argv) {
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

2
source/libs/sync/src/syncEnv.c
View File

@ -28,7 +28,7 @@ static void doSyncEnvStopTimer(SSyncEnv *pSyncEnv, tmr_h *pTimer);
int32_t syncEnvStart() {
int32_t ret;
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
gSyncEnv = (SSyncEnv *)malloc(sizeof(SSyncEnv));
assert(gSyncEnv != NULL);
ret = doSyncEnvStart(gSyncEnv);

2
source/libs/sync/src/syncIO.c
View File

@ -44,7 +44,7 @@ int32_t syncIOStart(char *host, uint16_t port) {
gSyncIO = syncIOCreate(host, port);
assert(gSyncIO != NULL);
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
int32_t ret = syncIOStartInternal(gSyncIO);
assert(ret == 0);

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

@ -26,7 +26,7 @@ typedef struct SCliConn {
T_REF_DECLARE()
uv_connect_t connReq;
uv_stream_t* stream;
uv_write_t* writeReq;
uv_write_t writeReq;
void* hostThrd;
SConnBuffer readBuf;
void* data;
@ -34,12 +34,12 @@ typedef struct SCliConn {
uint64_t expireTime;
int8_t ctnRdCnt; // continue read count
int hThrdIdx;
bool broken; // link broken or not
int persist; //
// spi configure
char spi;
char secured;
int32_t ref;
char spi;
char secured;
// debug and log info
struct sockaddr_in addr;
struct sockaddr_in locaddr;
@ -54,11 +54,10 @@ typedef struct SCliMsg {
} SCliMsg;
typedef struct SCliThrdObj {
pthread_t thread;
uv_loop_t* loop;
// uv_async_t* cliAsync; //
pthread_t thread;
uv_loop_t* loop;
SAsyncPool* asyncPool;
uv_timer_t* timer;
uv_timer_t timer;
void* pool; // conn pool
// msg queue
@ -83,7 +82,7 @@ typedef struct SConnList {
// conn pool
// add expire timeout and capacity limit
static void* creatConnPool(int size);
static void* createConnPool(int size);
static void* destroyConnPool(void* pool);
static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port);
static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn);
@ -99,8 +98,10 @@ static void clientWriteCb(uv_write_t* req, int status);
// callback after conn to server
static void clientConnCb(uv_connect_t* req, int status);
static void clientAsyncCb(uv_async_t* handle);
static void clientDestroy(uv_handle_t* handle);
static void clientConnDestroy(SCliConn* pConn, bool clear /*clear tcp handle or not*/);
static SCliConn* clientConnCreate(SCliThrdObj* thrd);
static void clientConnDestroy(SCliConn* pConn, bool clear /*clear tcp handle or not*/);
static void clientDestroy(uv_handle_t* handle);
// process data read from server, add decompress etc later
static void clientHandleResp(SCliConn* conn);
@ -176,14 +177,14 @@ static void clientHandleResp(SCliConn* conn) {
conn->data = NULL;
// start thread's timer of conn pool if not active
if (!uv_is_active((uv_handle_t*)pThrd->timer) && pTransInst->idleTime > 0) {
// uv_timer_start((uv_timer_t*)pThrd->timer, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
if (!uv_is_active((uv_handle_t*)&pThrd->timer) && pTransInst->idleTime > 0) {
// uv_timer_start((uv_timer_t*)&pThrd->timer, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
}
}
static void clientHandleExcept(SCliConn* pConn) {
if (pConn->data == NULL) {
// handle conn except in conn pool
clientConnDestroy(pConn, true);
transUnrefCliHandle(pConn);
return;
}
SCliThrdObj* pThrd = pConn->hostThrd;
@ -209,7 +210,7 @@ static void clientHandleExcept(SCliConn* pConn) {
pConn->data = NULL;
tTrace("%s client conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn);
clientConnDestroy(pConn, true);
transUnrefCliHandle(pConn);
}
static void clientTimeoutCb(uv_timer_t* handle) {
@ -225,9 +226,7 @@ static void clientTimeoutCb(uv_timer_t* handle) {
SCliConn* c = QUEUE_DATA(h, SCliConn, conn);
if (c->expireTime < currentTime) {
QUEUE_REMOVE(h);
// uv_stream_t stm = *(c->stream);
// uv_close((uv_handle_t*)&stm, clientDestroy);
clientConnDestroy(c, true);
transUnrefCliHandle(c);
} else {
break;
}
@ -238,7 +237,7 @@ static void clientTimeoutCb(uv_timer_t* handle) {
pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
uv_timer_start(handle, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
}
static void* creatConnPool(int size) {
static void* createConnPool(int size) {
// thread local, no lock
return taosHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
}
@ -253,7 +252,7 @@ static void* destroyConnPool(void* pool) {
}
connList = taosHashIterate((SHashObj*)pool, connList);
}
taosHashClear(pool);
taosHashCleanup(pool);
}
static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
@ -328,26 +327,38 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
}
if (nread < 0) {
tError("%s client conn %p read error: %s", CONN_GET_INST_LABEL(conn), conn, uv_err_name(nread));
conn->broken = true;
clientHandleExcept(conn);
}
}
static SCliConn* clientConnCreate(SCliThrdObj* pThrd) {
SCliConn* conn = calloc(1, sizeof(SCliConn));
// read/write stream handle
conn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream));
conn->stream->data = conn;
conn->writeReq.data = conn;
conn->connReq.data = conn;
QUEUE_INIT(&conn->conn);
conn->hostThrd = pThrd;
conn->broken = false;
transRefCliHandle(conn);
return conn;
}
static void clientConnDestroy(SCliConn* conn, bool clear) {
//
conn->ref--;
if (conn->ref == 0) {
tTrace("%s client conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn);
QUEUE_REMOVE(&conn->conn);
if (clear) {
uv_close((uv_handle_t*)conn->stream, clientDestroy);
}
tTrace("%s client conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn);
QUEUE_REMOVE(&conn->conn);
if (clear) {
uv_close((uv_handle_t*)conn->stream, clientDestroy);
}
}
static void clientDestroy(uv_handle_t* handle) {
SCliConn* conn = handle->data;
free(conn->stream);
free(conn->writeReq);
tTrace("%s client conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
free(conn);
}
@ -359,7 +370,6 @@ static void clientWriteCb(uv_write_t* req, int status) {
tTrace("%s client conn %p data already was written out", CONN_GET_INST_LABEL(pConn), pConn);
SCliMsg* pMsg = pConn->data;
if (pMsg == NULL) {
// handle
return;
}
destroyUserdata(&pMsg->msg);
@ -410,7 +420,7 @@ static void clientWrite(SCliConn* pConn) {
TMSG_INFO(pHead->msgType), inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
inet_ntoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port));
uv_write(pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, clientWriteCb);
uv_write(&pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, clientWriteCb);
}
static void clientConnCb(uv_connect_t* req, int status) {
// impl later
@ -436,10 +446,10 @@ static void clientHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) {
tDebug("client work thread %p start to quit", pThrd);
destroyCmsg(pMsg);
destroyConnPool(pThrd->pool);
// transDestroyAsyncPool(pThr) uv_close((uv_handle_t*)pThrd->cliAsync, NULL);
uv_timer_stop(pThrd->timer);
uv_timer_stop(&pThrd->timer);
pThrd->quit = true;
// uv__async_stop(pThrd->cliAsync);
uv_stop(pThrd->loop);
}
static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
@ -463,7 +473,7 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
if (conn != NULL) {
conn->data = pMsg;
conn->writeReq->data = conn;
conn->writeReq.data = conn;
transDestroyBuffer(&conn->readBuf);
if (pThrd->quit) {
@ -472,21 +482,8 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
}
clientWrite(conn);
} else {
conn = calloc(1, sizeof(SCliConn));
conn->ref++;
// read/write stream handle
conn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream));
conn->stream->data = conn;
conn->writeReq = malloc(sizeof(uv_write_t));
conn->writeReq->data = conn;
QUEUE_INIT(&conn->conn);
conn->connReq.data = conn;
conn = clientConnCreate(pThrd);
conn->data = pMsg;
conn->hostThrd = pThrd;
int ret = transSetConnOption((uv_tcp_t*)conn->stream);
if (ret) {
@ -585,11 +582,10 @@ static SCliThrdObj* createThrdObj() {
pThrd->asyncPool = transCreateAsyncPool(pThrd->loop, 5, pThrd, clientAsyncCb);
pThrd->timer = malloc(sizeof(uv_timer_t));
uv_timer_init(pThrd->loop, pThrd->timer);
pThrd->timer->data = pThrd;
uv_timer_init(pThrd->loop, &pThrd->timer);
pThrd->timer.data = pThrd;
pThrd->pool = creatConnPool(4);
pThrd->pool = createConnPool(4);
pThrd->quit = false;
return pThrd;
@ -602,8 +598,8 @@ static void destroyThrdObj(SCliThrdObj* pThrd) {
pthread_join(pThrd->thread, NULL);
pthread_mutex_destroy(&pThrd->msgMtx);
transDestroyAsyncPool(pThrd->asyncPool);
// free(pThrd->cliAsync);
free(pThrd->timer);
uv_timer_stop(&pThrd->timer);
free(pThrd->loop);
free(pThrd);
}
@ -649,6 +645,7 @@ void transUnrefCliHandle(void* handle) {
}
int ref = T_REF_DEC((SCliConn*)handle);
if (ref == 0) {
clientConnDestroy((SCliConn*)handle, true);
}
// unref cli handle

4
source/libs/transport/src/transComm.c
View File

@ -226,9 +226,13 @@ int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf) {
uvBuf->base = p->buf;
uvBuf->len = CAPACITY;
} else if (p->total == -1 && p->len < CAPACITY) {
uvBuf->base = p->buf + p->len;
uvBuf->len = CAPACITY - p->len;
} else {
p->cap = p->total;
p->buf = realloc(p->buf, p->cap);
uvBuf->base = p->buf + p->len;
uvBuf->len = p->cap - p->len;
}

43
source/libs/transport/src/transSrv.c
View File

@ -19,11 +19,10 @@
typedef struct SSrvConn {
T_REF_DECLARE()
uv_tcp_t* pTcp;
uv_write_t* pWriter;
uv_timer_t* pTimer;
uv_tcp_t* pTcp;
uv_write_t pWriter;
uv_timer_t pTimer;
// uv_async_t* pWorkerAsync;
queue queue;
int ref;
int persist; // persist connection or not
@ -65,7 +64,7 @@ typedef struct SWorkThrdObj {
queue conn;
pthread_mutex_t msgMtx;
void* pTransInst;
bool stop;
bool quit;
} SWorkThrdObj;
typedef struct SServerObj {
@ -236,7 +235,7 @@ static void uvHandleReq(SSrvConn* pConn) {
inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), inet_ntoa(pConn->locaddr.sin_addr),
ntohs(pConn->locaddr.sin_port), rpcMsg.contLen);
(*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL);
// uv_timer_start(pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0);
// uv_timer_start(&pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0);
// auth
// validate msg type
}
@ -312,6 +311,7 @@ static void uvOnPipeWriteCb(uv_write_t* req, int status) {
} else {
tError("fail to dispatch conn to work thread");
}
free(req);
}
static void uvPrepareSendData(SSrvMsg* smsg, uv_buf_t* wb) {
@ -349,8 +349,8 @@ static void uvStartSendRespInternal(SSrvMsg* smsg) {
uvPrepareSendData(smsg, &wb);
SSrvConn* pConn = smsg->pConn;
uv_timer_stop(pConn->pTimer);
uv_write(pConn->pWriter, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnWriteCb);
uv_timer_stop(&pConn->pTimer);
uv_write(&pConn->pWriter, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnWriteCb);
}
static void uvStartSendResp(SSrvMsg* smsg) {
// impl
@ -417,8 +417,8 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
uv_stop(pThrd->loop);
} else {
destroyAllConn(pThrd);
uv_loop_close(pThrd->loop);
pThrd->stop = true;
// uv_loop_close(pThrd->loop);
pThrd->quit = true;
}
} else {
uvStartSendResp(msg);
@ -493,9 +493,8 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
pConn->pTransInst = pThrd->pTransInst;
/* init conn timer*/
pConn->pTimer = malloc(sizeof(uv_timer_t));
uv_timer_init(pThrd->loop, pConn->pTimer);
pConn->pTimer->data = pConn;
uv_timer_init(pThrd->loop, &pConn->pTimer);
pConn->pTimer.data = pConn;
pConn->hostThrd = pThrd;
@ -504,8 +503,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
uv_tcp_init(pThrd->loop, pConn->pTcp);
pConn->pTcp->data = pConn;
pConn->pWriter = calloc(1, sizeof(uv_write_t));
pConn->pWriter->data = pConn;
pConn->pWriter.data = pConn;
transSetConnOption((uv_tcp_t*)pConn->pTcp);
@ -633,17 +631,20 @@ static void destroyConn(SSrvConn* conn, bool clear) {
}
}
static void uvDestroyConn(uv_handle_t* handle) {
SSrvConn* conn = handle->data;
SSrvConn* conn = handle->data;
if (conn == NULL) {
return;
}
SWorkThrdObj* thrd = conn->hostThrd;
tDebug("server conn %p destroy", conn);
uv_timer_stop(conn->pTimer);
uv_timer_stop(&conn->pTimer);
QUEUE_REMOVE(&conn->queue);
free(conn->pTcp);
free(conn->pWriter);
free(conn);
// free(conn);
if (thrd->stop && QUEUE_IS_EMPTY(&thrd->conn)) {
if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) {
uv_loop_close(thrd->loop);
uv_stop(thrd->loop);
}
}
@ -680,7 +681,7 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
for (int i = 0; i < srv->numOfThreads; i++) {
SWorkThrdObj* thrd = (SWorkThrdObj*)calloc(1, sizeof(SWorkThrdObj));
thrd->stop = false;
thrd->quit = false;
srv->pThreadObj[i] = thrd;
srv->pipe[i] = (uv_pipe_t*)calloc(2, sizeof(uv_pipe_t));

4
source/libs/transport/test/pushClient.c
View File

@ -203,7 +203,7 @@ int main(int argc, char *argv[]) {
tInfo("client is initialized");
tInfo("threads:%d msgSize:%d requests:%d", appThreads, msgSize, numOfReqs);
gettimeofday(&systemTime, NULL);
taosGetTimeOfDay(&systemTime);
startTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
SInfo *pInfo = (SInfo *)calloc(1, sizeof(SInfo) * appThreads);
@ -226,7 +226,7 @@ int main(int argc, char *argv[]) {
taosUsleep(1);
} while (tcount < appThreads);
gettimeofday(&systemTime, NULL);
taosGetTimeOfDay(&systemTime);
endTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
float usedTime = (endTime - startTime) / 1000.0f; // mseconds

4
source/libs/transport/test/rclient.c
View File

@ -181,7 +181,7 @@ int main(int argc, char *argv[]) {
tInfo("client is initialized");
tInfo("threads:%d msgSize:%d requests:%d", appThreads, msgSize, numOfReqs);
gettimeofday(&systemTime, NULL);
taosGetTimeOfDay(&systemTime);
startTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
SInfo *pInfo = (SInfo *)calloc(1, sizeof(SInfo) * appThreads);
@ -204,7 +204,7 @@ int main(int argc, char *argv[]) {
taosUsleep(1);
} while (tcount < appThreads);
gettimeofday(&systemTime, NULL);
taosGetTimeOfDay(&systemTime);
endTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
float usedTime = (endTime - startTime) / 1000.0f; // mseconds

4
source/libs/transport/test/rsclient.c
View File

@ -158,7 +158,7 @@ int main(int argc, char *argv[]) {
tInfo("client is initialized");
gettimeofday(&systemTime, NULL);
taosGetTimeOfDay(&systemTime);
startTime = systemTime.tv_sec*1000000 + systemTime.tv_usec;
SInfo *pInfo = (SInfo *)calloc(1, sizeof(SInfo)*appThreads);
@ -181,7 +181,7 @@ int main(int argc, char *argv[]) {
taosUsleep(1);
} while ( tcount < appThreads);
gettimeofday(&systemTime, NULL);
taosGetTimeOfDay(&systemTime);
endTime = systemTime.tv_sec*1000000 + systemTime.tv_usec;
float usedTime = (endTime - startTime)/1000.0; // mseconds

4
source/libs/transport/test/syncClient.c
View File

@ -181,7 +181,7 @@ int main(int argc, char *argv[]) {
tInfo("client is initialized");
tInfo("threads:%d msgSize:%d requests:%d", appThreads, msgSize, numOfReqs);
gettimeofday(&systemTime, NULL);
taosGetTimeOfDay(&systemTime);
startTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
SInfo *pInfo = (SInfo *)calloc(1, sizeof(SInfo) * appThreads);
@ -204,7 +204,7 @@ int main(int argc, char *argv[]) {
taosUsleep(1);
} while (tcount < appThreads);
gettimeofday(&systemTime, NULL);
taosGetTimeOfDay(&systemTime);
endTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
float usedTime = (endTime - startTime) / 1000.0f; // mseconds

4
source/os/src/osRand.c
View File

@ -33,11 +33,11 @@ uint32_t taosSafeRand(void) {
pFile = taosOpenFile("/dev/urandom", TD_FILE_READ);
if (pFile == NULL) {
seed = (int)time(0);
seed = (int)taosGetTimestampSec();
} else {
int len = taosReadFile(pFile, &seed, sizeof(seed));
if (len < 0) {
seed = (int)time(0);
seed = (int)taosGetTimestampSec();
}
taosCloseFile(&pFile);
}

692
source/os/src/osStrptime.c
View File

@ -38,366 +38,368 @@
//
//#include "lukemftp.h"
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
// #if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
#include <time.h>
#include <stdlib.h>
#include <string.h>
#include <winsock2.h>
//#define TM_YEAR_BASE 1970 //origin
#define TM_YEAR_BASE 1900 //slguan
/*
* We do not implement alternate representations. However, we always
* check whether a given modifier is allowed for a certain conversion.
*/
#define ALT_E 0x01
#define ALT_O 0x02
#define LEGAL_ALT(x) { if (alt_format & ~(x)) return (0); }
// #include <time.h>
// #include <stdlib.h>
// #include <string.h>
// #include <winsock2.h>
// //#define TM_YEAR_BASE 1970 //origin
// #define TM_YEAR_BASE 1900 //slguan
// /*
// * We do not implement alternate representations. However, we always
// * check whether a given modifier is allowed for a certain conversion.
// */
// #define ALT_E 0x01
// #define ALT_O 0x02
// #define LEGAL_ALT(x) { if (alt_format & ~(x)) return (0); }
static int conv_num(const char **, int *, int, int);
// static int conv_num(const char **buf, int *dest, int llim, int ulim)
// {
// int result = 0;
static const char *day[7] = {
"Sunday", "Monday", "Tuesday", "Wednesday", "Thursday",
"Friday", "Saturday"
};
static const char *abday[7] = {
"Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"
};
static const char *mon[12] = {
"January", "February", "March", "April", "May", "June", "July",
"August", "September", "October", "November", "December"
};
static const char *abmon[12] = {
"Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec"
};
static const char *am_pm[2] = {
"AM", "PM"
};
// /* The limit also determines the number of valid digits. */
// int rulim = ulim;
// if (**buf < '0' || **buf > '9')
// return (0);
// do {
// result *= 10;
// result += *(*buf)++ - '0';
// rulim /= 10;
// } while ((result * 10 <= ulim) && rulim && **buf >= '0' && **buf <= '9');
// if (result < llim || result > ulim)
// return (0);
// *dest = result;
// return (1);
// }
// static const char *day[7] = {
// "Sunday", "Monday", "Tuesday", "Wednesday", "Thursday",
// "Friday", "Saturday"
// };
// static const char *abday[7] = {
// "Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"
// };
// static const char *mon[12] = {
// "January", "February", "March", "April", "May", "June", "July",
// "August", "September", "October", "November", "December"
// };
// static const char *abmon[12] = {
// "Jan", "Feb", "Mar", "Apr", "May", "Jun",
// "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"
// };
// static const char *am_pm[2] = {
// "AM", "PM"
// };
// #endif
// char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
// #if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
// char c;
// const char *bp;
// size_t len = 0;
// int alt_format, i, split_year = 0;
// bp = buf;
// while ((c = *fmt) != '\0') {
// /* Clear `alternate' modifier prior to new conversion. */
// alt_format = 0;
// /* Eat up white-space. */
// if (isspace(c)) {
// while (isspace(*bp))
// bp++;
// fmt++;
// continue;
// }
// if ((c = *fmt++) != '%')
// goto literal;
char *
strptime(const char *buf, const char *fmt, struct tm *tm)
{
char c;
const char *bp;
size_t len = 0;
int alt_format, i, split_year = 0;
// again: switch (c = *fmt++) {
// case '%': /* "%%" is converted to "%". */
// literal :
// if (c != *bp++)
// return (0);
// break;
bp = buf;
// /*
// * "Alternative" modifiers. Just set the appropriate flag
// * and start over again.
// */
// case 'E': /* "%E?" alternative conversion modifier. */
// LEGAL_ALT(0);
// alt_format |= ALT_E;
// goto again;
while ((c = *fmt) != '\0') {
/* Clear `alternate' modifier prior to new conversion. */
alt_format = 0;
// case 'O': /* "%O?" alternative conversion modifier. */
// LEGAL_ALT(0);
// alt_format |= ALT_O;
// goto again;
/* Eat up white-space. */
if (isspace(c)) {
while (isspace(*bp))
bp++;
// /*
// * "Complex" conversion rules, implemented through recursion.
// */
// case 'c': /* Date and time, using the locale's format. */
// LEGAL_ALT(ALT_E);
// if (!(bp = taosStrpTime(bp, "%x %X", tm)))
// return (0);
// break;
fmt++;
continue;
}
// case 'D': /* The date as "%m/%d/%y". */
// LEGAL_ALT(0);
// if (!(bp = taosStrpTime(bp, "%m/%d/%y", tm)))
// return (0);
// break;
if ((c = *fmt++) != '%')
goto literal;
// case 'R': /* The time as "%H:%M". */
// LEGAL_ALT(0);
// if (!(bp = taosStrpTime(bp, "%H:%M", tm)))
// return (0);
// break;
// case 'r': /* The time in 12-hour clock representation. */
// LEGAL_ALT(0);
// if (!(bp = taosStrpTime(bp, "%I:%M:%S %p", tm)))
// return (0);
// break;
// case 'T': /* The time as "%H:%M:%S". */
// LEGAL_ALT(0);
// if (!(bp = taosStrpTime(bp, "%H:%M:%S", tm)))
// return (0);
// break;
// case 'X': /* The time, using the locale's format. */
// LEGAL_ALT(ALT_E);
// if (!(bp = taosStrpTime(bp, "%H:%M:%S", tm)))
// return (0);
// break;
// case 'x': /* The date, using the locale's format. */
// LEGAL_ALT(ALT_E);
// if (!(bp = taosStrpTime(bp, "%m/%d/%y", tm)))
// return (0);
// break;
// /*
// * "Elementary" conversion rules.
// */
// case 'A': /* The day of week, using the locale's form. */
// case 'a':
// LEGAL_ALT(0);
// for (i = 0; i < 7; i++) {
// /* Full name. */
// len = strlen(day[i]);
// if (strncmp(day[i], bp, len) == 0)
// break;
// /* Abbreviated name. */
// len = strlen(abday[i]);
// if (strncmp(abday[i], bp, len) == 0)
// break;
// }
// /* Nothing matched. */
// if (i == 7)
// return (0);
// tm->tm_wday = i;
// bp += len;
// break;
// case 'B': /* The month, using the locale's form. */
// case 'b':
// case 'h':
// LEGAL_ALT(0);
// for (i = 0; i < 12; i++) {
// /* Full name. */
// len = strlen(mon[i]);
// if (strncmp(mon[i], bp, len) == 0)
// break;
// /* Abbreviated name. */
// len = strlen(abmon[i]);
// if (strncmp(abmon[i], bp, len) == 0)
// break;
// }
// /* Nothing matched. */
// if (i == 12)
// return (0);
// tm->tm_mon = i;
// bp += len;
// break;
// case 'C': /* The century number. */
// LEGAL_ALT(ALT_E);
// if (!(conv_num(&bp, &i, 0, 99)))
// return (0);
// if (split_year) {
// tm->tm_year = (tm->tm_year % 100) + (i * 100);
// }
// else {
// tm->tm_year = i * 100;
// split_year = 1;
// }
// break;
// case 'd': /* The day of month. */
// case 'e':
// LEGAL_ALT(ALT_O);
// if (!(conv_num(&bp, &tm->tm_mday, 1, 31)))
// return (0);
// break;
// case 'k': /* The hour (24-hour clock representation). */
// LEGAL_ALT(0);
// /* FALLTHROUGH */
// case 'H':
// LEGAL_ALT(ALT_O);
// if (!(conv_num(&bp, &tm->tm_hour, 0, 23)))
// return (0);
// break;
// case 'l': /* The hour (12-hour clock representation). */
// LEGAL_ALT(0);
// /* FALLTHROUGH */
// case 'I':
// LEGAL_ALT(ALT_O);
// if (!(conv_num(&bp, &tm->tm_hour, 1, 12)))
// return (0);
// if (tm->tm_hour == 12)
// tm->tm_hour = 0;
// break;
// case 'j': /* The day of year. */
// LEGAL_ALT(0);
// if (!(conv_num(&bp, &i, 1, 366)))
// return (0);
// tm->tm_yday = i - 1;
// break;
// case 'M': /* The minute. */
// LEGAL_ALT(ALT_O);
// if (!(conv_num(&bp, &tm->tm_min, 0, 59)))
// return (0);
// break;
// case 'm': /* The month. */
// LEGAL_ALT(ALT_O);
// if (!(conv_num(&bp, &i, 1, 12)))
// return (0);
// tm->tm_mon = i - 1;
// break;
// case 'p': /* The locale's equivalent of AM/PM. */
// LEGAL_ALT(0);
// /* AM? */
// if (strcmp(am_pm[0], bp) == 0) {
// if (tm->tm_hour > 11)
// return (0);
// bp += strlen(am_pm[0]);
// break;
// }
// /* PM? */
// else if (strcmp(am_pm[1], bp) == 0) {
// if (tm->tm_hour > 11)
// return (0);
// tm->tm_hour += 12;
// bp += strlen(am_pm[1]);
// break;
// }
// /* Nothing matched. */
// return (0);
// case 'S': /* The seconds. */
// LEGAL_ALT(ALT_O);
// if (!(conv_num(&bp, &tm->tm_sec, 0, 61)))
// return (0);
// break;
// case 'U': /* The week of year, beginning on sunday. */
// case 'W': /* The week of year, beginning on monday. */
// LEGAL_ALT(ALT_O);
// /*
// * XXX This is bogus, as we can not assume any valid
// * information present in the tm structure at this
// * point to calculate a real value, so just check the
// * range for now.
// */
// if (!(conv_num(&bp, &i, 0, 53)))
// return (0);
// break;
// case 'w': /* The day of week, beginning on sunday. */
// LEGAL_ALT(ALT_O);
// if (!(conv_num(&bp, &tm->tm_wday, 0, 6)))
// return (0);
// break;
// case 'Y': /* The year. */
// LEGAL_ALT(ALT_E);
// if (!(conv_num(&bp, &i, 0, 9999)))
// return (0);
// tm->tm_year = i - TM_YEAR_BASE;
// break;
// case 'y': /* The year within 100 years of the epoch. */
// LEGAL_ALT(ALT_E | ALT_O);
// if (!(conv_num(&bp, &i, 0, 99)))
// return (0);
// if (split_year) {
// tm->tm_year = ((tm->tm_year / 100) * 100) + i;
// break;
// }
// split_year = 1;
// if (i <= 68)
// tm->tm_year = i + 2000 - TM_YEAR_BASE;
// else
// tm->tm_year = i + 1900 - TM_YEAR_BASE;
// break;
// /*
// * Miscellaneous conversions.
// */
// case 'n': /* Any kind of white-space. */
// case 't':
// LEGAL_ALT(0);
// while (isspace(*bp))
// bp++;
// break;
again: switch (c = *fmt++) {
case '%': /* "%%" is converted to "%". */
literal :
if (c != *bp++)
return (0);
break;
/*
* "Alternative" modifiers. Just set the appropriate flag
* and start over again.
*/
case 'E': /* "%E?" alternative conversion modifier. */
LEGAL_ALT(0);
alt_format |= ALT_E;
goto again;
case 'O': /* "%O?" alternative conversion modifier. */
LEGAL_ALT(0);
alt_format |= ALT_O;
goto again;
/*
* "Complex" conversion rules, implemented through recursion.
*/
case 'c': /* Date and time, using the locale's format. */
LEGAL_ALT(ALT_E);
if (!(bp = strptime(bp, "%x %X", tm)))
return (0);
break;
case 'D': /* The date as "%m/%d/%y". */
LEGAL_ALT(0);
if (!(bp = strptime(bp, "%m/%d/%y", tm)))
return (0);
break;
case 'R': /* The time as "%H:%M". */
LEGAL_ALT(0);
if (!(bp = strptime(bp, "%H:%M", tm)))
return (0);
break;
case 'r': /* The time in 12-hour clock representation. */
LEGAL_ALT(0);
if (!(bp = strptime(bp, "%I:%M:%S %p", tm)))
return (0);
break;
case 'T': /* The time as "%H:%M:%S". */
LEGAL_ALT(0);
if (!(bp = strptime(bp, "%H:%M:%S", tm)))
return (0);
break;
case 'X': /* The time, using the locale's format. */
LEGAL_ALT(ALT_E);
if (!(bp = strptime(bp, "%H:%M:%S", tm)))
return (0);
break;
case 'x': /* The date, using the locale's format. */
LEGAL_ALT(ALT_E);
if (!(bp = strptime(bp, "%m/%d/%y", tm)))
return (0);
break;
/*
* "Elementary" conversion rules.
*/
case 'A': /* The day of week, using the locale's form. */
case 'a':
LEGAL_ALT(0);
for (i = 0; i < 7; i++) {
/* Full name. */
len = strlen(day[i]);
if (strncmp(day[i], bp, len) == 0)
break;
/* Abbreviated name. */
len = strlen(abday[i]);
if (strncmp(abday[i], bp, len) == 0)
break;
}
/* Nothing matched. */
if (i == 7)
return (0);
tm->tm_wday = i;
bp += len;
break;
case 'B': /* The month, using the locale's form. */
case 'b':
case 'h':
LEGAL_ALT(0);
for (i = 0; i < 12; i++) {
/* Full name. */
len = strlen(mon[i]);
if (strncmp(mon[i], bp, len) == 0)
break;
/* Abbreviated name. */
len = strlen(abmon[i]);
if (strncmp(abmon[i], bp, len) == 0)
break;
}
/* Nothing matched. */
if (i == 12)
return (0);
tm->tm_mon = i;
bp += len;
break;
case 'C': /* The century number. */
LEGAL_ALT(ALT_E);
if (!(conv_num(&bp, &i, 0, 99)))
return (0);
if (split_year) {
tm->tm_year = (tm->tm_year % 100) + (i * 100);
}
else {
tm->tm_year = i * 100;
split_year = 1;
}
break;
case 'd': /* The day of month. */
case 'e':
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_mday, 1, 31)))
return (0);
break;
case 'k': /* The hour (24-hour clock representation). */
LEGAL_ALT(0);
/* FALLTHROUGH */
case 'H':
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_hour, 0, 23)))
return (0);
break;
case 'l': /* The hour (12-hour clock representation). */
LEGAL_ALT(0);
/* FALLTHROUGH */
case 'I':
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_hour, 1, 12)))
return (0);
if (tm->tm_hour == 12)
tm->tm_hour = 0;
break;
case 'j': /* The day of year. */
LEGAL_ALT(0);
if (!(conv_num(&bp, &i, 1, 366)))
return (0);
tm->tm_yday = i - 1;
break;
case 'M': /* The minute. */
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_min, 0, 59)))
return (0);
break;
case 'm': /* The month. */
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &i, 1, 12)))
return (0);
tm->tm_mon = i - 1;
break;
case 'p': /* The locale's equivalent of AM/PM. */
LEGAL_ALT(0);
/* AM? */
if (strcmp(am_pm[0], bp) == 0) {
if (tm->tm_hour > 11)
return (0);
bp += strlen(am_pm[0]);
break;
}
/* PM? */
else if (strcmp(am_pm[1], bp) == 0) {
if (tm->tm_hour > 11)
return (0);
tm->tm_hour += 12;
bp += strlen(am_pm[1]);
break;
}
/* Nothing matched. */
return (0);
case 'S': /* The seconds. */
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_sec, 0, 61)))
return (0);
break;
case 'U': /* The week of year, beginning on sunday. */
case 'W': /* The week of year, beginning on monday. */
LEGAL_ALT(ALT_O);
/*
* XXX This is bogus, as we can not assume any valid
* information present in the tm structure at this
* point to calculate a real value, so just check the
* range for now.
*/
if (!(conv_num(&bp, &i, 0, 53)))
return (0);
break;
case 'w': /* The day of week, beginning on sunday. */
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_wday, 0, 6)))
return (0);
break;
case 'Y': /* The year. */
LEGAL_ALT(ALT_E);
if (!(conv_num(&bp, &i, 0, 9999)))
return (0);
tm->tm_year = i - TM_YEAR_BASE;
break;
case 'y': /* The year within 100 years of the epoch. */
LEGAL_ALT(ALT_E | ALT_O);
if (!(conv_num(&bp, &i, 0, 99)))
return (0);
if (split_year) {
tm->tm_year = ((tm->tm_year / 100) * 100) + i;
break;
}
split_year = 1;
if (i <= 68)
tm->tm_year = i + 2000 - TM_YEAR_BASE;
else
tm->tm_year = i + 1900 - TM_YEAR_BASE;
break;
/*
* Miscellaneous conversions.
*/
case 'n': /* Any kind of white-space. */
case 't':
LEGAL_ALT(0);
while (isspace(*bp))
bp++;
break;
// default: /* Unknown/unsupported conversion. */
// return (0);
// }
default: /* Unknown/unsupported conversion. */
return (0);
}
// }
// /* LINTED functional specification */
// return ((char *)bp);
// #elif defined(_TD_DARWIN_64)
// return strptime(buf, fmt, tm);
// #else
// return strptime(buf, fmt, tm);
// #endif
// }
}
/* LINTED functional specification */
return ((char *)bp);
}
static int
conv_num(const char **buf, int *dest, int llim, int ulim)
{
int result = 0;
/* The limit also determines the number of valid digits. */
int rulim = ulim;
if (**buf < '0' || **buf > '9')
return (0);
do {
result *= 10;
result += *(*buf)++ - '0';
rulim /= 10;
} while ((result * 10 <= ulim) && rulim && **buf >= '0' && **buf <= '9');
if (result < llim || result > ulim)
return (0);
*dest = result;
return (1);
}
#endif

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

@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define ALLOW_FORBID_FUNC
#define _BSD_SOURCE
#ifdef DARWIN
@ -26,16 +27,373 @@
#include "os.h"
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
/*
* windows implementation
*/
#include <time.h>
#include <stdlib.h>
#include <string.h>
#include <winsock2.h>
//#define TM_YEAR_BASE 1970 //origin
#define TM_YEAR_BASE 1900 //slguan
/*
* We do not implement alternate representations. However, we always
* check whether a given modifier is allowed for a certain conversion.
*/
#define ALT_E 0x01
#define ALT_O 0x02
#define LEGAL_ALT(x) { if (alt_format & ~(x)) return (0); }
int taosGetTimeOfDay(struct timeval *tv, struct timezone *tz) {
static int conv_num(const char **buf, int *dest, int llim, int ulim)
{
int result = 0;
/* The limit also determines the number of valid digits. */
int rulim = ulim;
if (**buf < '0' || **buf > '9')
return (0);
do {
result *= 10;
result += *(*buf)++ - '0';
rulim /= 10;
} while ((result * 10 <= ulim) && rulim && **buf >= '0' && **buf <= '9');
if (result < llim || result > ulim)
return (0);
*dest = result;
return (1);
}
static const char *day[7] = {
"Sunday", "Monday", "Tuesday", "Wednesday", "Thursday",
"Friday", "Saturday"
};
static const char *abday[7] = {
"Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"
};
static const char *mon[12] = {
"January", "February", "March", "April", "May", "June", "July",
"August", "September", "October", "November", "December"
};
static const char *abmon[12] = {
"Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec"
};
static const char *am_pm[2] = {
"AM", "PM"
};
#else
#include <sys/time.h>
#endif
char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm) {
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
char c;
const char *bp;
size_t len = 0;
int alt_format, i, split_year = 0;
bp = buf;
while ((c = *fmt) != '\0') {
/* Clear `alternate' modifier prior to new conversion. */
alt_format = 0;
/* Eat up white-space. */
if (isspace(c)) {
while (isspace(*bp))
bp++;
fmt++;
continue;
}
if ((c = *fmt++) != '%')
goto literal;
again: switch (c = *fmt++) {
case '%': /* "%%" is converted to "%". */
literal :
if (c != *bp++)
return (0);
break;
/*
* "Alternative" modifiers. Just set the appropriate flag
* and start over again.
*/
case 'E': /* "%E?" alternative conversion modifier. */
LEGAL_ALT(0);
alt_format |= ALT_E;
goto again;
case 'O': /* "%O?" alternative conversion modifier. */
LEGAL_ALT(0);
alt_format |= ALT_O;
goto again;
/*
* "Complex" conversion rules, implemented through recursion.
*/
case 'c': /* Date and time, using the locale's format. */
LEGAL_ALT(ALT_E);
if (!(bp = taosStrpTime(bp, "%x %X", tm)))
return (0);
break;
case 'D': /* The date as "%m/%d/%y". */
LEGAL_ALT(0);
if (!(bp = taosStrpTime(bp, "%m/%d/%y", tm)))
return (0);
break;
case 'R': /* The time as "%H:%M". */
LEGAL_ALT(0);
if (!(bp = taosStrpTime(bp, "%H:%M", tm)))
return (0);
break;
case 'r': /* The time in 12-hour clock representation. */
LEGAL_ALT(0);
if (!(bp = taosStrpTime(bp, "%I:%M:%S %p", tm)))
return (0);
break;
case 'T': /* The time as "%H:%M:%S". */
LEGAL_ALT(0);
if (!(bp = taosStrpTime(bp, "%H:%M:%S", tm)))
return (0);
break;
case 'X': /* The time, using the locale's format. */
LEGAL_ALT(ALT_E);
if (!(bp = taosStrpTime(bp, "%H:%M:%S", tm)))
return (0);
break;
case 'x': /* The date, using the locale's format. */
LEGAL_ALT(ALT_E);
if (!(bp = taosStrpTime(bp, "%m/%d/%y", tm)))
return (0);
break;
/*
* "Elementary" conversion rules.
*/
case 'A': /* The day of week, using the locale's form. */
case 'a':
LEGAL_ALT(0);
for (i = 0; i < 7; i++) {
/* Full name. */
len = strlen(day[i]);
if (strncmp(day[i], bp, len) == 0)
break;
/* Abbreviated name. */
len = strlen(abday[i]);
if (strncmp(abday[i], bp, len) == 0)
break;
}
/* Nothing matched. */
if (i == 7)
return (0);
tm->tm_wday = i;
bp += len;
break;
case 'B': /* The month, using the locale's form. */
case 'b':
case 'h':
LEGAL_ALT(0);
for (i = 0; i < 12; i++) {
/* Full name. */
len = strlen(mon[i]);
if (strncmp(mon[i], bp, len) == 0)
break;
/* Abbreviated name. */
len = strlen(abmon[i]);
if (strncmp(abmon[i], bp, len) == 0)
break;
}
/* Nothing matched. */
if (i == 12)
return (0);
tm->tm_mon = i;
bp += len;
break;
case 'C': /* The century number. */
LEGAL_ALT(ALT_E);
if (!(conv_num(&bp, &i, 0, 99)))
return (0);
if (split_year) {
tm->tm_year = (tm->tm_year % 100) + (i * 100);
}
else {
tm->tm_year = i * 100;
split_year = 1;
}
break;
case 'd': /* The day of month. */
case 'e':
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_mday, 1, 31)))
return (0);
break;
case 'k': /* The hour (24-hour clock representation). */
LEGAL_ALT(0);
/* FALLTHROUGH */
case 'H':
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_hour, 0, 23)))
return (0);
break;
case 'l': /* The hour (12-hour clock representation). */
LEGAL_ALT(0);
/* FALLTHROUGH */
case 'I':
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_hour, 1, 12)))
return (0);
if (tm->tm_hour == 12)
tm->tm_hour = 0;
break;
case 'j': /* The day of year. */
LEGAL_ALT(0);
if (!(conv_num(&bp, &i, 1, 366)))
return (0);
tm->tm_yday = i - 1;
break;
case 'M': /* The minute. */
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_min, 0, 59)))
return (0);
break;
case 'm': /* The month. */
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &i, 1, 12)))
return (0);
tm->tm_mon = i - 1;
break;
case 'p': /* The locale's equivalent of AM/PM. */
LEGAL_ALT(0);
/* AM? */
if (strcmp(am_pm[0], bp) == 0) {
if (tm->tm_hour > 11)
return (0);
bp += strlen(am_pm[0]);
break;
}
/* PM? */
else if (strcmp(am_pm[1], bp) == 0) {
if (tm->tm_hour > 11)
return (0);
tm->tm_hour += 12;
bp += strlen(am_pm[1]);
break;
}
/* Nothing matched. */
return (0);
case 'S': /* The seconds. */
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_sec, 0, 61)))
return (0);
break;
case 'U': /* The week of year, beginning on sunday. */
case 'W': /* The week of year, beginning on monday. */
LEGAL_ALT(ALT_O);
/*
* XXX This is bogus, as we can not assume any valid
* information present in the tm structure at this
* point to calculate a real value, so just check the
* range for now.
*/
if (!(conv_num(&bp, &i, 0, 53)))
return (0);
break;
case 'w': /* The day of week, beginning on sunday. */
LEGAL_ALT(ALT_O);
if (!(conv_num(&bp, &tm->tm_wday, 0, 6)))
return (0);
break;
case 'Y': /* The year. */
LEGAL_ALT(ALT_E);
if (!(conv_num(&bp, &i, 0, 9999)))
return (0);
tm->tm_year = i - TM_YEAR_BASE;
break;
case 'y': /* The year within 100 years of the epoch. */
LEGAL_ALT(ALT_E | ALT_O);
if (!(conv_num(&bp, &i, 0, 99)))
return (0);
if (split_year) {
tm->tm_year = ((tm->tm_year / 100) * 100) + i;
break;
}
split_year = 1;
if (i <= 68)
tm->tm_year = i + 2000 - TM_YEAR_BASE;
else
tm->tm_year = i + 1900 - TM_YEAR_BASE;
break;
/*
* Miscellaneous conversions.
*/
case 'n': /* Any kind of white-space. */
case 't':
LEGAL_ALT(0);
while (isspace(*bp))
bp++;
break;
default: /* Unknown/unsupported conversion. */
return (0);
}
}
/* LINTED functional specification */
return ((char *)bp);
#else
return strptime(buf, fmt, tm);
#endif
}
FORCE_INLINE int32_t taosGetTimeOfDay(struct timeval *tv) {
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
time_t t;
t = time(NULL);
t = taosGetTimestampSec();
SYSTEMTIME st;
GetLocalTime(&st);
@ -43,26 +401,18 @@ int taosGetTimeOfDay(struct timeval *tv, struct timezone *tz) {
tv->tv_usec = st.wMilliseconds * 1000;
return 0;
#else
return gettimeofday(tv, NULL);
#endif
}
struct tm *localtime_r(const time_t *timep, struct tm *result) {
struct tm *taosLocalTime(const time_t *timep, struct tm *result) {
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
localtime_s(result, timep);
#else
localtime_r(timep, result);
#endif
return result;
}
#else
/*
* linux and darwin implementation
*/
#include <sys/time.h>
// #include "monotonic.h"
FORCE_INLINE int32_t taosGetTimeOfDay(struct timeval *tv) {
return gettimeofday(tv, NULL);
}
int32_t taosGetTimestampSec() { return (int32_t)time(NULL); }
#endif

111
source/os/src/osTimer.c
View File

@ -13,15 +13,11 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define ALLOW_FORBID_FUNC
#define _DEFAULT_SOURCE
#include "os.h"
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
/*
* windows implementation
*/
#include <Mmsystem.h>
#include <Windows.h>
#include <stdint.h>
@ -39,24 +35,9 @@ void WINAPI taosWinOnTimer(UINT wTimerID, UINT msg, DWORD_PTR dwUser, DWORD_PTR
}
static MMRESULT timerId;
int taosInitTimer(win_timer_f callback, int ms) {
DWORD_PTR param = *((int64_t *)&callback);
timerId = timeSetEvent(ms, 1, (LPTIMECALLBACK)taosWinOnTimer, param, TIME_PERIODIC);
if (timerId == 0) {
return -1;
}
return 0;
}
void taosUninitTimer() { timeKillEvent(timerId); }
#elif defined(_TD_DARWIN_64)
/*
* darwin implementation
*/
#include <sys/event.h>
#include <sys/syscall.h>
#include <unistd.h>
@ -88,53 +69,12 @@ static void* timer_routine(void* arg) {
return NULL;
}
int taosInitTimer(void (*callback)(int), int ms) {
int r = 0;
timer_kq = -1;
timer_stop = 0;
timer_ms = ms;
timer_callback = callback;
timer_kq = kqueue();
if (timer_kq == -1) {
fprintf(stderr, "==%s[%d]%s()==failed to create timer kq\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__);
// since no caller of this func checks the return value for the moment
abort();
}
r = pthread_create(&timer_thread, NULL, timer_routine, NULL);
if (r) {
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();
}
return 0;
}
void taosUninitTimer() {
int r = 0;
timer_stop = 1;
r = pthread_join(timer_thread, NULL);
if (r) {
fprintf(stderr, "==%s[%d]%s()==failed to join timer thread\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__);
// since no caller of this func checks the return value for the moment
abort();
}
close(timer_kq);
timer_kq = -1;
}
void taos_block_sigalrm(void) {
// we don't know if there's any specific API for SIGALRM to deliver to specific thread
// this implementation relies on kqueue rather than SIGALRM
}
#else
/*
* linux implementation
*/
#include <sys/syscall.h>
#include <unistd.h>
@ -200,8 +140,39 @@ static void * taosProcessAlarmSignal(void *tharg) {
return NULL;
}
#endif
int taosInitTimer(void (*callback)(int), int ms) {
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
DWORD_PTR param = *((int64_t *)&callback);
timerId = timeSetEvent(ms, 1, (LPTIMECALLBACK)taosWinOnTimer, param, TIME_PERIODIC);
if (timerId == 0) {
return -1;
}
return 0;
#elif defined(_TD_DARWIN_64)
int r = 0;
timer_kq = -1;
timer_stop = 0;
timer_ms = ms;
timer_callback = callback;
timer_kq = kqueue();
if (timer_kq == -1) {
fprintf(stderr, "==%s[%d]%s()==failed to create timer kq\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__);
// since no caller of this func checks the return value for the moment
abort();
}
r = pthread_create(&timer_thread, NULL, timer_routine, NULL);
if (r) {
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();
}
return 0;
#else
stopTimer = false;
pthread_attr_t tattr;
pthread_attr_init(&tattr);
@ -215,13 +186,29 @@ int taosInitTimer(void (*callback)(int), int ms) {
}
return 0;
#endif
}
void taosUninitTimer() {
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
timeKillEvent(timerId);
#elif defined(_TD_DARWIN_64)
int r = 0;
timer_stop = 1;
r = pthread_join(timer_thread, NULL);
if (r) {
fprintf(stderr, "==%s[%d]%s()==failed to join timer thread\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__);
// since no caller of this func checks the return value for the moment
abort();
}
close(timer_kq);
timer_kq = -1;
#else
stopTimer = true;
// printf("join timer thread:0x%08" PRIx64, taosGetPthreadId(timerThread));
pthread_join(timerThread, NULL);
#endif
}
int64_t taosGetMonotonicMs() {
@ -239,5 +226,3 @@ const char *taosMonotonicInit() {
return NULL;
#endif
}
#endif

8
source/os/src/osTimezone.c
View File

@ -128,9 +128,9 @@ void taosGetSystemTimezone(char *outTimezone) {
* Enforce set the correct daylight saving time(DST) flag according
* to current time
*/
time_t tx1 = time(NULL);
time_t tx1 = taosGetTimestampSec();
struct tm tm1;
localtime_r(&tx1, &tm1);
taosLocalTime(&tx1, &tm1);
/*
* format example:
@ -147,9 +147,9 @@ void taosGetSystemTimezone(char *outTimezone) {
* Enforce set the correct daylight saving time(DST) flag according
* to current time
*/
time_t tx1 = time(NULL);
time_t tx1 = taosGetTimestampSec();
struct tm tm1;
localtime_r(&tx1, &tm1);
taosLocalTime(&tx1, &tm1);
/* load time zone string from /etc/timezone */
// FILE *f = fopen("/etc/timezone", "r");

2
source/util/src/tdes.c
View File

@ -31,7 +31,7 @@ void process_message(uint8_t* message_piece, uint8_t* processed_piece, key_set*
#if 0
int64_t taosDesGenKey() {
uint32_t iseed = (uint32_t)time(NULL);
uint32_t iseed = (uint32_t)taosGetTimestampSec();
taosSeedRand(iseed);
uint8_t key[8] = {0};

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

@ -401,7 +401,7 @@ static inline int32_t taosBuildLogHead(char *buffer, const char *flags) {
taosGetTimeOfDay(&timeSecs);
time_t curTime = timeSecs.tv_sec;
ptm = localtime_r(&curTime, &Tm);
ptm = taosLocalTime(&curTime, &Tm);
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);

2
source/util/src/tskiplist.c
View File

@ -82,7 +82,7 @@ SSkipList *tSkipListCreate(uint8_t maxLevel, uint8_t keyType, uint16_t keyLen, _
}
}
taosSeedRand((uint32_t)time(NULL));
taosSeedRand((uint32_t)taosGetTimestampSec());
#if SKIP_LIST_RECORD_PERFORMANCE
pSkipList->state.nTotalMemSize += sizeof(SSkipList);

4
source/util/test/codingTests.cpp
View File

@ -150,7 +150,7 @@ static bool test_variant_int64(int64_t value) {
}
TEST(codingTest, fixed_encode_decode) {
taosSeedRand(time(0));
taosSeedRand(taosGetTimestampSec());
// uint16_t
for (uint16_t value = 0; value <= UINT16_MAX; value++) {
@ -204,7 +204,7 @@ TEST(codingTest, fixed_encode_decode) {
}
TEST(codingTest, variant_encode_decode) {
taosSeedRand(time(0));
taosSeedRand(taosGetTimestampSec());
// uint16_t
for (uint16_t value = 0; value <= UINT16_MAX; value++) {

2
source/util/test/pageBufferTest.cpp
View File

@ -161,7 +161,7 @@ void recyclePageTest() {
TEST(testCase, resultBufferTest) {
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
simpleTest();
writeDownTest();
recyclePageTest();

4
source/util/test/skiplistTest.cpp
View File

@ -74,7 +74,7 @@ void randKeyTest() {
false, getkey);
int32_t size = 200000;
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
printf("generated %d keys is: \n", size);
@ -337,7 +337,7 @@ void duplicatedKeyTest() {
TEST(testCase, skiplist_test) {
assert(sizeof(SSkipListKey) == 8);
taosSeedRand(time(NULL));
taosSeedRand(taosGetTimestampSec());
stringKeySkiplistTest();
doubleSkipListTest();

2
tests/script/api/batchprepare.c
View File

@ -18,7 +18,7 @@ void taosMsleep(int mseconds);
unsigned long long getCurrentTime(){
struct timeval tv;
if (gettimeofday(&tv, NULL) != 0) {
if (taosGetTimeOfDay(&tv) != 0) {
perror("Failed to get current time in ms");
exit(EXIT_FAILURE);
}

2
tests/script/api/stmtBatchTest.c
View File

@ -42,7 +42,7 @@ int g_runTimes = 5;
unsigned long long getCurrentTime(){
struct timeval tv;
if (gettimeofday(&tv, NULL) != 0) {
if (taosGetTimeOfDay(&tv) != 0) {
perror("Failed to get current time in ms");
exit(EXIT_FAILURE);
}

2
tests/test/c/tmqDemo.c
View File

@ -609,7 +609,7 @@ void printParaIntoFile() {
};
g_fp = pFile;
time_t tTime = time(NULL);
time_t tTime = taosGetTimestampSec();
struct tm tm = *localtime(&tTime);
taosFprintfFile(pFile, "###################################################################\n");