Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
57b4e7ce
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
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,发现更多精彩内容 >>
提交
57b4e7ce
编写于
3月 09, 2022
作者:
M
Minghao Li
浏览文件
操作
浏览文件
下载
差异文件
Merge branch '3.0' of
https://github.com/taosdata/TDengine
into feature/3.0_mhli
上级
2fcc970d
290c04d7
变更
8
展开全部
隐藏空白更改
内联
并排
Showing
8 changed file
with
642 addition
and
490 deletion
+642
-490
include/os/osSocket.h
include/os/osSocket.h
+46
-39
include/os/osSysinfo.h
include/os/osSysinfo.h
+1
-0
source/dnode/mgmt/daemon/src/dmnMain.c
source/dnode/mgmt/daemon/src/dmnMain.c
+5
-0
source/libs/transport/src/rpcTcp.c
source/libs/transport/src/rpcTcp.c
+57
-63
source/libs/transport/src/rpcUdp.c
source/libs/transport/src/rpcUdp.c
+22
-23
source/libs/transport/src/thttp.c
source/libs/transport/src/thttp.c
+8
-9
source/os/src/osSocket.c
source/os/src/osSocket.c
+494
-356
source/os/src/osSysinfo.c
source/os/src/osSysinfo.c
+9
-0
未找到文件。
include/os/osSocket.h
浏览文件 @
57b4e7ce
...
...
@@ -21,7 +21,10 @@
#define socket SOCKET_FUNC_TAOS_FORBID
#define bind BIND_FUNC_TAOS_FORBID
#define listen LISTEN_FUNC_TAOS_FORBID
// #define accept ACCEPT_FUNC_TAOS_FORBID
#define accept ACCEPT_FUNC_TAOS_FORBID
#define epoll_create EPOLL_CREATE_FUNC_TAOS_FORBID
#define epoll_ctl EPOLL_CTL_FUNC_TAOS_FORBID
#define epoll_wait EPOLL_WAIT_FUNC_TAOS_FORBID
#endif
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
...
...
@@ -38,31 +41,6 @@
extern
"C"
{
#endif
#define TAOS_EPOLL_WAIT_TIME 500
typedef
int32_t
SOCKET
;
typedef
SOCKET
EpollFd
;
#define EpollClose(pollFd) taosCloseSocket(pollFd)
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
typedef
SOCKET
SocketFd
;
#else
typedef
int32_t
SocketFd
;
#endif
int32_t
taosSendto
(
SocketFd
fd
,
void
*
msg
,
int
len
,
unsigned
int
flags
,
const
struct
sockaddr
*
to
,
int
tolen
);
int32_t
taosWriteSocket
(
SocketFd
fd
,
void
*
msg
,
int
len
);
int32_t
taosReadSocket
(
SocketFd
fd
,
void
*
msg
,
int
len
);
int32_t
taosCloseSocketNoCheck
(
SocketFd
fd
);
int32_t
taosCloseSocket
(
SocketFd
fd
);
void
taosShutDownSocketRD
(
SOCKET
fd
);
void
taosShutDownSocketWR
(
SOCKET
fd
);
int32_t
taosSetNonblocking
(
SOCKET
sock
,
int32_t
on
);
int32_t
taosSetSockOpt
(
SOCKET
socketfd
,
int32_t
level
,
int32_t
optname
,
void
*
optval
,
int32_t
optlen
);
int32_t
taosGetSockOpt
(
SOCKET
socketfd
,
int32_t
level
,
int32_t
optname
,
void
*
optval
,
int32_t
*
optlen
);
uint32_t
taosInetAddr
(
const
char
*
ipAddr
);
const
char
*
taosInetNtoa
(
struct
in_addr
ipInt
);
#if (defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32))
#define htobe64 htonll
#if defined(_TD_GO_DLL_)
...
...
@@ -74,25 +52,54 @@ const char *taosInetNtoa(struct in_addr ipInt);
#define htobe64 htonll
#endif
int32_t
taosReadn
(
SOCKET
sock
,
char
*
buffer
,
int32_t
len
);
int32_t
taosWriteMsg
(
SOCKET
fd
,
void
*
ptr
,
int32_t
nbytes
);
int32_t
taosReadMsg
(
SOCKET
fd
,
void
*
ptr
,
int32_t
nbytes
);
int32_t
taosNonblockwrite
(
SOCKET
fd
,
char
*
ptr
,
int32_t
nbytes
);
int64_t
taosCopyFds
(
SOCKET
sfd
,
int32_t
dfd
,
int64_t
len
);
int32_t
taosSetNonblocking
(
SOCKET
sock
,
int32_t
on
);
#define TAOS_EPOLL_WAIT_TIME 500
typedef
struct
TdSocketServer
*
TdSocketServerPtr
;
typedef
struct
TdSocket
*
TdSocketPtr
;
typedef
struct
TdEpoll
*
TdEpollPtr
;
int32_t
taosSendto
(
TdSocketPtr
pSocket
,
void
*
msg
,
int
len
,
unsigned
int
flags
,
const
struct
sockaddr
*
to
,
int
tolen
);
int32_t
taosWriteSocket
(
TdSocketPtr
pSocket
,
void
*
msg
,
int
len
);
int32_t
taosReadSocket
(
TdSocketPtr
pSocket
,
void
*
msg
,
int
len
);
int32_t
taosReadFromSocket
(
TdSocketPtr
pSocket
,
void
*
buf
,
int32_t
len
,
int32_t
flags
,
struct
sockaddr
*
destAddr
,
socklen_t
*
addrLen
);
int32_t
taosCloseSocket
(
TdSocketPtr
*
ppSocket
);
int32_t
taosCloseSocketServer
(
TdSocketServerPtr
*
ppSocketServer
);
int32_t
taosShutDownSocketRD
(
TdSocketPtr
pSocket
);
int32_t
taosShutDownSocketServerRD
(
TdSocketServerPtr
pSocketServer
);
int32_t
taosShutDownSocketWR
(
TdSocketPtr
pSocket
);
int32_t
taosShutDownSocketServerWR
(
TdSocketServerPtr
pSocketServer
);
int32_t
taosShutDownSocketRDWR
(
TdSocketPtr
pSocket
);
int32_t
taosShutDownSocketServerRDWR
(
TdSocketServerPtr
pSocketServer
);
int32_t
taosSetNonblocking
(
TdSocketPtr
pSocket
,
int32_t
on
);
int32_t
taosSetSockOpt
(
TdSocketPtr
pSocket
,
int32_t
level
,
int32_t
optname
,
void
*
optval
,
int32_t
optlen
);
int32_t
taosGetSockOpt
(
TdSocketPtr
pSocket
,
int32_t
level
,
int32_t
optname
,
void
*
optval
,
int32_t
*
optlen
);
int32_t
taosWriteMsg
(
TdSocketPtr
pSocket
,
void
*
ptr
,
int32_t
nbytes
);
int32_t
taosReadMsg
(
TdSocketPtr
pSocket
,
void
*
ptr
,
int32_t
nbytes
);
int32_t
taosNonblockwrite
(
TdSocketPtr
pSocket
,
char
*
ptr
,
int32_t
nbytes
);
int64_t
taosCopyFds
(
TdSocketPtr
pSrcSocket
,
TdSocketPtr
pDestSocket
,
int64_t
len
);
SOCKET
taosOpenUdpSocket
(
uint32_t
localIp
,
uint16_t
localPort
);
SOCKET
taosOpenTcpClientSocket
(
uint32_t
ip
,
uint16_t
port
,
uint32_t
localIp
);
SOCKET
taosOpenTcpServerSocket
(
uint32_t
ip
,
uint16_t
port
);
int32_t
taosKeepTcpAlive
(
SOCKET
sockFd
);
TdSocketPtr
taosOpenUdpSocket
(
uint32_t
localIp
,
uint16_t
localPort
);
TdSocketPtr
taosOpenTcpClientSocket
(
uint32_t
ip
,
uint16_t
port
,
uint32_t
localIp
);
TdSocketServerPtr
taosOpenTcpServerSocket
(
uint32_t
ip
,
uint16_t
port
);
int32_t
taosKeepTcpAlive
(
TdSocketPtr
pSocket
);
TdSocketPtr
taosAcceptTcpConnectSocket
(
TdSocketServerPtr
pServerSocket
,
struct
sockaddr
*
destAddr
,
socklen_t
*
addrLen
);
void
taosBlockSIGPIPE
();
int32_t
taosGetSocketName
(
TdSocketPtr
pSocket
,
struct
sockaddr
*
destAddr
,
socklen_t
*
addrLen
);
void
taosBlockSIGPIPE
();
uint32_t
taosGetIpv4FromFqdn
(
const
char
*
);
int32_t
taosGetFqdn
(
char
*
);
void
tinet_ntoa
(
char
*
ipstr
,
uint32_t
ip
);
uint32_t
ip2uint
(
const
char
*
const
ip_addr
);
void
taosIgnSIGPIPE
();
void
taosSetMaskSIGPIPE
();
void
taosIgnSIGPIPE
();
void
taosSetMaskSIGPIPE
();
uint32_t
taosInetAddr
(
const
char
*
ipAddr
);
const
char
*
taosInetNtoa
(
struct
in_addr
ipInt
);
TdEpollPtr
taosCreateEpoll
(
int32_t
size
);
int32_t
taosCtlEpoll
(
TdEpollPtr
pEpoll
,
int32_t
epollOperate
,
TdSocketPtr
pSocket
,
struct
epoll_event
*
event
);
int32_t
taosWaitEpoll
(
TdEpollPtr
pEpoll
,
struct
epoll_event
*
event
,
int32_t
maxEvents
,
int32_t
timeout
);
int32_t
taosCloseEpoll
(
TdEpollPtr
*
ppEpoll
);
#ifdef __cplusplus
}
...
...
include/os/osSysinfo.h
浏览文件 @
57b4e7ce
...
...
@@ -33,6 +33,7 @@ typedef struct {
SDiskSize
size
;
}
SDiskSpace
;
bool
taosCheckSystemIsSmallEnd
();
void
taosGetSystemInfo
();
int32_t
taosGetEmail
(
char
*
email
,
int32_t
maxLen
);
int32_t
taosGetOsReleaseName
(
char
*
releaseName
,
int32_t
maxLen
);
...
...
source/dnode/mgmt/daemon/src/dmnMain.c
浏览文件 @
57b4e7ce
...
...
@@ -97,6 +97,11 @@ int32_t dmnRunDnode() {
}
int
main
(
int
argc
,
char
const
*
argv
[])
{
if
(
!
taosCheckSystemIsSmallEnd
())
{
uError
(
"TDengine does not run on non-small-end machines."
);
return
-
1
;
}
if
(
dmnParseOption
(
argc
,
argv
)
!=
0
)
{
return
-
1
;
}
...
...
source/libs/transport/src/rpcTcp.c
浏览文件 @
57b4e7ce
...
...
@@ -24,7 +24,7 @@
#ifndef USE_UV
typedef
struct
SFdObj
{
void
*
signature
;
SOCKET
fd
;
// TCP socket FD
TdSocketPtr
pSocket
;
// TCP socket FD
void
*
thandle
;
// handle from upper layer, like TAOS
uint32_t
ip
;
uint16_t
port
;
...
...
@@ -40,7 +40,7 @@ typedef struct SThreadObj {
pthread_mutex_t
mutex
;
uint32_t
ip
;
bool
stop
;
EpollFd
pollFd
;
TdEpollPtr
pEpoll
;
int
numOfFds
;
int
threadId
;
char
label
[
TSDB_LABEL_LEN
];
...
...
@@ -56,20 +56,20 @@ typedef struct {
}
SClientObj
;
typedef
struct
{
SOCKET
fd
;
uint32_t
ip
;
uint16_t
port
;
int8_t
stop
;
int8_t
reserve
;
char
label
[
TSDB_LABEL_LEN
];
int
numOfThreads
;
void
*
shandle
;
SThreadObj
**
pThreadObj
;
pthread_t
thread
;
TdSocketServerPtr
pSocketServer
;
uint32_t
ip
;
uint16_t
port
;
int8_t
stop
;
int8_t
reserve
;
char
label
[
TSDB_LABEL_LEN
];
int
numOfThreads
;
void
*
shandle
;
SThreadObj
**
pThreadObj
;
pthread_t
thread
;
}
SServerObj
;
static
void
*
taosProcessTcpData
(
void
*
param
);
static
SFdObj
*
taosMallocFdObj
(
SThreadObj
*
pThreadObj
,
SOCKET
fd
);
static
SFdObj
*
taosMallocFdObj
(
SThreadObj
*
pThreadObj
,
TdSocketPtr
pSocket
);
static
void
taosFreeFdObj
(
SFdObj
*
pFdObj
);
static
void
taosReportBrokenLink
(
SFdObj
*
pFdObj
);
static
void
*
taosAcceptTcpConnection
(
void
*
arg
);
...
...
@@ -85,7 +85,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
return
NULL
;
}
pServerObj
->
fd
=
-
1
;
pServerObj
->
pSocketServer
=
NULL
;
taosResetPthread
(
&
pServerObj
->
thread
);
pServerObj
->
ip
=
ip
;
pServerObj
->
port
=
port
;
...
...
@@ -118,7 +118,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
}
pServerObj
->
pThreadObj
[
i
]
=
pThreadObj
;
pThreadObj
->
p
ollFd
=
-
1
;
pThreadObj
->
p
Epoll
=
NULL
;
taosResetPthread
(
&
pThreadObj
->
thread
);
pThreadObj
->
processData
=
fp
;
tstrncpy
(
pThreadObj
->
label
,
label
,
sizeof
(
pThreadObj
->
label
));
...
...
@@ -135,8 +135,8 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
break
;
}
pThreadObj
->
p
ollFd
=
(
EpollFd
)
epoll_create
(
10
);
// size does not matter
if
(
pThreadObj
->
p
ollFd
<
0
)
{
pThreadObj
->
p
Epoll
=
taosCreateEpoll
(
10
);
// size does not matter
if
(
pThreadObj
->
p
Epoll
==
NULL
)
{
tError
(
"%s failed to create TCP epoll"
,
label
);
code
=
-
1
;
break
;
...
...
@@ -151,8 +151,8 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
pThreadObj
->
threadId
=
i
;
}
pServerObj
->
fd
=
taosOpenTcpServerSocket
(
pServerObj
->
ip
,
pServerObj
->
port
);
if
(
pServerObj
->
fd
<
0
)
code
=
-
1
;
pServerObj
->
pSocketServer
=
taosOpenTcpServerSocket
(
pServerObj
->
ip
,
pServerObj
->
port
);
if
(
pServerObj
->
pSocketServer
==
NULL
)
code
=
-
1
;
if
(
code
==
0
)
{
code
=
pthread_create
(
&
pServerObj
->
thread
,
&
thattr
,
taosAcceptTcpConnection
,
(
void
*
)
pServerObj
);
...
...
@@ -196,8 +196,8 @@ void taosStopTcpServer(void *handle) {
if
(
pServerObj
==
NULL
)
return
;
pServerObj
->
stop
=
1
;
if
(
pServerObj
->
fd
>=
0
)
{
taosShutDownSocket
RD
(
pServerObj
->
fd
);
if
(
pServerObj
->
pSocketServer
!=
NULL
)
{
taosShutDownSocket
ServerRD
(
pServerObj
->
pSocketServer
);
}
if
(
taosCheckPthreadValid
(
pServerObj
->
thread
))
{
if
(
taosComparePthread
(
pServerObj
->
thread
,
pthread_self
()))
{
...
...
@@ -227,7 +227,7 @@ void taosCleanUpTcpServer(void *handle) {
}
static
void
*
taosAcceptTcpConnection
(
void
*
arg
)
{
SOCKET
connFd
=
-
1
;
TdSocketPtr
pSocket
=
NULL
;
struct
sockaddr_in
caddr
;
int
threadId
=
0
;
SThreadObj
*
pThreadObj
;
...
...
@@ -239,13 +239,13 @@ static void *taosAcceptTcpConnection(void *arg) {
while
(
1
)
{
socklen_t
addrlen
=
sizeof
(
caddr
);
connFd
=
accept
(
pServerObj
->
fd
,
(
struct
sockaddr
*
)
&
caddr
,
&
addrlen
);
pSocket
=
taosAcceptTcpConnectSocket
(
pServerObj
->
pSocketServer
,
(
struct
sockaddr
*
)
&
caddr
,
&
addrlen
);
if
(
pServerObj
->
stop
)
{
tDebug
(
"%s TCP server stop accepting new connections"
,
pServerObj
->
label
);
break
;
}
if
(
connFd
==
-
1
)
{
if
(
pSocket
==
NULL
)
{
if
(
errno
==
EINVAL
)
{
tDebug
(
"%s TCP server stop accepting new connections, exiting"
,
pServerObj
->
label
);
break
;
...
...
@@ -255,11 +255,11 @@ static void *taosAcceptTcpConnection(void *arg) {
continue
;
}
taosKeepTcpAlive
(
connFd
);
taosKeepTcpAlive
(
pSocket
);
struct
timeval
to
=
{
5
,
0
};
int32_t
ret
=
taosSetSockOpt
(
connFd
,
SOL_SOCKET
,
SO_RCVTIMEO
,
&
to
,
sizeof
(
to
));
int32_t
ret
=
taosSetSockOpt
(
pSocket
,
SOL_SOCKET
,
SO_RCVTIMEO
,
&
to
,
sizeof
(
to
));
if
(
ret
!=
0
)
{
taosCloseSocket
(
connFd
);
taosCloseSocket
(
&
pSocket
);
tError
(
"%s failed to set recv timeout fd(%s)for connection from:%s:%hu"
,
pServerObj
->
label
,
strerror
(
errno
),
taosInetNtoa
(
caddr
.
sin_addr
),
htons
(
caddr
.
sin_port
));
continue
;
...
...
@@ -268,14 +268,14 @@ static void *taosAcceptTcpConnection(void *arg) {
// pick up the thread to handle this connection
pThreadObj
=
pServerObj
->
pThreadObj
[
threadId
];
SFdObj
*
pFdObj
=
taosMallocFdObj
(
pThreadObj
,
connFd
);
SFdObj
*
pFdObj
=
taosMallocFdObj
(
pThreadObj
,
pSocket
);
if
(
pFdObj
)
{
pFdObj
->
ip
=
caddr
.
sin_addr
.
s_addr
;
pFdObj
->
port
=
htons
(
caddr
.
sin_port
);
tDebug
(
"%s new TCP connection from %s:%hu,
fd:%d
FD:%p numOfFds:%d"
,
pServerObj
->
label
,
taosInetNtoa
(
caddr
.
sin_addr
),
pFdObj
->
port
,
connFd
,
pFdObj
,
pThreadObj
->
numOfFds
);
tDebug
(
"%s new TCP connection from %s:%hu, FD:%p numOfFds:%d"
,
pServerObj
->
label
,
taosInetNtoa
(
caddr
.
sin_addr
),
pFdObj
->
port
,
pFdObj
,
pThreadObj
->
numOfFds
);
}
else
{
taosCloseSocket
(
connFd
);
taosCloseSocket
(
&
pSocket
);
tError
(
"%s failed to malloc FdObj(%s) for connection from:%s:%hu"
,
pServerObj
->
label
,
strerror
(
errno
),
taosInetNtoa
(
caddr
.
sin_addr
),
htons
(
caddr
.
sin_port
));
}
...
...
@@ -285,7 +285,7 @@ static void *taosAcceptTcpConnection(void *arg) {
threadId
=
threadId
%
pServerObj
->
numOfThreads
;
}
taosCloseSocket
(
pServerObj
->
fd
);
taosCloseSocket
Server
(
&
pServerObj
->
pSocketServer
);
return
NULL
;
}
...
...
@@ -339,8 +339,8 @@ void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int numOfThread
break
;
}
pThreadObj
->
p
ollFd
=
(
int64_t
)
epoll_create
(
10
);
// size does not matter
if
(
pThreadObj
->
p
ollFd
<
0
)
{
pThreadObj
->
p
Epoll
=
taosCreateEpoll
(
10
);
// size does not matter
if
(
pThreadObj
->
p
Epoll
==
NULL
)
{
tError
(
"%s failed to create TCP epoll"
,
label
);
code
=
-
1
;
break
;
...
...
@@ -388,21 +388,17 @@ void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uin
atomic_store_32
(
&
pClientObj
->
index
,
index
+
1
);
SThreadObj
*
pThreadObj
=
pClientObj
->
pThreadObj
[
index
];
SOCKET
fd
=
taosOpenTcpClientSocket
(
ip
,
port
,
pThreadObj
->
ip
);
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
if
(
fd
==
(
SOCKET
)
-
1
)
return
NULL
;
#else
if
(
fd
<=
0
)
return
NULL
;
#endif
TdSocketPtr
pSocket
=
taosOpenTcpClientSocket
(
ip
,
port
,
pThreadObj
->
ip
);
if
(
pSocket
==
NULL
)
return
NULL
;
struct
sockaddr_in
sin
;
uint16_t
localPort
=
0
;
unsigned
int
addrlen
=
sizeof
(
sin
);
if
(
getsockname
(
fd
,
(
struct
sockaddr
*
)
&
sin
,
&
addrlen
)
==
0
&&
sin
.
sin_family
==
AF_INET
&&
addrlen
==
sizeof
(
sin
))
{
if
(
taosGetSocketName
(
pSocket
,
(
struct
sockaddr
*
)
&
sin
,
&
addrlen
)
==
0
&&
sin
.
sin_family
==
AF_INET
&&
addrlen
==
sizeof
(
sin
))
{
localPort
=
(
uint16_t
)
ntohs
(
sin
.
sin_port
);
}
SFdObj
*
pFdObj
=
taosMallocFdObj
(
pThreadObj
,
fd
);
SFdObj
*
pFdObj
=
taosMallocFdObj
(
pThreadObj
,
pSocket
);
if
(
pFdObj
)
{
pFdObj
->
thandle
=
thandle
;
...
...
@@ -415,7 +411,7 @@ void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uin
ipport
,
localPort
,
pFdObj
,
pThreadObj
->
numOfFds
);
}
else
{
tError
(
"%s failed to malloc client FdObj(%s)"
,
pThreadObj
->
label
,
strerror
(
errno
));
taosCloseSocket
(
fd
);
taosCloseSocket
(
&
pSocket
);
}
return
pFdObj
;
...
...
@@ -430,7 +426,7 @@ void taosCloseTcpConnection(void *chandle) {
// pFdObj->thandle = NULL;
pFdObj
->
closedByApp
=
1
;
taosShutDownSocketWR
(
pFdObj
->
fd
);
taosShutDownSocketWR
(
pFdObj
->
pSocket
);
}
int
taosSendTcpData
(
uint32_t
ip
,
uint16_t
port
,
void
*
data
,
int
len
,
void
*
chandle
)
{
...
...
@@ -438,8 +434,8 @@ int taosSendTcpData(uint32_t ip, uint16_t port, void *data, int len, void *chand
if
(
pFdObj
==
NULL
||
pFdObj
->
signature
!=
pFdObj
)
return
-
1
;
SThreadObj
*
pThreadObj
=
pFdObj
->
pThreadObj
;
int
ret
=
taosWriteMsg
(
pFdObj
->
fd
,
data
,
len
);
tTrace
(
"%s %p TCP data is sent, FD:%p
fd:%d bytes:%d"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
pFdObj
,
pFdObj
->
fd
,
ret
);
int
ret
=
taosWriteMsg
(
pFdObj
->
pSocket
,
data
,
len
);
tTrace
(
"%s %p TCP data is sent, FD:%p
bytes:%d"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
pFdObj
,
ret
);
return
ret
;
}
...
...
@@ -449,7 +445,7 @@ static void taosReportBrokenLink(SFdObj *pFdObj) {
// notify the upper layer, so it will clean the associated context
if
(
pFdObj
->
closedByApp
==
0
)
{
taosShutDownSocketWR
(
pFdObj
->
fd
);
taosShutDownSocketWR
(
pFdObj
->
pSocket
);
SRecvInfo
recvInfo
;
recvInfo
.
msg
=
NULL
;
...
...
@@ -473,7 +469,7 @@ static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
SThreadObj
*
pThreadObj
=
pFdObj
->
pThreadObj
;
headLen
=
taosReadMsg
(
pFdObj
->
fd
,
&
rpcHead
,
sizeof
(
SRpcHead
));
headLen
=
taosReadMsg
(
pFdObj
->
pSocket
,
&
rpcHead
,
sizeof
(
SRpcHead
));
if
(
headLen
!=
sizeof
(
SRpcHead
))
{
tDebug
(
"%s %p read error, FD:%p headLen:%d"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
pFdObj
,
headLen
);
return
-
1
;
...
...
@@ -486,13 +482,12 @@ static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
tError
(
"%s %p TCP malloc(size:%d) fail"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
msgLen
);
return
-
1
;
}
else
{
tTrace
(
"%s %p read data, FD:%p fd:%d TCP malloc mem:%p"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
pFdObj
,
pFdObj
->
fd
,
buffer
);
tTrace
(
"%s %p read data, FD:%p TCP malloc mem:%p"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
pFdObj
,
buffer
);
}
msg
=
buffer
+
tsRpcOverhead
;
leftLen
=
msgLen
-
headLen
;
retLen
=
taosReadMsg
(
pFdObj
->
fd
,
msg
+
headLen
,
leftLen
);
retLen
=
taosReadMsg
(
pFdObj
->
pSocket
,
msg
+
headLen
,
leftLen
);
if
(
leftLen
!=
retLen
)
{
tError
(
"%s %p read error, leftLen:%d retLen:%d FD:%p"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
leftLen
,
retLen
,
pFdObj
);
...
...
@@ -532,7 +527,7 @@ static void *taosProcessTcpData(void *param) {
setThreadName
(
name
);
while
(
1
)
{
int
fdNum
=
epoll_wait
(
pThreadObj
->
pollFd
,
events
,
maxEvents
,
TAOS_EPOLL_WAIT_TIME
);
int
fdNum
=
taosWaitEpoll
(
pThreadObj
->
pEpoll
,
events
,
maxEvents
,
TAOS_EPOLL_WAIT_TIME
);
if
(
pThreadObj
->
stop
)
{
tDebug
(
"%s TCP thread get stop event, exiting..."
,
pThreadObj
->
label
);
break
;
...
...
@@ -561,7 +556,7 @@ static void *taosProcessTcpData(void *param) {
}
if
(
taosReadTcpData
(
pFdObj
,
&
recvInfo
)
<
0
)
{
shutdown
(
pFdObj
->
fd
,
SHUT_WR
);
taosShutDownSocketWR
(
pFdObj
->
pSocket
);
continue
;
}
...
...
@@ -572,9 +567,9 @@ static void *taosProcessTcpData(void *param) {
if
(
pThreadObj
->
stop
)
break
;
}
if
(
pThreadObj
->
p
ollFd
>=
0
)
{
EpollClose
(
pThreadObj
->
pollFd
);
pThreadObj
->
p
ollFd
=
-
1
;
if
(
pThreadObj
->
p
Epoll
!=
NULL
)
{
taosCloseEpoll
(
&
pThreadObj
->
pEpoll
);
pThreadObj
->
p
Epoll
=
NULL
;
}
while
(
pThreadObj
->
pHead
)
{
...
...
@@ -590,7 +585,7 @@ static void *taosProcessTcpData(void *param) {
return
NULL
;
}
static
SFdObj
*
taosMallocFdObj
(
SThreadObj
*
pThreadObj
,
SOCKET
fd
)
{
static
SFdObj
*
taosMallocFdObj
(
SThreadObj
*
pThreadObj
,
TdSocketPtr
pSocket
)
{
struct
epoll_event
event
;
SFdObj
*
pFdObj
=
(
SFdObj
*
)
calloc
(
sizeof
(
SFdObj
),
1
);
...
...
@@ -599,13 +594,13 @@ static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd) {
}
pFdObj
->
closedByApp
=
0
;
pFdObj
->
fd
=
fd
;
pFdObj
->
pSocket
=
pSocket
;
pFdObj
->
pThreadObj
=
pThreadObj
;
pFdObj
->
signature
=
pFdObj
;
event
.
events
=
EPOLLIN
|
EPOLLRDHUP
;
event
.
data
.
ptr
=
pFdObj
;
if
(
epoll_ctl
(
pThreadObj
->
pollFd
,
EPOLL_CTL_ADD
,
fd
,
&
event
)
<
0
)
{
if
(
taosCtlEpoll
(
pThreadObj
->
pEpoll
,
EPOLL_CTL_ADD
,
pSocket
,
&
event
)
<
0
)
{
tfree
(
pFdObj
);
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
return
NULL
;
...
...
@@ -635,8 +630,8 @@ static void taosFreeFdObj(SFdObj *pFdObj) {
}
pFdObj
->
signature
=
NULL
;
epoll_ctl
(
pThreadObj
->
pollFd
,
EPOLL_CTL_DEL
,
pFdObj
->
fd
,
NULL
);
taosCloseSocket
(
pFdObj
->
fd
);
taosCtlEpoll
(
pThreadObj
->
pEpoll
,
EPOLL_CTL_DEL
,
pFdObj
->
pSocket
,
NULL
);
taosCloseSocket
(
&
pFdObj
->
pSocket
);
pThreadObj
->
numOfFds
--
;
if
(
pThreadObj
->
numOfFds
<
0
)
...
...
@@ -655,8 +650,7 @@ static void taosFreeFdObj(SFdObj *pFdObj) {
pthread_mutex_unlock
(
&
pThreadObj
->
mutex
);
tDebug
(
"%s %p TCP connection is closed, FD:%p fd:%d numOfFds:%d"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
pFdObj
,
pFdObj
->
fd
,
pThreadObj
->
numOfFds
);
tDebug
(
"%s %p TCP connection is closed, FD:%p numOfFds:%d"
,
pThreadObj
->
label
,
pFdObj
->
thandle
,
pFdObj
,
pThreadObj
->
numOfFds
);
tfree
(
pFdObj
);
}
...
...
source/libs/transport/src/rpcUdp.c
浏览文件 @
57b4e7ce
...
...
@@ -30,17 +30,17 @@
#define RPC_MAX_UDP_SIZE 65480
typedef
struct
{
int
index
;
SOCKET
fd
;
uint16_t
port
;
// peer port
uint16_t
localPort
;
// local port
char
label
[
TSDB_LABEL_LEN
];
// copy from udpConnSet;
pthread_t
thread
;
void
*
hash
;
void
*
shandle
;
// handle passed by upper layer during server initialization
void
*
pSet
;
void
*
(
*
processData
)(
SRecvInfo
*
pRecv
);
char
*
buffer
;
// buffer to receive data
int
index
;
TdSocketPtr
pSocket
;
uint16_t
port
;
// peer port
uint16_t
localPort
;
// local port
char
label
[
TSDB_LABEL_LEN
];
// copy from udpConnSet;
pthread_t
thread
;
void
*
hash
;
void
*
shandle
;
// handle passed by upper layer during server initialization
void
*
pSet
;
void
*
(
*
processData
)(
SRecvInfo
*
pRecv
);
char
*
buffer
;
// buffer to receive data
}
SUdpConn
;
typedef
struct
{
...
...
@@ -86,8 +86,8 @@ void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int threads
for
(
i
=
0
;
i
<
threads
;
++
i
)
{
pConn
=
pSet
->
udpConn
+
i
;
ownPort
=
(
port
?
port
+
i
:
0
);
pConn
->
fd
=
taosOpenUdpSocket
(
ip
,
ownPort
);
if
(
pConn
->
fd
<
0
)
{
pConn
->
pSocket
=
taosOpenUdpSocket
(
ip
,
ownPort
);
if
(
pConn
->
pSocket
==
NULL
)
{
tError
(
"%s failed to open UDP socket %x:%hu"
,
label
,
ip
,
port
);
break
;
}
...
...
@@ -100,7 +100,7 @@ void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int threads
struct
sockaddr_in
sin
;
unsigned
int
addrlen
=
sizeof
(
sin
);
if
(
getsockname
(
pConn
->
fd
,
(
struct
sockaddr
*
)
&
sin
,
&
addrlen
)
==
0
&&
sin
.
sin_family
==
AF_INET
&&
if
(
taosGetSocketName
(
pConn
->
pSocket
,
(
struct
sockaddr
*
)
&
sin
,
&
addrlen
)
==
0
&&
sin
.
sin_family
==
AF_INET
&&
addrlen
==
sizeof
(
sin
))
{
pConn
->
localPort
=
(
uint16_t
)
ntohs
(
sin
.
sin_port
);
}
...
...
@@ -138,9 +138,9 @@ void taosStopUdpConnection(void *handle) {
for
(
int
i
=
0
;
i
<
pSet
->
threads
;
++
i
)
{
pConn
=
pSet
->
udpConn
+
i
;
if
(
pConn
->
fd
>=
0
)
shutdown
(
pConn
->
fd
,
SHUT_RDWR
);
if
(
pConn
->
fd
>=
0
)
taosCloseSocket
(
pConn
->
fd
);
pConn
->
fd
=
-
1
;
if
(
pConn
->
pSocket
!=
NULL
)
taosShutDownSocketRDWR
(
pConn
->
pSocket
);
if
(
pConn
->
pSocket
!=
NULL
)
taosCloseSocket
(
&
pConn
->
pSocket
);
pConn
->
pSocket
=
NULL
;
}
for
(
int
i
=
0
;
i
<
pSet
->
threads
;
++
i
)
{
...
...
@@ -163,7 +163,7 @@ void taosCleanUpUdpConnection(void *handle) {
for
(
int
i
=
0
;
i
<
pSet
->
threads
;
++
i
)
{
pConn
=
pSet
->
udpConn
+
i
;
if
(
pConn
->
fd
>=
0
)
taosCloseSocket
(
pConn
->
fd
);
if
(
pConn
->
pSocket
!=
NULL
)
taosCloseSocket
(
&
pConn
->
pSocket
);
}
tDebug
(
"%s UDP is cleaned up"
,
pSet
->
label
);
...
...
@@ -199,13 +199,12 @@ static void *taosRecvUdpData(void *param) {
setThreadName
(
"recvUdpData"
);
while
(
1
)
{
dataLen
=
recvfrom
(
pConn
->
fd
,
pConn
->
buffer
,
RPC_MAX_UDP_SIZE
,
0
,
(
struct
sockaddr
*
)
&
sourceAdd
,
&
addLen
);
dataLen
=
taosReadFromSocket
(
pConn
->
pSocket
,
pConn
->
buffer
,
RPC_MAX_UDP_SIZE
,
0
,
(
struct
sockaddr
*
)
&
sourceAdd
,
&
addLen
);
if
(
dataLen
<=
0
)
{
tDebug
(
"%s UDP socket was closed, exiting(%s), dataLen:%d fd:%d"
,
pConn
->
label
,
strerror
(
errno
),
(
int32_t
)
dataLen
,
pConn
->
fd
);
tDebug
(
"%s UDP socket was closed, exiting(%s), dataLen:%d"
,
pConn
->
label
,
strerror
(
errno
),
(
int32_t
)
dataLen
);
// for windows usage, remote shutdown also returns - 1 in windows client
if
(
pConn
->
fd
==
-
1
)
{
if
(
pConn
->
pSocket
==
NULL
)
{
break
;
}
else
{
continue
;
...
...
@@ -255,7 +254,7 @@ int taosSendUdpData(uint32_t ip, uint16_t port, void *data, int dataLen, void *c
destAdd
.
sin_addr
.
s_addr
=
ip
;
destAdd
.
sin_port
=
htons
(
port
);
int
ret
=
(
int
)
taosSendto
(
pConn
->
fd
,
data
,
(
size_t
)
dataLen
,
0
,
(
struct
sockaddr
*
)
&
destAdd
,
sizeof
(
destAdd
));
int
ret
=
taosSendto
(
pConn
->
pSocket
,
data
,
(
size_t
)
dataLen
,
0
,
(
struct
sockaddr
*
)
&
destAdd
,
sizeof
(
destAdd
));
return
ret
;
}
...
...
source/libs/transport/src/thttp.c
浏览文件 @
57b4e7ce
...
...
@@ -173,7 +173,7 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32
#else
int32_t
taosSendHttpReport
(
const
char
*
server
,
uint16_t
port
,
char
*
pCont
,
int32_t
contLen
,
EHttpCompFlag
flag
)
{
int32_t
code
=
-
1
;
SOCKET
fd
=
0
;
TdSocketPtr
pSocket
=
NULL
;
uint32_t
ip
=
taosGetIpv4FromFqdn
(
server
);
if
(
ip
==
0xffffffff
)
{
...
...
@@ -182,8 +182,8 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32
goto
SEND_OVER
;
}
fd
=
taosOpenTcpClientSocket
(
ip
,
port
,
0
);
if
(
fd
<
0
)
{
pSocket
=
taosOpenTcpClientSocket
(
ip
,
port
,
0
);
if
(
pSocket
==
NULL
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
uError
(
"failed to create http socket to %s:%u since %s"
,
server
,
port
,
terrstr
());
goto
SEND_OVER
;
...
...
@@ -200,21 +200,20 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32
char
header
[
1024
]
=
{
0
};
int32_t
headLen
=
taosBuildHttpHeader
(
server
,
contLen
,
header
,
sizeof
(
header
),
flag
);
if
(
taosWriteSocket
(
fd
,
header
,
headLen
)
<
0
)
{
if
(
taosWriteMsg
(
pSocket
,
header
,
headLen
)
<
0
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
uError
(
"failed to send http header to %s:%u since %s"
,
server
,
port
,
terrstr
());
goto
SEND_OVER
;
}
if
(
taosWrite
Socket
(
fd
,
(
void
*
)
pCont
,
contLen
)
<
0
)
{
if
(
taosWrite
Msg
(
pSocket
,
(
void
*
)
pCont
,
contLen
)
<
0
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
uError
(
"failed to send http content to %s:%u since %s"
,
server
,
port
,
terrstr
());
goto
SEND_OVER
;
}
// read something to avoid nginx error 499
if
(
taos
ReadSocket
(
fd
,
header
,
10
)
<
0
)
{
if
(
taos
WriteMsg
(
pSocket
,
header
,
10
)
<
0
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
uError
(
"failed to receive response from %s:%u since %s"
,
server
,
port
,
terrstr
());
goto
SEND_OVER
;
...
...
@@ -223,8 +222,8 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32
code
=
0
;
SEND_OVER:
if
(
fd
!=
0
)
{
taosCloseSocket
(
fd
);
if
(
pSocket
!=
NULL
)
{
taosCloseSocket
(
&
pSocket
);
}
return
code
;
...
...
source/os/src/osSocket.c
浏览文件 @
57b4e7ce
此差异已折叠。
点击以展开。
source/os/src/osSysinfo.c
浏览文件 @
57b4e7ce
...
...
@@ -16,6 +16,15 @@
#define _DEFAULT_SOURCE
#include "os.h"
bool
taosCheckSystemIsSmallEnd
()
{
union
check
{
int16_t
i
;
char
ch
[
2
];
}
c
;
c
.
i
=
1
;
return
c
.
ch
[
0
]
==
1
;
}
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
/*
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录