add missing features

Makefile

@@ -31,6 +31,7 @@ lib := \
 	flow.o \
 	hexdump.o \
 	histo.o \
+	irq.o \
 	logging.o \
 	loop.o \
 	or_die.o \
@@ -63,6 +64,11 @@ psp_rr-objs := psp_rr_main.o psp_rr.o rr.o psp_lib.o $(lib)
 
 ext-libs := -lm -lrt -lpthread
 
+ifeq ($(WITH_IO_URING),1)
+	CFLAGS += -DWITH_IO_URING
+	ext-libs += -luring
+endif
+
 tcp_rr: $(tcp_rr-objs)
 	$(CC) $(LDFLAGS) -o $@ $^ $(ext-libs)
 

check_all_options.c

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #include "common.h"
 #include "check_all_options.h"
 
@@ -132,3 +134,4 @@ void check_options_psp_common(struct options *opts, struct callbacks *cb)
                       "PSP server requires server IP (-H) for device lookup.");
         }
 }
+/* clang-format on */

check_all_options.h

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #ifndef THIRD_PARTY_NEPER_CHECK_ALL_OPTIONS_H_
 #define THIRD_PARTY_NEPER_CHECK_ALL_OPTIONS_H_
 
@@ -34,3 +36,4 @@ void check_options_psp_common(struct options *opts, struct callbacks *cb);
 
 #endif  // THIRD_PARTY_NEPER_CHECK_ALL_OPTIONS_H_
 
+/* clang-format on */

coef.c

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #include "coef.h"
 #include "common.h"
 
@@ -102,3 +104,4 @@ struct neper_coef *neper_coef(void)
 
         return coef;
 }
+/* clang-format on */

coef.h

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #ifndef THIRD_PARTY_NEPER_COEF_H
 #define THIRD_PARTY_NEPER_COEF_H
 
@@ -44,3 +46,4 @@ struct neper_coef {
 struct neper_coef *neper_coef(void);
 
 #endif
+/* clang-format on */

common.c

@@ -14,18 +14,23 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #include <fcntl.h>
 #include <netinet/tcp.h>
+#include <linux/errqueue.h>
 #include <stddef.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <sys/mman.h>
 
 #include "common.h"
+#include "flow.h"
 #include "lib.h"
 #include "logging.h"
 #include "or_die.h"
+#include "thread.h"
 
 #define kilo (1000)
 #define kibi (1024)
@@ -346,3 +351,100 @@ int create_suicide_timeout(int sec_to_suicide)
         }
         return 0;
 }
