Merge pull request #79 from insa-unyte/development

Development

Author: ahuangfeng

.gitignore

@@ -19,6 +19,7 @@ client_ebpf_user
 vmlinux.h
 
 docker/tmp
+.DS_Store
 
 **/*.log
 **/logs/run*

README.md

@@ -9,20 +9,16 @@ See [INSTALL](INSTALL.md)
 The collector allows to read and parse UDP-notif protocol messages from a ip/port specified on the parameters. It allows to get directly the buffer and the metadata of the message in a struct.
 
 The api is in `unyte_udp_collector.h`:
-- `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)
-- `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)
+- `int unyte_udp_create_socket(char *address, char *port, uint64_t buffer_size)` from `unyte_udp_utils.h`: Helper that creates and binds a socket to an address and port.
+- `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: socketfd of the socket to listen to, recvmmsg_vlen (vlen used on recvmmsg syscall meaning how many messages to receive on every syscall, by default 10)...
 - `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.
 - `int unyte_udp_free_all(unyte_seg_met_t *seg)` from `unyte_udp_collector.h`: free all struct used on a message received.
 
 Simple example of usage :
 ```c
 #include <stdio.h>
-#include <stdlib.h>
-#include <pthread.h>
-#include <stdint.h>
-#include <string.h>
-#include <signal.h>
 #include <unistd.h>
+#include <arpa/inet.h>
 
 // include installed library headers
 #include <unyte-udp-notif/unyte_udp_collector.h>
@@ -34,10 +30,13 @@ Simple example of usage :
 
 int main()
 {
+  // Initialize socket and bind it to the address
+  int socketfd = unyte_udp_create_socket(ADDR, PORT, DEFAULT_SK_BUFF_SIZE);
+
   // Initialize collector options
   unyte_udp_options_t options = {0};
-  options.address = ADDR;
-  options.port = PORT;
+  // add socket fd reference to options
+  options.socket_fd = socketfd
   // if argument set to 0, defaults are used
   options.recvmmsg_vlen = 0;       // vlen parameter for recvmmsg. Default: 10
   options.output_queue_size = 0;   // output queue size. Default: 1000
@@ -113,8 +112,7 @@ To activate this thread, you must initiate the monitoring thread queue size (`mo
 ```c
 typedef struct
 {
-  char *address;
-  char *port;
+  int socket_fd;                // socket file descriptor
   ...
   uint monitoring_queue_size;   // monitoring queue size if wanted to activate the monitoring thread. Default: 0. Recommended: 500.
   uint monitoring_delay;        // monitoring queue frequence in seconds. Default: 5 seconds
@@ -155,7 +153,6 @@ typedef struct unyte_message
 
 Simple usage of the sender :
 ```c
-#include <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>
 

VERSION

@@ -1 +1 @@
-0.5.0-git
+0.5.1-git

examples/client_monitoring.c

@@ -1,10 +1,6 @@
 #include <stdio.h>
-#include <stdlib.h>
-#include <pthread.h>
-#include <stdint.h>
-#include <string.h>
-#include <signal.h>
 #include <unistd.h>
+#include <arpa/inet.h>
 
 #include "../src/hexdump.h"
 #include "../src/unyte_udp_collector.h"
@@ -42,17 +38,18 @@ int main(int argc, char *argv[])
     exit(1);
   }
 
+  // Create a udp socket with default socket buffer
+  int socketfd = unyte_udp_create_socket(argv[1], argv[2], DEFAULT_SK_BUFF_SIZE);
+  printf("Listening on %s:%s\n", argv[1], argv[2]);
+
   // Initialize collector options
   unyte_udp_options_t options = {0};
-  options.address = argv[1];
-  options.port = argv[2];
   options.recvmmsg_vlen = USED_VLEN;
+  options.socket_fd = socketfd; // passing socket file descriptor to listen to
   options.monitoring_delay = 2;
   options.monitoring_queue_size = 500;
 
-  printf("Listening on %s:%s\n", options.address, options.port);
-
-  /* Initialize collector */
+  // Initialize collector
   unyte_udp_collector_t *collector = unyte_udp_start_collector(&options);
   int recv_count = 0;
   int max = MAX_TO_RECEIVE;

examples/client_sample.c

@@ -1,9 +1,4 @@
 #include <stdio.h>
-#include <stdlib.h>
-#include <pthread.h>
-#include <stdint.h>
-#include <string.h>
-#include <signal.h>
 #include <unistd.h>
 #include <arpa/inet.h>
 
