Skip to content

Commit 2b0bcb2

Browse files
ahuangfengahuangfeng
committed
feat: new api supporting socket file descriptor as an option
1 parent 9a29f54 commit 2b0bcb2

File tree

6 files changed

+324
-58
lines changed

6 files changed

+324
-58
lines changed

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ examples/resources
1414
sender_continuous
1515
client_continuous
1616
client_monitoring
17+
client_socket
1718

1819
docker/tmp
1920

‎README.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ The collector allows to read and parse UDP-notif protocol messages from a ip/por
1010

1111
The api is in `unyte_udp_collector.h`:
1212
- `unyte_udp_collector_t *unyte_udp_start_collector(unyte_udp_options_t *options)` from `unyte_udp_collector.h`: Initialize the UDP-notif messages collector. It accepts a struct with different options: address (the IP address to listen to), port (port to listen to), recvmmsg_vlen (vlen used on recvmmsg syscall meaning how many messages to receive on every syscall, by default 10)
13+
- `unyte_udp_collector_t *unyte_udp_start_collector_sk(unyte_udp_sk_options_t *options)` from `unyte_udp_collector.h`: Initialize the UDP-notif messages collector binded to a socket. It accepts a struct with different options: socket file descriptor to listen on, recvmmsg_vlen (vlen used on recvmmsg syscall meaning how many messages to receive on every syscall, by default 10)
1314
- `void *unyte_udp_queue_read(unyte_udp_queue_t *queue)` from `unyte_udp_queue.h` : read from a queue a struct with all the message buffer and metadata.
1415
- `int unyte_udp_free_all(unyte_seg_met_t *seg)` from `unyte_udp_collector.h`: free all struct used on a message received.
1516

@@ -205,6 +206,7 @@ int main()
205206
There are some samples implemented during the development of the project [here](examples).
206207
- `client_sample.c` : simple example for minimal usage of the collector library.
207208
- `client_monitoring.c` : sample implementing the monitoring thread to read packets statistics.
209+
- `client_socket.c` : example using a custom socket instead of creating a new one from the library.
208210
- `sender_sample.c` : simple example for minimal usage of the sender library.
209211
- `sender_json.c` : sample reading a json file and sending the bytes by the library.
210212

‎examples/Makefile.am‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ AM_LDFLAGS = $(EXTRA_LDFLAGS)
22

33
ACLOCAL_AMFLAGS = -I m4
44

5-
bin_PROGRAMS = client_monitoring client_sample sender_json sender_sample
5+
bin_PROGRAMS = client_monitoring client_sample client_socket sender_json sender_sample
66

77
client_monitoring_SOURCES = client_monitoring.c
88
client_monitoring_CFLAGS = -I$(srcdir)/../src -Wextra -Wall -ansi -g -std=c11 -D_GNU_SOURCE -fPIC -pthread
@@ -12,6 +12,10 @@ client_sample_SOURCES = client_sample.c
1212
client_sample_CFLAGS = -I$(srcdir)/../src -Wextra -Wall -ansi -g -std=c11 -D_GNU_SOURCE -fPIC -pthread
1313
client_sample_LDADD = $(srcdir)/../src/libunyte-udp-notif.la
1414

15+
client_socket_SOURCES = client_socket.c
16+
client_socket_CFLAGS = -I$(srcdir)/../src -Wextra -Wall -ansi -g -std=c11 -D_GNU_SOURCE -fPIC -pthread
17+
client_socket_LDADD = $(srcdir)/../src/libunyte-udp-notif.la
18+
1519
sender_json_SOURCES = sender_json.c
1620
sender_json_CFLAGS = -I$(srcdir)/../src -Wextra -Wall -ansi -g -std=c11 -D_GNU_SOURCE -fPIC -pthread
1721
sender_json_LDADD = $(srcdir)/../src/libunyte-udp-notif.la

‎examples/client_socket.c‎

