gfwleak/av-frag-rssb
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1. keepalive add state

Browse Source 
2. support UDP send in SOQ
3. 上述功能写入安装使用手册
This commit is contained in:
lishu
2019-03-15 16:59:18 +08:00
parent 24bc0055e3
commit e2c4b4e6a1
10 changed files with 274 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/AV_sendback.h
View File

@@ -22,7 +22,7 @@
#define PROTOCOL_MMS 0x09
#define PROTOCOL_RTMP 0x0A
#define PROTOCOL_SIP 0x0B
/*ý<><C3BD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>*/
#define MEDIA_TYPE_UNKNOWN 0x00
/*ý<><C3BD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>:<3A><>Ƶ֧<C6B5>ֵ<EFBFBD>ý<EFBFBD><C3BD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>*/

20
src/frag_send.c
View File

@@ -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)
{
/*<2A><><EFBFBD><EFBFBD>UDP<44>ش<EFBFBD>*/
#if K_PROJECT
send_data_by_udp(data, datalen, thread_id);
/*<2A><><EFBFBD><EFBFBD>unix socket<65>ش<EFBFBD>*/
#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
{
/*<2A><><EFBFBD><EFBFBD>unix socket<65>ش<EFBFBD>*/
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);

376
src/hard_keepalive.c
View File

@@ -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)
{
/*<2A><><EFBFBD><EFBFBD>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)
{
/*<2A><><EFBFBD><EFBFBD>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);
/*<2A><><EFBFBD>Ͷ<EFBFBD><CDB6><EFBFBD>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;
}

7
src/hard_keepalive.h
View File

@@ -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<69><74>state*/
unsigned char detect_time_multiplier;
unsigned char length;
unsigned char my_discriminator[4]; /* ҵ?º? ²²â*/

45
src/log.c
View File

@@ -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()

3
src/log.h
View File

@@ -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
{

14
src/main.c
View File

@@ -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 :<3A><>ƴװ<C6B4><D7B0>ǰ<EFBFBD>˷<EFBFBD><CBB7><EFBFBD><EFBFBD>ݣ<EFBFBD><DDA3><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD>˿<EFBFBD>*/
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;

9
src/main.h
View File

@@ -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; /*ÿһ<C3BF><D2BB>bit<69><74>һ<EFBFBD>μ<EFBFBD><CEBC><EFBFBD>״̬<D7B4><CCAC><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>64<36><34>*/
uint32_t bfd_recv_port;
int16_t bfd_down_index; /*<2A><>ǰ<EFBFBD>ڼ<EFBFBD><DABC><EFBFBD>*/
int16_t bfd_threshold;
int16_t bfd_down_num;
int16_t send_udp_switch;
}frag_rssb_configure_t;
typedef struct frag_rssb_status_s