1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <sys/prctl.h>
#include "wy_singleflow_entry.h"
extern struct wysf_global_info g_wysf_global_info;
static int udp_broadcast_server_socket_init(void *log_runtime, u_int16_t port)
{
struct sockaddr_in servaddr;
int on = 1, socket_fd;
if((socket_fd = socket(AF_INET,SOCK_DGRAM,0))<0)//AF_INET:IPV4 AF_INET6:IPV6
{
return -1;
}
if(setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on))<0)
{
MESA_RUNTIME_LOGV3(log_runtime, RLOG_LV_FATAL, "setsockopt for udp_port %u failed, because:%s!!", port, strerror(errno));
}
on = 5242880;
if(setsockopt(socket_fd, SOL_SOCKET, SO_RCVBUF, &on, sizeof(on))<0)
{
MESA_RUNTIME_LOGV3(log_runtime, RLOG_LV_FATAL, "setsockopt for udp_port %u failed, because:%s!!", port, strerror(errno));
}
bzero(&servaddr,sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(port);
if(bind(socket_fd,(struct sockaddr *)&servaddr,sizeof(servaddr))<0)
{
MESA_RUNTIME_LOGV3(log_runtime, RLOG_LV_FATAL, "bind udp socket port %u failed, because: %s!!", port, strerror(errno));
close(socket_fd);
return -2;
}
evutil_make_socket_nonblocking(socket_fd);
return socket_fd;
}
static long broadcast_stream_tuple_htable_cb(void *data, const uchar *key, uint size, void *user_arg)
{
int32_t clj_ip=*(int32_t *)user_arg;
std::map<int32_t, struct judian_as_group*>::iterator iter;
struct judian_as_group *group;
if((iter = g_wysf_global_info.forwardip2judian->find(clj_ip)) == g_wysf_global_info.forwardip2judian->end())
{
char ipbuffer[32];
inet_ntop(AF_INET, &clj_ip, ipbuffer, 32);
MESA_RUNTIME_LOGV3(g_wysf_global_info.log_runtime, RLOG_LV_FATAL, "lookup group for clj_ip: %s not found!!", ipbuffer);
return -1;
}
group = iter->second;
if(data != NULL)
{
MESA_htable_del(g_wysf_global_info.stream_tuple_mapping, key, size, NULL);
}
MESA_htable_add(g_wysf_global_info.stream_tuple_mapping, key, size, group);
return 0;
}
void udp_server_read_broadcast_cb(evutil_socket_t fd, short events, void *arg)
{
struct udp_broadcast_msg *broadmsg;
char buffer[4096];
int from_len = 0, buf_len, clj_ip;
struct sockaddr_in saddr;
long cb_ret;
from_len = sizeof (struct sockaddr);
while((buf_len = recvfrom (fd, buffer, 4096, 0, (struct sockaddr*)&saddr, (socklen_t *)&from_len))>0)
{
broadmsg = (struct udp_broadcast_msg *)buffer;
if((u_int32_t)buf_len!=sizeof(struct udp_broadcast_msg) || broadmsg->header.magic_number != WYSF_MAGIC_NUMBER ||
broadmsg->header.total_len!=sizeof(struct broadcast_message))
{
continue;
}
clj_ip = broadmsg->msg.clj_ip;
broadmsg->msg.clj_ip = 0;
MESA_htable_search_cb(g_wysf_global_info.stream_tuple_mapping, (uchar*)&broadmsg->msg, sizeof(struct broadcast_message),
broadcast_stream_tuple_htable_cb, &clj_ip, &cb_ret);
from_len = sizeof (struct sockaddr);
}
}
void *thread_fwdnode_breoadcast(void *arg)
{
struct event_base *evbase;
struct event udp_server_event;
evbase = event_base_new();
if((g_wysf_global_info.broadcast_udp_servert_sockfd = udp_broadcast_server_socket_init(g_wysf_global_info.log_runtime, g_wysf_global_info.broadcast_udp_server_port)) < 0)
{
assert(0);return NULL;
}
event_assign(&udp_server_event, evbase, g_wysf_global_info.broadcast_udp_servert_sockfd,
EV_READ|EV_PERSIST, udp_server_read_broadcast_cb, NULL);
event_add(&udp_server_event, NULL);
event_base_dispatch(evbase);
printf("Libevent dispath error, should not run here.\n");
MESA_RUNTIME_LOGV3(g_wysf_global_info.log_runtime, RLOG_LV_FATAL, "Libevent dispath error, should not run here.");
assert(0);return NULL;
}
|