Lines changed: 162 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,162 @@
1+
#include <stdio.h>
2+
#include <stdlib.h>
3+
#include <pthread.h>
4+
#include <stdint.h>
5+
#include <string.h>
6+
#include <signal.h>
7+
#include <unistd.h>
8+
#include <arpa/inet.h>
9+
#include <netdb.h>
10+
11+
#include "../src/hexdump.h"
12+
#include "../src/unyte_udp_collector.h"
13+
#include "../src/unyte_udp_utils.h"
14+
#include "../src/unyte_udp_queue.h"
15+
16+
#define USED_VLEN 1
17+
#define MAX_TO_RECEIVE 20
18+
19+
/**
20+
* Creates own custom socket
21+
*/
22+
int create_socket(char *address, char *port)
23+
{
24+
struct addrinfo *addr_info;
25+
struct addrinfo hints;
26+
27+
memset(&hints, 0, sizeof(hints));
28+
29+
hints.ai_socktype = SOCK_DGRAM;
30+
hints.ai_family = AF_UNSPEC;
31+
hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV;
32+
33+
// Using getaddrinfo to support both IPv4 and IPv6
34+
int rc = getaddrinfo(address, port, &hints, &addr_info);
35+
36+
if (rc != 0) {
37+
printf("getaddrinfo error: %s\n", gai_strerror(rc));
38+
exit(EXIT_FAILURE);
39+
}
40+
41+
printf("Address type: %s | %d\n", (addr_info->ai_family == AF_INET) ? "IPv4" : "IPv6", ntohs(((struct sockaddr_in *)addr_info->ai_addr)->sin_port));
42+
43+
// create socket on UDP protocol
44+
int sockfd = socket(addr_info->ai_family, addr_info->ai_socktype, addr_info->ai_protocol);
45+
46+
// handle error
47+
if (sockfd < 0)
48+
{
49+
perror("Cannot create socket");
50+
exit(EXIT_FAILURE);
51+
}
52+
53+
// Use SO_REUSEPORT to be able to launch multiple collector on the same address
54+
int optval = 1;
55+
if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(int)) < 0)
56+
{
57+
perror("Cannot set SO_REUSEPORT option on socket");
58+
exit(EXIT_FAILURE);
59+
}
60+
61+
// Setting socket buffer to default 20 MB
62+
uint64_t receive_buf_size = DEFAULT_SK_BUFF_SIZE;
63+
if (setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &receive_buf_size, sizeof(receive_buf_size)) < 0)
64+
{
65+
perror("Cannot set buffer size");
66+
exit(EXIT_FAILURE);
67+
}
68+
69+
if (bind(sockfd, addr_info->ai_addr, (int)addr_info->ai_addrlen) == -1)
70+
{
71+
perror("Bind failed");
72+
close(sockfd);
73+
exit(EXIT_FAILURE);
74+
}
75+
76+
// free addr_info after usage
77+
freeaddrinfo(addr_info);
78+
79+
return sockfd;
80+
}
81+
82+
int main(int argc, char *argv[])
83+
{
84+
if (argc != 3)
85+
{
86+
printf("Error: arguments not valid\n");
87+
printf("Usage: ./client_socket <ip> <port>\n");
88+
exit(1);
89+
}
90+
91+
int sockfd = create_socket(argv[1], argv[2]);
92+
93+
// Initialize collector options
94+
unyte_udp_sk_options_t options = {0};
95+
options.recvmmsg_vlen = USED_VLEN;
96+
options.socket_fd = sockfd;
97+
printf("Listening on socket %d\n", options.socket_fd);
98+
99+
/* Initialize collector */
100+
unyte_udp_collector_t *collector = unyte_udp_start_collector_sk(&options);
101+
int recv_count = 0;
102+
int max = MAX_TO_RECEIVE;
103+
104+
while (recv_count < max)
105+
{
106+
/* Read queue */
107+
void *seg_pointer = unyte_udp_queue_read(collector->queue);
108+
if (seg_pointer == NULL)
109+
{
110+
printf("seg_pointer null\n");
111+
fflush(stdout);
112+
}
113+
unyte_seg_met_t *seg = (unyte_seg_met_t *)seg_pointer;
114+
115+
// printf("unyte_udp_get_version: %u\n", unyte_udp_get_version(seg));
116+
// printf("unyte_udp_get_space: %u\n", unyte_udp_get_space(seg));
117+
// printf("unyte_udp_get_encoding_type: %u\n", unyte_udp_get_encoding_type(seg));
118+
// printf("unyte_udp_get_header_length: %u\n", unyte_udp_get_header_length(seg));
119+
// printf("unyte_udp_get_message_length: %u\n", unyte_udp_get_message_length(seg));
120+
// printf("unyte_udp_get_generator_id: %u\n", unyte_udp_get_generator_id(seg));
121+
// printf("unyte_udp_get_message_id: %u\n", unyte_udp_get_message_id(seg));
122+
// printf("unyte_udp_get_src[family]: %u\n", unyte_udp_get_src(seg)->ss_family);
123+
// printf("unyte_udp_get_dest_addr[family]: %u\n", unyte_udp_get_dest_addr(seg)->ss_family);
124+
char ip_canonical[100];
125+
if (unyte_udp_get_src(seg)->ss_family == AF_INET) {
126+
printf("src IPv4: %s\n", inet_ntop(unyte_udp_get_src(seg)->ss_family, &((struct sockaddr_in*)unyte_udp_get_src(seg))->sin_addr.s_addr, ip_canonical, sizeof ip_canonical));
127+
printf("src port: %u\n", ntohs(((struct sockaddr_in*)unyte_udp_get_src(seg))->sin_port));
128+
} else {
129+
printf("src IPv6: %s\n", inet_ntop(unyte_udp_get_src(seg)->ss_family, &((struct sockaddr_in6*)unyte_udp_get_src(seg))->sin6_addr.s6_addr, ip_canonical, sizeof ip_canonical));
130+
printf("src port: %u\n", ntohs(((struct sockaddr_in6*)unyte_udp_get_src(seg))->sin6_port));
131+
}
132+
char ip_dest_canonical[100];
133+
if (unyte_udp_get_src(seg)->ss_family == AF_INET) {
134+
printf("dest IPv4: %s\n", inet_ntop(unyte_udp_get_dest_addr(seg)->ss_family, &((struct sockaddr_in*)unyte_udp_get_dest_addr(seg))->sin_addr.s_addr, ip_dest_canonical, sizeof ip_dest_canonical));
135+
printf("dest port: %u\n", ntohs(((struct sockaddr_in*)unyte_udp_get_dest_addr(seg))->sin_port));
136+
} else {
137+
printf("dest IPv6: %s\n", inet_ntop(unyte_udp_get_dest_addr(seg)->ss_family, &((struct sockaddr_in6*)unyte_udp_get_dest_addr(seg))->sin6_addr.s6_addr, ip_dest_canonical, sizeof ip_dest_canonical));
138+
printf("dest port: %u\n", ntohs(((struct sockaddr_in6*)unyte_udp_get_dest_addr(seg))->sin6_port));
139+
}
140+
// printf("unyte_udp_get_payload: %s\n", unyte_udp_get_payload(seg));
141+
// printf("unyte_udp_get_payload_length: %u\n", unyte_udp_get_payload_length(seg));
142+
143+
/* Processing sample */
144+
recv_count++;
145+
print_udp_notif_header(seg->header, stdout);
146+
// hexdump(seg->payload, seg->header->message_length - seg->header->header_length);
147+
148+
fflush(stdout);
149+
150+
/* Struct frees */
151+
unyte_udp_free_all(seg);
152+
}
153+
154+
printf("Shutdown the socket\n");
155+
close(*collector->sockfd);
156+
pthread_join(*collector->main_thread, NULL);
157+
158+
// freeing collector mallocs
159+
unyte_udp_free_collector(collector);
160+
fflush(stdout);
161+
return 0;
162+
}

0 commit comments

Comments
 (0)