Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
d2c59701
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
d2c59701
编写于
3月 13, 2022
作者:
C
Cary Xu
浏览文件
操作
浏览文件
下载
差异文件
Merge branch '3.0' into feature/TD-11463-3.0
上级
421d087a
02981347
变更
19
展开全部
隐藏空白更改
内联
并排
Showing
19 changed file
with
404 addition
and
650 deletion
+404
-650
include/client/taos.h
include/client/taos.h
+21
-21
include/libs/executor/dataSinkMgt.h
include/libs/executor/dataSinkMgt.h
+0
-1
include/libs/transport/trpc.h
include/libs/transport/trpc.h
+0
-9
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+23
-14
source/libs/executor/src/dataDispatcher.c
source/libs/executor/src/dataDispatcher.c
+2
-17
source/libs/executor/src/executorMain.c
source/libs/executor/src/executorMain.c
+1
-1
source/libs/executor/src/executorimpl.c
source/libs/executor/src/executorimpl.c
+186
-166
source/libs/executor/test/executorTests.cpp
source/libs/executor/test/executorTests.cpp
+6
-6
source/libs/nodes/src/nodesCloneFuncs.c
source/libs/nodes/src/nodesCloneFuncs.c
+0
-1
source/libs/nodes/src/nodesCodeFuncs.c
source/libs/nodes/src/nodesCodeFuncs.c
+0
-2
source/libs/nodes/src/nodesUtilFuncs.c
source/libs/nodes/src/nodesUtilFuncs.c
+0
-1
source/libs/parser/src/parTranslater.c
source/libs/parser/src/parTranslater.c
+0
-3
source/libs/planner/test/plannerTest.cpp
source/libs/planner/test/plannerTest.cpp
+1
-1
source/libs/qworker/src/qworker.c
source/libs/qworker/src/qworker.c
+1
-1
source/libs/scalar/src/sclvector.c
source/libs/scalar/src/sclvector.c
+20
-21
source/libs/transport/src/transCli.c
source/libs/transport/src/transCli.c
+136
-118
source/libs/transport/src/transSrv.c
source/libs/transport/src/transSrv.c
+7
-7
source/libs/transport/test/CMakeLists.txt
source/libs/transport/test/CMakeLists.txt
+0
-18
source/libs/transport/test/pushClient.c
source/libs/transport/test/pushClient.c
+0
-242
未找到文件。
include/client/taos.h
浏览文件 @
d2c59701
...
...
@@ -31,27 +31,27 @@ typedef void TAOS_SUB;
typedef
void
**
TAOS_ROW
;
// Data type definition
#define TSDB_DATA_TYPE_NULL
0
// 1 bytes
#define TSDB_DATA_TYPE_BOOL
1
// 1 bytes
#define TSDB_DATA_TYPE_TINYINT
2
// 1 byte
#define TSDB_DATA_TYPE_SMALLINT
3
// 2 bytes
#define TSDB_DATA_TYPE_INT
4
// 4 bytes
#define TSDB_DATA_TYPE_BIGINT
5
// 8 bytes
#define TSDB_DATA_TYPE_FLOAT
6
// 4 bytes
#define TSDB_DATA_TYPE_DOUBLE
7
// 8 bytes
#define TSDB_DATA_TYPE_
BINARY 8
// string, alias for varchar
#define TSDB_DATA_TYPE_TIMESTAMP
9
// 8 bytes
#define TSDB_DATA_TYPE_NCHAR
10
// unicode string
#define TSDB_DATA_TYPE_UTINYINT
11
// 1 byte
#define TSDB_DATA_TYPE_USMALLINT
12
// 2 bytes
#define TSDB_DATA_TYPE_UINT
13
// 4 bytes
#define TSDB_DATA_TYPE_UBIGINT
14
// 8 bytes
#define TSDB_DATA_TYPE_
VARCHAR 15 //
string
#define TSDB_DATA_TYPE_VARBINARY
16
// binary
#define TSDB_DATA_TYPE_
JSON 17 // json
#define TSDB_DATA_TYPE_
DECIMAL 18 // decimal
#define TSDB_DATA_TYPE_
BLOB 19 // binary
#define TSDB_DATA_TYPE_
MEDIUMBLOB 20
#define TSDB_DATA_TYPE_NULL
0
// 1 bytes
#define TSDB_DATA_TYPE_BOOL
1
// 1 bytes
#define TSDB_DATA_TYPE_TINYINT
2
// 1 byte
#define TSDB_DATA_TYPE_SMALLINT
3
// 2 bytes
#define TSDB_DATA_TYPE_INT
4
// 4 bytes
#define TSDB_DATA_TYPE_BIGINT
5
// 8 bytes
#define TSDB_DATA_TYPE_FLOAT
6
// 4 bytes
#define TSDB_DATA_TYPE_DOUBLE
7
// 8 bytes
#define TSDB_DATA_TYPE_
VARCHAR 8
// string, alias for varchar
#define TSDB_DATA_TYPE_TIMESTAMP
9
// 8 bytes
#define TSDB_DATA_TYPE_NCHAR
10
// unicode string
#define TSDB_DATA_TYPE_UTINYINT
11
// 1 byte
#define TSDB_DATA_TYPE_USMALLINT
12
// 2 bytes
#define TSDB_DATA_TYPE_UINT
13
// 4 bytes
#define TSDB_DATA_TYPE_UBIGINT
14
// 8 bytes
#define TSDB_DATA_TYPE_
JSON 15 // json
string
#define TSDB_DATA_TYPE_VARBINARY
16
// binary
#define TSDB_DATA_TYPE_
DECIMAL 17 // decimal
#define TSDB_DATA_TYPE_
BLOB 18 // binary
#define TSDB_DATA_TYPE_
MEDIUMBLOB 19
#define TSDB_DATA_TYPE_
BINARY TSDB_DATA_TYPE_VARCHAR // string
typedef
enum
{
TSDB_OPTION_LOCALE
,
...
...
include/libs/executor/dataSinkMgt.h
浏览文件 @
d2c59701
...
...
@@ -41,7 +41,6 @@ int32_t dsDataSinkMgtInit(SDataSinkMgtCfg *cfg);
typedef
struct
SInputData
{
const
struct
SSDataBlock
*
pData
;
SHashObj
*
pTableRetrieveTsMap
;
}
SInputData
;
typedef
struct
SOutputData
{
...
...
include/libs/transport/trpc.h
浏览文件 @
d2c59701
...
...
@@ -29,7 +29,6 @@ extern "C" {
extern
int
tsRpcHeadSize
;
typedef
struct
SRpcPush
SRpcPush
;
typedef
struct
SRpcConnInfo
{
uint32_t
clientIp
;
...
...
@@ -45,16 +44,8 @@ typedef struct SRpcMsg {
int32_t
code
;
void
*
handle
;
// rpc handle returned to app
void
*
ahandle
;
// app handle set by client
int
persist
;
// keep handle or not, default 0
SRpcPush
*
push
;
}
SRpcMsg
;
typedef
struct
SRpcPush
{
void
*
arg
;
int
(
*
callback
)(
void
*
arg
,
SRpcMsg
*
rpcMsg
);
}
SRpcPush
;
typedef
struct
SRpcInit
{
uint16_t
localPort
;
// local port
...
...
source/libs/executor/inc/executorimpl.h
浏览文件 @
d2c59701
...
...
@@ -253,6 +253,11 @@ typedef struct STaskIdInfo {
char
*
str
;
}
STaskIdInfo
;
typedef
struct
STaskBufInfo
{
int32_t
bufSize
;
// total available buffer size in bytes
int32_t
remainBuf
;
// remain buffer size
}
STaskBufInfo
;
typedef
struct
SExecTaskInfo
{
STaskIdInfo
id
;
char
*
content
;
...
...
@@ -264,7 +269,8 @@ typedef struct SExecTaskInfo {
uint64_t
totalRows
;
// total number of rows
STableGroupInfo
tableqinfoGroupInfo
;
// this is a group array list, including SArray<STableQueryInfo*> structure
char
*
sql
;
// query sql string
jmp_buf
env
;
//
jmp_buf
env
;
// when error occurs, abort
STaskBufInfo
bufInfo
;
// available buffer info this task
struct
SOperatorInfo
*
pRoot
;
}
SExecTaskInfo
;
...
...
@@ -307,9 +313,11 @@ typedef struct STaskRuntimeEnv {
}
STaskRuntimeEnv
;
enum
{
OP_IN_EXECUTING
=
1
,
OP_RES_TO_RETURN
=
2
,
OP_EXEC_DONE
=
3
,
OP_NOT_OPENED
=
0x0
,
OP_OPENED
=
0x1
,
OP_IN_EXECUTING
=
0x3
,
OP_RES_TO_RETURN
=
0x5
,
OP_EXEC_DONE
=
0x9
,
};
typedef
struct
SOperatorInfo
{
...
...
@@ -322,12 +330,14 @@ typedef struct SOperatorInfo {
SExprInfo
*
pExpr
;
STaskRuntimeEnv
*
pRuntimeEnv
;
// todo remove it
SExecTaskInfo
*
pTaskInfo
;
SOperatorCostInfo
cost
;
struct
SOperatorInfo
**
pDownstream
;
// downstram pointer list
int32_t
numOfDownstream
;
// number of downstream. The value is always ONE expect for join operator
__optr_
open_fn_t
open
Fn
;
__optr_fn_t
nextData
Fn
;
__optr_
fn_t
getNext
Fn
;
__optr_fn_t
cleanup
Fn
;
__optr_close_fn_t
closeFn
;
__optr_open_fn_t
_openFn
;
// DO NOT invoke this function directly
}
SOperatorInfo
;
typedef
struct
{
...
...
@@ -378,9 +388,9 @@ typedef struct STaskParam {
}
STaskParam
;
enum
{
DATA_NOT_READY
=
0x1
,
DATA_READY
=
0x2
,
DATA_EXHAUSTED
=
0x3
,
EX_SOURCE_
DATA_NOT_READY
=
0x1
,
EX_SOURCE_
DATA_READY
=
0x2
,
EX_SOURCE_
DATA_EXHAUSTED
=
0x3
,
};
typedef
struct
SSourceDataInfo
{
...
...
@@ -639,12 +649,16 @@ SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntim
SOperatorInfo
*
createAggregateOperatorInfo
(
SOperatorInfo
*
downstream
,
SArray
*
pExprInfo
,
SSDataBlock
*
pResultBlock
,
SExecTaskInfo
*
pTaskInfo
,
const
STableGroupInfo
*
pTableGroupInfo
);
SOperatorInfo
*
createMultiTableAggOperatorInfo
(
SOperatorInfo
*
downstream
,
SArray
*
pExprInfo
,
SSDataBlock
*
pResultBlock
,
SExecTaskInfo
*
pTaskInfo
,
const
STableGroupInfo
*
pTableGroupInfo
);
SOperatorInfo
*
createProjectOperatorInfo
(
SOperatorInfo
*
downstream
,
SArray
*
pExprInfo
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createOrderOperatorInfo
(
SOperatorInfo
*
downstream
,
SArray
*
pExprInfo
,
SArray
*
pOrderVal
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createSortedMergeOperatorInfo
(
SOperatorInfo
**
downstream
,
int32_t
numOfDownstream
,
SArray
*
pExprInfo
,
SArray
*
pOrderVal
,
SArray
*
pGroupInfo
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createSysTableScanOperatorInfo
(
void
*
pSysTableReadHandle
,
const
SArray
*
pExprInfo
,
const
SSchema
*
pSchema
,
int32_t
tableType
,
SEpSet
epset
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createLimitOperatorInfo
(
STaskRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
downstream
);
SOperatorInfo
*
createIntervalOperatorInfo
(
SOperatorInfo
*
downstream
,
SArray
*
pExprInfo
,
SInterval
*
pInterval
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createLimitOperatorInfo
(
STaskRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
downstream
);
SOperatorInfo
*
createAllTimeIntervalOperatorInfo
(
STaskRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
downstream
,
SExprInfo
*
pExpr
,
int32_t
numOfOutput
);
SOperatorInfo
*
createSWindowOperatorInfo
(
STaskRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
downstream
,
SExprInfo
*
pExpr
,
...
...
@@ -674,10 +688,6 @@ SOperatorInfo* createSLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorI
SOperatorInfo
*
createJoinOperatorInfo
(
SOperatorInfo
**
pdownstream
,
int32_t
numOfDownstream
,
SSchema
*
pSchema
,
int32_t
numOfOutput
);
SOperatorInfo
*
createOrderOperatorInfo
(
SOperatorInfo
*
downstream
,
SArray
*
pExprInfo
,
SArray
*
pOrderVal
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createSortedMergeOperatorInfo
(
SOperatorInfo
**
downstream
,
int32_t
numOfDownstream
,
SArray
*
pExprInfo
,
SArray
*
pOrderVal
,
SArray
*
pGroupInfo
,
SExecTaskInfo
*
pTaskInfo
);
// SSDataBlock* doSLimit(void* param, bool* newgroup);
// int32_t doCreateFilterInfo(SColumnInfo* pCols, int32_t numOfCols, int32_t numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo, uint64_t qId);
void
doSetFilterColumnInfo
(
SSingleColumnFilterInfo
*
pFilterInfo
,
int32_t
numOfFilterCols
,
SSDataBlock
*
pBlock
);
...
...
@@ -691,7 +701,6 @@ void* doDestroyFilterInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFil
void
setInputDataBlock
(
SOperatorInfo
*
pOperator
,
SqlFunctionCtx
*
pCtx
,
SSDataBlock
*
pBlock
,
int32_t
order
);
void
finalizeQueryResult
(
SOperatorInfo
*
pOperator
,
SqlFunctionCtx
*
pCtx
,
SResultRowInfo
*
pResultRowInfo
,
int32_t
*
rowCellInfoOffset
);
void
updateOutputBuf
(
SOptrBasicInfo
*
pBInfo
,
int32_t
*
bufCapacity
,
int32_t
numOfInputRows
);
void
clearOutputBuf
(
SOptrBasicInfo
*
pBInfo
,
int32_t
*
bufCapacity
);
void
copyTsColoum
(
SSDataBlock
*
pRes
,
SqlFunctionCtx
*
pCtx
,
int32_t
numOfOutput
);
...
...
source/libs/executor/src/dataDispatcher.c
浏览文件 @
d2c59701
...
...
@@ -21,8 +21,6 @@
#include "tqueue.h"
#include "executorimpl.h"
#define DATA_META_LENGTH(tables) (sizeof(int32_t) + sizeof(STableIdInfo) * taosHashGetSize(tables) + sizeof(SRetrieveTableRsp))
typedef
struct
SDataDispatchBuf
{
int32_t
useSize
;
int32_t
allocSize
;
...
...
@@ -90,19 +88,6 @@ static void copyData(const SInputData* pInput, const SDataBlockDescNode* pSchema
data
+=
pColRes
->
info
.
bytes
*
pInput
->
pData
->
info
.
rows
;
}
}
int32_t
numOfTables
=
(
int32_t
)
taosHashGetSize
(
pInput
->
pTableRetrieveTsMap
);
*
(
int32_t
*
)
data
=
htonl
(
numOfTables
);
data
+=
sizeof
(
int32_t
);
STableIdInfo
*
item
=
taosHashIterate
(
pInput
->
pTableRetrieveTsMap
,
NULL
);
while
(
item
)
{
STableIdInfo
*
pDst
=
(
STableIdInfo
*
)
data
;
pDst
->
uid
=
htobe64
(
item
->
uid
);
pDst
->
key
=
htobe64
(
item
->
key
);
data
+=
sizeof
(
STableIdInfo
);
item
=
taosHashIterate
(
pInput
->
pTableRetrieveTsMap
,
item
);
}
}
// data format with compress: SDataCacheEntry | cols_data_offset | col1_data col2_data ... | numOfTables | STableIdInfo STableIdInfo ...
...
...
@@ -113,7 +98,7 @@ static void toDataCacheEntry(const SDataDispatchHandle* pHandle, const SInputDat
pEntry
->
numOfRows
=
pInput
->
pData
->
info
.
rows
;
pEntry
->
dataLen
=
0
;
pBuf
->
useSize
=
DATA_META_LENGTH
(
pInput
->
pTableRetrieveTsMa
p
);
pBuf
->
useSize
=
sizeof
(
SRetrieveTableRs
p
);
copyData
(
pInput
,
pHandle
->
pSchema
,
pEntry
->
data
,
pEntry
->
compressed
,
&
pEntry
->
dataLen
);
if
(
0
==
pEntry
->
compressed
)
{
pEntry
->
dataLen
=
pHandle
->
pSchema
->
resultRowSize
*
pInput
->
pData
->
info
.
rows
;
...
...
@@ -130,7 +115,7 @@ static bool allocBuf(SDataDispatchHandle* pDispatcher, const SInputData* pInput,
return
false
;
}
pBuf
->
allocSize
=
DATA_META_LENGTH
(
pInput
->
pTableRetrieveTsMa
p
)
+
pDispatcher
->
pSchema
->
resultRowSize
*
pInput
->
pData
->
info
.
rows
;
pBuf
->
allocSize
=
sizeof
(
SRetrieveTableRs
p
)
+
pDispatcher
->
pSchema
->
resultRowSize
*
pInput
->
pData
->
info
.
rows
;
pBuf
->
pData
=
malloc
(
pBuf
->
allocSize
);
if
(
pBuf
->
pData
==
NULL
)
{
qError
(
"SinkNode failed to malloc memory, size:%d, code:%d"
,
pBuf
->
allocSize
,
TAOS_SYSTEM_ERROR
(
errno
));
...
...
source/libs/executor/src/executorMain.c
浏览文件 @
d2c59701
...
...
@@ -158,7 +158,7 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
int64_t
st
=
0
;
st
=
taosGetTimestampUs
();
*
pRes
=
pTaskInfo
->
pRoot
->
nextData
Fn
(
pTaskInfo
->
pRoot
,
&
newgroup
);
*
pRes
=
pTaskInfo
->
pRoot
->
getNext
Fn
(
pTaskInfo
->
pRoot
,
&
newgroup
);
uint64_t
el
=
(
taosGetTimestampUs
()
-
st
);
pTaskInfo
->
cost
.
elapsedTime
+=
el
;
...
...
source/libs/executor/src/executorimpl.c
浏览文件 @
d2c59701
此差异已折叠。
点击以展开。
source/libs/executor/test/executorTests.cpp
浏览文件 @
d2c59701
...
...
@@ -201,9 +201,9 @@ SOperatorInfo* createDummyOperator(int32_t startVal, int32_t numOfBlocks, int32_
pOperator
->
name
=
"dummyInputOpertor4Test"
;
if
(
numOfCols
==
1
)
{
pOperator
->
nextData
Fn
=
getDummyBlock
;
pOperator
->
getNext
Fn
=
getDummyBlock
;
}
else
{
pOperator
->
nextData
Fn
=
get2ColsDummyBlock
;
pOperator
->
getNext
Fn
=
get2ColsDummyBlock
;
}
SDummyInputInfo
*
pInfo
=
(
SDummyInputInfo
*
)
calloc
(
1
,
sizeof
(
SDummyInputInfo
));
...
...
@@ -971,7 +971,7 @@ TEST(testCase, inMem_sort_Test) {
SOperatorInfo* pOperator = createOrderOperatorInfo(createDummyOperator(10000, 5, 1000, data_asc, 1), pExprInfo, pOrderVal, NULL);
bool newgroup = false;
SSDataBlock* pRes = pOperator->
nextData
Fn(pOperator, &newgroup);
SSDataBlock* pRes = pOperator->
getNext
Fn(pOperator, &newgroup);
SColumnInfoData* pCol1 = static_cast<SColumnInfoData*>(taosArrayGet(pRes->pDataBlock, 0));
SColumnInfoData* pCol2 = static_cast<SColumnInfoData*>(taosArrayGet(pRes->pDataBlock, 1));
...
...
@@ -1049,7 +1049,7 @@ TEST(testCase, external_sort_Test) {
while(1) {
int64_t s = taosGetTimestampUs();
pRes = pOperator->
nextData
Fn(pOperator, &newgroup);
pRes = pOperator->
getNext
Fn(pOperator, &newgroup);
int64_t e = taosGetTimestampUs();
if (t++ == 1) {
...
...
@@ -1121,7 +1121,7 @@ TEST(testCase, sorted_merge_Test) {
while(1) {
int64_t s = taosGetTimestampUs();
pRes = pOperator->
nextData
Fn(pOperator, &newgroup);
pRes = pOperator->
getNext
Fn(pOperator, &newgroup);
int64_t e = taosGetTimestampUs();
if (t++ == 1) {
...
...
@@ -1199,7 +1199,7 @@ TEST(testCase, time_interval_Operator_Test) {
while(1) {
int64_t s = taosGetTimestampUs();
pRes = pOperator->
nextData
Fn(pOperator, &newgroup);
pRes = pOperator->
getNext
Fn(pOperator, &newgroup);
int64_t e = taosGetTimestampUs();
if (t++ == 1) {
...
...
source/libs/nodes/src/nodesCloneFuncs.c
浏览文件 @
d2c59701
...
...
@@ -131,7 +131,6 @@ static SNode* valueNodeCopy(const SValueNode* pSrc, SValueNode* pDst) {
case
TSDB_DATA_TYPE_DOUBLE
:
COPY_SCALAR_FIELD
(
datum
.
d
);
break
;
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_NCHAR
:
case
TSDB_DATA_TYPE_VARCHAR
:
case
TSDB_DATA_TYPE_VARBINARY
:
...
...
source/libs/nodes/src/nodesCodeFuncs.c
浏览文件 @
d2c59701
...
...
@@ -978,7 +978,6 @@ static int32_t datumToJson(const void* pObj, SJson* pJson) {
case
TSDB_DATA_TYPE_DOUBLE
:
code
=
tjsonAddDoubleToObject
(
pJson
,
jkValueDatum
,
pNode
->
datum
.
d
);
break
;
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_NCHAR
:
case
TSDB_DATA_TYPE_VARCHAR
:
case
TSDB_DATA_TYPE_VARBINARY
:
...
...
@@ -1042,7 +1041,6 @@ static int32_t jsonToDatum(const SJson* pJson, void* pObj) {
case
TSDB_DATA_TYPE_DOUBLE
:
code
=
tjsonGetDoubleValue
(
pJson
,
jkValueDatum
,
&
pNode
->
datum
.
d
);
break
;
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_NCHAR
:
case
TSDB_DATA_TYPE_VARCHAR
:
case
TSDB_DATA_TYPE_VARBINARY
:
...
...
source/libs/nodes/src/nodesUtilFuncs.c
浏览文件 @
d2c59701
...
...
@@ -386,7 +386,6 @@ void* nodesGetValueFromNode(SValueNode *pNode) {
case
TSDB_DATA_TYPE_FLOAT
:
case
TSDB_DATA_TYPE_DOUBLE
:
return
(
void
*
)
&
pNode
->
datum
.
d
;
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_NCHAR
:
case
TSDB_DATA_TYPE_VARCHAR
:
case
TSDB_DATA_TYPE_VARBINARY
:
...
...
source/libs/parser/src/parTranslater.c
浏览文件 @
d2c59701
...
...
@@ -287,7 +287,6 @@ static EDealRes translateValue(STranslateContext* pCxt, SValueNode* pVal) {
pVal
->
datum
.
d
=
strtold
(
pVal
->
literal
,
&
endPtr
);
break
;
}
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_NCHAR
:
case
TSDB_DATA_TYPE_VARCHAR
:
case
TSDB_DATA_TYPE_VARBINARY
:
{
...
...
@@ -601,7 +600,6 @@ static int32_t translateStar(STranslateContext* pCxt, SSelectStmt* pSelect, bool
static
int32_t
getPositionValue
(
const
SValueNode
*
pVal
)
{
switch
(
pVal
->
node
.
resType
.
type
)
{
case
TSDB_DATA_TYPE_NULL
:
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_TIMESTAMP
:
case
TSDB_DATA_TYPE_NCHAR
:
case
TSDB_DATA_TYPE_VARCHAR
:
...
...
@@ -1438,7 +1436,6 @@ static void valueNodeToVariant(const SValueNode* pNode, SVariant* pVal) {
case
TSDB_DATA_TYPE_DOUBLE
:
pVal
->
d
=
pNode
->
datum
.
d
;
break
;
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_NCHAR
:
case
TSDB_DATA_TYPE_VARCHAR
:
case
TSDB_DATA_TYPE_VARBINARY
:
...
...
source/libs/planner/test/plannerTest.cpp
浏览文件 @
d2c59701
...
...
@@ -56,7 +56,7 @@ protected:
const
string
syntaxTreeStr
=
toString
(
query_
->
pRoot
,
false
);
SLogicNode
*
pLogicPlan
=
nullptr
;
SPlanContext
cxt
=
{
.
queryId
=
1
,
.
pAstRoot
=
query_
->
pRoot
};
SPlanContext
cxt
=
{
.
queryId
=
1
,
.
acctId
=
0
,
.
pAstRoot
=
query_
->
pRoot
};
code
=
createLogicPlan
(
&
cxt
,
&
pLogicPlan
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
cout
<<
"sql:["
<<
cxt_
.
pSql
<<
"] logic plan code:"
<<
code
<<
", strerror:"
<<
tstrerror
(
code
)
<<
endl
;
...
...
source/libs/qworker/src/qworker.c
浏览文件 @
d2c59701
...
...
@@ -495,7 +495,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) {
ASSERT
(
pRes
->
info
.
rows
>
0
);
SInputData
inputData
=
{.
pData
=
pRes
,
.
pTableRetrieveTsMap
=
NULL
};
SInputData
inputData
=
{.
pData
=
pRes
};
code
=
dsPutDataBlock
(
sinkHandle
,
&
inputData
,
&
qcontinue
);
if
(
code
)
{
QW_TASK_ELOG
(
"dsPutDataBlock failed, code:%s"
,
tstrerror
(
code
));
...
...
source/libs/scalar/src/sclvector.c
浏览文件 @
d2c59701
...
...
@@ -411,27 +411,26 @@ int32_t vectorConvertImpl(SScalarParam* pIn, SScalarParam* pOut) {
}
int8_t
gConvertTypes
[
TSDB_DATA_TYPE_BLOB
+
1
][
TSDB_DATA_TYPE_BLOB
+
1
]
=
{
/* NULL BOOL TINY SMAL INT BIG FLOA DOUB BINA TIME NCHA UTIN USMA UINT UBIG VARC VARB JSON DECI BLOB */
/*NULL*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*BOOL*/
0
,
0
,
0
,
3
,
4
,
5
,
6
,
7
,
7
,
9
,
7
,
0
,
12
,
13
,
14
,
7
,
7
,
0
,
0
,
0
,
/*TINY*/
0
,
0
,
0
,
3
,
4
,
5
,
6
,
7
,
7
,
9
,
7
,
3
,
4
,
5
,
7
,
7
,
7
,
0
,
0
,
0
,
/*SMAL*/
0
,
0
,
0
,
0
,
4
,
5
,
6
,
7
,
7
,
9
,
7
,
3
,
4
,
5
,
7
,
7
,
7
,
0
,
0
,
0
,
/*INT */
0
,
0
,
0
,
0
,
0
,
5
,
6
,
7
,
7
,
9
,
7
,
4
,
4
,
5
,
7
,
7
,
7
,
0
,
0
,
0
,
/*BIGI*/
0
,
0
,
0
,
0
,
0
,
0
,
6
,
7
,
7
,
0
,
7
,
5
,
5
,
5
,
7
,
7
,
7
,
0
,
0
,
0
,
/*FLOA*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
7
,
6
,
7
,
6
,
6
,
6
,
6
,
7
,
7
,
0
,
0
,
0
,
/*DOUB*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
7
,
7
,
7
,
7
,
7
,
7
,
7
,
7
,
0
,
0
,
0
,
/*BINA*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
0
,
7
,
7
,
7
,
7
,
0
,
0
,
0
,
0
,
0
,
/*TIME*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
9
,
9
,
9
,
7
,
7
,
7
,
0
,
0
,
0
,
/*NCHA*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
7
,
7
,
7
,
0
,
0
,
0
,
0
,
0
,
/*UTIN*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
12
,
13
,
14
,
7
,
7
,
0
,
0
,
0
,
/*USMA*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
13
,
14
,
7
,
7
,
0
,
0
,
0
,
/*UINT*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
14
,
7
,
7
,
0
,
0
,
0
,
/*UBIG*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
7
,
0
,
0
,
0
,
/*VARC*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*VARB*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*JSON*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*DECI*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*BLOB*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
/* NULL BOOL TINY SMAL INT BIG FLOA DOUB VARC TIME NCHA UTIN USMA UINT UBIG VARB JSON DECI BLOB */
/*NULL*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*BOOL*/
0
,
0
,
0
,
3
,
4
,
5
,
6
,
7
,
7
,
9
,
7
,
0
,
12
,
13
,
14
,
7
,
0
,
0
,
0
,
/*TINY*/
0
,
0
,
0
,
3
,
4
,
5
,
6
,
7
,
7
,
9
,
7
,
3
,
4
,
5
,
7
,
7
,
0
,
0
,
0
,
/*SMAL*/
0
,
0
,
0
,
0
,
4
,
5
,
6
,
7
,
7
,
9
,
7
,
3
,
4
,
5
,
7
,
7
,
0
,
0
,
0
,
/*INT */
0
,
0
,
0
,
0
,
0
,
5
,
6
,
7
,
7
,
9
,
7
,
4
,
4
,
5
,
7
,
7
,
0
,
0
,
0
,
/*BIGI*/
0
,
0
,
0
,
0
,
0
,
0
,
6
,
7
,
7
,
0
,
7
,
5
,
5
,
5
,
7
,
7
,
0
,
0
,
0
,
/*FLOA*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
7
,
6
,
7
,
6
,
6
,
6
,
6
,
7
,
0
,
0
,
0
,
/*DOUB*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
7
,
7
,
7
,
7
,
7
,
7
,
7
,
0
,
0
,
0
,
/*VARC*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
0
,
7
,
7
,
7
,
7
,
0
,
0
,
0
,
0
,
/*TIME*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
9
,
9
,
9
,
7
,
7
,
0
,
0
,
0
,
/*NCHA*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
7
,
7
,
7
,
0
,
0
,
0
,
0
,
/*UTIN*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
12
,
13
,
14
,
7
,
0
,
0
,
0
,
/*USMA*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
13
,
14
,
7
,
0
,
0
,
0
,
/*UINT*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
14
,
7
,
0
,
0
,
0
,
/*UBIG*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
7
,
0
,
0
,
0
,
/*VARB*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*JSON*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*DECI*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
/*BLOB*/
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
,
0
};
int32_t
vectorGetConvertType
(
int32_t
type1
,
int32_t
type2
)
{
...
...
source/libs/transport/src/transCli.c
浏览文件 @
d2c59701
此差异已折叠。
点击以展开。
source/libs/transport/src/transSrv.c
浏览文件 @
d2c59701
...
...
@@ -92,9 +92,9 @@ static int uvAuthMsg(SSrvConn* pConn, char* msg, int msgLen);
static
void
uvAllocConnBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
);
static
void
uvAllocReadBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
);
static
void
uvOnRe
ad
Cb
(
uv_stream_t
*
cli
,
ssize_t
nread
,
const
uv_buf_t
*
buf
);
static
void
uvOnRe
cv
Cb
(
uv_stream_t
*
cli
,
ssize_t
nread
,
const
uv_buf_t
*
buf
);
static
void
uvOnTimeoutCb
(
uv_timer_t
*
handle
);
static
void
uvOn
Write
Cb
(
uv_write_t
*
req
,
int
status
);
static
void
uvOn
Send
Cb
(
uv_write_t
*
req
,
int
status
);
static
void
uvOnPipeWriteCb
(
uv_write_t
*
req
,
int
status
);
static
void
uvOnAcceptCb
(
uv_stream_t
*
stream
,
int
status
);
static
void
uvOnConnectionCb
(
uv_stream_t
*
q
,
ssize_t
nread
,
const
uv_buf_t
*
buf
);
...
...
@@ -240,7 +240,7 @@ static void uvHandleReq(SSrvConn* pConn) {
// validate msg type
}
void
uvOnRe
ad
Cb
(
uv_stream_t
*
cli
,
ssize_t
nread
,
const
uv_buf_t
*
buf
)
{
void
uvOnRe
cv
Cb
(
uv_stream_t
*
cli
,
ssize_t
nread
,
const
uv_buf_t
*
buf
)
{
// opt
SSrvConn
*
conn
=
cli
->
data
;
SConnBuffer
*
pBuf
=
&
conn
->
readBuf
;
...
...
@@ -282,7 +282,7 @@ void uvOnTimeoutCb(uv_timer_t* handle) {
tError
(
"server conn %p time out"
,
pConn
);
}
void
uvOn
Write
Cb
(
uv_write_t
*
req
,
int
status
)
{
void
uvOn
Send
Cb
(
uv_write_t
*
req
,
int
status
)
{
SSrvConn
*
conn
=
req
->
data
;
transClearBuffer
(
&
conn
->
readBuf
);
if
(
status
==
0
)
{
...
...
@@ -350,7 +350,7 @@ static void uvStartSendRespInternal(SSrvMsg* smsg) {
SSrvConn
*
pConn
=
smsg
->
pConn
;
uv_timer_stop
(
&
pConn
->
pTimer
);
uv_write
(
&
pConn
->
pWriter
,
(
uv_stream_t
*
)
pConn
->
pTcp
,
&
wb
,
1
,
uvOn
Write
Cb
);
uv_write
(
&
pConn
->
pWriter
,
(
uv_stream_t
*
)
pConn
->
pTcp
,
&
wb
,
1
,
uvOn
Send
Cb
);
}
static
void
uvStartSendResp
(
SSrvMsg
*
smsg
)
{
// impl
...
...
@@ -526,7 +526,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
return
;
}
uv_read_start
((
uv_stream_t
*
)(
pConn
->
pTcp
),
uvAllocReadBufferCb
,
uvOnRe
ad
Cb
);
uv_read_start
((
uv_stream_t
*
)(
pConn
->
pTcp
),
uvAllocReadBufferCb
,
uvOnRe
cv
Cb
);
}
else
{
tDebug
(
"failed to create new connection"
);
...
...
@@ -641,7 +641,7 @@ static void uvDestroyConn(uv_handle_t* handle) {
uv_timer_stop
(
&
conn
->
pTimer
);
QUEUE_REMOVE
(
&
conn
->
queue
);
free
(
conn
->
pTcp
);
//
free(conn);
free
(
conn
);
if
(
thrd
->
quit
&&
QUEUE_IS_EMPTY
(
&
thrd
->
conn
))
{
uv_loop_close
(
thrd
->
loop
);
...
...
source/libs/transport/test/CMakeLists.txt
浏览文件 @
d2c59701
...
...
@@ -3,7 +3,6 @@ add_executable(client "")
add_executable
(
server
""
)
add_executable
(
transUT
""
)
add_executable
(
syncClient
""
)
add_executable
(
pushClient
""
)
add_executable
(
pushServer
""
)
target_sources
(
transUT
...
...
@@ -27,10 +26,6 @@ target_sources (syncClient
"syncClient.c"
)
target_sources
(
pushClient
PRIVATE
"pushClient.c"
)
target_sources
(
pushServer
PRIVATE
"pushServer.c"
...
...
@@ -102,19 +97,6 @@ target_link_libraries (syncClient
transport
)
target_include_directories
(
pushClient
PUBLIC
"
${
CMAKE_SOURCE_DIR
}
/include/libs/transport"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/../inc"
)
target_link_libraries
(
pushClient
os
util
common
gtest_main
transport
)
target_include_directories
(
pushServer
PUBLIC
"
${
CMAKE_SOURCE_DIR
}
/include/libs/transport"
...
...
source/libs/transport/test/pushClient.c
已删除
100644 → 0
浏览文件 @
421d087a
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <sys/time.h>
#include <tdatablock.h>
#include "os.h"
#include "rpcLog.h"
#include "taoserror.h"
#include "tglobal.h"
#include "trpc.h"
#include "tutil.h"
typedef
struct
{
int
index
;
SEpSet
epSet
;
int
num
;
int
numOfReqs
;
int
msgSize
;
tsem_t
rspSem
;
tsem_t
*
pOverSem
;
pthread_t
thread
;
void
*
pRpc
;
}
SInfo
;
static
void
processResponse
(
void
*
pParent
,
SRpcMsg
*
pMsg
,
SEpSet
*
pEpSet
)
{
SInfo
*
pInfo
=
(
SInfo
*
)
pMsg
->
ahandle
;
tDebug
(
"thread:%d, response is received, type:%d contLen:%d code:0x%x"
,
pInfo
->
index
,
pMsg
->
msgType
,
pMsg
->
contLen
,
pMsg
->
code
);
if
(
pEpSet
)
pInfo
->
epSet
=
*
pEpSet
;
rpcFreeCont
(
pMsg
->
pCont
);
// tsem_post(&pInfo->rspSem);
tsem_post
(
&
pInfo
->
rspSem
);
}
static
int
tcount
=
0
;
typedef
struct
SPushArg
{
tsem_t
sem
;
}
SPushArg
;
// ping
int
pushCallback
(
void
*
arg
,
SRpcMsg
*
msg
)
{
SPushArg
*
push
=
arg
;
tsem_post
(
&
push
->
sem
);
}
SRpcPush
*
createPushArg
()
{
SRpcPush
*
push
=
calloc
(
1
,
sizeof
(
SRpcPush
));
push
->
arg
=
calloc
(
1
,
sizeof
(
SPushArg
));
tsem_init
(
&
(((
SPushArg
*
)
push
->
arg
)
->
sem
),
0
,
0
);
push
->
callback
=
pushCallback
;
return
push
;
}
static
void
*
sendRequest
(
void
*
param
)
{
SInfo
*
pInfo
=
(
SInfo
*
)
param
;
SRpcMsg
rpcMsg
=
{
0
};
tDebug
(
"thread:%d, start to send request"
,
pInfo
->
index
);
tDebug
(
"thread:%d, reqs: %d"
,
pInfo
->
index
,
pInfo
->
numOfReqs
);
int
u100
=
0
;
int
u500
=
0
;
int
u1000
=
0
;
int
u10000
=
0
;
while
(
pInfo
->
numOfReqs
==
0
||
pInfo
->
num
<
pInfo
->
numOfReqs
)
{
SRpcPush
*
push
=
createPushArg
();
pInfo
->
num
++
;
rpcMsg
.
pCont
=
rpcMallocCont
(
pInfo
->
msgSize
);
rpcMsg
.
contLen
=
pInfo
->
msgSize
;
rpcMsg
.
ahandle
=
pInfo
;
rpcMsg
.
msgType
=
1
;
rpcMsg
.
push
=
push
;
// tDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
int64_t
start
=
taosGetTimestampUs
();
rpcSendRequest
(
pInfo
->
pRpc
,
&
pInfo
->
epSet
,
&
rpcMsg
,
NULL
);
if
(
pInfo
->
num
%
20000
==
0
)
tInfo
(
"thread:%d, %d requests have been sent"
,
pInfo
->
index
,
pInfo
->
num
);
tsem_wait
(
&
pInfo
->
rspSem
);
// ping->pong
// tsem_wait(&pInfo->rspSem);
SPushArg
*
arg
=
push
->
arg
;
/// e
tsem_wait
(
&
arg
->
sem
);
// push callback
// query_fetch(client->h)
int64_t
end
=
taosGetTimestampUs
()
-
start
;
if
(
end
<=
100
)
{
u100
++
;
}
else
if
(
end
>
100
&&
end
<=
500
)
{
u500
++
;
}
else
if
(
end
>
500
&&
end
<
1000
)
{
u1000
++
;
}
else
{
u10000
++
;
}
tDebug
(
"recv response succefully"
);
// taosSsleep(100);
}
tError
(
"send and recv sum: %d, %d, %d, %d"
,
u100
,
u500
,
u1000
,
u10000
);
tDebug
(
"thread:%d, it is over"
,
pInfo
->
index
);
tcount
++
;
return
NULL
;
}
int
main
(
int
argc
,
char
*
argv
[])
{
SRpcInit
rpcInit
;
SEpSet
epSet
;
int
msgSize
=
128
;
int
numOfReqs
=
0
;
int
appThreads
=
1
;
char
serverIp
[
40
]
=
"127.0.0.1"
;
char
secret
[
20
]
=
"mypassword"
;
struct
timeval
systemTime
;
int64_t
startTime
,
endTime
;
pthread_attr_t
thattr
;
// server info
epSet
.
inUse
=
0
;
addEpIntoEpSet
(
&
epSet
,
serverIp
,
7000
);
addEpIntoEpSet
(
&
epSet
,
"192.168.0.1"
,
7000
);
// client info
memset
(
&
rpcInit
,
0
,
sizeof
(
rpcInit
));
rpcInit
.
localPort
=
0
;
rpcInit
.
label
=
"APP"
;
rpcInit
.
numOfThreads
=
1
;
rpcInit
.
cfp
=
processResponse
;
rpcInit
.
sessions
=
100
;
rpcInit
.
idleTime
=
100
;
rpcInit
.
user
=
"michael"
;
rpcInit
.
secret
=
secret
;
rpcInit
.
ckey
=
"key"
;
rpcInit
.
spi
=
1
;
rpcInit
.
connType
=
TAOS_CONN_CLIENT
;
for
(
int
i
=
1
;
i
<
argc
;
++
i
)
{
if
(
strcmp
(
argv
[
i
],
"-p"
)
==
0
&&
i
<
argc
-
1
)
{
epSet
.
eps
[
0
].
port
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-i"
)
==
0
&&
i
<
argc
-
1
)
{
tstrncpy
(
epSet
.
eps
[
0
].
fqdn
,
argv
[
++
i
],
sizeof
(
epSet
.
eps
[
0
].
fqdn
));
}
else
if
(
strcmp
(
argv
[
i
],
"-t"
)
==
0
&&
i
<
argc
-
1
)
{
rpcInit
.
numOfThreads
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-m"
)
==
0
&&
i
<
argc
-
1
)
{
msgSize
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-s"
)
==
0
&&
i
<
argc
-
1
)
{
rpcInit
.
sessions
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-n"
)
==
0
&&
i
<
argc
-
1
)
{
numOfReqs
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-a"
)
==
0
&&
i
<
argc
-
1
)
{
appThreads
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-o"
)
==
0
&&
i
<
argc
-
1
)
{
tsCompressMsgSize
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-u"
)
==
0
&&
i
<
argc
-
1
)
{
rpcInit
.
user
=
argv
[
++
i
];
}
else
if
(
strcmp
(
argv
[
i
],
"-k"
)
==
0
&&
i
<
argc
-
1
)
{
rpcInit
.
secret
=
argv
[
++
i
];
}
else
if
(
strcmp
(
argv
[
i
],
"-spi"
)
==
0
&&
i
<
argc
-
1
)
{
rpcInit
.
spi
=
atoi
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-d"
)
==
0
&&
i
<
argc
-
1
)
{
rpcDebugFlag
=
atoi
(
argv
[
++
i
]);
}
else
{
printf
(
"
\n
usage: %s [options]
\n
"
,
argv
[
0
]);
printf
(
" [-i ip]: first server IP address, default is:%s
\n
"
,
serverIp
);
printf
(
" [-p port]: server port number, default is:%d
\n
"
,
epSet
.
eps
[
0
].
port
);
printf
(
" [-t threads]: number of rpc threads, default is:%d
\n
"
,
rpcInit
.
numOfThreads
);
printf
(
" [-s sessions]: number of rpc sessions, default is:%d
\n
"
,
rpcInit
.
sessions
);
printf
(
" [-m msgSize]: message body size, default is:%d
\n
"
,
msgSize
);
printf
(
" [-a threads]: number of app threads, default is:%d
\n
"
,
appThreads
);
printf
(
" [-n requests]: number of requests per thread, default is:%d
\n
"
,
numOfReqs
);
printf
(
" [-o compSize]: compression message size, default is:%d
\n
"
,
tsCompressMsgSize
);
printf
(
" [-u user]: user name for the connection, default is:%s
\n
"
,
rpcInit
.
user
);
printf
(
" [-k secret]: password for the connection, default is:%s
\n
"
,
rpcInit
.
secret
);
printf
(
" [-spi SPI]: security parameter index, default is:%d
\n
"
,
rpcInit
.
spi
);
printf
(
" [-d debugFlag]: debug flag, default:%d
\n
"
,
rpcDebugFlag
);
printf
(
" [-h help]: print out this help
\n\n
"
);
exit
(
0
);
}
}
taosInitLog
(
"client.log"
,
10
);
void
*
pRpc
=
rpcOpen
(
&
rpcInit
);
if
(
pRpc
==
NULL
)
{
tError
(
"failed to initialize RPC"
);
return
-
1
;
}
tInfo
(
"client is initialized"
);
tInfo
(
"threads:%d msgSize:%d requests:%d"
,
appThreads
,
msgSize
,
numOfReqs
);
taosGetTimeOfDay
(
&
systemTime
);
startTime
=
systemTime
.
tv_sec
*
1000000
+
systemTime
.
tv_usec
;
SInfo
*
pInfo
=
(
SInfo
*
)
calloc
(
1
,
sizeof
(
SInfo
)
*
appThreads
);
pthread_attr_init
(
&
thattr
);
pthread_attr_setdetachstate
(
&
thattr
,
PTHREAD_CREATE_JOINABLE
);
for
(
int
i
=
0
;
i
<
appThreads
;
++
i
)
{
pInfo
->
index
=
i
;
pInfo
->
epSet
=
epSet
;
pInfo
->
numOfReqs
=
numOfReqs
;
pInfo
->
msgSize
=
msgSize
;
tsem_init
(
&
pInfo
->
rspSem
,
0
,
0
);
pInfo
->
pRpc
=
pRpc
;
pthread_create
(
&
pInfo
->
thread
,
&
thattr
,
sendRequest
,
pInfo
);
pInfo
++
;
}
do
{
taosUsleep
(
1
);
}
while
(
tcount
<
appThreads
);
taosGetTimeOfDay
(
&
systemTime
);
endTime
=
systemTime
.
tv_sec
*
1000000
+
systemTime
.
tv_usec
;
float
usedTime
=
(
endTime
-
startTime
)
/
1000
.
0
f
;
// mseconds
tInfo
(
"it takes %.3f mseconds to send %d requests to server"
,
usedTime
,
numOfReqs
*
appThreads
);
tInfo
(
"Performance: %.3f requests per second, msgSize:%d bytes"
,
1000
.
0
*
numOfReqs
*
appThreads
/
usedTime
,
msgSize
);
int
ch
=
getchar
();
UNUSED
(
ch
);
taosCloseLog
();
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录