Af xdp multi threading

build/build.sh

@@ -23,5 +23,3 @@ echo "--- prepare dependencies ---"
 
 echo "--- building arion-agent ---"
 cmake . && make
-
-fi

build/machine-init.sh

@@ -129,7 +129,10 @@ echo "5--- installing ebpf dependencies ---" && \
     cd /var/local/git && \
     git clone https://github.com/futurewei-cloud/zeta && \
     cd zeta && \
-    ./build.sh && \
+    git submodule update --init --recursive && \
+    cd src/extern/libbpf/src && \
+    mkdir build root && \
+    BUILD_STATIC_ONLY=y OBJDIR=build DESTDIR=root make install && \
     cd ~
 
 echo "6--- installing sqlite3 database ---" && \
@@ -149,7 +152,16 @@ echo "7--- installing sqlite orm lib dependencies ---" && \
     sudo cmake --build build --target install && \
     cd ~
 
-echo "8--- installing double conversion for folly ---" && \
+echo "8--- installing glog for folly ---" && \
+    cd /var/local/git && \
+    git clone https://github.com/google/glog.git && \
+    cd glog && \
+    sudo cmake -S . -B build -G "Unix Makefiles" && \
+    sudo cmake --build build && \
+    sudo cmake --build build --target install && \
+    cd ~
+
+echo "9--- installing double conversion for folly ---" && \
     cd /var/local/git && \
     git clone https://github.com/google/double-conversion.git && \
     cd double-conversion && \
@@ -158,7 +170,7 @@ echo "8--- installing double conversion for folly ---" && \
     sudo make install && \
     cd ~
 
-echo "9--- installing folly lib for concurrent hashmap ---" && \
+echo "10--- installing folly lib for concurrent hashmap ---" && \
     cd /var/local/git && \
     git clone https://github.com/facebook/folly.git && \
     cd folly && \
@@ -168,3 +180,13 @@ echo "9--- installing folly lib for concurrent hashmap ---" && \
     make && \
     sudo make install && \
     cd ~
+
+echo "11--- downloading xdp-project files"
+    cd /var/local/git && \
+    git clone https://github.com/xdp-project/xdp-tutorial&& \
+#    cd xdp-tutorial && \
+#    git submodule update --init --recursive && \
+#    cd src/extern/libbpf/src && \
+#    mkdir build root && \
+#    BUILD_STATIC_ONLY=y OBJDIR=build DESTDIR=root make install && \
+    cd ~

include/af_xdp_user_multi_thread.h

@@ -0,0 +1,33 @@
+//
+// Created by ubuntu on 10/4/22.
+//
+
+#ifndef ARIONAGENT_AF_XDP_USER_MULTI_THREADED_H
+#define ARIONAGENT_AF_XDP_USER_MULTI_THREADED_H
+
+#include "logger.h"
+#include <bpf.h>
+#include <xsk.h>
+#include <errno.h>
+#include <string>
+#ifdef __cplusplus
+extern "C"
+{
+#include "common_params.h"
+#include "common_user_bpf_xdp.h"
+#include "common_libbpf.h"
+}
+#endif
+static const char *__d__ = "AF_XDP kernel bypass example multi threaded\n";
+
+class af_xdp_user_multi_thread {
+public:
+    af_xdp_user_multi_thread() {
+        printf("%s", "Start of multithread af_xdp userspace program.");
+    }
+    static void* run_af_xdp_multi_threaded(void* args/*std::string table_name_neighbor_ebpf_map*/);
+private:
+
+};
+
+#endif //ARIONAGENT_AF_XDP_USER_H

include/grpc_client.h

