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

add a new definition foe label length

This commit is contained in:
Jeff Tao 2020-06-07 06:07:15 +00:00
parent 6f29506e17
commit dad3b9f19c
5 changed files with 41 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/inc/taosdef.h
View File

@ -221,6 +221,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
#define TSDB_COUNTRY_LEN 20
#define TSDB_LOCALE_LEN 64
#define TSDB_TIMEZONE_LEN 64
#define TSDB_LABEL_LEN 8
#define TSDB_FQDN_LEN 128
#define TSDB_EP_LEN (TSDB_FQDN_LEN+6)

4
src/rpc/src/rpcMain.c
View File

@ -47,7 +47,7 @@ typedef struct {
uint16_t localPort;
int8_t connType;
int index; // for UDP server only, round robin for multiple threads
char label[12];
char label[TSDB_LABEL_LEN];
char user[TSDB_UNI_LEN]; // meter ID
char spi; // security parameter index
@ -88,7 +88,7 @@ typedef struct {
} SRpcReqContext;
typedef struct SRpcConn {
char info[50];// debug info: label + pConn + ahandle
char info[48];// debug info: label + pConn + ahandle
int sid; // session ID
uint32_t ownId; // own link ID
uint32_t peerId; // peer link ID

64
src/rpc/src/rpcTcp.c
View File

@ -16,6 +16,7 @@
#include "os.h"
#include "tsocket.h"
#include "tutil.h"
#include "taosdef.h"
#include "taoserror.h"
#include "rpcLog.h"
#include "rpcHead.h"
@ -46,7 +47,7 @@ typedef struct SThreadObj {
int pollFd;
int numOfFds;
int threadId;
char label[12];
char label[TSDB_LABEL_LEN];
void *shandle; // handle passed by upper layer during server initialization
void *(*processData)(SRecvInfo *pPacket);
} SThreadObj;
@ -55,7 +56,7 @@ typedef struct {
int fd;
uint32_t ip;
uint16_t port;
char label[12];
char label[TSDB_LABEL_LEN];
int numOfThreads;
void * shandle;
SThreadObj *pThreadObj;
@ -79,6 +80,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
return NULL;
}
pServerObj->thread = 0;
pServerObj->ip = ip;
pServerObj->port = port;
tstrncpy(pServerObj->label, label, sizeof(pServerObj->label));
@ -93,8 +95,14 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
}
int code = 0;
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
pThreadObj = pServerObj->pThreadObj;
for (int i = 0; i < numOfThreads; ++i) {
pThreadObj->pollFd = -1;
pThreadObj->thread = 0;
pThreadObj->processData = fp;
tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
pThreadObj->shandle = shandle;
@ -114,11 +122,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
break;
}
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
pthread_attr_destroy(&thattr);
if (code != 0) {
tError("%s failed to create TCP process data thread(%s)", label, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
@ -130,11 +134,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
}
if (code == 0) {
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
code = pthread_create(&(pServerObj->thread), &thattr, (void *)taosAcceptTcpConnection, (void *)(pServerObj));
pthread_attr_destroy(&thattr);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
tError("%s failed to create TCP accept thread(%s)", label, strerror(errno));
@ -142,38 +142,39 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
}
if (code != 0) {
free(pServerObj->pThreadObj);
free(pServerObj);
taosCleanUpTcpServer(pServerObj);
pServerObj = NULL;
} else {
tTrace("%s TCP server is initialized, ip:0x%x port:%hu numOfThreads:%d", label, ip, port, numOfThreads);
}
pthread_attr_destroy(&thattr);
return (void *)pServerObj;
}
static void taosStopTcpThread(SThreadObj* pThreadObj) {
pThreadObj->stop = true;
eventfd_t fd = -1;
// signal the thread to stop, try graceful method first,
// and use pthread_cancel when failed
struct epoll_event event = { .events = EPOLLIN };
eventfd_t fd = eventfd(1, 0);
if (fd == -1) {
// failed to create eventfd, call pthread_cancel instead, which may result in data corruption:
tError("%s, failed to create eventfd(%s)", pThreadObj->label, strerror(errno));
pthread_cancel(pThreadObj->thread);
} else if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, fd, &event) < 0) {
// failed to call epoll_ctl, call pthread_cancel instead, which may result in data corruption:
tError("%s, failed to call epoll_ctl(%s)", pThreadObj->label, strerror(errno));
pthread_cancel(pThreadObj->thread);
if (pThreadObj->thread && pThreadObj->pollFd >=0) {
// signal the thread to stop, try graceful method first,
// and use pthread_cancel when failed
struct epoll_event event = { .events = EPOLLIN };
fd = eventfd(1, 0);
if (fd == -1) {
// failed to create eventfd, call pthread_cancel instead, which may result in data corruption:
tError("%s, failed to create eventfd(%s)", pThreadObj->label, strerror(errno));
pthread_cancel(pThreadObj->thread);
} else if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, fd, &event) < 0) {
// failed to call epoll_ctl, call pthread_cancel instead, which may result in data corruption:
tError("%s, failed to call epoll_ctl(%s)", pThreadObj->label, strerror(errno));
pthread_cancel(pThreadObj->thread);
}
}
pthread_join(pThreadObj->thread, NULL);
close(pThreadObj->pollFd);
if (fd != -1) {
close(fd);
}
if (pThreadObj->thread) pthread_join(pThreadObj->thread, NULL);
if (pThreadObj->pollFd >=0) close(pThreadObj->pollFd);
if (fd != -1) close(fd);
while (pThreadObj->pHead) {
SFdObj *pFdObj = pThreadObj->pHead;
@ -188,9 +189,8 @@ void taosCleanUpTcpServer(void *handle) {
SThreadObj *pThreadObj;
if (pServerObj == NULL) return;
shutdown(pServerObj->fd, SHUT_RD);
pthread_join(pServerObj->thread, NULL);
if(pServerObj->fd >=0) shutdown(pServerObj->fd, SHUT_RD);
if(pServerObj->thread) pthread_join(pServerObj->thread, NULL);
for (int i = 0; i < pServerObj->numOfThreads; ++i) {
pThreadObj = pServerObj->pThreadObj + i;

7
src/rpc/src/rpcUdp.c
View File

@ -18,6 +18,7 @@
#include "tsystem.h"
#include "ttimer.h"
#include "tutil.h"
#include "taosdef.h"
#include "rpcLog.h"
#include "rpcUdp.h"
#include "rpcHead.h"
@ -33,7 +34,7 @@ typedef struct {
int fd;
uint16_t port; // peer port
uint16_t localPort; // local port
char label[12]; // copy from udpConnSet;
char label[TSDB_LABEL_LEN]; // copy from udpConnSet;
pthread_t thread;
void *hash;
void *shandle; // handle passed by upper layer during server initialization
@ -49,7 +50,7 @@ typedef struct {
uint16_t port; // local Port
void *shandle; // handle passed by upper layer during server initialization
int threads;
char label[12];
char label[TSDB_LABEL_LEN];
void *(*fp)(SRecvInfo *pPacket);
SUdpConn udpConn[];
} SUdpConnSet;
@ -93,7 +94,7 @@ void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int threads
}
struct sockaddr_in sin;
unsigned int addrlen = sizeof(sin);
unsigned int addrlen = sizeof(sin);
if (getsockname(pConn->fd, (struct sockaddr *)&sin, &addrlen) == 0 && sin.sin_family == AF_INET &&
addrlen == sizeof(sin)) {
pConn->localPort = (uint16_t)ntohs(sin.sin_port);

3
src/util/src/tsched.c
View File

@ -14,6 +14,7 @@
*/
#include "os.h"
#include "taosdef.h"
#include "tulog.h"
#include "tsched.h"
#include "ttimer.h"
@ -21,7 +22,7 @@
#define DUMP_SCHEDULER_TIME_WINDOW 30000 //every 30sec, take a snap shot of task queue.
typedef struct {
char label[16];
char label[TSDB_LABEL_LEN];
tsem_t emptySem;
tsem_t fullSem;
pthread_mutex_t queueMutex;