mcast: add source filter for kernel based backend (#631)

1. Add source filter for kernel based backend. 2. Add multicast support for AF_XDP backend. 3. Fix condition for leave report sending. --------- Signed-off-by: Ric Li <[email protected]>
OpenVisualCloud · Dec 8, 2023 · fde8c05 · fde8c05
1 parent f2eeb10
commit fde8c05
Show file tree

Hide file tree

Showing 11 changed files with 199 additions and 171 deletions.
diff --git a/lib/src/datapath/mt_dp_socket.c b/lib/src/datapath/mt_dp_socket.c
@@ -477,14 +477,23 @@ static int rx_socket_init_fd(struct mt_rx_socket_entry* entry, int fd, bool reus
     return ret;
   }
 
+  /* join multicast group, will drop automatically when socket fd closed */
   if (mt_is_multicast_ip(flow->dip_addr)) {
-    struct ip_mreq mreq;
-    memset(&mreq, 0, sizeof(mreq));
-    /* multicast addr */
-    memcpy(&mreq.imr_multiaddr.s_addr, flow->dip_addr, MTL_IP_ADDR_LEN);
-    /* local nic src ip */
-    memcpy(&mreq.imr_interface.s_addr, mt_sip_addr(impl, port), MTL_IP_ADDR_LEN);
-    ret = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
+    uint32_t source = *(uint32_t*)flow->sip_addr;
+    if (source == 0) {
+      struct ip_mreq mreq;
+      memset(&mreq, 0, sizeof(mreq));
+      memcpy(&mreq.imr_multiaddr.s_addr, flow->dip_addr, MTL_IP_ADDR_LEN);
+      memcpy(&mreq.imr_interface.s_addr, mt_sip_addr(impl, port), MTL_IP_ADDR_LEN);
+      ret = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
+    } else {
+      struct ip_mreq_source mreq;
+      memset(&mreq, 0, sizeof(mreq));
+      memcpy(&mreq.imr_multiaddr.s_addr, flow->dip_addr, MTL_IP_ADDR_LEN);
+      memcpy(&mreq.imr_interface.s_addr, mt_sip_addr(impl, port), MTL_IP_ADDR_LEN);
+      memcpy(&mreq.imr_sourceaddr.s_addr, flow->sip_addr, MTL_IP_ADDR_LEN);
+      ret = setsockopt(fd, IPPROTO_IP, IP_ADD_SOURCE_MEMBERSHIP, &mreq, sizeof(mreq));
+    }
     if (ret < 0) {
       err("%s(%d,%d), join multicast fail %d\n", __func__, port, fd, ret);
       return ret;

diff --git a/lib/src/dev/mt_af_xdp.c b/lib/src/dev/mt_af_xdp.c
@@ -1062,6 +1062,41 @@ struct mt_rx_xdp_entry* mt_rx_xdp_get(struct mtl_main_impl* impl, enum mtl_port
       entry->skip_all_check = false;
   }
 
+  /* join multicast group, will drop automatically when socket fd closed */
+  int mcast_fd = -1;
+  if (mt_is_multicast_ip(flow->dip_addr)) {
+    int ret;
+    mcast_fd = socket(AF_INET, SOCK_DGRAM, 0);
+    if (mcast_fd < 0) {
+      err("%s(%d,%u), create multicast socket fail\n", __func__, port, q);
+      mt_rx_xdp_put(entry);
+      return NULL;
+    }
+    uint32_t source = *(uint32_t*)flow->sip_addr;
+    if (source == 0) {
+      struct ip_mreq mreq;
+      memset(&mreq, 0, sizeof(mreq));
+      memcpy(&mreq.imr_multiaddr.s_addr, flow->dip_addr, MTL_IP_ADDR_LEN);
+      memcpy(&mreq.imr_interface.s_addr, mt_sip_addr(impl, port), MTL_IP_ADDR_LEN);
+      ret = setsockopt(mcast_fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
+    } else {
+      struct ip_mreq_source mreq;
+      memset(&mreq, 0, sizeof(mreq));
+      memcpy(&mreq.imr_multiaddr.s_addr, flow->dip_addr, MTL_IP_ADDR_LEN);
+      memcpy(&mreq.imr_interface.s_addr, mt_sip_addr(impl, port), MTL_IP_ADDR_LEN);
+      memcpy(&mreq.imr_sourceaddr.s_addr, flow->sip_addr, MTL_IP_ADDR_LEN);
+      ret =
+          setsockopt(mcast_fd, IPPROTO_IP, IP_ADD_SOURCE_MEMBERSHIP, &mreq, sizeof(mreq));
+    }
+    if (ret < 0) {
+      err("%s(%d), join multicast fail %d\n", __func__, port, ret);
+      mt_rx_xdp_put(entry);
+      return NULL;
+    }
+    info("%s(%d), join multicast succ\n", __func__, port);
+  }
+  entry->mcast_fd = mcast_fd;
+
   uint8_t* ip = flow->dip_addr;
   info("%s(%d,%u), ip %u.%u.%u.%u port %u\n", __func__, port, q, ip[0], ip[1], ip[2],
        ip[3], flow->dst_port);
@@ -1076,6 +1111,10 @@ int mt_rx_xdp_put(struct mt_rx_xdp_entry* entry) {
   struct mt_xdp_queue* xq = entry->xq;
   struct mt_xdp_priv* xdp = mt_if(impl, port)->xdp;
 
+  if (entry->mcast_fd > 0) {
+    close(entry->mcast_fd);
+  }
+
   if (entry->flow_rsp) {
     mt_rx_flow_free(impl, port, entry->flow_rsp);
     entry->flow_rsp = NULL;

diff --git a/lib/src/dev/mt_dev.c b/lib/src/dev/mt_dev.c
@@ -97,7 +97,7 @@ static const struct mt_dev_driver_info dev_drvs[] = {
         .drv_type = MT_DRV_NATIVE_AF_XDP,
         .flow_type = MT_FLOW_ALL,
         .flags = MT_DRV_F_NOT_DPDK_PMD | MT_DRV_F_NO_CNI | MT_DRV_F_USE_KERNEL_CTL |
-                 MT_DRV_F_RX_POOL_COMMON | MT_DRV_F_KERNEL_BASED,
+                 MT_DRV_F_RX_POOL_COMMON | MT_DRV_F_MCAST_IN_DP | MT_DRV_F_KERNEL_BASED,
     },
 };
 

diff --git a/lib/src/mt_main.h b/lib/src/mt_main.h
@@ -1095,6 +1095,7 @@ struct mt_rx_xdp_entry {
   struct mt_rx_flow_rsp* flow_rsp;
   bool skip_udp_port_check;
   bool skip_all_check;
+  int mcast_fd;
 };
 
 struct mt_flow_impl {