+
+#ifdef WITH_IO_URING
+static struct io_uring s_main_ring;
+void io_uring_init_main_ring(struct options *opts)
+{
+        io_uring_queue_init(opts->uring_size, &s_main_ring, 0);
+}
+
+struct io_uring *io_uring_get_main_ring()
+{
+        return &s_main_ring;
+}
+#endif
+
+static void process_cmsg(struct flow *f, struct callbacks *cb,
+                         struct cmsghdr *cmsg)
+{
+        struct sock_extended_err *err;
+
+        switch(cmsg->cmsg_level) {
+        case SOL_IP:
+        case SOL_IPV6:
+                switch (cmsg->cmsg_type) {
+                case IP_RECVERR:
+                case IPV6_RECVERR:
+                {
+                        err = (struct sock_extended_err *)CMSG_DATA(cmsg);
+                        switch (err->ee_origin) {
+                        case SO_EE_ORIGIN_ZEROCOPY:
+                        {
+                                uint32_t lo = err->ee_info;
+                                uint32_t hi = err->ee_data;
+                                uint64_t stat_zcopies = 0;
+                                int ee_copy_event = 0;
+
+                                if (err->ee_code & SO_EE_CODE_ZEROCOPY_COPIED)
+                                        ee_copy_event = 1;
+                                else
+                                        stat_zcopies = hi - lo + 1;
+
+                                flow_update_zstat(f, stat_zcopies,
+                                                  ee_copy_event);
+                                break;
+                        }
+                        default:
+                                LOG_ERROR(cb, "unsupported ee_origin %u\n",
+                                          err->ee_origin);
+                                break;
+                        }
+                        break;
+                }
+                default:
+                        LOG_ERROR(cb, "unsupported cmsg type: %u\n",
+                                  cmsg->cmsg_type);
+                }
+                break;
+
+        default:
+                LOG_ERROR(cb, "unsupported cmsg level: %u\n", cmsg->cmsg_level);
+        }
+}
+
+// This function is used to process error queue when TX_ZEROCOPY is used.
+int do_recvmsg_errqueue(struct thread *t, struct flow *f, uint32_t events)
+{
+        /*
+         * Right now we only process ZEROCOPY related cmsg.
+         */
+        char control[CMSG_SPACE(sizeof(struct sockaddr_in6)) +
+                     CMSG_SPACE(sizeof(struct sock_extended_err))];
+        struct msghdr msg = {
+                .msg_control = control,
+                .msg_controllen = sizeof(control),
+        };
+        struct cmsghdr *cmsg;
+        int fd = flow_fd(f);
+        ssize_t n;
+
+        if (events & EPOLLERR) {
+                do {
+                        n = recvmsg(fd, &msg, MSG_ERRQUEUE);
+                } while(n == -1 && errno == EINTR);
+                if (n == -1) {
+                        if (errno != EAGAIN)
+                                LOG_WARN(t->cb,
+                                         "recvmsg() on ERRQUEUE failed: %s\n",
+                                         strerror(errno));
+                        return errno;
+                }
+                for (cmsg = CMSG_FIRSTHDR(&msg); cmsg;
+                     cmsg = CMSG_NXTHDR(&msg, cmsg))
+                        process_cmsg(f, t->cb, cmsg);
+        }
+        return 0;
+}
+
+/* clang-format on */

common.h

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #ifndef THIRD_PARTY_NEPER_COMMON_H
 #define THIRD_PARTY_NEPER_COMMON_H
 
@@ -168,4 +170,14 @@ void print_unit(const char *name, const void *var, struct callbacks *);
 const struct rate_conversion *auto_unit(const double throughput,
                                         const struct rate_conversion *opt,
                                         struct callbacks *);
+
+int do_recvmsg_errqueue(struct thread *t, struct flow *f, uint32_t events);
+
+#ifdef WITH_IO_URING
+struct io_uring;
+void io_uring_init_main_ring(struct options *opts);
+struct io_uring *io_uring_get_main_ring();
+#endif
+
 #endif
+/* clang-format on */

control_plane.c

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #include "control_plane.h"
 #include <netinet/tcp.h>
 #include <inttypes.h>
@@ -514,3 +516,4 @@ void control_plane_destroy(struct control_plane *cp)
 {
         free(cp);
 }
+/* clang-format on */

control_plane.h

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #ifndef THIRD_PARTY_NEPER_CONTROL_PLANE_H
 #define THIRD_PARTY_NEPER_CONTROL_PLANE_H
 
@@ -36,3 +38,4 @@ int control_plane_incidents(struct control_plane *cp);
 void control_plane_destroy(struct control_plane *cp);
 
 #endif
+/* clang-format on */

countdown_cond.h

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #ifndef _COUNTDOWN_COND_H
 #define _COUNTDOWN_COND_H
 
@@ -92,3 +94,4 @@ static inline void countdown_cond_wait(struct countdown_cond *cc)
 }
 
 #endif
+/* clang-format on */

cpuinfo.c

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #include <ctype.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -87,3 +89,4 @@ int get_cpuinfo(struct cpuinfo *cpus, int max_cpus, struct callbacks *cb)
         fclose(f);
         return n;
 }
+/* clang-format on */

cpuinfo.h

@@ -13,6 +13,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
+/* clang-format off */
 #include "lib.h"
 
 #ifndef THIRD_PARTY_NEPER_CPUINFO_H
