提交 7a782907 编写于 作者: S Shengliang Guan

TD-1207 change socket to int32

上级 5d9ebb25
...@@ -30,10 +30,10 @@ static SCheckItem tsCheckItem[TSDB_CHECK_ITEM_MAX] = {{0}}; ...@@ -30,10 +30,10 @@ static SCheckItem tsCheckItem[TSDB_CHECK_ITEM_MAX] = {{0}};
int64_t tsMinFreeMemSizeForStart = 0; int64_t tsMinFreeMemSizeForStart = 0;
static int32_t bindTcpPort(int32_t port) { static int32_t bindTcpPort(int32_t port) {
SOCKET serverSocket; int32_t serverSocket;
struct sockaddr_in server_addr; struct sockaddr_in server_addr;
if ((serverSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) { if ((serverSocket = ( int32_t)socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
dError("socket() fail: %s", strerror(errno)); dError("socket() fail: %s", strerror(errno));
return -1; return -1;
} }
...@@ -60,10 +60,10 @@ static int32_t bindTcpPort(int32_t port) { ...@@ -60,10 +60,10 @@ static int32_t bindTcpPort(int32_t port) {
} }
static int32_t bindUdpPort(int32_t port) { static int32_t bindUdpPort(int32_t port) {
SOCKET serverSocket; int32_t serverSocket;
struct sockaddr_in server_addr; struct sockaddr_in server_addr;
if ((serverSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) { if ((serverSocket = (int32_t)socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) {
dError("socket() fail: %s", strerror(errno)); dError("socket() fail: %s", strerror(errno));
return -1; return -1;
} }
......
...@@ -200,7 +200,7 @@ static void sendTelemetryReport() { ...@@ -200,7 +200,7 @@ static void sendTelemetryReport() {
dTrace("failed to get IP address of " TELEMETRY_SERVER ", reason:%s", strerror(errno)); dTrace("failed to get IP address of " TELEMETRY_SERVER ", reason:%s", strerror(errno));
return; return;
} }
SOCKET fd = taosOpenTcpClientSocket(ip, TELEMETRY_PORT, 0); int32_t fd = taosOpenTcpClientSocket(ip, TELEMETRY_PORT, 0);
if (fd < 0) { if (fd < 0) {
dTrace("failed to create socket for telemetry, reason:%s", strerror(errno)); dTrace("failed to create socket for telemetry, reason:%s", strerror(errno));
return; return;
......
...@@ -39,7 +39,7 @@ int32_t taosRenameFile(char *fullPath, char *suffix, char delimiter, char **dstP ...@@ -39,7 +39,7 @@ int32_t taosRenameFile(char *fullPath, char *suffix, char delimiter, char **dstP
} }
// TAOS_OS_FUNC_FILE_SENDIFLE // TAOS_OS_FUNC_FILE_SENDIFLE
int64_t taosSendFile(SOCKET dfd, int32_t sfd, int64_t *offset, int64_t size); int64_t taosSendFile(int32_t dfd, int32_t sfd, int64_t *offset, int64_t size);
int64_t taosFSendFile(FILE *outfile, FILE *infile, int64_t *offset, int64_t size); int64_t taosFSendFile(FILE *outfile, FILE *infile, int64_t *offset, int64_t size);
#ifdef TAOS_RANDOM_FILE_FAIL #ifdef TAOS_RANDOM_FILE_FAIL
......
...@@ -33,7 +33,6 @@ extern "C" { ...@@ -33,7 +33,6 @@ extern "C" {
x = FD_INITIALIZER; \ x = FD_INITIALIZER; \
} \ } \
} }
typedef int32_t SOCKET;
#endif #endif
#ifndef TAOS_OS_DEF_EPOLL #ifndef TAOS_OS_DEF_EPOLL
...@@ -58,13 +57,13 @@ extern "C" { ...@@ -58,13 +57,13 @@ extern "C" {
#endif #endif
// TAOS_OS_FUNC_SOCKET // TAOS_OS_FUNC_SOCKET
int32_t taosSetNonblocking(SOCKET sock, int32_t on); int32_t taosSetNonblocking(int32_t sock, int32_t on);
void taosIgnSIGPIPE(); void taosIgnSIGPIPE();
void taosBlockSIGPIPE(); void taosBlockSIGPIPE();
void taosSetMaskSIGPIPE(); void taosSetMaskSIGPIPE();
// TAOS_OS_FUNC_SOCKET_SETSOCKETOPT // TAOS_OS_FUNC_SOCKET_SETSOCKETOPT
int32_t taosSetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen); int32_t taosSetSockOpt(int32_t socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen);
// TAOS_OS_FUNC_SOCKET_INET // TAOS_OS_FUNC_SOCKET_INET
uint32_t taosInetAddr(char *ipAddr); uint32_t taosInetAddr(char *ipAddr);
......
...@@ -51,7 +51,7 @@ int64_t taosFSendFile(FILE *out_file, FILE *in_file, int64_t *offset, int64_t co ...@@ -51,7 +51,7 @@ int64_t taosFSendFile(FILE *out_file, FILE *in_file, int64_t *offset, int64_t co
return writeLen; return writeLen;
} }
int64_t taosSendFile(SOCKET dfd, int32_t sfd, int64_t* offset, int64_t count) { int64_t taosSendFile(int32_t dfd, int32_t sfd, int64_t* offset, int64_t count) {
lseek(sfd, (int32_t)(*offset), 0); lseek(sfd, (int32_t)(*offset), 0);
int64_t writeLen = 0; int64_t writeLen = 0;
uint8_t buffer[_SEND_FILE_STEP_] = { 0 }; uint8_t buffer[_SEND_FILE_STEP_] = { 0 };
......
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
int taosSetSockOpt(SOCKET socketfd, int level, int optname, void *optval, int optlen) { int taosSetSockOpt(int32_t socketfd, int level, int optname, void *optval, int optlen) {
if (level == SOL_SOCKET && optname == SO_SNDBUF) { if (level == SOL_SOCKET && optname == SO_SNDBUF) {
return 0; return 0;
} }
......
...@@ -121,7 +121,7 @@ int64_t taosLSeekImp(int32_t fd, int64_t offset, int32_t whence) { ...@@ -121,7 +121,7 @@ int64_t taosLSeekImp(int32_t fd, int64_t offset, int32_t whence) {
#ifndef TAOS_OS_FUNC_FILE_SENDIFLE #ifndef TAOS_OS_FUNC_FILE_SENDIFLE
int64_t taosSendFile(SOCKET dfd, int32_t sfd, int64_t *offset, int64_t size) { int64_t taosSendFile(int32_t dfd, int32_t sfd, int64_t *offset, int64_t size) {
int64_t leftbytes = size; int64_t leftbytes = size;
int64_t sentbytes; int64_t sentbytes;
......
...@@ -19,7 +19,7 @@ ...@@ -19,7 +19,7 @@
#ifndef TAOS_OS_FUNC_SOCKET #ifndef TAOS_OS_FUNC_SOCKET
int32_t taosSetNonblocking(SOCKET sock, int32_t on) { int32_t taosSetNonblocking(int32_t sock, int32_t on) {
int32_t flags = 0; int32_t flags = 0;
if ((flags = fcntl(sock, F_GETFL, 0)) < 0) { if ((flags = fcntl(sock, F_GETFL, 0)) < 0) {
uError("fcntl(F_GETFL) error: %d (%s)\n", errno, strerror(errno)); uError("fcntl(F_GETFL) error: %d (%s)\n", errno, strerror(errno));
...@@ -67,7 +67,7 @@ void taosSetMaskSIGPIPE() { ...@@ -67,7 +67,7 @@ void taosSetMaskSIGPIPE() {
#ifndef TAOS_OS_FUNC_SOCKET_SETSOCKETOPT #ifndef TAOS_OS_FUNC_SOCKET_SETSOCKETOPT
int32_t taosSetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen) { int32_t taosSetSockOpt(int32_t socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen) {
return setsockopt(socketfd, level, optname, optval, (socklen_t)optlen); return setsockopt(socketfd, level, optname, optval, (socklen_t)optlen);
} }
......
...@@ -76,7 +76,7 @@ int64_t taosFSendFile(FILE *out_file, FILE *in_file, int64_t *offset, int64_t co ...@@ -76,7 +76,7 @@ int64_t taosFSendFile(FILE *out_file, FILE *in_file, int64_t *offset, int64_t co
return writeLen; return writeLen;
} }
int64_t taosSendFile(SOCKET dfd, int32_t sfd, int64_t *offset, int64_t count) { int64_t taosSendFile(int32_t dfd, int32_t sfd, int64_t *offset, int64_t count) {
if (offset != NULL) lseek(sfd, (int32_t)(*offset), 0); if (offset != NULL) lseek(sfd, (int32_t)(*offset), 0);
int64_t writeLen = 0; int64_t writeLen = 0;
......
...@@ -34,7 +34,7 @@ void taosWinSocketInit() { ...@@ -34,7 +34,7 @@ void taosWinSocketInit() {
} }
} }
int32_t taosSetNonblocking(SOCKET sock, int32_t on) { int32_t taosSetNonblocking(int32_t sock, int32_t on) {
u_long mode; u_long mode;
if (on) { if (on) {
mode = 1; mode = 1;
...@@ -50,7 +50,7 @@ void taosIgnSIGPIPE() {} ...@@ -50,7 +50,7 @@ void taosIgnSIGPIPE() {}
void taosBlockSIGPIPE() {} void taosBlockSIGPIPE() {}
void taosSetMaskSIGPIPE() {} void taosSetMaskSIGPIPE() {}
int32_t taosSetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen) { int32_t taosSetSockOpt(int32_t socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen) {
if (level == SOL_SOCKET && optname == TCP_KEEPCNT) { if (level == SOL_SOCKET && optname == TCP_KEEPCNT) {
return 0; return 0;
} }
......
...@@ -166,7 +166,7 @@ typedef struct HttpThread { ...@@ -166,7 +166,7 @@ typedef struct HttpThread {
HttpContext * pHead; HttpContext * pHead;
pthread_mutex_t threadMutex; pthread_mutex_t threadMutex;
bool stop; bool stop;
SOCKET pollFd; int32_t pollFd;
int32_t numOfContexts; int32_t numOfContexts;
int32_t threadId; int32_t threadId;
char label[HTTP_LABEL_SIZE]; char label[HTTP_LABEL_SIZE];
...@@ -179,7 +179,7 @@ typedef struct HttpServer { ...@@ -179,7 +179,7 @@ typedef struct HttpServer {
uint16_t serverPort; uint16_t serverPort;
int8_t stop; int8_t stop;
int8_t reserve; int8_t reserve;
SOCKET fd; int32_t fd;
int32_t numOfThreads; int32_t numOfThreads;
int32_t methodScannerLen; int32_t methodScannerLen;
int32_t requestNum; int32_t requestNum;
......
...@@ -260,7 +260,7 @@ bool httpInitConnect() { ...@@ -260,7 +260,7 @@ bool httpInitConnect() {
return false; return false;
} }
pThread->pollFd = (SOCKET)epoll_create(HTTP_MAX_EVENTS); // size does not matter pThread->pollFd = (int32_t)epoll_create(HTTP_MAX_EVENTS); // size does not matter
if (pThread->pollFd < 0) { if (pThread->pollFd < 0) {
httpError("http thread:%s, failed to create HTTP epoll", pThread->label); httpError("http thread:%s, failed to create HTTP epoll", pThread->label);
pthread_mutex_destroy(&(pThread->threadMutex)); pthread_mutex_destroy(&(pThread->threadMutex));
......
...@@ -31,7 +31,7 @@ ...@@ -31,7 +31,7 @@
typedef struct SFdObj { typedef struct SFdObj {
void *signature; void *signature;
SOCKET fd; // TCP socket FD int32_t fd; // TCP socket FD
int closedByApp; // 1: already closed by App int closedByApp; // 1: already closed by App
void *thandle; // handle from upper layer, like TAOS void *thandle; // handle from upper layer, like TAOS
uint32_t ip; uint32_t ip;
...@@ -47,7 +47,7 @@ typedef struct SThreadObj { ...@@ -47,7 +47,7 @@ typedef struct SThreadObj {
pthread_mutex_t mutex; pthread_mutex_t mutex;
uint32_t ip; uint32_t ip;
bool stop; bool stop;
SOCKET pollFd; int32_t pollFd;
int numOfFds; int numOfFds;
int threadId; int threadId;
char label[TSDB_LABEL_LEN]; char label[TSDB_LABEL_LEN];
...@@ -56,7 +56,7 @@ typedef struct SThreadObj { ...@@ -56,7 +56,7 @@ typedef struct SThreadObj {
} SThreadObj; } SThreadObj;
typedef struct { typedef struct {
SOCKET fd; int32_t fd;
uint32_t ip; uint32_t ip;
uint16_t port; uint16_t port;
int8_t stop; int8_t stop;
...@@ -69,7 +69,7 @@ typedef struct { ...@@ -69,7 +69,7 @@ typedef struct {
} SServerObj; } SServerObj;
static void *taosProcessTcpData(void *param); static void *taosProcessTcpData(void *param);
static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd); static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, int32_t fd);
static void taosFreeFdObj(SFdObj *pFdObj); static void taosFreeFdObj(SFdObj *pFdObj);
static void taosReportBrokenLink(SFdObj *pFdObj); static void taosReportBrokenLink(SFdObj *pFdObj);
static void *taosAcceptTcpConnection(void *arg); static void *taosAcceptTcpConnection(void *arg);
...@@ -134,7 +134,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread ...@@ -134,7 +134,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
break; break;
} }
pThreadObj->pollFd = (SOCKET)epoll_create(10); // size does not matter pThreadObj->pollFd = (int32_t)epoll_create(10); // size does not matter
if (pThreadObj->pollFd < 0) { if (pThreadObj->pollFd < 0) {
tError("%s failed to create TCP epoll", label); tError("%s failed to create TCP epoll", label);
code = -1; code = -1;
...@@ -227,7 +227,7 @@ void taosCleanUpTcpServer(void *handle) { ...@@ -227,7 +227,7 @@ void taosCleanUpTcpServer(void *handle) {
} }
static void *taosAcceptTcpConnection(void *arg) { static void *taosAcceptTcpConnection(void *arg) {
SOCKET connFd = -1; int32_t connFd = -1;
struct sockaddr_in caddr; struct sockaddr_in caddr;
int threadId = 0; int threadId = 0;
SThreadObj *pThreadObj; SThreadObj *pThreadObj;
...@@ -238,7 +238,7 @@ static void *taosAcceptTcpConnection(void *arg) { ...@@ -238,7 +238,7 @@ static void *taosAcceptTcpConnection(void *arg) {
while (1) { while (1) {
socklen_t addrlen = sizeof(caddr); socklen_t addrlen = sizeof(caddr);
connFd = accept(pServerObj->fd, (struct sockaddr *)&caddr, &addrlen); connFd = (int32_t)accept(pServerObj->fd, (struct sockaddr *)&caddr, &addrlen);
if (pServerObj->stop) { if (pServerObj->stop) {
tDebug("%s TCP server stop accepting new connections", pServerObj->label); tDebug("%s TCP server stop accepting new connections", pServerObj->label);
break; break;
...@@ -306,7 +306,7 @@ void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void * ...@@ -306,7 +306,7 @@ void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *
return NULL; return NULL;
} }
pThreadObj->pollFd = (SOCKET)epoll_create(10); // size does not matter pThreadObj->pollFd = (int32_t)epoll_create(10); // size does not matter
if (pThreadObj->pollFd < 0) { if (pThreadObj->pollFd < 0) {
tError("%s failed to create TCP client epoll", label); tError("%s failed to create TCP client epoll", label);
free(pThreadObj); free(pThreadObj);
...@@ -351,7 +351,7 @@ void taosCleanUpTcpClient(void *chandle) { ...@@ -351,7 +351,7 @@ void taosCleanUpTcpClient(void *chandle) {
void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) { void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) {
SThreadObj * pThreadObj = shandle; SThreadObj * pThreadObj = shandle;
SOCKET fd = taosOpenTcpClientSocket(ip, port, pThreadObj->ip); int32_t fd = taosOpenTcpClientSocket(ip, port, pThreadObj->ip);
if (fd < 0) return NULL; if (fd < 0) return NULL;
struct sockaddr_in sin; struct sockaddr_in sin;
...@@ -541,7 +541,7 @@ static void *taosProcessTcpData(void *param) { ...@@ -541,7 +541,7 @@ static void *taosProcessTcpData(void *param) {
return NULL; return NULL;
} }
static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd) { static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, int32_t fd) {
struct epoll_event event; struct epoll_event event;
SFdObj *pFdObj = (SFdObj *)calloc(sizeof(SFdObj), 1); SFdObj *pFdObj = (SFdObj *)calloc(sizeof(SFdObj), 1);
......
...@@ -31,7 +31,7 @@ ...@@ -31,7 +31,7 @@
typedef struct { typedef struct {
int index; int index;
SOCKET fd; int32_t fd;
uint16_t port; // peer port uint16_t port; // peer port
uint16_t localPort; // local port uint16_t localPort; // local port
char label[TSDB_LABEL_LEN]; // copy from udpConnSet; char label[TSDB_LABEL_LEN]; // copy from udpConnSet;
......
...@@ -82,8 +82,8 @@ typedef struct SsyncPeer { ...@@ -82,8 +82,8 @@ typedef struct SsyncPeer {
uint64_t sversion; // track the peer version in retrieve process uint64_t sversion; // track the peer version in retrieve process
uint64_t lastFileVer; // track the file version while retrieve uint64_t lastFileVer; // track the file version while retrieve
uint64_t lastWalVer; // track the wal version while retrieve uint64_t lastWalVer; // track the wal version while retrieve
SOCKET syncFd; int32_t syncFd;
SOCKET peerFd; // forward FD int32_t peerFd; // forward FD
int32_t numOfRetrieves; // number of retrieves tried int32_t numOfRetrieves; // number of retrieves tried
int32_t fileChanged; // a flag to indicate file is changed during retrieving process int32_t fileChanged; // a flag to indicate file is changed during retrieving process
int32_t refCount; int32_t refCount;
......
...@@ -27,12 +27,12 @@ typedef struct { ...@@ -27,12 +27,12 @@ typedef struct {
int32_t bufferSize; int32_t bufferSize;
void (*processBrokenLink)(int64_t handleId); void (*processBrokenLink)(int64_t handleId);
int32_t (*processIncomingMsg)(int64_t handleId, void *buffer); int32_t (*processIncomingMsg)(int64_t handleId, void *buffer);
void (*processIncomingConn)(SOCKET fd, uint32_t ip); void (*processIncomingConn)(int32_t fd, uint32_t ip);
} SPoolInfo; } SPoolInfo;
void *syncOpenTcpThreadPool(SPoolInfo *pInfo); void *syncOpenTcpThreadPool(SPoolInfo *pInfo);
void syncCloseTcpThreadPool(void *); void syncCloseTcpThreadPool(void *);
void *syncAllocateTcpConn(void *, int64_t rid, SOCKET connFd); void *syncAllocateTcpConn(void *, int64_t rid, int32_t connFd);
void syncFreeTcpConn(void *); void syncFreeTcpConn(void *);
#ifdef __cplusplus #ifdef __cplusplus
......
...@@ -28,16 +28,16 @@ ...@@ -28,16 +28,16 @@
#include "syncTcp.h" #include "syncTcp.h"
static void arbSignalHandler(int32_t signum); static void arbSignalHandler(int32_t signum);
static void arbProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp); static void arbProcessIncommingConnection(int32_t connFd, uint32_t sourceIp);
static void arbProcessBrokenLink(int64_t rid); static void arbProcessBrokenLink(int64_t rid);
static int32_t arbProcessPeerMsg(int64_t rid, void *buffer); static int32_t arbProcessPeerMsg(int64_t rid, void *buffer);
static tsem_t tsArbSem; static tsem_t tsArbSem;
static void * tsArbTcpPool; static void * tsArbTcpPool;
typedef struct { typedef struct {
char id[TSDB_EP_LEN + 24]; char id[TSDB_EP_LEN + 24];
SOCKET nodeFd; int32_t nodeFd;
void * pConn; void * pConn;
} SNodeConn; } SNodeConn;
int32_t main(int32_t argc, char *argv[]) { int32_t main(int32_t argc, char *argv[]) {
...@@ -106,7 +106,7 @@ int32_t main(int32_t argc, char *argv[]) { ...@@ -106,7 +106,7 @@ int32_t main(int32_t argc, char *argv[]) {
return 0; return 0;
} }
static void arbProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp) { static void arbProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
char ipstr[24]; char ipstr[24];
tinet_ntoa(ipstr, sourceIp); tinet_ntoa(ipstr, sourceIp);
sDebug("peer TCP connection from ip:%s", ipstr); sDebug("peer TCP connection from ip:%s", ipstr);
......
...@@ -45,7 +45,7 @@ static void syncCheckPeerConnection(void *param, void *tmrId); ...@@ -45,7 +45,7 @@ static void syncCheckPeerConnection(void *param, void *tmrId);
static int32_t syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack, int8_t type, uint16_t tranId); static int32_t syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack, int8_t type, uint16_t tranId);
static void syncProcessBrokenLink(int64_t rid); static void syncProcessBrokenLink(int64_t rid);
static int32_t syncProcessPeerMsg(int64_t rid, void *buffer); static int32_t syncProcessPeerMsg(int64_t rid, void *buffer);
static void syncProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp); static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp);
static void syncRemovePeer(SSyncPeer *pPeer); static void syncRemovePeer(SSyncPeer *pPeer);
static void syncAddArbitrator(SSyncNode *pNode); static void syncAddArbitrator(SSyncNode *pNode);
static void syncFreeNode(void *); static void syncFreeNode(void *);
...@@ -1114,8 +1114,8 @@ static void syncSetupPeerConnection(SSyncPeer *pPeer) { ...@@ -1114,8 +1114,8 @@ static void syncSetupPeerConnection(SSyncPeer *pPeer) {
return; return;
} }
SOCKET connFd = taosOpenTcpClientSocket(pPeer->ip, pPeer->port, 0); int32_t connFd = taosOpenTcpClientSocket(pPeer->ip, pPeer->port, 0);
if (connFd < 0) { if ((int32_t)connFd < 0) {
sDebug("%s, failed to open tcp socket since %s", pPeer->id, strerror(errno)); sDebug("%s, failed to open tcp socket since %s", pPeer->id, strerror(errno));
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer); taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);
return; return;
...@@ -1179,7 +1179,7 @@ static void syncCreateRestoreDataThread(SSyncPeer *pPeer) { ...@@ -1179,7 +1179,7 @@ static void syncCreateRestoreDataThread(SSyncPeer *pPeer) {
} }
} }
static void syncProcessIncommingConnection(SOCKET connFd, uint32_t sourceIp) { static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
char ipstr[24]; char ipstr[24];
int32_t i; int32_t i;
......
...@@ -35,7 +35,7 @@ ...@@ -35,7 +35,7 @@
typedef struct SThreadObj { typedef struct SThreadObj {
pthread_t thread; pthread_t thread;
bool stop; bool stop;
SOCKET pollFd; int32_t pollFd;
int32_t numOfFds; int32_t numOfFds;
struct SPoolObj *pPool; struct SPoolObj *pPool;
} SThreadObj; } SThreadObj;
...@@ -45,14 +45,14 @@ typedef struct SPoolObj { ...@@ -45,14 +45,14 @@ typedef struct SPoolObj {
SThreadObj **pThread; SThreadObj **pThread;
pthread_t thread; pthread_t thread;
int32_t nextId; int32_t nextId;
SOCKET acceptFd; // FD for accept new connection int32_t acceptFd; // FD for accept new connection
int8_t stop; int8_t stop;
} SPoolObj; } SPoolObj;
typedef struct { typedef struct {
SThreadObj *pThread; SThreadObj *pThread;
int64_t handleId; int64_t handleId;
SOCKET fd; int32_t fd;
int32_t closedByApp; int32_t closedByApp;
} SConnObj; } SConnObj;
...@@ -128,7 +128,7 @@ void syncCloseTcpThreadPool(void *param) { ...@@ -128,7 +128,7 @@ void syncCloseTcpThreadPool(void *param) {
tfree(pPool); tfree(pPool);
} }
void *syncAllocateTcpConn(void *param, int64_t rid, SOCKET connFd) { void *syncAllocateTcpConn(void *param, int64_t rid, int32_t connFd) {
struct epoll_event event; struct epoll_event event;
SPoolObj *pPool = param; SPoolObj *pPool = param;
...@@ -264,13 +264,13 @@ static void *syncAcceptPeerTcpConnection(void *argv) { ...@@ -264,13 +264,13 @@ static void *syncAcceptPeerTcpConnection(void *argv) {
while (1) { while (1) {
struct sockaddr_in clientAddr; struct sockaddr_in clientAddr;
socklen_t addrlen = sizeof(clientAddr); socklen_t addrlen = sizeof(clientAddr);
SOCKET connFd = accept(pPool->acceptFd, (struct sockaddr *)&clientAddr, &addrlen); int32_t connFd = (int32_t)accept(pPool->acceptFd, (struct sockaddr *)&clientAddr, &addrlen);
if (pPool->stop) { if (pPool->stop) {
sDebug("%p TCP server accept is stopped", pPool); sDebug("%p TCP server accept is stopped", pPool);
break; break;
} }
if (connFd < 0) { if ((int32_t)connFd < 0) {
if (errno == EINVAL) { if (errno == EINVAL) {
sDebug("%p TCP server accept is exiting...", pPool); sDebug("%p TCP server accept is exiting...", pPool);
break; break;
...@@ -298,7 +298,7 @@ static SThreadObj *syncGetTcpThread(SPoolObj *pPool) { ...@@ -298,7 +298,7 @@ static SThreadObj *syncGetTcpThread(SPoolObj *pPool) {
if (pThread == NULL) return NULL; if (pThread == NULL) return NULL;
pThread->pPool = pPool; pThread->pPool = pPool;
pThread->pollFd = (SOCKET)epoll_create(10); // size does not matter pThread->pollFd = (int32_t)epoll_create(10); // size does not matter
if (pThread->pollFd < 0) { if (pThread->pollFd < 0) {
tfree(pThread); tfree(pThread);
return NULL; return NULL;
......
...@@ -28,17 +28,17 @@ extern "C" { ...@@ -28,17 +28,17 @@ extern "C" {
#define EPOLLWAKEUP (1u << 29) #define EPOLLWAKEUP (1u << 29)
#endif #endif
int32_t taosReadn(SOCKET sock, char *buffer, int32_t len); int32_t taosReadn(int32_t sock, char *buffer, int32_t len);
int32_t taosWriteMsg(SOCKET fd, void *ptr, int32_t nbytes); int32_t taosWriteMsg(int32_t fd, void *ptr, int32_t nbytes);
int32_t taosReadMsg(SOCKET fd, void *ptr, int32_t nbytes); int32_t taosReadMsg(int32_t fd, void *ptr, int32_t nbytes);
int32_t taosNonblockwrite(SOCKET fd, char *ptr, int32_t nbytes); int32_t taosNonblockwrite(int32_t fd, char *ptr, int32_t nbytes);
int32_t taosCopyFds(SOCKET sfd, SOCKET dfd, int64_t len); int32_t taosCopyFds(int32_t sfd, int32_t dfd, int64_t len);
int32_t taosSetNonblocking(SOCKET sock, int32_t on); int32_t taosSetNonblocking(int32_t sock, int32_t on);
SOCKET taosOpenUdpSocket(uint32_t localIp, uint16_t localPort); int32_t taosOpenUdpSocket(uint32_t localIp, uint16_t localPort);
SOCKET taosOpenTcpClientSocket(uint32_t ip, uint16_t port, uint32_t localIp); int32_t taosOpenTcpClientSocket(uint32_t ip, uint16_t port, uint32_t localIp);
SOCKET taosOpenTcpServerSocket(uint32_t ip, uint16_t port); int32_t taosOpenTcpServerSocket(uint32_t ip, uint16_t port);
int32_t taosKeepTcpAlive(SOCKET sockFd); int32_t taosKeepTcpAlive(int32_t sockFd);
int32_t taosGetFqdn(char *); int32_t taosGetFqdn(char *);
uint32_t taosGetIpv4FromFqdn(const char *); uint32_t taosGetIpv4FromFqdn(const char *);
......
...@@ -39,7 +39,7 @@ typedef struct { ...@@ -39,7 +39,7 @@ typedef struct {
static void *taosNetBindUdpPort(void *sarg) { static void *taosNetBindUdpPort(void *sarg) {
STestInfo *pinfo = (STestInfo *)sarg; STestInfo *pinfo = (STestInfo *)sarg;
int32_t port = pinfo->port; int32_t port = pinfo->port;
SOCKET serverSocket; int32_t serverSocket;
char buffer[BUFFER_SIZE]; char buffer[BUFFER_SIZE];
int32_t iDataNum; int32_t iDataNum;
socklen_t sin_size; socklen_t sin_size;
...@@ -48,7 +48,7 @@ static void *taosNetBindUdpPort(void *sarg) { ...@@ -48,7 +48,7 @@ static void *taosNetBindUdpPort(void *sarg) {
struct sockaddr_in server_addr; struct sockaddr_in server_addr;
struct sockaddr_in clientAddr; struct sockaddr_in clientAddr;
if ((serverSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) { if ((serverSocket = (int32_t)socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) {
uError("failed to create UDP socket since %s", strerror(errno)); uError("failed to create UDP socket since %s", strerror(errno));
return NULL; return NULL;
} }
...@@ -104,14 +104,14 @@ static void *taosNetBindTcpPort(void *sarg) { ...@@ -104,14 +104,14 @@ static void *taosNetBindTcpPort(void *sarg) {
struct sockaddr_in server_addr; struct sockaddr_in server_addr;
struct sockaddr_in clientAddr; struct sockaddr_in clientAddr;
STestInfo *pinfo = sarg; STestInfo *pinfo = sarg;
int32_t port = pinfo->port; int32_t port = pinfo->port;
SOCKET serverSocket; int32_t serverSocket;
int32_t addr_len = sizeof(clientAddr); int32_t addr_len = sizeof(clientAddr);
SOCKET client; int32_t client;
char buffer[BUFFER_SIZE]; char buffer[BUFFER_SIZE];
if ((serverSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) { if ((serverSocket = (int32_t)socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
uError("failed to create TCP socket since %s", strerror(errno)); uError("failed to create TCP socket since %s", strerror(errno));
return NULL; return NULL;
} }
...@@ -148,7 +148,7 @@ static void *taosNetBindTcpPort(void *sarg) { ...@@ -148,7 +148,7 @@ static void *taosNetBindTcpPort(void *sarg) {
uInfo("TCP server at port:%d is listening", port); uInfo("TCP server at port:%d is listening", port);
while (1) { while (1) {
client = accept(serverSocket, (struct sockaddr *)&clientAddr, (socklen_t *)&addr_len); client = (int32_t)accept(serverSocket, (struct sockaddr *)&clientAddr, (socklen_t *)&addr_len);
if (client < 0) { if (client < 0) {
uDebug("TCP: failed to accept at port:%d since %s", port, strerror(errno)); uDebug("TCP: failed to accept at port:%d since %s", port, strerror(errno));
continue; continue;
...@@ -178,10 +178,10 @@ static void *taosNetBindTcpPort(void *sarg) { ...@@ -178,10 +178,10 @@ static void *taosNetBindTcpPort(void *sarg) {
} }
static int32_t taosNetCheckTcpPort(STestInfo *info) { static int32_t taosNetCheckTcpPort(STestInfo *info) {
SOCKET clientSocket; int32_t clientSocket;
char buffer[BUFFER_SIZE] = {0}; char buffer[BUFFER_SIZE] = {0};
if ((clientSocket = socket(AF_INET, SOCK_STREAM, 0)) < 0) { if ((clientSocket = (int32_t)socket(AF_INET, SOCK_STREAM, 0)) < 0) {
uError("failed to create TCP client socket since %s", strerror(errno)); uError("failed to create TCP client socket since %s", strerror(errno));
return -1; return -1;
} }
...@@ -226,14 +226,14 @@ static int32_t taosNetCheckTcpPort(STestInfo *info) { ...@@ -226,14 +226,14 @@ static int32_t taosNetCheckTcpPort(STestInfo *info) {
} }
static int32_t taosNetCheckUdpPort(STestInfo *info) { static int32_t taosNetCheckUdpPort(STestInfo *info) {
SOCKET clientSocket; int32_t clientSocket;
char buffer[BUFFER_SIZE] = {0}; char buffer[BUFFER_SIZE] = {0};
int32_t iDataNum = 0; int32_t iDataNum = 0;
int32_t bufSize = 1024000; int32_t bufSize = 1024000;
struct sockaddr_in serverAddr; struct sockaddr_in serverAddr;
if ((clientSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) { if ((clientSocket = (int32_t)socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) {
uError("failed to create udp client socket since %s", strerror(errno)); uError("failed to create udp client socket since %s", strerror(errno));
return -1; return -1;
} }
......
...@@ -102,7 +102,7 @@ uint32_t ip2uint(const char *const ip_addr) { ...@@ -102,7 +102,7 @@ uint32_t ip2uint(const char *const ip_addr) {
return *((uint32_t *)ip); return *((uint32_t *)ip);
} }
int32_t taosWriteMsg(SOCKET fd, void *buf, int32_t nbytes) { int32_t taosWriteMsg(int32_t fd, void *buf, int32_t nbytes) {
int32_t nleft, nwritten; int32_t nleft, nwritten;
char * ptr = (char *)buf; char * ptr = (char *)buf;
...@@ -128,7 +128,7 @@ int32_t taosWriteMsg(SOCKET fd, void *buf, int32_t nbytes) { ...@@ -128,7 +128,7 @@ int32_t taosWriteMsg(SOCKET fd, void *buf, int32_t nbytes) {
return (nbytes - nleft); return (nbytes - nleft);
} }
int32_t taosReadMsg(SOCKET fd, void *buf, int32_t nbytes) { int32_t taosReadMsg(int32_t fd, void *buf, int32_t nbytes) {
int32_t nleft, nread; int32_t nleft, nread;
char * ptr = (char *)buf; char * ptr = (char *)buf;
...@@ -159,7 +159,7 @@ int32_t taosReadMsg(SOCKET fd, void *buf, int32_t nbytes) { ...@@ -159,7 +159,7 @@ int32_t taosReadMsg(SOCKET fd, void *buf, int32_t nbytes) {
return (nbytes - nleft); return (nbytes - nleft);
} }
int32_t taosNonblockwrite(SOCKET fd, char *ptr, int32_t nbytes) { int32_t taosNonblockwrite(int32_t fd, char *ptr, int32_t nbytes) {
taosSetNonblocking(fd, 1); taosSetNonblocking(fd, 1);
int32_t nleft, nwritten, nready; int32_t nleft, nwritten, nready;
...@@ -201,7 +201,7 @@ int32_t taosNonblockwrite(SOCKET fd, char *ptr, int32_t nbytes) { ...@@ -201,7 +201,7 @@ int32_t taosNonblockwrite(SOCKET fd, char *ptr, int32_t nbytes) {
return (nbytes - nleft); return (nbytes - nleft);
} }
int32_t taosReadn(SOCKET fd, char *ptr, int32_t nbytes) { int32_t taosReadn(int32_t fd, char *ptr, int32_t nbytes) {
int32_t nread, nready, nleft = nbytes; int32_t nread, nready, nleft = nbytes;
fd_set fset; fd_set fset;
...@@ -239,9 +239,9 @@ int32_t taosReadn(SOCKET fd, char *ptr, int32_t nbytes) { ...@@ -239,9 +239,9 @@ int32_t taosReadn(SOCKET fd, char *ptr, int32_t nbytes) {
return (nbytes - nleft); return (nbytes - nleft);
} }
SOCKET taosOpenUdpSocket(uint32_t ip, uint16_t port) { int32_t taosOpenUdpSocket(uint32_t ip, uint16_t port) {
struct sockaddr_in localAddr; struct sockaddr_in localAddr;
SOCKET sockFd; int32_t sockFd;
int32_t bufSize = 1024000; int32_t bufSize = 1024000;
uDebug("open udp socket:0x%x:%hu", ip, port); uDebug("open udp socket:0x%x:%hu", ip, port);
...@@ -279,14 +279,14 @@ SOCKET taosOpenUdpSocket(uint32_t ip, uint16_t port) { ...@@ -279,14 +279,14 @@ SOCKET taosOpenUdpSocket(uint32_t ip, uint16_t port) {
return sockFd; return sockFd;
} }
SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clientIp) { int32_t taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clientIp) {
SOCKET sockFd = 0; int32_t sockFd = 0;
int32_t ret; int32_t ret;
struct sockaddr_in serverAddr, clientAddr; struct sockaddr_in serverAddr, clientAddr;
int32_t bufSize = 1024 * 1024; int32_t bufSize = 1024 * 1024;
sockFd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP); sockFd = (int32_t)socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
if (sockFd <= 2) { if (sockFd <= 2) {
uError("failed to open the socket: %d (%s)", errno, strerror(errno)); uError("failed to open the socket: %d (%s)", errno, strerror(errno));
taosCloseSocketNoCheck(sockFd); taosCloseSocketNoCheck(sockFd);
...@@ -346,7 +346,7 @@ SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clie ...@@ -346,7 +346,7 @@ SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clie
return sockFd; return sockFd;
} }
int32_t taosKeepTcpAlive(SOCKET sockFd) { int32_t taosKeepTcpAlive(int32_t sockFd) {
int32_t alive = 1; int32_t alive = 1;
if (taosSetSockOpt(sockFd, SOL_SOCKET, SO_KEEPALIVE, (void *)&alive, sizeof(alive)) < 0) { if (taosSetSockOpt(sockFd, SOL_SOCKET, SO_KEEPALIVE, (void *)&alive, sizeof(alive)) < 0) {
uError("fd:%d setsockopt SO_KEEPALIVE failed: %d (%s)", sockFd, errno, strerror(errno)); uError("fd:%d setsockopt SO_KEEPALIVE failed: %d (%s)", sockFd, errno, strerror(errno));
...@@ -394,9 +394,9 @@ int32_t taosKeepTcpAlive(SOCKET sockFd) { ...@@ -394,9 +394,9 @@ int32_t taosKeepTcpAlive(SOCKET sockFd) {
return 0; return 0;
} }
SOCKET taosOpenTcpServerSocket(uint32_t ip, uint16_t port) { int32_t taosOpenTcpServerSocket(uint32_t ip, uint16_t port) {
struct sockaddr_in serverAdd; struct sockaddr_in serverAdd;
SOCKET sockFd; int32_t sockFd;
int32_t reuse; int32_t reuse;
uDebug("open tcp server socket:0x%x:%hu", ip, port); uDebug("open tcp server socket:0x%x:%hu", ip, port);
...@@ -449,7 +449,7 @@ void tinet_ntoa(char *ipstr, uint32_t ip) { ...@@ -449,7 +449,7 @@ void tinet_ntoa(char *ipstr, uint32_t ip) {
#define COPY_SIZE 32768 #define COPY_SIZE 32768
// sendfile shall be used // sendfile shall be used
int32_t taosCopyFds(SOCKET sfd, SOCKET dfd, int64_t len) { int32_t taosCopyFds(int32_t sfd, int32_t dfd, int64_t len) {
int64_t leftLen; int64_t leftLen;
int32_t readLen, writeLen; int32_t readLen, writeLen;
char temp[COPY_SIZE]; char temp[COPY_SIZE];
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册