Raw Ethernet: working in tests

receiver/host/IBWrappers.cpp

@@ -5,6 +5,12 @@
 
 #include <thread>
 #include <chrono>
+#include <sys/mman.h>
+
+#ifdef JFJOCH_USE_NUMA
+#include <numaif.h>
+#include <numa.h>
+#endif
 
 #include "IBWrappers.h"
 #include "../../common/JFJochException.h"
@@ -122,11 +128,9 @@ void IBQueuePair::Init() {
     ibv_qp_attr qp_attr{};
     memset(&qp_attr, 0, sizeof(qp_attr));
 
-    int qp_flags = IBV_QP_STATE | IBV_QP_PORT |IBV_QP_PKEY_INDEX | IBV_QP_ACCESS_FLAGS;
+    int qp_flags = IBV_QP_STATE | IBV_QP_PORT;
     qp_attr.qp_state = IBV_QPS_INIT;
     qp_attr.port_num = 1;
-    qp_attr.qp_access_flags = 0;
-    qp_attr.pkey_index = 0;
 
     if (ibv_modify_qp(qp, &qp_attr, qp_flags))
         throw JFJochException(JFJochExceptionCategory::IBVerbs, "Failed modify IB queue pair to init.");
@@ -177,11 +181,11 @@ void IBQueuePair::PostReceiveWR(IBMemoryRegion &mr, uint32_t location, void *poi
     // pointer to packet buffer size and memory key of each packet buffer
     ib_sg_entry.length = size;
     ib_sg_entry.lkey = mr.GetLocalKey();
+    ib_sg_entry.addr = (uintptr_t)(pointer);
 
     ib_wr.num_sge = 1;
     ib_wr.sg_list = &ib_sg_entry;
     ib_wr.next = nullptr;
-    ib_sg_entry.addr = (uintptr_t)(pointer);
     ib_wr.wr_id = location;
 
     int ret;
@@ -193,7 +197,7 @@ void IBQueuePair::PostReceiveWR(IBMemoryRegion &mr, uint32_t location, void *poi
     }
 }
 
-void IBQueuePair::PostInlineSendWR(void *pointer, size_t size) {
+void IBQueuePair::PostSendWR(IBMemoryRegion &mr, void *pointer, size_t size) {
     // Send the frame via RDMA
     ibv_sge ib_sg{};
     ibv_send_wr ib_wr{};
@@ -204,13 +208,14 @@ void IBQueuePair::PostInlineSendWR(void *pointer, size_t size) {
 
     ib_sg.addr	 = (uintptr_t)(pointer);
     ib_sg.length = size;
+    ib_sg.lkey   = mr.GetLocalKey();
 
     ib_wr.wr_id      = UINT64_MAX;
     ib_wr.sg_list    = &ib_sg;
     ib_wr.num_sge    = 1;
 
     ib_wr.opcode     = IBV_WR_SEND;
-    ib_wr.send_flags = IBV_SEND_INLINE;
+    ib_wr.send_flags = IBV_SEND_SIGNALED;
 
     int ret;
     while ((ret = ibv_post_send(qp, &ib_wr, &ib_bad_wr))) {
@@ -221,14 +226,9 @@ void IBQueuePair::PostInlineSendWR(void *pointer, size_t size) {
     }
 }
 
-void IBQueuePair::FlowSteering(uint64_t mac_addr, uint32_t ipv4) {
-    uint8_t dst_mac_addr[6];
-    for (int i = 0; i < 6; i++)
-        dst_mac_addr[i] = (mac_addr & (0xFF << (i*8))) >> (i * 8);
-
+void IBQueuePair::FlowSteeringIPv4(uint32_t ipv4) {
     struct raw_eth_flow_attr {
         struct ibv_flow_attr attr;
-        struct ibv_flow_spec_eth spec_eth;
         struct ibv_flow_spec_ipv4 spec_ipv4;
     } __attribute__((packed)) flow_attr = {
             .attr = {
@@ -236,33 +236,10 @@ void IBQueuePair::FlowSteering(uint64_t mac_addr, uint32_t ipv4) {
                     .type = IBV_FLOW_ATTR_NORMAL,
                     .size = sizeof(flow_attr),
                     .priority = 0,
-                    .num_of_specs = 2,
+                    .num_of_specs = 1,
                     .port = 1,
                     .flags = 0,
             },
-            .spec_eth = {
-                    .type = IBV_FLOW_SPEC_ETH,
-                    .size = sizeof(struct ibv_flow_spec_eth),
-                    .val = {
-                            .dst_mac = {
-                                    dst_mac_addr[0],
-                                    dst_mac_addr[1],
-                                    dst_mac_addr[2],
-                                    dst_mac_addr[3],
-                                    dst_mac_addr[4],
-                                    dst_mac_addr[5]
-                                    },
-                            .src_mac = {0,0,0,0,0,0},
-                            .ether_type = 0x0800,
-                            .vlan_tag = 0x0
-                    },
-                    .mask = {
-                            .dst_mac = {0xFF,0xFF,0xFF,0xFF,0xFF,0xFF},
-                            .src_mac = {0,0,0,0,0,0},
-                            .ether_type = 0xFFFF,
-                            .vlan_tag = 0x0
-                    }
-            },
             .spec_ipv4 = {
                     .type = IBV_FLOW_SPEC_IPV4,
                     .size = sizeof(struct ibv_flow_spec_ipv4),
@@ -277,6 +254,7 @@ void IBQueuePair::FlowSteering(uint64_t mac_addr, uint32_t ipv4) {
             }
     };
 
+
     flow = ibv_create_flow(qp, &flow_attr.attr);
     if (flow == nullptr)
         throw JFJochException(JFJochExceptionCategory::IBVerbs, "Failed to set IB flow steering");
@@ -321,4 +299,48 @@ IBMemoryRegion::~IBMemoryRegion() {
     ibv_dereg_mr(buffer_mr);
 }
 
+IBRegBuffer::IBRegBuffer(IBProtectionDomain &pd, size_t in_loc_count, size_t in_loc_size, int16_t numa_node) :
+        loc_count(in_loc_count), loc_size(in_loc_size) {
+    buffer = (uint8_t *) mmap(nullptr, loc_size * loc_count, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
+    if (buffer == nullptr)
+        throw JFJochException(JFJochExceptionCategory::MemAllocFailed, "frame_buffer");
+
+#ifdef JFJOCH_USE_NUMA
+    if (numa_node >= 0) {
+        unsigned long nodemask = 1L << numa_node;;
+        if (numa_node > sizeof(nodemask)*8) {
+            Unmap();
+            throw JFJochException(JFJochExceptionCategory::MemAllocFailed, "Mask too small for NUMA node");
+        }
+        if (mbind(buffer, loc_size * loc_count, MPOL_BIND, &nodemask, sizeof(nodemask)*8, MPOL_MF_STRICT) == -1) {
+            Unmap();
+            throw JFJochException(JFJochExceptionCategory::MemAllocFailed, "Cannot apply NUMA policy");
+        }
+    }
+#endif
+    memset(buffer, 0, loc_size * loc_count);
+
+    mr = std::make_unique<IBMemoryRegion>(pd, buffer, loc_size * loc_count);
+}
+
+void IBRegBuffer::Unmap() {
+    munmap(buffer, loc_size * loc_count);
+}
+
+IBRegBuffer::~IBRegBuffer() {
+    mr.reset();
+    Unmap();
+}
+
+IBMemoryRegion *IBRegBuffer::GetMemoryRegion() {
+    return mr.get();
+}
+
+uint8_t *IBRegBuffer::GetLocation(uint64_t location) {
+    if (location < loc_count)
+        return buffer + loc_size * location;
+    else
+        throw JFJochException(JFJochExceptionCategory::ArrayOutOfBounds, "Location out of bounds");
+}
+
 #endif //JFJOCH_USE_IBVERBS