@@ -35,3 +37,4 @@ struct cpuinfo {
 int get_cpuinfo(struct cpuinfo *cpus, int max_cpus, struct callbacks *cb);
 
 #endif
+/* clang-format on */

define_all_flags.c

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #include "common.h"
 #include "flags.h"
 #include "lib.h"
@@ -54,9 +56,12 @@ struct flags_parser *add_flags_common(struct flags_parser *fp)
         DEFINE_FLAG(fp, const char *, all_samples,   NULL,    'A', "Print all samples? If yes, this is the output file name");
         DEFINE_FLAG_HAS_OPTIONAL_ARGUMENT(fp, all_samples);
         DEFINE_FLAG_PARSER(fp,        all_samples, parse_all_samples);
+        DEFINE_FLAG_NAMED(fp, bool,    tx_zerocopy,     false, "zerocopy", 'Z', "Set MSG_ZEROCOPY when sending");
         DEFINE_FLAG(fp, bool,         time_wait,     false,    0,  "Do not set SO_LINGER 0. Close gracefully. Active peer will enter TIME_WAIT state");
         DEFINE_FLAG(fp, unsigned long, iostat_ms,    0,        0,  "Print io stats snapshot every this many ms");
         DEFINE_FLAG(fp, unsigned long, wait_start,   0,        0,  "Wait this many seconds before starting any data flows.");
+        DEFINE_FLAG(fp, bool,          use_uring,    false,   'O', "Should use uring for both tx/rx path");
+        DEFINE_FLAG(fp, int,           uring_size,   128,      0,  "io_uring size");
 
         /* Return the updated fp */
         return (fp);
@@ -112,7 +117,6 @@ struct flags_parser *add_flags_stream(struct flags_parser *fp)
         DEFINE_FLAG(fp, int,           test_length,     10,      'l', "Test length in seconds");
         DEFINE_FLAG(fp, bool,          edge_trigger,    false,   'E', "Edge-triggered epoll");
         DEFINE_FLAG(fp, bool,          reuseaddr,       false,   'R', "Use SO_REUSEADDR on sockets");
-        DEFINE_FLAG_NAMED(fp, bool,    tx_zerocopy,     false, "zerocopy", 'Z', "Set MSG_ZEROCOPY when sending");
         DEFINE_FLAG(fp, const struct rate_conversion *, throughput_opt, neper_units_mb_pointer_hack, 0, "Units to display for throughput");
         DEFINE_FLAG_PARSER(fp,                          throughput_opt, parse_unit);
         DEFINE_FLAG_PRINTER(fp,                         throughput_opt, print_unit);
@@ -130,6 +134,7 @@ struct flags_parser *add_flags_tcp_rr(struct flags_parser *fp)
         DEFINE_FLAG(fp, bool,          reuseaddr,       false,   0, "Use SO_REUSEADDR on sockets");
         DEFINE_FLAG(fp, unsigned long, noburst,         0,       0, "noburst interval in ns (default), us, ms, or s");
         DEFINE_FLAG_PARSER(fp,         noburst, parse_duration);
+        DEFINE_FLAG(fp, int,           iov_len,         0,       0, "Number of bytes to put in 1 iovec");
 
         /* Return the updated fp */
         return (fp);
@@ -183,3 +188,4 @@ struct flags_parser *add_flags_udp_stream(struct flags_parser *fp)
         return (fp);
 }
 
+/* clang-format on */

define_all_flags.h

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #ifndef THIRD_PARTY_NEPER_DEFINE_ALL_FLAGS_H
 #define THIRD_PARTY_NEPER_DEFINE_ALL_FLAGS_H
 
@@ -32,3 +34,4 @@ struct flags_parser *add_flags_udp_rr(struct flags_parser *fp);
 struct flags_parser *add_flags_udp_stream(struct flags_parser *fp);
 
 #endif
+/* clang-format on */

flags.c

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #include "flags.h"
 
 #include <ctype.h>
@@ -369,3 +371,4 @@ void flags_parser_dump(struct flags_parser *fp)
                 print_flag(flag, fp->cb);
         }
 }
+/* clang-format on */

flags.h

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+/* clang-format off */
+
 #ifndef THIRD_PARTY_NEPER_FLAGS_H
 #define THIRD_PARTY_NEPER_FLAGS_H
 
@@ -85,3 +87,4 @@ void flags_parser_destroy(struct flags_parser *fp);
 } while (0)
 
 #endif
+/* clang-format on */