@@ -24,12 +19,14 @@ int main(int argc, char *argv[])
     exit(1);
   }
 
+  // Create a udp socket with default socket buffer
+  int socketfd = unyte_udp_create_socket(argv[1], argv[2], DEFAULT_SK_BUFF_SIZE);
+  printf("Listening on %s:%s\n", argv[1], argv[2]);
+
   // Initialize collector options
   unyte_udp_options_t options = {0};
   options.recvmmsg_vlen = USED_VLEN;
-  options.address = argv[1];
-  options.port = argv[2];
-  printf("Listening on %s:%s\n", options.address, options.port);
+  options.socket_fd = socketfd; // passing socket file descriptor to listen to
 
   /* Initialize collector */
   unyte_udp_collector_t *collector = unyte_udp_start_collector(&options);

examples/client_socket.c

@@ -1,9 +1,5 @@
 #include <stdio.h>
-#include <stdlib.h>
-#include <pthread.h>
-#include <stdint.h>
 #include <string.h>
-#include <signal.h>
 #include <unistd.h>
 #include <arpa/inet.h>
 #include <netdb.h>
@@ -19,7 +15,7 @@
 /**
  * Creates own custom socket
  */
-int create_socket(char *address, char *port)
+int create_custom_socket(char *address, char *port)
 {
   struct addrinfo *addr_info;
   struct addrinfo hints;
@@ -88,16 +84,16 @@ int main(int argc, char *argv[])
     exit(1);
   }
 
-  int sockfd = create_socket(argv[1], argv[2]);
+  int sockfd = create_custom_socket(argv[1], argv[2]);
 
   // Initialize collector options
-  unyte_udp_sk_options_t options = {0};
+  unyte_udp_options_t options = {0};
   options.recvmmsg_vlen = USED_VLEN;
   options.socket_fd = sockfd;
   printf("Listening on socket %d\n", options.socket_fd);
 
   /* Initialize collector */
-  unyte_udp_collector_t *collector = unyte_udp_start_collector_sk(&options);
+  unyte_udp_collector_t *collector = unyte_udp_start_collector(&options);
   int recv_count = 0;
   int max = MAX_TO_RECEIVE;
 

examples/eBPF/README.md

@@ -22,3 +22,11 @@ To build this directory, use the following options on `configure` script:
 ```shell
 $ ./configure --with-ebpf-example --with-linux=/usr/src/linux
 ```
+
+## Debug
+To show the maps:
+(`sudo` may needed)
+```shell
+$ bpftool map dump name tcp_balancing_t
+$ bpftool map dump name udp_balancing_t
+```

examples/eBPF/client_ebpf_user.c

@@ -13,15 +13,8 @@
  *    ./client_ebpf_user 192.168.1.17 10001 2 3
  */
 
-#include <stdio.h>
-#include <stdlib.h>
-#include <pthread.h>
-#include <stdint.h>
-#include <string.h>
-#include <signal.h>
 #include <unistd.h>
 #include <arpa/inet.h>
-#include <netdb.h>
 #include <assert.h>
 #include <bpf/bpf.h>
 #include <bpf/libbpf.h>
@@ -48,77 +41,22 @@ static int libbpf_print_fn(enum libbpf_print_level level, const char *format, va
 }
 
 /**
- * Creates own custom socket
+ * Open socket, loads eBPF program and attaches it to the opened socket.
+ * int socketfd : socket file descriptor to listen to.
+ * uint32_t key : index of the socket to be filled in the eBPF hash table.
+ * uint32_t balancer_count : max values to be used in eBPF reuse. Should be <= MAX_BALANCER_COUNT.
  */
