diff options
| -rw-r--r-- | bin/conf/main.conf | 8 | ||||
| -rw-r--r-- | frag_rssb安装使用手册.docx | bin | 80365 -> 81685 bytes | |||
| -rw-r--r-- | src/AV_sendback.h | 2 | ||||
| -rw-r--r-- | src/frag_send.c | 20 | ||||
| -rw-r--r-- | src/hard_keepalive.c | 376 | ||||
| -rw-r--r-- | src/hard_keepalive.h | 7 | ||||
| -rw-r--r-- | src/log.c | 45 | ||||
| -rw-r--r-- | src/log.h | 3 | ||||
| -rw-r--r-- | src/main.c | 14 | ||||
| -rw-r--r-- | src/main.h | 9 |
10 files changed, 274 insertions, 210 deletions
diff --git a/bin/conf/main.conf b/bin/conf/main.conf index 0820712..8b8607e 100644 --- a/bin/conf/main.conf +++ b/bin/conf/main.conf @@ -27,6 +27,7 @@ UnixSocketRecvAddr=/home/mesasoft/frag_rssb/un_recv #udp socket:send frag +UdpSendSwitch=0 UdpSendIP=127.0.0.1;127.0.0.1; UdpSendIPNum=2 UdpSendPort=33082;33083 @@ -237,4 +238,9 @@ capacity=32 cost=32 wlb_report_interval=10 -bfd_recv_port=0 +/*�˿�Ϊ0��ʾ�����������*/ +bfd_recv_port=3785 +/*wait_queue���õĹ�����ֵ*/ +down_threshold=90 +/*�����ж�����*/ +down_num=20 diff --git a/frag_rssb安装使用手册.docx b/frag_rssb安装使用手册.docx Binary files differindex 4e34160..0be6b1b 100644 --- a/frag_rssb安装使用手册.docx +++ b/frag_rssb安装使用手册.docx diff --git a/src/AV_sendback.h b/src/AV_sendback.h index 07de83b..151c1a8 100644 --- a/src/AV_sendback.h +++ b/src/AV_sendback.h @@ -22,7 +22,7 @@ #define PROTOCOL_MMS 0x09 #define PROTOCOL_RTMP 0x0A #define PROTOCOL_SIP 0x0B - + /*ý������*/ #define MEDIA_TYPE_UNKNOWN 0x00 /*ý������:��Ƶ֧�ֵ�ý������*/ diff --git a/src/frag_send.c b/src/frag_send.c index f13b68d..58b053c 100644 --- a/src/frag_send.c +++ b/src/frag_send.c @@ -271,14 +271,16 @@ void send_data_by_unixsocket(const char* data, uint32_t datalen, int thread_id) } void send_data(const char* data, uint32_t datalen, int thread_id) -{ - /*����UDP�ش�*/ -#if K_PROJECT - send_data_by_udp(data, datalen, thread_id); - /*����unix socket�ش�*/ -#else - send_data_by_unixsocket(data, datalen, thread_id); -#endif +{ + if(g_frag_cfg.send_udp_switch) + { + send_data_by_udp(data, datalen, thread_id); + } + else + { + /*����unix socket�ش�*/ + send_data_by_unixsocket(data, datalen, thread_id); + } } void send_data_bizman(const char* data, uint32_t datalen, uint64_t mid,uint32_t ip, int thread_id) @@ -660,7 +662,7 @@ void pack_and_send_media_info(media_info_t* media_info, frag_in_t* frg, int thre if(g_frag_run.usm_on_flag) { send_data_usm(sendbuf,sendbuflen,thread_id); - } + } else { send_data(sendbuf, sendbuflen, thread_id); diff --git a/src/hard_keepalive.c b/src/hard_keepalive.c index ded8991..e25fbea 100644 --- a/src/hard_keepalive.c +++ b/src/hard_keepalive.c @@ -1,174 +1,202 @@ - -#include <sys/socket.h> -#include <netinet/in.h> -#include <pthread.h> -#include <stdio.h> -#include <stdlib.h> -#include <unistd.h> -#include <fcntl.h> -#include <errno.h> -#include <string.h> -#include <sys/types.h> -#include <sys/socket.h> -#include <net/if.h> -#include <netinet/in.h> -#include <arpa/inet.h> -#include <netdb.h> -#include <sys/un.h> -#include <stddef.h>//offsetof - -#include "hard_keepalive.h" - -int udp_socket_recv(int sockfd, uint32_t *src_ip, uint16_t *src_port, uint8_t *buf, uint32_t buf_size) -{ - if (NULL == buf) return -1; - - int numbytes; - struct sockaddr_storage their_addr; - socklen_t addr_len = sizeof(their_addr); - - if ((numbytes = recvfrom(sockfd, buf, buf_size , 0,(struct sockaddr *)&their_addr, &addr_len)) == -1) - { - perror("recvfrom"); - return -1; - } - *src_ip = ((struct sockaddr_in *)&their_addr)->sin_addr.s_addr; - *src_port = ((struct sockaddr_in *)&their_addr)->sin_port; - return numbytes; -} - -// send udp packet -int udp_socket_send(int sockfd, uint32_t addr, uint16_t port, char *data, int datalen) -{ - struct sockaddr_in dst_addr; /* connector's address information */ - dst_addr.sin_family = AF_INET; /* host byte order */ - dst_addr.sin_port = port; /* short, network byte order */ - dst_addr.sin_addr.s_addr = addr; - bzero(&(dst_addr.sin_zero), 8); /* zero the rest of the struct */ - int to_send_len=datalen; - int already_sended_len=0; - while(to_send_len>0) - { - already_sended_len=sendto(sockfd,data, - to_send_len-already_sended_len, - 0, - (struct sockaddr *)&(dst_addr), - sizeof(dst_addr)); - if(already_sended_len==-1) - { - if((EAGAIN == errno)||( EINTR == errno )|| (EWOULDBLOCK==errno)) - { - continue; - } - else - { - return -1; - } - } - to_send_len-=already_sended_len; - } - return already_sended_len; -} - -int create_recv_udp_socket(uint16_t port) -{ - int sockfd = socket(AF_INET, SOCK_DGRAM, 0); - if (-1 == sockfd) - { - perror("listener: socket"); - return -1; - } - struct sockaddr_in my_addr; /* my address information */ - my_addr.sin_family = AF_INET; /* host byte order */ - my_addr.sin_port = port; /* short, network byte order */ - my_addr.sin_addr.s_addr = INADDR_ANY; /* auto-fill with my IP */ - bzero(&(my_addr.sin_zero), 8); /* zero the rest of the struct */ - - if (bind(sockfd, (struct sockaddr *)&my_addr, sizeof(struct sockaddr)) == -1) - { - perror("listener: bind"); - close(sockfd); - return -1; - } - return sockfd; -} - -void* thread_hard_keepalive(void *param) -{ - char buf[1500] = {0}; - char discriminator_temp[ID_SIZE] = {0}; - bfd_port_t* bfd_port = (bfd_port_t*)param; - int size = 0; - uint32_t src_ip = 0; - uint16_t src_port = 0; - int send_rec = 0; - fd_set rset; - int recv_pkt_sd = create_recv_udp_socket(htons(bfd_port->recv_port)); - - if(-1==recv_pkt_sd) - { - printf("hard_keepalive create socket error.\n"); - return NULL; - } - - while(1) - { - FD_ZERO(&rset); - FD_SET(recv_pkt_sd,&rset); - if(-1==select(recv_pkt_sd+1,&rset,NULL,NULL,NULL)) - { - continue; - } - if(FD_ISSET(recv_pkt_sd, &rset)) - { - size = udp_socket_recv(recv_pkt_sd, &src_ip, &src_port, (unsigned char*)buf, sizeof(buf)); - if(size>0) - { - /*����discriminator*/ - memcpy(discriminator_temp, buf+MY_ID_OFFSET, ID_SIZE); - memcpy(buf+MY_ID_OFFSET, buf+YOUR_ID_OFFSET, ID_SIZE); - memcpy(buf+YOUR_ID_OFFSET, discriminator_temp, ID_SIZE); - send_rec = udp_socket_send(recv_pkt_sd, - src_ip, - src_port, - (char*)buf,size); - if(-1==send_rec) - { - printf("hard_keepalive send pkt error.\n"); - } - } - } - else - { - continue; - } - } - - return NULL; -} - -int hard_keepalive_run(void* bfd_port) -{ - pthread_t thread_desc; - pthread_attr_t attr; - - memset(&thread_desc, 0, sizeof(thread_desc)); - memset(&attr, 0, sizeof(attr)); - - if(0 != pthread_attr_init(&(attr))) - { - return -1; - } - if(0 != pthread_attr_setdetachstate(&(attr), PTHREAD_CREATE_DETACHED)) - { - return -1; - } - if(0 != pthread_create(&(thread_desc), &(attr), thread_hard_keepalive, (void*)bfd_port)) - { - pthread_attr_destroy(&(attr)); - return -1; - } - pthread_attr_destroy(&(attr)); - return 0; -} - +
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <string.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <net/if.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <sys/un.h>
+#include <stddef.h>//offsetof
+
+#include "hard_keepalive.h"
+#include "main.h"
+
+extern frag_rssb_parameter_t g_frag_run;
+extern frag_rssb_configure_t g_frag_cfg;
+extern frag_rssb_status_t g_frag_stat;
+
+
+int udp_socket_recv(int sockfd, uint32_t *src_ip, uint16_t *src_port, uint8_t *buf, uint32_t buf_size)
+{
+ if (NULL == buf) return -1;
+
+ int numbytes;
+ struct sockaddr_storage their_addr;
+ socklen_t addr_len = sizeof(their_addr);
+
+ if ((numbytes = recvfrom(sockfd, buf, buf_size , 0,(struct sockaddr *)&their_addr, &addr_len)) == -1)
+ {
+ perror("recvfrom");
+ return -1;
+ }
+ *src_ip = ((struct sockaddr_in *)&their_addr)->sin_addr.s_addr;
+ *src_port = ((struct sockaddr_in *)&their_addr)->sin_port;
+ return numbytes;
+}
+
+// send udp packet
+int udp_socket_send(int sockfd, uint32_t addr, uint16_t port, char *data, int datalen)
+{
+ struct sockaddr_in dst_addr; /* connector's address information */
+ dst_addr.sin_family = AF_INET; /* host byte order */
+ dst_addr.sin_port = port; /* short, network byte order */
+ dst_addr.sin_addr.s_addr = addr;
+ bzero(&(dst_addr.sin_zero), 8); /* zero the rest of the struct */
+ int to_send_len=datalen;
+ int already_sended_len=0;
+ while(to_send_len>0)
+ {
+ already_sended_len=sendto(sockfd,data,
+ to_send_len-already_sended_len,
+ 0,
+ (struct sockaddr *)&(dst_addr),
+ sizeof(dst_addr));
+ if(already_sended_len==-1)
+ {
+ if((EAGAIN == errno)||( EINTR == errno )|| (EWOULDBLOCK==errno))
+ {
+ continue;
+ }
+ else
+ {
+ return -1;
+ }
+ }
+ to_send_len-=already_sended_len;
+ }
+ return already_sended_len;
+}
+
+int create_recv_udp_socket(uint16_t port)
+{
+ int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
+ if (-1 == sockfd)
+ {
+ perror("listener: socket");
+ return -1;
+ }
+ struct sockaddr_in my_addr; /* my address information */
+ my_addr.sin_family = AF_INET; /* host byte order */
+ my_addr.sin_port = port; /* short, network byte order */
+ my_addr.sin_addr.s_addr = INADDR_ANY; /* auto-fill with my IP */
+ bzero(&(my_addr.sin_zero), 8); /* zero the rest of the struct */
+
+ if (bind(sockfd, (struct sockaddr *)&my_addr, sizeof(struct sockaddr)) == -1)
+ {
+ perror("listener: bind");
+ close(sockfd);
+ return -1;
+ }
+ return sockfd;
+}
+
+void* thread_hard_keepalive(void *param)
+{
+ char buf[1500] = {0};
+ char discriminator_temp[ID_SIZE] = {0};
+ bfd_port_t* bfd_port = (bfd_port_t*)param;
+ int size = 0;
+ uint32_t src_ip = 0;
+ uint16_t src_port = 0;
+ int send_rec = 0;
+ fd_set rset;
+ int recv_pkt_sd = create_recv_udp_socket(htons(bfd_port->recv_port));
+
+ if(-1==recv_pkt_sd)
+ {
+ printf("hard_keepalive create socket error.\n");
+ return NULL;
+ }
+
+ while(1)
+ {
+ FD_ZERO(&rset);
+ FD_SET(recv_pkt_sd,&rset);
+ if(-1==select(recv_pkt_sd+1,&rset,NULL,NULL,NULL))
+ {
+ continue;
+ }
+ if(FD_ISSET(recv_pkt_sd, &rset))
+ {
+ size = udp_socket_recv(recv_pkt_sd, &src_ip, &src_port, (unsigned char*)buf, sizeof(buf));
+ if(size>0)
+ {
+ /*����discriminator*/
+ memcpy(discriminator_temp, buf+MY_ID_OFFSET, ID_SIZE);
+ memcpy(buf+MY_ID_OFFSET, buf+YOUR_ID_OFFSET, ID_SIZE);
+ memcpy(buf+YOUR_ID_OFFSET, discriminator_temp, ID_SIZE);
+ /*���Ͷ���state״̬*/
+ int state = 0;
+ for(int i=0;i<g_frag_cfg.bfd_down_num;i++)
+ {
+ state &= getbit(g_frag_cfg.bfd_down_stat,i);
+ }
+ if(state)
+ {
+ /*STATE_DOWN 2bit: 0 1*/
+ clrbit(((bfd_header_t*)buf
)->flags,8);
+ setbit(((bfd_header_t*)buf
)->flags,7);
+ printf("wait_queue is overflow, app is down.\n");
+ MESA_handle_runtime_log(g_frag_run.logger, RLOG_LV_FATAL, FRAG_REASSEMBLY_MODULE_NAME,
+ "{%s:%d} wait_queue is overflow, app is down.",
+ __FILE__,__LINE__);
+ }
+ else
+ {
+ /*STATE_UP 2bit: 1 1*/
+ setbit(((bfd_header_t*)buf
)->flags,8);
+ setbit(((bfd_header_t*)buf
)->flags,7);
+ }
+ send_rec = udp_socket_send(recv_pkt_sd,
+ src_ip,
+ src_port,
+ (char*)buf,size);
+ if(-1==send_rec)
+ {
+ printf("hard_keepalive send pkt error.\n");
+ }
+ }
+ }
+ else
+ {
+ continue;
+ }
+ }
+
+ return NULL;
+}
+
+int hard_keepalive_run(void* bfd_port)
+{
+ pthread_t thread_desc;
+ pthread_attr_t attr;
+
+ memset(&thread_desc, 0, sizeof(thread_desc));
+ memset(&attr, 0, sizeof(attr));
+
+ if(0 != pthread_attr_init(&(attr)))
+ {
+ return -1;
+ }
+ if(0 != pthread_attr_setdetachstate(&(attr), PTHREAD_CREATE_DETACHED))
+ {
+ return -1;
+ }
+ if(0 != pthread_create(&(thread_desc), &(attr), thread_hard_keepalive, (void*)bfd_port))
+ {
+ pthread_attr_destroy(&(attr));
+ return -1;
+ }
+ pthread_attr_destroy(&(attr));
+ return 0;
+}
+
diff --git a/src/hard_keepalive.h b/src/hard_keepalive.h index 7eaa7d3..22f08c2 100644 --- a/src/hard_keepalive.h +++ b/src/hard_keepalive.h @@ -10,10 +10,15 @@ typedef struct bfd_port_s #define MY_ID_OFFSET 4 #define YOUR_ID_OFFSET 8 #define ID_SIZE 4 + +#define setbit(x,y) x|=(1<<y) +#define clrbit(x,y) x&=~(1<<y) +#define getbit(x,y) ((x) >> (y)&1) + typedef struct bfd_header_s { unsigned char version_diag; - unsigned char flags; + unsigned char flags; /*flagǰ2bit��state*/ unsigned char detect_time_multiplier; unsigned char length; unsigned char my_discriminator[4]; /* ҵ?º? ²²â*/ @@ -194,23 +194,25 @@ void* thread_stat_output(void *param) char buf[32] = {0}; int send_dest_num = 0; -#if K_PROJECT - send_dest_num = g_frag_cfg.send_dest_udp_ip_num; - for(uint32_t m=0;m<g_frag_cfg.send_dest_udp_ip_num;m++) - { - inet_ntop(AF_INET, &g_frag_cfg.send_dest_udp_iplist[m], buf, sizeof(buf)); - g_frag_stat.sendlog_line_id[2*m]=FS_register(g_frag_stat.fs_handle,FS_STYLE_LINE,FS_CALC_CURRENT, "udp_send"); - g_frag_stat.sendlog_line_id[2*m+1]=FS_register(g_frag_stat.fs_handle,FS_STYLE_LINE,FS_CALC_CURRENT, "lq_count"); - } -#else - send_dest_num = g_frag_cfg.send_dest_addr_num; - for(uint32_t m=0;m<g_frag_cfg.send_dest_addr_num;m++) + if(g_frag_cfg.send_udp_switch) { - g_frag_stat.sendlog_line_id[2*m]=FS_register(g_frag_stat.fs_handle,FS_STYLE_LINE,FS_CALC_CURRENT, (char*)g_frag_cfg.send_dest_addr[m].sun_path); - g_frag_stat.sendlog_line_id[2*m+1]=FS_register(g_frag_stat.fs_handle,FS_STYLE_LINE,FS_CALC_CURRENT, "lq_count"); - } -#endif - + send_dest_num = g_frag_cfg.send_dest_udp_ip_num; + for(uint32_t m=0;m<g_frag_cfg.send_dest_udp_ip_num;m++) + { + inet_ntop(AF_INET, &g_frag_cfg.send_dest_udp_iplist[m], buf, sizeof(buf)); + g_frag_stat.sendlog_line_id[2*m]=FS_register(g_frag_stat.fs_handle,FS_STYLE_LINE,FS_CALC_CURRENT, "udp_send"); + g_frag_stat.sendlog_line_id[2*m+1]=FS_register(g_frag_stat.fs_handle,FS_STYLE_LINE,FS_CALC_CURRENT, "lq_count"); + } + } + else + { + send_dest_num = g_frag_cfg.send_dest_addr_num; + for(uint32_t m=0;m<g_frag_cfg.send_dest_addr_num;m++) + { + g_frag_stat.sendlog_line_id[2*m]=FS_register(g_frag_stat.fs_handle,FS_STYLE_LINE,FS_CALC_CURRENT, (char*)g_frag_cfg.send_dest_addr[m].sun_path); + g_frag_stat.sendlog_line_id[2*m+1]=FS_register(g_frag_stat.fs_handle,FS_STYLE_LINE,FS_CALC_CURRENT, "lq_count"); + } + } memset(buf, 0, sizeof(buf)); for(uint32_t m=0;m<g_frag_cfg.special_media_wins_port_num;m++) { @@ -369,6 +371,17 @@ void get_rssb_queuelog_count() { frag_rssb.sysinfo_stat[RSSB_WAIT_QUEUE][QUEUE_CURRENT] += MESA_lqueue_get_count(frag_rssb.wait_lq[i]); } + double queue_usage = (double)frag_rssb.sysinfo_stat[RSSB_WAIT_QUEUE][QUEUE_CURRENT]/(double)(g_frag_cfg.thread_num*frag_rssb.wait_lq_num)*100; + if(queue_usage < g_frag_cfg.bfd_threshold) + { + clrbit(g_frag_cfg.bfd_down_stat,g_frag_cfg.bfd_down_index); + } + else + { + setbit(g_frag_cfg.bfd_down_stat,g_frag_cfg.bfd_down_index); + } + g_frag_cfg.bfd_down_index++; + g_frag_cfg.bfd_down_index = g_frag_cfg.bfd_down_index%g_frag_cfg.bfd_down_num; } void get_rssb_hashlog_count() @@ -21,6 +21,9 @@ typedef long atomic_t; #define atomic_set(x,y) ((*(x))=(y)) #endif +#define setbit(x,y) x|=(1<<y) +#define clrbit(x,y) x&=~(1<<y) + /*********************************************debug log type***********************************/ typedef enum { @@ -42,9 +42,9 @@ const char* frag_rssb_version = "2018-08-13T09:00:00"; const char* frag_rssb_version_time = "2018-08-13T09:00:00"; const char* frag_rssb_version_des = "MESA@iie rssb_maskey"; -int FRAG_RSSB_VERSION_1_0_20190121 = 0; -const char* frag_rssb_version_time_in = "2019-01-21"; -const char* frag_rssb_version_des_in = "support youtube"; +int FRAG_RSSB_VERSION_1_0_20190315 = 0; +const char* frag_rssb_version_time_in = "2019-03-15"; +const char* frag_rssb_version_des_in = "support udp socket"; void frag_rssb_history() { //2015.11.15 v1.0 create the project @@ -229,6 +229,7 @@ void frag_rssb_history() //2018.12.11 v4.0//resp_checkresult_search_media_cb add K_PROJECT,and delete send_json_log //2018.12.24 v4.0//send_voip_full_json_log add survey judgement for K //2019.01.21 v4.0//support youtube; update req_frag and template + //2019.03.15 v4.0//keepalive; support udp send } frag_rssb_parameter_t g_frag_run; @@ -767,6 +768,8 @@ int read_conf_and_init(const char* filename) MESA_load_profile_uint_def(filename, "WLB", "wlb_report_interval", (unsigned int*)&g_frag_cfg.wlb_report_interval,10); MESA_load_profile_uint_def(filename, "WLB", "enable_override", (unsigned int*)&g_frag_cfg.enable_override,0); MESA_load_profile_uint_def(filename, "WLB", "bfd_recv_port", (unsigned int*)&g_frag_cfg.bfd_recv_port,0); + MESA_load_profile_short_def(filename, "WLB", "down_threshold", (short*)&g_frag_cfg.bfd_threshold,90); + MESA_load_profile_short_def(filename, "WLB", "down_num", (short*)&g_frag_cfg.bfd_down_num,20); /*send bizman :��ƴװ��ǰ�˷����ݣ�������Ӧ�˿�*/ MESA_load_profile_short_def(filename, "NETWORK", "BizmanAckPort", (short*)&g_frag_cfg.bizman_ack_port,22084); @@ -906,7 +909,8 @@ int read_conf_and_init(const char* filename) } } - /*udp socket : send av data by udp socket*/ + /*udp socket : send av data by udp socket*/ + MESA_load_profile_short_def(filename, "NETWORK", "UdpSendSwitch", (short*)&g_frag_cfg.send_udp_switch, 0); uint32_t* send_udp_ip_serial = NULL; MESA_load_profile_uint_def(filename, "NETWORK", "UdpSendIPNum", (uint32_t*)&g_frag_cfg.send_dest_udp_ip_num,0); memset(conf_buf,0,sizeof(conf_buf)); @@ -933,7 +937,6 @@ int read_conf_and_init(const char* filename) } } -#if K_PROJECT if(g_frag_cfg.send_dest_udp_ip_num>0) { for(i=0; i<g_frag_cfg.thread_num; i++) @@ -948,7 +951,6 @@ int read_conf_and_init(const char* filename) } } } -#endif /*udp socket : data msg send to windows system*/ uint32_t* wins_ip_serial = NULL; @@ -175,7 +175,7 @@ typedef struct frag_rssb_configure_s struct sockaddr_un send_dest_addr[DEST_MAXNUM]; //unix socket : send frag to dest /*send data by udp*/ - uint16_t send_dest_udp_port[DEST_MAXNUM]; //udp socket : send frag to dest + uint16_t send_dest_udp_port[DEST_MAXNUM]; //udp socket : send frag to dest uint32_t send_dest_udp_ip_num; in_addr_t send_dest_udp_iplist[DEST_MAXNUM]; //udp socket : send frag to dest @@ -240,7 +240,12 @@ typedef struct frag_rssb_configure_s uint32_t health_check_port; int16_t save_media; - uint32_t bfd_recv_port; + uint64_t bfd_down_stat; /*ÿһ��bit��һ�μ��״̬������������64��*/ + uint32_t bfd_recv_port; + int16_t bfd_down_index; /*��ǰ�ڼ���*/ + int16_t bfd_threshold; + int16_t bfd_down_num; + int16_t send_udp_switch; }frag_rssb_configure_t; typedef struct frag_rssb_status_s |