Merge pull request #4 from NetSys/vector_tx_unix

C++-ize RX/TX vector implementation for UnixSocketPort

Author: kot-begemot-uk

CONTRIBUTING.md

@@ -57,3 +57,4 @@ Please add your name to the end of this file and include this file to the PR, un
 * Eran Gampel
 * Tamás Lévai
 * Matthew Mussomele
+* Anton Ivanov

core/drivers/unix_socket.cc

@@ -96,17 +96,20 @@ void UnixSocketAcceptThread::Run() {
   }
 }
 
-static void replenishVector(struct iovec *iovecs, bess::Packet **pktvec, int cnt) {
-    for (int i=0; i < cnt; i++) {
-        pktvec[i] = current_worker.packet_pool()->Alloc();
-        if (pktvec[i]) {
-            iovecs[i].iov_base = pktvec[i]->data();
-            iovecs[i].iov_len = SNBUF_DATA;
-        } else {
-            iovecs[i].iov_base = NULL; /* vectors can have holes, it will just drop the packet */
-            iovecs[i].iov_len = 0;
-        }
+void UnixSocketPort::ReplenishRecvVector(int cnt) {
+  DCHECK_LE(cnt, bess::PacketBatch::kMaxBurst);
+  bool allocated =
+      current_worker.packet_pool()->AllocBulk(pkt_recv_vector_.data(), cnt);
+
+  for (int i = 0; i < cnt; i++) {
+    if (allocated) {
+      recv_iovecs_[i] = {.iov_base = pkt_recv_vector_[i]->data(),
+                         .iov_len = SNBUF_DATA};
+    } else {
+      // vectors can have holes, it will just drop the packet
+      recv_iovecs_[i] = {.iov_base = nullptr, .iov_len = 0};
     }
+  }
 }
 
 CommandResponse UnixSocketPort::Init(const bess::pb::UnixSocketPortArg &arg) {
@@ -171,36 +174,20 @@ CommandResponse UnixSocketPort::Init(const bess::pb::UnixSocketPortArg &arg) {
     return CommandFailure(errno, "unable to start accept thread");
   }
 
-  send_vector = new(std::nothrow) mmsghdr[VECTOR_QUANTUM]{};
-  send_iovecs = new(std::nothrow) iovec[VECTOR_QUANTUM * MAX_SEGS_IN_VECTOR]{};
-
-  recv_vector = new(std::nothrow) mmsghdr[VECTOR_QUANTUM]{};
-  recv_iovecs = new(std::nothrow) iovec[VECTOR_QUANTUM]{};
-  pkt_recv_vector = new(std::nothrow) bess::Packet*[VECTOR_QUANTUM];
-
-  for (int i=0; i < VECTOR_QUANTUM; i++) {
-    pkt_recv_vector[i] = nullptr;
-    recv_iovecs[i].iov_base = NULL;
-    recv_iovecs[i].iov_len = 0;
-    recv_vector[i].msg_hdr.msg_iov = &recv_iovecs[i];
-    recv_vector[i].msg_hdr.msg_iovlen = 1;
-    recv_vector[i].msg_hdr.msg_name = NULL;
-    recv_vector[i].msg_hdr.msg_namelen = 0;
-    recv_vector[i].msg_hdr.msg_control = NULL;
-    recv_vector[i].msg_hdr.msg_controllen = 0;
-    recv_vector[i].msg_hdr.msg_flags = 0;
-  }
-
-  replenishVector(recv_iovecs, pkt_recv_vector, VECTOR_QUANTUM);
-
-  if ((send_vector == nullptr) || (send_iovecs == nullptr) || (recv_iovecs == nullptr) || (recv_vector == nullptr) || (pkt_recv_vector == nullptr)) {
-    DeInit();
-    return CommandFailure(errno, "failed to allocate vector buffers");
+  for (size_t i = 0; i < bess::PacketBatch::kMaxBurst; i++) {
+    recv_vector_[i] = {.msg_hdr = {.msg_name = nullptr,
+                                   .msg_namelen = 0,
+                                   .msg_iov = &recv_iovecs_[i],
+                                   .msg_iovlen = 1,
+                                   .msg_control = nullptr,
+                                   .msg_controllen = 0,
+                                   .msg_flags = 0},
+                       .msg_len = 0};
   }
 
-  for (int i=0; i<VECTOR_QUANTUM; i++) {
-    send_vector[i].msg_hdr.msg_iov = &send_iovecs[i * MAX_SEGS_IN_VECTOR];
-  }
+  recv_iovecs_.fill({.iov_base = nullptr, .iov_len = 0});
+  pkt_recv_vector_.fill(nullptr);
+  ReplenishRecvVector(bess::PacketBatch::kMaxBurst);
 
   return CommandSuccess();
 }
@@ -215,21 +202,9 @@ void UnixSocketPort::DeInit() {
   if (client_fd_ != kNotConnectedFd) {
     close(client_fd_);
   }
-  if (send_vector != nullptr)
-        delete[] send_vector;
-  if (send_iovecs != nullptr)
-        delete[] send_iovecs;
-  if (recv_vector != nullptr)
-        delete[] recv_vector;
-  if (recv_iovecs != nullptr)
-        delete[] recv_iovecs;
-  if (pkt_recv_vector != nullptr) {
-        for (int i = 0; i < VECTOR_QUANTUM ; i++) {
-            if (pkt_recv_vector[i]) {
-                bess::Packet::Free(pkt_recv_vector[i]);
-            }
-        }
-        delete[] pkt_recv_vector;
+
+  for (auto *pkt : pkt_recv_vector_) {
+    bess::Packet::Free(pkt);
   }
 }
 
@@ -251,36 +226,31 @@ int UnixSocketPort::RecvPackets(queue_t qid, bess::Packet **pkts, int cnt) {
   int received = 0;
 
   while (received < cnt) {
-      int ret = recvmmsg(client_fd, recv_vector, std::min(cnt - received, VECTOR_QUANTUM), 0, NULL);
-
-      if (ret > 0) {
-            for (int i=0; i<ret; i++) {
-                if ((recv_iovecs[i].iov_base != NULL) && (recv_vector[i].msg_len > 0)) {
-                    pkt_recv_vector[i]->append(recv_vector[i].msg_len);
-                    pkts[received++] = pkt_recv_vector[i];
-                }
-            }
-            replenishVector(recv_iovecs, pkt_recv_vector, ret);
-      } else {
-            break;
+    int ret =
+        recvmmsg(client_fd, recv_vector_.data(), cnt - received, 0, nullptr);
+
+    if (ret > 0) {
+      for (int i = 0; i < ret; i++) {
+        if ((recv_iovecs_[i].iov_base != nullptr) &&
+            (recv_vector_[i].msg_len > 0)) {
+          pkt_recv_vector_[i]->append(recv_vector_[i].msg_len);
+          pkts[received++] = pkt_recv_vector_[i];
+        }
       }
+      ReplenishRecvVector(ret);
+    } else {
+      break;
+    }
   }
 
   last_idle_ns_ = (received == 0) ? now_ns : 0;
 
   return received;
 }
 
-static void enqueuePacket(struct iovec *iov,  bess::Packet *pkt, int nb_segs) {
-    for (int j = 0; j < nb_segs; j++) {
-      iov[j].iov_base = pkt->head_data();
-      iov[j].iov_len = pkt->head_len();
-      pkt = pkt->next();
-    }
-}
-
 int UnixSocketPort::SendPackets(queue_t qid, bess::Packet **pkts, int cnt) {
-  int sent = 0, vecindex = 0;
+  int i;
+  int sent = 0;
   int client_fd = client_fd_;
 
   DCHECK_EQ(qid, 0);
@@ -289,56 +259,39 @@ int UnixSocketPort::SendPackets(queue_t qid, bess::Packet **pkts, int cnt) {
     return 0;
   }
 
-  for (int i = 0; i < cnt; i++) {
+  size_t iovec_idx = 0;
+  for (i = 0; i < cnt; i++) {
     bess::Packet *pkt = pkts[i];
-
     int nb_segs = pkt->nb_segs();
-    if ((nb_segs <= MAX_SEGS_IN_VECTOR) || (vecindex > VECTOR_QUANTUM)) {
-        enqueuePacket(send_vector[vecindex].msg_hdr.msg_iov, pkt, nb_segs);
-        send_vector[vecindex].msg_hdr.msg_iovlen = nb_segs;
-        vecindex++;
-    } else {
-        if (vecindex > 0) {
-            int vecret;
-            vecret = sendmmsg(client_fd, send_vector, vecindex, 0);
-            vecindex = 0;
-            if (vecret < 0)
-                break;
-            sent += vecret;
-            if (vecret < vecindex)
-                break;
-        }
-        if (nb_segs > MAX_SEGS_IN_VECTOR) {
-            struct iovec iov[nb_segs];
-
-            struct msghdr msg = msghdr();
-            msg.msg_iov = iov;
-            msg.msg_iovlen = nb_segs;
-
-            enqueuePacket(iov, pkt, nb_segs);
 
-            ssize_t ret;
-        
-            ret = sendmsg(client_fd, &msg, 0);
-            if (ret < 0) {
-              break;
-            }
+    for (int j = 0; j < nb_segs; j++) {
+      if (iovec_idx >= send_iovecs_.size()) {
+        break;
+      }
+      send_iovecs_[iovec_idx++] = {
+          .iov_base = pkt->head_data(),
+          .iov_len = static_cast<size_t>(pkt->head_len())};
+      pkt = pkt->next();
+    }
 
-            sent++;
-        }
-     }
+    send_vector_[i] = {
+        .msg_hdr = {.msg_name = nullptr,
+                    .msg_namelen = 0,
+                    .msg_iov = &send_iovecs_[iovec_idx - nb_segs],
+                    .msg_iovlen = static_cast<size_t>(nb_segs),
+                    .msg_control = nullptr,
+                    .msg_controllen = 0,
+                    .msg_flags = 0},
+        .msg_len = 0};
   }
 
-  /* Final flush if there is an outstanding vector */
-
-  if (vecindex > 0) {
-     int vecret;
-     vecret = sendmmsg(client_fd, send_vector, vecindex, 0);
-     if (vecret > 0)
-         sent += vecret;
-  }
-  if (sent) {
-    bess::Packet::Free(pkts, sent);
+  if (!send_vector_.empty()) {
+    sent = sendmmsg(client_fd, send_vector_.data(), i, 0);
+    if (sent > 0) {
+      bess::Packet::Free(pkts, sent);
+    } else {
+      sent = 0;
+    }
   }
 
   return sent;

core/drivers/unix_socket.h

@@ -30,14 +30,13 @@
 
 #ifndef BESS_DRIVERS_UNIXSOCKET_H_
 #define BESS_DRIVERS_UNIXSOCKET_H_
-#ifndef _GNU_SOURCE
-#define _GNU_SOURCE
-#endif
-#include <sys/types.h>
+
 #include <sys/socket.h>
+#include <sys/types.h>
 #include <sys/un.h>
 #include <unistd.h>
 
+#include <array>
 #include <atomic>
 #include <thread>
 
@@ -46,9 +45,6 @@
 
 #include "../utils/syscallthread.h"
 
-#define MAX_SEGS_IN_VECTOR 8
-#define VECTOR_QUANTUM 64
-
 class UnixSocketPort;
 
 // We promise to block only in ppoll(),
@@ -103,11 +99,15 @@ class UnixSocketPort final : public Port {
   int SendPackets(queue_t qid, bess::Packet **pkts, int cnt) override;
 
  private:
-
-  // These rely on there beeing no multiqueue support !!!
-  struct mmsghdr *send_vector, *recv_vector;
-  bess::Packet **pkt_recv_vector;
-  struct iovec *send_iovecs, *recv_iovecs;
+  void ReplenishRecvVector(int cnt);
+
+  // These rely on there being no multiqueue support !!!
+  std::array<bess::Packet *, bess::PacketBatch::kMaxBurst> pkt_recv_vector_;
+  std::array<mmsghdr, bess::PacketBatch::kMaxBurst> recv_vector_;
+  std::array<iovec, bess::PacketBatch::kMaxBurst> recv_iovecs_;
+  // send_iovecs reserves *8 elements for segmented packets
+  std::array<mmsghdr, bess::PacketBatch::kMaxBurst> send_vector_;
+  std::array<iovec, bess::PacketBatch::kMaxBurst * 8> send_iovecs_;
 
   // Value for a disconnected socket.
   static const int kNotConnectedFd = -1;