@@ -38,14 +38,18 @@ class ArionMasterWatcherImpl final : public Watch::Service {
 
     explicit ArionMasterWatcherImpl() {}
 
-    void RequestNeighborRules(ArionWingRequest *request, grpc::CompletionQueue *cq);
+    void RequestArionMaster(std::vector<ArionWingRequest *> *request_vector, grpc::CompletionQueue *cq);
 
     void ConnectToArionMaster();
 
-    void RunClient(std::string ip, std::string port, std::string group, std::string table);
+    void RunClient(std::string ip, std::string port, std::string group, std::string neighbor_table, std::string security_group_rules_table);
 
     bool a = chan_ == nullptr;
 
+    int fd_neighbor_ebpf_map = -1;
+
+    int fd_security_group_ebpf_map = -1;
+
 private:
     std::string server_address;
 
@@ -55,18 +59,20 @@ class ArionMasterWatcherImpl final : public Watch::Service {
 
     std::string table_name_neighbor_ebpf_map;
 
-    int fd_neighbor_ebpf_map = -1;
+    std::string table_name_sg_ebpf_map;
 
     // key std::string is '<vni>-<vpc_ip>', value is inserted version of this neighbor
     folly::ConcurrentHashMap<std::string, int> neighbor_task_map;
 
+    // key std::string is 'securitygroupid', value is inserted version of this security group rule
+    folly::ConcurrentHashMap<std::string, int> security_group_rule_task_map;
     // segment lock for neighbor key version control
     SegmentLock segment_lock;
 };
 
 struct AsyncClientCall {
-    arion::schema::NeighborRule reply;
+    arion::schema::ArionWingResponse reply;
     grpc::ClientContext context;
     grpc::Status status;
-    std::unique_ptr<grpc::ClientAsyncReaderWriter<ArionWingRequest, NeighborRule> > stream;
+    std::unique_ptr<grpc::ClientAsyncReaderWriter<ArionWingRequest, ArionWingResponse> > stream;
 };

include/util.h

@@ -17,6 +17,8 @@
 
 #include <string>
 #include <arpa/inet.h>
+#include <cstring>
+using namespace std;
 
 // the number of characters needed to store the HEX form of IP address
 #define HEX_IP_BUFFER_SIZE 12
@@ -37,7 +39,7 @@
 static inline long ip4tol(const string ip) {
   struct sockaddr_in sa;
   if (inet_pton(AF_INET, ip.c_str(), &(sa.sin_addr)) != 1) {
-    throw std::invalid_argument("Virtual ipv4 address is not in the expected format");
+//    throw std::invalid_argument("Virtual ipv4 address is not in the expected format");
   }
   return sa.sin_addr.s_addr;
 }
@@ -49,4 +51,129 @@ static inline std::uint8_t getNum(char hexChar) {
     return (hexChar - 'A' + 10);
 }
 
+
+static inline __sum16 csum16_add(__sum16 csum, __be16 addend)
+{
+    uint16_t res = (uint16_t)csum;
+
+    res += (__u16)addend;
+    return (__sum16)(res + (res < (__u16)addend));
+}
+
+static inline __sum16 csum16_sub(__sum16 csum, __be16 addend)
+{
+    return csum16_add(csum, ~addend);
+}
+
+static inline void csum_replace2(__sum16 *sum, __be16 old, __be16 present)
+{
+    *sum = ~csum16_add(csum16_sub(~(*sum), old), present);
+}
+
+static inline void trn_set_mac(void *dst, unsigned char *mac)
+{
+    unsigned short *d = static_cast<unsigned short *>(dst);
+    unsigned short *s = (unsigned short *)mac;
+
+    d[0] = s[0];
+    d[1] = s[1];
+    d[2] = s[2];
+}
+
+static inline void trn_set_dst_mac(void *data, unsigned char *dst_mac)
+{
+    trn_set_mac(data, dst_mac);
+}
+
+static inline void trn_set_src_mac(void *data, unsigned char *src_mac)
+{
+    uint8_t *tmp = static_cast<uint8_t *>(data);
+    trn_set_mac((void*)(tmp + 6), src_mac);
+}
+
+static __be32 trn_get_vni(const __u8 *vni)
+{
+    /* Big endian! */
+    return (vni[0] << 16) | (vni[1] << 8) | vni[2];
+}
+
+static inline void trn_set_src_ip(void *data, void *data_end, __u32 saddr)
+{
+    int off = offsetof(struct iphdr, saddr);
+    uint8_t *tmp = static_cast<uint8_t *>(data);
+
+    __u32 *addr = (__u32*)(tmp + off);
+    if ((void *)addr > data_end)
+        return;
+
+    *addr = saddr;
+}
+
+static inline void trn_set_dst_ip(void *data, void *data_end, __u32 daddr)
+{
+    int off = offsetof(struct iphdr, daddr);
+    uint8_t *tmp = static_cast<uint8_t *>(data);
+
+    __u32 *addr = (__u32 *)(tmp + off);
+    if ((void *)addr > data_end)
+        return;
+
+    *addr = daddr;
+}
+
+static inline __u16 trn_csum_fold_helper(__u64 csum)
+{
+    int i;
+#pragma unroll
+    for (i = 0; i < 4; i++) {
+        if (csum >> 16)
+            csum = (csum & 0xffff) + (csum >> 16);
+    }
+    return ~csum;
+}
+
+static inline void trn_ipv4_csum_inline(void *iph, __u64 *csum)
+{
+    __u16 *next_iph_u16 = (__u16 *)iph;
+#pragma clang loop unroll(full)
+    for (int i = 0; i<sizeof(struct iphdr)>> 1; i++) {
+        *csum += *next_iph_u16++;
+    }
+    *csum = trn_csum_fold_helper(*csum);
+}
+
+static inline void trn_set_src_dst_ip_csum(struct iphdr *ip,
+                                           __u32 saddr, __u32 daddr, void *data_end)
+{
+    /* Since the packet destination is being rewritten we also
+	decrement the TTL */
+    ip->ttl--;
+
+    __u64 csum = 0;
+    trn_set_src_ip(ip, data_end, saddr);
+    trn_set_dst_ip(ip, data_end, daddr);
+    csum = 0;
+    ip->check = 0;
+    trn_ipv4_csum_inline(ip, &csum);
+    ip->check = csum;
+
+    //    printf("Modified IP Address, src: 0x%x, dst: 0x%x, csum: 0x%x\n",
+    //              ip->saddr, ip->daddr, ip->check);
+}
+
+static inline void trn_swap_src_dst_mac(void *data)
+{
+    unsigned short *p = static_cast<unsigned short *>(data);
+    unsigned short tmp[3];
+
+    tmp[0] = p[0];
+    tmp[1] = p[1];
+    tmp[2] = p[2];
+    p[0] = p[3];
+    p[1] = p[4];
+    p[2] = p[5];
+    p[3] = tmp[0];
+    p[4] = tmp[1];
+    p[5] = tmp[2];
+}
 #endif

include/xdp/trn_datamodel.h

@@ -29,6 +29,7 @@
 #include <linux/ip.h>
 #include <linux/udp.h>
 #include <linux/types.h>
+#include "bpf.h"
 #define __ALIGNED_64__ __attribute__((aligned(64)))
 #define __ALWAYS_INLINE__ __attribute__((__always_inline__))
 
@@ -147,6 +148,12 @@ typedef struct {
 	unsigned char hmac[6];
 } __attribute__((packed, aligned(4))) endpoint_t;
 
+typedef struct {
+    bool action; // 0 or 1
+    __u16 port_range[2]; // assume it supports only 1 range, such as [9000,9016]
+    __u16 remote_group;  // remote group ID
+} __attribute__((packed, aligned(4))) security_group_rule_t;
+
 typedef struct {
 	__u32 ip;   // IP used for ZGC access
 	__u16 announced;    // non-zero indicates the MAC has been announced locally
@@ -215,3 +222,81 @@ typedef struct {
 		dp_encap_opdata_t encap;
 	} opdata;
 } __attribute__((packed, aligned(8))) flow_ctx_t;
+
+// #if connTrack
+struct ipv4_tuple_t {
+        __u32 saddr;
+        __u32 daddr;
+
+        /* ports */
+        __u16 sport;
+        __u16 dport;
+
+        /* Addresses */
+        __u8 protocol;
+
+        /*TODO: include TCP flags, no use case for the moment! */
+
+} __attribute__((packed));
+
+
+typedef struct  {
+        __u32 vni;
+        struct ipv4_tuple_t tuple;
+} __attribute__((packed)) contrack_key_t;
+
+
+typedef struct  {
+        __u32 hip;
+        unsigned char mac[6];
+        unsigned char hmac[6];
+} __attribute__ ((packed, aligned(4))) contrack_t;
+
+// #endif
+
+// #if sgSupport
+
+typedef struct {
+        __u32 prefixlen; /* up to 32 for AF_INET, 128 for AF_INET6*/
+        __u32 vni;
+        __u16 port;
+        __u8  direction;
+        __u8  protocol;
+        __u32 local_ip;
+        __u32 remote_ip;
+} __attribute__((packed, aligned(4))) sg_cidr_key_t;
+
+
+typedef struct {
+        __u32 sg_id;
+        __u8 action;
+} __attribute__((packed, aligned(4))) sg_cidr_t;
+
+
+typedef struct  {
+        __u32 vni;
+        __u32 ip;
+        __u8  direction;
+} __attribute__((packed, aligned(4))) security_group_key_t;
+
+
+typedef struct {
+        __u32 sg_id;
+        __u8 action;
+} __attribute__((packed, aligned(4))) security_group_t;
+
+
+typedef struct {
+        __u32 vni;
+        __u32 ip;
+        __u8  direction;
+} __attribute__((packed, aligned(4))) port_range_key_t;
+
+typedef struct {
+        __u16 port_min1;
+        __u16 port_max1;
+        __u16 port_min2;
+        __u16 port_max2;
+} __attribute__((packed, aligned(4))) port_range_t;
+
+// #endif