-int create_socket(char *address, char *port)
-{
-  struct addrinfo *addr_info;
-  struct addrinfo hints;
-
-  memset(&hints, 0, sizeof(hints));
-
-  hints.ai_socktype = SOCK_DGRAM;
-  hints.ai_family = AF_UNSPEC;
-  hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV;
-
-  // Using getaddrinfo to support both IPv4 and IPv6
-  int rc = getaddrinfo(address, port, &hints, &addr_info);
-
-  if (rc != 0) {
-    printf("getaddrinfo error: %s\n", gai_strerror(rc));
-    exit(EXIT_FAILURE);
-  }
-
-  printf("Address type: %s | %d\n", (addr_info->ai_family == AF_INET) ? "IPv4" : "IPv6", ntohs(((struct sockaddr_in *)addr_info->ai_addr)->sin_port));
-
-  // create socket on UDP protocol
-  int sockfd = socket(addr_info->ai_family, addr_info->ai_socktype, addr_info->ai_protocol);
-
-  // handle error
-  if (sockfd < 0)
-  {
-    perror("Cannot create socket");
-    exit(EXIT_FAILURE);
-  }
-
-  // Use SO_REUSEPORT to be able to launch multiple collector on the same address
-  int optval = 1;
-  if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(int)) < 0)
-  {
-    perror("Cannot set SO_REUSEPORT option on socket");
-    exit(EXIT_FAILURE);
-  }
-
-  // Setting socket buffer to default 20 MB
-  uint64_t receive_buf_size = DEFAULT_SK_BUFF_SIZE;
-  if (setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &receive_buf_size, sizeof(receive_buf_size)) < 0)
-  {
-    perror("Cannot set buffer size");
-    exit(EXIT_FAILURE);
-  }
-
-  if (bind(sockfd, addr_info->ai_addr, (int)addr_info->ai_addrlen) == -1)
-  {
-    perror("Bind failed");
-    close(sockfd);
-    exit(EXIT_FAILURE);
-  }
-
-  // free addr_info after usage
-  freeaddrinfo(addr_info);
-
-  return sockfd;
-}
-
-int open_socket_attach_ebpf(char *address, char *port, uint32_t key, uint32_t balancer_count)
+int attach_ebpf_to_socket(int socketfd, uint32_t key, uint32_t balancer_count)
 {
   int umap_fd, size_map_fd, prog_fd;
   char filename[] = BPF_KERNEL_PRG;
-  int64_t usock;
+  int64_t usock = socketfd;
   long err = 0;
 
   assert(!balancer_count || key < balancer_count);
   assert(balancer_count <= MAX_BALANCER_COUNT);
+  assert(usock >= 0);
+
   printf("from args: Using hash bucket index %u", key);
   if (balancer_count > 0) printf(" (%u buckets in total)", balancer_count);
   puts("");
@@ -158,10 +96,6 @@ int open_socket_attach_ebpf(char *address, char *port, uint32_t key, uint32_t ba
   umap_fd = bpf_map__fd(udpmap);
   assert(umap_fd);
 
-  usock = create_socket(address, port);
-
-  assert(usock >= 0);
-
   if (setsockopt(usock, SOL_SOCKET, SO_ATTACH_REUSEPORT_EBPF, &prog_fd, sizeof(prog_fd)) != 0) {
     perror("Could not attach BPF prog");
     return 1;
@@ -200,26 +134,31 @@ int open_socket_attach_ebpf(char *address, char *port, uint32_t key, uint32_t ba
   return usock;
 }
 
-
 int main(int argc, char *argv[])
 {
   if (argc != 5)
   {
     printf("Error: arguments not valid\n");
     printf("Usage: ./client_ebpf_user <ip> <port> <index> <loadbalance_max>\n");
+    printf("Example: ./client_ebpf_user 10.0.2.15 10001 0 5\n");
     exit(1);
   }
 
-  int sockfd = open_socket_attach_ebpf(argv[1], argv[2], atoi(argv[3]), atoi(argv[4]));
+  printf("Listening on %s:%s\n", argv[1], argv[2]);
+
+  // Create a udp socket with default socket buffer
+  int socketfd = unyte_udp_create_socket(argv[1], argv[2], DEFAULT_SK_BUFF_SIZE);
+
+  attach_ebpf_to_socket(socketfd, atoi(argv[3]), atoi(argv[4]));
 
   // Initialize collector options
-  unyte_udp_sk_options_t options = {0};
+  unyte_udp_options_t options = {0};
   options.recvmmsg_vlen = USED_VLEN;
-  options.socket_fd = sockfd;
+  options.socket_fd = socketfd;
   printf("Listening on socket %d\n", options.socket_fd);
 
   /* Initialize collector */
-  unyte_udp_collector_t *collector = unyte_udp_start_collector_sk(&options);
+  unyte_udp_collector_t *collector = unyte_udp_start_collector(&options);
   int recv_count = 0;
   int max = MAX_TO_RECEIVE;
 

examples/sender_json.c

@@ -1,4 +1,3 @@
-#include <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>
 

examples/sender_sample.c

@@ -1,4 +1,3 @@
-#include <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>